panther-sdk 0.0.24

No project description provided

Panther SDK

The Panther SDK module allows you to configure detections for your Panther instance.

Getting Started

Install

The Panther SDK can be installed using PIP.

pip3 install panther-sdk==0.0.24

Writing a detection

You can use the detection module in Panther SDK to configure detections. The example below shows a detection that processing an HTTP log.

Assuming we have a log type Internal.HTTP.Traffic that has the following shape:

{
    "method": "POST",
    "useSSL": false,
    "path": "/api/some/endpoint",
    "host": "internal.megacorp.com"
}

We can write a detection to check for insecure POST and PUT requests:

# import the "detection" module, as we'll be creating a rule.
from panther_sdk import detection

# The next four functions are used in the detection definition that follows. Panther 
# detections allow for parts of a detection to be defined with custom Python code.


# "filter_insecure" is an example of python function that will be used to create
# a detection.PythonFilter that only matches insecure events. The event argument will
# have a python dict object corresponding to the JSON event.
def filter_insecure(event, params):
    return event["useSSL"] == False


# "filter_http_methods" is a slightly more complicated python function that we will use
# to create a detection.PythonFilter later. This function uses the second "params" argument
# to make the function's logic reusable.
def filter_http_methods(event, params):
    allowed_methods = params["methods"]
    actual_method = event["method"]
    return actual_method in allowed_methods


# "reference_generator" is another example where we will use some python code to dynamically
# change the information Panther presents on an Alert resulting from a Detection match. In
# this example, we're populating a query param with some relevant information from 
# the contextual event.
def reference_generator(event):
    origin = event["origin"]
    return f"https://wiki.internal.megacorp.com/hosts_db?host={origin}"


# "make_context" will be used to attach arbitrary data to the resulting Alert. This
# can be used to simply change how data is presented or be used to push machine readable information
# to a downstream Alert Destination.
def make_context(event):
    # The Panther SDK's hooks to arbitrary python come with some restrictions. When the
    # detection is saved, the Panther backend will capture the source of provided function to use
    # in realtime log data processing and other backend processes. Because of this, functions used
    # to define a detection must not have references that are not local to the function body. Therefore,
    # imports targeting any of the below _must_ be included in the function body:
    #
    #   - The python standard library
    #   - Libraries that are enabled on your Panther instance
    #   - Panther "Global Helpers" that you have configured on your instance
    
    import fnmatch 
    
    path = event["path"]
    return {
        "is_api_path": fnmatch.fnmatch(path, "/api/*")
    }


# "pick_severity" is a custom function we are using to define the severity of Alerts based on
# the data present in the current log. The detection.Rule declaration below shows how this method
# is registered with the definition of our Rule.
def pick_severity(event) -> str:
    from panther_sdk import detection # required. see notes in comment above "make_context"
    
    if event["origin"] != "internal.megacorp.com":
        return detection.SeverityInfo

    return detection.SeverityInfo


