Global Helper Functions

Overview
A common pattern in programming is to extract repeated code into helper functions—Panther supports this pattern with the global analysis type.
Global helpers are not best suited to frequent changes. Lookup Tables support automatic syncing with S3, which means they don't require code changes within Panther for updates.
Built-in globals
By default, Panther comes with built-in global helpers such as panther_default and panther_oss_helpers. panther_default is a default helper, and panther_oss_helpers provides boilerplate helpers to common caching and other use cases. 
While some globals require configuration, it is recommended to create a net-new global for any custom methods or logic that you would like to add. This reduces the chances of dealing with complex merge conflicts when updating your detection sources.
Using globals
Importing globals
Import global helpers in your detections by declared ID at the top of your analysis function body then call the global as if it were any other python library.
For example:
import panther_oss_helpers
​
​
def rule(event):
 return event['name'] == 'test-bucket'
​
​
def title(event):
 # Lookup the account name from an account Id
 account_name = panther_oss_helpers.lookup_aws_account_name(event['accountId'])
 return 'Suspicious request made to account ' + account_name
Adding new globals
New globals can be created from the Panther Analysis Tool or in your Panther Console.
Panther Console
Panther Analysis Tool
To create a new global in the Panther Console:
1.Log in to your Panther Console and navigate to Build > Helpers.
2.In the upper right corner, click Create New.
​
3.Type your Python functions, then click Create. This global can now be imported in your rules or policies.
Global functions allow common logic to be shared across either rules or policies. To declare them as code, add them into the global_helpers folder with a similar pattern to rules and policies.
Globals defined outside of the global_helpers folder will not be loaded.
​
1.Create your Python file (global_helpers/acmecorp.py):
from fnmatch import fnmatch
​
RESOURCE_PATTERN = 'acme-corp-*-[0-9]'
​
​
def matches_internal_naming(resource_name):
 return fnmatch(resource_name, RESOURCE_PATTERN)
​
 2. Create your specification file:
AnalysisType: global
GlobalID: acmecorp
Filename: acmecorp.py
Description: A set of helpers internal to acme-corp
​
 3. Use this helper in a policy (or a rule):
import acmecorp
​
​
def policy(resource):
 return acmecorp.matches_internal_naming(resource['Name'])
Removing references to helper functions from detections
If you decide to remove dependencies from your detections, we recommend staggering the changes.
Common helpers
deep_get()
Located in panther_base_helpers.
deep_get() can be used to return keys that are nested within Python dictionaries. This function is useful for safely returning nested keys and avoiding an AttributeError when a key is not present. 
If the key you are trying to access is nested inside a list, consider using deep_walk() instead.
def deep_get(dictionary: dict, *keys, default=None):
 """Safely return the value of an arbitrarily nested map
 Inspired by https://bit.ly/3a0hq9E
 """
 return reduce(
 lambda d, key: d.get(key, default) if isinstance(d, Mapping) else default, keys, dictionary
 )
Example
With the following JSON, the deep_get function would return the value of result.
{ "outcome": { "reason": "VERIFICATION_ERROR", "result": "FAILURE" }}
deep_get(event, "outcome", "result") == "FAILURE"
This can be found in the Geographically Improbable Okta Login detection.
default
deep_get() takes in an optional default parameter. If a key is not present at the expected location or the value at that location is None, the default value will be returned.
deep_get(event, "outcome", "nonexistent_key", default="Key Not Found") == "Key Not Found"
deep_walk()
Located in panther_base_helpers.
deep_walk() can be used to return values associated with keys that are deeply nested in Python dictionaries, which may contain any number of dictionaries or lists. This functionality is the key differentiator between deep_walk() and deep_get(). 
As with deep_get(), this traversal is safe and will avoid any exceptions or errors. In the event that a key is not present in the structure, the default value is returned.
def deep_walk(
 obj: Optional[Any], *keys: str, default: Optional[str] = None, return_val: str = "all"
) -> Union[Optional[Any], Optional[List[Any]]]:
 """Safely retrieve a value stored in complex dictionary structure
​
 Similar to deep_get but supports accessing dictionary keys within nested lists as well
​
 Parameters:
 obj (any): the original log event passed to rule(event)
 and nested objects retrieved recursively
 keys (str): comma-separated list of keys used to traverse the event object
 default (str): the default value to return if the desired key's value is not present
 return_val (str): string specifying which value to return
 possible values are "first", "last", or "all"
​
 Returns:
 any | list[any]: A single value if return_val is "first", "last",
 or if "all" is a list containing one element,
 otherwise a list of values
 """
​
 def _empty_list(sub_obj: Any):
 return (
 all(_empty_list(next_obj) for next_obj in sub_obj)
 if isinstance(sub_obj, Sequence) and not isinstance(sub_obj, str)
 else False
 )
​
 if not keys:
 return default if _empty_list(obj) else obj
​
 current_key = keys[0]
 found: OrderedDict = OrderedDict()
