Panther rules: crowdstrike

from panther_crowdstrike_fdr_helpers import crowdstrike_network_detection_alert_context

# U.S. Gov Sanctioned Destinations
EMBARGO_COUNTRY_CODES = {
    "CU",  # Cuba
    "IR",  # Iran
    "KP",  # DPRK
    "SY",  # Syria
}


def get_enrichment_obj(event):
    return event.deep_get("p_enrichment", "ipinfo_location", "p_any_ip_addresses", default=None)


def rule(event):
    enrichment_obj = get_enrichment_obj(event)
    # enrichment_object returns a list.
    # Iterate over list and check if the "country" value matches the country codes.
    if enrichment_obj:
        for i in enrichment_obj:
            if i.get("country") in EMBARGO_COUNTRY_CODES:
                return True
    return False


def title(event):
    enrichment_obj = get_enrichment_obj(event)
    country_codes = set(
        i.get("country") for i in enrichment_obj if i.get("country") in EMBARGO_COUNTRY_CODES
    )

    return f"Connection made to embargoed country: [{country_codes}]."


def alert_context(event):
    if event.get("p_log_type") == "Crowdstrike.FDREvent":
        return crowdstrike_network_detection_alert_context(event) | {
            "p_any_ip_addresses": event.get("p_any_ip_addresses")
        }

    return {"p_any_ip_addresses": event.get("p_any_ip_addresses")}

AnalysisType: rule
Description: "Detection to alert when internal asset is communicating with an sanctioned destination. This detection leverages Panther UDM and IPInfo enrichment."
Reference: "U.S. Sanctioned Destinations - https://www.bis.doc.gov/index.php/policy-guidance/country-guidance/sanctioned-destinations"
DisplayName: "Connection to Embargoed Country"
Enabled: true
DedupPeriodMinutes: 60
LogTypes:
  - Crowdstrike.FDREvent
RuleID: "Connection.to.Embargoed.Country"
Threshold: 1
Filename: crowdstrike_connection_to_embargoed_country.py
Severity: Low

from panther_crowdstrike_event_streams_helpers import audit_keys_dict, cs_alert_context

# List of priviledged roles.
# IMPORTANT: YOU MUST ADD ANY CUSTOM ADMIN ROLES YOURSELF
ADMIN_ROLES = {
    "billing_dashboard_admin",
    "falconhost_admin",
    "firewall_manager",
    "xdr_admin",  # NG SIEM Admin
    "remote_responder_three",  # Remote Responder Admin
}


def get_roles_assigned(event):
    """Returns a list of the roles assigned in this event."""
    # Extract the AuditKeyValues construct
    audit_keys = audit_keys_dict(event)
    # Return Roles
    return audit_keys.get("roles", "").split(",")


def rule(event):
    # Ignore non role-granting events
    if not all(
        [
            event.deep_get("event", "OperationName") == "grantUserRoles",
            event.deep_get("event", "Success"),
        ]
    ):
        return False

    # Raise alert if any of the admin roles were assigned
    roles_assigned = get_roles_assigned(event)
    return bool(ADMIN_ROLES & set(roles_assigned))


def title(event):
    audit_keys = audit_keys_dict(event)
    actor = audit_keys["actor_user"]
    target = audit_keys["target_name"]
    admin_roles = set(get_roles_assigned(event)) & ADMIN_ROLES
    return f"{actor} assigned admin roles to {target}: {', '.join(list(admin_roles))}"


def dedup(event):
    # The title includes the role names, but if the actor assigned more roles to the user, we
    #   dedup those alerts as well.
    audit_keys = audit_keys_dict(event)
    actor = audit_keys["actor_user"]
    target = audit_keys["target_name"]
    return f"{actor}-{target}"


def alert_context(event):
    context = cs_alert_context(event)
    actor = context.get("actor_user", "UNKNOWN_ACTOR")
    target = context.get("target_name", "UNKNOWN_TARGET")
    context["actor_target"] = f"{actor}-{target}"
    return context

AnalysisType: rule
Filename: crowdstrike_admin_role_assigned.py
RuleID: "Crowdstrike.AdminRoleAssigned"
DisplayName: "Crowdstrike Admin Role Assigned"
Enabled: true
LogTypes:
  - Crowdstrike.EventStreams
Severity: Medium
Reports:
  MITRE ATT&CK:
    - TA0003:T1098.003 # Persistence: Additional Cloud Roles
    - TA0004:T1098.003 # Priv Escalation: Additional Cloud Roles
Description: A user was assigned a priviledged role
DedupPeriodMinutes: 60
Threshold: 1
Runbook: Confirm the role assignment is justified.

from panther_crowdstrike_event_streams_helpers import audit_keys_dict, cs_alert_context


def rule(event):
    # Return True if allowlist is deleted
    if event.deep_get("event", "OperationName") == "DeleteAllowlistGroup":
        return True

    # Return True if allowlist is disabled
    if event.deep_get("event", "OperationName") == "UpdateAllowlistGroup":
        audit_keys = audit_keys_dict(event)
        return audit_keys.get("active") == "false" and audit_keys.get("old_active") == "true"

    return False


def title(event):
    actor = event.deep_get("event", "UserId")
    audit_keys = audit_keys_dict(event)
    list_name = audit_keys.get("group_name", "UNKNOWN_GROUP")

    verb = {"DeleteAllowlistGroup": "deleted", "UpdateAllowlistGroup": "disabled"}.get(
        event.deep_get("event", "OperationName"), "removed"
    )

    return f'{actor} {verb} IP allowlist "{list_name}"'


def dedup(event):
    # We wanna group alerts if a user disables, then deletes the same allowlist
    actor = event.deep_get("event", "UserId")
    audit_keys = audit_keys_dict(event)
    list_name = audit_keys.get("group_name", "UNKNOWN_GROUP")
    return f"{actor}-{list_name}"


def alert_context(event):
    return cs_alert_context(event)


def severity(event):
    # Downgrade severity if a disabled allowlist was deleted
    if all(
        [
            event.deep_get("event", "OperationName") == "DeleteAllowlistGroup",
            audit_keys_dict(event).get("enabled") == "false",
        ]
    ):
        return "INFO"
    return "DEFAULT"

AnalysisType: rule
Filename: crowdstrike_allowlist_removed.py
RuleID: "Crowdstrike.AllowlistRemoved"
DisplayName: "Crowdstrike Allowlist Removed"
Enabled: true
LogTypes:
  - Crowdstrike.EventStreams
Severity: Medium
Reports:
  MITRE ATT&CK:
    - TA0040:T1531 # Account Access removal
