Snowflake Table Copied Into Stage

AnalysisType: rule
Filename: snowflake_stream_table_copied_into_stage.py
RuleID: Snowflake.Stream.TableCopiedIntoStage
DisplayName: Snowflake Table Copied Into Stage
Enabled: true
LogTypes:
  - Snowflake.QueryHistory
Severity: Info
CreateAlert: false
Reports:
  MITRE ATT&CK:
    - TA0010:T1041      # Exfiltration Over C2 Channel
Description: A table was copied into 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: Copy from Table into Stage
    ExpectedResult: true
    Log:
      {
        "EXECUTION_STATUS": "SUCCESS",
        "QUERY_TEXT": "COPY INTO '@PANTHER_LOGS.PUBLIC.data_exfil/DATA.csv'
          FROM (SELECT * FROM PANTHER_LOGS.PUBLIC.amazon_eks_audit LIMIT 100)
          FILE_FORMAT = ( 
          TYPE='CSV' 
          COMPRESSION=GZIP
          FIELD_DELIMITER=',' 
          ESCAPE=NONE 
          ESCAPE_UNENCLOSED_FIELD=NONE 
          date_format='AUTO' 
          time_format='AUTO' 
          timestamp_format='AUTO'
          binary_format='UTF-8' 
          null_if='' 
          EMPTY_FIELD_AS_NULL = FALSE 
          )  
          overwrite=TRUE 
          single=FALSE 
          max_file_size=5368709120 
          header=TRUE",
        "QUERY_TYPE": "UNLOAD",
        "USER_NAME": "LEX_LUTHOR"
      }
  - Name: Copy from Stage into Table
    ExpectedResult: false
    Log:
      {
        "EXECUTION_STATUS": "SUCCESS",
        "QUERY_TEXT": "COPY INTO mytable\nFROM @mystage/result/data_ FILE_FORMAT =
      (FORMAT_NAME='CSV' COMPRESSION='GZIP');",
        "QUERY_TYPE": "UNLOAD",
        "USER_NAME": "LEX_LUTHOR"
      }


# ------ paired body: snowflake_stream_table_copied_into_stage.py ------

import re

STAGE_EXPR = re.compile(r"COPY\s+INTO\s+(?:\$\$|')?@([\w\.]+)", flags=re.I)
PATH_EXPR = re.compile(r"COPY\s+INTO\s+(?:\$\$|')?@([\w\./]+)(?:\$\$|')?\s+FROM", 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") == "UNLOAD",
            STAGE is not None,
            event.get("EXECUTION_STATUS") == "SUCCESS",
        )
    )


def alert_context(event):
    # pylint: disable=global-statement
    global STAGE
    path = PATH_EXPR.match(event.get("QUERY_TEXT", ""))
    return {"actor": event.get("USER_NAME"), "path": path.group(1), "stage": STAGE.group(1).lower()}