​
 if isinstance(obj, Mapping):
 next_key = obj.get(current_key, None)
 return (
 deep_walk(next_key, *keys[1:], default=default, return_val=return_val)
 if next_key is not None
 else default
 )
 if isinstance(obj, Sequence) and not isinstance(obj, str):
 for item in obj:
 value = deep_walk(item, *keys, default=default, return_val=return_val)
 if value is not None:
 if isinstance(value, Sequence) and not isinstance(value, str):
 for sub_item in value:
 found[sub_item] = None
 else:
 found[value] = None
​
 found_list: list[Any] = list(found.keys())
 if not found_list:
 return default
 return {
 "first": found_list[0],
 "last": found_list[-1],
 "all": found_list[0] if len(found_list) == 1 else found_list,
 }.get(return_val, "all")
Example
With the following object, deep_walk() would return the value of very_nested_key:
{"key": {"multiple_nested_lists_with_dict": [[[{"very_nested_key": "very_nested_value"}]]]}}
deep_walk(event, "key", "multiple_nested_lists_with_dict", "very_nested_key", default="") == "very_nested_value"
This can be found in the GCP Service Account Access Denied detection.
default
Like deep_get(), deep_walk() takes an optional default parameter. If a key is not present in the provided event, the key is None, or the key is an empty list, the default value is returned instead.
Using the above example:
deep_walk(event, "key", "multiple_nested_lists_with_dict", "very_nested_nonexistent_key", default="") == ""
return_val
Unlike deep_get(), deep_walk() can return three distinct value classifications:
all
first
last
all
By default, deep_walk() will return all values for a given key. This is useful for cases where a key is duplicated in an event; however, if the number of values returned by all is one, only that value is returned.
For example:
{"key": {"inner_key": [{"nested_key": "nested_value"}, {"nested_key": "nested_value2"}]}}
deep_walk(event, "key", "inner_key", "nested_key", default="") == ['nested_value', 'nested_value2']
When using all and returning multiple values, the elements in the list can be accessed like any other Python list. 
first
To return only the first found value for a key, specify return_val="first".
For example:
deep_walk(event, "key", "inner_key", "nested_key", default="", return_val="first") == "nested_value"
last
To return only the last found value for a key, specify return_val="last".
For example:
deep_walk(event, "key", "inner_key", "nested_key", default="", return_val="last") == "nested_value2"
is_ip_in_network()
Located in panther_base_helpers.
is_ip_in_network() is a function to check if an IP address is within a list of IP ranges. This function can be used with a list of known internal networks for added context to the detection.
def is_ip_in_network(ip_addr, networks):
 """Check that a given IP is within a list of IP ranges"""
 return any(ip_address(ip_addr) in ip_network(network) for network in networks)
Example:
SHARED_IP_SPACE = [
 "192.168.0.0/16",
]
​
if is_ip_in_network(event.get("ipaddr"), SHARED_IP_SPACE):
 ...
An example can be found in the OneLogin Active Login Activity detection.
pattern_match()
Located in panther_base_helpers.
Wrapper around fnmatch for basic pattern globs. This can be used when simple pattern matching is needed without the requirement of using regex.
def pattern_match(string_to_match: str, pattern: str):
 """Wrapper around fnmatch for basic pattern globs"""
 return fnmatch(string_to_match, pattern)
Example:
With the following JSON the pattern_match() function would return true. 
{ "operation": "REST.PUT.OBJECT" }
pattern_match(event.get("operation", ""), "REST.*.OBJECT")
An example can be found in the AWS S3 Access Error detection.
pattern_match_list()
Located in panther_base_helpers.
Similar to pattern_match(), pattern_match_list() can check that a string matches any pattern in a given list.
def pattern_match_list(string_to_match: str, patterns: Sequence[str]):
 """Check that a string matches any pattern in a given list"""
 return any(fnmatch(string_to_match, p) for p in patterns)
Example:
With the following JSON the pattern_match_list() function would return true. 
{ "userAgent": "aws-sdk-go/1.29.7 (go1.13.7; darwin; amd64) APN/1.0 HashiCorp/1.0 Terraform/0.12.24 (+https://www.terraform.io)" }
ALLOWED_USER_AGENTS = {
 "* HashiCorp/?.0 Terraform/*",
 # 'console.ec2.amazonaws.com',
 # 'cloudformation.amazonaws.com',
}
​
pattern_match_list(event.get("userAgent"), ALLOWED_USER_AGENTS)
An example can be found in the AWS EC2 Manual Security Group Change detection.
aws_strip_role_session_id()
Located in panther_base_helpers.
aws_strip_role_session_id() strips the session ID our of the arn. 
def aws_strip_role_session_id(user_identity_arn):
 # The ARN structure is arn:aws:sts::123456789012:assumed-role/RoleName/<sessionId>
 arn_parts = user_identity_arn.split("/")
 if arn_parts:
 return "/".join(arn_parts[:2])
 return user_identity_arn
Example:
With the following value, aws_strip_role_session_id() would return arn:aws:sts::123456789012:assumed-role/demo
{ "arn": "arn:aws:sts::123456789012:assumed-role/demo/sessionName" }
aws_strip_role_session_id(user_identity.get("arn", ""))
An example can be found in the AWS Unauthorized API Call detection.
Data Models
Modifying Detections with Inline Filters
Last modified 3mo ago