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[source]#

Bases: TypedDict

Delay range configuration.

minimum: float#

maximum: float#

class saf.process.test.TestProcessConfig(*, plugin: str, delay: Optional[float][Optional[float]] = None, delay_range: Optional[DelayRange] = None, child_events_count: Optional[int][Optional[int]] = None)[source]#

Bases: ProcessConfigBase

Test collector configuration.

delay: Optional[float]#

delay_range: Optional[DelayRange]#

child_events_count: Optional[int]#

model_config: ClassVar[ConfigDict] = {'frozen': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'child_events_count': FieldInfo(annotation=Union[int, NoneType], required=False, metadata=[Gt(gt=0), Lt(lt=9223372036854775807)]), 'delay': FieldInfo(annotation=Union[float, NoneType], required=False, metadata=[Gt(gt=0)]), 'delay_range': FieldInfo(annotation=Union[DelayRange, NoneType], required=False), 'plugin': FieldInfo(annotation=str, required=True)}#

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

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.

class saf.process.regex_mask.RegexMaskProcessConfig(*, plugin: str, rules: Dict[str, str], mask_char: Optional[str][Optional[str]] = None, mask_prefix: str = '<:', mask_suffix: str = ':>')[source]#

Bases: ProcessConfigBase

Configuration schema for the regex mask processor plugin.

rules: Dict[str, str]#

mask_char: Optional[str]#

mask_prefix: str#

mask_suffix: str#

model_config: ClassVar[ConfigDict] = {'frozen': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'mask_char': FieldInfo(annotation=Union[str, NoneType], required=False, metadata=[MinLen(min_length=1), MaxLen(max_length=1)]), 'mask_prefix': FieldInfo(annotation=str, required=False, default='<:'), 'mask_suffix': FieldInfo(annotation=str, required=False, default=':>'), 'plugin': FieldInfo(annotation=str, required=True), 'rules': FieldInfo(annotation=Dict[str, str], required=True)}#

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

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.

class saf.process.shannon_mask.ShannonMaskProcessConfig(*, plugin: str, mask_str: str = 'HIGH-ENTROPY', mask_char: Optional[str][Optional[str]] = None, mask_prefix: str = '<:', mask_suffix: str = ':>', h_threshold: float[float] = 0.9, length_threshold: int[int] = 16, delimeter: str[str] = ' ', alphabet: str[str] = 'abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789+/=')[source]#

Bases: ProcessConfigBase

Configuration schema for the Shannon mask processor plugin.

mask_str: str#

mask_char: Optional[str]#

mask_prefix: str#

mask_suffix: str#

h_threshold: float#

length_threshold: int#

delimeter: str#

alphabet: str#

model_config: ClassVar[ConfigDict] = {'frozen': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'alphabet': FieldInfo(annotation=str, required=False, default='abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789+/=', metadata=[MinLen(min_length=1)]), 'delimeter': FieldInfo(annotation=str, required=False, default=' ', metadata=[MinLen(min_length=1), MaxLen(max_length=1)]), 'h_threshold': FieldInfo(annotation=float, required=False, default=0.9, metadata=[Ge(ge=0.0), Le(le=1.0)]), 'length_threshold': FieldInfo(annotation=int, required=False, default=16, metadata=[Gt(gt=1)]), 'mask_char': FieldInfo(annotation=Union[str, NoneType], required=False, metadata=[MinLen(min_length=1), MaxLen(max_length=1)]), 'mask_prefix': FieldInfo(annotation=str, required=False, default='<:'), 'mask_str': FieldInfo(annotation=str, required=False, default='HIGH-ENTROPY'), 'mask_suffix': FieldInfo(annotation=str, required=False, default=':>'), 'plugin': FieldInfo(annotation=str, required=True)}#

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

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.