Databricks User Account Created

AnalysisType: rule
Filename: databricks_user_account_created.py
RuleID: "Databricks.Audit.UserAccountCreated"
DisplayName: "Databricks User Account 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 new user accounts in Databricks. Account creation may be part of
  normal onboarding or could indicate an attacker establishing persistence.
Runbook: |
  1. Verify the account creation was part of an approved onboarding workflow
  2. Check if the new account was immediately granted elevated privileges
  3. Review the actor who created the account
Reference: https://github.com/databricks-solutions/cybersec-workspace-detection-app/tree/main/base/detections/behavioral
Tests:
  - Name: User Created via createUser
    ExpectedResult: true
    Log:
      serviceName: "accounts"
      actionName: "createUser"
      userIdentity:
        email: "admin@example.com"
      requestParams:
        targetUserName: "newuser@example.com"
      response:
        statusCode: 200


  - Name: Different Service - Should Not Alert
    ExpectedResult: false
    Log:
      serviceName: "workspace"
      actionName: "createUser"
      userIdentity:
        email: "admin@example.com"

  - Name: Different Action - Should Not Alert
    ExpectedResult: false
    Log:
      serviceName: "accounts"
      actionName: "deleteUser"
      userIdentity:
        email: "admin@example.com"


# ------ paired body: databricks_user_account_created.py ------

from panther_databricks_helpers import databricks_alert_context


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

    action = event.get("actionName")
    return action in ("createUser", "addUser")


def title(event):
    actor = event.deep_get("userIdentity", "email", default="Unknown Actor")
    target = event.deep_get("requestParams", "targetUserName", default="Unknown User")
    endpoint = event.deep_get("requestParams", "endpoint", default="")
    source = f" via {endpoint}" if endpoint else ""
    return f"User account created: {target} by {actor}{source}"


def dedup(event):
    target = event.deep_get("requestParams", "targetUserName", default="unknown")
    return f"user_created_{target}"


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