Description: A user deleted an allowlist
DedupPeriodMinutes: 60
Threshold: 1
Runbook: Confirm if the deleted allowlist is needed.

from panther_crowdstrike_event_streams_helpers import cs_alert_context


def rule(event):
    return all(
        [
            event.deep_get("event", "OperationName") == "CreateAPIClient",
            event.deep_get("event", "Success"),
        ]
    )


def title(event):
    user = event.deep_get("event", "UserId")
    service = event.deep_get("event", "ServiceName")
    return f"{user} created a new API key in {service}"


def alert_context(event):
    return cs_alert_context(event)

AnalysisType: rule
Filename: crowdstrike_api_key_created.py
RuleID: "Crowdstrike.API.Key.Created"
DisplayName: "Crowdstrike API Key Created"
Reports:
    MITRE ATT&CK:
        - TA0006:T1098.001 # Additional Cloud Credentials
Enabled: true
LogTypes:
  - Crowdstrike.EventStreams
Severity: Medium
Description: A user created an API Key in CrowdStrike
DedupPeriodMinutes: 60
Threshold: 1
Runbook: Reach out to the user if needed to validate the activity.

from panther_crowdstrike_event_streams_helpers import cs_alert_context


def rule(event):
    return all(
        [
            event.deep_get("event", "OperationName") == "DeleteAPIClients",
            event.deep_get("event", "Success"),
        ]
    )


def title(event):
    user = event.deep_get("event", "UserId")
    service = event.deep_get("event", "ServiceName")
    return f"{user} deleted an API key in {service}"


def alert_context(event):
    return cs_alert_context(event)

AnalysisType: rule
Filename: crowdstrike_api_key_deleted.py
RuleID: "Crowdstrike.API.Key.Deleted"
DisplayName: "Crowdstrike API Key Deleted"
Reports:
    MITRE ATT&CK:
        - TA0040:T1531 # Account Access Removal
        - TA0005:T1070 # Indicator Removal
Enabled: true
LogTypes:
  - Crowdstrike.EventStreams
Severity: Medium
Description: A user deleted an API Key in CrowdStrike
DedupPeriodMinutes: 60
Threshold: 1
Runbook: Validate this action was authorized.

import uuid

from panther_crowdstrike_fdr_helpers import (
    crowdstrike_detection_alert_context,
    get_crowdstrike_field,
)


def rule(event):
    return (
        get_crowdstrike_field(event, "ExternalApiType", default="none")
        == "Event_EppDetectionSummaryEvent"
    )


def title(event):
    return (
        f"Crowdstrike Alert ({get_crowdstrike_field(event, 'Technique')}) - "
        + f"{get_crowdstrike_field(event, 'Hostname')}"
        + f"({get_crowdstrike_field(event, 'UserName')})"
    )


def alert_context(event):
    return crowdstrike_detection_alert_context(event)


def severity(event):
    # First, try returning the severity based on the SeverityName
    sevname = get_crowdstrike_field(event, "SeverityName").upper()
    allowed_values = ("INFO", "LOW", "MEDIUM", "HIGH", "CRITICAL")
    if sevname == "INFORMATIONAL":
        sevname = "INFO"
    if sevname in allowed_values:
        return sevname

    # Else, fallback on the numerical value, falling back on MEDIUM if we still don't have a value
    sevval = get_crowdstrike_field(event, "Severity")
    return {1: "INFO", 2: "LOW", 3: "MEDIUM", 4: "HIGH", 5: "CRITICAL", 6: "CRITICAL"}.get(
        sevval, "DEFAULT"
    )


def dedup(event):
    # CompositeId is unique per detection indicator and present in both
    # DetectionSummary and FDREvent formats. EventUUID is null in FDREvent,
    # which previously caused all detections to share the dedup key "None ".
    composite_id = get_crowdstrike_field(event, "CompositeId")
    if composite_id:
        return composite_id
    # Fallback: generate a unique ID so detections never silently merge
    return str(uuid.uuid4())

AnalysisType: rule
Filename: crowdstrike_detection_passthrough.py
RuleID: "Crowdstrike.Detection.passthrough"
DisplayName: "Crowdstrike Detection Passthrough"
Enabled: true
Severity: Medium
LogTypes:
  - Crowdstrike.DetectionSummary
  - Crowdstrike.FDREvent
Tags:
  - Crowdstrike
Description: Crowdstrike Falcon has detected malicious activity on a host.
Runbook: Follow the Falcon console link and follow the IR process as needed.
Reference: https://www.crowdstrike.com/blog/tech-center/hunt-threat-activity-falcon-endpoint-protection/
DedupPeriodMinutes: 60
SummaryAttributes:
  - p_any_ip_addresses

from panther_core import PantherEvent
from panther_crowdstrike_event_streams_helpers import cs_alert_context


def rule(event: PantherEvent):
    return event.deep_get("metadata", "eventType") == "EppDetectionSummaryEvent"


def title(event: PantherEvent):
    alert_title = event.deep_get("event", "Name", default="New CrowdStrike Detection")
    alert_desc = event.deep_get("event", "Description")
    return f"{alert_title}: {alert_desc}" if alert_desc else alert_title


def dedup(event: PantherEvent):
    if alert_id := event.deep_get("event", "CompositeId"):
        return alert_id
    # Else, fall back on title string
    return title(event)


def severity(event: PantherEvent):
    # First, try returning the severity based on the SeverityName
    sevname = str(event.deep_get("event", "SeverityName")).upper()
    allowed_values = ("INFO", "LOW", "MEDIUM", "HIGH", "CRITICAL")
    if sevname == "INFORMATIONAL":
        sevname = "INFO"
    if sevname in allowed_values:
        return sevname

    # Else, fallback on the numerical value, falling back on MEDIUM if we still don't have a value
    sevval = event.deep_get("event", "Severity") // 20
    return {0: "INFO", 1: "LOW", 2: "MEDIUM", 3: "HIGH", 4: "CRITICAL", 5: "CRITICAL"}.get(
        sevval, "DEFAULT"
    )


def reference(event: PantherEvent):
    return event.deep_get("event", "FalconHostLink", default="")


def alert_context(event: PantherEvent):
    context = cs_alert_context(event)
    context.update(
        {
            "FalconLink": event.deep_get("event", "FalconHostLink", default="<NO LINK PROVIDED>"),
            "CompositeId": event.deep_get("event", "CompositeId", default="<NO ID PROVIDED>"),
            "FileName": event.deep_get("event", "FileName", default="<NO FILENAME PROVIDED>"),
            "FilePath": event.deep_get("event", "FilePath", default="<NO FILEPATH PROVIDED>"),
            "UserName": event.deep_get("event", "UserName", default="<NO USERNAME PROVIDED>"),
        }
    )
    return context

