Panther rules: databricks

from panther_databricks_helpers import SYSTEM_USERS, databricks_alert_context


def rule(event):
    # Filter out system users and unknown
    user = event.deep_get("userIdentity", "email", default="")
    if user in SYSTEM_USERS or user in ("", "unknown"):
        return False

    # Must have workspace context
    return event.get("workspaceId") is not None


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


def unique(event):
    workspace = event.get("workspaceId", "unknown")
    return workspace


def title(event):
    user = event.deep_get("userIdentity", "email", default="Unknown User")
    return f"User accessing multiple workspaces: {user} (≥5 workspaces/day)"


def alert_context(event):
    return databricks_alert_context(event)

AnalysisType: rule
Filename: databricks_access_to_multiple_workspaces.py
RuleID: "Databricks.Audit.AccessToMultipleWorkspaces"
DisplayName: "Databricks Access to Multiple Workspaces"
Enabled: true
Status: Experimental
LogTypes:
  - Databricks.Audit
Tags:
  - Databricks
  - Lateral Movement
  - Reconnaissance
Reports:
  MITRE ATT&CK:
    - TA0008:T1021 # Remote Services
Severity: Medium
Threshold: 5
DedupPeriodMinutes: 1440
Description: >
  Detects users accessing 5 or more distinct workspaces within 24 hours, which may indicate
  lateral movement, reconnaissance, or compromised credentials.
Runbook: |
  1. Query audit logs for all workspace access by this user in the past 7 days to establish normal patterns
  2. Check if the user performed unusual actions (data downloads, permission changes) across multiple workspaces
  3. Find all users accessing multiple workspaces in the past 30 days to establish baseline
Reference: https://github.com/andyweaves/system-tables-audit-logs/blob/main/resources/queries_and_alerts.json
SummaryAttributes:
  - actor

from panther_databricks_helpers import databricks_alert_context


def rule(event):
    if event.get("serviceName") != "accounts":
        return False

    return event.get("actionName") == "revokeDbToken"


def title(event):
    actor = event.deep_get("userIdentity", "email", default="Unknown Actor")
    token_id = event.deep_get("requestParams", "tokenId", default="Unknown Token")
    return f"Access token revoked: {token_id} by {actor}"


def dedup(event):
    token_id = event.deep_get("requestParams", "tokenId", default="unknown")
    return f"token_revoked_{token_id}"


def alert_context(event):
    return databricks_alert_context(
        event,
        additional_fields={
            "token_id": event.deep_get("requestParams", "tokenId"),
        },
    )

AnalysisType: rule
Filename: databricks_access_token_revoked.py
RuleID: "Databricks.Audit.AccessTokenRevoked"
DisplayName: "Databricks Access Token Revoked"
Enabled: true
Status: Experimental
LogTypes:
  - Databricks.Audit
Tags:
  - Databricks
  - Defense Evasion
Reports:
  MITRE ATT&CK:
    - TA0005:T1070 # Indicator Removal
Severity: Info
Description: >
  Detects revocation of Databricks access tokens. Token revocation may be routine
  credential rotation or could indicate an attacker covering their tracks after
  using a compromised token.
Runbook: |
  1. Verify the token revocation was intentional and authorized
  2. Check if the token was recently created or used from unusual IPs
  3. Look for preceding suspicious activity associated with this token
Reference: https://github.com/databricks-solutions/cybersec-workspace-detection-app/tree/main/base/detections/behavioral

from panther_databricks_helpers import (
    ADMIN_PRIVILEGE_ACTIONS,
    databricks_alert_context,
    extract_group_identifier,
    extract_target_principal,
    get_principal_type,
    is_admin_privilege_action,
)

REMOVAL_ACTIONS = ["removeAdmin", "removePrincipalFromGroup"]


def rule(event):
    # Must be account-level event
    if event.get("auditLevel") != "ACCOUNT_LEVEL":
        return False

    # Exclude privilege removals — this rule detects grants only
    if event.get("actionName") in REMOVAL_ACTIONS:
        return False

    return is_admin_privilege_action(event)


def severity(event):
    status_code = event.deep_get("response", "statusCode")
    return "HIGH" if status_code == 200 else "MEDIUM"


def title(event):
    action = event.get("actionName", "Unknown Action")
    actor = event.deep_get("userIdentity", "email", default="Unknown Actor")
    target = extract_target_principal(event) or "Unknown Principal"
    status_code = event.deep_get("response", "statusCode")
    status = "Granted" if status_code == 200 else "Attempted to grant"

    # Check if it's direct admin action or group-based
    if action in ADMIN_PRIVILEGE_ACTIONS["direct"]:
        return f"{status} account admin privileges to {target} by {actor}"
    group = extract_group_identifier(event)
    return f"{status} admin group membership ({group}) to {target} by {actor}"


def dedup(event):
    target_principal = extract_target_principal(event) or "unknown"
    return f"account_admin_privilege_assignment_{target_principal}"


def alert_context(event):
    target_principal = extract_target_principal(event)
    principal_type = get_principal_type(target_principal) if target_principal else "Unknown"

    return databricks_alert_context(
        event,
        additional_fields={
            "privilege_scope": "ACCOUNT_LEVEL",
            "target_principal": target_principal,
            "principal_type": principal_type,
            "target_group": extract_group_identifier(event),
        },
    )

