Panther rules: salesforce

def rule(event):
    return event.get("EVENT_TYPE", "<NO_EVENT_TYPE_FOUND>") == "LoginAs"


def title(event):
    admin = event.get("DELEGATED_USER_NAME", "<NO_ADMIN_FOUND>")
    user_id = event.get("USER_ID", "<NO_USER_ID_FOUND>")
    return f"Salesforce admin [{admin}] logged in as a regular user with the user id [{user_id}]."

AnalysisType: rule
Description: "Salesforce detection that alerts when an admin logs in as another user. "
DisplayName: "Salesforce Admin Login As User"
Enabled: true
Filename: salesforce_admin_login_as_user.py
Runbook: "Please do an indicator search on USER_ID to find which user was assumed. "
Reference: https://help.salesforce.com/s/articleView?id=sf.logging_in_as_another_user.htm&type=5
Severity: Info
DedupPeriodMinutes: 60
LogTypes:
  - Salesforce.LoginAs
RuleID: "Salesforce.Admin.Login.As.User"
Threshold: 1

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"),
    }

AnalysisType: rule
Description: "Salesforce Real-Time Event Monitoring has detected anomalous API activity. This could indicate compromised credentials, automated abuse, data exfiltration attempts, or other suspicious API usage patterns."
DisplayName: "Salesforce API Anomaly Detection (RET Passthrough)"
Enabled: true
Filename: salesforce_api_anomaly_passthrough.py
Runbook: |
  1. Review the anomaly score and summary to understand what triggered the alert
  2. Investigate the user's recent API activity in Salesforce EventLogFile
  3. Check if the source IP is expected for this user
  4. Review session details and authentication method
  5. Look for other anomalous activity from the same user or session
  6. If confirmed malicious, reset user credentials and review data access
  7. Consider enabling additional API monitoring or rate limiting
Reference: https://developer.salesforce.com/docs/atlas.en-us.platform_events.meta/platform_events/sforce_api_objects_apianomalyevent.htm
Severity: Medium
DedupPeriodMinutes: 60
LogTypes:
  - Salesforce.RealtimeEvent
RuleID: "Salesforce.API.Anomaly.Passthrough"
Threshold: 1
Tags:
  - Salesforce
  - API Security
  - Real-Time Event Monitoring
  - Anomaly Detection
Reports:
  MITRE ATT&CK:
    - TA0001:T1078  # Initial Access: Valid Accounts
    - TA0006:T1110  # Credential Access: Brute Force
    - TA0009:T1530  # Collection: Data from Cloud Storage Object
    - TA0010:T1567  # Exfiltration: Exfiltration Over Web Service

def rule(event):
    # Alert on Salesforce Bulk API Result Events
    # These events are generated when bulk API jobs complete
    # and can indicate large-scale data exfiltration attempts
    return event.get("EVENT_TYPE") == "BulkApiResultEventStore"


def title(event):
    # Create descriptive title with operation and volume details
    user = event.get("USER_NAME", event.get("USER_ID", "<UNKNOWN_USER>"))
    operation = event.get("OPERATION_TYPE", "<UNKNOWN_OPERATION>")
    records = event.get("RECORDS_PROCESSED", 0)
    entity = event.get("ENTITY_NAME", "<UNKNOWN_ENTITY>")

    return f"Salesforce Bulk API: {operation} on {entity} ({records:,} records) - User: {user}"


def severity(event):
    # Map based on operation type and volume
    operation = event.get("OPERATION_TYPE", "")
    records = event.get("RECORDS_PROCESSED", 0)
    # Ensure records is numeric
    records = records if isinstance(records, (int, float)) else 0

    # Query operations are most concerning for data exfiltration
    if operation in ["query", "queryAll"]:
        if records >= 100000:
            severity_level = "CRITICAL"
        elif records >= 50000:
            severity_level = "HIGH"
        elif records >= 10000:
            severity_level = "MEDIUM"
        else:
            severity_level = "DEFAULT"
    # Other operations (insert, update, delete) are also notable
    elif operation in ["delete", "hardDelete"]:
        if records >= 10000:
            severity_level = "HIGH"
        elif records >= 1000:
            severity_level = "MEDIUM"
        else:
            severity_level = "DEFAULT"
    # Default for other operations
    else:
        if records >= 50000:
            severity_level = "HIGH"
        elif records >= 10000:
            severity_level = "MEDIUM"
        else:
            severity_level = "DEFAULT"

    return severity_level


