saf.process package#

Submodules#

saf.process.test module#

The NOOP process plugins exists as an implementation example.

It doesn’t really do anything to the collected event

class saf.process.test.DelayRange(**kwargs)[source]#

Bases: TypedDict

Delay range configuration.

minimum: float#

maximum: float#

pydantic model saf.process.test.TestProcessConfig[source]#

Bases: ProcessConfigBase

Test collector configuration.

Show JSON schema

{
   "title": "TestProcessConfig",
   "description": "Test collector configuration.",
   "type": "object",
   "properties": {
      "plugin": {
         "title": "Plugin",
         "type": "string"
      },
      "delay": {
         "title": "Delay",
         "exclusiveMinimum": 0,
         "type": "number"
      },
      "delay_range": {
         "$ref": "#/definitions/DelayRange"
      },
      "child_events_count": {
         "title": "Child Events Count",
         "exclusiveMinimum": 0,
         "exclusiveMaximum": 9223372036854775807,
         "type": "integer"
      }
   },
   "required": [
      "plugin"
   ],
   "definitions": {
      "DelayRange": {
         "title": "DelayRange",
         "type": "object",
         "properties": {
            "minimum": {
               "title": "Minimum",
               "type": "number"
            },
            "maximum": {
               "title": "Maximum",
               "type": "number"
            }
         },
         "required": [
            "minimum",
            "maximum"
         ]
      }
   }
}

Config:

allow_mutation: bool = False
underscore_attrs_are_private: bool = True

Fields:

child_events_count (Optional[int])
delay (Optional[float])
delay_range (Optional[saf.process.test.DelayRange])

field delay: Optional[float] = None#

Constraints:

exclusiveMinimum = 0

field delay_range: Optional[DelayRange] = None#

field child_events_count: Optional[int] = None#

Constraints:

exclusiveMinimum = 0
exclusiveMaximum = 9223372036854775807

saf.process.test.get_config_schema() → Type[TestProcessConfig][source]#: Get the test collect plugin configuration schema.

async saf.process.test.process(*, ctx: PipelineRunContext[TestProcessConfig], event: CollectedEvent) → AsyncIterator[CollectedEvent][source]#: Method called to collect events, in this case, generate.

saf.process.regex_mask module#

Mask data based on provided regex rules.

pydantic model saf.process.regex_mask.RegexMaskProcessConfig[source]#

Bases: ProcessConfigBase

Configuration schema for the regex mask processor plugin.

Show JSON schema

{
   "title": "RegexMaskProcessConfig",
   "description": "Configuration schema for the regex mask processor plugin.",
   "type": "object",
   "properties": {
      "plugin": {
         "title": "Plugin",
         "type": "string"
      },
      "rules": {
         "title": "Rules",
         "type": "object",
         "additionalProperties": {
            "type": "string"
         }
      },
      "mask_char": {
         "title": "Mask Char",
         "maxLength": 1,
         "minLength": 1,
         "type": "string"
      },
      "mask_prefix": {
         "title": "Mask Prefix",
         "default": "<:",
         "type": "string"
      },
      "mask_suffix": {
         "title": "Mask Suffix",
         "default": ":>",
         "type": "string"
      }
   },
   "required": [
      "plugin",
      "rules"
   ]
}

Config:

allow_mutation: bool = False
underscore_attrs_are_private: bool = True

Fields:

mask_char (Optional[str])
mask_prefix (str)
mask_suffix (str)
rules (Dict[str, str])

field rules: Dict[str, str] [Required]#

field mask_char: Optional[str] = None#

Constraints:

maxLength = 1
minLength = 1

field mask_prefix: str = '<:'#

field mask_suffix: str = ':>'#

saf.process.regex_mask.get_config_schema() → Type[ProcessConfigBase][source]#: Get the regex mask processor plugin configuration schema.

async saf.process.regex_mask.process(*, ctx: PipelineRunContext[RegexMaskProcessConfig], event: CollectedEvent) → AsyncIterator[CollectedEvent][source]#: Method called to mask the data based on provided regex rules.

saf.process.shannon_mask module#

Mask data based using a length-relative normalized version of the Shannon Index as an indicator of entropy.

pydantic model saf.process.shannon_mask.ShannonMaskProcessConfig[source]#

Bases: ProcessConfigBase

Configuration schema for the Shannon mask processor plugin.

Show JSON schema

{
   "title": "ShannonMaskProcessConfig",
   "description": "Configuration schema for the Shannon mask processor plugin.",
   "type": "object",
   "properties": {
      "plugin": {
         "title": "Plugin",
         "type": "string"
      },
      "mask_str": {
         "title": "Mask Str",
         "default": "HIGH-ENTROPY",
         "type": "string"
      },
      "mask_char": {
         "title": "Mask Char",
         "maxLength": 1,
         "minLength": 1,
         "type": "string"
      },
      "mask_prefix": {
         "title": "Mask Prefix",
         "default": "<:",
         "type": "string"
      },
      "mask_suffix": {
         "title": "Mask Suffix",
         "default": ":>",
         "type": "string"
      },
      "h_threshold": {
         "title": "H Threshold",
         "default": 0.9,
         "minimum": 0.0,
         "maximum": 1.0,
         "type": "number"
      },
      "length_threshold": {
         "title": "Length Threshold",
         "default": 16,
         "exclusiveMinimum": 1,
         "type": "integer"
      },
      "delimeter": {
         "title": "Delimeter",
         "default": " ",
         "maxLength": 1,
         "minLength": 1,
         "type": "string"
      },
      "alphabet": {
         "title": "Alphabet",
         "default": "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789+/=",
         "minLength": 1,
         "type": "string"
      }
   },
   "required": [
      "plugin"
   ]
}

Config:

allow_mutation: bool = False
underscore_attrs_are_private: bool = True

Fields:

alphabet (str)
delimeter (str)
h_threshold (float)
length_threshold (int)
mask_char (Optional[str])
mask_prefix (str)
mask_str (str)
mask_suffix (str)

field mask_str: str = 'HIGH-ENTROPY'#

field mask_char: Optional[str] = None#

Constraints:

maxLength = 1
minLength = 1

field mask_prefix: str = '<:'#

field mask_suffix: str = ':>'#

field h_threshold: float = 0.9#

Constraints:

minimum = 0.0
maximum = 1.0

field length_threshold: int = 16#

Constraints:

exclusiveMinimum = 1

field delimeter: str = ' '#

Constraints:

maxLength = 1
minLength = 1

field alphabet: str = 'abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789+/='#

Constraints:

minLength = 1

saf.process.shannon_mask.get_config_schema() → Type[ProcessConfigBase][source]#: Get the Shannon mask processor plugin configuration schema.

async saf.process.shannon_mask.process(*, ctx: PipelineRunContext[ShannonMaskProcessConfig], event: CollectedEvent) → AsyncIterator[CollectedEvent][source]#: Method called to mask the data based on normalized Shannon index values.