AnalysisType: rule
Filename: databricks_account_admin_privileged_role_assignment.py
RuleID: "Databricks.Audit.AccountAdminPrivilegedRoleAssignment"
DisplayName: "Databricks Account Admin Privileged Role Assignment"
Enabled: true
Status: Experimental
LogTypes:
  - Databricks.Audit
Tags:
  - Databricks
  - Privilege Escalation
  - Persistence
Reports:
  MITRE ATT&CK:
    - TA0004:T1098 # Account Manipulation
    - TA0003:T1136 # Create Account
Severity: Medium
Description: >
  Detects when account-level admin privileges are granted in Databricks through direct role
  assignments or administrative group membership. Account admins have extensive control across
  all workspaces and should be carefully monitored. Successful grants are elevated to HIGH severity.
Runbook: |
  1. Query audit logs for all account-level administrative actions by the target principal in the 24 hours after this privilege grant
  2. Check if the target principal accessed multiple workspaces or performed bulk operations in the 6 hours after receiving admin rights
  3. Find all account admin grants in the past 90 days to identify unusual patterns or privilege escalation chains
Reference: https://github.com/databricks-solutions/cybersec-workspace-detection-app/blob/main/base/detections/event-based/account_admin_privileged_role_assignment.py
SummaryAttributes:
  - actor
  - target_principal
  - principal_type

from panther_databricks_helpers import databricks_alert_context, is_config_change


def rule(event):
    # Must be account-level audit event
    if event.get("auditLevel") != "ACCOUNT_LEVEL":
        return False

    # Account settings changes
    if is_config_change(event, config_category="account"):
        return True

    # SSO configuration changes
    if is_config_change(event, config_category="sso"):
        return True

    return False


def title(event):
    action = event.get("actionName", "Unknown Action")
    service = event.get("serviceName", "Unknown Service")
    actor = event.deep_get("userIdentity", "email", default="Unknown Actor")
    status_code = event.deep_get("response", "statusCode")
    status = "Success" if status_code == 200 else "Failed"

    return f"Account-level configuration change ({service}.{action}) by {actor} - {status}"


def dedup(event):
    service = event.get("serviceName", "unknown")
    action = event.get("actionName", "unknown")
    return f"account_config_change_{service}_{action}"


def alert_context(event):
    return databricks_alert_context(
        event,
        additional_fields={
            "change_scope": "ACCOUNT_LEVEL",
            "request_params": event.get("requestParams"),
        },
    )

AnalysisType: rule
Filename: databricks_config_changes_account_level.py
RuleID: "Databricks.Audit.ConfigChangesAccountLevel"
DisplayName: "Databricks Account-Level Configuration Changes"
Enabled: true
Status: Experimental
LogTypes:
  - Databricks.Audit
Tags:
  - Databricks
  - Persistence
Reports:
  MITRE ATT&CK:
    - TA0003:T1098 # Account Manipulation
Severity: Info
Description: >
  Detects configuration changes at the Databricks account level, including account settings,
  metastore configurations, and SSO settings. Account-level changes affect all workspaces and
  should be monitored for unauthorized modifications.
Runbook: |
  1. Query audit logs for all account-level changes by the actor in the 24 hours around this event
  2. Check if this configuration change affected multiple workspaces based on activity in the 6 hours after the change
  3. Find all account-level configuration changes in the past 30 days to identify patterns
Reference: https://github.com/databricks-solutions/cybersec-workspace-detection-app/blob/main/base/detections/event-based/configuration_changes_account_level.py

from panther_databricks_helpers import databricks_alert_context, filter_noise


def rule(event):
    if event.get("serviceName") != "accounts":
        return False

    if event.get("actionName") != "IpAccessDenied":
        return False

    # Filter out system noise using helper
    if filter_noise(event):
        return False

    return True


def title(event):
    user = event.deep_get("userIdentity", "email", default="Unknown User")
    source_ip = event.get("sourceIPAddress", "Unknown IP")
    workspace = event.get("workspaceId", "Unknown Workspace")
    return (
        f"Blocked login attempt from denied IP {source_ip} for user {user} to workspace {workspace}"
    )


def dedup(event):
    source_ip = event.get("sourceIPAddress", "unknown")
    workspace = event.get("workspaceId", "unknown")
    return f"denied_ip_login_{workspace}_{source_ip}"


def alert_context(event):
    return databricks_alert_context(
        event, additional_fields={"path": event.deep_get("requestParams", "path")}
    )

AnalysisType: rule
Filename: databricks_attempted_logon_from_denied_ip.py
RuleID: "Databricks.Audit.AttemptedLogonFromDeniedIP"
DisplayName: "Databricks Attempted Logon From Denied IP"
Enabled: true
Status: Experimental
LogTypes:
  - Databricks.Audit
Tags:
  - Databricks
  - Initial Access
  - Reconnaissance
Reports:
  MITRE ATT&CK:
    - TA0001:T1078 # Valid Accounts
