Salesforce API Anomaly Detection (RET Passthrough)

def rule(event):
    # Alert on any Salesforce API Anomaly Event
    # These are generated by Salesforce's Real-Time Event Monitoring
    # when it detects anomalous API activity
    return event.get("EVENT_TYPE") == "ApiAnomalyEventStore"


def title(event):
    # Create descriptive title with user and anomaly details
    user = event.get("USERNAME", event.get("USER_ID", "<UNKNOWN_USER>"))
    summary = event.get("SUMMARY", "API Anomaly Detected")
    score = event.get("SCORE", "<UNKNOWN>")
    return f"Salesforce API Anomaly: {summary} (Score: {score}) - User: {user}"


def severity(event):
    # Map anomaly score to Panther severity
    # Salesforce scores range from 0-100, higher = more anomalous
    score = event.get("SCORE", 0)
    # Ensure score is numeric
    score = score if isinstance(score, (int, float)) else 0

    if score >= 80:
        return "CRITICAL"
    if score >= 60:
        return "HIGH"
    if score >= 40:
        return "MEDIUM"
    if score >= 20:
        return "LOW"
    return "DEFAULT"


def dedup(event):
    # Deduplicate by unique event identifier
    # Use a combination of user, session, and timestamp to avoid duplicate alerts
    user_id = event.get("USER_ID", "unknown")
    session_key = event.get("SESSION_KEY", "unknown")
    timestamp = event.get("TIMESTAMP", "unknown")
    return f"SF_API_ANOMALY_{user_id}_{session_key}_{timestamp}"


def alert_context(event):
    # Provide key context for investigation
    return {
        "User ID": event.get("USER_ID"),
        "Username": event.get("USERNAME"),
        "Session Key": event.get("SESSION_KEY"),
        "Source IP": event.get("SOURCE_IP"),
        "User Type": event.get("USER_TYPE"),
        "Anomaly Score": event.get("SCORE"),
        "Summary": event.get("SUMMARY"),
        "Request ID": event.get("REQUEST_ID"),
        "Organization ID": event.get("ORGANIZATION_ID"),
        "Event Date": event.get("EVENT_DATE"),
    }