def dedup(event):
    # Deduplicate by job ID to avoid duplicate alerts for the same job
    job_id = event.get("JOB_ID", "unknown")
    return f"SF_BULK_API_{job_id}"


def alert_context(event):
    # Provide comprehensive context for investigation
    return {
        "Job ID": event.get("JOB_ID"),
        "User ID": event.get("USER_ID"),
        "Username": event.get("USER_NAME"),
        "Operation Type": event.get("OPERATION_TYPE"),
        "Entity Name": event.get("ENTITY_NAME"),
        "Records Processed": event.get("RECORDS_PROCESSED"),
        "Number of Batches": event.get("NUMBER_OF_BATCHES"),
        "API Version": event.get("API_VERSION"),
        "Source IP": event.get("SOURCE_IP"),
        "Request ID": event.get("REQUEST_ID"),
        "Organization ID": event.get("ORGANIZATION_ID"),
        "Job Type": event.get("JOB_TYPE"),
    }

AnalysisType: rule
Description: |
  Detects Salesforce Bulk API operations that could indicate data exfiltration attempts. The Bulk API allows users to process large volumes of records (up to millions) asynchronously, making it a common vector for data theft.

  This detection triggers on all Bulk API job completions and adjusts severity based on:
  - Operation type (query operations are highest risk for exfiltration)
  - Volume of records processed
  - Entity/object type being accessed
DisplayName: "Salesforce Bulk API Data Exfiltration"
Enabled: true
Filename: salesforce_bulk_data_exfiltration.py
Runbook: |
  1. Review the operation type and number of records processed
  2. Identify if this is expected behavior for the user (data analyst, integration user, etc.)
  3. Check the user's recent activity for other suspicious patterns
  4. Verify the source IP is expected for this user
  5. Review what entity/object was accessed (sensitive data like Contacts, Accounts, etc.)
  6. Check if data was exported or if this was an internal operation
  7. If confirmed malicious:
     - Immediately disable the user's API access
     - Reset credentials
     - Review all data accessed in the session
     - Check for data exfiltration to external systems
  8. Consider implementing IP restrictions or rate limits for bulk API access
Reference: https://developer.salesforce.com/docs/atlas.en-us.api_asynch.meta/api_asynch/
Severity: Medium
DedupPeriodMinutes: 60
LogTypes:
  - Salesforce.RealtimeEvent
RuleID: "Salesforce.BulkAPI.DataExfiltration"
Threshold: 1
Tags:
  - Salesforce
  - Data Exfiltration
  - Bulk API
  - Insider Threat
Reports:
  MITRE ATT&CK:
    - TA0010:T1567  # Exfiltration: Exfiltration Over Web Service
    - TA0009:T1530  # Collection: Data from Cloud Storage Object
    - TA0010:T1020  # Exfiltration: Automated Exfiltration

def rule(event):
    # Alert on OAuth-related events that may indicate credential abuse
    event_type = event.get("EVENT_TYPE", "")

    # Monitor OAuth token usage and authentication events
    oauth_events = [
        "OAuthTokenRevoked",
        "OAuthTokenRefreshFailed",
        "ApiTotalUsage",
        "ApiConnectedApp",
    ]

    return event_type in oauth_events or "oauth" in str(event_type).lower()


def title(event):
    # Create descriptive title based on event type
    event_type = event.get("EVENT_TYPE", "<UNKNOWN_EVENT>")
    user = event.get("USER_NAME", event.get("USER_ID", "<UNKNOWN_USER>"))
    app_name = event.get("CONNECTED_APP_NAME", event.get("CLIENT_NAME", "<UNKNOWN_APP>"))

    # Special handling for different event types
    if "Revoked" in event_type:
        return f"Salesforce OAuth Token Revoked: {app_name} - User: {user}"
    if "Failed" in event_type:
        return f"Salesforce OAuth Token Refresh Failed: {app_name} - User: {user}"

    return f"Salesforce OAuth Activity: {event_type} - {app_name} - User: {user}"