Severity: Info
Description: >
  Detects blocked login attempts from IP addresses explicitly denied by workspace IP access
  control policies. This excludes known service agents and telemetry operations. While these
  attempts were successfully blocked, they may indicate reconnaissance or unauthorized access attempts.
Runbook: |
  1. Count all login attempts from the source IP (sourceIPAddress) in the 1 hour before and after this blocked attempt
  2. Check if the source IP is associated with known VPN services, cloud providers, or threat intelligence feeds
  3. Find all successful and failed login attempts for this user in the 24 hours around the alert to identify credential stuffing patterns
Reference: https://github.com/databricks-solutions/cybersec-workspace-detection-app/blob/main/base/detections/event-based/attempted_logon_from_denied_ip.py

from panther_databricks_helpers import DOWNLOAD_ACTIONS, databricks_alert_context


def rule(event):
    action = event.get("actionName")
    if action not in DOWNLOAD_ACTIONS:
        return False

    # Exclude source exports
    if action == "workspaceExport":
        export_format = event.deep_get("requestParams", "workspaceExportFormat")
        if export_format == "SOURCE":
            return False

    # Exclude arrows format
    if action == "downloadQueryResult":
        file_type = event.deep_get("requestParams", "fileType")
        if file_type == "arrows":
            return False

    return True


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


def title(event):
    user = event.deep_get("userIdentity", "email", default="Unknown User")
    action = event.get("actionName", "download")
    return f"High volume data downloads by {user} ({action})"


def alert_context(event):
    return databricks_alert_context(event)

AnalysisType: rule
Filename: databricks_data_downloads_from_control_plane.py
RuleID: "Databricks.Audit.DataDownloadsFromControlPlane"
DisplayName: "Databricks Data Downloads From Control Plane"
Enabled: true
Status: Experimental
LogTypes:
  - Databricks.Audit
Tags:
  - Databricks
  - Exfiltration
Reports:
  MITRE ATT&CK:
    - TA0010:T1567 # Exfiltration Over Web Service
Severity: Medium
Threshold: 21
DedupPeriodMinutes: 60
Description: >
  Detects high volume data downloads from the control plane which may indicate
  data exfiltration. Monitors download actions including query results, notebooks, and models.
Runbook: |
  1. Query audit logs for all download actions by this user in the past 24 hours to calculate total volume
  2. Check if the downloaded data contains sensitive classifications or PII in the 6 hours around this alert
  3. Find all users with high download rates in the past 30 days to establish baseline
Reference: https://github.com/andyweaves/system-tables-audit-logs/blob/main/resources/queries_and_alerts.json

SummaryAttributes:
  - actor

from panther_databricks_helpers import DATA_MOVEMENT_CREDENTIAL_ACTIONS, databricks_alert_context


def rule(event):
    action = event.get("actionName")
    # mount is covered by databricks_mount_point_creation; skip it here
    return action in DATA_MOVEMENT_CREDENTIAL_ACTIONS and action != "mount"


def title(event):
    actor = event.deep_get("userIdentity", "email", default="Unknown Actor")
    action = event.get("actionName", "unknown")
    workspace = event.get("workspaceId", "Unknown Workspace")
    return f"Data movement credential operation: {action} in workspace {workspace} by {actor}"


def dedup(event):
    actor = event.deep_get("userIdentity", "email", default="unknown")
    action = event.get("actionName", "unknown")
    return f"data_movement_cred_{actor}_{action}"


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

AnalysisType: rule
Filename: databricks_data_movement_explicit_credentials.py
RuleID: "Databricks.Audit.DataMovementExplicitCredentials"
DisplayName: "Databricks Data Movement with Explicit Credentials"
Enabled: true
Status: Experimental
LogTypes:
  - Databricks.Audit
Tags:
  - Databricks
  - Exfiltration
Reports:
  MITRE ATT&CK:
    - TA0010:T1537 # Transfer Data to Cloud Account
Severity: Info
Description: >
  Detects creation or modification of storage credentials, connections, and external
  locations that could facilitate data exfiltration. These operations establish direct
  paths to external storage and may indicate data movement preparation. Mount point
  creation is covered separately by Databricks.Audit.MountPointCreation.
Runbook: |
  1. Verify the credential/connection creation was part of an approved workflow
  2. Check the target storage location for sensitivity
  3. Monitor for subsequent data access through the new credential
Reference: https://github.com/databricks-solutions/cybersec-workspace-detection-app/tree/main/base/detections/behavioral

from panther_databricks_helpers import databricks_alert_context

# Keywords that indicate IP-based access denials
IP_DENIAL_KEYWORDS = ["address", "network", "cidr", "allowlist", "blocklist"]


def rule(event):
    if event.get("serviceName") != "deltaSharingAccess":
        return False

    status_code = event.deep_get("response", "statusCode")
    if status_code not in [403, 401]:
        return False

    error_message = event.deep_get("response", "errorMessage", default="").lower()
    return any(keyword in error_message for keyword in IP_DENIAL_KEYWORDS)


def title(event):
    recipient = event.deep_get("requestParams", "recipientName", default="Unknown Recipient")
    source_ip = event.get("sourceIPAddress", "Unknown IP")
    return f"Delta Sharing access blocked from {source_ip} for recipient {recipient}"


