# Writing Python Detections ## Overview You can write your own Python detections in the Panther Console or locally, following the [CLI workflow](https://docs.panther.com/panther-developer-workflows/detections-repo/ci-cd). When writing Python detections, note [these best practices](#python-detection-writing-best-practices), and remember that [certain alert fields can be set dynamically](#alert-functions-in-python-detections). Rules written in Python can be used in [detection derivation](https://docs.panther.com/detections/rules/derived). You can alternatively use the [no-code detection builder](https://docs.panther.com/detections/rules/simple-detection-builder) in the Console to create rules, or [write them locally as Simple Detections](https://docs.panther.com/detections/rules/writing-simple-detections). If you aren't sure whether to write detections locally as Simple Detections or Python detections, see the [Using Python vs. Simple Detections YAML](https://docs.panther.com/detections/rules/..#using-python-vs.-simple-detections-yaml) section. {% hint style="info" %} Before you write a new Python detection, see if there's a [Panther-managed detection](https://docs.panther.com/detections/panther-managed) that meets your needs (or *almost* meets your needs—Panther-managed rules can be tuned with [Inline Filters](https://docs.panther.com/detections/rules/inline-filters)). Leveraging a Panther-managed detection not only saves you the effort of writing one yourself, but also provides the ongoing benefit of continuous updates to core detection logic as Panther releases new versions. {% endhint %} {% hint style="warning" %} It is highly discouraged to make external API requests from within your detections in Panther. In general, detections are processed at a very high scale, and making API requests can overload receiving systems and cause your rules to exceed the [15-second runtime limit](https://docs.panther.com/detections/rules/..#rule-errors-and-scheduled-rule-errors). {% endhint %} ## How to create detections in Python ### How to create a rule in Python You can write a Python rule in both the Panther Console and CLI workflow.
Creating a rule in Python in the Console 1. In the left-hand navigation bar of your Panther Console, click **Detections**. 2. Click **Create New**. 3. On the **Python Rule** tile, click **Start**. 4. On the create page, configure your rule: * **Name**: Enter a descriptive name for the rule. * **ID** (optional)**:** Click the pen icon and enter a unique ID for your rule. * In the upper-right corner, the **Enabled** toggle will be set to `ON` by default. If you'd like to disable the rule, flip the toggle to `OFF`. * In the **For the Following Source** section: * **Log Types**: Select the log types this rule should apply to. * In the **Detect** section: * In the **Rule Function** text editor, write a Python `rule` function to define your detection. * For detection templates and examples, see the [panther\_analysis GitHub repository](https://github.com/panther-labs/panther-analysis/tree/master/templates). * In the **Create Alert** section, set the **Create Alert** `ON/OFF` toggle. This indicates whether an [alert](https://docs.panther.com/alerts) should be created when there are matches, or only a [Signal](https://docs.panther.com/detections/signals). If you set this toggle to `ON`: * **Severity**: Select a [severity level](#alert-severity) for the alerts triggered by this detection. * In the **Optional Fields** section, optionally provide values for the following fields: * **Description**: Enter additional context about the rule. * **Runbook**: Enter the procedures and operations relating to this rule. * Learn more on [Alert Runbooks](https://docs.panther.com/alerts/alert-runbooks). * It's recommended to provide a descriptive runbook, as [Panther AI alert triage](https://docs.panther.com/alerts#panther-ai-alert-triage) will take it into consideration. * **Reference**: Enter an external link to more information relating to this rule. * **Destination Overrides:** Choose destinations to receive alerts for this detection, regardless of severity. Note that destinations can also be set dynamically, in the rule function. See [Routing Order Precedence](https://docs.panther.com/alerts/destinations#routing-order-precedence) to learn more about routing precedence. * **Deduplication Period** and **Events Threshold**: Enter the deduplication period and threshold for rule matches. To learn how deduplication works, see [Deduplication](#deduplication). * **Summary Attributes**: Enter the attributes you want to showcase in the alerts that are triggered by this detection. * To use a nested field as a summary attribute, use the Snowflake dot notation in the Summary Attribute field to traverse a path in a JSON object: `:..` The alert summary will then be generated for the referenced object in the alert. [Learn more about traversing semi-structured data in Snowflake here.](https://docs.snowflake.com/en/user-guide/querying-semistructured.html#label-traversing-semistructured-data) * For more information on Alert Summaries, see [Assigning and Managing Alerts](https://docs.panther.com/alerts/alert-management). * **Custom Tags**: Enter custom tags to help you understand the rule at a glance (e.g., `HIPAA`.) * In the **Framework Mapping** section: 1. Click **Add New** to enter a report. 2. Provide values for the following fields: * **Report Key**: Enter a key relevant to your report. * **Report Values**: Enter values for that report. * In the **Test** section: * In the **Unit Test** section, click **Add New** to [create a test](https://docs.panther.com/detections/testing) for the rule you defined in the previous step. 5. In the upper-right corner, click **Save**. After you have created a rule, you can modify it using [Inline Filters](https://docs.panther.com/detections/rules/inline-filters).
Creating a rule in Python in the CLI workflow If you're writing detections locally (instead of in the Panther Console), we recommend managing your local detection files in a version control system like GitHub or GitLab. We advise that you start your custom detection content by creating either [a public fork](https://docs.panther.com/panther-developer-workflows/ci-cd/detections-repo/public-fork) or a [private cloned repo](https://docs.panther.com/panther-developer-workflows/ci-cd/detections-repo/private-cloned-repo) from Panther's [open-source panther-analysis repository](https://github.com/panther-labs/panther-analysis). **Folder setup** If you group your rules into folders, each folder name must contain `rules` in order for them to be found during upload (using either PAT or the bulk uploader in the Console). We recommend grouping rules into folders based on log/resource type, e.g., `suricata_rules` or `aws_s3_policies`. You can use the [panther-analysis](https://github.com/panther-labs/panther-analysis) repo as a reference. **File setup** Each rule and scheduled rule consists of: * A Python file (a file with a `.py` extension) containing your detection logic. * A YAML specification file (a file with a `.yml` extension) containing metadata attributes of the detection. * By convention, we give this file the same name as the Python file. Rules are Python functions to detect suspicious behaviors. Returning a value of `True` indicates suspicious activity, which triggers an alert. 1. Write your rule and save it (in your folder of choice) as `my_new_rule.py`: ```python def rule(event): return 'prod' in event.get('hostName') ``` 2. Create a metadata file using the template below: ```yaml AnalysisType: rule DedupPeriodMinutes: 60 # 1 hour DisplayName: Example Rule to Check the Format of the Spec Enabled: true Filename: my_new_rule.py RuleID: Type.Behavior.MoreContext Severity: High LogTypes: - LogType.GoesHere Reports: ReportName (like CIS, MITRE ATT&CK): - The specific report section relevant to this rule Tags: - Tags - Go - Here Description: > This rule exists to validate the CLI workflows of the Panther CLI Runbook: > First, find out who wrote this the spec format, then notify them with feedback. Reference: https://www.a-clickable-link-to-more-info.com ``` When this rule is uploaded, each of the fields you would normally populate in the Panther Console will be auto-filled. See [Rule specification reference](#python-rule-specification-reference) for a complete list of required and optional fields.
### How to create a scheduled rule in Python You can write a Python scheduled rule in both the Panther Console and CLI workflow.
Creating a scheduled rule in Python in the Console 1. In the left-hand navigation bar of your Panther Console, click **Detections**. 2. Click **Create New**. 3. On the **Scheduled Rule** tile, click **Start**. 4. On the create page, configure your scheduled rule: * **Name**: Enter a descriptive name for the scheduled rule. * **ID** (optional)**:** Click the pen icon and enter a unique ID for your scheduled rule. * In the upper-right corner, the **Enabled** toggle will be set to `ON` by default. If you'd like to disable the scheduled rule, flip the toggle to `OFF`. * In the **For the Following Scheduled Queries** section: * **Scheduled Queries**: Select one or more [Scheduled Searches](https://docs.panther.com/search/scheduled-searches) this scheduled rule should apply to. * In the **Detect** section: * In the **Rule Function** text editor, write a Python `rule` function to define your detection. * If all your filtering logic is already taken care of in the SQL of the associated [scheduled query](https://docs.panther.com/search/scheduled-searches), you can configure the `rule` function to simply return `true` for each row: ```python def rule(event): return True ``` * For detection templates and examples, see the [panther\_analysis GitHub repository](https://github.com/panther-labs/panther-analysis/tree/master/templates) * In the **Create Alert** section, set the **Create Alert** `ON/OFF` toggle. This indicates whether an [alert](https://docs.panther.com/alerts) should be created when there are matches, or only a [Signal](https://docs.panther.com/detections/signals). If you set this toggle to `ON`: * **Severity**: Select a [severity level](#alert-severity) for the alerts triggered by this detection. * In the **Optional Fields** section, optionally provide values for the following fields: * **Description**: Enter additional context about the rule. * **Runbook**: Enter the procedures and operations relating to this rule. * Learn more on [Alert Runbooks](https://docs.panther.com/alerts/alert-runbooks). * It's recommended to provide a descriptive runbook, as [Panther AI alert triage](https://docs.panther.com/alerts#panther-ai-alert-triage) will take it into consideration. * **Reference**: Enter an external link to more information relating to this rule. * **Destination Overrides:** Choose destinations to receive alerts for this detection, regardless of severity. Note that destinations can also be set dynamically, in the rule function. See [Routing Order Precedence](https://docs.panther.com/alerts/destinations#routing-order-precedence) to learn more about routing precedence. * **Deduplication Period** and **Events Threshold**: Enter the deduplication period and threshold for rule matches. To learn how deduplication works, see [Deduplication](#deduplication). * **Summary Attributes**: Enter the attributes you want to showcase in the alerts that are triggered by this detection. * To use a nested field as a summary attribute, use the Snowflake dot notation in the Summary Attribute field to traverse a path in a JSON object: `:..` The alert summary will then be generated for the referenced object in the alert. [Learn more about traversing semi-structured data in Snowflake here.](https://docs.snowflake.com/en/user-guide/querying-semistructured.html#label-traversing-semistructured-data) * For more information on Alert Summaries, see [Assigning and Managing Alerts](https://docs.panther.com/alerts/alert-management). * **Custom Tags**: Enter custom tags to help you understand the rule at a glance (e.g., `HIPAA`.) * In the **Framework Mapping** section: 1. Click **Add New** to enter a report. 2. Provide values for the following fields: * **Report Key**: Enter a key relevant to your report. * **Report Values**: Enter values for that report. * In the **Test** section: * In the **Unit Test** section, click **Add New** to [create a test](https://docs.panther.com/detections/testing) for the rule you defined in the previous step. 5. In the upper-right corner, click **Save**. * Once you've clicked **Save**, the scheduled rule will become active. The SQL returned from the associated [scheduled query](https://docs.panther.com/search/scheduled-searches) (at the interval defined in the query) will be run through the scheduled rule (if, that is, any rows are returned). After you have created a rule, you can modify it using [Inline Filters](https://docs.panther.com/detections/rules/inline-filters).
Creating a scheduled rule in Python in the CLI workflow If you're writing detections locally (instead of in the Panther Console), we recommend managing your local detection files in a version control system like GitHub or GitLab. We advise that you start your custom detection content by creating either [a public fork](https://docs.panther.com/panther-developer-workflows/ci-cd/detections-repo/public-fork) or a [private cloned repo](https://docs.panther.com/panther-developer-workflows/ci-cd/detections-repo/private-cloned-repo) from Panther's [open-source panther-analysis repository](https://github.com/panther-labs/panther-analysis). **Folder setup** If you group your rules into folders, each folder name must contain the string `rules` in order for them to be found during upload (using either PAT or the bulk uploader in the Console). We recommend grouping rules into folders based on log/resource type, e.g., `suricata_rules` or `aws_s3_policies`. You can use the [panther-analysis](https://github.com/panther-labs/panther-analysis) repo as a reference. **File setup** Each scheduled rule consists of: * A Python file (a file with a `.py` extension) containing your detection logic. * A YAML specification file (a file with a `.yml` extension) containing metadata attributes of the detection. * By convention, we give this file the same name as the Python file. Scheduled rules allow you to analyze the output of a [scheduled search](https://docs.panther.com/search/scheduled-searches) with Python. Returning a value of `True` indicates suspicious activity, which triggers an alert. 1. Write your query and save it as `my_new_scheduled_query.yml`: ```yaml AnalysisType: scheduled_query QueryName: My New Scheduled Query Name Enabled: true Tags: - Optional - Tags Description: > An optional Description Query: 'SELECT * FROM panther_logs.aws_cloudtrail LIMIT 10' Schedule: # Note: CronExpression and RateMinutes are mutually exclusive, only # configure one or the other CronExpression: '0 * * * *' RateMinutes: 1 TimeoutMinutes: 1 ``` 2. Write your rule and save it as `my_new_rule.py`: ```python # Note: See an example rule for more options # https://github.com/panther-labs/panther-analysis/blob/master/templates/example_rule.py def rule(_): # Note: You may add additional logic here return True ``` 3. Create a metadata file and save it as `my_new_schedule_rule.yml`: ```yaml AnalysisType: scheduled_rule Filename: my_new_rule.py RuleID: My.New.Rule DisplayName: A More Friendly Name Enabled: true ScheduledQueries: - My New Scheduled Query Name Tags: - Tag Severity: Medium Description: > An optional Description Runbook: > An optional Runbook Reference: An optional reference.link Tests: - Name: Name ExpectedResult: true Log: { "JSON": "string" } ``` When this scheduled rule is uploaded, each of the files will connect a scheduled query with a rule, and fill in the fields you would normally populate in the Panther Console will be auto-filled. See [Rule specification reference below](#python-rule-specification-reference) for a complete list of required and optional fields.
### How to create a policy in Python * To learn how to create a policy, see the [How to write a policy instructions on Policies](https://docs.panther.com/policies#how-to-write-a-policy). ## Python detection syntax A local Python detection is made up of two files: a Python file and a YAML file. When a Python detection is created in the Panther Console, there is only a Python text editor (not a YAML one). The keys listed in the YAML column, below, are set in fields in the user interface.
The Python file can contain:The YAML file can contain:

  • Detection logic

    def rule(event): # or def policy(resource):

  • Alert functions (dynamic)

    def severity(event):
def title(event):
def dedup(event):
def unique(event):
def destinations(event):
def runbook(event):
def reference(event):
def description(event):
def alert_context(event):

  • Filter key

    InlineFilters:

  • Metadata keys

    AnalysisType: # rule, scheduled_rule, or policy
Enabled: 
FileName: 
CreatedBy:
RuleID: # or PolicyId:
LogTypes: 
Reports: 
Tags: 
Tests: 
ScheduledQueries: # only applicable to scheduled rules
Suppressions: # only applicable to policies
CreateAlert: # not applicable to policies

  • Alert keys (static)

    Severity:
Description:
DedupPeriodMinutes:
Threshold: 
DisplayName:
OutputIds:
Reference:
Runbook:
SummaryAttributes:
### **Basic Python rule structure** Only a `rule()` function and the YAML keys shown below are required for a Python rule. Additional Python alert functions, however, can make your alerts more dynamic. Additional YAML keys are available, too—see [Python rule specification reference](#python-rule-specification-reference).
rule.pyrule.yml
def rule(event): 
    if event.get("Something"): 
        return True 
    return False

AnalysisType: rule
Enabled: true
Filename: rule.py
RuleID: my.rule
LogTypes: 
    - Some.Schema
Severity: INFO
For more templates, see the [panther-analysis repo on GitHub](https://github.com/panther-labs/panther-analysis/tree/master/templates). ### `InlineFilters` Learn more about using `InlineFilters` in Python rules on [Modifying Detections with Inline Filters](https://docs.panther.com/detections/inline-filters#creating-filters-in-the-developer-workflow). ### Alert functions in Python detections Panther's detection auxiliary functions are Python functions that control analysis logic, generated alert title, event grouping, routing of alerts, and metadata overrides. Rules are customizable and can import from standard Python libraries or [global helpers](https://docs.panther.com/detections/rules/python/globals). Applicable to both rules and policies, each function below takes a single argument of `event` (rules) or `resource` (policies). Advanced users may define functions, variables, or classes outside of the functions defined below. If you are using [alert deduplication](https://docs.panther.com/detections/rules/..#deduplication-of-alerts), the *first* event to match the detection is used as a parameter for these alert functions. Each of the below alert functions are optional, but can add dynamic context to your alerts.
Alert functionDescriptionDefault valueReturn value
severityThe level of urgency of the alert

In YAML: Severity key

In Console: Severity field

INFO, LOW, MEDIUM, HIGH,CRITICAL, or DEFAULT
titleThe generated alert title

In YAML: DisplayName > RuleID or PolicyID

In Console: Name field > ID field

String
dedupThe string to group related events with, limited to 1000 characters

In Python/YAML: title() > DisplayName > RuleID or PolicyID

In Console: title() > Name field > ID field

String
uniqueThe value to track for unique threshold detectionN/A - Uses standard event count thresholdingString
alert_contextAdditional context to pass to the alert destination(s)Dict[String: Any]
runbookA list of instructions to follow once the alert is generated. It's recommended to provide a descriptive runbook, as Panther AI alert triage will take it into consideration.

In YAML: Runbook key

In Console: Runbook field

String
descriptionAn explanation about why the rule exists

In YAML: Description key

In Console: Description field

String
referenceA reference URL to an internal document or online resource about the rule

In YAML: Reference key

In Console: Reference field

String
destinationsThe ID(s) of the destination(s) to send alerts to.

In YAML: OutputIds key

In Console: Destination Overrides field

List[Destination Name/ID]
#### `severity` In some scenarios, you may need to upgrade or downgrade the severity level of an alert. The severity levels of an alert can be mapped to `INFO`, `LOW`, `MEDIUM`, `HIGH`, `CRITICAL`, or `DEFAULT`. Return `DEFAULT` to fall back to the statically defined rule severity. The severity string is case insensitive, meaning you can return, for example, `Critical` or `default`, depending on your style preferences. In the example below, if an API token has been created, a `HIGH` severity alert is generated—otherwise an `INFO` level alert is generated: ```python def severity(event): if event.get('eventType') == 'system.api_token.create': return "HIGH" return "INFO" ``` Reference: [Template Rule](https://github.com/panther-labs/panther-analysis/blob/master/templates/example_rule.py#L33) Example using `DEFAULT`: ```python def severity(event): if event.get('eventType') == 'system.api_token.create': return "HIGH" return "DEFAULT" ``` #### `title` The `title()` function is optional, but it is recommended to include it to provide additional context in an alert. In the example below, the log type, username, and a static string are sent to the alert destination. The function checks to see if the event is related the AWS.CloudTrail log type and, if so, returns the AWS Account Name. Learn more about how an alert title is set on [Rules and Scheduled Rules](https://docs.panther.com/detections/rules/..#title-of-associated-alerts). Example: ```python def title(event): # use unified data model field in title log_type = event.get("p_log_type") title_str = ( f"{log_type}: User [{event.udm('actor_user')}] has exceeded the failed logins threshold" ) if log_type == "AWS.CloudTrail": title_str += f" in [{lookup_aws_account_name(event.get('recipientAccountId'))}]" return title_str ``` Reference: [Template Rule](https://github.com/panther-labs/panther-analysis/blob/master/templates/example_rule.py#L15) #### `dedup` Deduplication is the process of grouping related events into a single alert to prevent receiving duplicate alerts. Events triggering the same detection that also share a deduplication string, within the deduplication period, are grouped together in a single alert. The `dedup` function is one way to define a deduplication string. It is limited to 1000 characters. Learn more about deduplication on [Rules and Scheduled Rules](https://docs.panther.com/detections/rules/..#deduplication). Example: ```python def dedup(event): user_identity = event.get("userIdentity", {}) if user_identity.get("type") == "AssumedRole": return helper_strip_role_session_id(user_identity.get("arn", "")) return user_identity.get("arn") ``` Reference: [AWS S3 Bucket Deleted Rule](https://github.com/panther-labs/panther-analysis/blob/4b6b79846fb4cb1596908fd31ce983c75a39baaa/aws_cloudtrail_rules/aws_s3_bucket_deleted.py) #### `unique` The `unique()` function enables unique value threshold detection, allowing your rule to alert based on the count of distinct values rather than total event count. This is useful for detecting distributed attacks, data exfiltration, and administrative abuse scenarios. Example detecting login brute force across multiple users: ```python def rule(event): return ( event.get("eventType") == "user.session.start" and event.get("outcome", {}).get("result") == "FAILURE" ) def unique(event): # Track unique usernames being targeted return event.get("actor", {}).get("alternateId", "unknown_user") ``` With a threshold of 5, this rule will alert when failed login attempts target 5 or more unique usernames within the deduplication period. The `unique()` function should always return a string. For multi-field uniqueness, concatenate the fields: ```python def unique(event): user = event.get("username", "") ip = event.get("sourceIP", "") return f"{user}@{ip}" ``` Learn more about unique value threshold detection, including additional examples, on [Unique value threshold detection](#unique-value-threshold-detection). #### **`destinations`** The `destinations()` function is a way to indicate which alert destination(s) alerts from the rule should be sent to. It supersedes all other alert destination configurations (i.e., it is [Scenario 1](https://docs.panther.com/alerts/destinations#scenario-1-dynamically-defined-destination-s-on-the-detection) in the [Alert routing scenarios](https://docs.panther.com/alerts/destinations#alert-routing-scenarios)). The `destinations()` function should return a list of one or more alert destination names or UUIDs. If the list returned by the `destinations()` function is empty (`[]`), the alert will not be routed to any destination. For example, the rule below is associated with multiple log types. If the log type is `AWS.CloudTrail`, `destinations()` routes alerts to a the `slack-security-alerts` destination. If the log type is not `AWS.CloudTrail`, no alerts to external destinations are sent—indicated by `return []`. ```python def destinations(event): if event.get("p_log_type") == "AWS.CloudTrail": return ["slack-security-alerts"] # Name or UUID of destination # Do not send alert to an external destination return [] ``` See another example in the [panther-analysis example\_rule](https://github.com/panther-labs/panther-analysis/blob/main/templates/example_rule.py#L66). #### `alert_context` This function allows the detection to pass any event details as additional context, such as usernames, IP addresses, or success/failure, to the alert destination(s). Values included in the alert context dictionary must be JSON-compliant. Examples of non-compliant values include Python's `nan`, `inf`, and `-inf`. Example: The code below returns all event data in the alert context. ```python def rule(event): return ( event.get("actionName") == "UPDATE_SAML_SETTINGS" and event.get("actionResult") == "SUCCEEDED" ) def alert_context(event): return { "user": event.udm("actor_user"), "ip": event.udm("source_ip") } ``` #### `runbook` The `runbook` function output should provide actionable investigation steps for triaging an alert generated by this detection. When [Panther AI triages an alert](https://docs.panther.com/alerts#panther-ai-alert-triage), it will read and autonomously execute a runbook. Learn more about how to write an effective runbook on [alert-runbooks](https://docs.panther.com/alerts/alert-runbooks "mention"). Example: ```python def runbook(event): user_arn = event.deep_get("userIdentity", "arn", default="this user") source_ip = event.deep_get("sourceIPAddress", default="this IP address") return f""" 1. Find all API calls by {user_arn} in the 24 hours before the alert 2. Check if the source IP {source_ip} is associated with known cloud provider IP ranges or VPN endpoints 3. Look for other alerts from {user_arn} or {source_ip} in the past 7 days """ ``` #### `reference` and `description` The `reference` and `description` functions can provide additional context on why an alert was triggered and how to resolve the related issue. The example below dynamically provides a link within the `reference` field in an alert: ```python def reference(event): log_type = event.get("p_log_type") if log_type == "OnePassword.SignInAttempt": return: f"" elif log_type == "Okta.SystemLog": return: f"" else: return: f"" ``` ## Event object functions In a Python detection, the `rule()` function and all [dynamic alert functions](#alert-functions-in-python-detections) take in a single argument: the `event` object. This event object has built-in functions to enable simple extraction of event values. ### `get()` {% code title="Function signature" %} ```python def get(self, key, default=None) -> Any: ``` {% endcode %} Use `get()` to access a top-level event field. You can provide a default value that will be returned if the key is not found. It is also possible to access a top-level field using [`deep_get()`](#deep_get) and [`deep_walk()`](#deep_walk). Learn more about [accessing top-level fields safely below](#accessing-top-level-fields-safely). Example: {% code title="Example event" %} ```json { "key": "value" } ``` {% endcode %} {% code title="Using get()" %} ```python def rule(event): return event.get("key") == "value" # The above would return true ``` {% endcode %} ### `deep_get()` {% code title="Function signature" %} ```python def deep_get(self, *keys: str, default: Any = None) -> Any: ``` {% endcode %} Use `deep_get()` to return keys that are nested within Python dictionaries. If the value you need to retrieve lives within a list, use [`deep_walk()`](#deep_walk) instead. {% hint style="info" %} This function is [also represented as a global helper](https://docs.panther.com/detections/rules/globals#deep_get), but for convenience it is recommended to use this event object function. {% endhint %} Example: Given an event with the following structure {% code title="Example event" %} ```json { "object": { "nested": { "key": "here" } } } ``` {% endcode %} {% code title="Using deep\_get()" %} ```python def rule(event): return event.deep_get("object", "nested", "key") == "here" # The above would return true ``` {% endcode %} ### `deep_walk()` {% code title="Function signature" %} ```python def deep_walk( self, *keys: str, default: Optional[str] = None, return_val: str = "all" ) -> Union[Optional[Any], Optional[List[Any]]]: ``` {% endcode %} Use `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. {% hint style="info" %} This function is [also represented as a global helper](https://docs.panther.com/detections/rules/globals#deep_walk), but for convenience it is recommended to use this event object function. {% endhint %} Example: {% code title="Example event" %} ```json { "object": { "nested": { "list": [ { "key": "first" }, { "key": "second" } ] } } } ``` {% endcode %} {% code title="Using deep\_walk()" %} ```python def rule(event): return "first" in event.deep_walk("object", "nested", "list", "key", default=[]) # The above would return true ``` {% endcode %} ### `lookup()` {% code title="Function signature" %} ```python def lookup(self, lookup_table_name: str, lookup_key: str) -> Any: ``` {% endcode %} The `lookup()` function lets you dynamically access data from [Custom Lookup Tables](https://docs.panther.com/enrichment) and [Panther-managed Enrichment providers](https://docs.panther.com/enrichment) from your detections. The `lookup()` function may be useful if your incoming logs do not contain an exact match for a value in your Lookup Table's primary key column. You can use Python to modify an event value before passing it into `lookup()` to fetch enrichment data. `lookup()` takes two arguments: * The name of the Lookup Table * The Lookup Table name passed to `lookup()` must be as it appears on the **Enrichment Providers** or **Lookup Tables** pages in the Panther Console. This name may differ syntactically from how it appears in a search query; for example, `My-Custom-LUT` instead of `my_custom_lut`. * A Lookup Table primary key If a match is found in the Lookup Table for the provided key, the full Lookup Table row is returned as a Python dictionary. If no match is found, `None` is returned. {% hint style="info" %} This `lookup()` function is different from "automatic" event enrichment, which happens when the value of an event field designated as a Selector exactly matches a value in the Lookup Table's primary key column. In that case, the Lookup Table data is appended to an event's `p_enrichment` field. Learn more in [How is data matched between logs and Lookup Tables? on Custom Lookup Tables](https://docs.panther.com/enrichment/custom#how-is-data-matched-between-logs-and-lookup-tables). If you are using "automatic" enrichment in this fashion, access nested enrichment data using [`deep_walk()`](#deep_walk) instead. {% endhint %} Example using `lookup()`: ```python # Imagine you have a Lookup Table named user_roles with the following entries: # row 1: {"id": "your@email.com", "role": "admin"} # row 2: {"id": "vistor@email.com", "role": "guest"} # In this rule, we want to return True if the user has a non-admin role # We want to fetch role from the Lookup Table, but the event doesn't # contain the Lookup Table's primary key (the email address) def rule(event): lookup_table_name = "user_roles" # On the event, we *do* have access to the username, from which we can # generate the email address user_name = event.get("username", "").lower() lookup_key = f"{user_name}@email.com" # Dynamically compose the lookup key, or "Selector" lookup_data = event.lookup(lookup_table_name, lookup_key) # If a match occurs, `lookup_data` will contain the full row of data # Otherwise it will return None if (lookup_data and lookup_data.get("role") != "admin") or lookup_data is None: return True return False ``` #### Unit testing detections that use `lookup()` When [unit tests](https://docs.panther.com/detections/testing) are run, `lookup()` does not retrieve live data. To emulate the lookup functionality, add a `_mocked_lookup_data_` field in the event payload of each unit test to mock the Lookup Table data. You cannot use the [enrich test data button or CLI command](https://docs.panther.com/testing#enrich-test-data) with `lookup()`. `_mocked_lookup_data_` should be structured like the following example: {% code title="Mocking event.lookup()" %} ```json5 { "_mocked_lookup_data_": { "user_roles": { # This key is the name of your Lookup Table # The keys in this object should be the Lookup Table key # The values in this object should be the Lookup Table data object "your@email.com": {"id": "your@email.com", "role": "admin"}, "vistor@email.com": {"id": "vistor@email.com", "role": "guest"} } } } ``` {% endcode %} If you do not specify a `_mocked_lookup_data_` field in your unit test, attempts to call `lookup()` will return `None/null`. ### `udm()` {% code title="Function signature" %} ```python def udm(self, *key: str, default: Any = None) -> Any: ``` {% endcode %} The `udm()` function is primarily intended to allow you to access [Data Model](https://docs.panther.com/detections/rules/python/data-models), but can also be used to access event fields. Here is how the `udm()` function works: 1. The function first checks to see if there is a [Data Model](https://docs.panther.com/detections/rules/python/data-models) key mapping defined for the value passed in to `udm()`. If so, the value of the Data Model is returned. * If a [Data Model](https://docs.panther.com/detections/rules/python/data-models) key is defined for the value passed in to `udm()`, the function returns its value and does not move on to step 2, below. This is true even if the event being evaluated does not contain the key path defined in the Data Model mapping—in this case, `null` is returned. 2. If there is no [Data Model](https://docs.panther.com/detections/rules/python/data-models) defined for the value passed into `udm()`, the function then checks whether there is an event field with that name. If so, its value is returned. * In this case, `udm()` checks all event fields, even nested ones. Its behavior is analogous to [`deep_get()`](#deep_get). {% hint style="info" %} The behavior outlined above means it is only possible to use `udm()` to access an event field value if there is not also a [Data Model](https://docs.panther.com/detections/rules/python/data-models) mapping defined with the same key. {% endhint %} {% code title="Sample udm usage" %} ```python # Example usage when operating on data models def rule(event): return event.udm('field_on_data_model') # Example usage when operating on data models with default def rule(event): # The default parameter is only respected when using path-based mappings on # your data model. If your data model maps to a function, whatever value your # function returns will be respected return event.udm('field_on_data_model', default='') ``` {% endcode %} #### Example using `udm()` to access a [Data Model](https://docs.panther.com/detections/rules/python/data-models) value: {% code title="Data Model example" %} ```yaml Mappings: - Name: source_ip Path: nested.srcIp ``` {% endcode %} {% code title="Example event" %} ```json { "nested": { "srcIp": "127.0.0.1" } } ``` {% endcode %} {% code title="Using udm()" %} ```python def rule(event): return event.udm("source_ip") == "127.0.0.1" # The above would return true ``` {% endcode %} ## Unique value threshold detection Panther supports unique value threshold detection, which allows rules to alert based on the number of unique values observed (e.g., "alert when 10+ unique IP addresses are seen") rather than just total event count. This is useful for detecting scenarios like: * Multiple login attempts from different IP addresses * Brute force attacks targeting multiple usernames * Data exfiltration to various external domains * Administrative actions affecting multiple resources ### The `unique()` function To enable unique value threshold detection, define a `unique()` function in your Python rule that returns a string value representing what should be counted as unique. {% code title="Function signature" %} ```python def unique(event) -> str: ``` {% endcode %} The `unique()` function should return a string that identifies the unique value to track. Common examples include IP addresses, usernames, domain names, or resource IDs. #### Example: Multiple IP login detection ```python def rule(event): # Check if this is a successful login event return ( event.get("eventName") == "ConsoleLogin" and event.deep_get("responseElements", "ConsoleLogin") == "Success" ) def unique(event): # Track unique source IP addresses return event.get("sourceIPAddress") # With threshold=5, this will alert when 5+ unique IPs # have successful logins within the deduplication period ``` #### Example: Multi-target brute force detection ```python def rule(event): # Check for failed login attempts return ( event.get("eventType") == "user.session.start" and event.get("outcome", {}).get("result") == "FAILURE" ) def unique(event): # Track unique target usernames being attacked return event.get("actor", {}).get("alternateId") # With threshold=10, this will alert when 10+ different users # experience failed login attempts within the deduplication period ``` #### Example: Data exfiltration detection ```python def rule(event): # Look for file downloads from cloud storage return ( event.get("eventName") == "GetObject" and event.get("responseElements", {}).get("x-amz-request-charged") != "requester" ) def unique(event): # Track unique files being accessed bucket = event.get("requestParameters", {}).get("bucketName", "") key = event.get("requestParameters", {}).get("key", "") return f"{bucket}/{key}" # With threshold=50, this will alert when 50+ unique files # are downloaded within the deduplication period ``` ### How unique value thresholding works When you define a `unique()` function: 1. **Value extraction**: For each matching event, Panther calls your `unique()` function to extract the unique value to track 2. **Probabilistic counting**: Panther uses a memory-efficient probabilistic algorithm to estimate the number of unique values observed 3. **Threshold checking**: When the estimated unique count reaches your configured threshold, an alert is generated 4. **Deduplication**: The unique count resets based on your rule's deduplication period ### Configuration requirements To use unique value thresholding: * Define a `unique()` function that returns a string * Set your rule's **Threshold** to the minimum number of unique values needed to trigger an alert (default: 1) * Configure an appropriate **Deduplication Period** for your use case (default: 1 hour) ### Performance considerations * **Accuracy**: The unique counting algorithm provides \~90% accuracy for cardinalities up to 1,000 unique values * **Memory efficiency**: Uses approximately 16KB of storage per alert regardless of the number of unique values * **Recommended limits**: Best performance with thresholds under 1,000 unique values * **Concatenation**: For multi-field uniqueness, concatenate fields in your `unique()` function: `return f"{ip}:{username}"` ### Backward compatibility Rules without a `unique()` function continue to use standard event count thresholding. Adding or removing the `unique()` function from an existing rule will change its alerting behavior. ## Python best practices Python Enhancement Proposals [publishes resources](https://peps.python.org/pep-0008/) on how to cleanly and effectively write and style your Python code. For example, you can use [autopep8](https://pypi.org/project/autopep8/) to automatically ensure that your written detections all follow a consistent style. ### Available Python libraries The following Python libraries are available to be used in Panther in addition to `boto3`, provided by [AWS Lambda](https://docs.aws.amazon.com/lambda/latest/dg/lambda-runtimes.html):
PackageVersionDescriptionLicense
PackageVersionDescriptionLicense
jsonpath-ng1.5.2JSONPath ImplementationApache v2
policyuniverse1.3.3.20210223Parse AWS ARNs and PoliciesApache v2
requests2.23.0Easy HTTP RequestsApache v2
## Python detection writing best practices ### Writing tests for your detections Before enabling new detections, it is [recommended to write tests ](https://docs.panther.com/detections/testing)that define scenarios where alerts should or should not be generated. Best practice dictates at least one positive and one negative to ensure the most reliability. ### Casing for event fields Lookups for event fields are not case sensitive. `event.get("Event_Type")` or `event.get("event_type")` will return the same result. ### Understanding top level fields and nested fields Top-level fields represent the parent fields in a nested data structure. For example, a record may contain a field called `user` under which there are other fields such as `ip_address`. In this case, `user` is the top-level field, and `ip_address` is a nested field underneath it. Nesting can occur many layers deep, and so it is valuable to understand the schema structure and know how to access a given field for a detection. ### **Accessing top-level fields safely** Basic Rules match a field’s value in the event, and a best practice to avoid errors is to leverage Python’s built-in [`get()`](#get) function. The example below is a best practice because it leverages a `get()` function. `get()` will look for a field, and if the field doesn't exist, it will return `None` instead of an error, which will result in the detection returning `False`. ```python def rule(event): return event.get('field') == 'value' ``` In the example below, if the field exists, the value of the field will be returned. Otherwise, `False` will be returned: ```python def rule(event): if event.get('field') return event.get('field') return False ``` {% hint style="warning" %} **Bad practice example**\ The rule definition below is bad practice because the code is explicit about the field name. If the field doesn't exist, Python will throw a `KeyError`: ```python def rule(event): return event['field'] == 'value' ``` {% endhint %} ### Using Global Helper functions Once many detections are written, a set of patterns and repeated code will begin to emerge. This is a great use case for [Global Helper functions](https://docs.panther.com/detections/rules/python/globals), which provide a centralized location for this logic to exist across all detections. ### Accessing nested fields safely If you'd like to access a filed nested deeply within an event, use the [`deep_get()`](#deep_get) and [`deep_walk()`](#deep_walk) functions available on the event object. These functions are also represented as [Global Helper functions](https://docs.panther.com/detections/rules/python/globals), but for convenience, it's recommended to use the event object version instead. Example: AWS CloudTrail logs nest the `type` of user accessing the console underneath `userIdentity`. Here is a snippet of a JSON CloudTrail root activity log: ```json { "eventVersion": "1.05", "userIdentity": { "type": "Root", "principalId": "1111", "arn": "arn:aws:iam::123456789012:root", "accountId": "123456789012", "userName": "root" }, ... } ``` See how to check the value of `type` safely using both forms of `deep_get()`: {% tabs %} {% tab title="deep\_get() event object function" %} Checking the event value using the [event object `deep_get()` function](#deep_get): ```python def rule(event): return event.deep_get("userIdentity", "type") == "Root" ``` {% endtab %} {% tab title="deep\_get() Global Helper function" %} Checking the event value using the [`deep_get()` Global Helper function](https://docs.panther.com/detections/rules/globals#deep_get): ```python from panther_base_helpers import deep_get def rule(event): return deep_get(event, "userIdentity", "type") == "Root" ``` {% endtab %} {% endtabs %} ### **Checking fields for specific values** You may want to know when a specific event has occurred. If it did occur, then the detection should trigger an alert. Since Panther stores everything as normalized JSON, you can check the value of a field against the criteria you specify. For example, to detect the action of granting Box technical support access to your Box account, the Python below would be used to match events where the `event_type` equals `ACCESS_GRANTED`: ```python def rule(event): return event.get("event_type") == "ACCESS_GRANTED" ``` If the field is `event_type` and the value is equal to `ACCESS_GRANTED` then the rule function will return `true` and an Alert will be created. ### Checking fields for Integer values You may need to compare the value of a field against integers. This allows you to use any of Python’s built-in comparisons against your events. For example, you can create an alert based on HTTP response status codes: ```python # returns True if 'status_code' equals 404 def rule(event): if event.get("status_code"): return event.get("status_code") == 404 else: return False # returns True if 'status_code' greater than 400 def rule(event): if event.get("status_code"): return event.get("status_code") > 404 else: return False ``` Reference: * [box\_access\_granted.py](https://github.com/panther-labs/panther-analysis/blob/cd220c87982011d4ad156c7daecd2857c358d154/rules/box_rules/box_access_granted.py) * [Python Operators](https://www.w3schools.com/python/python_operators.asp) ### **Using the Universal Data Model** [Data Models](https://docs.panther.com/detections/rules/python/data-models) provide a way to configure a set of unified fields across all log types. By default, Panther comes with built-in Data Models for several log types. Custom Data Models can be added in the Panther Console or via the [Panther Analysis Tool](https://docs.panther.com/writing-detections/panther-analysis-tool#data-models). [`event.udm()`](#udm) can only be used with log types that have an existing Data Model in your Panther environment. Example: ```python import panther_event_type_helpers as event_type def rule(event): # filter events on unified data model field ‘event_type’ return event.udm("event_type") == event_type.FAILED_LOGIN ``` References: * [Data Models Guide](https://docs.panther.com/writing-detections/data-models) * [Data Models](https://github.com/panther-labs/panther-analysis/tree/master/data_models) * [Brute Force by IP](https://github.com/panther-labs/panther-analysis/blob/cd220c87982011d4ad156c7daecd2857c358d154/indexes/standard.md) ### Using multiple conditions The `and` keyword is a logical operator and is used to combine conditional statements. It is often required to match multiple fields in an event using the `and` keyword. When using `and`, all statements must be true:\ `"string_a" == "this"`**`and`**`"string_b" == "that"` Example: To track down successful root user access to the AWS console you need to look at several fields: ```python from panther_base_helpers import deep_get def rule(event): return (event.get("eventName") == "ConsoleLogin" and deep_get(event, "userIdentity", "type") == "Root" and deep_get(event, "responseElements", "ConsoleLogin") == "Success") ``` The `or` keyword is a logical operator and is used to combine conditional statements. When using `or`, either of the statements may be true:\ "` string_a" == "this"`` `` `**`or`**` `` ``"string_b" == "that" ` Example: This example detects if the field contains either Port 80 **or** Port 22: ```python # returns True if 'port_number' is 80 or 22 def rule(event): return event.get("port_number") == 80 or event.get("port_number") == 22 ``` ### Searching values in lists Comparing event values against a list (containing, for example, IP addresses or users) is quick in Python. It's a common pattern to set your rule logic to not match when an event value also exists in the list. This can help reduce false positives for known behavior in your environment. When checking whether event values are in some collection, it's recommended to use a Python set—sets are more performant (i.e., memory efficient) than lists and tuples in Python. Lists and tuples, unlike sets, require iterating through each item in the collection to check for inclusion. If the set against which you're performing the comparison is static, it's recommended to define it at the global level, rather than inside the `rule()` function. Global variables are initialized only once per Lambda invocation. Because a single Lambda invocation can process multiple events, a global variable is usually more efficient than initializing it each time `rule()` is invoked. Example: ```python # Set - Recommended over tuples and lists for performance ALLOW_IP = {'192.0.0.1', '192.0.0.2', '192.0.0.3'} def rule(event): return event.get("ip_address") not in ALLOW_IP ``` In the example below, we use the Panther helper `pattern_match_list`: ```python from panther_base_helpers import pattern_match_list USER_CREATE_PATTERNS = [ "chage", # user password expiry "passwd", # change passwords for users "user*", # create, modify, and delete users ] def rule(event): # Filter the events if event.get("event") != "session.command": return False # Check that the program matches our list above return pattern_match_list(event.get("program", ""), USER_CREATE_PATTERNS) ``` Reference: [Teleport Create User Accounts](https://github.com/panther-labs/panther-analysis/blob/cd220c87982011d4ad156c7daecd2857c358d154/rules/gravitational_teleport_rules/teleport_create_user_accounts.py) ### Matching events with regex If you want to match against events using regular expressions - to match subdomains, file paths, or a prefix/suffix of a general string - you can use regex. In Python, regex can be used by importing the `re` library and looking for a matching value. In the example below, the regex pattern will match Administrator or administrator against the nested value of the privilegeGranted field. ```python import re from panther_base_helpers import deep_get #The regex pattern is stored in a variable # Note: This is better performance than putting it in the rule function, which is evaluated on each event ADMIN_PATTERN = re.compile(r"[aA]dministrator") def rule(event): # using the deep_get function we can pull out the nested value under the "privilegeGranted" field value_to_search = deep_get(event, "debugContext", "debugData", "privilegeGranted") # finally we use the regex object we created earlier to check against our value # if there is a match, "True" is returned return (bool(ADMIN_PATTERN.search(value_to_search, default=""))) ``` In the example below, we use the Panther helper `pattern_match`: ```python from panther_base_helpers import pattern_match def rule(event): return pattern_match(event.get("operation", ""), "REST.*.OBJECT") ``` References: * [re.compile](https://docs.python.org/3/library/re.html#functions) * [Pythex: simple RegEx editor and tester](https://pythex.org/) * [AWS S3 Insecure Access](https://github.com/panther-labs/panther-analysis/blob/cd220c87982011d4ad156c7daecd2857c358d154/rules/aws_s3_rules/aws_s3_insecure_access.py) ## Python rule specification reference Required fields are in **bold**.
Field NameDescriptionExpected Value
Field NameDescriptionExpected Value
AnalysisTypeIndicates whether this analysis is a rule, scheduled_rule, policy, or globalRules: rule
Scheduled Rules: scheduled_rule
EnabledWhether this rule is enabledBoolean
FileNameThe path (with file extension) to the python rule bodyString
RuleIDThe unique identifier of the ruleString
Cannot include %
LogTypesThe list of logs to apply this rule toList of strings
SeverityWhat severity this rule isOne of the following strings: Info, Low, Medium, High, or Critical
ScheduledQueries (field only for Scheduled Rules)The list of Scheduled Query names to apply this rule toList of strings
CreateAlertWhether the rule should generate rule matches/an alert on matches (default true)Boolean
DescriptionA brief description of the ruleString
DedupPeriodMinutesThe time period (in minutes) during which similar events of an alert will be grouped together15,30,60,180 (3 hours),720 (12 hours), or 1440 (24 hours)
DisplayNameA friendly name to show in the UI and alerts. The RuleID will be displayed if this field is not set.String
OutputIdsStatic destination overrides. These will be used to determine how alerts from this rule are routed, taking priority over default routing based on severity.List of strings
ReferenceThe reason this rule exists, often a link to documentationString
ReportsA mapping of framework or report names to values this rule covers for that frameworkMap of strings to list of strings
RunbookActions an analyst or Panther AI can take to triage associated alerts.String
SummaryAttributesA list of fields that alerts should summarize.List of strings
ThresholdHow many events need to trigger this rule before an alert will be sent.Integer
TagsTags used to categorize this ruleList of strings
TestsUnit tests for this rule.List of maps
CreatedByThe author of this detection. Can be set as a Panther user UUID, email address, or an arbitrary text value. Learn more in The CreatedBy detection field.String
## Python Policy Specification Reference Required fields are in **bold**. A complete list of policy specification fields:
Field NameDescriptionExpected Value
Field NameDescriptionExpected Value
AnalysisTypeIndicates whether this specification is defining a policy or a rulepolicy
EnabledWhether this policy is enabledBoolean
FileNameThe path (with file extension) to the python policy bodyString
PolicyIDThe unique identifier of the policyString
Cannot include %
ResourceTypesWhat resource types this policy will apply toList of strings
SeverityWhat severity this policy isOne of the following strings: Info, Low, Medium, High, or Critical
DescriptionA brief description of the policyString
DisplayNameWhat name to display in the UI and alerts. The PolicyID will be displayed if this field is not set.String
ReferenceThe reason this policy exists, often a link to documentationString
ReportsA mapping of framework or report names to values this policy covers for that frameworkMap of strings to list of strings
RunbookActions an analyst or Panther AI can take to triage associated alerts.String
SuppressionsPatterns to ignore, e.g., aws::s3::*List of strings
TagsTags used to categorize this policyList of strings
TestsUnit tests for this policy.List of maps
CreatedByThe author of this detection. Can be set as a Panther user UUID, email address, or an arbitrary text value. Learn more in The CreatedBy detection field.String
## Troubleshooting Detections Visit the Panther Knowledge Base to [view articles about detections](https://help.panther.com/Detections) that answer frequently asked questions and help you resolve common errors and issues.