Time: ~90 minutes · Difficulty: Beginner · You need: a terminal and Python 3.11 or newer. No Panther account and no AWS account: everything runs against a sample file.
How this page was checked: a script extracted every code block from this page and ran it in order with panther_analysis_tool 1.10.0 on Python 3.12.14 and 3.13.15 (macOS, in bash and in zsh), and every output shown is pasted from that run. Two things run only inside a Panther deployment and were not executed: counting matches against a threshold (step 6 counts them with a plain-Python stand-in) and evaluating the correlation rule (in step 7 the tool skips its tests, and the page shows what that means).
Detections as Python functions
Panther is a SIEM (security information and event management) platform. It collects logs such as AWS (Amazon Web Services) CloudTrail, adds fields of its own whose names start with p_ (p_log_type, p_event_time), and, as each event arrives, runs every enabled rule written for that kind of log. A rule is two files, by convention with the same name: a .py file holding the logic and a .yml file, in the plain-text YAML format, holding its ID, severity, alert settings and tests. Panther recommends keeping them in version control such as git, and its open-source CLI (command-line interface), panther_analysis_tool, tests them on your machine and uploads them to a Panther deployment.
Detection engineers write these, but the Python involved is small: read a field, compare it, check it against a list, return the answer. This page teaches that much Python and no more, and points at the rules Panther publishes in its open panther-analysis repository as it goes. Panther also runs scheduled searches: queries in SQL (Structured Query Language) that run on a timer over the logs Panther has stored, with a Python rule applied to each row they return. You need no SQL for this page; the SQL guide teaches it from zero.
The distinction that matters: the Python sees one event at a time. "Five refusals from one address in ten minutes" and "a user created, then given a key" both need several events, so neither is written in Python. Counting is set in the YAML file, and a sequence is a separate correlation rule. That split also decides what you can test at home: Python rules run entirely on your machine, while thresholds and correlation rules are evaluated only by a Panther deployment.
On this page
One event in, True or False out
A CloudTrail event is a JSON (JavaScript Object Notation) object: named fields with values, where a value can itself hold named fields. Python reads one into a dictionary. A Panther rule is a function that receives one event and returns True (alert) or False (ignore). This needs nothing but Python:
python3 - <<'PY'
# One trimmed CloudTrail event. Python calls this a dictionary.
event = {"eventName": "StopLogging", "userIdentity": {"userName": "backup-svc"}}
print(event.get("eventName"), event.get("errorCode")) # errorCode is missing
def rule(event): # def names a function; event is its input
return event.get("eventName") in {"StopLogging", "DeleteTrail"}
print(rule(event)) # return hands back the function's answer
PYStopLogging None True
In Python, = gives a value a name, text after # is a comment, and .get() reads a field by its name, giving None, Python's value for "nothing", when the field is missing. The function is a complete Panther rule. in asks "is it one of these?", and {"StopLogging", "DeleteTrail"} is a set, Python's way of writing a group of alternatives. == compares two values, and and, or and not combine comparisons. The indented line belongs to the function: Python uses indentation where many languages use brackets. Inside Panther, event is Panther's own event object, which reads the same way and adds event.deep_get("userIdentity", "userName") for a field inside another field.
Walk-through: a leaked CI key, five detections
The lab file tells one story. The access key of build-bot, a CI (continuous integration) user, has leaked. From the IP (Internet Protocol) address 203.0.113.50 someone checks whose key it is, tries to list IAM (Identity and Access Management) users, EC2 (Elastic Compute Cloud) resources and secrets and is mostly refused, lists the S3 (Simple Storage Service) buckets, then creates a user called backup-svc, gives it AdministratorAccess and an access key, and uses that new identity to stop the CloudTrail trail. Around it runs ordinary work: alice, an administrator, signs in with MFA (multi-factor authentication) and creates svc-reporting, giving it a key hours later; bob signs in without MFA; someone fails to sign in as alice; a reports-app role is refused one S3 object three times in half an hour; a CI role describes instances.
Five detections run through the rest of the page:
- D1, trail tampering: CloudTrail logging stopped, deleted or changed.
- D2, denied-call burst: one source address refused five or more times.
- D3, new user given a key: a user created and handed an access key within 10 minutes.
- D4, console login without MFA: a successful console sign-in without multi-factor authentication.
- D5, administrator policy attached:
AdministratorAccessattached to a user.
1. Install the CLI
Run python3 --version first. panther_analysis_tool requires Python 3.11 or newer, and Panther tests it on 3.11. If yours is older, install a newer Python from python.org or your package manager and use its name (for example python3.12) instead of python3 on the second line.
mkdir -p ~/panther-lab && cd ~/panther-lab python3 -m venv .venv && . .venv/bin/activate pip install --quiet panther-analysis-tool==1.10.0 pat --version
1.10.0
The second line makes a private Python for this lab (a virtual environment) and switches to it, so nothing is installed system-wide; in a new terminal, cd ~/panther-lab && . .venv/bin/activate brings it back. pat and panther_analysis_tool are two names for the same program. Keep the version pin: on Python 3.10 an unpinned install does not fail but quietly installs 0.57.0 from March 2025, the last release that did not require 3.11. Each run, pat also asks PyPI (the Python Package Index) whether a newer release exists and prints a WARNING line if so.
2. Load the lab data
Twenty-four events, one JSON object per line: the same file every guide in this series uses. Paste the whole block, including the final EOF line. Step 3 counts the events, which confirms the paste.
cat > cloudtrail-lab.jsonl <<'EOF'
{"eventTime":"2026-09-15T08:55:12Z","eventSource":"signin.amazonaws.com","eventName":"ConsoleLogin","awsRegion":"us-east-1","sourceIPAddress":"198.51.100.20","userAgent":"Mozilla/5.0","userIdentity":{"type":"IAMUser","arn":"arn:aws:iam::111122223333:user/alice","userName":"alice"},"responseElements":{"ConsoleLogin":"Success"},"additionalEventData":{"MFAUsed":"Yes"}}
{"eventTime":"2026-09-15T09:02:40Z","eventSource":"iam.amazonaws.com","eventName":"CreateUser","awsRegion":"us-east-1","sourceIPAddress":"198.51.100.20","userAgent":"AWS Internal","userIdentity":{"type":"IAMUser","arn":"arn:aws:iam::111122223333:user/alice","userName":"alice"},"requestParameters":{"userName":"svc-reporting"},"readOnly":false}
{"eventTime":"2026-09-15T09:03:05Z","eventSource":"s3.amazonaws.com","eventName":"ListBuckets","awsRegion":"us-east-1","sourceIPAddress":"198.51.100.20","userAgent":"AWS Internal","userIdentity":{"type":"IAMUser","arn":"arn:aws:iam::111122223333:user/alice","userName":"alice"},"readOnly":true}
{"eventTime":"2026-09-15T09:10:22Z","eventSource":"ec2.amazonaws.com","eventName":"DescribeInstances","awsRegion":"us-east-1","sourceIPAddress":"192.0.2.10","userAgent":"aws-cli/2.17.40","userIdentity":{"type":"AssumedRole","arn":"arn:aws:sts::111122223333:assumed-role/ci-deploy/runner-4821"},"readOnly":true}
{"eventTime":"2026-09-15T09:15:47Z","eventSource":"signin.amazonaws.com","eventName":"ConsoleLogin","awsRegion":"us-east-1","sourceIPAddress":"198.51.100.21","userAgent":"Mozilla/5.0","userIdentity":{"type":"IAMUser","arn":"arn:aws:iam::111122223333:user/bob","userName":"bob"},"responseElements":{"ConsoleLogin":"Success"},"additionalEventData":{"MFAUsed":"No"}}
{"eventTime":"2026-09-15T09:20:03Z","eventSource":"signin.amazonaws.com","eventName":"ConsoleLogin","awsRegion":"us-east-1","sourceIPAddress":"192.0.2.77","userAgent":"Mozilla/5.0","userIdentity":{"type":"IAMUser","arn":"arn:aws:iam::111122223333:user/alice","userName":"alice"},"errorMessage":"Failed authentication","responseElements":{"ConsoleLogin":"Failure"},"additionalEventData":{"MFAUsed":"No"}}
{"eventTime":"2026-09-15T09:31:10Z","eventSource":"s3.amazonaws.com","eventName":"GetObject","awsRegion":"us-east-1","sourceIPAddress":"192.0.2.20","userAgent":"Boto3/1.35.10","userIdentity":{"type":"AssumedRole","arn":"arn:aws:sts::111122223333:assumed-role/reports-app/i-0a12b34c56d78e90f"},"errorCode":"AccessDenied","requestParameters":{"bucketName":"finance-reports","key":"2026/09/summary.csv"},"readOnly":true}
{"eventTime":"2026-09-15T09:47:55Z","eventSource":"s3.amazonaws.com","eventName":"GetObject","awsRegion":"us-east-1","sourceIPAddress":"192.0.2.20","userAgent":"Boto3/1.35.10","userIdentity":{"type":"AssumedRole","arn":"arn:aws:sts::111122223333:assumed-role/reports-app/i-0a12b34c56d78e90f"},"errorCode":"AccessDenied","requestParameters":{"bucketName":"finance-reports","key":"2026/09/summary.csv"},"readOnly":true}
{"eventTime":"2026-09-15T10:01:30Z","eventSource":"s3.amazonaws.com","eventName":"GetObject","awsRegion":"us-east-1","sourceIPAddress":"192.0.2.20","userAgent":"Boto3/1.35.10","userIdentity":{"type":"AssumedRole","arn":"arn:aws:sts::111122223333:assumed-role/reports-app/i-0a12b34c56d78e90f"},"errorCode":"AccessDenied","requestParameters":{"bucketName":"finance-reports","key":"2026/09/summary.csv"},"readOnly":true}
{"eventTime":"2026-09-15T10:02:05Z","eventSource":"sts.amazonaws.com","eventName":"GetCallerIdentity","awsRegion":"us-east-1","sourceIPAddress":"203.0.113.50","userAgent":"aws-cli/2.17.40","userIdentity":{"type":"IAMUser","arn":"arn:aws:iam::111122223333:user/build-bot","userName":"build-bot"},"readOnly":true}
{"eventTime":"2026-09-15T10:02:31Z","eventSource":"iam.amazonaws.com","eventName":"ListUsers","awsRegion":"us-east-1","sourceIPAddress":"203.0.113.50","userAgent":"aws-cli/2.17.40","userIdentity":{"type":"IAMUser","arn":"arn:aws:iam::111122223333:user/build-bot","userName":"build-bot"},"errorCode":"AccessDenied","readOnly":true}
{"eventTime":"2026-09-15T10:02:38Z","eventSource":"iam.amazonaws.com","eventName":"ListRoles","awsRegion":"us-east-1","sourceIPAddress":"203.0.113.50","userAgent":"aws-cli/2.17.40","userIdentity":{"type":"IAMUser","arn":"arn:aws:iam::111122223333:user/build-bot","userName":"build-bot"},"errorCode":"AccessDenied","readOnly":true}
{"eventTime":"2026-09-15T10:02:52Z","eventSource":"iam.amazonaws.com","eventName":"GetAccountAuthorizationDetails","awsRegion":"us-east-1","sourceIPAddress":"203.0.113.50","userAgent":"aws-cli/2.17.40","userIdentity":{"type":"IAMUser","arn":"arn:aws:iam::111122223333:user/build-bot","userName":"build-bot"},"errorCode":"AccessDenied","readOnly":true}
{"eventTime":"2026-09-15T10:03:05Z","eventSource":"ec2.amazonaws.com","eventName":"DescribeInstances","awsRegion":"us-east-1","sourceIPAddress":"203.0.113.50","userAgent":"aws-cli/2.17.40","userIdentity":{"type":"IAMUser","arn":"arn:aws:iam::111122223333:user/build-bot","userName":"build-bot"},"errorCode":"Client.UnauthorizedOperation","readOnly":true}
{"eventTime":"2026-09-15T10:03:09Z","eventSource":"ec2.amazonaws.com","eventName":"DescribeVpcs","awsRegion":"us-east-1","sourceIPAddress":"203.0.113.50","userAgent":"aws-cli/2.17.40","userIdentity":{"type":"IAMUser","arn":"arn:aws:iam::111122223333:user/build-bot","userName":"build-bot"},"errorCode":"Client.UnauthorizedOperation","readOnly":true}
{"eventTime":"2026-09-15T10:03:14Z","eventSource":"ec2.amazonaws.com","eventName":"DescribeSecurityGroups","awsRegion":"us-east-1","sourceIPAddress":"203.0.113.50","userAgent":"aws-cli/2.17.40","userIdentity":{"type":"IAMUser","arn":"arn:aws:iam::111122223333:user/build-bot","userName":"build-bot"},"errorCode":"Client.UnauthorizedOperation","readOnly":true}
{"eventTime":"2026-09-15T10:03:22Z","eventSource":"secretsmanager.amazonaws.com","eventName":"ListSecrets","awsRegion":"us-east-1","sourceIPAddress":"203.0.113.50","userAgent":"aws-cli/2.17.40","userIdentity":{"type":"IAMUser","arn":"arn:aws:iam::111122223333:user/build-bot","userName":"build-bot"},"errorCode":"AccessDenied","readOnly":true}
{"eventTime":"2026-09-15T10:03:40Z","eventSource":"s3.amazonaws.com","eventName":"ListBuckets","awsRegion":"us-east-1","sourceIPAddress":"203.0.113.50","userAgent":"aws-cli/2.17.40","userIdentity":{"type":"IAMUser","arn":"arn:aws:iam::111122223333:user/build-bot","userName":"build-bot"},"readOnly":true}
{"eventTime":"2026-09-15T10:06:12Z","eventSource":"iam.amazonaws.com","eventName":"CreateUser","awsRegion":"us-east-1","sourceIPAddress":"203.0.113.50","userAgent":"aws-cli/2.17.40","userIdentity":{"type":"IAMUser","arn":"arn:aws:iam::111122223333:user/build-bot","userName":"build-bot"},"requestParameters":{"userName":"backup-svc"},"readOnly":false}
{"eventTime":"2026-09-15T10:06:30Z","eventSource":"iam.amazonaws.com","eventName":"AttachUserPolicy","awsRegion":"us-east-1","sourceIPAddress":"203.0.113.50","userAgent":"aws-cli/2.17.40","userIdentity":{"type":"IAMUser","arn":"arn:aws:iam::111122223333:user/build-bot","userName":"build-bot"},"requestParameters":{"userName":"backup-svc","policyArn":"arn:aws:iam::aws:policy/AdministratorAccess"},"readOnly":false}
{"eventTime":"2026-09-15T10:06:51Z","eventSource":"iam.amazonaws.com","eventName":"CreateAccessKey","awsRegion":"us-east-1","sourceIPAddress":"203.0.113.50","userAgent":"aws-cli/2.17.40","userIdentity":{"type":"IAMUser","arn":"arn:aws:iam::111122223333:user/build-bot","userName":"build-bot"},"requestParameters":{"userName":"backup-svc"},"responseElements":{"accessKey":{"userName":"backup-svc","accessKeyId":"AKIAIOSFODNN7EXAMPLE","status":"Active"}},"readOnly":false}
{"eventTime":"2026-09-15T10:09:03Z","eventSource":"cloudtrail.amazonaws.com","eventName":"StopLogging","awsRegion":"us-east-1","sourceIPAddress":"203.0.113.50","userAgent":"aws-cli/2.17.40","userIdentity":{"type":"IAMUser","arn":"arn:aws:iam::111122223333:user/backup-svc","userName":"backup-svc"},"requestParameters":{"name":"arn:aws:cloudtrail:us-east-1:111122223333:trail/org-trail"},"readOnly":false}
{"eventTime":"2026-09-15T10:40:18Z","eventSource":"ec2.amazonaws.com","eventName":"DescribeInstances","awsRegion":"us-east-1","sourceIPAddress":"192.0.2.10","userAgent":"aws-cli/2.17.40","userIdentity":{"type":"AssumedRole","arn":"arn:aws:sts::111122223333:assumed-role/ci-deploy/runner-4821"},"readOnly":true}
{"eventTime":"2026-09-15T11:40:26Z","eventSource":"iam.amazonaws.com","eventName":"CreateAccessKey","awsRegion":"us-east-1","sourceIPAddress":"198.51.100.20","userAgent":"AWS Internal","userIdentity":{"type":"IAMUser","arn":"arn:aws:iam::111122223333:user/alice","userName":"alice"},"requestParameters":{"userName":"svc-reporting"},"responseElements":{"accessKey":{"userName":"svc-reporting","accessKeyId":"AKIAI44QH8DHBEXAMPLE","status":"Active"}},"readOnly":false}
EOF3. Your first rule, run over every event (D1)
Rules live in a rules/ folder. The first one catches anyone blinding CloudTrail: stopping a trail, deleting it, or changing what it records.
mkdir -p rules
cat > rules/trail_tampering.py <<'EOF'
# Each of these calls stops, removes or narrows what CloudTrail records.
TAMPERING = {"StopLogging", "DeleteTrail", "UpdateTrail", "PutEventSelectors"}
def rule(event):
return event.get("eventName") in TAMPERING
def title(event):
# An f-string: the parts in {braces} are filled in from the event.
return f"CloudTrail {event.get('eventName')} by {event.deep_get('userIdentity', 'arn')}"
EOFrule() is the only function Panther requires. title() writes the alert's headline, and Panther calls it only for events where rule() returned True. To run a rule over the whole file, save this helper. You do not need to follow every line: it is scaffolding, not a rule.
cat > fire.py <<'EOF'
# fire.py: run one rule file over every lab event, then group the matches
# the way Panther groups them into alerts: by dedup(), or else by title().
import json, runpy, sys
from collections import Counter
from panther_core import PantherEvent # the event object pat hands to rule()
f = runpy.run_path(sys.argv[1]) # the functions in the rule file
group = f.get("dedup") or f.get("title") or (lambda event: "one group")
events = [PantherEvent(json.loads(line)) for line in open("cloudtrail-lab.jsonl")]
matches = [e for e in events if f["rule"](e)]
for e in matches:
print(e["eventTime"], e["sourceIPAddress"], e["eventName"], f["alert_context"](e) if "alert_context" in f else "")
print(len(matches), "of", len(events), "events matched. Matches per group:", dict(Counter(map(group, matches))))
EOF
python3 fire.py rules/trail_tampering.py2026-09-15T10:09:03Z 203.0.113.50 StopLogging
1 of 24 events matched. Matches per group: {'CloudTrail StopLogging by arn:aws:iam::111122223333:user/backup-svc': 1}One match out of 24: backup-svc's StopLogging at 10:09:03, and silence on the other 23. PantherEvent comes with the CLI and wraps each event the way Panther does, which is what gives it deep_get(). The last line groups matches the way Panther groups them into alerts, which step 6 explains.
4. Add the metadata and tests, then run pat test
The YAML file names the rule, sets its severity, maps it to MITRE ATT&CK (Adversarial Tactics, Techniques, and Common Knowledge, a public catalogue of attacker behaviour) and carries its tests. Each test is an event pasted from the lab file (a JSON object is valid YAML as it stands) plus the answer rule() should give: one event that must match and one that must not.
cat > rules/trail_tampering.yml <<'EOF'
AnalysisType: rule
Filename: trail_tampering.py
RuleID: Lab.CloudTrail.TrailTampering
Enabled: true
LogTypes:
- AWS.CloudTrail
Severity: High
Runbook: Confirm the change was planned. If not, restart logging and review every call from the same identity.
Reports:
MITRE ATT&CK:
- TA0112:T1685.002 # Defense Impairment: Disable or Modify Cloud Log
Tests:
- Name: backup-svc stops the trail
ExpectedResult: true
Log: {"eventTime":"2026-09-15T10:09:03Z","eventSource":"cloudtrail.amazonaws.com","eventName":"StopLogging","awsRegion":"us-east-1","sourceIPAddress":"203.0.113.50","userAgent":"aws-cli/2.17.40","userIdentity":{"type":"IAMUser","arn":"arn:aws:iam::111122223333:user/backup-svc","userName":"backup-svc"},"requestParameters":{"name":"arn:aws:cloudtrail:us-east-1:111122223333:trail/org-trail"},"readOnly":false}
- Name: the same attacker listing buckets is not tampering
ExpectedResult: false
Log: {"eventTime":"2026-09-15T10:03:40Z","eventSource":"s3.amazonaws.com","eventName":"ListBuckets","awsRegion":"us-east-1","sourceIPAddress":"203.0.113.50","userAgent":"aws-cli/2.17.40","userIdentity":{"type":"IAMUser","arn":"arn:aws:iam::111122223333:user/build-bot","userName":"build-bot"},"readOnly":true}
EOF
pat testINFO: Testing analysis items in .
Lab.CloudTrail.TrailTampering
[PASS] backup-svc stops the trail
[PASS] [rule] true
[PASS] [title] CloudTrail StopLogging by arn:aws:iam::111122223333:user/backup-svc
[PASS] [dedup] CloudTrail StopLogging by arn:aws:iam::111122223333:user/backup-svc
[PASS] the same attacker listing buckets is not tampering
[PASS] [rule] false
--------------------------
Test Summary
Path: .
Passed: 1
Skipped: 0
Failed: 0
Invalid: 0pat indents with tabs, shown here as spaces. For a test that expects a match it prints what each function returned; for one that expects none, only rule(). [dedup] repeats the title because this rule has no dedup(), which matters in step 6. RuleID must be unique in your deployment (the Lab. prefix keeps yours apart from Panther's own, such as AWS.CloudTrail.Stopped), LogTypes says which data the rule receives (AWS.CloudTrail is Panther's name for CloudTrail), and Severity is one of Info, Low, Medium, High or Critical.
Reports maps the rule to frameworks, each ATT&CK entry written tactic, colon, technique. ATT&CK version 19 moved cloud-log tampering to T1685.002, Disable or Modify Cloud Log, under the new Defense Impairment tactic (TA0112). pat and Panther's own mapping check test only the shape of an entry, so they accept it, but Panther's published rules still use the older IDs: AWS.CloudTrail.Stopped says TA0005:T1562, a technique v19 revoked, under the tactic v19 renamed Stealth.
5. Nested fields: console logins without MFA (D4)
MFA status sits inside another field, at additionalEventData.MFAUsed. event.deep_get() takes the path one name at a time and returns None if any step is missing. The obvious first attempt:
cat > rules/console_login_no_mfa.py <<'EOF'
def rule(event):
return (
event.get("eventName") == "ConsoleLogin"
and event.deep_get("additionalEventData", "MFAUsed") == "No"
)
EOF
python3 fire.py rules/console_login_no_mfa.py2026-09-15T09:15:47Z 198.51.100.21 ConsoleLogin
2026-09-15T09:20:03Z 192.0.2.77 ConsoleLogin
2 of 24 events matched. Matches per group: {'one group': 2}Two matches, and the second is wrong. At 09:20:03 someone at 192.0.2.77 failed to sign in as alice: that event says MFAUsed: No too, but nobody got in. The outcome lives in a different nested field, responseElements.ConsoleLogin, so the rule needs a third condition (the brackets let one condition run over several lines):
cat > rules/console_login_no_mfa.py <<'EOF'
def rule(event):
return (
event.get("eventName") == "ConsoleLogin"
and event.deep_get("responseElements", "ConsoleLogin") == "Success"
and event.deep_get("additionalEventData", "MFAUsed") == "No"
)
EOF
python3 fire.py rules/console_login_no_mfa.py2026-09-15T09:15:47Z 198.51.100.21 ConsoleLogin
1 of 24 events matched. Matches per group: {'one group': 1}Bob only, at 09:15:47. The failed login is the negative test this rule most needs. AWS's own check for this, control CloudWatch.3 among the Security Hub CloudWatch controls, and Panther's AWS.Console.LoginWithoutMFA both require the successful outcome too, and both test MFAUsed != "Yes" (!= means "is not equal to") rather than == "No", so a successful login with the field missing still counts as no MFA. When you give this rule its YAML file, its Reports entry is TA0001:T1078.004: Initial Access, Valid Accounts: Cloud Accounts.
6. Counting belongs in the YAML, not the Python (D2)
A burst of refused calls from one address is a common sign that someone is trying out a stolen key. rule() cannot count: it sees one event and answers once. Panther counts, using three settings. dedup() returns a string, and matches with the same string form one group (with no dedup(), Panther groups by the title). Threshold is how many matches a group needs before an alert is sent, and DedupPeriodMinutes is how long, from its first match, a group keeps counting. The first version treats AccessDenied as "refused":
cat > rules/denied_burst.py <<'EOF'
def rule(event):
return event.get("errorCode") == "AccessDenied"
def dedup(event):
return event.get("sourceIPAddress", "no address") # one group per address
EOF
cat > rules/denied_burst.yml <<'EOF'
AnalysisType: rule
Filename: denied_burst.py
RuleID: Lab.CloudTrail.DeniedCallBurst
Enabled: true
LogTypes:
- AWS.CloudTrail
Severity: Medium
Threshold: 5 # alert when one group reaches 5 matches
DedupPeriodMinutes: 10 # counted over 10 minutes from its first match
EOF
python3 fire.py rules/denied_burst.py | tail -n 17 of 24 events matched. Matches per group: {'192.0.2.20': 3, '203.0.113.50': 4}The final | tail -n 1 keeps only the last line of fire.py's output. The count per group is what Panther compares with Threshold: four from the attacker, three from reports-app, neither reaches five, so no alert. Give this rule a test each way, as in step 4, and both pass. (fire.py ignores time. The attacker's refusals fall within one minute, so a 10-minute period changes nothing for them; reports-app's three are spread over half an hour, which only lowers its count.)
A query that returns nothing and a query that cannot work return the same thing. A rule is no different, and green tests prove only that it does what you wrote, not that you wrote the right thing. Before trusting a quiet rule, drop the condition and look at what the data actually contains.
python3 - <<'PY'
import json
from collections import Counter
print(Counter(json.loads(line).get("errorCode") for line in open("cloudtrail-lab.jsonl")))
PYCounter({None: 14, 'AccessDenied': 7, 'Client.UnauthorizedOperation': 3})Ten refusals, spelled two ways. EC2 reports a refusal as Client.UnauthorizedOperation, and all three of those came from the attacker. AWS's own check for unauthorised calls, Security Hub control CloudWatch.2 (on the AWS page linked in step 5), counts any error code that starts with AccessDenied or ends with UnauthorizedOperation, which also covers AccessDeniedException and a bare UnauthorizedOperation. Do the same. startswith() and endswith() check how a piece of text begins and ends, and the default "" matters: without it, an event with no errorCode gives None, which has no startswith().
cat > rules/denied_burst.py <<'EOF'
def rule(event):
code = event.get("errorCode", "") # "" when the call was not refused
return code.startswith("AccessDenied") or code.endswith("UnauthorizedOperation")
def dedup(event):
return event.get("sourceIPAddress", "no address")
EOF
python3 fire.py rules/denied_burst.py | tail -n 110 of 24 events matched. Matches per group: {'192.0.2.20': 3, '203.0.113.50': 7}Seven from 203.0.113.50, over the threshold of five: one alert, where the first version produced none. Now add the EC2 refusal from the lab file as a test that expects true, so the gap cannot reopen (written before the fix, that test fails, which proves it can). Its Reports entries are TA0007:T1087.004 and TA0007:T1580: Discovery of cloud accounts and of cloud infrastructure. This is not a hypothetical trap. Panther's published AWS.IAMUser.ReconAccessDenied treats EC2 Describe calls as reconnaissance but accepts only AccessDenied, so run over the lab file it matches 3 of the attacker's 7 refusals, and the three EC2 ones never count.
7. Two events, one story: signals and a correlation rule (D3)
"A new user given an access key within ten minutes" joins two events, which one call to rule() never sees together. Panther's answer has two parts. First, two signal rules: ordinary rules with CreateAlert: false, which record their matches for other rules to use but alert nobody. Each returns the user name from alert_context() under the same key, and Panther stores it in the match's p_alert_context field. (not event.get("errorCode") is True when the call was not refused, and default="" is what deep_get() returns if the field is missing.)
cat > rules/iam_user_created.py <<'EOF'
def rule(event):
return event.get("eventName") == "CreateUser" and not event.get("errorCode")
def alert_context(event):
return {"request_username": event.deep_get("requestParameters", "userName", default="")}
EOF
cat > rules/iam_user_created.yml <<'EOF'
AnalysisType: rule
Filename: iam_user_created.py
RuleID: Lab.CloudTrail.IAMUserCreated
Enabled: true
CreateAlert: false
LogTypes:
- AWS.CloudTrail
Severity: Info
EOF
sed 's/CreateUser/CreateAccessKey/' rules/iam_user_created.py > rules/iam_access_key_created.py
sed 's/iam_user_created/iam_access_key_created/; s/IAMUserCreated/IAMAccessKeyCreated/' rules/iam_user_created.yml > rules/iam_access_key_created.yml
python3 fire.py rules/iam_user_created.py
python3 fire.py rules/iam_access_key_created.py2026-09-15T09:02:40Z 198.51.100.20 CreateUser {'request_username': 'svc-reporting'}
2026-09-15T10:06:12Z 203.0.113.50 CreateUser {'request_username': 'backup-svc'}
2 of 24 events matched. Matches per group: {'one group': 2}
2026-09-15T10:06:51Z 203.0.113.50 CreateAccessKey {'request_username': 'backup-svc'}
2026-09-15T11:40:26Z 198.51.100.20 CreateAccessKey {'request_username': 'svc-reporting'}
2 of 24 events matched. Matches per group: {'one group': 2}The two sed lines write the second signal rule by copying the first with its names swapped (s/old/new/ replaces text). Both users were created and later given a key: backup-svc after 39 seconds, svc-reporting after 2 hours 37 minutes. Only a time window separates them. (The signal rules carry no tests here, to save space; give them one each way as in step 4.) Second, the correlation rule itself, a YAML file with no Python: a Sequence of the two signal rules, and a Transition requiring the second within 10 minutes of the first, for the same user name. Its tests describe signals rather than events, so the times below are copied from the output above.
mkdir -p correlation_rules
cat > correlation_rules/new_user_key_within_10m.yml <<'EOF'
AnalysisType: correlation_rule
RuleID: Lab.CloudTrail.NewUserKeyWithin10m
DisplayName: New IAM User Given An Access Key Within 10 Minutes
Enabled: true
Severity: High
Detection:
- Sequence:
- ID: UserCreated
RuleID: Lab.CloudTrail.IAMUserCreated
- ID: KeyCreated
RuleID: Lab.CloudTrail.IAMAccessKeyCreated
Transitions:
- ID: key follows user
From: UserCreated
To: KeyCreated
WithinTimeFrameMinutes: 10
Match:
- On: p_alert_context.request_username
LookbackWindowMinutes: 30 # twice the window, plus 10 minutes for log delivery
Schedule:
RateMinutes: 10 # run every 10 minutes
TimeoutMinutes: 5
Tests:
- Name: backup-svc is given a key 39 seconds after it is created
ExpectedResult: true
RuleOutputs:
- {ID: UserCreated, Matches: {p_alert_context.request_username: {backup-svc: ["2026-09-15T10:06:12Z"]}}}
- {ID: KeyCreated, Matches: {p_alert_context.request_username: {backup-svc: ["2026-09-15T10:06:51Z"]}}}
- Name: svc-reporting waits 2 hours 37 minutes, outside the window
ExpectedResult: false
RuleOutputs:
- {ID: UserCreated, Matches: {p_alert_context.request_username: {svc-reporting: ["2026-09-15T09:02:40Z"]}}}
- {ID: KeyCreated, Matches: {p_alert_context.request_username: {svc-reporting: ["2026-09-15T11:40:26Z"]}}}
EOFIts Reports entries are TA0003:T1136.003 and TA0003:T1098.001: Persistence, by creating a cloud account and by adding cloud credentials. Each run of the rule searches the last LookbackWindowMinutes of signals. Panther's correlation rule documentation, where the feature is marked beta, recommends that a rule running at least as often as its window look back twice the window plus the delay before logs arrive, so that a pair split across two runs is still seen. Now run the tests, with one expectation deliberately wrong, because a clean result here needs a control:
mkdir -p planted sed 's/ExpectedResult: false/ExpectedResult: true/' correlation_rules/new_user_key_within_10m.yml > planted/wrong.yml pat test --path planted --show-failures-only; echo "exit code: $?" rm -r planted
INFO: Testing analysis items in planted
--------------------------
Skipped Tests Summary
~/panther-lab/planted/wrong.yml
Lab.CloudTrail.NewUserKeyWithin10m
--------------------------
Test Summary
Path: planted
Passed: 0
Skipped: 1
Failed: 0
Invalid: 0
exit code: 0A test that claims svc-reporting should alert, and still exit code 0: Skipped: 1 means the tests never ran. Only a Panther deployment evaluates correlation rules, and Panther's documentation says pat test needs an API (application programming interface) token for them and for Simple Detections, rules written entirely in YAML. Locally, pat checks only the file's structure: misspell WithinTimeFrameMinutes and it rejects the file. Given --api-host and --api-token from your deployment, the same command sends these tests there to run. Panther's own published correlation rules, 28 at the commit read for this page, all use a Group form, which matches signals in any order, and ship disabled; the correlation rule documentation covers both forms.
8. Getting rules into Panther
pat upload tests the folder and sends it to your deployment, given --api-host and --api-token (or the PANTHER_API_HOST and PANTHER_API_TOKEN environment variables), and pat zip builds an archive to upload in the Panther Console instead. Panther's own repository shows the usual continuous-integration shape: every pull request runs panther_analysis_tool test --show-failures-only, with the API host and token added where they are available so correlation tests run too, and every push to its develop branch runs panther_analysis_tool upload. Add --minimum-tests 2 to your test step and any rule without at least one true and one false test fails the build.
Hands-on exercises
Write rules/admin_policy_attached.py for D5: return True when the AWS managed AdministratorAccess policy is attached to an IAM user. Check it with fire.py, then name the lab events you would use as its positive and negative tests.
Show the answer
cat > rules/admin_policy_attached.py <<'EOF'
ADMIN = "arn:aws:iam::aws:policy/AdministratorAccess"
def rule(event):
policy = event.deep_get("requestParameters", "policyArn")
return event.get("eventName") == "AttachUserPolicy" and policy == ADMIN and not event.get("errorCode")
EOF
python3 fire.py rules/admin_policy_attached.py2026-09-15T10:06:30Z 203.0.113.50 AttachUserPolicy
1 of 24 events matched. Matches per group: {'one group': 1}One match, build-bot attaching the policy to backup-svc at 10:06:30. That event is the positive test; a good negative is the CreateAccessKey call 21 seconds later, the same actor acting on the same user. Map it to TA0004:T1098.003 (Privilege Escalation: Additional Cloud Roles). Comparing the whole ARN (Amazon Resource Name, AWS's full name for a resource) is a choice: Panther's AWS.IAM.AttachAdminUserPolicy checks only that it ends in AdministratorAccess, which also catches a customer-made policy that borrows the name. Neither sees administrator rights granted another way, such as AttachRolePolicy or an inline policy.
A colleague's burst rule uses the corrected check from step 6, has Threshold: 5, matches all ten refusals, and has never alerted. It has no dedup(), and its title is f"Refused {event.get('eventName')} from {event.get('sourceIPAddress')}". Why is it silent?
Show the answer
With no dedup(), Panther groups matches by the title, and this title contains the event name, so every different call is a group of its own. Save the rule outside rules/ and run it through fire.py: the ten matches fall into eight groups, the largest holding reports-app's three GetObject refusals, so none reaches five. The fix is a dedup() that returns what you are counting per, the source address. pat test hints at this too: its [dedup] line shows the key each test event would be grouped under, and a key that repeats the title word for word deserves a second look.
A colleague raises the correlation rule's WithinTimeFrameMinutes from 10 to 180 "to be safe" and changes nothing else. pat test is still green. What is wrong with the change?
Show the answer
Two things. The window exists to exclude svc-reporting, whose key came about 158 minutes after the user: at 180 the transition accepts that pair, which is exactly what the negative test says must not happen, and locally that test is skipped, so green proves nothing. And the edit breaks a documented limit: WithinTimeFrameMinutes may be no larger than LookbackWindowMinutes, still 30, because a run that looks back 30 minutes never holds both halves of a pair 158 minutes apart. pat checks locally only that both are whole numbers, so neither problem can surface until a Panther deployment sees the rule, for example through the tests run with --api-host and --api-token.
From 203.0.113.50, two identities acted within minutes: build-bot, then backup-svc. Which Panther features would alert when one address uses two or more distinct identities within an hour, and does anything else in the lab file reach two?
Show the answer
Counting distinct values is still counting, so it still belongs outside rule(). Panther's unique-value thresholding does it: add a unique() function returning the identity, keep dedup() returning the address, and set Threshold: 2 and DedupPeriodMinutes: 60. Threshold then counts distinct unique() values per group, as an estimate rather than an exact count, and only Python rules can use it. Nothing else in the lab file reaches two: every other address used a single identity.
Common mistakes
- Counting inside
rule(). It is called once per event and answers once. "Five in ten minutes" isThresholdandDedupPeriodMinutes, grouped bydedup(). - Checking one spelling of "refused". EC2 writes
Client.UnauthorizedOperationwhere most services writeAccessDenied, and the missing spelling fails silently. List the codes your data contains before you write the rule. - Leaving out
dedup()on a threshold rule. Matches are then grouped by the title, and a title with per-event detail in it splits the count until no group reaches the threshold. - Returning something other than
TrueorFalse.return event.get("errorCode") and ...hands backNonefor an event without that field, becauseandreturns one of its values, andpatfails the test withFunctionReturnTypeError. Build the answer from comparisons such asis not None. - Reading fields with
event["name"]. On an event without that field it raisesKeyErrorand the test fails. Useevent.get()andevent.deep_get(), which returnNoneor a default you choose. - Believing a green
pat testfor anything that is not a Python rule. Without API credentials, correlation rule tests are reported as skipped, and a Simple Detection whose test is wrong is reported as passed, both with exit code 0. The check that eachLogTypesvalue exists is skipped too, soAWS.CloudTrialpasses every local test. Run the tests with credentials in CI. - Hand-writing correlation test data. A correlation test feeds in signal matches you typed, so it cannot notice when the signal rules emit a different key. Copy the values from what the signal rules actually return, as step 7 does.
Where next
- Panther's panther-analysis repository: the published rules, a
templates/folder showing every optional function, and the correlation rules. Read the rules for a log source before writing your own. - Panther's detection documentation and the panther_analysis_tool repository for the parts this page could not run: thresholds, correlation and upload.
- SQL for security data lakes for the language of Panther's scheduled searches, and Sigma and YARA for the vendor-neutral way to write these detections.
- Detection engineering for the lifecycle around rule files: review, tuning and retirement.
- The CloudTrail to SIEM lab for real CloudTrail data to point these rules at.