AnalysisType: rule
Filename: crowdstrike_detection_summary.py
RuleID: "Crowdstrike.EppDetectionSummary"
DisplayName: "Crowdstrike Detection Summary"
Enabled: true
LogTypes:
  - Crowdstrike.EventStreams
Severity: Medium
Description: Forwards any alerts generated by CrowdStrike to your Panther destinations.
DedupPeriodMinutes: 60
Threshold: 1
Runbook: Followup with the alert according to CrowdStrike's recommendations.
Tags:
  - Crowdstrike
  - Passthrough

AnalysisType: correlation_rule
RuleID: "Crowdstrike.EphemeralUserAccount"
DisplayName: "Crowdstrike Ephemeral User Account"
Enabled: true
Severity: High
Description: Detects when a user account is created and deleted within 12 hours. This aims to detect ephemeral user accounts infiltrators might use to avoid suspicion.
Reference: ""
Reports:
  MITRE ATT&CK:
    - TA0003:T1136.003 # Persistence: Create Cloud Account
    - TA0005:T1070 # Defense Evasion: Indicator Removal
Detection:
  - Sequence:
      - ID: AccountCreated
        RuleID: Crowdstrike.NewUserCreated
      - ID: AccountDeleted
        RuleID: Crowdstrike.UserDeleted
    Transitions:
      - ID: User Created FOLLOWED BY User Deleted
        From: AccountCreated
        To: AccountDeleted
        WithinTimeFrameMinutes: 720 # 12 hours
        Match:
          - On: p_alert_context.target_name
    LookbackWindowMinutes: 2160
    Schedule:
      RateMinutes: 1440
      TimeoutMinutes: 1

from panther_crowdstrike_fdr_helpers import crowdstrike_process_alert_context

LOLBAS_EXE = {
    "AppInstaller.exe",
    "At.exe",
    "Atbroker.exe",
    "Bash.exe",
    "Bitsadmin.exe",
    "CertOC.exe",
    "CertReq.exe",
    "Certutil.exe",
    "Cmd.exe",
    "Cmdkey.exe",
    "cmdl32.exe",
    "Cmstp.exe",
    "ConfigSecurityPolicy.exe",
    "Conhost.exe",
    "Control.exe",
    "Csc.exe",
    "Cscript.exe",
    "CustomShellHost.exe",
    "DataSvcUtil.exe",
    "Desktopimgdownldr.exe",
    "DeviceCredentialDeployment.exe",
    "Dfsvc.exe",
    "Diantz.exe",
    "Diskshadow.exe",
    "Dnscmd.exe",
    "Esentutl.exe",
    "Eventvwr.exe",
    "Expand.exe",
    "Explorer.exe",
    "Extexport.exe",
    "Extrac32.exe",
    "Findstr.exe",
    "Finger.exe",
    "fltMC.exe",
    "Forfiles.exe",
    "Ftp.exe",
    "Gpscript.exe",
    "Hh.exe",
    "IMEWDBLD.exe",
    "Ie4uinit.exe",
    "Ieexec.exe",
    "Ilasm.exe",
    "Infdefaultinstall.exe",
    "Installutil.exe",
    "Jsc.exe",
    "Ldifde.exe",
    "Makecab.exe",
    "Mavinject.exe",
    "Mmc.exe",
    "MpCmdRun.exe",
    "Msbuild.exe",
    "Msconfig.exe",
    "Msdt.exe",
    "Msedge.exe",
    "Mshta.exe",
    "Msiexec.exe",
    "Netsh.exe",
    "Odbcconf.exe",
    "OfflineScannerShell.exe",
    "OneDriveStandaloneUpdater.exe",
    "Pcalua.exe",
    "Pcwrun.exe",
    "Pktmon.exe",
    "Pnputil.exe",
    "Presentationhost.exe",
    "Print.exe",
    "PrintBrm.exe",
    "Psr.exe",
    "Rasautou.exe",
    "rdrleakdiag.exe",
    "Reg.exe",
    "Regasm.exe",
    "Regedit.exe",
    "Regini.exe",
    "Regsvcs.exe",
    "Regsvr32.exe",
    "Replace.exe",
    "Rpcping.exe",
    "Rundll32.exe",
    "Runexehelper.exe",
    "Runonce.exe",
    "Runscripthelper.exe",
    "Sc.exe",
    "Schtasks.exe",
    "Scriptrunner.exe",
    "Setres.exe",
    "SettingSyncHost.exe",
    "ssh.exe",
    "Stordiag.exe",
    "SyncAppvPublishingServer.exe",
    "Ttdinject.exe",
    "Tttracer.exe",
    "Unregmp2.exe",
    "vbc.exe",
    "Verclsid.exe",
    "Wab.exe",
    "winget.exe",
    "Wlrmdr.exe",
    "Wmic.exe",
    "WorkFolders.exe",
    "Wscript.exe",
    "Wsreset.exe",
    "wuauclt.exe",
    "Xwizard.exe",
    "fsutil.exe",
    "wt.exe",
}


def rule(event):
    if event.deep_get("event", "event_simpleName") == "ProcessRollup2":
        if event.deep_get("event", "event_platform") == "Win":
            exe = event.udm("process_name")
            return bool(exe.lower() in [x.lower() for x in LOLBAS_EXE])
    return False


def title(event):
    exe = event.deep_get("event", "ImageFileName").split("\\")[-1]
    return f'Crowdstrike: LOLBAS execution - [{exe}] - [{event.deep_get("event", "CommandLine")}]'


def dedup(event):
    # dedup string on "{aid}-{exe}"
    exe = event.udm("process_name")
    return f'{event.deep_get("event", "aid")}-{exe}'


def alert_context(event):
    return crowdstrike_process_alert_context(event)

AnalysisType: rule
LogTypes:
  - Crowdstrike.FDREvent
RuleID: "Crowdstrike.FDR.LOLBAS"
DisplayName: "Crowdstrike FDR LOLBAS"
Description: "Living off the land binaries and script usage"
Reference: "https://lolbas-project.github.io/"
Threshold: 1
DedupPeriodMinutes: 1440
Enabled: false
Filename: crowdstrike_lolbas.py
Severity: Info
Tags:
  - Configuration Required

from panther_crowdstrike_event_streams_helpers import audit_keys_dict, cs_alert_context, str_to_list