def alert_context(event):
    return databricks_alert_context(event)

AnalysisType: rule
Filename: databricks_delta_sharing_ip_access_failures.py
RuleID: "Databricks.Audit.DeltaSharingIPAccessFailures"
DisplayName: "Databricks Delta Sharing IP Access Failures"
Enabled: true
Status: Experimental
LogTypes:
  - Databricks.Audit
Tags:
  - Databricks
  - Delta Sharing
  - Initial Access
Reports:
  MITRE ATT&CK:
    - TA0001:T1078 # Valid Accounts
Severity: Medium
Description: >
  Detects blocked Delta Sharing access attempts due to IP access list restrictions,
  which may indicate unauthorized access attempts from unexpected locations.
Runbook: |
  1. Query audit logs for all Delta Sharing access attempts from this IP in the past 24 hours
  2. Check if the source IP is associated with known threats or unexpected geographic locations
  3. Find all Delta Sharing access failures in the past 7 days to identify patterns
Reference: https://github.com/andyweaves/system-tables-audit-logs/blob/main/resources/queries_and_alerts.json

from panther_databricks_helpers import databricks_alert_context


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

    # Check if IP access list is configured
    # Alert if IP ACL is missing, empty string, or empty list (all falsy)
    ip_access_list = event.deep_get("requestParams", "ipAccessList")
    return not ip_access_list


def title(event):
    actor = event.deep_get("userIdentity", "email", default="Unknown Actor")
    recipient = event.deep_get("requestParams", "name", default="Unknown Recipient")
    return f"Delta Sharing recipient created without IP ACLs: {recipient} by {actor}"


def alert_context(event):
    return databricks_alert_context(
        event,
        additional_fields={"recipient_name": event.deep_get("requestParams", "name")},
    )

AnalysisType: rule
Filename: databricks_delta_sharing_recipient_without_ip_acls.py
RuleID: "Databricks.Audit.DeltaSharingRecipientWithoutIPACLs"
DisplayName: "Databricks Delta Sharing Recipient Without IP ACLs"
Enabled: true
Status: Experimental
LogTypes:
  - Databricks.Audit
Tags:
  - Databricks
  - Delta Sharing
  - Defense Evasion
Reports:
  MITRE ATT&CK:
    - TA0005:T1562 # Impair Defenses
Severity: Medium
Description: >
  Detects creation of Delta Sharing recipients without IP access list restrictions,
  which could allow unauthorized data access from any location.
Runbook: |
  1. Query audit logs for all Delta Sharing recipient creations in the past 30 days
  2. Check if this recipient has accessed shared data in the 24 hours after creation
  3. Find all recipients without IP ACLs to establish security posture
Reference: https://github.com/andyweaves/system-tables-audit-logs/blob/main/resources/queries_and_alerts.json

from panther_databricks_helpers import SYSTEM_USERS, databricks_alert_context

# Destructive action prefixes/names to match
DESTRUCTIVE_PREFIXES = ["delete", "drop", "trash", "destroy", "purge"]


def rule(event):
    action = event.get("actionName", "").lower()

    # Exclude system users
    user = event.deep_get("userIdentity", "email", default="")
    if user in SYSTEM_USERS:
        return False

    # Exclude non-destructive actions that contain "delete" as substring
    if action.startswith("undelete") or action.startswith("restore"):
        return False

    # Check for destructive action prefixes
    return any(action.startswith(prefix) for prefix in DESTRUCTIVE_PREFIXES)


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


def title(event):
    user = event.deep_get("userIdentity", "email", default="Unknown User")
    action = event.get("actionName", "delete")
    return f"High volume destructive activities by {user} (>50/day, action: {action})"


def alert_context(event):
    return databricks_alert_context(event)

AnalysisType: rule
Filename: databricks_destructive_activities.py
RuleID: "Databricks.Audit.DestructiveActivities"
DisplayName: "Databricks Destructive Activities"
Enabled: true
Status: Experimental
LogTypes:
  - Databricks.Audit
Tags:
  - Databricks
  - Impact
Reports:
  MITRE ATT&CK:
    - TA0040:T1485 # Data Destruction
Severity: Medium
Threshold: 50
DedupPeriodMinutes: 1440
Description: >
  Detects high volume destructive activities by a single user which may indicate
  malicious data destruction, ransomware, or insider threats.
Runbook: |
  1. Query audit logs for all delete actions by this user in the past 7 days to identify patterns
  2. Check if deleted resources can be recovered or if backups exist
  3. Find all users with high deletion rates in the past 30 days to establish baseline
Reference: https://github.com/andyweaves/system-tables-audit-logs/blob/main/resources/queries_and_alerts.json
SummaryAttributes:
  - actor

from panther_databricks_helpers import (
    databricks_alert_context,
    is_databricks_employee_auth,
    is_login_action,
)


def rule(event):
    if event.get("serviceName") != "accounts":
        return False

    if not is_login_action(event):
        return False

    if not is_databricks_employee_auth(event):
        return False

    # Check for successful response
    status_code = event.deep_get("response", "statusCode")
    if status_code != 200:
        return False

    # Check for workspace-level audit
    if event.get("auditLevel") != "WORKSPACE_LEVEL":
        return False

    return True


