Merge branch 'develop' into ben/revert-helper-backport

arielkr256 · web-flow · commit fdd4715ce85c · 2025-03-14T11:55:21.000-06:00
diff --git a/.cursor/rules/panther-rules.mdc b/.cursor/rules/panther-rules.mdc
@@ -0,0 +1,74 @@
+---
+description: Understanding how to write optimal Panther Rules
+globs: *.py
+alwaysApply: false
+---
+You are an expert cybersecurity detection engineer specialzed in cloud infrastructure, SaaS security, and MITRE ATT&CK for mapping attacker techniques. Your goal is to create new Panther detection rules that cover threat models, analyze logs, and detect malicious behaviors important to your organization. Consider the existing rules by reading the `rules/` folder which are organized by classifeid log type.
+
+# System Context
+Panther contains several types of Detections:
+- Rules (`rules/`): Streaming Python rules that analyze events one at a time, best applied towards high-fidelity events such as alerts from IDS systems (GuardDuty, Wiz, etc) or very high-confidence events like a cronjob containing a wget command or exfiltration from an S3 bucket.
+- Signal (found in `rules/`): A special mode of a Rule where no alert is generated and events are labeled, dictated by the CreateAlert attribute being set to false. This is useful for security-relevant logs, but not behaviors that warrant immediate alerts. Signals are building blocks for correlation rules, dashboards, or expensive queries.
+- Scheduled Rules: An aggregate style detection sourced from scheduled queries (`queries/`) declared in SQL + YAML. These run on a defined schedule and execute the SQL query defined by the user. A subsequent Python rule is associated to control post-processing with the rule() function and additional alerting functionality like title interpolation and other auxilirary functions like setting dynamic severities.
+
+All log data is normalized per a strictly-defined schema prior to being passed into the detection engine.
+
+Thresholding and deduplication are handled by the Panther platform. DO NOT implement this logic in the rule.
+
+# Conventions
+
+## Event Functions
+- Use `event.get()` to safely access `event` fields that may not exist: `bucket_name = event.get('requestParameters')`
+- Use `event.deep_get()` to access nested `event` fields: `bucket_name = event.deep_get('requestParameters', 'bucketName')` DO NOT IMPORT THIS FUNCTION. IT'S DIRECTLY ACCESSIBLE ON THE EVENT.
+- Use `event.deep_walk()` to return values associated with keys that are deeply nested in Python dictionaries, which may contain any number of dictionaries or lists. If it matches multiple event fields, an array of matches will be returned; if only one match is made, the value of that match will be returned.
+
+## Style
+- ONLY ASSIGN VARIABLES WHEN REUSE IS REQUIRED.
+- Don't write Rule methods with type annotations.
+- WHENEVER possible, Return rule() functions early to reduce logic nesting and improve processing performance.
+- Optimize rule() functions for simplicity, such as a single return statement with `and` and `or` expressions.
+- Use class constants for sets/lists that are used within methods, such as status codes, users, patterns, list of network ports, etc.
+- Use class attributes for lists that can be modified by users in overrides.
+
+# Python Rule Syntax
+
+A Python detection MUST CONTAIN TWO FILES: a Python file for logic and a YML file containing metadata.
+
+The YML file has the following structure:
+AnalysisType: # rule, scheduled_rule, correlation_rule, or policy
+Enabled: # boolean
+FileName: # the Python file name
+RuleID: # or PolicyId
+LogTypes: 
+Tags: 
+Tests: 
+ScheduledQueries: # only applicable to scheduled rules
+Suppressions: # only applicable to policies
+CreateAlert: # not applicable to policies
+Severity:
+Description:
+DedupPeriodMinutes:
+Threshold: 
+DisplayName:
+OutputIds:
+Reference:
+Runbook:
+SummaryAttributes:
+
+The Python file can contain the following functions:
+- `rule(event: Dict[str, Any]) -> bool`: The main detection logic that determines if an alert is sent. Returns `True` if the event matches the rule criteria, `False` otherwise. REQUIRED FOR ALL DETECTIONS.
+- `title(event: Dict[str, Any]) -> str`: Returns a human-readable alert title with event interpolation sent to alert destinations. THIS IS ALSO THE DEFAULT DEDUP(). Do not make it too unique, otherwise too many alerts will be sent. REQUIRED FOR ALL DETECTIONS BUT NOT FOR SIGNALS.
+- `dedup(event: Dict[str, Any]) -> str`: A deduplication key for the alert. OPTIONAL. Only use if specifically instructed by user.
+- `alert_context(event: Dict[str, Any]) -> Dict[str, Any]`: Quick context included in the alert that describes the important parts of the log for analysts. OPTIONAL.
+
+ONLY USE THE FOLLOWING FUNCTIONS IF DYNAMIC SELECTION IS REQUESTED BY THE USER EXPLICITLY. OTHERWISE, use the related YAML field.
+- `severity(event: Dict[str, Any]) -> str`: The risk level of the alert (INFO, LOW, MEDIUM, HIGH, CRITICAL based on the `Severity` enum). Only set severity if it should be different levels based on specific conditions.
+- `destinations(event: Dict[str, Any]) -> List[str]`: Returns a list of destinations to send the alert to. Only add this when the user specifies.
+- `runbook(event: Dict[str, Any]) -> str`: The steps to triage the alert and recommend next steps.
+- `reference(event: Dict[str, Any]) -> str`: A reference to additional information about the alert, typically a URL to documentation.
+
+If a user asks to create a Signal, then:
+1. Set `CreateAlert` to false
+2. Set `Severity` to INFO
+3. ONLY include the rule method
+4. Ignore alert-related metadata like deduplication
diff --git a/.github/workflows/validate.yml b/.github/workflows/validate.yml
@@ -1,14 +1,13 @@
 on:
-  pull_request:
-    types:
-      - closed
+  pull_request_review:
+    types: [submitted]
 
 permissions:
   contents: read
 
 jobs:
   validate:
-    if: github.event.pull_request.merged == true
+    if: github.event.review.state == 'approved' && github.event.pull_request.head.repo.fork == false
     name: Validate
     runs-on: ubuntu-latest
     env:
diff --git a/queries/snowflake_queries/snowflake_0108977_configuration_drift_query.yml b/queries/snowflake_queries/snowflake_0108977_configuration_drift_query.yml
@@ -17,7 +17,7 @@ Query: |
       start_time as p_event_time,
       end_time
   FROM snowflake.account_usage.query_history
-    WHERE execution_status = 'SUCCESS'
+    WHERE ((execution_status = 'SUCCESS'
       AND query_type NOT in ('SELECT')
       AND user_name NOT in ('PANTHER_ADMIN', 'PANTHERACCOUNTADMIN')
       AND (query_text ILIKE '%create role%'
@@ -41,11 +41,11 @@ Query: |
           OR query_text ILIKE '%drop_network_policy%'
           OR query_text ILIKE '%copy%'
           )
-      OR (
+      ) OR (
         query_text ilike '%grant%accountadmin%to%'
         AND query_type = 'GRANT'
         AND execution_status = 'SUCCESS'
-      )
+      ))
     AND p_occurs_since('1 day')
     ORDER BY end_time desc
     LIMIT 100;
diff --git a/queries/snowflake_queries/snowflake_0108977_configuration_drift_threat_hunting.yml b/queries/snowflake_queries/snowflake_0108977_configuration_drift_threat_hunting.yml
@@ -14,7 +14,7 @@ Query: |
       start_time as p_event_time,
       end_time
   FROM snowflake.account_usage.query_history
-    WHERE execution_status = 'SUCCESS'
+    WHERE ((execution_status = 'SUCCESS'
       AND query_type NOT in ('SELECT')
       AND user_name NOT in ('PANTHER_ADMIN', 'PANTHERACCOUNTADMIN')
       AND (query_text ILIKE '%create role%'
@@ -38,10 +38,10 @@ Query: |
           OR query_text ILIKE '%drop_network_policy%'
           OR query_text ILIKE '%copy%'
           )
-      OR (
+      ) OR (
         query_text ilike '%grant%accountadmin%to%'
         AND query_type = 'GRANT'
         AND execution_status = 'SUCCESS'
-      )
+      ))
     ORDER BY end_time desc
     LIMIT 100;
diff --git a/rules/aws_cloudtrail_rules/aws_resource_made_public.py b/rules/aws_cloudtrail_rules/aws_resource_made_public.py
@@ -5,67 +5,106 @@
 from policyuniverse.policy import Policy
 
 
-# Check that the IAM policy allows resource accessibility via the Internet
-def policy_is_internet_accessible(json_policy):
-    if json_policy is None:
+# Check if a policy (string or JSON) allows resource accessibility via the Internet
+# pylint: disable=too-complex
+def policy_is_internet_accessible(policy):
+    """
+    Check if a policy (string or JSON) allows resource accessibility via the Internet.
+
+    Args:
+        policy: A policy object that can be either a string or a JSON object
+
+    Returns:
+        bool: True if the policy allows internet access, False otherwise
+    """
+    # Handle empty policies (None, empty strings, empty dicts, etc.)
+    if not policy:
         return False
-    return Policy(json_policy).is_internet_accessible()
 
+    # Handle string policies by converting to JSON
+    if isinstance(policy, str):
+        try:
+            policy = json.loads(policy)
+        except json.JSONDecodeError:
+            return False
 
