Snowflake User Created

AnalysisType: rule
Filename: snowflake_stream_user_created.py
RuleID: Snowflake.Stream.UserCreated
DisplayName: Snowflake User Created
Enabled: false
LogTypes:
  - Snowflake.QueryHistory
Severity: Info
Reports:
  MITRE ATT&CK:
    - TA0003:T1136
Description: Detect new users created in Snowflake.
Tags:
  - Snowflake
  - '[MITRE] Persistence'
  - '[MITRE] Create Account'
Tests:
  - Name: User Created
    ExpectedResult: true
    Log:
      {
        "p_event_time": "2024-10-09 19:43:05.007000000",
        "p_log_type": "Snowflake.QueryHistory",
        "BYTES_DELETED": 0,
        "EXECUTION_STATUS": "SUCCESS",
        "QUERY_TEXT": "CREATE USER MERCY\nPASSWORD = '\u263a\u263a\u263a\u263a\u263a'\nDEFAULT_ROLE = PUBLIC;",
        "QUERY_TYPE": "CREATE_USER",
        "ROLE_NAME": "ACCOUNTADMIN",
        "USER_NAME": "LEX_LUTHOR",
        "WAREHOUSE_NAME": "ADMIN_WH"
      }


# ------ paired body: snowflake_stream_user_created.py ------

import re

from panther_snowflake_helpers import query_history_alert_context

CREATE_USER_EXPR = re.compile(r"create user (\w+).*", flags=re.I)

CREATE_USER = ""


def rule(event):
    # pylint: disable=global-statement
    global CREATE_USER
    CREATE_USER = CREATE_USER_EXPR.match(event.get("QUERY_TEXT", ""))
    return all(
        (
            event.get("EXECUTION_STATUS") == "SUCCESS",
            event.get("QUERY_TYPE") == "CREATE_USER",
            CREATE_USER is not None,
        )
    )


def title(event):
    # pylint: disable=global-statement
    global CREATE_USER
    new_user = CREATE_USER.group(1)
    actor = event.get("user_name", "<UNKNOWN ACTOR>")
    source = event.get("p_source_label", "<UNKNOWN SOURCE>")
    return f"{source}: Snowflake user {new_user} created by {actor}"


def alert_context(event):
    return query_history_alert_context(event)