def title(event):
    user = event.deep_get("userIdentity", "email", default="Unknown User")
    workspace = event.get("workspaceId", "Unknown Workspace")
    return f"Databricks employee logged into workspace {workspace} as {user}"


def alert_context(event):
    return databricks_alert_context(
        event,
        additional_fields={"auth_method": event.deep_get("requestParams", "authentication_method")},
    )

AnalysisType: rule
Filename: databricks_employee_logon.py
RuleID: "Databricks.Audit.EmployeeLogon"
DisplayName: "Databricks Employee Logon"
Enabled: true
Status: Experimental
LogTypes:
  - Databricks.Audit
Tags:
  - Databricks
  - Initial Access
Reports:
  MITRE ATT&CK:
    - TA0001:T1078 # Valid Accounts
Severity: Info
Description: >
  Detects when a Databricks employee successfully logs into a workspace using GENIE_AUTH authentication.
  This is typically for legitimate support purposes but should be tracked for awareness.
Runbook: |
  1. Query Databricks audit logs for all actions by this Databricks employee (userIdentity.email) in the 24 hours before and after the alert
  2. Check if there are open support cases or authorized maintenance windows that would explain this employee access
  3. Find all other Databricks employee logins to this workspace in the past 30 days to identify patterns
Reference: https://github.com/databricks-solutions/cybersec-workspace-detection-app/blob/main/base/detections/event-based/databricks_employee_logon.py

from panther_databricks_helpers import databricks_alert_context


def rule(event):
    return event.get("serviceName") == "globalInitScripts"


def title(event):
    action = event.get("actionName", "Unknown Action")
    actor = event.deep_get("userIdentity", "email", default="Unknown Actor")
    script_name = event.deep_get("requestParams", "name", default="Unknown Script")
    return f"Global init script {action}: {script_name} by {actor}"


def dedup(event):
    script_name = event.deep_get("requestParams", "name", default="unknown")
    script_id = event.deep_get("requestParams", "script_id", default="unknown")
    return f"global_init_script_{script_id}_{script_name}"


def alert_context(event):
    return databricks_alert_context(
        event,
        additional_fields={
            "script_name": event.deep_get("requestParams", "name"),
            "script_id": event.deep_get("requestParams", "script_id"),
            "script_enabled": event.deep_get("requestParams", "enabled"),
            "script_sha256": event.deep_get("requestParams", "script-SHA256"),
        },
    )

AnalysisType: rule
Filename: databricks_global_init_script_changes.py
RuleID: "Databricks.Audit.GlobalInitScriptChanges"
DisplayName: "Databricks Global Init Script Changes"
Enabled: true
Status: Experimental
LogTypes:
  - Databricks.Audit
Tags:
  - Databricks
  - Persistence
  - Execution
Reports:
  MITRE ATT&CK:
    - TA0003:T1037 # Boot or Logon Initialization Scripts
    - TA0002:T1059 # Command and Scripting Interpreter
Severity: Info
Description: >
  Detects modifications to global initialization scripts which run on all clusters at startup.
  These scripts can be used for persistence or to execute malicious code across the environment.
  All script creations, updates, and deletions are monitored.
Runbook: |
  1. Query audit logs for all global init script changes by this actor in the past 30 days
  2. Check if new clusters were created shortly after the script modification in the 6 hours after this change
  3. Find all script modifications across all workspaces in the past 7 days to identify coordinated changes
Reference: https://github.com/andyweaves/system-tables-audit-logs/blob/main/resources/queries_and_alerts.json
SummaryAttributes:
  - actor
  - script_name
  - action

from panther_databricks_helpers import databricks_alert_context, extract_group_identifier


def rule(event):
    if event.get("serviceName") != "accounts":
        return False

    return event.get("actionName") == "createGroup"


def title(event):
    actor = event.deep_get("userIdentity", "email", default="Unknown Actor")
    group = extract_group_identifier(event) or event.deep_get(
        "requestParams", "groupName", default="Unknown Group"
    )
    return f"Group created: {group} by {actor}"


def dedup(event):
    group = extract_group_identifier(event) or "unknown"
    return f"group_created_{group}"


def alert_context(event):
    return databricks_alert_context(
        event,
        additional_fields={
            "group_name": extract_group_identifier(event),
        },
    )

AnalysisType: rule
Filename: databricks_group_created.py
RuleID: "Databricks.Audit.GroupCreated"
DisplayName: "Databricks Group Created"
Enabled: true
Status: Experimental
LogTypes:
  - Databricks.Audit
Tags:
  - Databricks
  - Persistence
Reports:
  MITRE ATT&CK:
    - TA0003:T1136 # Create Account
Severity: Info
Description: >
  Detects creation of user groups in Databricks. Group creation may be part of normal
  administration or could indicate privilege escalation preparation by creating a group
  that will later receive elevated permissions.
Runbook: |
  1. Verify the group creation was part of an approved workflow
  2. Check if the group was subsequently granted admin or elevated permissions
  3. Review group membership additions following creation
