### Install a Plugin Source: https://github.com/sigmahq/pysigma/blob/main/docs/Plugin_System.md Installs a discovered plugin. This method is called on a `SigmaPlugin` object obtained from `get_plugins()`. ```python plugin.install() ``` -------------------------------- ### Install Git Hooks with pre-commit Source: https://github.com/sigmahq/pysigma/blob/main/README.md Install git hooks using pre-commit after cloning the repository and installing dependencies. This ensures code quality checks are performed before commits. ```bash poetry run pre-commit install ``` -------------------------------- ### Get Available Plugins Source: https://github.com/sigmahq/pysigma/blob/main/docs/Plugin_System.md Retrieves a list of plugins matching specified types and states, filtered for compatibility with the current pySigma version. Use `plugin.install()` to install a discovered plugin. ```python plugins.get_plugins( plugin_types={ SigmaPluginType.BACKEND }, plugin_state={ SigmaPluginState.STABLE }, compatible_only=True, ) ``` -------------------------------- ### Install pySigma Source: https://github.com/sigmahq/pysigma/blob/main/README.md Install pySigma using pip, pipenv, or poetry. Ensure you have Python 3.10 or later. ```bash pip install pysigma ``` ```bash pipenv install pysigma ``` ```bash poetry add pysigma ``` -------------------------------- ### Auto-discover Installed Sigma Plugins Source: https://context7.com/sigmahq/pysigma/llms.txt InstalledSigmaPlugins scans namespaces to automatically discover all installed pySigma plugin packages. Instantiate discovered backends by name and convert Sigma rules to queries. ```python from sigma.plugins import InstalledSigmaPlugins # Auto-discover all installed plugins plugins = InstalledSigmaPlugins.autodiscover() print("Backends:", list(plugins.backends.keys())) # e.g. ['splunk', 'elasticsearch', 'qradar', ...] print("Pipelines:", list(plugins.pipelines.keys())) # e.g. ['sysmon', 'ecs_windows', 'crowdstrike', ...] print("Validators:", list(plugins.validators.keys())) # e.g. ['sigma_tags', 'sigma_status', ...] # Instantiate a discovered backend by name if "splunk" in plugins.backends: BackendClass = plugins.backends["splunk"] pipeline_fn = plugins.pipelines.get("sysmon") pipeline = pipeline_fn() if pipeline_fn else None backend = BackendClass(pipeline) from sigma.collection import SigmaCollection rules = SigmaCollection.load_ruleset(["rules/"]) queries = backend.convert(rules) for q in queries: print(q) ``` -------------------------------- ### Autodiscover Installed Plugins Source: https://github.com/sigmahq/pysigma/blob/main/docs/Plugin_System.md Instantiates InstalledSigmaPlugins by discovering all available plugin classes using defined conventions. This allows referencing plugin components by their identifiers. ```python plugins = InstalledSigmaPlugins.autodiscover() ``` -------------------------------- ### Reference Installed Plugin Components Source: https://github.com/sigmahq/pysigma/blob/main/docs/Plugin_System.md Accesses installed plugin components like backends, pipelines, or validators using their registered identifiers. A pipeline resolver can be obtained via `plugins.get_pipeline_resolver()`. ```python plugins.backends["backend-indetifier"] plugins.pipelines["pipeline-indetifier"] plugins.validators["validator-indetifier"] ``` -------------------------------- ### YAML Processing Pipeline Example Source: https://github.com/sigmahq/pysigma/blob/main/docs/Processing_Pipelines.md Defines a custom processing pipeline named 'Custom Sysmon field naming' with a priority of 100. It includes a field name mapping transformation and a rule condition to apply it only to Sysmon logs. ```yaml name: Custom Sysmon field naming priority: 100 transformations: - id: field_mapping type: field_name_mapping mapping: CommandLine: command_line rule_conditions: - type: logsource service: sysmon ``` -------------------------------- ### Instantiate SigmaValidator from YAML configuration Source: https://github.com/sigmahq/pysigma/blob/main/docs/Rule_Validation.md Load SigmaValidator configuration, including validators and exclusions, from a YAML file. This allows for flexible and persistent validation setups. ```python with open("config.yml") as validation_config: rule_validator = SigmaValidator.from_yaml(validation_config.read()) ``` -------------------------------- ### Nested Processing Transformation Source: https://github.com/sigmahq/pysigma/blob/main/docs/Processing_Pipelines.md Allows nesting multiple transformations. This example shows a field name mapping followed by setting a state. ```yaml transformations: type: nest items: - type: field_name_mapping mapping: EventID: EventCode CommandLine: - command_line - cmdline - type: set_state state: processed ``` -------------------------------- ### Validate Sigma Rules with SigmaValidator Source: https://context7.com/sigmahq/pysigma/llms.txt Use SigmaValidator to run a set of SigmaRuleValidator subclasses across rules to detect quality and consistency issues. Validators maintain state across all rules for accurate uniqueness checks. Load validator configurations from YAML for flexible setup. ```python from sigma.validation import SigmaValidator from sigma.validators.core import ( SigmaTagsValidator, SigmaStatusExistsValidator, SigmaDescriptionLengthValidator, ) from sigma.validators.base import SigmaValidationIssueSeverity from sigma.collection import SigmaCollection collection = SigmaCollection.from_yaml(""" title: Missing Status Rule logsource: category: process_creation product: windows detection: sel: CommandLine: evil.exe condition: sel """, collect_errors=True) validator = SigmaValidator( validators=[ SigmaTagsValidator, SigmaStatusExistsValidator, SigmaDescriptionLengthValidator, ] ) issues = validator.validate_rules(iter(collection)) for issue in issues: print(f"[{issue.severity.name}] {issue.rule.title}: {issue.description}") # [MEDIUM] Missing Status Rule: Rule has no status # Load validator config from YAML from sigma.validators.core import validators as all_validators validator_yaml = """ validators: - all - -sigma_description_length # exclude one validator exclusions: 12345678-0000-0000-0000-000000000001: - sigma_tags # skip tag validation for this specific rule ID """ validator_from_config = SigmaValidator.from_yaml(validator_yaml, all_validators) ``` -------------------------------- ### Instantiate SigmaPluginDirectory Source: https://github.com/sigmahq/pysigma/blob/main/docs/Plugin_System.md Instantiates the SigmaPluginDirectory with the current content of the plugin directory. Use `from_url()` for alternative directories. ```python plugins = SigmaPluginDirectory.default_plugin_directory() ``` -------------------------------- ### InstalledSigmaPlugins Source: https://context7.com/sigmahq/pysigma/llms.txt Auto-discover backends, pipelines, and validators by scanning namespaces. ```APIDOC ## InstalledSigmaPlugins ### Description `InstalledSigmaPlugins` scans the `sigma.backends`, `sigma.pipelines`, and `sigma.validators` namespaces to automatically discover all installed pySigma plugin packages and expose them as structured dictionaries. ### Method Use the `autodiscover()` class method to scan for plugins. ### Usage ```python from sigma.plugins import InstalledSigmaPlugins # Auto-discover all installed plugins plugins = InstalledSigmaPlugins.autodiscover() print("Backends:", list(plugins.backends.keys())) # e.g. ['splunk', 'elasticsearch', 'qradar', ...] print("Pipelines:", list(plugins.pipelines.keys())) # e.g. ['sysmon', 'ecs_windows', 'crowdstrike', ...] print("Validators:", list(plugins.validators.keys())) # e.g. ['sigma_tags', 'sigma_status', ...] # Instantiate a discovered backend by name if "splunk" in plugins.backends: BackendClass = plugins.backends["splunk"] pipeline_fn = plugins.pipelines.get("sysmon") pipeline = pipeline_fn() if pipeline_fn else None backend = BackendClass(pipeline) from sigma.collection import SigmaCollection rules = SigmaCollection.load_ruleset(["rules/"]) queries = backend.convert(rules) for q in queries: print(q) ``` ``` -------------------------------- ### Implement Custom TextQueryBackend Source: https://context7.com/sigmahq/pysigma/llms.txt Shows how to subclass TextQueryBackend to create a new backend for a custom query language. Configures behavior via class variables for operators, quoting, and expressions. ```python import re from sigma.conversion.base import TextQueryBackend from sigma.processing.pipeline import ProcessingPipeline class MyQueryBackend(TextQueryBackend): """Example backend generating a simple query language.""" name = "MyQuery Backend" formats = {"default": "Plain query string", "json": "JSON-wrapped queries"} # Boolean operators or_token = "OR" and_token = "AND" not_token = "NOT" eq_token = ":" # String quoting and escaping str_quote = '"' escape_char = "\" wildcard_multi = "*" wildcard_single = "?" # Field quoting: quote fields that contain non-word characters field_quote = "`" field_quote_pattern = re.compile(r"^\w+$") field_quote_pattern_negation = True # quote if pattern does NOT match # Operator expressions startswith_expression = "{field}:{value}*" endswith_expression = "{field}:*{value}" contains_expression = "{field}:*{value}*" re_expression = "{field}:/regex:{regex}/" # Null / existence field_null_expression = "{field}:null" field_exists_expression = "_exists_:{field}" # IN-list optimisation convert_or_as_in = True field_in_list_expression = "{field} IN ({list})" or_in_operator = "IN" list_separator = ", " def finalize_output_default(self, queries): return queries # return list[str] def finalize_output_json(self, queries): import json return json.dumps({"queries": queries}) # Usage backend = MyQueryBackend() from sigma.collection import SigmaCollection rules = SigmaCollection.from_yaml(""" title: Test status: test logsource: category: process_creation product: windows detection: sel: CommandLine|contains: 'malware' Image|endswith: '.exe' condition: sel """,) print(backend.convert(rules)) ``` -------------------------------- ### Convert SigmaCollection to Queries with TextQueryTestBackend Source: https://context7.com/sigmahq/pysigma/llms.txt Demonstrates converting a SigmaCollection to queries using the built-in test backend. Shows default output and named formats. Handles conversion errors by collecting them instead of raising exceptions. ```python from sigma.collection import SigmaCollection from sigma.processing.pipeline import ProcessingPipeline, ProcessingItem from sigma.processing.transformations import FieldMappingTransformation # Assumes pySigma-backend-splunk is installed: pip install pySigma-backend-splunk # from sigma.backends.splunk import SplunkBackend # Demonstration using the built-in test backend from sigma.backends.test import TextQueryTestBackend sigma_yaml = """ title: Failed Logon status: stable logsource: category: authentication product: windows detection: sel: EventID: 4625 LogonType: - 2 - 3 SubjectUserName|startswith: 'ADMIN' filter_local: IpAddress: '127.0.0.1' condition: sel and not filter_local level: medium """ pipeline = ProcessingPipeline(items=[ ProcessingItem( FieldMappingTransformation({"EventID": "event.code", "LogonType": "winlog.logon.type"}) ) ]) backend = TextQueryTestBackend(pipeline) collection = SigmaCollection.from_yaml(sigma_yaml) queries = backend.convert(collection) print(queries) # ["event.code=4625 and 'winlog.logon.type' in (2, 3) and 'SubjectUserName' startswith \"ADMIN\" and not 'IpAddress'=\"127.0.0.1\"] # Use a named output format queries_test_fmt = backend.convert(collection, output_format="test") print(type(queries_test_fmt)) # list # Collect conversion errors instead of raising backend_safe = TextQueryTestBackend(collect_errors=True) backend_safe.convert(collection) for rule, error in backend_safe.errors: print(f"Conversion error in '{rule.title}': {error}") ``` -------------------------------- ### Create and Load Multi-event Correlation Rules with SigmaCollection Source: https://context7.com/sigmahq/pysigma/llms.txt Demonstrates how to define a multi-event correlation rule in YAML and load it into a SigmaCollection. The collection can then be inspected to verify the presence of both standard Sigma rules and correlation rules. ```python from sigma.collection import SigmaCollection combined_yaml = """ title: Failed Login Base Rule name: failed_login status: stable logsource: category: authentication product: windows detection: sel: EventID: 4625 condition: sel --- title: Brute Force Detection name: brute_force_detection status: stable type: correlation rules: failed_login: failed_login group-by: - TargetUserName - IpAddress timespan: 5m condition: gte: 10 level: high """ collection = SigmaCollection.from_yaml(combined_yaml) print(len(collection)) # 2: one SigmaRule + one SigmaCorrelationRule from sigma.correlations import SigmaCorrelationRule, SigmaCorrelationType corr_rule = collection[1] assert isinstance(corr_rule, SigmaCorrelationRule) assert corr_rule.type == SigmaCorrelationType.EVENT_COUNT print(corr_rule.timespan.spec) # "5m" ``` -------------------------------- ### Initialize SigmaValidator with all validators Source: https://github.com/sigmahq/pysigma/blob/main/docs/Rule_Validation.md Instantiate SigmaValidator with all available validator checks. This is useful for a comprehensive validation run. ```python from sigma.validators.core import validators rule_validator = SigmaValidator(validators.values()) ``` -------------------------------- ### Build pySigma Package Source: https://github.com/sigmahq/pysigma/blob/main/README.md Build your own pySigma package using the 'poetry build' command. ```bash poetry build ``` -------------------------------- ### Backend.convert Source: https://context7.com/sigmahq/pysigma/llms.txt The main entry point for backend-based conversion. It takes a SigmaCollection and returns converted queries in the backend's default or specified output format. The result type typically depends on the backend implementation, often a list of strings for text-based query languages. ```APIDOC ## Backend.convert ### Description Converts a `SigmaCollection` into target queries using a specified backend. This is the primary method for generating output queries from Sigma rules. ### Method `convert(collection: SigmaCollection, output_format: str = 'default', collect_errors: bool = False)` ### Parameters - **collection** (`SigmaCollection`) - The collection of Sigma rules to convert. - **output_format** (`str`, optional) - The desired output format. Defaults to 'default'. - **collect_errors** (`bool`, optional) - If True, conversion errors are collected instead of raising exceptions. Defaults to False. ### Request Example ```python from sigma.collection import SigmaCollection from sigma.backends.test import TextQueryTestBackend sigma_yaml = """ title: Failed Logon status: stable logsource: category: authentication product: windows detection: sel: EventID: 4625 LogonType: - 2 - 3 SubjectUserName|startswith: 'ADMIN' filter_local: IpAddress: '127.0.0.1' condition: sel and not filter_local level: medium """ backend = TextQueryTestBackend() collection = SigmaCollection.from_yaml(sigma_yaml) queries = backend.convert(collection) print(queries) ``` ### Response #### Success Response - **queries** (`list[str]` or other backend-specific type) - A list of converted queries or the backend's default output. #### Response Example ``` ["event.code=4625 and 'winlog.logon.type' in (2, 3) and 'SubjectUserName' startswith \"ADMIN\" and not 'IpAddress'=\"127.0.0.1\""] ``` ``` -------------------------------- ### Configure Custom MITRE Data URLs Source: https://github.com/sigmahq/pysigma/blob/main/docs/Rule_Validation.md Set custom URLs for MITRE ATT&CK and D3FEND data if you maintain your own mirror. This allows pySigma to fetch data from your specified locations. ```python mitre_attack.set_url("https://your-mirror.example.com/enterprise-attack.json") mitre_d3fend.set_url("https://your-mirror.example.com/d3fend.json") ``` -------------------------------- ### TextQueryBackend Source: https://context7.com/sigmahq/pysigma/llms.txt Base class for implementing new backends that generate text-based query languages. Configuration is done via class variables for tokens, expressions, and quoting rules. ```APIDOC ## TextQueryBackend ### Description Base class for creating custom text-based query language backends in pySigma. Subclass this to define how Sigma rules are translated into a specific query language by configuring various class variables. ### Usage Subclass `TextQueryBackend` and configure its class variables to define the syntax and behavior of your query language. Then, instantiate your custom backend and use its `convert` method. ### Class Variables - **name** (`str`): The name of the backend. - **formats** (`dict`): A dictionary mapping format names to their descriptions. - **or_token**, **and_token**, **not_token** (`str`): Boolean operators. - **eq_token** (`str`): Equality operator. - **str_quote**, **escape_char** (`str`): String quoting and escaping characters. - **wildcard_multi**, **wildcard_single** (`str`): Wildcard characters. - **field_quote**, **field_quote_pattern**, **field_quote_pattern_negation** (`str`, `re.Pattern`, `bool`): Rules for quoting field names. - **startswith_expression**, **endswith_expression**, **contains_expression**, **re_expression** (`str`): Patterns for string matching operators. - **field_null_expression**, **field_exists_expression** (`str`): Expressions for null or existence checks. - **convert_or_as_in** (`bool`): Whether to convert OR operations into IN-lists. - **field_in_list_expression**, **or_in_operator**, **list_separator** (`str`): Configuration for IN-list optimization. ### Methods - **finalize_output_default(self, queries: list[str]) -> str**: Finalizes the output for the default format. - **finalize_output_json(self, queries: list[str]) -> str**: Finalizes the output for the JSON format. ### Request Example ```python import re from sigma.conversion.base import TextQueryBackend from sigma.collection import SigmaCollection class MyQueryBackend(TextQueryBackend): name = "MyQuery Backend" formats = {"default": "Plain query string", "json": "JSON-wrapped queries"} or_token = "OR" and_token = "AND" not_token = "NOT" eq_token = ":" str_quote = '"' escape_char = "\" wildcard_multi = "*" wildcard_single = "?" field_quote = "`" field_quote_pattern = re.compile(r"^\w+$") field_quote_pattern_negation = True startswith_expression = "{field}:{value}*" endswith_expression = "{field}:*{value}" contains_expression = "{field}:*{value}*" re_expression = "{field}:/regex:{regex}/" field_null_expression = "{field}:null" field_exists_expression = "_exists_:{field}" convert_or_as_in = True field_in_list_expression = "{field} IN ({list})" or_in_operator = "IN" list_separator = ", " def finalize_output_default(self, queries): return queries def finalize_output_json(self, queries): import json return json.dumps({"queries": queries}) backend = MyQueryBackend() rules = SigmaCollection.from_yaml(""" title: Test status: test logsource: category: process_creation product: windows detection: sel: CommandLine|contains: 'malware' Image|endswith: '.exe' condition: sel """) print(backend.convert(rules)) ``` ### Response Example ``` ['CommandLine:*malware* AND Image:*.exe'] ``` ``` -------------------------------- ### Load MITRE ATT&CK data from a local file Source: https://github.com/sigmahq/pysigma/blob/main/docs/Rule_Validation.md Configure pySigma to load MITRE ATT&CK data from a local JSON file instead of downloading it automatically. This is useful for environments with restricted internet access. ```python from sigma.data import mitre_attack # Load MITRE ATT&CK data from a local file mitre_attack.set_url("/path/to/enterprise-attack.json") ``` -------------------------------- ### Load MITRE D3FEND Data Locally Source: https://github.com/sigmahq/pysigma/blob/main/docs/Rule_Validation.md Use this to load MITRE D3FEND data from a local JSON file. Ensure the path to the file is correctly specified. ```python mitre_d3fend.set_url("/path/to/d3fend.json") ``` -------------------------------- ### Resolve Processing Pipelines with ProcessingPipelineResolver Source: https://context7.com/sigmahq/pysigma/llms.txt Illustrates how to use `ProcessingPipelineResolver` to map string identifiers or file paths to `ProcessingPipeline` objects. This is useful for composing pipelines from multiple sources in a prioritized manner. ```python from sigma.processing.resolver import ProcessingPipelineResolver from sigma.processing.pipeline import ProcessingPipeline, ProcessingItem from sigma.processing.transformations import FieldMappingTransformation ecs_pipeline = ProcessingPipeline( name="ecs_basic", priority=10, items=[ProcessingItem(FieldMappingTransformation({"CommandLine": "process.command_line"}))], ) custom_pipeline = ProcessingPipeline( name="custom_env", priority=20, items=[ProcessingItem(FieldMappingTransformation({"process.command_line": "cmd_line_full"}))], ) resolver = ProcessingPipelineResolver({ "ecs": ecs_pipeline, "custom": custom_pipeline, }) # Resolve a single pipeline by identifier resolved_ecs = resolver.resolve_pipeline("ecs") # Resolve and merge multiple pipelines in priority order merged_pipeline = resolver.resolve(["ecs", "custom"]) print(merged_pipeline.name) # merged result applying ecs first, then custom # Resolve from a YAML file path pipeline_from_file = resolver.resolve_pipeline("/etc/sigma/my_pipeline.yml") ``` -------------------------------- ### Run All Tests Source: https://github.com/sigmahq/pysigma/blob/main/README.md Execute all tests using pytest. For a coverage report, use the --cov=sigma flag. ```bash pytest ``` ```bash pytest --cov=sigma ``` -------------------------------- ### Convert Sigma rules to Splunk queries Source: https://github.com/sigmahq/pysigma/blob/main/docs/index.md Convert a Sigma rule collection into a list of queries using a specific backend (e.g., SplunkBackend) and an optional processing pipeline. ```default from sigma.pipelines.sysmon import sysmon_pipeline pipeline = sysmon_pipeline() backend = SplunkBackend(pipeline) rules = SigmaCollection.from_yaml(sigma_rule_yaml) print("Result: " + "\n".join(backend.convert(rules))) ``` -------------------------------- ### Load Sigma Ruleset Recursively Source: https://context7.com/sigmahq/pysigma/llms.txt Loads all .yml rule files from specified paths and directories recursively. Supports callbacks for filtering rules during loading and collects errors encountered. ```python from pathlib import Path from sigma.collection import SigmaCollection # Load all rules under a directory tree ruleset = SigmaCollection.load_ruleset( inputs=["rules/windows/", "rules/linux/custom_rule.yml"], collect_errors=True, recursion_pattern="**/*.yml", ) print(f"Loaded {len(ruleset)} rules") # Skip rules with errors for rule in ruleset: if rule.errors: print(f"[WARN] {rule.title}: {rule.errors[0]}") # Filter rules during load with a callback def skip_test_rules(path: Path): if "test" in path.stem: return None # returning None skips this file return path filtered_ruleset = SigmaCollection.load_ruleset( inputs=["rules/"], on_beforeload=skip_test_rules, collect_errors=True, ) ``` -------------------------------- ### Load Processing Pipeline from YAML Source: https://context7.com/sigmahq/pysigma/llms.txt Loads a processing pipeline definition from a YAML string. This method is standard for distributing and sharing pipelines. ```python from sigma.processing.pipeline import ProcessingPipeline pipeline_from_file = ProcessingPipeline.from_yaml(""" name: sysmon_pipeline priority: 10 transformations: - id: field_map type: field_name_mapping mapping: CommandLine: process.args rule_conditions: - type: logsource service: sysmon ") ``` -------------------------------- ### Access rule properties Source: https://github.com/sigmahq/pysigma/blob/main/docs/index.md Access simple rule properties like title or author directly. For complex attributes like logsource or detection, use their respective classes. ```default print("Rule: " + rule.title) ``` ```default if rule.logsource not in SigmaLogSource(None, "windows", "sysmon"): print("This is not a Sysmon rule!") print("Consolidated condition: " + "or".join(rule.detection.condition)) ``` -------------------------------- ### SigmaCollection.load_ruleset Source: https://context7.com/sigmahq/pysigma/llms.txt Recursively loads Sigma rules from specified files and directories into a SigmaCollection. It supports optional callbacks for filtering or transforming rules during the loading process and can collect errors encountered. ```APIDOC ## SigmaCollection.load_ruleset — Load a full ruleset from files or directories Recursively loads all `.yml` rule files from a list of file paths and directories into a single `SigmaCollection`. Accepts optional `on_beforeload` and `on_load` callbacks for filtering or transforming individual rules during loading. ### Parameters - **inputs** (list[str | Path]) - Required - A list of file paths or directories to load rules from. - **collect_errors** (bool) - Optional - If True, errors encountered during loading will be collected. - **recursion_pattern** (str) - Optional - A glob pattern to use for recursive loading of files. - **on_beforeload** (callable) - Optional - A callback function executed before a rule is loaded. Returning `None` skips the rule. - **on_load** (callable) - Optional - A callback function executed after a rule is loaded. ### Request Example ```python from pathlib import Path from sigma.collection import SigmaCollection # Load all rules under a directory tree ruleset = SigmaCollection.load_ruleset( inputs=["rules/windows/", "rules/linux/custom_rule.yml"], collect_errors=True, recursion_pattern="**/*.yml", ) print(f"Loaded {len(ruleset)} rules") # Skip rules with errors for rule in ruleset: if rule.errors: print(f"[WARN] {rule.title}: {rule.errors[0]}") # Filter rules during load with a callback def skip_test_rules(path: Path): if "test" in path.stem: return None # returning None skips this file return path filtered_ruleset = SigmaCollection.load_ruleset( inputs=["rules/"], on_beforeload=skip_test_rules, collect_errors=True, ) ``` ### Response #### Success Response (SigmaCollection) - **ruleset** (SigmaCollection) - A collection of loaded Sigma rules. ``` -------------------------------- ### ProcessingPipeline.from_yaml Source: https://context7.com/sigmahq/pysigma/llms.txt Deserializes a processing pipeline from a YAML string or file. This is the standard method for loading and sharing pipelines, often used for pre-defined configurations like the Sysmon pipeline. ```APIDOC ## ProcessingPipeline.from_yaml — Load a pipeline from YAML Deserialises a processing pipeline from a YAML string or file. This is the standard way to distribute and share pipelines (e.g., the Sysmon pipeline from `pySigma-pipeline-sysmon`). ### Parameters - **yaml_data** (str) - Required - A string containing the YAML definition of the pipeline. ### Request Example ```python from sigma.processing.pipeline import ProcessingPipeline yaml_pipeline = """ name: Custom Field Normalizer priority: 15 transformations: - id: rename_winlog_fields type: field_name_mapping mapping: winlog.event_data.CommandLine: process.command_line winlog.event_data.Image: process.executable rule_conditions: - type: logsource product: windows - id: drop_empty_commandline type: drop_detection_item detection_item_conditions: - type: match_string field: CommandLine pattern: "" postprocessing: - id: wrap_query type: embed prefix: "search " suffix: "" """ pipeline = ProcessingPipeline.from_yaml(yaml_pipeline) print(pipeline.name) # "Custom Field Normalizer" print(pipeline.priority) # 15 ``` ### Response #### Success Response (ProcessingPipeline) - **pipeline** (ProcessingPipeline) - The deserialized processing pipeline object. ``` -------------------------------- ### Deserialize Processing Pipeline from YAML String Source: https://context7.com/sigmahq/pysigma/llms.txt Deserializes a processing pipeline from a YAML string, defining field mappings, conditions, and postprocessing steps. Demonstrates loading a custom pipeline with specific transformations. ```python from sigma.processing.pipeline import ProcessingPipeline yaml_pipeline = """ name: Custom Field Normalizer priority: 15 transformations: - id: rename_winlog_fields type: field_name_mapping mapping: winlog.event_data.CommandLine: process.command_line winlog.event_data.Image: process.executable rule_conditions: - type: logsource product: windows - id: drop_empty_commandline type: drop_detection_item detection_item_conditions: - type: match_string field: CommandLine pattern: "" postprocessing: - id: wrap_query type: embed prefix: "search " suffix: "" """ pipeline = ProcessingPipeline.from_yaml(yaml_pipeline) print(pipeline.name) # "Custom Field Normalizer" print(pipeline.priority) # 15 ``` -------------------------------- ### Configure validators in YAML Source: https://github.com/sigmahq/pysigma/blob/main/docs/Rule_Validation.md Define which validator checks to use or exclude in a YAML configuration. The 'all' keyword enables all validators, with subsequent entries specifying exclusions. ```yaml validators: - all - -tlptag - -tlpv1_tag ``` -------------------------------- ### Define Processing Pipeline with Field Mapping Source: https://context7.com/sigmahq/pysigma/llms.txt Builds a processing pipeline to map Sigma fields to ECS field names for Windows logs and replaces backslashes in paths. Uses `ProcessingItem` with `LogsourceCondition`. ```python from sigma.processing.pipeline import ProcessingPipeline, ProcessingItem from sigma.processing.transformations import ( FieldMappingTransformation, AddFieldnamePrefixTransformation, ReplaceStringTransformation, ) from sigma.processing.conditions import LogsourceCondition # Build a pipeline that maps Sysmon-style fields to ECS field names pipeline = ProcessingPipeline( name="windows_to_ecs", priority=20, items=[ ProcessingItem( transformation=FieldMappingTransformation({ "CommandLine": "process.command_line", "Image": "process.executable", "ParentImage": "process.parent.executable", "User": "user.name", }), rule_conditions=[LogsourceCondition(product="windows")], identifier="windows_field_mapping", ), ProcessingItem( transformation=ReplaceStringTransformation( regex=r"\\", replacement="/", ), rule_conditions=[LogsourceCondition(product="windows")], ), ], ) ``` -------------------------------- ### Lint Code with Black Source: https://github.com/sigmahq/pysigma/blob/main/README.md Format the code using Black. To check for linting errors without reformatting, use the --check flag. ```bash poetry run black ``` ```bash poetry run black --check ``` -------------------------------- ### Load Sigma Rules Collection from YAML String Source: https://context7.com/sigmahq/pysigma/llms.txt Use `SigmaCollection.from_yaml` to parse a YAML string containing one or more Sigma rule documents. This method supports collection actions like `global`, `repeat`, and `reset` for sharing common fields across rules. ```python from sigma.collection import SigmaCollection multi_rule_yaml = """ title: Rule One status: test logsource: category: network_connection product: windows detection: sel: DestinationPort: 4444 condition: sel --- title: Rule Two status: test logsource: category: network_connection product: windows detection: sel: DestinationPort: 1337 condition: sel """ collection = SigmaCollection.from_yaml(multi_rule_yaml) print(len(collection)) # 2 print(collection[0].title) # "Rule One" print(collection[1].title) # "Rule Two" # Iterate over all rules for rule in collection: print(rule.title, rule.logsource.category) ``` -------------------------------- ### Configure Validator Checks Source: https://github.com/sigmahq/pysigma/blob/main/docs/Rule_Validation.md Validator checks can be configured using a dictionary passed to the 'config' parameter. This dictionary maps validator identifiers to parameter-value pairs for their constructors. ```yaml config: description_length: min_length: 100 ``` -------------------------------- ### Parse multiple Sigma rules from YAML Source: https://github.com/sigmahq/pysigma/blob/main/docs/index.md Use SigmaCollection.from_yaml to parse multiple Sigma rules from a YAML string into a SigmaCollection object. ```default rules = SigmaCollection.from_yaml(string_with_sigma_rule_yaml) ``` -------------------------------- ### Use Convenience Validator Base Classes Source: https://github.com/sigmahq/pysigma/blob/main/docs/Rule_Validation.md pySigma offers convenience base classes for validating specific parts of Sigma rules, such as detection definitions, items, values, and tags. ```python from sigma.validators.base import SigmaDetectionValidator, SigmaValueValidator, SigmaTagValueValidator # Example usage (conceptual): # class MyDetectionValidator(SigmaDetectionValidator): # ... # class MyValueValidator(SigmaValueValidator): # ... # class MyTagValidator(SigmaTagValueValidator): # ... ``` -------------------------------- ### Parse a single Sigma rule from YAML Source: https://github.com/sigmahq/pysigma/blob/main/docs/index.md Use SigmaRule.from_yaml to parse a single Sigma rule from a YAML string into a SigmaRule object. ```default rule = SigmaRule.from_yaml(string_with_sigma_rule_yaml) ``` -------------------------------- ### SigmaCollection.from_yaml Source: https://context7.com/sigmahq/pysigma/llms.txt Parses a YAML string that may contain one or multiple Sigma rule documents into a SigmaCollection. It supports collection actions like `global`, `repeat`, and `reset` for managing shared fields. ```APIDOC ## SigmaCollection.from_yaml ### Description Parses a YAML string that may contain one or multiple rule documents (separated by `---`) into a `SigmaCollection`. Supports `action: global`, `action: repeat`, and `action: reset` collection actions to share common fields across multiple rules in a single file. ### Method Signature `SigmaCollection.from_yaml(yaml_string: str) -> SigmaCollection` ### Parameters - **yaml_string** (str) - Required - The YAML string containing one or more Sigma rules. ### Request Example ```python from sigma.collection import SigmaCollection multi_rule_yaml = """ title: Rule One status: test logsource: category: network_connection product: windows detection: sel: DestinationPort: 4444 condition: sel --- title: Rule Two status: test logsource: category: network_connection product: windows detection: sel: DestinationPort: 1337 condition: sel """ collection = SigmaCollection.from_yaml(multi_rule_yaml) print(len(collection)) # 2 print(collection[0].title) # "Rule One" print(collection[1].title) # "Rule Two" # Iterate over all rules for rule in collection: print(rule.title, rule.logsource.category) ``` ### Response - **SigmaCollection** - A collection of SigmaRule objects parsed from the YAML string. ``` -------------------------------- ### ProcessingPipeline Source: https://context7.com/sigmahq/pysigma/llms.txt Defines a transformation pipeline for processing Sigma rules. A pipeline consists of an ordered list of processing items, each containing a transformation and optional conditions for its application. It can also include postprocessing items and finalizers. ```APIDOC ## ProcessingPipeline — Define transformation pipelines A `ProcessingPipeline` consists of an ordered list of `ProcessingItem` objects, each pairing a `Transformation` with optional conditions that control whether the transformation is applied to a given rule or detection item. Pipelines can also include `postprocessing_items` (applied to generated queries) and `finalizers` (applied to the full output set). ### Parameters - **name** (str) - Optional - The name of the pipeline. - **priority** (int) - Optional - The priority of the pipeline, used for ordering. - **items** (list[ProcessingItem]) - Optional - A list of `ProcessingItem` objects to be applied to rules. - **postprocessing_items** (list[ProcessingItem]) - Optional - A list of `ProcessingItem` objects for postprocessing queries. - **finalizers** (list[ProcessingItem]) - Optional - A list of `ProcessingItem` objects for finalization. ### Request Example ```python from sigma.processing.pipeline import ProcessingPipeline, ProcessingItem from sigma.processing.transformations import ( FieldMappingTransformation, AddFieldnamePrefixTransformation, ReplaceStringTransformation, ) from sigma.processing.conditions import LogsourceCondition # Build a pipeline that maps Sysmon-style fields to ECS field names pipeline = ProcessingPipeline( name="windows_to_ecs", priority=20, items=[ ProcessingItem( transformation=FieldMappingTransformation({ "CommandLine": "process.command_line", "Image": "process.executable", "ParentImage": "process.parent.executable", "User": "user.name", }), rule_conditions=[LogsourceCondition(product="windows")], identifier="windows_field_mapping", ), ProcessingItem( transformation=ReplaceStringTransformation( regex=r"\\", replacement="/", ), rule_conditions=[LogsourceCondition(product="windows")], ), ], ) # Load a pipeline from a YAML file pipeline_from_file = ProcessingPipeline.from_yaml(""" name: sysmon_pipeline priority: 10 transformations: - id: field_map type: field_name_mapping mapping: CommandLine: process.args rule_conditions: - type: logsource service: sysmon """) ``` ### Response #### Success Response (ProcessingPipeline) - **pipeline** (ProcessingPipeline) - The created processing pipeline object. ``` -------------------------------- ### Validate a set of Sigma rules Source: https://github.com/sigmahq/pysigma/blob/main/docs/Rule_Validation.md Run validation on a collection of Sigma rules using the configured SigmaValidator. The result is a list of validation issues. ```python issues = rule_validator.validate_rules(sigma_rules) ``` -------------------------------- ### Map String Transformation Source: https://github.com/sigmahq/pysigma/blob/main/docs/Processing_Pipelines.md Maps specific string values to other string values. Can map a single value to another or to a list of values. ```yaml transformations: type: map_string mapping: value1: mapped1 value2: - mapped2A - mapped2B ``` -------------------------------- ### SigmaRule.from_yaml Source: https://context7.com/sigmahq/pysigma/llms.txt Parses a YAML string containing a single Sigma rule document into a SigmaRule object. Errors can be collected instead of raised by setting `collect_errors=True`. ```APIDOC ## SigmaRule.from_yaml ### Description Parses a YAML string containing exactly one Sigma rule document into a `SigmaRule` object. All standard rule fields (`title`, `id`, `author`, `logsource`, `detection`, `tags`, `level`, etc.) are available as typed attributes. Set `collect_errors=True` to capture parsing errors in `rule.errors` instead of raising immediately. ### Method Signature `SigmaRule.from_yaml(yaml_string: str, collect_errors: bool = False) -> SigmaRule` ### Parameters - **yaml_string** (str) - Required - The YAML string containing the Sigma rule. - **collect_errors** (bool) - Optional - If True, parsing errors are collected in `rule.errors` instead of raising an exception. ### Request Example ```python from sigma.rule import SigmaRule yaml_str = """ title: Suspicious PowerShell Execution id: 12345678-1234-1234-1234-123456789abc status: stable description: Detects suspicious PowerShell command-line patterns author: Jane Doe date: 2024-01-15 logsource: category: process_creation product: windows detection: sel: CommandLine|contains: - '-EncodedCommand' - '-Enc ' - 'IEX' Image|endswith: 'powershell.exe' condition: sel level: high tags: - attack.execution - attack.t1059.001 """ rule = SigmaRule.from_yaml(yaml_str) print(rule.title) # "Suspicious PowerShell Execution" print(rule.level) # SigmaLevel.HIGH print(str(rule.id)) # "12345678-1234-1234-1234-123456789abc" print(rule.tags[0].name) # "attack.execution" print(rule.logsource.product) # "windows" print(rule.logsource.category) # "process_creation" # Collect errors instead of raising rule_with_errors = SigmaRule.from_yaml(yaml_str, collect_errors=True) if rule_with_errors.errors: for err in rule_with_errors.errors: print(f"Parse error: {err}") ``` ### Response - **SigmaRule** - The parsed Sigma rule object. ``` -------------------------------- ### Parametrize Validator Checks with Dataclasses Source: https://github.com/sigmahq/pysigma/blob/main/docs/Rule_Validation.md Parametrize validator checks by decorating the class with @dataclass(frozen=True). Parameters are defined as dataclass members and passed as keyword arguments to the constructor. ```python from dataclasses import dataclass from sigma.validators.base import SigmaRuleValidator @dataclass(frozen=True) class MyParametrizedValidator(SigmaRuleValidator): min_length: int def validate(self, rule): # Use self.min_length for validation pass ``` -------------------------------- ### SigmaRule.from_dict Source: https://context7.com/sigmahq/pysigma/llms.txt Converts a Python dictionary, typically loaded from a YAML parser, into a SigmaRule object. This is useful when rules are sourced from sources other than direct YAML files. ```APIDOC ## SigmaRule.from_dict ### Description Converts a dict (e.g., already loaded by a YAML parser) into a `SigmaRule` object. Useful when rules are sourced from databases, APIs, or custom pre-processors rather than YAML files directly. ### Method Signature `SigmaRule.from_dict(rule_dict: dict) -> SigmaRule` ### Parameters - **rule_dict** (dict) - Required - A dictionary representing the Sigma rule. ### Request Example ```python from sigma.rule import SigmaRule rule_dict = { "title": "Mimikatz in Memory", "status": "experimental", "logsource": {"category": "process_creation", "product": "windows"}, "detection": { "sel": {"CommandLine|contains": "sekurlsa"}, "condition": "sel", }, "level": "critical", } rule = SigmaRule.from_dict(rule_dict) print(rule.title) # "Mimikatz in Memory" print(rule.level) # SigmaLevel.CRITICAL # Serialise back to dict d = rule.to_dict() assert d["title"] == "Mimikatz in Memory" ``` ### Response - **SigmaRule** - The converted Sigma rule object. ``` -------------------------------- ### Parse Sigma Rule from YAML String Source: https://context7.com/sigmahq/pysigma/llms.txt Use `SigmaRule.from_yaml` to parse a single Sigma rule from a YAML string. Set `collect_errors=True` to capture parsing errors in `rule.errors` instead of raising them immediately. ```python from sigma.rule import SigmaRule from sigma.exceptions import SigmaError yaml_str = """ title: Suspicious PowerShell Execution id: 12345678-1234-1234-1234-123456789abc status: stable description: Detects suspicious PowerShell command-line patterns author: Jane Doe date: 2024-01-15 logsource: category: process_creation product: windows detection: sel: CommandLine|contains: - '-EncodedCommand' - '-Enc ' - 'IEX' Image|endswith: 'powershell.exe' condition: sel level: high tags: - attack.execution - attack.t1059.001 """ rule = SigmaRule.from_yaml(yaml_str) print(rule.title) # "Suspicious PowerShell Execution" print(rule.level) # SigmaLevel.HIGH print(str(rule.id)) # "12345678-1234-1234-1234-123456789abc" print(rule.tags[0].name) # "attack.execution" print(rule.logsource.product) # "windows" print(rule.logsource.category) # "process_creation" # Collect errors instead of raising rule_with_errors = SigmaRule.from_yaml(yaml_str, collect_errors=True) if rule_with_errors.errors: for err in rule_with_errors.errors: print(f"Parse error: {err}") ``` -------------------------------- ### Apply Sigma Filters for False-Positive Suppression Source: https://context7.com/sigmahq/pysigma/llms.txt Sigma filter rules narrow detection scope by applying additional exclusion conditions. Filters can be automatically applied when loading a collection or applied manually after collecting them separately. ```python from sigma.collection import SigmaCollection rules_and_filters_yaml = """ title: Suspicious Net.exe Execution name: suspicious_net_exe status: stable logsource: category: process_creation product: windows detection: sel: Image|endswith: '\\net.exe' CommandLine|contains: 'user' condition: sel level: medium --- title: Filter - Domain Controllers filter: rules: - suspicious_net_exe definition: Exclude known domain controller activity selection: SubjectUserName|endswith: '$' condition: not selection """ collection = SigmaCollection.from_yaml( rules_and_filters_yaml, collect_filters=False, # False = automatically apply filters to rules ) for rule in collection: print(rule.title) # The filter condition is merged into the rule's detection automatically # Apply filters manually after collecting them separately collection_manual = SigmaCollection.from_yaml( rules_and_filters_yaml, collect_filters=True, # True = collect but do NOT apply yet ) collection_manual.apply_filters(collection_manual.filters) ``` -------------------------------- ### Implement Custom Sigma Rule Validator Source: https://github.com/sigmahq/pysigma/blob/main/docs/Rule_Validation.md Create custom validator checks by inheriting from SigmaRuleValidator. The validate() method is called for each rule, and finalize() can be used for checks after all rules are processed. ```python from sigma.validators.base import SigmaRuleValidator class MyCustomValidator(SigmaRuleValidator): def validate(self, rule): # Perform checks on the rule pass def finalize(self): # Perform checks after all rules are processed pass ``` -------------------------------- ### SigmaValidator Source: https://context7.com/sigmahq/pysigma/llms.txt Validate Sigma rules against configurable rule validators to detect quality and consistency issues. ```APIDOC ## SigmaValidator ### Description `SigmaValidator` runs a set of `SigmaRuleValidator` subclasses across one or more rules to detect quality and consistency issues (e.g., missing title, invalid tags, duplicate IDs). Validators maintain state across all rules so uniqueness checks work correctly. ### Method Instantiate `SigmaValidator` with a list of validator classes. ### Usage ```python from sigma.validation import SigmaValidator from sigma.validators.core import ( SigmaTagsValidator, SigmaStatusExistsValidator, SigmaDescriptionLengthValidator, ) from sigma.collection import SigmaCollection collection = SigmaCollection.from_yaml(""" title: Missing Status Rule logsource: category: process_creation product: windows detection: sel: CommandLine: evil.exe condition: sel """, collect_errors=True) validator = SigmaValidator( validators=[ SigmaTagsValidator, SigmaStatusExistsValidator, SigmaDescriptionLengthValidator, ] ) issues = validator.validate_rules(iter(collection)) for issue in issues: print(f"[{issue.severity.name}] {issue.rule.title}: {issue.description}") # [MEDIUM] Missing Status Rule: Rule has no status ``` ### Loading Validator Configuration from YAML ### Usage ```python from sigma.validators.core import validators as all_validators validator_yaml = """ validators: - all - -sigma_description_length # exclude one validator exclusions: 12345678-0000-0000-0000-000000000001: - sigma_tags # skip tag validation for this specific rule ID """ validator_from_config = SigmaValidator.from_yaml(validator_yaml, all_validators) ``` ```