Helper reorg

STYLE_GUIDE.md

@@ -91,7 +91,7 @@ Panther's [dynamic auxiliary functions](https://docs.panther.com/detections/rule
 Check for `alert_context` functions in `global_helpers` for the LogType you are developing against.  Alert context can be extended in specific rules, for example:
 
 ```python
-from panther_base_helpers import aws_rule_context
+from panther_aws_helpers import aws_rule_context
 
 def alert_context(event):
     return aws_rule_context(event) | {'another_field': 'another_value'}

data_models/gcp_data_model.py

@@ -2,7 +2,8 @@
 from fnmatch import fnmatch
 
 import panther_event_type_helpers as event_type
-from panther_base_helpers import deep_get, get_binding_deltas
+from panther_base_helpers import deep_get
+from panther_gcp_helpers import get_binding_deltas
 
 ADMIN_ROLES = {
     # Primitive Rolesx

data_models/gsuite_data_model.py

@@ -1,6 +1,6 @@
 import panther_event_type_helpers as event_type
 from panther_base_helpers import deep_get
-from panther_base_helpers import gsuite_details_lookup as details_lookup
+from panther_gsuite_helpers import gsuite_details_lookup as details_lookup
 
 
 def get_event_type(event):

data_models/zendesk_data_model.py

@@ -1,5 +1,5 @@
 import panther_event_type_helpers as event_type
-from panther_base_helpers import ZENDESK_CHANGE_DESCRIPTION, zendesk_get_roles
+from panther_zendesk_helpers import ZENDESK_CHANGE_DESCRIPTION, zendesk_get_roles
 
 ZENDESK_TWO_FACTOR_SOURCES = {
     "Two-Factor authentication for all admins and agents",

global_helpers/default_test.py

@@ -6,7 +6,7 @@
 import unittest
 
 sys.path.append(os.path.dirname(__file__))
-import panther_default as p_d  # pylint: disable=C0413
+import panther_aws_helpers as p_d  # pylint: disable=C0413
 
 
 class TestAWSKeyAccountId(unittest.TestCase):

global_helpers/global_helpers_test.py

@@ -19,14 +19,16 @@
 
 import panther_asana_helpers as p_a_h  # pylint: disable=C0413
 import panther_auth0_helpers as p_auth0_h  # pylint: disable=C0413
+import panther_aws_helpers as p_aws_h  # pylint: disable=C0413
 import panther_azuresignin_helpers as p_asi_h  # pylint: disable=C0413
 import panther_base_helpers as p_b_h  # pylint: disable=C0413
+import panther_box_helpers as p_box_h  # pylint: disable=C0413
 import panther_cloudflare_helpers as p_cf_h  # pylint: disable=C0413
+import panther_crowdstrike_fdr_helpers as p_cf_fdr_h  # pylint: disable=C0413
 import panther_greynoise_helpers as p_greynoise_h  # pylint: disable=C0413
 import panther_ipinfo_helpers as p_i_h  # pylint: disable=C0413
 import panther_lookuptable_helpers as p_l_h  # pylint: disable=C0413
 import panther_notion_helpers as p_notion_h  # pylint: disable=C0413
-import panther_oss_helpers as p_o_h  # pylint: disable=C0413
 import panther_snyk_helpers as p_snyk_h  # pylint: disable=C0413
 import panther_tailscale_helpers as p_tscale_h  # pylint: disable=C0413
 import panther_tines_helpers as p_tines_h  # pylint: disable=C0413
@@ -94,7 +96,7 @@ def setUp(self):
         )
 
     def test_complete_event(self):
-        response = p_b_h.eks_panther_obj_ref(self.event)
+        response = p_aws_h.eks_panther_obj_ref(self.event)
         self.assertEqual(response.get("actor", ""), "kubernetes-admin")
         self.assertEqual(response.get("object", ""), "some-job-xxx1y")
         self.assertEqual(response.get("ns", ""), "default")
@@ -112,7 +114,7 @@ def test_all_missing_event(self):
         del temp_event["verb"]
         del temp_event["p_source_label"]
         temp_event = PantherEvent(temp_event)
-        response = p_b_h.eks_panther_obj_ref(temp_event)
+        response = p_aws_h.eks_panther_obj_ref(temp_event)
         self.assertEqual(response.get("actor", ""), "<NO_USERNAME>")
         self.assertEqual(response.get("object", ""), "<NO_OBJECT_NAME>")
         self.assertEqual(response.get("ns", ""), "<NO_OBJECT_NAMESPACE>")
@@ -126,7 +128,7 @@ def test_missing_subresource_event(self):
         temp_event = self.event.to_dict()
         del temp_event["objectRef"]["subresource"]
         temp_event = PantherEvent(temp_event)
-        response = p_b_h.eks_panther_obj_ref(temp_event)
+        response = p_aws_h.eks_panther_obj_ref(temp_event)
         self.assertEqual(response.get("resource", ""), "pods")
 
 
@@ -168,37 +170,37 @@ def setUp(self):
 
     def test_additional_details_string(self):
         event = ImmutableCaseInsensitiveDict({"additional_details": self.initial_str})
-        returns = p_b_h.box_parse_additional_details(event)
+        returns = p_box_h.box_parse_additional_details(event)
         self.assertEqual(returns.get("t", 0), 10)
 
     # in the case of a byte array, we expect the empty dict
     def test_additional_details_bytes(self):
         event = ImmutableCaseInsensitiveDict({"additional_details": self.initial_bytes})
-        returns = p_b_h.box_parse_additional_details(event)
+        returns = p_box_h.box_parse_additional_details(event)
         self.assertEqual(len(returns), 0)
 
     # In the case of a list ( not a string or bytes array ), expect un-altered return
     def test_additional_details_list(self):
         event = ImmutableCaseInsensitiveDict({"additional_details": self.initial_list})
-        returns = p_b_h.box_parse_additional_details(event)
+        returns = p_box_h.box_parse_additional_details(event)
         self.assertEqual(len(returns), 4)
 
     # in the case of a dict or similar, we expect it to be returned un-altered
     def test_additional_details_dict(self):
         event = ImmutableCaseInsensitiveDict({"additional_details": self.initial_dict})
-        returns = p_b_h.box_parse_additional_details(event)
+        returns = p_box_h.box_parse_additional_details(event)
         self.assertEqual(returns.get("t", 0), 10)
 
     # If it's a string with no json object to be decoded, we expect an empty dict back
     def test_additional_details_plain_str(self):
         event = ImmutableCaseInsensitiveDict({"additional_details": self.initial_str_no_json})
-        returns = p_b_h.box_parse_additional_details(event)
+        returns = p_box_h.box_parse_additional_details(event)
         self.assertEqual(len(returns), 0)
 
     # If it's a string with a json list, we expect the list
     def test_additional_details_str_list_json(self):
         event = ImmutableCaseInsensitiveDict({"additional_details": self.initial_str_list_json})
-        returns = p_b_h.box_parse_additional_details(event)
+        returns = p_box_h.box_parse_additional_details(event)
         self.assertEqual(len(returns), 4)
 
 
@@ -1100,11 +1102,11 @@ def setUp(self):
         )
 
     def test_is_different_with_fdr_event_type_provided(self):
-        response = p_b_h.filter_crowdstrike_fdr_event_type(self.input, "SomethingElse")
+        response = p_cf_fdr_h.filter_crowdstrike_fdr_event_type(self.input, "SomethingElse")
         self.assertEqual(response, True)
 
     def test_is_same_with_the_fdr_event_type_provided(self):
-        response = p_b_h.filter_crowdstrike_fdr_event_type(self.input, "DnsRequest")
+        response = p_cf_fdr_h.filter_crowdstrike_fdr_event_type(self.input, "DnsRequest")
         self.assertEqual(response, False)
 
     def test_is_entirely_different_type(self):
@@ -1115,7 +1117,7 @@ def test_is_entirely_different_type(self):
                 "event": {"foo": "bar"},
             }
         )
-        response = p_b_h.filter_crowdstrike_fdr_event_type(self.input, "DnsRequest")
+        response = p_cf_fdr_h.filter_crowdstrike_fdr_event_type(self.input, "DnsRequest")
         self.assertEqual(response, False)
 
 
@@ -1131,30 +1133,30 @@ def setUp(self):
         )
 
     def test_input_key_default_works(self):
-        response = p_b_h.get_crowdstrike_field(self.input, "zee", default="hello")
+        response = p_cf_fdr_h.get_crowdstrike_field(self.input, "zee", default="hello")
         self.assertEqual(response, "hello")
 
     def test_input_key_does_not_exist(self):
-        response = p_b_h.get_crowdstrike_field(self.input, "zee")
+        response = p_cf_fdr_h.get_crowdstrike_field(self.input, "zee")
         self.assertEqual(response, None)
 
     def test_input_key_exists(self):
-        response = p_b_h.get_crowdstrike_field(self.input, "cid")
+        response = p_cf_fdr_h.get_crowdstrike_field(self.input, "cid")
         self.assertEqual(response, "something")
 
     def test_input_key_can_be_found_in_event(self):
-        response = p_b_h.get_crowdstrike_field(self.input, "foo")
+        response = p_cf_fdr_h.get_crowdstrike_field(self.input, "foo")
         self.assertEqual(response, "bar")
 
     def test_input_key_can_be_found_in_unknown(self):
-        response = p_b_h.get_crowdstrike_field(self.input, "field")
+        response = p_cf_fdr_h.get_crowdstrike_field(self.input, "field")
         self.assertEqual(response, "is")
 
     def test_precedence(self):
         temp_event = self.input.to_dict()
         temp_event["event"]["field"] = "found"
         temp_event = PantherEvent(temp_event)
-        response = p_b_h.get_crowdstrike_field(temp_event, "field")
+        response = p_cf_fdr_h.get_crowdstrike_field(temp_event, "field")
         self.assertEqual(response, "found")
 
 
@@ -1974,10 +1976,10 @@ def setUp(self):
         )
 
     def test_distances(self):
-        nyc_to_sfo = p_o_h.km_between_ipinfo_loc(self.loc_nyc, self.loc_sfo)
-        nyc_to_athens = p_o_h.km_between_ipinfo_loc(self.loc_nyc, self.loc_athens)
-        nyc_to_aukland = p_o_h.km_between_ipinfo_loc(self.loc_nyc, self.loc_aukland)
-        aukland_to_nyc = p_o_h.km_between_ipinfo_loc(self.loc_aukland, self.loc_nyc)
+        nyc_to_sfo = p_i_h.km_between_ipinfo_loc(self.loc_nyc, self.loc_sfo)
+        nyc_to_athens = p_i_h.km_between_ipinfo_loc(self.loc_nyc, self.loc_athens)
+        nyc_to_aukland = p_i_h.km_between_ipinfo_loc(self.loc_nyc, self.loc_aukland)
+        aukland_to_nyc = p_i_h.km_between_ipinfo_loc(self.loc_aukland, self.loc_nyc)
         # I used https://www.nhc.noaa.gov/gccalc.shtml to get test comparison distances
         #
         # delta is set to 0.5% of total computed distanc from gccalc

global_helpers/panther_default.py → global_helpers/panther_aws_helpers.py

@@ -1,40 +1,63 @@
-# Define common code here that all of your policies and rules can share.
-#
-# Example usage:
-#
-# import panther_default
-# def policy(resource):
-#     return panther.example_helper()
-#
-
-
 import base64
 import binascii
 from typing import List
 
+from panther_config import config
 
-def example_helper():
-    return True
+AWS_ACCOUNTS = config.AWS_ACCOUNTS
 
 
-AWS_ACCOUNTS = {
-    # Add your AWS account IDs/names below:
-    "123456789012": "sample-account",
-}
+def aws_strip_role_session_id(user_identity_arn):
+    # The ARN structure is arn:aws:sts::123456789012:assumed-role/RoleName/<sessionId>
+    arn_parts = user_identity_arn.split("/")
+    if arn_parts:
+        return "/".join(arn_parts[:2])
+    return user_identity_arn
 
 
-def lookup_aws_account_name(account_id):
-    """Lookup the AWS account name, return the ID if not found
+def aws_rule_context(event: dict):
+    return {
+        "eventName": event.get("eventName", "<MISSING_EVENT_NAME>"),
+        "eventSource": event.get("eventSource", "<MISSING_ACCOUNT_ID>"),
+        "awsRegion": event.get("awsRegion", "<MISSING_AWS_REGION>"),
+        "recipientAccountId": event.get("recipientAccountId", "<MISSING_ACCOUNT_ID>"),
+        "sourceIPAddress": event.get("sourceIPAddress", "<MISSING_SOURCE_IP>"),
+        "userAgent": event.get("userAgent", "<MISSING_USER_AGENT>"),
+        "userIdentity": event.get("userIdentity", "<MISSING_USER_IDENTITY>"),
+    }
 
-    Args:
-        account_id (str): The AWS account ID
 
-    Returns:
-        str: The name of the AWS account ID
-        or
-        str: The AWS account ID (unnamed account)
-    """
-    return AWS_ACCOUNTS.get(account_id, f"{account_id} (unnamed account)")
+def aws_guardduty_context(event: dict):
+    return {
+        "description": event.get("description", "<MISSING DESCRIPTION>"),
+        "severity": event.get("severity", "<MISSING SEVERITY>"),
+        "id": event.get("id", "<MISSING ID>"),
+        "type": event.get("type", "<MISSING TYPE>"),
+        "resource": event.get("resource", {}),
+        "service": event.get("service", {}),
+    }
+
+
+def eks_panther_obj_ref(event):
+    user = event.deep_get("user", "username", default="<NO_USERNAME>")
+    source_ips = event.get("sourceIPs", ["0.0.0.0"])  # nosec
+    verb = event.get("verb", "<NO_VERB>")
+    obj_name = event.deep_get("objectRef", "name", default="<NO_OBJECT_NAME>")
+    obj_ns = event.deep_get("objectRef", "namespace", default="<NO_OBJECT_NAMESPACE>")
+    obj_res = event.deep_get("objectRef", "resource", default="<NO_OBJECT_RESOURCE>")
+    obj_subres = event.deep_get("objectRef", "subresource", default="")
+    p_source_label = event.get("p_source_label", "<NO_P_SOURCE_LABEL>")
+    if obj_subres:
+        obj_res = "/".join([obj_res, obj_subres])
+    return {
+        "actor": user,
+        "ns": obj_ns,
+        "object": obj_name,
+        "resource": obj_res,
+        "sourceIPs": source_ips,
+        "verb": verb,
+        "p_source_label": p_source_label,
+    }
 
 
 def aws_cloudtrail_success(event):
@@ -125,3 +148,17 @@ def aws_regions() -> List[str]:
         "us-west-1",
         "us-west-2",
     ]
+
+
+def lookup_aws_account_name(account_id):
+    """Lookup the AWS account name, return the ID if not found
+
+    Args:
+        account_id (str): The AWS account ID
+
+    Returns:
+        str: The name of the AWS account ID
+        or
+        str: The AWS account ID (unnamed account)
+    """
+    return AWS_ACCOUNTS.get(account_id, f"{account_id} (unnamed account)")

global_helpers/panther_aws_helpers.yml

@@ -0,0 +1,5 @@
+AnalysisType: global
+Filename: panther_aws_helpers.py
+GlobalID: "panther_aws_helpers"
+Description: >
+  Used to define global helpers and variables for AWS events.