Databricks Repeated Access to Secrets

AnalysisType: rule
Filename: databricks_repeated_access_to_secrets.py
RuleID: "Databricks.Audit.RepeatedAccessToSecrets"
DisplayName: "Databricks Repeated Access to Secrets"
Enabled: true
Status: Experimental
LogTypes:
  - Databricks.Audit
Tags:
  - Databricks
  - Credential Access
  - Collection
Reports:
  MITRE ATT&CK:
    - TA0006:T1555 # Credentials from Password Stores
Severity: Medium
Threshold: 10
DedupPeriodMinutes: 60
Description: >
  Detects repeated secret access (≥10 times in 60 minutes) which may indicate credential
  harvesting or unauthorized secret enumeration.
Runbook: |
  1. Query audit logs for all secret access by this user in the past 24 hours to identify patterns
  2. Check if the accessed secrets were used in API calls or notebook execution in the 6 hours after access
  3. Find all users with high secret access rates in the past 7 days to establish baseline
Reference: https://github.com/andyweaves/system-tables-audit-logs/blob/main/resources/queries_and_alerts.json
SummaryAttributes:
  - actor
  - secret_scope
Tests:
  - Name: Secret Access
    ExpectedResult: true
    Log:
      timestamp: 1704067200000
      serviceName: "secrets"
      actionName: "getSecret"
      userIdentity:
        email: "user@example.com"
      requestParams:
        scope: "production-keys"
        key: "api-token"
      response:
        statusCode: 200
  - Name: System User Access
    ExpectedResult: false
    Log:
      timestamp: 1704067200000
      serviceName: "secrets"
      actionName: "getSecret"
      userIdentity:
        email: "System-User"
      requestParams:
        scope: "internal"
        key: "config"


# ------ paired body: databricks_repeated_access_to_secrets.py ------

from panther_databricks_helpers import SYSTEM_USERS, databricks_alert_context


def rule(event):
    if event.get("actionName") != "getSecret":
        return False

    # Filter out system users
    user = event.deep_get("userIdentity", "email", default="")
    return user not in SYSTEM_USERS


def dedup(event):
    user = event.deep_get("userIdentity", "email", default="unknown")
    return f"secret_access_{user}"


def title(event):
    user = event.deep_get("userIdentity", "email", default="Unknown User")
    scope = event.deep_get("requestParams", "scope", default="Unknown Scope")
    return f"Repeated secret access by {user} in scope {scope}"


def alert_context(event):
    return databricks_alert_context(
        event,
        additional_fields={
            "secret_scope": event.deep_get("requestParams", "scope"),
            "secret_key": event.deep_get("requestParams", "key"),
        },
    )