Reference: https://github.com/databricks-solutions/cybersec-workspace-detection-app/tree/main/base/detections/behavioral

from panther_databricks_helpers import (
    databricks_alert_context,
    extract_group_identifier,
    should_alert_on_group_change,
)


def rule(event):
    return should_alert_on_group_change(event, change_type="delete")


def severity(event):
    status_code = event.deep_get("response", "statusCode")
    return "HIGH" if status_code == 200 else "LOW"


def title(event):
    group_id = extract_group_identifier(event) or "Unknown Group"
    actor = event.deep_get("userIdentity", "email", default="Unknown Actor")
    status_code = event.deep_get("response", "statusCode")
    status = "Success" if status_code == 200 else "Failed"
    return f"Group deletion {status}: {group_id} by {actor}"


def dedup(event):
    group_id = extract_group_identifier(event) or "unknown"
    return f"group_deleted_{group_id}"


def alert_context(event):
    group_id = extract_group_identifier(event)
    return databricks_alert_context(event, additional_fields={"group_id": group_id})

AnalysisType: rule
Filename: databricks_group_deleted.py
RuleID: "Databricks.Audit.GroupDeleted"
DisplayName: "Databricks Group Deleted"
Enabled: true
Status: Experimental
LogTypes:
  - Databricks.Audit
Tags:
  - Databricks
  - Impact
Reports:
  MITRE ATT&CK:
    - TA0040:T1531 # Account Access Removal
Severity: Low
Description: >
  Detects group deletions in Databricks accounts. While often part of normal cleanup processes,
  unauthorized group deletions could indicate access control dismantling. Successful deletions
  are elevated to HIGH severity.
Runbook: |
  1. Query audit logs for all members who were in this group in the 24 hours before deletion
  2. Check if this group had admin privileges or access to sensitive resources based on past 30 days of activity
  3. Find all group deletions by this actor in the past 30 days to identify bulk deletion patterns
Reference: https://github.com/databricks-solutions/cybersec-workspace-detection-app/blob/main/base/detections/event-based/group_deleted.py

from panther_databricks_helpers import (
    databricks_alert_context,
    get_config_key_value,
    is_critical_config_change,
)


def rule(event):
    if not is_critical_config_change(event):
        return False

    # Verbose audit logging disabled is handled by a dedicated rule
    config_key, config_value = get_config_key_value(event)
    if config_key == "enableVerboseAuditLogs" and config_value == "false":
        return False

    return True


def severity(event):
    status_code = event.deep_get("response", "statusCode")
    success = status_code == 200
    action = event.get("actionName", "Unknown Action")

    # Determine severity from action type
    #                        Success    Failure
    # IP access list deleted HIGH       MEDIUM
    # Other critical configs MEDIUM     LOW
    if action == "deleteIpAccessList":
        return "HIGH" if success else "MEDIUM"
    return "MEDIUM" if success else "LOW"


def title(event):
    action = event.get("actionName", "Unknown Action")
    actor = event.deep_get("userIdentity", "email", default="Unknown Actor")
    status_code = event.deep_get("response", "statusCode")
    status = "Successfully" if status_code == 200 else "Attempted to"

    # IP access list changes
    if "IpAccessList" in action:
        return f"{status} {action} by {actor}"

    # Workspace configuration edits
    config_key, config_value = get_config_key_value(event)
    if config_key:
        return f"{status} modify {config_key} to {config_value} by {actor}"

    return f"Critical configuration change by {actor}"


def dedup(event):
    config_key, _ = get_config_key_value(event)
    # IP access list events don't have workspaceConfKeys, so fall back to actionName
    key = config_key or event.get("actionName", "unknown")
    return f"critical_config_change_{key}"


def alert_context(event):
    config_key, config_value = get_config_key_value(event)
    return databricks_alert_context(
        event,
        additional_fields={
            "config_key": config_key,
            "config_value": config_value,
            "change_category": (
                "IP Access List"
                if "IpAccessList" in event.get("actionName", "")
                else "Workspace Configuration"
            ),
        },
    )

AnalysisType: rule
Filename: databricks_config_changes_high_priority.py
RuleID: "Databricks.Audit.ConfigChangesHighPriority"
DisplayName: "Databricks High Priority Configuration Changes"
Enabled: true
Status: Experimental
LogTypes:
  - Databricks.Audit
Tags:
  - Databricks
  - Defense Evasion
  - Persistence
Reports:
  MITRE ATT&CK:
    - TA0005:T1562.008 # Impair Defenses: Disable Cloud Logs
    - TA0003:T1098 # Account Manipulation
Severity: Medium
Description: >
  Detects high-priority security configuration changes including audit logging modifications,
  IP access list changes, and security-critical workspace settings. Severity is elevated for
  successful changes to high-risk settings.
Runbook: |
  1. Query audit logs for all configuration changes by the actor in the 24 hours before and after this critical change
  2. Check if this configuration change has been performed by this actor in the past 90 days to establish legitimacy
  3. Find all other high-risk configuration changes (IP access lists, audit settings) across all workspaces in the past 7 days
