PyPanther Detections Style Guide

PyPanther Detections are in closed beta starting with Panther version 1.108. Please share any bug reports and feature requests with your Panther support team.

Repository structure recommendations

Get up and running quickly by cloning the pypanther-starter-kit repository.

In your code repository where your PyPanther Detections are stored, it's recommended to:

• Maintain a top-level module, content, in which all of your custom Python code is stored (except for the main.py file).

  The top-level directory we are calling content/ can be named anything except src/, which is a reserved repository name in Panther.

  Within this folder, it's recommended to:

  • Store custom rule definitions in a rules directory.

  • Store logic that makes overrides on Panther-managed rules in an overrides directory.

    • Define an apply_overrides function in each file in overrides.

  • Store custom helpers in a helpers directory.

# Recommended repository structure
.
├── README.md
├── content
│   ├── __init__.py
│   ├── helpers
│   │   ├── __init__.py
│   │   ├── cloud.py
│   │   └── custom_log_types.py
│   ├── overrides
│   │   ├── __init__.py
│   │   ├── aws_cloudtrail.py└
│   │   └── aws_guardduty.py
│   ├── rules
│   │   ├── __init__.py
│   │   ├── my_custom_rule.py
│   │   └── my_inherited_rule.py
│   └── schemas
│       └── schema1.yml
└── main.py

main.py content recommendations

It's recommended for your main.py file to:

• Import Panther-managed rules (which you do or don't want to make overrides on) using get_panther_rules

  • If you defined apply_overrides functions, call pypanther's apply_overrides() to apply all of your changes. Learn more in Call apply_overrides(), below.
• Import custom rules using get_rules

# Example main.py file
from pypanther import get_panther_rules, get_rules, register, apply_overrides

from content import rules, overrides
from content.helpers.custom_log_types import CustomLogType

# Load base rules
base_rules = get_panther_rules(
    # log_types=[
    #     LogType.AWS_CLOUDTRAIL,
    #     LogType.AWS_GUARDDUTY,
    #     LogType.PANTHER_AUDIT,
    # ],
    # default_severity=[
    #     Severity.CRITICAL,
    #     Severity.HIGH,
    # ],
)
# Load all local custom rules
custom_rules = get_rules(module=rules)
# Omit rules with custom log types, since they must be present in the Panther instance for upload to work
custom_rules = [rule for rule in custom_rules if not any(custom in rule.log_types for custom in CustomLogType)]

# Apply overrides
apply_overrides(module=overrides, rules=base_rules)

# Register all rules
register(base_rules + custom_rules)

Call apply_overrides()

The pypanther apply_overrides() convenience function lets you, in main.py, efficiently apply all detection overrides made in a separate file or folder.

apply_overrides() takes an imported package (folder) or module (file) name and an optional list of rules, and runs all functions named apply_overrides() from that package or module against the list of rules. (This is similar to how get_rules() works.) It's recommended to name this package or module overrides.

Each folder containing rules imported with apply_overrides() must contain an __init__.py file.

In the following example, the apply_overrides() functions in general.py and aws_cloudtrail.py are applied when apply_overrides(overrides, all_rules) is called in main.py:

# main.py
import overrides

all_rules = get_panther_rules()
apply_overrides(overrides, all_rules)

register(all_rules)

# general.py in /overrides
def apply_overrides(rules):
    for rule in rules:
        rule.override(enabled=True)

# aws_cloudtrail.py in /overrides
from pypanther import Severity
from pypanther.rules.aws_cloudtrail import AWSCloudTrailStopped

def apply_overrides(rules):
    AWSCloudTrailStopped.override(
        default_severity=Severity.LOW,
        default_runbook=(
            "If the account is in production, investigate why CloudTrail was stopped. "
            "If it was intentional, ensure that the account is monitored by another CloudTrail. "
            "If it was not intentional, investigate the account for unauthorized access."
        ),
    )

Best practices for PyPanther Detection writing

Use filters instead of overriding rule()

If you would like to alter the logic of a Panther-managed PyPanther Detection, it's recommended to use include/exclude filters instead of overriding the rule's rule() function. Filters are designed for this purpose—to be applied on top of existing rule logic. They are executed against each incoming event before the rule() logic, in order to determine if the rule should indeed process the event.

If you are significantly altering the rule logic, you might also consider writing a custom rule instead.

Use upgrade() or downgrade() in severity()

While each PyPanther rule must define a default_severity, it can also define a severity() function, whose value overrides default_severity to set the severity level of resulting alerts.

Within severity(), it's common practice to dynamically set the severity based on an event field value. One way to do this is to add some condition, which, if satisfied, returns a hard-coded Severity value. In the example below, if the actor associated to the event has an admin role, Severity.MEDIUM is returned, regardless of the value of self.default_severity:

# Example NOT using upgrade() or downgrade()
class MyRule(Rule):
    ...
    default_severity = Severity.LOW
    ...

def severity(self, event):
    if event.deep_get("actor", "role") == "admin":
        return Severity.MEDIUM                     # returns MEDIUM
    return self.default_severity                   # returns LOW

Note that you can reference the default_severity value with self.default_severity.

In the example above, if MyRule's default_severity was ever changed from Severity.LOW to, say, Severity.MEDIUM, you would also need to remember to update the Severity.MEDIUM in severity() (presumably to Severity.HIGH), to preserve the escalation.

Using upgrade() instead of hard-coding a Severity

Instead of setting the return value of severity() as a hard-coded Severity value (as is shown in the example above), it's recommended to call the Severity class's upgrade() and downgrade() functions on self.default_severity.

In this model, if your detection's default_severity value ever changes, you won't also need to make changes in the severity() function. In the example below, default_severity has been changed to Severity.MEDIUM, and the return value within the condition automatically adjusts to returning Severity.HIGH because it uses upgrade():

# Example using upgrade()
class MyRule(Rule):
    ...
    default_severity = Severity.MEDIUM
    ...

def severity(self, event):
    if event.deep_get("actor", "role") == "admin":
        return self.default_severity.upgrade()        # returns HIGH
    return self.default_severity                      # returns MEDIUM

In order to use self.default_severity.upgrade() or self.default_severity.downgrade(), the detection's default_severity value must be a Severity object, not a string literal.

There may be rare instances when using a static value within severity() is preferable—for example, you may always want to return Severity.INFO when an event originates in a dev account, even if the default_severity later changes. This might look like:

def severity(self, event):
    if is_dev_env(event.get("accountId")):
        return Severity.INFO
    return self.default_severity

Optionally use explicit typing

The pypanther base Rule class is typed, so when you create a custom detection (i.e., inherit from Rule), explicit typing is optional. Still, if you prefer to use explicit types, you can do so.

Example of a custom rule with explicit typing

from time import strptime
from typing import Dict, List

from panther_core.enriched_event import PantherEvent
from pypanther import LogType, Rule, RuleTest, Severity
from pypanther.base import SeverityType
from pypanther.helpers.aws import aws_guardduty_context
from pypanther.severity import SEVERITY_DEFAULT


class MyTypedRule(Rule):
    log_types: List[LogType | str] = [LogType.AWS_GUARDDUTY]
    id: str = "AWS.GuardDuty.HighVolFindings"
    create_alert: bool = True
    dedup_period_minutes = 45
    display_name: str = "High volume of GuardDuty findings"
    enabled: bool = True
    threshold = 100
    tags: List[str] = ["GuardDuty", "Security"]
    reports: Dict[str, List[str]] = {"MITRE ATT&CK": ["TA0010:T1499"]}

    default_severity: Severity | str = Severity.HIGH
    default_destinations: List[str] = ["slack:my-channel"]
    default_description: str = "This rule tracks high volumes of GuardDuty findings"

    def rule(self, event: PantherEvent) -> bool:
        if event.deep_get("service", "additionalInfo", "sample"):
            # in case of sample data
            # https://docs.aws.amazon.com/guardduty/latest/ug/sample_findings.html
            return False
        return 7.0 <= float(event.get("severity", 0)) <= 8.9

    def title(self, event: PantherEvent) -> str:
        return event.get("title", "GuardDuty finding")

    def severity(self, event: PantherEvent) -> SeverityType:
        # Parse timestamp: "createdAt": "2020-02-14T18:12:22.316Z"
        timestamp = strptime(event.get("createdAt", "1970-01-01T00:00:00Z"), "%Y-%m-%dT%H:%M:%S.%fZ")
        # Increase severity if it's the weekend
        if timestamp.tm_wday in (5, 6):
            return Severity.CRITICAL
        return SEVERITY_DEFAULT

    def alert_context(self, event: PantherEvent) -> dict:
        return aws_guardduty_context(event)

    tests: List[RuleTest] = [
        RuleTest(
            name="High Sev Finding",
            expected_result=True,
            log={
                "schemaVersion": "2.0",
                "accountId": "123456789012",
                "region": "us-east-1",
                "partition": "aws",
                "arn": "arn:aws:guardduty:us-west-2:123456789012:detector/111111bbbbbbbbbb5555555551111111/finding/90b82273685661b9318f078d0851fe9a",
                "type": "PrivilegeEscalation:IAMUser/AdministrativePermissions",
                "service": {
                    "serviceName": "guardduty",
                    "detectorId": "111111bbbbbbbbbb5555555551111111",
                    "action": {
                        "actionType": "AWS_API_CALL",
                        "awsApiCallAction": {
                            "api": "PutRolePolicy",
                            "serviceName": "iam.amazonaws.com",
                            "callerType": "Domain",
                            "domainDetails": {"domain": "cloudformation.amazonaws.com"},
                            "affectedResources": {"AWS::IAM::Role": "arn:aws:iam::123456789012:role/IAMRole"},
                        },
                    },
                    "resourceRole": "TARGET",
                    "additionalInfo": {},
                    "evidence": None,
                    "eventFirstSeen": "2020-02-14T17:59:17Z",
                    "eventLastSeen": "2020-02-14T17:59:17Z",
                    "archived": False,
                    "count": 1,
                },
                "severity": 8,
                "id": "eeb88ab56556eb7771b266670dddee5a",
                "createdAt": "2020-02-14T18:12:22.316Z",
                "updatedAt": "2020-02-14T18:12:22.316Z",
                "title": "Principal AssumedRole:IAMRole attempted to add a policy to themselves that is highly permissive.",
                "description": "Principal AssumedRole:IAMRole attempted to add a highly permissive policy to themselves.",
            },
        ),
        RuleTest(
            name="High Sev Finding As Sample Data",
            expected_result=False,
            log={
                "schemaVersion": "2.0",
                "accountId": "123456789012",
                "region": "us-east-1",
                "partition": "aws",
                "arn": "arn:aws:guardduty:us-west-2:123456789012:detector/111111bbbbbbbbbb5555555551111111/finding/90b82273685661b9318f078d0851fe9a",
                "type": "PrivilegeEscalation:IAMUser/AdministrativePermissions",
                "service": {
                    "serviceName": "guardduty",
                    "detectorId": "111111bbbbbbbbbb5555555551111111",
                    "action": {
                        "actionType": "AWS_API_CALL",
                        "awsApiCallAction": {
                            "api": "PutRolePolicy",
                            "serviceName": "iam.amazonaws.com",
                            "callerType": "Domain",
                            "domainDetails": {"domain": "cloudformation.amazonaws.com"},
                            "affectedResources": {"AWS::IAM::Role": "arn:aws:iam::123456789012:role/IAMRole"},
                        },
                    },
                    "resourceRole": "TARGET",
                    "additionalInfo": {"sample": True},
                    "evidence": None,
                    "eventFirstSeen": "2020-02-14T17:59:17Z",
                    "eventLastSeen": "2020-02-14T17:59:17Z",
                    "archived": False,
                    "count": 1,
                },
                "severity": 8,
                "id": "eeb88ab56556eb7771b266670dddee5a",
                "createdAt": "2020-02-14T18:12:22.316Z",
                "updatedAt": "2020-02-14T18:12:22.316Z",
                "title": "Principal AssumedRole:IAMRole attempted to add a policy to themselves that is highly permissive.",
                "description": "Principal AssumedRole:IAMRole attempted to add a highly permissive policy to themselves.",
            },
        ),
    ]