def severity(event):
    # Map based on event type and context
    event_type = event.get("EVENT_TYPE", "")
    status = str(event.get("STATUS", "")).lower()

    # Token revocation may indicate compromise
    if "Revoked" in event_type:
        return "HIGH"

    # Failed token operations are suspicious
    if "Failed" in event_type or "fail" in status:
        return "MEDIUM"

    # Excessive API usage may indicate abuse
    api_calls = event.get("API_TOTAL_COUNT", 0)
    # Ensure api_calls is numeric
    api_calls = api_calls if isinstance(api_calls, (int, float)) else 0
    if api_calls > 10000:
        return "HIGH"
    if api_calls > 5000:
        return "MEDIUM"

    return "DEFAULT"


def dedup(event):
    # Deduplicate by event type, user, and app
    event_type = event.get("EVENT_TYPE", "unknown")
    user_id = event.get("USER_ID", "unknown")
    app_id = event.get("CONNECTED_APP_ID", event.get("CLIENT_ID", "unknown"))
    return f"SF_OAUTH_ABUSE_{event_type}_{user_id}_{app_id}"


def alert_context(event):
    # Provide comprehensive context for investigation
    return {
        "Event Type": event.get("EVENT_TYPE"),
        "User ID": event.get("USER_ID"),
        "Username": event.get("USER_NAME"),
        "Connected App ID": event.get("CONNECTED_APP_ID"),
        "Connected App Name": event.get("CONNECTED_APP_NAME"),
        "Client ID": event.get("CLIENT_ID"),
        "Client Name": event.get("CLIENT_NAME"),
        "Source IP": event.get("SOURCE_IP"),
        "Status": event.get("STATUS"),
        "API Total Count": event.get("API_TOTAL_COUNT"),
        "Request ID": event.get("REQUEST_ID"),
        "Organization ID": event.get("ORGANIZATION_ID"),
        "Session Key": event.get("SESSION_KEY"),
    }

AnalysisType: rule
Description: |
  Detects OAuth credential abuse and suspicious token usage patterns in Salesforce. OAuth tokens provide API access and can be abused if compromised, making this detection critical for:
  - Stolen or leaked OAuth tokens
  - Token replay attacks
  - Excessive API usage indicating automated abuse
  - Failed token refresh attempts (potential brute force)
  - Unauthorized token revocations

  This detection triggers on OAuth-related security events and adjusts severity based on:
  - Token revocation events (may indicate compromise response)
  - Failed OAuth operations (potential attack attempts)
  - Excessive API usage patterns
DisplayName: "Salesforce OAuth Credential Abuse Detection"
Enabled: true
Filename: salesforce_oauth_credential_abuse.py
Runbook: |
  1. Identify the event type and severity of the OAuth activity
  2. Determine if this is a revocation (possible compromise) or failure (attack attempt)
  3. Review the user and connected app involved
  4. Check for multiple failed attempts from the same source
  5. Investigate the source IP and geolocation
  6. Review API usage patterns for anomalies
  7. If token compromise suspected:
     - Immediately revoke all active OAuth tokens for the affected app/user
     - Reset user credentials
     - Review all API calls made using the compromised token
     - Check for data exfiltration or unauthorized changes
     - Contact the connected app owner if third-party
  8. Consider implementing OAuth token rotation policies and IP restrictions
Reference: https://developer.salesforce.com/docs/atlas.en-us.object_reference.meta/object_reference/sforce_api_objects_oauthtoken.htm
Severity: Medium
DedupPeriodMinutes: 60
LogTypes:
  - Salesforce.RealtimeEvent
RuleID: "Salesforce.OAuth.Credential.Abuse"
Threshold: 1
Tags:
  - Salesforce
  - OAuth
  - Credential Abuse
  - Token Theft
  - API Abuse
Reports:
  MITRE ATT&CK:
    - TA0006:T1528  # Credential Access: Steal Application Access Token
    - TA0006:T1110  # Credential Access: Brute Force
    - TA0005:T1550  # Defense Evasion: Use Alternate Authentication Material
    - TA0010:T1020  # Exfiltration: Automated Exfiltration

def rule(event):
    # Alert on Connected App usage events
    # These track OAuth authorizations and third-party integrations
    return event.get("EVENT_TYPE") in [
        "ConnectedAppUsageEventStore",
        "ApiConnectedApp",
    ]


