Snowflake File Downloaded

AnalysisType: rule
Filename: snowflake_stream_file_downloaded.py
RuleID: Snowflake.Stream.FileDownloaded
DisplayName: Snowflake File Downloaded
Enabled: true
LogTypes:
  - Snowflake.QueryHistory
Severity: Info
CreateAlert: false
Reports:
  MITRE ATT&CK:
    - TA0010:T1041 # Exfiltration Over C2 Channel
Description: A file was downloaded from a stage.
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: Worksheet File Downloaded
    ExpectedResult: false
    Log:
      {
        "p_event_time": "2024-10-09 19:38:06.158000000",
        "p_log_type": "Snowflake.QueryHistory",
        "p_source_label": "SF-Ben",
        "EXECUTION_STATUS": "SUCCESS",
        "QUERY_TEXT": "GET '@~/worksheet_data/metadata' 'file:///'",
        "QUERY_TYPE": "GET_FILES",
        "ROLE_NAME": "PUBLIC",
        "USER_NAME": "CLARK_KENT"
      }
  - Name: Other File Downloaded
    ExpectedResult: true
    Log:
      {
        "p_event_time": "2024-10-09 19:38:06.158000000",
        "p_log_type": "Snowflake.QueryHistory",
        "p_source_label": "SF-Ben",
        "EXECUTION_STATUS": "SUCCESS",
        "QUERY_TEXT": "GET @PANTHER_LOGS.PUBLIC.data_exfil/DATA.csv 'file:///Users/lex.luthor/Documents'",
        "QUERY_TYPE": "GET_FILES",
        "ROLE_NAME": "PUBLIC",
        "USER_NAME": "LEX_LUTHOR"
      }


# ------ paired body: snowflake_stream_file_downloaded.py ------

import re

from panther_snowflake_helpers import query_history_alert_context

PATH_EXPR = re.compile(r"GET\s+(?:\$\$|')?@([a-zA-Z0-9_\./]+)(?:\$\$|')?\s", flags=re.I)
STAGE_EXPR = re.compile(r"GET\s+(?:\$\$|')?@([a-zA-Z0-9_\.]+)", flags=re.I)

PATH = ""
STAGE = ""


def rule(event):
    # pylint: disable=global-statement
    # Check these conditions first to avoid running an expensive regex on every log
    if not all(
        (
            event.get("QUERY_TYPE") == "GET_FILES",
            event.get("EXECUTION_STATUS") == "SUCCESS",
            # Avoid alerting for fetching worksheets:
            event.get("QUERY_TEXT") != "GET '@~/worksheet_data/metadata' 'file:///'",
        )
    ):
        return False

    global PATH
    PATH = PATH_EXPR.search(event.get("QUERY_TEXT", ""))

    return PATH is not None


def alert_context(event):
    # pylint: disable=global-statement
    global PATH
    global STAGE
    STAGE = STAGE_EXPR.match(event.get("QUERY_TEXT", ""))
    return query_history_alert_context(event) | {
        "path": PATH.group(1),
        "stage": None if not STAGE else STAGE.group(1).lower(),
    }