Reference: https://github.com/databricks-solutions/cybersec-workspace-detection-app/blob/main/base/detections/event-based/configuration_changes_high_priority.py
SummaryAttributes:
  - actor
  - config_key
  - change_category

from panther_databricks_helpers import databricks_alert_context


def rule(event):
    return event.get("actionName") == "installLibraryOnAllClusters"


def title(event):
    actor = event.deep_get("userIdentity", "email", default="Unknown Actor")
    workspace = event.get("workspaceId", "Unknown Workspace")
    return f"Library installed on all clusters in workspace {workspace} by {actor}"


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

AnalysisType: rule
Filename: databricks_install_library_all_clusters.py
RuleID: "Databricks.Audit.InstallLibraryAllClusters"
DisplayName: "Databricks Install Library on All Clusters"
Enabled: true
Status: Experimental
LogTypes:
  - Databricks.Audit
Tags:
  - Databricks
  - Execution
  - Persistence
Reports:
  MITRE ATT&CK:
    - TA0002:T1203 # Exploitation for Client Execution
    - TA0003:T1543 # Create or Modify System Process
Severity: Medium
Description: >
  Detects use of the deprecated installLibraryOnAllClusters action. This anti-pattern can
  introduce security risks by installing potentially malicious libraries across the entire
  environment without proper review or controls.
Runbook: |
  1. Query audit logs for the library installation details and identify what library was installed
  2. Check if this library has been used in notebook or job execution in the 24 hours after installation
  3. Find all library installations by this user in the past 30 days to identify patterns
Reference: https://github.com/andyweaves/system-tables-audit-logs/blob/main/resources/queries_and_alerts.json
SummaryAttributes:
  - actor
  - workspace_id

from panther_databricks_helpers import databricks_alert_context


def _token_duration_hours(event):
    """Calculate token lifetime in hours, or None if not determinable."""
    try:
        token_expiration = int(event.deep_get("requestParams", "tokenExpirationTime", default=0))
        event_time_ms = event.get("timestamp", 0)
        if not token_expiration or not event_time_ms:
            return None
        return (token_expiration - event_time_ms) / (1000 * 3600)
    except (ValueError, TypeError):
        return None


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

    duration = _token_duration_hours(event)
    return duration is not None and duration > 72


def severity(event):
    duration = _token_duration_hours(event)
    if duration is None:
        return "LOW"

    #                    Severity
    # >1 year (8760h)   HIGH
    # >90 days (2160h)  MEDIUM
    # otherwise         LOW
    if duration > 8760:
        return "HIGH"
    if duration > 2160:
        return "MEDIUM"
    return "LOW"


def title(event):
    actor = event.deep_get("userIdentity", "email", default="Unknown User")
    duration = _token_duration_hours(event)
    if duration is not None:
        return f"Long-lifetime token ({int(duration / 24)} days) generated by {actor}"
    return f"Long-lifetime token generated by {actor}"


def alert_context(event):
    duration = _token_duration_hours(event)
    duration_days = duration / 24 if duration is not None else None

    return databricks_alert_context(
        event,
        additional_fields={
            "token_expiration_time": event.deep_get("requestParams", "tokenExpirationTime"),
            "token_hash": event.deep_get("requestParams", "tokenHash"),
            "token_duration_hours": duration,
            "token_duration_days": duration_days,
        },
    )

AnalysisType: rule
Filename: databricks_long_lifetime_token_generated.py
RuleID: "Databricks.Audit.LongLifetimeTokenGenerated"
DisplayName: "Databricks Long-Lifetime Token Generated"
Enabled: true
Status: Experimental
LogTypes:
  - Databricks.Audit
Tags:
  - Databricks
  - Credential Access
  - Persistence
Reports:
  MITRE ATT&CK:
    - TA0006:T1550 # Use Alternate Authentication Material
    - TA0003:T1098 # Account Manipulation
Severity: Low
Description: >
  Detects generation of personal access tokens (PATs) with lifetime exceeding 72 hours.
  Long-lived tokens increase the risk of credential theft and unauthorized access if compromised.
  Tokens with lifetime >90 days are elevated to MEDIUM, >1 year to HIGH severity.
Runbook: |
  1. Query audit logs for all token generation by this user in the past 30 days to identify patterns
  2. Check if the generated token has been used for API calls in the 24 hours after creation
  3. Find all other long-lifetime tokens (>72 hours) created in the past 90 days to establish baseline
Reference: https://github.com/andyweaves/system-tables-audit-logs/blob/main/resources/queries_and_alerts.json
SummaryAttributes:
  - actor
  - token_duration_days

from panther_databricks_helpers import databricks_alert_context, is_metastore_admin_action


def rule(event):
    # Use helper to check for metastore admin actions
    if not is_metastore_admin_action(event):
        return False

    action = event.get("actionName", "")

    # For group membership changes, only alert on additions (not removals)
    # and ensure it's at the account level
    if action != "updateMetastore":
        # Exclude removals
        if action == "removePrincipalFromGroup":
            return False
        # Must be account-level event
        if event.get("serviceName") != "accounts":
            return False

    return True


