replace single-layer deep_get with get

global_helpers/global_filter_azuresignin.py

@@ -14,7 +14,7 @@ def filter_include_event(event) -> bool:  # pylint: disable=unused-argument
     #
     # # example: event['tenantId']
     # # if tenantId were missing, we want default behavior to be to alert on this event.
-    # tenant_id = event.deep_get("tenantId", default="")
+    # tenant_id = event.get("tenantId", "")
     # return event_origin in ["333333eb-a222-33cc-9baf-4a1111111111", ""]
     #
     return True

global_helpers/global_filter_github.py

@@ -16,7 +16,7 @@ def filter_include_event(event) -> bool:  # pylint: disable=unused-argument
     #
     # # not all github enterprise events have org
     # # example: enterprise.self_hosted_runner_online
-    # org = event.deep_get("org", default="")
+    # org = event.get("org", "")
     # return org in ["my-prod-org", ""]
     #
     return True

global_helpers/global_filter_notion.py

@@ -14,7 +14,7 @@ def filter_include_event(event) -> bool:  # pylint: disable=unused-argument
     #
     # # example: workspace_id
     # # if we don't know the workspace_id, we want default behavior to be to alert on this event.
-    # workspace_id = event.deep_get("workspace_id", default="")
+    # workspace_id = event.get("workspace_id", "")
     # return workspace_id in ["ea65b016-6abc-4dcf-808b-e000099999999", ""]
     #
     return True

global_helpers/global_filter_snyk.py

@@ -13,7 +13,7 @@ def filter_include_event(event) -> bool:  # pylint: disable=unused-argument
     #
     # # not all snyk audit events have orgId & projectId
     # # example: group.user.add, sometimes api.access
-    # org = event.deep_get("orgId", default="")
+    # org = event.get("orgId", "")
     # return org in ["21111111-a222-4eee-8ddd-a99999999999", ""]
     #
     return True

global_helpers/global_filter_tines.py

@@ -11,7 +11,7 @@ def filter_include_event(event) -> bool:  # pylint: disable=unused-argument
     #    1. the specific tenant_id mentioned.
     #    2. events where tenant_id is undefined
     #
-    # tenant_id = event.deep_get("tenant_id", default="")
+    # tenant_id = event.get("tenant_id", "")
     # return tenant_id in ["1234", ""]
     #
     return True

global_helpers/panther_azuresignin_helpers.py

@@ -1,5 +1,5 @@
 def actor_user(event):
-    category = event.deep_get("category", default="")
+    category = event.get("category", "")
     if category in {"ServicePrincipalSignInLogs"}:
         return event.deep_get("properties", "servicePrincipalName")
     if category in {"SignInLogs", "NonInteractiveUserSignInLogs"}:
@@ -8,15 +8,15 @@ def actor_user(event):
 
 
 def is_sign_in_event(event):
-    return event.deep_get("operationName", default="") == "Sign-in activity"
+    return event.get("operationName", "") == "Sign-in activity"
 
 
 def azure_signin_alert_context(event) -> dict:
     ac_actor_user = actor_user(event)
     if ac_actor_user is None:
         ac_actor_user = "<NO_ACTORUSER>"
     a_c = {}