def rule(event):
    # Only alert if an allow list is created or edited
    op_name = event.deep_get("event", "OperationName")
    if op_name not in ("CreateAllowlistGroup", "UpdateAllowlistGroup"):
        return False

    return True


def title(event):
    actor = event.deep_get("event", "UserId")
    action = {
        "CreateAllowlistGroup": "created a new",
        "UpdateAllowlistGroup": "made changes to",
    }.get(event.deep_get("event", "OperationName"))
    group = audit_keys_dict(event).get("group_name", "UNKNWOWN GROUP")
    return f"{actor} {action} Crowdstrike IP allowlist group: {group}"


def alert_context(event):
    context = cs_alert_context(event)

    # Be nice and concert the "lists" into actual lists so customers can easily process the alert
    #   context
    for key in ("cidrs", "old_cidrs", "contexts", "old_contexts"):
        if context.get(key):
            try:
                context[key] = str_to_list(context[key])
            except ValueError:
                pass  # Just ignore if we can't unmarshal it

    # Find out what entries were removed, and which were added
    op_name = event.deep_get("event", "OperationName")
    audit_keys = audit_keys_dict(event)
    added_cidrs = []
    removed_cidrs = []
    added_contexts = []
    removed_contexts = []

    def getlist(key: str):
        return str_to_list(audit_keys.get(key))

    match op_name:
        case "UpdateAllowlistGroup":
            new_cidrs = getlist("cidrs")
            old_cidrs = getlist("old_cidrs")
            new_ctx = getlist("contexts")
            old_ctx = getlist("old_contexts")
            added_cidrs = get_unique_entries(new_cidrs, old_cidrs)
            removed_cidrs = get_unique_entries(old_cidrs, new_cidrs)
            added_contexts = get_unique_entries(new_ctx, old_ctx)
            removed_contexts = get_unique_entries(old_ctx, new_ctx)
        case "CreateAllowlistGroup":
            added_cidrs = str_to_list(audit_keys.get("cidrs", []))
            added_contexts = str_to_list(audit_keys.get("contexts", []))

    context.update(
        {
            "changes": {
                "cidr_added": added_cidrs,
                "cidr_removed": removed_cidrs,
                "context_added": added_contexts,
                "context_removed": removed_contexts,
            }
        }
    )

    return context


def get_unique_entries(list1: list, list2: list) -> list:
    """Returns items in l1 that are not in l2."""
    return list(set(list1) - set(list2))

AnalysisType: rule
Filename: crowdstrike_ip_allowlist_changed.py
RuleID: "Crowdstrike.IpAllowlistChanged"
DisplayName: "Crowdstrike IP Allowlist Changed"
Enabled: true
LogTypes:
  - Crowdstrike.EventStreams
Severity: Info
CreateAlert: false
Reports:
  MITRE ATT&CK:
    - TA0003:T1556.009 # Persistsnce: Modify Authentication Process: Conditional Access Policies
    - TA0005:T1556.009 # Defense Evasion: Modify Authentication Process: Conditional Access Policies
    - TA0006:T1556.009 # Credential Access: Modify Authentication Process: Conditional Access Policies
Description: Updates were made to Falcon console's allowlist. This could indicate a bad actor permitting access from another machine, or could be attackers preventing legitimate actors from accessing the console.
DedupPeriodMinutes: 60
Threshold: 1
Runbook: Validate this action was authorized.

AnalysisType: scheduled_query
Description: Detects creation of large zip files, which can indicate attempts of exfiltration
Enabled: false
SnowflakeQuery: |
  select
    ppr.commandline as parent_commandline,
    zip_proc.*
  from
    (
    select
      zips.*,
      pr2.targetprocessid as process_targetpid,
      pr2.parentprocessid as process_parentpid,
      pr2.commandline as process_commandline
    from
      (
        select
          *
        from
          panther_logs.public.crowdstrike_unknown
        where
          event_simpleName IN (
            'GzipFileWritten',
            'SevenZipFileWritten',
            'ZipFileWritten',
            'BZip2FileWritten'
          )
          and p_occurs_since('1 day')
          and CAST(unknown_payload:Size as integer) > 10000000
      ) zips
      left join panther_logs.public.crowdstrike_processrollup2 pr2
      on zips.ContextProcessId = pr2.targetprocessid

      where
      pr2.commandline like any(
        '%zip%'
      )

      and not (
        pr2.commandline like any (
          '%curl%',
          '/Application%',
          '%install%'
        )
      )

    ) zip_proc
    LEFT JOIN panther_logs.public.crowdstrike_processrollup2 ppr
    on zip_proc.process_parentpid = ppr.targetprocessid
  where
    (
      (parent_commandline is null) or
      not (parent_commandline like any (
        '%homebrew%',
        '%Homebrew%',
        '/Application%',
        '%install%'
        )
      )
    )

DatabricksQuery: |
  select
    ppr.commandline as parent_commandline,
    zip_proc.*
  from
    (
    select
      zips.*,
      pr2.targetprocessid as process_targetpid,
      pr2.parentprocessid as process_parentpid,
      pr2.commandline as process_commandline
    from
      (
        select
          *
        from
          panther_logs.crowdstrike_unknown
        where
          event_simpleName IN (
            'GzipFileWritten',
            'SevenZipFileWritten',
            'ZipFileWritten',
            'BZip2FileWritten'
          )
          and p_occurs_since('1 day')
          and CAST(unknown_payload:Size as BIGINT) > 10000000
      ) zips
      left join panther_logs.crowdstrike_processrollup2 pr2
      on zips.ContextProcessId = pr2.targetprocessid

      where
      pr2.commandline LIKE '%zip%'

      and not (
        pr2.commandline LIKE '%curl%'
        OR pr2.commandline LIKE '/Application%'
        OR pr2.commandline LIKE '%install%'
      )

    ) zip_proc
    LEFT JOIN panther_logs.crowdstrike_processrollup2 ppr
    on zip_proc.process_parentpid = ppr.targetprocessid
  where
    (
      (parent_commandline is null) or
      not (
        parent_commandline LIKE '%homebrew%'
        OR parent_commandline LIKE '%Homebrew%'
        OR parent_commandline LIKE '/Application%'
        OR parent_commandline LIKE '%install%'
      )
    )
QueryName: "CrowdStrike Large Zip Creation"
Schedule:
  RateMinutes: 1440
  TimeoutMinutes: 5

