Snowflake User Enabled

AnalysisType: rule
Filename: snowflake_stream_user_enabled.py
RuleID: Snowflake.Stream.UserEnabled
DisplayName: Snowflake User Enabled
Enabled: true
LogTypes:
  - Snowflake.QueryHistory
Severity: Info
Reports:
  MITRE ATT&CK:
    - TA0003:T1136
Description: Detects users being re-enabled in your environment.
Tags:
  - Snowflake
  - '[MITRE] Persistence'
  - '[MITRE] Create Account'
Tests:
  - Name: User Enabled
    ExpectedResult: true
    Log:
      {
        "p_event_time": "2024-10-09 21:03:25.750000000",
        "p_log_type": "Snowflake.QueryHistory",
        "p_row_id": "6283439ab35193e891ac9ea1227b",
        "p_source_label": "SF-Ben",
        "EXECUTION_STATUS": "SUCCESS",
        "QUERY_TEXT": "ALTER USER CLARK_KENT SET DISABLED=FALSE;",
        "QUERY_TYPE": "ALTER_USER",
        "ROLE_NAME": "ACCOUNTADMIN",
        "USER_NAME": "LEX_LUTHOR",
        "WAREHOUSE_NAME": "DATAOPS_WH"
      }
  - Name: User Disabled
    ExpectedResult: false
    Log:
      {
        "p_event_time": "2024-10-09 21:03:25.750000000",
        "p_log_type": "Snowflake.QueryHistory",
        "p_row_id": "6283439ab35193e891ac9ea1227b",
        "p_source_label": "SF-Ben",
        "EXECUTION_STATUS": "SUCCESS",
        "QUERY_TEXT": "ALTER USER CLARK_KENT SET DISABLED=TRUE;",
        "QUERY_TYPE": "ALTER_USER",
        "ROLE_NAME": "ACCOUNTADMIN",
        "USER_NAME": "PERRY_WHITE",
        "WAREHOUSE_NAME": "DATAOPS_WH"
      }


# ------ paired body: snowflake_stream_user_enabled.py ------

import re

from panther_snowflake_helpers import query_history_alert_context

USER_ENABLED_EXPR = re.compile(r"alter\s+user\s+(.+?)\s+.*?set\s+disabled\s*=\s*false", flags=re.I)

USER_ENABLED = ""


def rule(event):
    # pylint: disable=global-statement
    global USER_ENABLED
    USER_ENABLED = USER_ENABLED_EXPR.match(event.get("QUERY_TEXT", ""))

    # Exit out early to avoid needless regex
    return all(
        (
            event.get("QUERY_TYPE") == "ALTER_USER",
            event.get("EXECUTION_STATUS") == "SUCCESS",
            USER_ENABLED is not None,
        )
    )


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


def alert_context(event):
    return query_history_alert_context(event)