# Declare a rule. Every call to "detection.Rule" in your repo will create a Rule in the Panther backend.
# This example uses a subset of the fields available to define a Rule. See "Full Dataclass API" for
# all the available fields.
detection.Rule(
    # Give the Rule an id. This name must be unique on your Panther 
    # instance. Dot separated names following "namespace" pattern is recommended. 
    rule_id="Internal.HTTP.Traffic.Insecure",
    
    # Pick a human friendly name for the rule. This will be used
    # to present the Rule in the Panther Console UI
    name="Detect insecure internal HTTP traffic",
    
    # Specify one or more log types for your Rule. This setting is what
    # tells the Panther backend to run this Rule against the log data depicted above.
    log_types=["Internal.HTTP.Traffic"],
    
    # Specify an enabled state for the Rule. Detections can be uploaded in a
    # disabled state and will not begin processing log data until enabled.
    enabled=True,
    
    # Optionally, you can specify a list of tags to associate with the detection.
    tags=["internal"],

    # The "filters" list defines the sequence of matching logic that an event
    # will be tested against to determine whether there is a match. Returning "True" proceeds to the
    # next check. If the final check returns "True", there is a match.
    filters=[
        # If the HTTP transaction was secure, we're not interested in that log in this detection.
        # Therefore, the first filter uses the "filter_insecure" function we defined to filter to only
        # events that have useSSL set to "false". 
        detection.PythonFilter(func=filter_insecure),
        
        # For now, we're only interested in getting alerted for insecure POST and PUT requests.
        # We therefore define the second filter step using "filter_http_methods". Unlike "filter_insecure",
        # we pass some parameters to this function. 
        detection.PythonFilter(func=filter_http_methods, params={'methods': ["POST", "PUT"]}),
    ],

    # On this line, we're defining the severity that should be associated with any resulting alerts.
    # We have the option of defining a static severity (simply: severity="INFO") but, in this case, we want
    # to make the severity dynamic based on data from the event. To do this, we reference the "pick_severity"
    # function we defined at the top of the file. We also provide a fallback value in case the dynamic function
    # cannot be processed. This fallback will also be the Severity value displayed in the Panther Console UI
    severity=detection.DynamicStringField(
        func=pick_severity,
        fallback=detection.SeverityInfo,
    ),

    # Below we specify the "reference" field. This optional field is used to attach a link to any Alerts 
    # resulting from this detection. Similar to "severity" we can make this field dynamic based on 
    # event data. There is also a fallback value specified.
    reference=detection.DynamicStringField(
        func=reference_generator,
        fallback="https://wiki.internal.megacorp.com/hosts_db",
    ),
    
    # Fields like "runbook" and "description" are also available to further customize how any 
    # resulting Alerts will be displayed. These can have dynamic handlers defined, but in this 
    # example static string values will work just fine.
    runbook="Optional runbook content",
    description="A helpful description",
    
    # Finally, we use the optional "alert_context" field to tell the Panther backend to use our 
    # "make_context" function to generate custom data that will be attached to any resulting Alerts.
    alert_context=make_context,
)

There is support for override behavior that can be very helpful for writing factory functions for your detections. You can provide defaults for your detection while letting customizations be set when the function is called. For example:

from panther_sdk import detection

# This is an example of a factory rule function that might live in panther_detections or your own library of detections.
# This function accepts an optional overrides object, which is passed through to the Rule instantiation. Any attributes
# set on the RuleOverrides object will override the attributes defined for the rule.
def factory_rule_func(overrides: Optional[detection.RuleOverrides] = None) -> detection.Rule:
    return detection.Rule(
        rule_id="default.rule.id",
        filters=[...],
        log_types=["AWS.ALB"],
        severity=detection.SeverityMedium,
        overrides=overrides,
    )


# Here is an example of using the rule factory function.
def make_detections() -> None:
    # We can pass no overrides at all, and it will be the default rule the function creates
    r = factory_rule_func()
    print(r)

    # We can override the id and severity to create a more customized rule using our function from above
    r = factory_rule_func(overrides=detection.RuleOverrides(
        rule_id="not.important.rule",
        severity=detection.SeverityInfo,
    ))
    print(r)

    r = factory_rule_func(overrides=detection.RuleOverrides(
        rule_id="very.important.rule",
        severity=detection.SeverityHigh,
    ))
    print(r)


# Now we call our main detection making function to execute it when running this python script
make_detections()

Similar to overrides, you have the ability to add extensions to your Panther content. These can be used the same way as overrides on fields that have lists.

from panther_sdk import detection

# This is an example of a factory rule function that might live in panther_detections or your own library of detections.
# This function accepts an optional extensions object, which is passed through to the Rule instantiation. Any attributes
# set on the RuleExtensions object will override the attributes defined for the rule.
def factory_rule_func(extensions: Optional[detection.RuleExtensions] = None) -> detection.Rule:
    return detection.Rule(
        rule_id="default.rule.id",
        filters=[...],
        log_types=["AWS.ALB"],
        severity=detection.SeverityMedium,
        extensions=extensions,
    )