-# Normally this check helps avoid overly complex functions that are doing too many things,
-# but in this case we explicitly want to handle 10 different cases in 10 different ways.
-# Any solution that avoids too many return statements only increases the complexity of this rule.
-# pylint: disable=too-many-return-statements, too-complex
-def rule(event):
-    if not aws_cloudtrail_success(event):
-        return False
+    # Check if the policy has a wildcard principal but also has organization ID restrictions
+    # which should not be considered internet accessible
+    policy_obj = Policy(policy)
 
-    parameters = event.get("requestParameters", {})
-    # Ignore events that are missing request params
-    if not parameters:
+    # If policyuniverse thinks it's not internet accessible, trust that
+    if not policy_obj.is_internet_accessible():
         return False
 
-    policy = ""
-
-    # S3
-    if event["eventName"] == "PutBucketPolicy":
-        return policy_is_internet_accessible(parameters.get("bucketPolicy"))
+    # For policies with multiple statements, we need to check each statement individually
+    # If ANY statement is truly internet accessible, the policy is internet accessible
+    has_internet_accessible_statement = False
 
-    # ECR
-    if event["eventName"] == "SetRepositoryPolicy":
-        policy = parameters.get("policyText", {})
+    for statement in policy_obj.statements:
+        if statement.effect != "Allow" or "*" not in statement.principals:
+            continue
 
-    # Elasticsearch
-    if event["eventName"] in ["CreateElasticsearchDomain", "UpdateElasticsearchDomainConfig"]:
-        policy = parameters.get("accessPolicies", {})
+        # Check if there are organization ID conditions which restrict access
+        has_org_condition = False
+        for condition in statement.condition_entries:
+            if condition.category == "organization":
+                has_org_condition = True
+                break
 
-    # KMS
-    if event["eventName"] in ["CreateKey", "PutKeyPolicy"]:
-        policy = parameters.get("policy", {})
+        # If this statement has a wildcard principal but no organization ID restrictions,
+        # it's truly internet accessible
+        if not has_org_condition:
+            has_internet_accessible_statement = True
+            break
 
-    # S3 Glacier
-    if event["eventName"] == "SetVaultAccessPolicy":
-        policy = deep_get(parameters, "policy", "policy", default={})
+    return has_internet_accessible_statement
 
-    # SNS & SQS
-    if event["eventName"] in ["SetQueueAttributes", "CreateTopic"]:
-        policy = deep_get(parameters, "attributes", "Policy", default={})
 
-    # SNS
-    if (
-        event["eventName"] == "SetTopicAttributes"
-        and parameters.get("attributeName", "") == "Policy"
-    ):
-        policy = parameters.get("attributeValue", {})
+def rule(event):
+    if not aws_cloudtrail_success(event):
+        return False
 
-    # SecretsManager
-    if event["eventName"] == "PutResourcePolicy":
-        policy = parameters.get("resourcePolicy", {})
+    parameters = event.get("requestParameters", {})
+    # Ignore events that are missing request params
+    if not parameters:
+        return False
 
-    if not policy:
+    event_name = event.get("eventName", "")
+
+    # Special case for SNS topic attributes that need additional attribute name check
+    if event_name == "SetTopicAttributes" and parameters.get("attributeName", "") == "Policy":
+        policy_value = parameters.get("attributeValue", {})
+        return policy_is_internet_accessible(policy_value)
+
+    # Map of event names to policy locations in parameters
+    policy_location_map = {
+        # S3
+        "PutBucketPolicy": lambda p: p.get("bucketPolicy", {}),
+        # ECR
+        "SetRepositoryPolicy": lambda p: p.get("policyText", {}),
+        # Elasticsearch
+        "CreateElasticsearchDomain": lambda p: p.get("accessPolicies", {}),
+        "UpdateElasticsearchDomainConfig": lambda p: p.get("accessPolicies", {}),
+        # KMS
+        "CreateKey": lambda p: p.get("policy", {}),
+        "PutKeyPolicy": lambda p: p.get("policy", {}),
+        # S3 Glacier
+        "SetVaultAccessPolicy": lambda p: deep_get(p, "policy", "policy", default={}),
+        # SNS & SQS
+        "SetQueueAttributes": lambda p: deep_get(p, "attributes", "Policy", default={}),
+        "CreateTopic": lambda p: deep_get(p, "attributes", "Policy", default={}),
+        # SecretsManager
+        "PutResourcePolicy": lambda p: p.get("resourcePolicy", {}),
+    }
+
+    # Get the policy extraction function for this event name
+    policy_extractor = policy_location_map.get(event_name)
+    if not policy_extractor:
         return False
 
-    return policy_is_internet_accessible(json.loads(policy))
+    # Extract the policy using the appropriate function
+    policy = policy_extractor(parameters)
+    return policy_is_internet_accessible(policy)
 
 
 def title(event):
diff --git a/rules/aws_cloudtrail_rules/aws_resource_made_public.yml b/rules/aws_cloudtrail_rules/aws_resource_made_public.yml
