Snowflake Temporary Stage Created

AnalysisType: rule
Filename: snowflake_stream_temp_stage_created.py
RuleID: Snowflake.Stream.TempStageCreated
DisplayName: Snowflake Temporary Stage Created
Enabled: true
LogTypes:
  - Snowflake.QueryHistory
Severity: Info
CreateAlert: false
Reports:
  MITRE ATT&CK:
    - TA0010:T1041      # Exfiltration Over C2 Channel
Description: A temporary stage was created.
Reference: 
  https://cloud.google.com/blog/topics/threat-intelligence/unc5537-snowflake-data-theft-extortion/
Tags:
  - Snowflake
  - '[MITRE] Exfiltration'
  - '[MITRE] Exfiltration Over C2 Channel'
Tests:
  - Name: Successful Temp Stage Created
    ExpectedResult: true
    Log:
      {
        "p_event_time": "2024-10-09 21:06:03.631000000",
        "p_log_type": "Snowflake.QueryHistory",
        "p_source_id": "132d65cd-d6e4-4981-a209-a1d5902afd59",
        "p_source_label": "SF-Ben",
        "EXECUTION_STATUS": "SUCCESS",
        "QUERY_TEXT": "CREATE OR REPLACE TEMP STAGE panther_logs.PUBLIC.data_exfil;",
        "QUERY_TYPE": "CREATE",
        "USER_NAME": "LEX_LUTHOR",
        "WAREHOUSE_NAME": "ADMIN_WH"
      }
  - Name: Successful Temp Stage Created or Replaced
    ExpectedResult: true
    Log:
      {
        "p_event_time": "2024-10-09 21:06:03.631000000",
        "p_log_type": "Snowflake.QueryHistory",
        "p_source_id": "132d65cd-d6e4-4981-a209-a1d5902afd59",
        "p_source_label": "SF-Ben",
        "EXECUTION_STATUS": "SUCCESS",
        "QUERY_TEXT": "CREATE OR REPLACE TEMP STAGE my_temp_stage;",
        "QUERY_TYPE": "CREATE",
        "USER_NAME": "LEX_LUTHOR",
        "WAREHOUSE_NAME": "ADMIN_WH"
      }
  - Name: Unsuccessful Temp Stage Created
    ExpectedResult: false
    Log:
      {
        "p_event_time": "2024-10-09 21:06:03.631000000",
        "p_log_type": "Snowflake.QueryHistory",
        "p_source_id": "132d65cd-d6e4-4981-a209-a1d5902afd59",
        "p_source_label": "SF-Ben",
        "EXECUTION_STATUS": "FAIL",
        "QUERY_TEXT": "CREATE TEMP STAGE my_temp_stage;",
        "QUERY_TYPE": "CREATE",
        "USER_NAME": "LEX_LUTHOR",
        "WAREHOUSE_NAME": "ADMIN_WH"
      }


# ------ paired body: snowflake_stream_temp_stage_created.py ------

import re

from panther_snowflake_helpers import query_history_alert_context

STAGE_EXPR = re.compile(
    (
        r"CREATE\s+(?:OR\s+REPLACE\s+)?(?:TEMPORARY\s+|TEMP\s+)STAGE\s+"
        r"(?:IF\s+NOT\s+EXISTS\s+)?([a-zA-Z0-9_\.]+)"
    ),
    flags=re.I,
)

STAGE = ""


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

    return all(
        (
            event.get("QUERY_TYPE") == "CREATE",
            event.get("EXECUTION_STATUS") == "SUCCESS",
            STAGE is not None,
        )
    )


def alert_context(event):
    # pylint: disable=global-statement
    global STAGE
    return query_history_alert_context(event) | {"stage": STAGE.group(1).lower()}