-    a_c["tenantId"] = event.deep_get("tenantId", default="<NO_TENANTID>")
+    a_c["tenantId"] = event.get("tenantId", "<NO_TENANTID>")
     a_c["source_ip"] = event.deep_get("properties", "ipAddress", default="<NO_SOURCEIP>")
     a_c["actor_user"] = ac_actor_user
     a_c["resourceDisplayName"] = event.deep_get(

global_helpers/panther_sublime_helpers.py

@@ -1,6 +1,6 @@
 def sublime_alert_context(event) -> dict:
     context = {}
-    context["events_type"] = event.get("type", default="<TYPE_NOT_FOUND>")
+    context["events_type"] = event.get("type", "<TYPE_NOT_FOUND>")
     context["users_emails"] = event.deep_get(
         "created_by", "email_address", default="<EMAIL_NOT_FOUND>"
     )

rules/appomni_rules/appomni_alert_passthrough.py

@@ -33,4 +33,4 @@ def dedup(event):
 
 def alert_context(event):
     # 'Threat' and 'related' data to be included in the alert sent to the alert destination
-    return {"threat": event.deep_get("rule", "threat"), "related": event.deep_get("related")}
+    return {"threat": event.deep_get("rule", "threat"), "related": event.get("related")}

rules/auth0_rules/auth0_cic_credential_stuffing.py

@@ -16,7 +16,7 @@ def title(event):
     user = event.deep_get(
         "data", "details", "request", "auth", "user", "email", default="<NO_USER_FOUND>"
     )
-    p_source_label = event.deep_get("p_source_label", default="<NO_P_SOURCE_LABEL_FOUND>")
+    p_source_label = event.get("p_source_label", "<NO_P_SOURCE_LABEL_FOUND>")
     return (
         f"Auth0 User [{user}] had a suspicious [{event_type}] event in "
         f"your organization's tenant [{p_source_label}]."

rules/auth0_rules/auth0_custom_role_created.py

@@ -30,7 +30,7 @@ def title(event):
     else:
         role_type = "custom"
 
-    p_source_label = event.deep_get("p_source_label", default="<NO_P_SOURCE_LABEL_FOUND>")
+    p_source_label = event.get("p_source_label", "<NO_P_SOURCE_LABEL_FOUND>")
     return (
         f"Auth0 User [{user}] created a "
         f"role [{request_body_name}] with [{role_type}] "

rules/auth0_rules/auth0_integration_installed.py

@@ -22,7 +22,7 @@ def title(event):
     user = event.deep_get(
         "data", "details", "request", "auth", "user", "email", default="<NO_USER_FOUND>"
     )
-    p_source_label = event.deep_get("p_source_label", default="<NO_P_SOURCE_LABEL_FOUND>")
+    p_source_label = event.get("p_source_label", "<NO_P_SOURCE_LABEL_FOUND>")
     return (
         f"Auth0 User [{user}] installed an integration from the actions library for "
         f"your organization's tenant [{p_source_label}]."

rules/auth0_rules/auth0_mfa_factor_setting_enabled.py

@@ -21,7 +21,7 @@ def title(event):
         "data", "details", "request", "auth", "user", "email", default="<NO_USER_FOUND>"
     )
     path = event.deep_get("data", "details", "request", "path", default="<NO_PATH_FOUND>")
-    p_source_label = event.deep_get("p_source_label", default="<NO_P_SOURCE_LABEL_FOUND>")
+    p_source_label = event.get("p_source_label", "<NO_P_SOURCE_LABEL_FOUND>")
     return (
         f"Auth0 User [{user}] enabled mfa factor settings for [{path}] "
         f"in your organization’s tenant [{p_source_label}]."

rules/auth0_rules/auth0_mfa_policy_disabled.py

@@ -24,7 +24,7 @@ def title(event):
     user = event.deep_get(
         "data", "details", "request", "auth", "user", "email", default="<NO_USER_FOUND>"
     )
-    p_source_label = event.deep_get("p_source_label", default="<NO_P_SOURCE_LABEL_FOUND>")
+    p_source_label = event.get("p_source_label", "<NO_P_SOURCE_LABEL_FOUND>")
     return (
         f"Auth0 User [{user}] set mfa requirement settings to 'Never' for your "
         f"organization's tenant [{p_source_label}]."

rules/auth0_rules/auth0_mfa_risk_assessment_disabled.py

@@ -26,7 +26,7 @@ def title(event):
     user = event.deep_get(
         "data", "details", "request", "auth", "user", "email", default="<NO_USER_FOUND>"
     )
-    p_source_label = event.deep_get("p_source_label", default="<NO_P_SOURCE_LABEL_FOUND>")
+    p_source_label = event.get("p_source_label", "<NO_P_SOURCE_LABEL_FOUND>")
     return (
         f"Auth0 User [{user}] disabled mfa risk assessment settings for your "
         f"organization’s tenant [{p_source_label}]."

rules/auth0_rules/auth0_mfa_risk_assessment_enabled.py

@@ -26,7 +26,7 @@ def title(event):
     user = event.deep_get(
         "data", "details", "request", "auth", "user", "email", default="<NO_USER_FOUND>"
     )
-    p_source_label = event.deep_get("p_source_label", default="<NO_P_SOURCE_LABEL_FOUND>")
+    p_source_label = event.get("p_source_label", "<NO_P_SOURCE_LABEL_FOUND>")
     return (
         f"Auth0 User [{user}] enabled mfa risk assessment settings for your "
         f"organization’s tenant [{p_source_label}]."

rules/auth0_rules/auth0_post_login_action_flow.py

@@ -25,7 +25,7 @@ def title(event):
     user = event.deep_get(
         "data", "details", "request", "auth", "user", "email", default="<NO_USER_FOUND>"
     )
-    p_source_label = event.deep_get("p_source_label", default="<NO_P_SOURCE_LABEL_FOUND>")
+    p_source_label = event.get("p_source_label", "<NO_P_SOURCE_LABEL_FOUND>")
     request_bindings = event.deep_get("data", "details", "request", "body", "bindings", default=[])
     response_bindings = event.deep_get(
         "data", "details", "response", "body", "bindings", default=[]

rules/auth0_rules/auth0_user_invitation_created.py

@@ -28,7 +28,7 @@ def title(event):
     inviter = event.deep_get(
         "data", "details", "request", "auth", "user", "email", default="<NO_INVITER>"
     )
-    source = event.deep_get("p_source_label", default="<NO_PSOURCE>")
+    source = event.get("p_source_label", "<NO_PSOURCE>")
     return f"Auth0 User [{inviter}] invited [{invitee}] to {inv_type} [{source}]]"
 
 

rules/auth0_rules/auth0_user_joined_tenant.py

@@ -30,7 +30,7 @@ def title(event):
     user = event.deep_get(
         "data", "details", "request", "auth", "user", "email", default="<NO_USER_FOUND>"
     )
-    p_source_label = event.deep_get("p_source_label", default="<NO_P_SOURCE_LABEL_FOUND>")
+    p_source_label = event.get("p_source_label", "<NO_P_SOURCE_LABEL_FOUND>")
     return (
         f"Auth0 User [{user}] has accepted an invitation to join your "
         f"organization's tenant [{p_source_label}]."

rules/aws_cloudtrail_rules/aws_cloudtrail_useraccesskeyauth.py

@@ -14,7 +14,7 @@ def title(event):
 
 def alert_context(event):
     return {
-        "ip_accessKeyId": event.get("sourceIpAddress", default="{not found}")
+        "ip_accessKeyId": event.get("sourceIpAddress", "{not found}")
         + ":"
         + event.deep_get("userIdentity", "accessKeyId", default="{not found}")
     }

rules/aws_cloudtrail_rules/aws_iam_compromised_key_quarantine.py

@@ -19,6 +19,6 @@ def rule(event):
 
 
 def title(event):
-    account_id = event.deep_get("recipientAccountId", default="<ACCOUNT_ID_NOT_FOUND>")
+    account_id = event.get("recipientAccountId", "<ACCOUNT_ID_NOT_FOUND>")
     user_name = event.deep_get("requestParameters", "userName", default="<USER_NAME_NOT_FOUND>")
     return f"Compromised Key quarantined for [{user_name}] in AWS Account [{account_id}]"

rules/aws_cloudtrail_rules/aws_rds_snapshot_shared.py

@@ -21,7 +21,7 @@ def rule(event):
 
 
 def title(event):
-    account_id = event.get("recipientAccountId", default="<ACCOUNT_ID_NOT_FOUND>")
+    account_id = event.get("recipientAccountId", "<ACCOUNT_ID_NOT_FOUND>")
     rds_instance_id = event.deep_get(
         "responseElements", "dBInstanceIdentifier", default="<DB_INSTANCE_ID_NOT_FOUND>"
     )

rules/aws_cloudtrail_rules/aws_s3_activity_greynoise.py

@@ -68,7 +68,7 @@ def rule(event):
 
 def title(event):
     # Group by ip-arn combinations
-    ip = event.deep_get("sourceIPAddress")
+    ip = event.get("sourceIPAddress")
     arn = event.deep_get("userIdentity", "arn")
     return f"GreyNoise malicious S3 events detected by {ip} from {arn}"
 

rules/crowdstrike_rules/crowdstrike_macos_add_trusted_cert.py

@@ -2,8 +2,8 @@
 
 
 def rule(event):
-    event_platform = event.deep_get("event_platform", default="<UNKNOWN_PLATFORM>")
-    fdr_event_type = event.deep_get("fdr_event_type", default="<UNKNOWN_FDR_EVENT_TYPE>")
+    event_platform = event.get("event_platform", "<UNKNOWN_PLATFORM>")
+    fdr_event_type = event.get("fdr_event_type", "<UNKNOWN_FDR_EVENT_TYPE>")
     image_filename = event.deep_get("event", "ImageFileName", default="<UNKNOWN_IMAGE_FILE_NAME>")
     command_line = event.deep_get("event", "CommandLine", default="<UNKNOWN_COMMAND_LINE>")
     return all(

rules/crowdstrike_rules/crowdstrike_macos_osascript_administrator.py

@@ -2,8 +2,8 @@
 
 
 def rule(event):
-    event_platform = event.deep_get("event_platform", default="<UNKNOWN_PLATFORM>")
-    event_simplename = event.deep_get("event_simplename", default="<UNKNOWN_EVENT_SIMPLENAME>")
+    event_platform = event.get("event_platform", "<UNKNOWN_PLATFORM>")
+    event_simplename = event.get("event_simplename", "<UNKNOWN_EVENT_SIMPLENAME>")
     image_filename = event.deep_get("event", "ImageFileName", default="<UNKNOWN_IMAGE_FILE_NAME>")
     command_line = event.deep_get("event", "CommandLine", default="<UNKNOWN_COMMAND_LINE>")
     return all(

rules/crowdstrike_rules/crowdstrike_macos_plutil_usage.py

@@ -9,8 +9,8 @@ def rule(event):
     ):
         return False
 
-    event_platform = event.deep_get("event_platform", default="<UNKNOWN_PLATFORM>")
-    fdr_event_type = event.deep_get("fdr_event_type", default="<UNKNOWN_FDR_EVENT_TYPE>")
+    event_platform = event.get("event_platform", "<UNKNOWN_PLATFORM>")
+    fdr_event_type = event.get("fdr_event_type", "<UNKNOWN_FDR_EVENT_TYPE>")
     image_filename = event.deep_get("event", "ImageFileName", default="<UNKNOWN_IMAGE_FILE_NAME>")
 
     return all(

rules/gcp_audit_rules/gcp_access_attempts_violating_vpc_service_controls.py

@@ -1,5 +1,5 @@
 def rule(event):
-    severity = event.deep_get("severity", default="")
+    severity = event.get("severity", "")
     status_code = event.deep_get("protoPayload", "status", "code", default="")
     violation_types = event.deep_walk(
         "protoPayload", "status", "details", "violations", "type", default=[]

rules/gcp_audit_rules/gcp_workforce_pool_created_or_updated.py

@@ -16,7 +16,7 @@ def title(event):
         "protoPayload", "request", "workforcePool", "name", default=""
     ).split("/")[-1]
 
-    resource = organization_id = event.deep_get("logName", default="<LOG_NAME_NOT_FOUND>").split(
+    resource = organization_id = event.get("logName", "<LOG_NAME_NOT_FOUND>").split(
         "/"
     )
 

rules/gitlab_rules/gitlab_production_password_reset_multiple_emails.py

@@ -3,12 +3,12 @@
 
 
 def rule(event):
-    path = event.get("path", default="")
+    path = event.get("path", "")
 
     if path != "/users/password":
         return False
 
-    params = event.get("params", default=[])
+    params = event.get("params", [])
     for param in params:
         if param.get("key") == "user":
             email = deep_get(param, "value", "email", default=[])

rules/mongodb_rules/mongodb_2fa_disabled.py

@@ -2,11 +2,11 @@
 
 
 def rule(event):
-    return event.deep_get("eventTypeName", default="") == "ORG_TWO_FACTOR_AUTH_OPTIONAL"
+    return event.get("eventTypeName", "") == "ORG_TWO_FACTOR_AUTH_OPTIONAL"
 
 
 def title(event):
-    user = event.deep_get("username", default="<USER_NOT_FOUND>")
+    user = event.get("username", "<USER_NOT_FOUND>")
     return f"MongoDB Atlas: [{user}] has disabled 2FA"
 
 

rules/mongodb_rules/mongodb_access_allowed_from_anywhere.py

@@ -3,20 +3,20 @@
 
 def rule(event):
     if (
-        event.deep_get("eventTypeName", default="") == "NETWORK_PERMISSION_ENTRY_ADDED"
-        and event.deep_get("whitelistEntry", default="") == "0.0.0.0/0"
+        event.get("eventTypeName", "") == "NETWORK_PERMISSION_ENTRY_ADDED"
+        and event.get("whitelistEntry", "") == "0.0.0.0/0"
     ):
         return True
     return False
 
 
 def title(event):
-    user = event.deep_get("username", default="<USER_NOT_FOUND>")
-    group_id = event.deep_get("groupId", default="<GROUP_NOT_FOUND>")
+    user = event.get("username", "<USER_NOT_FOUND>")
+    group_id = event.get("groupId", "<GROUP_NOT_FOUND>")
     return f"MongoDB: [{user}] has allowed access to group [{group_id}] from anywhere"
 
 
 def alert_context(event):
     context = mongodb_alert_context(event)
-    context["groupId"] = event.deep_get("groupId", default="<GROUP_NOT_FOUND>")
+    context["groupId"] = event.get("groupId", "<GROUP_NOT_FOUND>")
     return context

rules/mongodb_rules/mongodb_alerting_disabled.py

@@ -2,19 +2,19 @@
 
 
 def rule(event):
-    return event.deep_get("eventTypeName", default="") in [
+    return event.get("eventTypeName", "") in [
         "ALERT_CONFIG_DISABLED_AUDIT",
         "ALERT_CONFIG_DELETED_AUDIT",
     ]
 
 
 def title(event):
-    user = event.deep_get("username", default="<USER_NOT_FOUND>")
-    alert_id = event.deep_get("alertConfigId", default="<ALERT_NOT_FOUND>")
+    user = event.get("username", "<USER_NOT_FOUND>")
+    alert_id = event.get("alertConfigId", "<ALERT_NOT_FOUND>")
     return f"MongoDB: [{user}] has disabled or deleted security alert [{alert_id}]"
 
 
 def alert_context(event):
     context = mongodb_alert_context(event)
-    context["alertConfigId"] = event.deep_get("alertConfigId", default="<ALERT_NOT_FOUND>")
+    context["alertConfigId"] = event.get("alertConfigId", "<ALERT_NOT_FOUND>")
     return context

rules/mongodb_rules/mongodb_atlas_api_key_created.py

@@ -2,12 +2,12 @@
 
 
 def rule(event):
-    return event.deep_get("eventTypeName", default="") == "API_KEY_ACCESS_LIST_ENTRY_ADDED"
+    return event.get("eventTypeName", "") == "API_KEY_ACCESS_LIST_ENTRY_ADDED"
 
 
 def title(event):
-    user = event.deep_get("username", default="<USER_NOT_FOUND>")
-    public_key = event.deep_get("targetPublicKey", default="<PUBLIC_KEY_NOT_FOUND>")
+    user = event.get("username", "<USER_NOT_FOUND>")
+    public_key = event.get("targetPublicKey", "<PUBLIC_KEY_NOT_FOUND>")
     return f"MongoDB Atlas: [{user}] updated the allowed access list for API Key [{public_key}]"
 
 
@@ -16,8 +16,8 @@ def alert_context(event):
     links = event.deep_walk("links", "href", return_val="first", default="<LINKS_NOT_FOUND>")
     extra_context = {
         "links": links,
-        "event_type_name": event.deep_get("eventTypeName", default="<EVENT_TYPE_NOT_FOUND>"),
-        "target_public_key": event.deep_get("targetPublicKey", default="<PUBLIC_KEY_NOT_FOUND>"),
+        "event_type_name": event.get("eventTypeName", "<EVENT_TYPE_NOT_FOUND>"),
+        "target_public_key": event.get("targetPublicKey", "<PUBLIC_KEY_NOT_FOUND>"),
     }
     context.update(extra_context)
 

rules/mongodb_rules/mongodb_external_user_invited.py

@@ -11,8 +11,8 @@ def rule(event):
     global ALLOWED_DOMAINS  # pylint: disable=global-statement
     if isinstance(ALLOWED_DOMAINS, MagicMock):
         ALLOWED_DOMAINS = json.loads(ALLOWED_DOMAINS())  # pylint: disable=not-callable
-    if event.deep_get("eventTypeName", default="") == "INVITED_TO_ORG":
-        target_user = event.deep_get("targetUsername", default="")
+    if event.get("eventTypeName", "") == "INVITED_TO_ORG":
+        target_user = event.get("targetUsername", "")
         target_domain = target_user.split("@")[-1]
         return target_domain not in ALLOWED_DOMAINS
     return False

rules/mongodb_rules/mongodb_external_user_invited_no_config.py

@@ -2,11 +2,11 @@
 
 
 def rule(event):
-    if event.deep_get("eventTypeName", default="") != "INVITED_TO_ORG":
+    if event.get("eventTypeName", "") != "INVITED_TO_ORG":
         return False
 
-    user_who_sent_an_invitation = event.deep_get("username", default="")
-    user_who_was_invited = event.deep_get("targetUsername", default="")
+    user_who_sent_an_invitation = event.get("username", "")
+    user_who_was_invited = event.get("targetUsername", "")
     domain = user_who_sent_an_invitation.split("@")[-1]
 
     email_domains_are_different = not user_who_was_invited.endswith(domain)

rules/mongodb_rules/mongodb_identity_provider_activity.py

@@ -16,7 +16,7 @@ def rule(event):
         "OIDC_IDENTITY_PROVIDER_ENABLED",
         "OIDC_IDENTITY_PROVIDER_DISABLED",
     }
-    return event.deep_get("eventTypeName") in important_event_types
+    return event.get("eventTypeName") in important_event_types
 
 
 def title(event):