def title(event):
    # Create descriptive title with app and user details
    app_id = event.get("CONNECTED_APP_ID", "<UNKNOWN_APP>")
    app_name = event.get("CONNECTED_APP_NAME", app_id)
    user = event.get("USER_NAME", event.get("USER_ID", "<UNKNOWN_USER>"))
    connection_type = event.get("CONNECTION_TYPE", "<UNKNOWN_TYPE>")

    return f"Salesforce Connected App Access: {app_name} via {connection_type} - User: {user}"


def severity(event):
    # Map based on connection type and context
    connection_type = str(event.get("CONNECTION_TYPE", "")).lower()
    app_name = str(event.get("CONNECTED_APP_NAME", "")).lower()

    # OAuth refresh token grants are sensitive (persistent access)
    if "refresh" in connection_type:
        return "HIGH"

    # New app authorizations are notable
    if "authorization" in connection_type:
        return "MEDIUM"

    # Unknown or suspicious app names
    # Check for common test/development naming patterns that may indicate
    # non-production or unapproved apps
    suspicious_keywords = ["test", "dev", "temp", "demo", "unknown", "sandbox", "trial"]
    # Only flag if the app name starts with or exactly matches these keywords
    # to reduce false positives from legitimate apps containing these terms
    if any(app_name.startswith(keyword) or app_name == keyword for keyword in suspicious_keywords):
        return "MEDIUM"

    return "DEFAULT"


def dedup(event):
    # Deduplicate by app, user, and connection type
    app_id = event.get("CONNECTED_APP_ID", "unknown")
    user_id = event.get("USER_ID", "unknown")
    connection = event.get("CONNECTION_TYPE", "unknown")
    return f"SF_CONNECTED_APP_{app_id}_{user_id}_{connection}"


def alert_context(event):
    # Provide comprehensive context for investigation
    return {
        "Connected App ID": event.get("CONNECTED_APP_ID"),
        "Connected App Name": event.get("CONNECTED_APP_NAME"),
        "Connection Type": event.get("CONNECTION_TYPE"),
        "User ID": event.get("USER_ID"),
        "Username": event.get("USER_NAME"),
        "Source IP": event.get("SOURCE_IP"),
        "User Type": event.get("USER_TYPE"),
        "Request ID": event.get("REQUEST_ID"),
        "Organization ID": event.get("ORGANIZATION_ID"),
        "API Version": event.get("API_VERSION"),
        "OAuth Scopes": event.get("OAUTH_SCOPES"),
    }

AnalysisType: rule
Description: |
  Monitors third-party integrations and OAuth connected apps accessing Salesforce. Connected apps use OAuth for authorization and can access data on behalf of users, making them a potential vector for:
  - Unauthorized data access
  - Shadow IT applications
  - Compromised OAuth tokens
  - Over-privileged integrations

  This detection triggers on connected app usage events and adjusts severity based on:
  - Connection type (refresh tokens are higher risk)
  - App authorization events
  - Suspicious app naming patterns
DisplayName: "Salesforce Third-Party Integration Monitoring"
Enabled: true
Filename: salesforce_third_party_integration.py
Runbook: |
  1. Identify the connected app and review its purpose and authorization
  2. Verify the app is approved and expected for this user
  3. Review OAuth scopes granted to the app (what permissions it has)
  4. Check if the source IP is expected for this integration
  5. Investigate the user's intent for authorizing this app
  6. Review the app's access history and data accessed
  7. If unauthorized or suspicious:
     - Revoke the OAuth token immediately
     - Review all data accessed by the app
     - Disable the connected app if not needed
     - Educate the user on approved applications
  8. Consider implementing connected app policies and IP restrictions
Reference: https://help.salesforce.com/s/articleView?id=sf.connected_app_overview.htm
Severity: Medium
DedupPeriodMinutes: 60
LogTypes:
  - Salesforce.RealtimeEvent
RuleID: "Salesforce.ThirdParty.Integration.Monitoring"
Threshold: 1
Tags:
  - Salesforce
  - OAuth
  - Connected Apps
  - Third-Party Integration
  - Shadow IT
Reports:
  MITRE ATT&CK:
    - TA0001:T1199  # Initial Access: Trusted Relationship
    - TA0003:T1098  # Persistence: Account Manipulation
    - TA0005:T1550  # Defense Evasion: Use Alternate Authentication Material
    - TA0006:T1528  # Credential Access: Steal Application Access Token