# This file is the part of the Crowdstrike FDREvent migration, and it's the equivalent of
# https://github.com/panther-labs/panther-analysis/blob/b61db1ecf3967c5f6a44c1782f8891fd5f54384d/queries/crowdstrike_queries/CrowdStrike_Large_Zip_Creation.yml
#
AnalysisType: scheduled_query
Description: Detects creation of large zip files, which can indicate attempts of exfiltration (crowdstrike_fdrevent table)
Enabled: false
SnowflakeQuery: |
  select
    ppr.event:CommandLine as parent_commandline,
    zip_proc.*
  from
    (
    select
      zips.*,
      pr2.event:TargetProcessId as process_targetpid,
      pr2.event:ParentProcessId as process_parentpid,
      pr2.event:CommandLine as process_commandline
    from
      (
        select
          *
        from
          panther_logs.public.crowdstrike_fdrevent
        where
          event_simpleName IN (
            'GzipFileWritten',
            'SevenZipFileWritten',
            'ZipFileWritten',
            'BZip2FileWritten'
          )
          and p_occurs_since('1 day')
          and CAST(event:Size as integer) > 10000000
      ) zips
      left join panther_logs.public.crowdstrike_fdrevent pr2
      on zips.ContextProcessId = pr2.TargetProcessId_decimal and pr2.fdr_event_type = 'ProcessRollup2'

      where
      pr2.event:CommandLine like any(
        '%zip%'
      )

      and not (
        pr2.event:CommandLine like any (
          '%curl%',
          '/Application%',
          '%install%'
        )
      )

    ) zip_proc
    LEFT JOIN panther_logs.public.crowdstrike_fdrevent ppr
    on zip_proc.process_parentpid = ppr.TargetProcessId_decimal and ppr.fdr_event_type = 'ProcessRollup2'
  where
    (
      (parent_commandline is null) or
      not (parent_commandline like any (
        '%homebrew%',
        '%Homebrew%',
        '/Application%',
        '%install%'
        )
      )
    )

DatabricksQuery: |
  select
    ppr.event.CommandLine as parent_commandline,
    zip_proc.*
  from
    (
    select
      zips.*,
      pr2.event.TargetProcessId as process_targetpid,
      pr2.event.ParentProcessId as process_parentpid,
      pr2.event.CommandLine as process_commandline
    from
      (
        select
          *
        from
          panther_logs.crowdstrike_fdrevent
        where
          event_simpleName IN (
            'GzipFileWritten',
            'SevenZipFileWritten',
            'ZipFileWritten',
            'BZip2FileWritten'
          )
          and p_occurs_since('1 day')
          and CAST(event:Size as BIGINT) > 10000000
      ) zips
      left join panther_logs.crowdstrike_fdrevent pr2
      on zips.ContextProcessId = pr2.TargetProcessId_decimal and pr2.fdr_event_type = 'ProcessRollup2'

      where
      pr2.event.CommandLine LIKE '%zip%'

      and not (
        pr2.event.CommandLine LIKE '%curl%'
        OR pr2.event.CommandLine LIKE '/Application%'
        OR pr2.event.CommandLine LIKE '%install%'
      )

    ) zip_proc
    LEFT JOIN panther_logs.crowdstrike_fdrevent ppr
    on zip_proc.process_parentpid = ppr.TargetProcessId_decimal and ppr.fdr_event_type = 'ProcessRollup2'
  where
    (
      (parent_commandline is null) or
      not (
        parent_commandline LIKE '%homebrew%'
        OR parent_commandline LIKE '%Homebrew%'
        OR parent_commandline LIKE '/Application%'
        OR parent_commandline LIKE '%install%'
      )
    )
QueryName: "CrowdStrike Large Zip Creation (crowdstrike_fdrevent table)"
Schedule:
  RateMinutes: 1440
  TimeoutMinutes: 5

from panther_crowdstrike_fdr_helpers import crowdstrike_process_alert_context


def rule(event):
    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(
        [
            event_platform == "Mac",
            fdr_event_type == "ProcessRollup2",
            image_filename == "/usr/bin/security",
            "add-trusted-cert" in command_line,
        ]
    )


def title(event):
    host = event.get("ComputerName") or event.get("aid", "<AID_NOT_FOUND>")
    return f"Crowdstrike: New trusted cert added on device [{host}]"


def alert_context(event):
    return crowdstrike_process_alert_context(event)

AnalysisType: rule
DisplayName: CrowdStrike MacOS Added Trusted Cert
Description: Detects attempt to install a root certificate on MacOS
Enabled: true
Filename: crowdstrike_macos_add_trusted_cert.py
RuleID: Crowdstrike.Macos.Add.Trusted.Cert
Reference: https://docs.panther.com/data-onboarding/supported-logs/crowdstrike#crowdstrike.processrollup2
Severity: Medium
LogTypes:
  - Crowdstrike.FDREvent

from panther_crowdstrike_fdr_helpers import crowdstrike_process_alert_context


def rule(event):
    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(
        [
            event_platform == "Mac",
            event_simplename == "ProcessRollup2",
            image_filename == "/usr/bin/osascript",
            "with administrator privileges" in command_line,
        ]
    )


def title(event):
    host = event.get("ComputerName") or event.get("aid", "<AID_NOT_FOUND>")
    return f"Crowdstrike: Osascript run with administrator privileges on [{host}]"


def alert_context(event):
    return crowdstrike_process_alert_context(event)

AnalysisType: rule
DisplayName: CrowdStrike MacOS Osascript as Administrator
Description: Detects usage of osascript with administrator privileges
Enabled: true
Filename: crowdstrike_macos_osascript_administrator.py
RuleID: Crowdstrike.Macos.Osascript.Administrator
Reference: https://www.sentinelone.com/blog/how-offensive-actors-use-applescript-for-attacking-macos/
Severity: Medium
LogTypes:
  - Crowdstrike.FDREvent

AnalysisType: scheduled_query
QueryName: CrowdStrike MacOS plutil Novel Plist Modification
Enabled: false
Description: Detects when plutil performs MODIFICATION operations (insert, replace, remove, create) on plist files it hasn't modified in the previous 30 days. Uses behavioral filtering to exclude read-only operations (convert, print, lint) that generate noise.
Tags:
  - Anomaly Detection
  - CrowdStrike
  - macOS
  - Persistence
  - T1547.011