# Here is an example of using the rule factory function.
def make_detections() -> None:
    # We can pass no extensions at all, and it will be the default rule the function creates
    r = factory_rule_func()
    print(r)

    # We can extend filters to add logic to the rule
    r = factory_rule_func(extensions=detection.RuleExtensions(
        filters=[detection.PythonFilter(func=my_very_important_filter)],
    ))
    print(r)


# Now we call our main detection making function to execute it when running this python script
make_detections()

Full Dataclass API

detection module

DynamicStringField

Make a field dynamic based on the detection input

Field	Description	Type
fallback	Fallback value in case the dynamic handler fails	str
func	Dynamic handler	Optional[Callable[[PantherEvent], str]]

DynamicDestinations

Make destinations dynamic based on the detection input

Field	Description	Type
fallback	Fallback value in case the dynamic handler fails	Optional[List[str]]
func	Dynamic handler	Callable[[PantherEvent], List[str]]

AlertGrouping

Configuration for how an alert is grouped

Field	Description	Type
period_minutes	How long should matches be grouped into an alert after the first match	int
group_by	Function to generate a key for grouping matches	Optional[Callable[[PantherEvent], str]]

PythonFilter

Create a filter by referencing a python function

Field	Description	Type
func	Provide a function whose python source will be used as the filter definition	Callable[[PantherEvent], bool]

UnitTestMock

Mock for a unit test

Field	Description	Type
name	name of the object to mock	str
return_value	string to assign as the return value for the mock	str

JSONUnitTest

Unit test with json content

Field	Description	Type
name	name of the unit test	str
expect_match	whether the data should match and trigger an alert	bool
data	json string	str
mocks	list of mocks to use in the test	Optional[List[UnitTestMock]]

Policy

Define a Policy-type detection to execute against log data in your Panther instance

Field	Description	Type
policy_id	ID for the Policy	str
ignore_patterns	Patterns of resource ids to ignore for the policy	Optional[Union[str, List[str]]]
destinations	Alert destinations for the policy	Optional[Union[str, List[str], DynamicDestinations]]
filters	Define event filters for the policy	Union[Union[PythonFilter], List[Union[PythonFilter]]]
enabled	Whether the policy is enabled or not	bool
resource_types	What resource types this policy will apply to	Union[str, List[str]]
severity	What severity alerts generated from this policy get assigned	Union[str, DynamicStringField]
name	What name to display in the UI and alerts. The PolicyID will be displayed if this field is not set.	Optional[str]
description	Description for the policy	Optional[Union[str, DynamicStringField]]
reference	The reason this policy exists, often a link to documentation	Optional[Union[str, DynamicStringField]]
reports	A mapping of framework or report names to values this policy covers for that framework	Optional[Dict[str, List[str]]]
runbook	The actions to be carried out if this policy fails, often a link to documentation	Optional[Union[str, DynamicStringField]]
tags	Tags used to categorize this policy	Optional[Union[str, List[str]]]
unit_tests	Unit tests for this policy	Optional[Union[Union[JSONUnitTest], List[Union[JSONUnitTest]]]]
alert_title	Title to use in the alert	Optional[Callable[[PantherEvent], str]]
alert_context	Optional JSON to attach to alerts generated by this policy	Optional[Callable[[PantherEvent], Dict[str, Any]]]
alert_grouping	Configuration for how an alert is grouped	Optional[AlertGrouping]

Rule

Define a Rule-type detection to execute against log data in your Panther instance

Field	Description	Type
rule_id	ID for the rule	str
severity	Severity for the rule	Union[str, DynamicStringField]
threshold	Number of matches received before an alert is triggered	int
name	Display name for the rule	Optional[str]
log_types	Log Types to associate with this rule	Union[str, List[str]]
filters	Define event filters for the rule	Union[Union[PythonFilter], List[Union[PythonFilter]]]
enabled	Whether the rule is enabled or not	bool
unit_tests	Define event filters for the rule	Optional[Union[Union[JSONUnitTest], List[Union[JSONUnitTest]]]]
tags	Tags for the rule	Optional[Union[str, List[str]]]
reference	Reference for the rule	Optional[Union[str, DynamicStringField]]
runbook	Runbook for the rule	Optional[Union[str, DynamicStringField]]
description	Description for the rule	Optional[Union[str, DynamicStringField]]
summary_attrs	Summary Attributes for the rule	Optional[List[str]]
reports	Report mappings for the rule	Optional[Dict[str, List[str]]]
destinations	Alert destinations for the rule	Optional[Union[str, List[str], DynamicDestinations]]
alert_title	Title to use in the alert	Optional[Callable[[PantherEvent], str]]
alert_context	Optional JSON to attach to alerts generated by this rule	Optional[Callable[[PantherEvent], Dict[str, Any]]]
alert_grouping	Configuration for how an alert is grouped	Optional[AlertGrouping]