def title(event):
    action = event.get("actionName", "Unknown Action")
    actor = event.deep_get("userIdentity", "email", default="Unknown Actor")

    if action == "updateMetastore":
        new_owner = event.deep_get("requestParams", "owner", default="Unknown Owner")
        return f"Metastore ownership changed to {new_owner} by {actor}"
    target_group = event.deep_get("requestParams", "targetGroupName", default="Unknown Group")
    principal = event.deep_get("requestParams", "principal") or event.deep_get(
        "requestParams", "targetUserName", default="Unknown Principal"
    )
    return f"Principal {principal} added to metastore admin group {target_group} by {actor}"


def dedup(event):
    action = event.get("actionName", "unknown")
    if action == "updateMetastore":
        metastore_id = event.deep_get("requestParams", "metastoreId", default="unknown")
        return f"metastore_admin_{metastore_id}"
    principal = event.deep_get("requestParams", "principal") or event.deep_get(
        "requestParams", "targetUserName", default="unknown"
    )
    return f"metastore_admin_group_{principal}"


def alert_context(event):
    return databricks_alert_context(
        event,
        additional_fields={
            "new_owner": event.deep_get("requestParams", "owner"),
            "target_group": event.deep_get("requestParams", "targetGroupName"),
            "principal": event.deep_get("requestParams", "principal")
            or event.deep_get("requestParams", "targetUserName"),
        },
    )

AnalysisType: rule
Filename: databricks_metastore_admin_privilege_granted.py
RuleID: "Databricks.Audit.MetastoreAdminPrivilegeGranted"
DisplayName: "Databricks Metastore Admin Privilege Granted"
Enabled: true
Status: Experimental
LogTypes:
  - Databricks.Audit
Tags:
  - Databricks
  - Privilege Escalation
  - Unity Catalog
Reports:
  MITRE ATT&CK:
    - TA0004:T1098 # Account Manipulation
Severity: Medium
Description: >
  Detects when metastore admin privileges are granted in Databricks through direct metastore
  ownership changes or addition to metastore admin groups. Metastore admins have extensive
  control over data access and governance policies in Unity Catalog.
Runbook: |
  1. Query Unity Catalog audit logs for all metastore operations by the target principal in the 24 hours after this privilege grant
  2. Check if the target principal accessed sensitive catalogs or tables in the 6 hours after receiving admin rights
  3. Find all metastore admin grants in the past 90 days to identify unusual patterns
Reference: https://github.com/databricks-solutions/cybersec-workspace-detection-app/blob/main/base/detections/event-based/metastore_admin_privilege_granted.py

from panther_databricks_helpers import MFA_ACTIONS, databricks_alert_context


def rule(event):
    if event.get("serviceName") != "accounts":
        return False

    action = event.get("actionName")
    return action in MFA_ACTIONS["add"] or action in MFA_ACTIONS["delete"]


def title(event):
    action = event.get("actionName", "unknown")
    actor = event.deep_get("userIdentity", "email", default="Unknown Actor")
    change_type = "added" if action in MFA_ACTIONS["add"] else "deleted"
    return f"MFA key {change_type} by {actor}"


def dedup(event):
    actor = event.deep_get("userIdentity", "email", default="unknown")
    action = event.get("actionName", "unknown")
    return f"mfa_key_change_{actor}_{action}"


def alert_context(event):
    return databricks_alert_context(
        event,
        additional_fields={
            "change_type": "add" if event.get("actionName") in MFA_ACTIONS["add"] else "delete",
        },
    )

AnalysisType: rule
Filename: databricks_mfa_key_change.py
RuleID: "Databricks.Audit.MFAKeyChange"
DisplayName: "Databricks MFA Key Change"
Enabled: true
Status: Experimental
LogTypes:
  - Databricks.Audit
Tags:
  - Databricks
  - Credential Access
Reports:
  MITRE ATT&CK:
    - TA0006:T1556 # Modify Authentication Process
Severity: Info
Description: >
  Detects addition or deletion of MFA keys on Databricks accounts. MFA key deletion
  may indicate an attacker weakening account security, while unexpected additions may
  indicate enrollment of attacker-controlled authenticators.
Runbook: |
  1. Verify the actor intended to modify their MFA settings
  2. For deletions, check if a replacement key was added
  3. For additions, verify the new key was enrolled by the legitimate user
Reference: https://github.com/databricks-solutions/cybersec-workspace-detection-app/tree/main/base/detections/behavioral

from panther_databricks_helpers import databricks_alert_context


def rule(event):
    return event.get("actionName") == "mount"


def title(event):
    actor = event.deep_get("userIdentity", "email", default="Unknown Actor")
    mount_point = event.deep_get("requestParams", "mountPoint", default="Unknown Mount")
    workspace = event.get("workspaceId", "Unknown Workspace")
    return f"Mount point created: {mount_point} in workspace {workspace} by {actor}"


def dedup(event):
    mount_point = event.deep_get("requestParams", "mountPoint", default="unknown")
    workspace = event.get("workspaceId", "unknown")
    return f"mount_point_{workspace}_{mount_point}"


def alert_context(event):
    return databricks_alert_context(
        event,
        additional_fields={
            "mount_point": event.deep_get("requestParams", "mountPoint"),
            "mount_config": event.get("requestParams"),
        },
    )