Databricks Repeated Unauthorized Unity Catalog Requests

AnalysisType: rule
Filename: databricks_repeated_unauthorized_uc_requests.py
RuleID: "Databricks.Audit.RepeatedUnauthorizedUCRequests"
DisplayName: "Databricks Repeated Unauthorized Unity Catalog Requests"
Enabled: true
Status: Experimental
LogTypes:
  - Databricks.Audit
Tags:
  - Databricks
  - Discovery
  - Unity Catalog
Reports:
  MITRE ATT&CK:
    - TA0007:T1087 # Account Discovery
Severity: Medium
Threshold: 26
DedupPeriodMinutes: 60
Description: >
  Detects repeated unauthorized Unity Catalog API requests (>25 per hour) which may indicate
  reconnaissance, privilege enumeration, or unauthorized data access attempts.
Runbook: |
  1. Query Unity Catalog audit logs for all unauthorized attempts by this user in the past 24 hours
  2. Check if the user attempted to access specific catalogs, schemas, or tables repeatedly
  3. Find all users with high unauthorized request rates in the past 7 days
Reference: https://github.com/andyweaves/system-tables-audit-logs/blob/main/resources/queries_and_alerts.json
SummaryAttributes:
  - actor
  - uc_action
Tests:
  - Name: Unauthorized UC Request - 403
    ExpectedResult: true
    Log:
      timestamp: 1704067200000
      serviceName: "unityCatalog"
      actionName: "getCatalog"
      userIdentity:
        email: "user@example.com"
      response:
        statusCode: 403
        errorMessage: "Access denied"
  - Name: Unauthorized UC Request - 401
    ExpectedResult: true
    Log:
      timestamp: 1704067200000
      serviceName: "unityCatalog"
      actionName: "getSchema"
      userIdentity:
        email: "user@example.com"
      response:
        statusCode: 401
  - Name: Successful UC Request
    ExpectedResult: false
    Log:
      timestamp: 1704067200000
      serviceName: "unityCatalog"
      actionName: "getCatalog"
      userIdentity:
        email: "user@example.com"
      response:
        statusCode: 200
  - Name: Wrong Service
    ExpectedResult: false
    Log:
      timestamp: 1704067200000
      serviceName: "workspace"
      actionName: "getCatalog"
      userIdentity:
        email: "user@example.com"
      response:
        statusCode: 403


# ------ paired body: databricks_repeated_unauthorized_uc_requests.py ------

from panther_databricks_helpers import databricks_alert_context


def rule(event):
    # Must be Unity Catalog service
    if event.get("serviceName") != "unityCatalog":
        return False

    # Check for unauthorized status codes
    status_code = event.deep_get("response", "statusCode")
    return status_code in [401, 403]


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


def title(event):
    user = event.deep_get("userIdentity", "email", default="Unknown User")
    action = event.get("actionName", "Unknown Action")
    return f"Repeated unauthorized Unity Catalog requests by {user} ({action})"


def alert_context(event):
    return databricks_alert_context(
        event,
        additional_fields={"uc_action": event.get("actionName")},
    )