ScheduledRule

Define a ScheduledRule-type detection to execute against query results in your Panther instance

Field	Description	Type
rule_id	ID for the scheduled rule	str
severity	What severity alerts generated from this scheduled rule get assigned	Union[str, DynamicStringField]
threshold	Number of matches received before an alert is triggered	int
name	Display name for the scheduled rule	Optional[str]
scheduled_queries	Scheduled queries to associate with this scheduled rule	Union[str, List[str]]
filters	Define event filters for the scheduled rule	Union[Union[PythonFilter], List[Union[PythonFilter]]]
enabled	Short description for the scheduled rule	bool
unit_tests	Define event filters for the scheduled rule	Optional[Union[Union[JSONUnitTest], List[Union[JSONUnitTest]]]]
tags	Tags for the scheduled rule	Optional[Union[str, List[str]]]
reference	Reference for the scheduled rule	Optional[Union[str, DynamicStringField]]
runbook	Runbook for the scheduled rule	Optional[Union[str, DynamicStringField]]
description	Description for the scheduled rule	Optional[Union[str, DynamicStringField]]
summary_attrs	Summary Attributes for the scheduled rule	Optional[List[str]]
reports	Report mappings for the scheduled rule	Optional[Dict[str, List[str]]]
destinations	Alert destinations for the scheduled rule	Optional[Union[str, List[str], DynamicDestinations]]
alert_title	Title to use in the alert	Optional[Callable[[PantherEvent], str]]
alert_context	Optional JSON to attach to alerts generated by this rule	Optional[Callable[[PantherEvent], Dict[str, Any]]]
alert_grouping	Configuration for how an alert is grouped	Optional[AlertGrouping]

query module

CronSchedule

Cron expression based schedule definition for a query

Field	Description	Type
expression	Defines how often queries using this schedule run	str
timeout_minutes	Defines the timeout applied to queries with this schedule	int

IntervalSchedule

Interval based schedule definition for a query

Field	Description	Type
rate_minutes	Defines how often queries using this schedule run	int
timeout_minutes	Defines the timeout applied to queries with this schedule	int

Query

A saved or scheduled query

Field	Description	Type
name	Unique name for the query	str
description	Short description for the query	str
default_database	Default database for the query	str
sql	SQL statement	str
enabled	Whether the query is enabled or not	bool
tags	Tags for the query	Optional[Union[str, List[str]]]
schedule	Schedule attached to the query	Optional[Union[IntervalSchedule, CronSchedule]]

schema module

DataModelMapping

Field	Description	Type
name	Name of the data model field. This will be the name used when accessing the field from within detections.	str
path	Path to the target value in the Panther Event. This can be a simple field name or complete JSON path starting with a $. JSON path syntax must be compatible with the jsonpath-ng Python package.	Optional[str]
func	A Python function to access the target value. The input is the Panther Event and output is the target value in the Panther Event.	Optional[Callable[[PantherEvent], Any]]

DataModel

Data Models provide a way to configure a set of unified fields across all log types.

Field	Description	Type
data_model_id	The unique identifier of the data model.	str
name	What name to display in the UI and alerts. The data_model_id will be displayed if this field is not set.	Optional[str]
enabled	Whether this data model is enabled.	Optional[bool]
log_type	What log type this data model will apply to.	str
mappings	Mapping from source field name or method to unified data model field name.	Union[DataModelMapping, List[DataModelMapping]]