Query: |-
    -- pragma: template

    {% import 'anomalies' new_unique_values %}

    with subquery as (
        SELECT
            aid as device_id,
            p_event_time,
            event:CommandLine as command_line,
            -- Extract plist file path from command line (robust extraction)
            -- 1. Try regex match for absolute/relative paths ending in .plist
            -- 2. Fall back to last space-delimited argument
            -- 3. Strip quotes to handle: "/path with spaces/file.plist"
            -- Handles: /Library/file.plist, ~/file.plist, "path with spaces.plist"
            REPLACE(
                REPLACE(
                    COALESCE(
                        REGEXP_SUBSTR(event:CommandLine, '[~/][^\\s]*\\.plist'),
                        SPLIT_PART(event:CommandLine, ' ', -1)
                    ),
                    '"', ''
                ),
                '''', ''
            ) as plist_file
        FROM panther_logs.public.crowdstrike_fdrevent
        WHERE
            p_occurs_since('60 day')
            AND event_platform = 'Mac'
            AND fdr_event_type = 'ProcessRollup2'
            AND event:ImageFileName = '/usr/bin/plutil'
            -- BEHAVIORAL FILTERING: Focus on modification operations
            -- Only alert on operations that CHANGE plists, not just read them
            -- CAUGHT (modifications):
            --   plutil -insert RunAtLoad -bool true file.plist
            --   plutil --replace ProgramArguments -array /tmp/evil file.plist
            --   plutil -remove SomeKey file.plist
            --   plutil --create xml1 file.plist
            -- FILTERED OUT (read-only):
            --   plutil -convert binary1 file.plist (format conversion)
            --   plutil -p file.plist (print/read)
            --   plutil -lint file.plist (syntax check)
            --   plutil -extract Key xml1 file.plist (read value)
            -- Use ILIKE for case-insensitive matching (handles -INSERT, --Replace, etc.)
            AND (
                event:CommandLine ILIKE '%--insert%'      -- Adding new keys
                OR event:CommandLine ILIKE '%-insert%'
                OR event:CommandLine ILIKE '%--replace%'  -- Modifying existing keys
                OR event:CommandLine ILIKE '%-replace%'
                OR event:CommandLine ILIKE '%--remove%'   -- Deleting keys
                OR event:CommandLine ILIKE '%-remove%'
                OR event:CommandLine ILIKE '%--create%'   -- Creating new plists
                OR event:CommandLine ILIKE '%-create%'
            )
            -- Exclude known benign operations
            AND event:CommandLine != 'plutil -convert binary1 /Library/Preferences/com.tinyspeck.slackmacgap.plist'
    ),
    {{ new_unique_values('subquery', 'device_id', 'plist_file', '1day') }}

Schedule:
  CronExpression: '0 */4 * * *'
  TimeoutMinutes: 5

def rule(event):
    # The scheduled query returns only new plist files that haven't been modified before
    # Add additional filtering here if needed
    plist_file = event.get("plist_file", "")

    # Filter out empty, malformed, or invalid results
    if not plist_file:
        return False

    # Filter out whitespace-only strings
    if isinstance(plist_file, str) and not plist_file.strip():
        return False

    # Filter out placeholder/error values from query
    if plist_file == "<UNKNOWN_FILE>":
        return False

    # Any valid result from the query should trigger an alert
    return True


def dedup(event):
    # Group alerts by device and plist file
    device_id = event.get("device_id", "<UNKNOWN_DEVICE>")
    plist_file = event.get("plist_file", "<UNKNOWN_FILE>")
    return f"{device_id}:{plist_file}"


def title(event):
    plist_file = event.get("plist_file", "<UNKNOWN_FILE>")
    return f"Crowdstrike: plutil modified new plist file: {plist_file}"


def severity(event):
    plist_file = event.get("plist_file", "")

    # Critical: System-level persistence with elevated privileges
    if any(
        loc in plist_file for loc in ["/System/Library/LaunchDaemons/", "/Library/LaunchDaemons/"]
    ):
        return "HIGH"

    # High: System-level LaunchAgents or system directory modifications
    if "/Library/LaunchAgents/" in plist_file or "/System/" in plist_file:
        return "HIGH"

    # Default: Medium for all other novel modifications
    return "DEFAULT"


def alert_context(event):
    plist_file = event.get("plist_file", "<UNKNOWN_FILE>")

    # Determine risk indicators based on plist location
    risk_indicators = []
    if any(loc in plist_file for loc in ["/Library/LaunchAgents/", "/Library/LaunchDaemons/"]):
        risk_indicators.append("System-level persistence location")
    elif any(loc in plist_file for loc in ["LaunchAgents/", "LaunchDaemons/"]):
        risk_indicators.append("User-level persistence location")
    elif "/Applications/" in plist_file:
        risk_indicators.append("Application bundle modification (potential tampering)")
    elif "/System/" in plist_file:
        risk_indicators.append("System directory modification (elevated privileges)")

    return {
        "device_id": event.get("device_id", "<UNKNOWN_DEVICE>"),
        "plist_file": plist_file,
        "detection_type": "Anomaly - New Modification Detected",
        "baseline_period": "30 days",
        "risk_indicators": risk_indicators if risk_indicators else ["Non-standard plist location"],
    }

AnalysisType: scheduled_rule
Filename: crowdstrike_macos_plutil_newfiles.py
DisplayName: CrowdStrike MacOS plutil Novel Plist Modification (Anomaly Detection)
Enabled: false
Status: Experimental
Description: |
  Detects when plutil performs modification operations (insert, replace, remove, create) on plist
  files it hasn't modified in the previous 30 days. Uses behavioral filtering to exclude read-only
  operations (convert, print, lint). This anomaly-based approach reduces noise from legitimate
  repeated operations while catching novel persistence attempts, even in /Applications/.
Severity: Medium
DedupPeriodMinutes: 1440
Threshold: 1
Reference: https://attack.mitre.org/techniques/T1547/011/
Runbook: |
  1. Verify the plist modification was authorized. Query CrowdStrike for the full plutil command line on the affected device to determine the exact operation performed (insert, replace, remove, create) and identify the parent process that executed the command.
  2. Review the modification context including the user account, source process, and timing. Assess the risk level based on the plist location - LaunchAgents/LaunchDaemons indicate persistence attempts (high risk), while /Applications/ modifications could be legitimate software updates (medium risk). Check for suspicious parent processes such as bash, python, curl, or remote shell activity.
  3. If the modification is suspicious or unauthorized, quarantine the device and analyze the modified plist content for malicious payloads or references to external scripts. Review all processes currently running that may have been launched by the modified plist, revoke any persistence mechanisms, and hunt for related IOCs across other endpoints. Escalate for forensic analysis if compromise is confirmed.
Tags:
  - Anomaly Detection
  - macOS
  - Persistence
  - CrowdStrike
  - T1547.011
InlineFilters:
  - All: []
ScheduledQueries:
  - CrowdStrike MacOS plutil Novel Plist Modification
RuleID: Crowdstrike.Macos.Plutil.NewFiles

from panther_crowdstrike_fdr_helpers import crowdstrike_process_alert_context


def rule(event):
    command_line = event.deep_get("event", "CommandLine", default="<UNKNOWN_COMMAND_LINE>")
    if (
        command_line
        == "plutil -convert binary1 /Library/Preferences/com.tinyspeck.slackmacgap.plist"
    ):
        return False

    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(
        [
            event_platform == "Mac",
            fdr_event_type == "ProcessRollup2",
            image_filename == "/usr/bin/plutil",
        ]
    )


def dedup(event):
    command_line = event.deep_get("event", "CommandLine", default="<UNKNOWN_COMMAND_LINE>")
    file_name = command_line.split(" ")[-1]
    return file_name


def title(_):
    return "Crowdstrike: plutil was used to modify a plist file on one or more devices"


def alert_context(event):
    return crowdstrike_process_alert_context(event)

AnalysisType: rule
DisplayName: CrowdStrike MacOS plutil Usage
Description: Detects the usage of plutil to modify plist files. Plist files run on start up and are often used by attackers to maintain persistence.
Enabled: true
Filename: crowdstrike_macos_plutil_usage.py
RuleID: Crowdstrike.Macos.Plutil.Usage
Reference: https://www.crowdstrike.com/blog/reconstructing-command-line-activity-on-macos/#:~:text=Terminal.savedState/.-,Windows.plist,-The%20file%20windows
Severity: Medium
LogTypes:
  - Crowdstrike.FDREvent

AnalysisType: correlation_rule
RuleID: "Crowdstrike.NewAdminUserCreated"
DisplayName: "Crowdstrike New Admin User Created"
Enabled: true
Severity: High
Description: Detects when a user account is created and assigned admin permissions
Reports:
  MITRE ATT&CK:
    - TA0003:T1136.003 # Persistence: Create Cloud Account
    - TA0003:T1098.003 # Persistence: Additional Cloud Roles
    - TA0004:T1098.003 # Priv Escalation: Additional Cloud Roles
Detection:
  - Sequence:
      - ID: AccountCreated
        RuleID: Crowdstrike.NewUserCreated
      - ID: AdminRoleAssigned
        RuleID: Crowdstrike.AdminRoleAssigned
    Transitions:
      - ID: AcountCreated FOLLOWED BY AdminRoleAssigned ON target AND actor
        From: AccountCreated
        To: AdminRoleAssigned
        WithinTimeFrameMinutes: 45
        Match:
          - On: p_alert_context.actor_target
    LookbackWindowMinutes: 2160
    Schedule:
      RateMinutes: 1440
      TimeoutMinutes: 1

from panther_base_helpers import key_value_list_to_dict
from panther_crowdstrike_event_streams_helpers import cs_alert_context


def rule(event):
    return all(
        [
            event.deep_get("event", "OperationName") == "createUser",
            event.deep_get("event", "Success"),
        ]
    )


def title(event):
    audit_keys = key_value_list_to_dict(
        event.deep_get("event", "AuditKeyValues"), "Key", "ValueString"
    )

    actor = event.deep_get("event", "UserId", "UNKNOWN USER")
    target = audit_keys.get("target_name")

    return f"[{actor}] created a new user: [{target}]"


def alert_context(event):
    context = cs_alert_context(event)
    actor = context.get("actor_user", "UNKNOWN_ACTOR")
    target = context.get("target_name", "UNKNOWN_TARGET")
    context["actor_target"] = f"{actor}-{target}"
    return context

AnalysisType: rule
Filename: crowdstrike_new_user_created.py
RuleID: "Crowdstrike.NewUserCreated"
DisplayName: "Crowdstrike New User Created"
Enabled: true
LogTypes:
  - Crowdstrike.EventStreams
Severity: Info
CreateAlert: false
Reports:
  MITRE ATT&CK:
    - TA0003:T1136.003 # Create Cloud Account
Description: A new Crowdstrike user was created
DedupPeriodMinutes: 60
Threshold: 1
Runbook: Confirm the new user is valid.

from panther_crowdstrike_fdr_helpers import get_crowdstrike_field


def rule(event):
    return (
        get_crowdstrike_field(event, "ExternalApiType", default="<unknown-ExternalApiType>")
        == "Event_RemoteResponseSessionStartEvent"
    )


def title(event):
    user_name = get_crowdstrike_field(event, "UserName", default="<unknown-UserName>")
    hostname_field = get_crowdstrike_field(
        event, "HostnameField", default="<unknown-HostNameField>"
    )
    return f"{user_name} started a Crowdstrike Real-Time Response (RTR) shell on {hostname_field}"


def severity(event):
    if get_crowdstrike_field(event, "UserName", default="").lower().endswith("@crowdstrike.com"):
        return "INFO"
    return "DEFAULT"


def dedup(event):
    return get_crowdstrike_field(event, "UserName", default="<unknown-UserName>")


def alert_context(event):
    return {
        "Start Time": get_crowdstrike_field(
            event, "StartTimestamp", default="<unknown-StartTimestamp>"
        ),
        "SessionId": get_crowdstrike_field(event, "SessionId", default="<unknown-SessionId>"),
        "Actor": get_crowdstrike_field(event, "UserName", default="<unknown-UserName>"),
        "Target Host": get_crowdstrike_field(
            event, "HostnameField", default="<unknown-HostnameField>"
        ),
    }

AnalysisType: rule
DedupPeriodMinutes: 60 # 1 hour
DisplayName: "Crowdstrike Real Time Response (RTS) Session"
Enabled: true
Filename: crowdstrike_real_time_response_session.py
RuleID: "Crowdstrike.RealTimeResponse.Session"
Severity: Medium
LogTypes:
  - Crowdstrike.Unknown
  - Crowdstrike.FDREvent
Tags:
  - Crowdstrike
Description: >
  Alert when someone uses Crowdstrike’s RTR (real-time response) capability to access a machine remotely to run commands.
Runbook: >
  Validate the real-time response session started by the Actor.
Reference: https://falcon.us-2.crowdstrike.com/documentation/71/real-time-response-and-network-containment#reviewing-real-time-response-audit-logs

from panther_crowdstrike_fdr_helpers import crowdstrike_detection_alert_context

REMOTE_ACCESS_EXECUTABLES = {
    "teamviewer_service.exe",
    "winvnc.exe",
    "racwinvnc.exe",
    "tvnserver.exe",
    "ultravnc.exe",
    "remotelyanywhere.exe",
    "logmein.exe",
    "g2svc.exe",
    "vncserver.exe",
    "awhost32.exe",
    "r_server.exe",
    "raabout.exe",
    "anydesk.exe",
    "ammyyadmin.exe",
    "putty.exe",
    "mstsc.exe",
    "chrome-remote-desktop-host.exe",
}


def rule(event):
    if event.get("fdr_event_type", "") == "ProcessRollup2":
        if event.get("event_platform", "") == "Win":
            process_name = (
                event.deep_get("event", "ImageFileName", default="").lower().split("\\")[-1]
            )
            return process_name in REMOTE_ACCESS_EXECUTABLES
    return False


def title(event):
    tool = (
        event.deep_get("event", "ImageFileName", default="<TOOL_NOT_FOUND>").lower().split("\\")[-1]
    )
    return f"Crowdstrike: Remote access tool [{tool}] detected"


def alert_context(event):
    return crowdstrike_detection_alert_context(event)

AnalysisType: rule
Description: Detects usage of common remote access tools.
DisplayName: "Crowdstrike Remote Access Tool Execution"
Enabled: true
Filename: crowdstrike_remote_access_tool_execution.py
Reference: https://attack.mitre.org/techniques/T1219/
Severity: Info
DedupPeriodMinutes: 60
LogTypes:
  - Crowdstrike.FDREvent
RuleID: "Crowdstrike.Remote.Access.Tool.Execution"
Threshold: 1

from panther_crowdstrike_fdr_helpers import crowdstrike_detection_alert_context

REMOTE_SHELL_TOOLS = {
    #   process name: reverse shell signature
    "nc.exe": ["cmd.exe", "powershell.exe", "command.exe"],
    "ncat.exe": ["cmd.exe", "powershell.exe", "command.exe"],
    "socat.exe": ["cmd.exe", "powershell.exe", "command.exe"],
    "psexec.exe": ["cmd.exe", "powershell.exe", "command.exe"],
    "python.exe": ["cmd.exe", "powershell.exe", "command.exe"],
    "powershell.exe": ["System.Net.Sockets.TcpClient"],
    "certutil.exe": ["-urlcache"],
    "php.exe": ["fsockopen", "cmd.exe", "powershell.exe", "command.exe"],
}


def rule(event):
    if event.get("fdr_event_type", "") == "ProcessRollup2":
        if event.get("event_platform", "") == "Win":
            process_name = (
                event.deep_get("event", "ImageFileName", default="").lower().split("\\")[-1]
            )
            command_line = event.deep_get("event", "CommandLine", default="")
            signatures = REMOTE_SHELL_TOOLS.get(process_name, [])
            for signature in signatures:
                if signature in command_line:
                    return True
    return False


def title(event):
    tool = (
        event.deep_get("event", "ImageFileName", default="<TOOL_NOT_FOUND>").lower().split("\\")[-1]
    )
    host = event.get("ComputerName") or event.get("aid", "<AID_NOT_FOUND>")
    return f"Crowdstrike: Reverse shell tool [{tool}] detected on host [{host}]"


def alert_context(event):
    return crowdstrike_detection_alert_context(event)

AnalysisType: rule
Description: Detects usage of tools commonly used to to establish reverse shells on Windows machines.
DisplayName: "Crowdstrike Reverse Shell Tool Executed"
Enabled: true
Filename: crowdstrike_reverse_shell_tool_executed.py
Reference: https://attack.mitre.org/techniques/T1059/
Severity: High
DedupPeriodMinutes: 60
LogTypes:
  - Crowdstrike.FDREvent
RuleID: "Crowdstrike.Reverse.Shell.Tool.Executed"
Threshold: 1

from panther_crowdstrike_event_streams_helpers import audit_keys_dict, cs_alert_context, str_to_list


def get_single_ips(event, fieldname="cidrs") -> list[str]:
    """Searches the "cidrs" field of the event audit keys, and returns any cidr entries which
    are actually just single IP addresses."""
    single_ips = []
    audit_keys = audit_keys_dict(event)
    cidrs = str_to_list(audit_keys.get(fieldname, []))
    for entry in cidrs:
        if "/" not in entry:
            single_ips.append(entry)
        elif entry.endswith("/32"):
            # A 32-bit CIDR range is the same as a single IP address
            single_ips.append(entry[:-3])
    return single_ips


def rule(event):
    # Only alert if an allow list is created or edited
    op_name = event.deep_get("event", "OperationName")
    if op_name not in ("CreateAllowlistGroup", "UpdateAllowlistGroup"):
        return False

    # Only alert if there's a single IP address allowed by the allowlist
    single_ips = get_single_ips(event)

    if op_name == "UpdateAllowlistGroup":
        # Remove IPs from single_ips if the weren't recently added
        old_single_ips = set(get_single_ips(event, "old_cidrs"))
        single_ips = [ip for ip in single_ips if ip not in old_single_ips]

    # Return true if there were any single IPs
    return len(single_ips) > 0


def title(event):
    # Title format: {actor} granted {contexts_str} access to {a, X} single ip{s}
    single_ips = get_single_ips(event)
    actor = event.deep_get("event", "UserId")

    # contexts_str: one of API, UI, or API & UI
    #   Also a more general case: API, UI, and XX (for if they add extra contexts in the future)
    contexts = str_to_list(audit_keys_dict(event).get("contexts", ""))
    if len(contexts) == 0:
        contexts_str = "no contexts"
    elif len(contexts) == 1:
        contexts_str = contexts[0]
    else:
        contexts_str = ", ".join(contexts[:-1]) + " & " + contexts[-1]

    num_ips_str = "a single ip" if len(contexts) == 1 else f"{len(single_ips)} single ips"

    return f"{actor} granted {contexts_str} access to {num_ips_str}"


def alert_context(event):
    context = cs_alert_context(event)
    context.update({"single_ips": get_single_ips(event)})
    return context

AnalysisType: rule
Filename: crowdstrike_single_ip_allowlisted.py
RuleID: "Crowdstrike.SingleIpAllowlisted"
DisplayName: "Crowdstrike Single IP Allowlisted"
Enabled: true
LogTypes:
  - Crowdstrike.EventStreams
Severity: Medium
Reports:
  MITRE ATT&CK:
    - TA0003:T1556.009 # Persistsnce: Modify Authentication Process: Conditional Access Policies
    - TA0005:T1556.009 # Defense Evasion: Modify Authentication Process: Conditional Access Policies
    - TA0006:T1556.009 # Credential Access: Modify Authentication Process: Conditional Access Policies
Description: A single IP (instead of a CIDR range) was allowlisted. This could indicate a bad actor permitting access from another machine.
DedupPeriodMinutes: 60
Threshold: 1
Runbook: Validate this action was authorized, and determine the client to which the IP belongs to.