Registering, Testing, and Uploading PyPanther Detections
Overview
It's recommended to use Python (v1) detections instead of PyPanther. Future support of PyPanther is being evaluated, and feature development is paused.
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.
A PyPanther Detection must be registered before it is uploaded to your Panther instance. It's also recommended to define tests for your PyPanther Detections.
Registering PyPanther Detections
Registering a PyPanther rule means including it in your Panther deployment package.
To register a rule, pass it in to register() in your main.py file. When you run the pypantherupload or test commands, only rules passed in to register() will be uploaded to your Panther instance or tested.
If a previously uploaded rule is not passed in to register() on a subsequent invocation of upload, it will be deleted in your Panther instance.
Registering a rule more than once is not allowed. This means each collection passed into register() must contain unique rules.
It’s not required to register rules used as base rules with inheritance so long as you do not want the base rules themselves uploaded into your Panther instance.
# Sample main.py filefrom pypanther import register, get_panther_rules# Register a single rule (that was defined as a class called "BoxNewLogin")register(BoxNewLogin)# Register all Panther-managed rulesregister(get_panther_rules())
Viewing registered rules
To see all rules that are registered given your currently configured repository, run the pypanther list rules CLI command.
You can use the get_rules() function to easily fetch rules you might want to register in main.py. get_rules() takes in an imported package (folder) or module (file) name, and returns all rules defined in that package or module that inherit the pypantherRule class.
Each folder containing rules imported with get_rules() must contain an __init__.py file. Learn more about recommended repository structure in the PyPanther Detections Style Guide.
Example using get_rules():
Registering vs. enabling a rule
All rules have an enabled property, which is different from being registered. See the table below for all possible outcomes:
enabled = True
enabled = False
Registered
Uploaded to Panther and enabled. Unit tests are run during testing.
Uploaded to Panther and disabled. Unit tests are run during testing.
Not registered
Not uploaded to Panther. If uploaded previously, deleted. Unit tests are not run during testing.
Not uploaded to Panther. If uploaded previously, deleted. Unit tests are not run during testing.
Testing PyPanther Detections
Defining tests
Define tests for PyPanther Detections by creating instances of the pypantherRuleTest class. Tests are associated with rules by assigning them to a rule’s tests field. Tests can be defined directly within a rule, or separately set to a variable (that's either local or imported).
Each instance of RuleTest must set a name, expected_result, and log. Each test can also optionally define:
A mocks field, which takes a list of RuleMocks.
(When expected_result=True) Additional fields, prepended with expected_, that verify the output of alert functions. For example, expected_severity and expected_title.
These alert function output verification fields should only be included on RuleTests where expected_result=True. Including them on a test where expected_result=False will cause the test to fail. This is because alert functions are only run when a detection returns True.
Note that this rule's tests (defined above the rule class) use mocks, expected_title, expected_dedup, and expected_alert_context.
Example: Rule with tests
Examples: Mocking entities defined both inside and outside the rule class
You can use RuleMock to define mock values for variables and methods defined either inside or outside of the rule class.
Example: Mocking a variable and function inside the rule class
In the example below, both the INSIDE variable and inside method are defined inside of the rule class, MockTestRule, and mocked in RuleMocks.
Example: Mocking a variable and function defined outside the rule class
In the example below, both the OUTSIDE variable and outside method are defined outside of the rule class, OutsideMockTest, and mocked in RuleMocks.
Running tests
To run all tests defined for your PyPanther Detections, run:
To run only a subset of tests, filter the detections for which tests are run by using a filter flag with test, such as --id or --log-types. See a full list of filter flags by running pypanther test --help.
The test command:
Only tests those detections passed into register()
Skips tests for Panther-managed rules
If any rules fail a test, the error will print next to the rule name that was tested:
Testing custom helper functions and data fixtures
The pypanther test command tests all PyPanther Detection classes that have a tests attribute set. In addition to testing rule() and alert function output in this way, if you have created custom helper functions for use in rules, you may want to write targeted tests for these helper functions. To do this, it's recommended to use a common Python testing framework, such as pytest or unittest.
Currently, pypanther will not run these tests during pypanther test, but they can be run locally or as part of your CI/CD workflow.
Example: using pytest to test a custom function
Say you'd like to write a reusable function that reads a list of AWS account IDs from a file and filters it to include only the production ones. (This function could then be used in an include or exclude filter.) This function might look like the following:
Given this function, there are a few ways in which you may want to test both the data fixture (AWS_ACCOUNTS) itself, as well as the function is_prod_aws_account(). For instance:
To verify that the AWS_ACCOUNTS data fixture:
Has a certain number of rows
Has a certain required field
To verify that certain account IDs are included in the prod list (and others are not)
To test these characteristics using pytest, you would add the following functions (all with the test_ prefix):
After you define these tests, you can invoke them by running pytest from your command line.
Uploading PyPanther Detections to Panther
In order to use the pypanther upload functionality, it must first be enabled for you. If you would like to upload detections, please reach out to your Panther Support team.
To upload all PyPanther Detections that are registered, run:
You can see all upload options by running pypanther upload -h, but may find the following options particularly useful:
--verbose: Generates verbose output, which includes a list of tests by detection (and their pass/fail statuses), a list of registered detections, and a list of included files
--dry-run: Do not upload, but show a summary of the changes that will be applied in the next upload
--output {text, json}: Prints output in the provided format
text is the default value, but json can be useful if you plan to port the output into another workflow
Sample output for pypanther upload --verbose --output json
If a previously uploaded rule is not passed in to register() on a subsequent invocation of upload, it will be deleted in your Panther instance.
Uploading a PyPanther rule with the same id as an existing v1 rule's RuleId will overwrite it. The same is true in the other direction—i.e., uploading a PyPanther rule with the same RuleId as an existing PyPanther rule's id will overwrite it.
upload limitations
The following limitations currently apply to pypanther upload:
A maximum of 500 custom rules can be uploaded at once.
This limit does not include Panther-managed rules (i.e., those returned by the get_panther_rules() function)
The total size of the zip file pypanther produces for upload cannot exceed 10MB.
from pypanther import LogType, Rule, RuleMock, RuleTest, Severity, panther_managed
from pypanther.helpers.base import deep_get
from pypanther.helpers.default import lookup_aws_account_name
from pypanther.helpers.oss import geoinfo_from_ip_formatted
aws_console_root_login_tests: list[RuleTest] = [
RuleTest(
name="Successful Root Login",
expected_result=True,
expected_title="AWS root login detected from [111.111.111.111] (111.111.111.111 in SF, California in USA) in account [sample-account]",
expected_dedup="123456789012-ConsoleLogin-2019-01-01T00:00:00Z",
expected_alert_context={
"sourceIPAddress": "111.111.111.111",
"userIdentityAccountId": "123456789012",
"userIdentityArn": "arn:aws:iam::123456789012:root",
"eventTime": "2019-01-01T00:00:00Z",
"mfaUsed": "No",
},
mocks=[
RuleMock(object_name="geoinfo_from_ip_formatted", return_value="111.111.111.111 in SF, California in USA")
],
log={
"eventVersion": "1.05",
"userIdentity": {
"type": "Root",
"principalId": "1111",
"arn": "arn:aws:iam::123456789012:root",
"accountId": "123456789012",
"userName": "root",
},
"eventTime": "2019-01-01T00:00:00Z",
"eventSource": "signin.amazonaws.com",
"eventName": "ConsoleLogin",
"awsRegion": "us-east-1",
"sourceIPAddress": "111.111.111.111",
"userAgent": "Mozilla",
"requestParameters": None,
"responseElements": {"ConsoleLogin": "Success"},
"additionalEventData": {
"LoginTo": "https://console.aws.amazon.com/console/",
"MobileVersion": "No",
"MFAUsed": "No",
},
"eventID": "1",
"eventType": "AwsConsoleSignIn",
"recipientAccountId": "123456789012",
},
),
RuleTest(
name="Non-Login Event",
expected_result=False,
mocks=[
RuleMock(object_name="geoinfo_from_ip_formatted", return_value="111.111.111.111 in SF, California in USA")
],
log={
"eventVersion": "1.06",
"userIdentity": {
"type": "AssumedRole",
"principalId": "1111:tester",
"arn": "arn:aws:sts::123456789012:user/tester",
"accountId": "123456789012",
"accessKeyId": "1",
"sessionContext": {
"sessionIssuer": {
"type": "Role",
"principalId": "1111",
"arn": "arn:aws:iam::123456789012:user/tester",
"accountId": "123456789012",
"userName": "tester",
},
"attributes": {"creationDate": "2019-01-01T00:00:00Z", "mfaAuthenticated": "true"},
},
},
"eventTime": "2019-01-01T00:00:00Z",
"eventSource": "dynamodb.amazonaws.com",
"eventName": "DescribeTable",
"awsRegion": "us-west-2",
"sourceIPAddress": "111.111.111.111",
"userAgent": "console.amazonaws.com",
"requestParameters": {"tableName": "table"},
"responseElements": None,
"requestID": "1",
"eventID": "1",
"readOnly": True,
"resources": [{"accountId": "123456789012", "type": "AWS::DynamoDB::Table", "ARN": "arn::::table/table"}],
"eventType": "AwsApiCall",
"apiVersion": "2012-08-10",
"managementEvent": True,
"recipientAccountId": "123456789012",
},
),
]
@panther_managed
class AWSConsoleRootLogin(Rule):
id = "AWS.Console.RootLogin-prototype"
display_name = "Root Console Login"
dedup_period_minutes = 15
log_types = [LogType.AWS_CLOUDTRAIL]
tags = [
"AWS",
"Identity & Access Management",
"Authentication",
"DemoThreatHunting",
"Privilege Escalation:Valid Accounts",
]
reports = {"CIS": ["3.6"], "MITRE ATT&CK": ["TA0004:T1078"]}
default_severity = Severity.HIGH
default_description = "The root account has been logged into."
default_runbook = "Investigate the usage of the root account. If this root activity was not authorized, immediately change the root credentials and investigate what actions the root account took.\n"
default_reference = "https://docs.aws.amazon.com/IAM/latest/UserGuide/id_root-user.html"
summary_attributes = ["userAgent", "sourceIpAddress", "recipientAccountId", "p_any_aws_arns"]
tests = aws_console_root_login_tests
def rule(self, event):
return (
event.get("eventName") == "ConsoleLogin"
and deep_get(event, "userIdentity", "type") == "Root"
and (deep_get(event, "responseElements", "ConsoleLogin") == "Success")
)
def title(self, event):
ip_address = event.get("sourceIPAddress")
return f"AWS root login detected from [{ip_address}] ({geoinfo_from_ip_formatted(ip_address)}) in account [{lookup_aws_account_name(event.get('recipientAccountId'))}]"
def dedup(self, event):
# Each Root login should generate a unique alert
return "-".join([event.get("recipientAccountId"), event.get("eventName"), event.get("eventTime")])
def alert_context(self, event):
return {
"sourceIPAddress": event.get("sourceIPAddress"),
"userIdentityAccountId": deep_get(event, "userIdentity", "accountId"),
"userIdentityArn": deep_get(event, "userIdentity", "arn"),
"eventTime": event.get("eventTime"),
"mfaUsed": deep_get(event, "additionalEventData", "MFAUsed"),
}
from pypanther import LogType, Rule, RuleMock, RuleTest, Severity
class MockTestRule(Rule):
id = "mock_test_rule"
display_name = "Mock Test Rule"
default_description = "This rule provides an example of how to mock objects within a rule in rule tests."
log_types = [LogType.AZURE_AUDIT]
default_severity = Severity.INFO
create_alert = False
INSIDE: list[str] = []
def inside(self, event):
return self.INSIDE
def rule(self, event):
return self.inside(event) != []
tests = [
RuleTest(
name="Mock new Test",
expected_result=True,
mocks=[
RuleMock(
# Mock the object INSIDE with a new object ["test"]
object_name="INSIDE",
new=["test"],
)
],
log={"action": "Blocked", "internalIp": ""},
),
RuleTest(
name="Mock side_effect Test",
expected_result=True,
mocks=[
RuleMock(
# Mock the method inside() with a side effect that checks if the action is "Blocked"
object_name="inside",
side_effect=lambda e: e.get("action") == "Blocked",
)
],
log={"action": "Blocked", "internalIp": ""},
),
RuleTest(
name="Mock return_value Test",
expected_result=True,
mocks=[
RuleMock(
# Mock the method inside() with a return value of True
object_name="inside",
return_value=True,
)
],
log={"action": "Blocked", "internalIp": ""},
),
]
from pypanther import LogType, Rule, RuleMock, RuleTest, Severity
OUTSIDE = "outside"
def outside():
return OUTSIDE
class OutsideMockTest(Rule):
id = "outside_mock_test"
display_name = "Outside Mock Test"
default_description = (
"This rule provides an example of how to mock objects outside of the rule class in rule tests."
)
log_types = [LogType.AZURE_AUDIT]
default_severity = Severity.INFO
create_alert = False
def rule(self, event):
return outside() == "outside"
tests = [
RuleTest(
name="Mock outside new",
expected_result=False,
mocks=[
RuleMock(
# Mock the object OUTSIDE with a new value "inside"
object_name="OUTSIDE",
new="inside",
)
],
log={"action": "Blocked", "internalIp": ""},
),
RuleTest(
name="Mock outside return_value",
expected_result=False,
mocks=[
RuleMock(
# Mock the function outside() with a return value "inside"
object_name="outside",
return_value="inside",
)
],
log={"action": "Blocked", "internalIp": ""},
),
RuleTest(
name="Mock outside side_effect",
expected_result=False,
mocks=[
RuleMock(
# Mock the function outside() with a side effect that returns "inside"
object_name="outside",
side_effect=lambda: "inside",
)
],
log={"action": "Blocked", "internalIp": ""},
),
]
$ pypanther test
MyRuleNameI:
FAIL: my test name
- Expected rule() to return 'True', but got 'False'
import pandas as pd
# A sample list of account data you may use with your detections
_account_data = [
{"account_id": "012345678901", "account_age_days": 1543, "environment": "prod"},
{"account_id": "112345678901", "account_age_days": 1753, "environment": "staging"},
...
{"account_id": "756239201432", "account_age_days": 32, "environment": "development"}
]
# Load account data into a pandas dataframe for easy access
AWS_ACCOUNTS = pd.DataFrame.from_dict(_account_data)
def is_prod_aws_account(account_id: str) -> bool:
"""
Checks the AWS_ACCOUNTS dataframe to see if an account_id is in production
"""
return AWS_ACCOUNTS.loc[AWS_ACCOUNTS['account_id'] == account_id, "environment"] == "prod"
def test_accounts_metadata_is_correct_size():
"""
Ensure that the account list has 100 rows.
"""
assert AWS_ACCOUNTS.shape[0] == 100
def test_critical_list_is_marked_prod():
"""
Ensure that specific accounts are marked as production while other ones are not
"""
expected_prod_list = ["012345678901", "012345678902", "012345678903"]
for account_id in expected_prod_list:
assert is_prod_aws_account(account_id)
expected_non_prod_list = ["012345678905", "012345678906", "012345678907"]
for account_id in expected_non_prod_list:
assert not is_prod_aws_account(account_id)
def test_accounts_metadata_has_account_age_field():
"""
Ensure the metadata data fixture has a field we may require for another helper function
"""
assert "account_age_days" in AWS_ACCOUNTS.columns
