diff --git a/.gitignore b/.gitignore index 0f7e1118..536bb35e 100644 --- a/.gitignore +++ b/.gitignore @@ -8,4 +8,4 @@ coverage.xml .python-version .idea .venv -.mypy_cache/ \ No newline at end of file +.mypy_cache/ diff --git a/docs/guides/index.rst b/docs/guides/index.rst index 4d6ff82f..e3889ea2 100644 --- a/docs/guides/index.rst +++ b/docs/guides/index.rst @@ -12,3 +12,4 @@ These guides provide step-by-step instructions for common tasks with pySigma. processing_pipelines rule_validation plugin_system + matching diff --git a/docs/guides/matching.rst b/docs/guides/matching.rst new file mode 100644 index 00000000..d1dd55e2 --- /dev/null +++ b/docs/guides/matching.rst @@ -0,0 +1,245 @@ +Matching Events Against Rules +============================== + +The ``sigma.matching`` module allows you to evaluate log events directly against Sigma rules in +Python, without converting rules to a query language first. It is useful for testing, rapid +prototyping, and offline analysis where query-language conversion is not required. + +.. admonition:: Performance Note + + pySigma's matching engine is **not designed for high-throughput production workloads**. It is a + pure-Python interpreter optimised for correctness and ease of use. If your use case demands + performant, production-grade event matching, consider the Rust-based Sigma engine + `rsigma `_, which is purpose-built for speed. + +Overview +-------- + +The matching pipeline has three layers: + +1. **Event** — wraps a raw log record with a timestamp and logsource descriptor. +2. **SigmaMatcher** — orchestrates matching: routes events to the right rules and handles + correlation rule state. +3. **SigmaMatch** — the result returned when a rule fires. + +Quick Start +----------- + +Load a rule collection, construct a :class:`~sigma.matching.SigmaMatcher`, and call +:meth:`~sigma.matching.SigmaMatcher.match` for each incoming event: + +.. code-block:: python + + from datetime import datetime, timezone + from sigma.collection import SigmaCollection + from sigma.matching import Event, EventLogSource, SigmaMatcher + + rules_yaml = """ + title: Suspicious PowerShell Encoded Command + status: test + logsource: + category: process_creation + product: windows + detection: + selection: + CommandLine|contains: '-EncodedCommand' + condition: selection + """ + + collection = SigmaCollection.from_yaml(rules_yaml) + matcher = SigmaMatcher(collection) + + event = Event( + timestamp=datetime.now(timezone.utc), + logsource=EventLogSource(category="process_creation", product="windows"), + data={ + "Image": "C:\\Windows\\System32\\WindowsPowerShell\\v1.0\\powershell.exe", + "CommandLine": "powershell.exe -EncodedCommand dABlAHMAdAA=", + }, + ) + + matches = matcher.match(event) + for m in matches: + print(f"Rule fired: {m.rule.title}") + +Creating Events +--------------- + +An :class:`~sigma.matching.Event` carries three required fields: + +* ``timestamp`` — a timezone-aware :class:`datetime` indicating when the event occurred. +* ``logsource`` — an :class:`~sigma.matching.EventLogSource` with optional ``category``, + ``product``, and ``service`` fields that mirror the Sigma rule ``logsource`` block. +* ``data`` — a flat ``dict[str, Any]`` containing the parsed event fields. + +An optional ``raw`` string can be supplied when keyword (field-less) detection items need to +search the raw log text: + +.. code-block:: python + + event = Event( + timestamp=datetime.now(timezone.utc), + logsource=EventLogSource(product="windows", service="security"), + data={"EventID": 4625, "TargetUserName": "Administrator"}, + raw="...original log line...", + ) + +If ``raw`` is omitted, the matcher falls back to a JSON serialisation of ``data`` for keyword +searches. + +Logsource Routing +----------------- + +:class:`~sigma.matching.SigmaMatcher` dispatches each event only to rules whose ``logsource`` +block is *contained by* the event's logsource. This mirrors how Sigma logsource matching works: +a rule with ``product: windows`` will fire for an event with both ``product: windows`` and +``service: security``, but not the other way around. + +.. code-block:: python + + # This event matches rules that require only `product: windows` + # as well as rules that require both `product: windows, service: security`. + event = Event( + timestamp=datetime.now(timezone.utc), + logsource=EventLogSource(product="windows", service="security"), + data={"EventID": 4624}, + ) + +Applying a Processing Pipeline +------------------------------- + +Pass a :class:`~sigma.processing.pipeline.ProcessingPipeline` to +:class:`~sigma.matching.SigmaMatcher` to apply field-name mappings or other transformations +before matching: + +.. code-block:: python + + from sigma.processing.pipeline import ProcessingPipeline + from sigma.processing.transformations import FieldMappingTransformation + from sigma.processing.conditions import LogsourceCondition + + pipeline = ProcessingPipeline( + transformations=[ + FieldMappingTransformation({"CommandLine": "process.command_line"}), + ] + ) + + matcher = SigmaMatcher(collection, pipeline=pipeline) + + event = Event( + timestamp=datetime.now(timezone.utc), + logsource=EventLogSource(category="process_creation", product="windows"), + data={"process.command_line": "powershell.exe -EncodedCommand dABlAHMAdAA="}, + ) + + matches = matcher.match(event) + +Bulk Matching +------------- + +When processing a stream or batch of events, use +:meth:`~sigma.matching.SigmaMatcher.bulk_match` to iterate over all events and collect every +:class:`~sigma.matching.SigmaMatch` in one call: + +.. code-block:: python + + events = [ + Event( + timestamp=datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc), + logsource=EventLogSource(category="process_creation", product="windows"), + data={"CommandLine": "cmd.exe /c whoami"}, + ), + Event( + timestamp=datetime(2024, 1, 1, 12, 0, 5, tzinfo=timezone.utc), + logsource=EventLogSource(category="process_creation", product="windows"), + data={"CommandLine": "powershell.exe -EncodedCommand dABlAHMAdAA="}, + ), + ] + + all_matches = matcher.bulk_match(events) + print(f"{len(all_matches)} match(es) found") + +Working with Matches +-------------------- + +Each :class:`~sigma.matching.SigmaMatch` exposes: + +* ``rule`` — the :class:`~sigma.rule.rule.SigmaRule` (or correlation rule) that fired. +* ``events`` — the list of :class:`~sigma.matching.Event` objects that caused the match. +* ``timestamp`` — the timestamp of the triggering event. +* ``group_values`` — for correlation rules, the ``group-by`` field values that formed the + matching group (``None`` for plain rules). + +.. code-block:: python + + for m in matches: + print(f"Rule : {m.rule.title}") + print(f"At : {m.timestamp}") + print(f"Events : {len(m.events)}") + if m.group_values: + print(f"Group : {m.group_values}") + +Using as a Context Manager +-------------------------- + +:class:`~sigma.matching.SigmaMatcher` implements the context manager protocol, which is useful +when integrating with resource-managed environments: + +.. code-block:: python + + with SigmaMatcher(collection) as matcher: + for event in event_stream: + for m in matcher.match(event): + handle_match(m) + +Correlation Rules +----------------- + +Correlation rules are handled transparently. When a base rule fires, the resulting +:class:`~sigma.matching.SigmaMatch` is forwarded internally to any +:class:`~sigma.matching.SigmaCorrelationRuleMatcher` that references that base rule by name. +The correlation matcher maintains a sliding time window and emits a new +:class:`~sigma.matching.SigmaMatch` whenever its condition is satisfied. + +.. code-block:: python + + rules_yaml = """ + --- + title: Failed Logon + name: failed_logon + status: test + logsource: + product: windows + service: security + detection: + selection: + EventID: 4625 + condition: selection + --- + title: Brute Force Detection + name: brute_force + status: test + type: event_count + rules: + failed_logon: '*' + group-by: + - TargetUserName + timespan: 5m + condition: + gte: 10 + """ + + collection = SigmaCollection.from_yaml(rules_yaml) + matcher = SigmaMatcher(collection) + + for event in event_stream: + for m in matcher.match(event): + if m.rule.name == "brute_force": + print(f"Brute force against: {m.group_values}") + +Two constructor parameters control memory use for correlation state: + +* ``max_events_per_match`` (default ``100``) — caps the number of events stored per + correlation match to avoid unbounded memory growth on high-cardinality rules. +* ``max_correlation_state`` (default ``None``, unlimited) — limits the total number of + entries retained in the sliding window per group key. diff --git a/docs/reference/index.rst b/docs/reference/index.rst index b9b87393..e7d77dda 100644 --- a/docs/reference/index.rst +++ b/docs/reference/index.rst @@ -17,4 +17,5 @@ Complete reference documentation for all pySigma modules, classes, and functions types modifiers correlations + matching exceptions diff --git a/docs/reference/matching.rst b/docs/reference/matching.rst new file mode 100644 index 00000000..0ede05d6 --- /dev/null +++ b/docs/reference/matching.rst @@ -0,0 +1,74 @@ +Matching +======== + +The ``sigma.matching`` package provides a pure-Python runtime for evaluating log events directly +against Sigma rules, without converting them to a query language. + +.. admonition:: Performance Note + + This engine is **not designed for high-throughput production workloads**. It prioritises + correctness and ease of use over speed. For performance-sensitive use cases, consider the + Rust-based Sigma engine `rsigma `_. + +See also: :doc:`/guides/matching` for a step-by-step introduction with examples. + +Event +----- + +.. autoclass:: sigma.matching.Event + :members: + :undoc-members: + :show-inheritance: + +.. autoclass:: sigma.matching.EventLogSource + :members: + :undoc-members: + :show-inheritance: + +SigmaMatcher +------------ + +.. autoclass:: sigma.matching.SigmaMatcher + :members: + :undoc-members: + :show-inheritance: + +SigmaMatch +---------- + +.. autoclass:: sigma.matching.SigmaMatch + :members: + :undoc-members: + :show-inheritance: + +SigmaRuleMatcher +---------------- + +.. autoclass:: sigma.matching.SigmaRuleMatcher + :members: + :undoc-members: + :show-inheritance: + +SigmaCorrelationRuleMatcher +--------------------------- + +.. autoclass:: sigma.matching.SigmaCorrelationRuleMatcher + :members: + :undoc-members: + :show-inheritance: + +Internal Components +------------------- + +The following classes are used internally by :class:`~sigma.matching.SigmaMatcher` and are not +part of the public API, but are documented here for completeness. + +.. autoclass:: sigma.matching.compiler.MatchCompiler + :members: + :undoc-members: + :show-inheritance: + +.. autoclass:: sigma.matching.state.SlidingWindowState + :members: + :undoc-members: + :show-inheritance: diff --git a/sigma/matching/__init__.py b/sigma/matching/__init__.py new file mode 100644 index 00000000..1dde9ae8 --- /dev/null +++ b/sigma/matching/__init__.py @@ -0,0 +1,14 @@ +from sigma.matching.event import Event, EventLogSource +from sigma.matching.match import SigmaMatch +from sigma.matching.matcher import SigmaMatcher +from sigma.matching.rule_matcher import SigmaRuleMatcher +from sigma.matching.correlation_matcher import SigmaCorrelationRuleMatcher + +__all__ = [ + "Event", + "EventLogSource", + "SigmaMatch", + "SigmaMatcher", + "SigmaRuleMatcher", + "SigmaCorrelationRuleMatcher", +] diff --git a/sigma/matching/channel.py b/sigma/matching/channel.py new file mode 100644 index 00000000..3604c145 --- /dev/null +++ b/sigma/matching/channel.py @@ -0,0 +1,39 @@ +from __future__ import annotations + +import queue +from abc import ABC, abstractmethod +from typing import Generic, TypeVar + +T = TypeVar("T") + + +class MessageChannel(ABC, Generic[T]): + """Abstract message channel for passing typed messages between components.""" + + @abstractmethod + def send(self, message: T) -> None: ... + + @abstractmethod + def receive(self, timeout: float | None = None) -> T | None: ... + + def close(self) -> None: + pass + + +class InProcessChannel(MessageChannel[T]): + """Queue-based in-process channel.""" + + def __init__(self) -> None: + self._queue: queue.Queue[T] = queue.Queue() + + def send(self, message: T) -> None: + self._queue.put(message) + + def receive(self, timeout: float | None = None) -> T | None: + try: + return self._queue.get(timeout=timeout) + except queue.Empty: + return None + + def close(self) -> None: + pass diff --git a/sigma/matching/compiler.py b/sigma/matching/compiler.py new file mode 100644 index 00000000..312933da --- /dev/null +++ b/sigma/matching/compiler.py @@ -0,0 +1,233 @@ +from __future__ import annotations + +import re +from ipaddress import ip_address, ip_network +from typing import Any, Callable + +from sigma.conditions import ( + ConditionAND, + ConditionFieldEqualsValueExpression, + ConditionItem, + ConditionNOT, + ConditionOR, + ConditionValueExpression, +) +from sigma.rule.rule import SigmaRule +from sigma.types import ( + CompareOperators, + SigmaBool, + SigmaCasedString, + SigmaCIDRExpression, + SigmaCompareExpression, + SigmaExists, + SigmaExpansion, + SigmaFieldReference, + SigmaNull, + SigmaNumber, + SigmaRegularExpression, + SigmaRegularExpressionFlag, + SigmaString, + SpecialChars, +) + +# Type for compiled matcher callable: (data_dict, raw_log_or_none) -> bool +CompiledMatcher = Callable[[dict[str, Any], str | None], bool] + + +class MatchCompiler: + """Compiles Sigma rule detection logic into optimized Python callables.""" + + def compile_rule(self, rule: SigmaRule) -> CompiledMatcher: + """Compile a SigmaRule's conditions into a single callable matcher.""" + # A rule can have multiple conditions (one per condition string). + # They are OR-linked by default in pySigma. + conditions = rule.detection.parsed_condition + closures: dict[str, Any] = {} + parts: list[str] = [] + + for cond in conditions: + parsed = cond.parsed + if parsed is None: + parts.append("False") + else: + parts.append(self._compile_node(parsed, closures)) + + if len(parts) == 1: + expr = parts[0] + else: + expr = "(" + " or ".join(parts) + ")" + + func_code = f"def _match(data, raw):\n return {expr}\n" + namespace: dict[str, Any] = {"__builtins__": {"str": str}} + namespace.update(closures) + code_obj = compile(func_code, "", "exec") + exec(code_obj, namespace) # noqa: S102 + return namespace["_match"] # type: ignore[no-any-return] + + def _compile_node(self, node: Any, closures: dict[str, Any]) -> str: + if isinstance(node, ConditionAND): + sub_list = [self._compile_node(arg, closures) for arg in node.args] + return "(" + " and ".join(sub_list) + ")" + elif isinstance(node, ConditionOR): + sub_list = [self._compile_node(arg, closures) for arg in node.args] + return "(" + " or ".join(sub_list) + ")" + elif isinstance(node, ConditionNOT): + sub_str = self._compile_node(node.args[0], closures) + return f"(not {sub_str})" + elif isinstance(node, ConditionFieldEqualsValueExpression): + return self._compile_field_value(node.field, node.value, closures) + elif isinstance(node, ConditionValueExpression): + return self._compile_keyword(node.value, closures) + else: + return "False" + + def _compile_field_value(self, field_name: str, value: Any, closures: dict[str, Any]) -> str: + if isinstance(value, SigmaNull): + return f"(data.get({field_name!r}) is None)" + + elif isinstance(value, SigmaBool): + bool_val = bool(value) + return f"(data.get({field_name!r}) is {bool_val})" + + elif isinstance(value, SigmaExists): + if value.exists: + return f"({field_name!r} in data)" + else: + return f"({field_name!r} not in data)" + + elif isinstance(value, SigmaFieldReference): + ref_field = value.field + if value.starts_with and value.ends_with: + # contains + return f"(str(data.get({ref_field!r}, '')) in str(data.get({field_name!r}, '')))" + elif value.starts_with: + return f"(str(data.get({field_name!r}, '')).startswith(str(data.get({ref_field!r}, ''))))" + elif value.ends_with: + return f"(str(data.get({field_name!r}, '')).endswith(str(data.get({ref_field!r}, ''))))" + else: + return f"(data.get({field_name!r}) == data.get({ref_field!r}))" + + elif isinstance(value, SigmaCompareExpression): + num_val = value.number.number + op_map = { + CompareOperators.LT: "<", + CompareOperators.LTE: "<=", + CompareOperators.GT: ">", + CompareOperators.GTE: ">=", + CompareOperators.NEQ: "!=", + } + op_str = op_map[value.op] + return f"(data.get({field_name!r}, 0) {op_str} {num_val!r})" + + elif isinstance(value, SigmaCIDRExpression): + var_name = f"_cidr_{len(closures)}" + closures[var_name] = value.network + closures["_try_cidr"] = _try_cidr + return f"(_try_cidr(data.get({field_name!r}, ''), {var_name}))" + + elif isinstance(value, SigmaExpansion): + # OR over all expansion values + sub_parts = [] + for sub_val in value.values: + sub_parts.append(self._compile_field_value(field_name, sub_val, closures)) + return "(" + " or ".join(sub_parts) + ")" + + elif isinstance(value, SigmaRegularExpression): + var_name = f"_re_{len(closures)}" + flags = 0 + for flag in value.flags: + flags |= value.sigma_to_python_flags[flag] + closures[var_name] = re.compile(str(value.regexp), flags) + return f"({var_name}.search(str(data.get({field_name!r}, ''))) is not None)" + + elif isinstance(value, SigmaNumber): + num = value.number + return f"(data.get({field_name!r}) == {num!r})" + + elif isinstance(value, SigmaCasedString): + # Case-sensitive matching + return self._compile_string_match(field_name, value, closures, case_sensitive=True) + + elif isinstance(value, SigmaString): + # Case-insensitive matching + return self._compile_string_match(field_name, value, closures, case_sensitive=False) + + else: + return "False" + + def _compile_string_match( + self, + field_name: str, + value: SigmaString, + closures: dict[str, Any], + case_sensitive: bool, + ) -> str: + if not value.contains_special(): + # Plain string equality + plain = str(value) + if case_sensitive: + return f"(str(data.get({field_name!r}, '')) == {plain!r})" + else: + lower_val = plain.lower() + return f"(str(data.get({field_name!r}, '')).lower() == {lower_val!r})" + else: + # Contains wildcards - compile to regex + regex_str = self._sigma_string_to_regex(value) + flags = 0 if case_sensitive else re.IGNORECASE + var_name = f"_pat_{len(closures)}" + closures[var_name] = re.compile(regex_str, flags) + return f"({var_name}.search(str(data.get({field_name!r}, ''))) is not None)" + + def _sigma_string_to_regex(self, value: SigmaString) -> str: + """Convert a SigmaString with wildcards to a regex pattern (anchored).""" + parts: list[str] = [] + for item in value.s: + if isinstance(item, str): + parts.append(re.escape(item)) + elif item == SpecialChars.WILDCARD_MULTI: + parts.append(".*") + elif item == SpecialChars.WILDCARD_SINGLE: + parts.append(".") + pattern = "".join(parts) + # Anchor: if no leading wildcard, anchor start; if no trailing, anchor end + if not value.startswith(SpecialChars.WILDCARD_MULTI): + pattern = "^" + pattern + if not value.endswith(SpecialChars.WILDCARD_MULTI): + pattern = pattern + "$" + return pattern + + def _compile_keyword(self, value: Any, closures: dict[str, Any]) -> str: + """Compile a keyword (value-only) expression that searches in raw log.""" + if isinstance(value, SigmaString): + if not value.contains_special(): + plain = str(value).lower() + return f"({plain!r} in (raw or '').lower())" + else: + regex_str = self._sigma_string_to_regex(value) + # For keywords, remove anchors - search anywhere + regex_str = regex_str.lstrip("^").rstrip("$") + var_name = f"_kw_{len(closures)}" + closures[var_name] = re.compile(regex_str, re.IGNORECASE) + return f"({var_name}.search(raw or '') is not None)" + elif isinstance(value, SigmaRegularExpression): + var_name = f"_kw_{len(closures)}" + flags = 0 + for flag in value.flags: + flags |= value.sigma_to_python_flags[flag] + closures[var_name] = re.compile(str(value.regexp), flags) + return f"({var_name}.search(raw or '') is not None)" + elif isinstance(value, SigmaExpansion): + sub_parts = [] + for sub_val in value.values: + sub_parts.append(self._compile_keyword(sub_val, closures)) + return "(" + " or ".join(sub_parts) + ")" + else: + return "False" + + +def _try_cidr(value: Any, network: Any) -> bool: + """Check if value is an IP within the given network.""" + try: + return ip_address(str(value)) in network + except (ValueError, TypeError): + return False diff --git a/sigma/matching/correlation_matcher.py b/sigma/matching/correlation_matcher.py new file mode 100644 index 00000000..1c65e69a --- /dev/null +++ b/sigma/matching/correlation_matcher.py @@ -0,0 +1,380 @@ +from __future__ import annotations + +from collections import defaultdict +from statistics import median +from typing import Any + +from sigma.correlations import ( + CorrelationConditionAND, + CorrelationConditionItem, + CorrelationConditionNOT, + CorrelationConditionOR, + SigmaCorrelationCondition, + SigmaCorrelationConditionOperator, + SigmaCorrelationRule, + SigmaCorrelationType, + SigmaExtendedCorrelationCondition, + SigmaRuleReference, +) +from sigma.matching.event import Event +from sigma.matching.match import SigmaMatch +from sigma.matching.state import SlidingWindowState + + +class SigmaCorrelationRuleMatcher: + """Evaluates a SigmaCorrelationRule by tracking matches from referenced base rules.""" + + def __init__( + self, + rule: SigmaCorrelationRule, + max_state: int | None = None, + max_events_per_match: int | None = 100, + ) -> None: + self.rule = rule + self.max_events_per_match = max_events_per_match + self._state = SlidingWindowState( + window_seconds=rule.timespan.seconds, + max_entries=max_state, + ) + # Track which base rule names are relevant + self._rule_names: set[str] = set() + if rule.rules: + for ref in rule.rules: + self._rule_names.add(ref.reference) + elif isinstance(rule.condition, SigmaExtendedCorrelationCondition): + for name in rule.condition.get_referenced_rules(): + self._rule_names.add(name) + + def receive_match(self, match: SigmaMatch, source_rule_name: str) -> list[SigmaMatch]: + """Process an incoming match from a base rule and return correlation matches if any.""" + group_key = self._extract_group_key(match) + self._state.add(group_key, match) + + window_matches = self._state.get_window(group_key, match.timestamp) + result = self._evaluate(window_matches, group_key, match.timestamp) + return result + + def _extract_group_key(self, match: SigmaMatch) -> tuple[Any, ...]: + """Extract group-by field values from match events.""" + if not self.rule.group_by: + return () + + values: list[Any] = [] + event = match.events[0] if match.events else None + if event is None: + return tuple(None for _ in self.rule.group_by) + + for field_name in self.rule.group_by: + # Check aliases + resolved_field = self._resolve_alias(field_name, match) + values.append(event.data.get(resolved_field)) + return tuple(values) + + def _resolve_alias(self, field_name: str, match: SigmaMatch) -> str: + """Resolve field alias to actual field name based on source rule.""" + if field_name in self.rule.aliases.aliases: + alias = self.rule.aliases.aliases[field_name] + # Try to find mapping for the source rule + for ref, mapped_field in alias.mapping.items(): + if hasattr(ref, "rule") and ref.rule is match.rule: + return mapped_field + return field_name + + def _evaluate( + self, + window_matches: list[SigmaMatch], + group_key: tuple[Any, ...], + now: Any, + ) -> list[SigmaMatch]: + """Evaluate the correlation condition against current window state.""" + ctype = self.rule.type + condition = self.rule.condition + + if ctype == SigmaCorrelationType.EVENT_COUNT: + return self._eval_event_count(window_matches, group_key, now, condition) + elif ctype == SigmaCorrelationType.VALUE_COUNT: + return self._eval_value_count(window_matches, group_key, now, condition) + elif ctype == SigmaCorrelationType.TEMPORAL: + return self._eval_temporal(window_matches, group_key, now) + elif ctype == SigmaCorrelationType.TEMPORAL_ORDERED: + return self._eval_temporal_ordered(window_matches, group_key, now) + elif ctype == SigmaCorrelationType.VALUE_SUM: + return self._eval_value_aggregate(window_matches, group_key, now, condition, "sum") + elif ctype == SigmaCorrelationType.VALUE_AVG: + return self._eval_value_aggregate(window_matches, group_key, now, condition, "avg") + elif ctype == SigmaCorrelationType.VALUE_MEDIAN: + return self._eval_value_aggregate(window_matches, group_key, now, condition, "median") + elif ctype == SigmaCorrelationType.VALUE_PERCENTILE: + return self._eval_value_aggregate( + window_matches, group_key, now, condition, "percentile" + ) + return [] + + def _compare( + self, op: SigmaCorrelationConditionOperator, value: Any, threshold: int | float + ) -> bool: + """Compare a value against a threshold using the given operator.""" + if value is None: + return False + op_map = { + SigmaCorrelationConditionOperator.LT: lambda a, b: a < b, + SigmaCorrelationConditionOperator.LTE: lambda a, b: a <= b, + SigmaCorrelationConditionOperator.GT: lambda a, b: a > b, + SigmaCorrelationConditionOperator.GTE: lambda a, b: a >= b, + SigmaCorrelationConditionOperator.EQ: lambda a, b: a == b, + SigmaCorrelationConditionOperator.NEQ: lambda a, b: a != b, + } + return bool(op_map[op](value, threshold)) + + def _collect_events(self, matches: list[SigmaMatch]) -> list[Event]: + """Collect events from matches, respecting the max_events_per_match limit.""" + events: list[Event] = [] + limit = self.max_events_per_match + if limit is None: + for m in matches: + events.extend(m.events) + else: + for m in matches: + for e in m.events: + if len(events) >= limit: + return events + events.append(e) + return events + + def _make_group_values(self, group_key: tuple[Any, ...]) -> dict[str, Any] | None: + if not self.rule.group_by: + return None + return dict(zip(self.rule.group_by, group_key)) + + def _eval_event_count( + self, + matches: list[SigmaMatch], + group_key: tuple[Any, ...], + now: Any, + condition: Any, + ) -> list[SigmaMatch]: + if not isinstance(condition, SigmaCorrelationCondition): + return [] + count = len(matches) + if self._compare(condition.op, count, condition.count): + return [ + SigmaMatch( + rule=self.rule, + events=self._collect_events(matches), + timestamp=now, + group_values=self._make_group_values(group_key), + ) + ] + return [] + + def _eval_value_count( + self, + matches: list[SigmaMatch], + group_key: tuple[Any, ...], + now: Any, + condition: Any, + ) -> list[SigmaMatch]: + if not isinstance(condition, SigmaCorrelationCondition): + return [] + field_ref = condition.fieldref + if field_ref is None: + return [] + # Count distinct values of the field + values: set[Any] = set() + for m in matches: + for e in m.events: + val = e.data.get(field_ref) # type: ignore[arg-type] + if val is not None: + values.add(val) + if self._compare(condition.op, len(values), condition.count): + return [ + SigmaMatch( + rule=self.rule, + events=self._collect_events(matches), + timestamp=now, + group_values=self._make_group_values(group_key), + ) + ] + return [] + + def _eval_extended_expr( + self, node: CorrelationConditionItem | SigmaRuleReference, fired_rules: set[str] + ) -> bool: + """Recursively evaluate a boolean extended correlation condition expression.""" + if isinstance(node, SigmaRuleReference): + return node.reference in fired_rules + elif isinstance(node, CorrelationConditionAND): + return all(self._eval_extended_expr(arg, fired_rules) for arg in node.args) + elif isinstance(node, CorrelationConditionOR): + return any(self._eval_extended_expr(arg, fired_rules) for arg in node.args) + elif isinstance(node, CorrelationConditionNOT): + return not self._eval_extended_expr(node.args[0], fired_rules) + return False + + def _eval_ordered_extended_expr( + self, + node: CorrelationConditionItem | SigmaRuleReference, + rule_first_fire: dict[str, Any], + ) -> tuple[bool, Any]: + """Evaluate a boolean expression with ordering constraints. + + Returns (satisfied, last_timestamp) where last_timestamp is the timestamp of the + last positively matched rule in the sub-expression (None for negated sub-expressions). + For AND, each successive operand's first-fire timestamp must be >= previous operand's. + """ + if isinstance(node, SigmaRuleReference): + ts = rule_first_fire.get(node.reference) + return ts is not None, ts + elif isinstance(node, CorrelationConditionAND): + last_ts: Any = None + for arg in node.args: + ok, ts = self._eval_ordered_extended_expr(arg, rule_first_fire) + if not ok: + return False, None + if last_ts is not None and ts is not None and ts < last_ts: + return False, None + if ts is not None: + last_ts = ts + return True, last_ts + elif isinstance(node, CorrelationConditionOR): + for arg in node.args: + ok, ts = self._eval_ordered_extended_expr(arg, rule_first_fire) + if ok: + return True, ts + return False, None + elif isinstance(node, CorrelationConditionNOT): + ok, _ = self._eval_ordered_extended_expr(node.args[0], rule_first_fire) + return not ok, None + return False, None + + def _eval_temporal( + self, + matches: list[SigmaMatch], + group_key: tuple[Any, ...], + now: Any, + ) -> list[SigmaMatch]: + """All referenced rules must fire within the time window.""" + if isinstance(self.rule.condition, SigmaExtendedCorrelationCondition): + fired_rules: set[str] = { + name for m in matches if (name := getattr(m.rule, "name", None)) + } + if self._eval_extended_expr(self.rule.condition.parsed, fired_rules): + return [ + SigmaMatch( + rule=self.rule, + events=self._collect_events(matches), + timestamp=now, + group_values=self._make_group_values(group_key), + ) + ] + return [] + matched_rules: set[str] = set() + for m in matches: + if hasattr(m.rule, "name") and m.rule.name: + matched_rules.add(m.rule.name) + if self._rule_names and self._rule_names.issubset(matched_rules): + return [ + SigmaMatch( + rule=self.rule, + events=self._collect_events(matches), + timestamp=now, + group_values=self._make_group_values(group_key), + ) + ] + return [] + + def _eval_temporal_ordered( + self, + matches: list[SigmaMatch], + group_key: tuple[Any, ...], + now: Any, + ) -> list[SigmaMatch]: + """All referenced rules must fire in order within the time window.""" + if isinstance(self.rule.condition, SigmaExtendedCorrelationCondition): + rule_first_fire: dict[str, Any] = {} + for m in sorted(matches, key=lambda x: x.timestamp): + name = getattr(m.rule, "name", None) + if name and name not in rule_first_fire: + rule_first_fire[name] = m.timestamp + ok, _ = self._eval_ordered_extended_expr(self.rule.condition.parsed, rule_first_fire) + if ok: + return [ + SigmaMatch( + rule=self.rule, + events=self._collect_events(matches), + timestamp=now, + group_values=self._make_group_values(group_key), + ) + ] + return [] + if not self.rule.rules: + return [] + expected_order = [ref.reference for ref in self.rule.rules] + # Find first occurrence of each in order + seen_idx = 0 + for m in sorted(matches, key=lambda x: x.timestamp): + if hasattr(m.rule, "name") and m.rule.name == expected_order[seen_idx]: + seen_idx += 1 + if seen_idx >= len(expected_order): + return [ + SigmaMatch( + rule=self.rule, + events=self._collect_events(matches), + timestamp=now, + group_values=self._make_group_values(group_key), + ) + ] + return [] + + def _eval_value_aggregate( + self, + matches: list[SigmaMatch], + group_key: tuple[Any, ...], + now: Any, + condition: Any, + agg_type: str, + ) -> list[SigmaMatch]: + if not isinstance(condition, SigmaCorrelationCondition): + return [] + field_ref = condition.fieldref + if field_ref is None: + return [] + + values: list[float] = [] + for m in matches: + for e in m.events: + val = e.data.get(field_ref) # type: ignore[arg-type] + if val is not None: + try: + values.append(float(val)) + except (ValueError, TypeError): + pass + + if not values: + return [] + + if agg_type == "sum": + result = sum(values) + elif agg_type == "avg": + result = sum(values) / len(values) + elif agg_type == "median": + result = median(values) + elif agg_type == "percentile": + percentile = condition.percentile or 95 + sorted_vals = sorted(values) + idx = int(len(sorted_vals) * percentile / 100) + idx = min(idx, len(sorted_vals) - 1) + result = sorted_vals[idx] + else: + return [] + + if self._compare(condition.op, result, condition.count): + return [ + SigmaMatch( + rule=self.rule, + events=self._collect_events(matches), + timestamp=now, + group_values=self._make_group_values(group_key), + ) + ] + return [] diff --git a/sigma/matching/event.py b/sigma/matching/event.py new file mode 100644 index 00000000..e4588652 --- /dev/null +++ b/sigma/matching/event.py @@ -0,0 +1,37 @@ +from __future__ import annotations + +import json +from dataclasses import dataclass, field +from datetime import datetime +from typing import Any + + +@dataclass(frozen=True) +class EventLogSource: + """Lightweight logsource descriptor for events.""" + + category: str | None = None + product: str | None = None + service: str | None = None + + def as_tuple(self) -> tuple[str | None, str | None, str | None]: + return (self.category, self.product, self.service) + + +@dataclass +class Event: + """A parsed log event to be matched against Sigma rules.""" + + timestamp: datetime + logsource: EventLogSource + data: dict[str, Any] + raw: str | None = None + _raw_cache: str | None = field(init=False, default=None, repr=False, compare=False) + + def get_raw(self) -> str: + """Return raw log representation, falling back to JSON-serialized data.""" + if self.raw is not None: + return self.raw + if self._raw_cache is None: + self._raw_cache = json.dumps(self.data) + return self._raw_cache diff --git a/sigma/matching/match.py b/sigma/matching/match.py new file mode 100644 index 00000000..38fb9965 --- /dev/null +++ b/sigma/matching/match.py @@ -0,0 +1,19 @@ +from __future__ import annotations + +from dataclasses import dataclass, field +from datetime import datetime +from typing import TYPE_CHECKING, Any + +if TYPE_CHECKING: + from sigma.matching.event import Event + from sigma.rule.base import SigmaRuleBase + + +@dataclass +class SigmaMatch: + """Result of a successful Sigma rule match against one or more events.""" + + rule: SigmaRuleBase + events: list[Event] + timestamp: datetime + group_values: dict[str, Any] | None = None diff --git a/sigma/matching/matcher.py b/sigma/matching/matcher.py new file mode 100644 index 00000000..cd231049 --- /dev/null +++ b/sigma/matching/matcher.py @@ -0,0 +1,129 @@ +from __future__ import annotations + +from collections import defaultdict +from typing import Any, Iterable + +from typing_extensions import Self + +from sigma.collection import SigmaCollection +from sigma.correlations import SigmaCorrelationRule, SigmaExtendedCorrelationCondition +from sigma.matching.compiler import MatchCompiler +from sigma.matching.correlation_matcher import SigmaCorrelationRuleMatcher +from sigma.matching.event import Event, EventLogSource +from sigma.matching.match import SigmaMatch +from sigma.matching.rule_matcher import SigmaRuleMatcher +from sigma.processing.pipeline import ProcessingPipeline +from sigma.rule.logsource import SigmaLogSource +from sigma.rule.rule import SigmaRule + + +class SigmaMatcher: + """ + Orchestrator that matches events against a collection of Sigma rules. + + Manages logsource dispatch, threading, and correlation. + """ + + def __init__( + self, + collection: SigmaCollection, + pipeline: ProcessingPipeline | None = None, + max_workers: int | None = None, + max_events_per_match: int | None = 100, + max_correlation_state: int | None = None, + ) -> None: + self._compiler = MatchCompiler() + + # Apply pipeline if provided + if pipeline is not None: + for rule in collection.rules: + pipeline.apply(rule) + + # Build dispatch table and correlation matchers + self._dispatch: dict[tuple[str | None, str | None, str | None], list[SigmaRuleMatcher]] = ( + defaultdict(list) + ) + self._all_rule_matchers: list[SigmaRuleMatcher] = [] + # Map rule name -> list of correlation matchers that reference it + self._correlation_map: dict[str, list[SigmaCorrelationRuleMatcher]] = defaultdict(list) + self._correlation_matchers: list[SigmaCorrelationRuleMatcher] = [] + + for rule in collection.rules: + if isinstance(rule, SigmaRule): + rm = SigmaRuleMatcher(rule, self._compiler) + self._all_rule_matchers.append(rm) + key = ( + rule.logsource.category, + rule.logsource.product, + rule.logsource.service, + ) + self._dispatch[key].append(rm) + elif isinstance(rule, SigmaCorrelationRule): + cm = SigmaCorrelationRuleMatcher( + rule, + max_state=max_correlation_state, + max_events_per_match=max_events_per_match, + ) + self._correlation_matchers.append(cm) + if rule.rules: + for ref in rule.rules: + self._correlation_map[ref.reference].append(cm) + elif isinstance(rule.condition, SigmaExtendedCorrelationCondition): + for name in rule.condition.get_referenced_rules(): + self._correlation_map[name].append(cm) + + def match(self, event: Event) -> list[SigmaMatch]: + """Match a single event against all relevant rules. Returns list of matches.""" + matchers = self._get_matchers(event.logsource) + matches: list[SigmaMatch] = [] + + for rm in matchers: + result = rm.matches(event) + if result is not None: + matches.append(result) + + # Route matches to correlation matchers + correlation_matches: list[SigmaMatch] = [] + for m in matches: + rule_name = getattr(m.rule, "name", None) + if rule_name and rule_name in self._correlation_map: + for cm in self._correlation_map[rule_name]: + correlation_matches.extend(cm.receive_match(m, rule_name)) + + matches.extend(correlation_matches) + return matches + + def bulk_match(self, events: Iterable[Event]) -> list[SigmaMatch]: + """Match multiple events, returning all matches. Processes sequentially for ordering.""" + all_matches: list[SigmaMatch] = [] + for event in events: + all_matches.extend(self.match(event)) + return all_matches + + def _get_matchers(self, logsource: EventLogSource) -> list[SigmaRuleMatcher]: + """Get all rule matchers whose logsource contains the event's logsource.""" + result: list[SigmaRuleMatcher] = [] + event_ls = SigmaLogSource( + category=logsource.category, + product=logsource.product, + service=logsource.service, + ) + for key, matchers in self._dispatch.items(): + rule_ls = SigmaLogSource( + category=key[0], + product=key[1], + service=key[2], + ) + if event_ls in rule_ls: + result.extend(matchers) + return result + + def close(self) -> None: + """Clean up resources.""" + pass + + def __enter__(self) -> Self: + return self + + def __exit__(self, *args: Any) -> None: + self.close() diff --git a/sigma/matching/rule_matcher.py b/sigma/matching/rule_matcher.py new file mode 100644 index 00000000..4eb11381 --- /dev/null +++ b/sigma/matching/rule_matcher.py @@ -0,0 +1,30 @@ +from __future__ import annotations + +from datetime import datetime, timezone +from typing import TYPE_CHECKING + +from sigma.matching.compiler import CompiledMatcher, MatchCompiler +from sigma.matching.event import Event +from sigma.matching.match import SigmaMatch + +if TYPE_CHECKING: + from sigma.rule.rule import SigmaRule + + +class SigmaRuleMatcher: + """Evaluates a single SigmaRule against events using a precompiled matcher.""" + + def __init__(self, rule: SigmaRule, compiler: MatchCompiler) -> None: + self.rule = rule + self._matcher: CompiledMatcher = compiler.compile_rule(rule) + + def matches(self, event: Event) -> SigmaMatch | None: + """Return a SigmaMatch if the event matches this rule, else None.""" + raw = event.get_raw() + if self._matcher(event.data, raw): + return SigmaMatch( + rule=self.rule, + events=[event], + timestamp=event.timestamp, + ) + return None diff --git a/sigma/matching/state.py b/sigma/matching/state.py new file mode 100644 index 00000000..259f0ed7 --- /dev/null +++ b/sigma/matching/state.py @@ -0,0 +1,42 @@ +from __future__ import annotations + +from collections import defaultdict, deque +from datetime import datetime, timedelta +from typing import Any + +from sigma.matching.match import SigmaMatch + + +class SlidingWindowState: + """Manages sliding time window state for correlation rules.""" + + def __init__(self, window_seconds: int, max_entries: int | None = None) -> None: + self.window_seconds = window_seconds + self.max_entries = max_entries + # group_key -> deque of (timestamp, match) + self._state: dict[tuple[Any, ...], deque[tuple[datetime, SigmaMatch]]] = defaultdict(deque) + + def add(self, group_key: tuple[Any, ...], match: SigmaMatch) -> None: + """Add a match to the state for the given group key.""" + entries = self._state[group_key] + entries.append((match.timestamp, match)) + if self.max_entries is not None and len(entries) > self.max_entries: + entries.popleft() + + def get_window(self, group_key: tuple[Any, ...], now: datetime) -> list[SigmaMatch]: + """Get all matches within the time window for the given group key.""" + self._evict_group(group_key, now) + return [match for _, match in self._state[group_key]] + + def _evict_group(self, group_key: tuple[Any, ...], now: datetime) -> None: + """Remove entries outside the time window.""" + entries = self._state[group_key] + cutoff = now - timedelta(seconds=self.window_seconds) + while entries and entries[0][0] < cutoff: + entries.popleft() + + def evict(self) -> None: + """Remove empty groups.""" + empty_keys = [k for k, v in self._state.items() if not v] + for k in empty_keys: + del self._state[k] diff --git a/tests/test_matching.py b/tests/test_matching.py new file mode 100644 index 00000000..07438f64 --- /dev/null +++ b/tests/test_matching.py @@ -0,0 +1,1445 @@ +"""Tests for the sigma.matching module.""" + +from datetime import datetime, timedelta, timezone + +import pytest + +from sigma.collection import SigmaCollection +from sigma.matching import Event, EventLogSource, SigmaMatch, SigmaMatcher + + +@pytest.fixture +def simple_rule_yaml(): + return """ +title: Test Rule +name: test_rule +status: test +logsource: + category: process_creation + product: windows +detection: + selection: + CommandLine|contains: "mimikatz" + condition: selection +""" + + +@pytest.fixture +def simple_collection(simple_rule_yaml): + return SigmaCollection.from_yaml(simple_rule_yaml) + + +class TestEvent: + def test_event_creation(self): + event = Event( + timestamp=datetime.now(tz=timezone.utc), + logsource=EventLogSource(category="process_creation", product="windows"), + data={"CommandLine": "mimikatz.exe"}, + ) + assert event.data["CommandLine"] == "mimikatz.exe" + + def test_event_get_raw_with_raw(self): + event = Event( + timestamp=datetime.now(tz=timezone.utc), + logsource=EventLogSource(), + data={"key": "val"}, + raw="raw log line", + ) + assert event.get_raw() == "raw log line" + + def test_event_get_raw_without_raw(self): + event = Event( + timestamp=datetime.now(tz=timezone.utc), + logsource=EventLogSource(), + data={"key": "val"}, + ) + assert '"key"' in event.get_raw() + + +class TestSigmaMatcher: + def test_basic_match(self, simple_collection): + with SigmaMatcher(simple_collection) as matcher: + event = Event( + timestamp=datetime.now(tz=timezone.utc), + logsource=EventLogSource(category="process_creation", product="windows"), + data={"CommandLine": "C:\\tools\\mimikatz.exe"}, + ) + matches = matcher.match(event) + assert len(matches) == 1 + assert matches[0].rule.name == "test_rule" + + def test_no_match(self, simple_collection): + with SigmaMatcher(simple_collection) as matcher: + event = Event( + timestamp=datetime.now(tz=timezone.utc), + logsource=EventLogSource(category="process_creation", product="windows"), + data={"CommandLine": "notepad.exe"}, + ) + matches = matcher.match(event) + assert len(matches) == 0 + + def test_logsource_mismatch(self, simple_collection): + with SigmaMatcher(simple_collection) as matcher: + event = Event( + timestamp=datetime.now(tz=timezone.utc), + logsource=EventLogSource(category="network_connection", product="windows"), + data={"CommandLine": "mimikatz.exe"}, + ) + matches = matcher.match(event) + assert len(matches) == 0 + + def test_bulk_match(self, simple_collection): + with SigmaMatcher(simple_collection) as matcher: + events = [ + Event( + timestamp=datetime.now(tz=timezone.utc), + logsource=EventLogSource(category="process_creation", product="windows"), + data={"CommandLine": "mimikatz.exe"}, + ), + Event( + timestamp=datetime.now(tz=timezone.utc), + logsource=EventLogSource(category="process_creation", product="windows"), + data={"CommandLine": "notepad.exe"}, + ), + ] + matches = matcher.bulk_match(events) + assert len(matches) == 1 + + +class TestCompilerTypes: + """Test various Sigma types compile correctly.""" + + def _match(self, yaml_str: str, data: dict, logsource: EventLogSource | None = None): + collection = SigmaCollection.from_yaml(yaml_str) + ls = logsource or EventLogSource(category="test", product="test") + with SigmaMatcher(collection) as matcher: + event = Event( + timestamp=datetime.now(tz=timezone.utc), + logsource=ls, + data=data, + ) + return matcher.match(event) + + def test_plain_string(self): + yaml_str = """ +title: Test +name: test +status: test +logsource: + category: test + product: test +detection: + selection: + field: "value" + condition: selection +""" + assert len(self._match(yaml_str, {"field": "value"})) == 1 + assert len(self._match(yaml_str, {"field": "VALUE"})) == 1 # case insensitive + assert len(self._match(yaml_str, {"field": "other"})) == 0 + + def test_wildcard(self): + yaml_str = """ +title: Test +name: test +status: test +logsource: + category: test + product: test +detection: + selection: + field|contains: "test" + condition: selection +""" + assert len(self._match(yaml_str, {"field": "this is a test string"})) == 1 + assert len(self._match(yaml_str, {"field": "no match here"})) == 0 + + def test_number(self): + yaml_str = """ +title: Test +name: test +status: test +logsource: + category: test + product: test +detection: + selection: + field: 42 + condition: selection +""" + assert len(self._match(yaml_str, {"field": 42})) == 1 + assert len(self._match(yaml_str, {"field": 43})) == 0 + + def test_null(self): + yaml_str = """ +title: Test +name: test +status: test +logsource: + category: test + product: test +detection: + selection: + field: null + condition: selection +""" + assert len(self._match(yaml_str, {"field": None})) == 1 + assert len(self._match(yaml_str, {"other": "val"})) == 1 # field not present = None + assert len(self._match(yaml_str, {"field": "something"})) == 0 + + def test_boolean(self): + yaml_str = """ +title: Test +name: test +status: test +logsource: + category: test + product: test +detection: + selection: + field: true + condition: selection +""" + assert len(self._match(yaml_str, {"field": True})) == 1 + assert len(self._match(yaml_str, {"field": False})) == 0 + + def test_regex(self): + yaml_str = """ +title: Test +name: test +status: test +logsource: + category: test + product: test +detection: + selection: + field|re: "^test\\\\d+" + condition: selection +""" + assert len(self._match(yaml_str, {"field": "test123"})) == 1 + assert len(self._match(yaml_str, {"field": "notest"})) == 0 + + def test_cidr(self): + yaml_str = """ +title: Test +name: test +status: test +logsource: + category: test + product: test +detection: + selection: + field|cidr: "192.168.1.0/24" + condition: selection +""" + assert len(self._match(yaml_str, {"field": "192.168.1.50"})) == 1 + assert len(self._match(yaml_str, {"field": "10.0.0.1"})) == 0 + + def test_compare_gt(self): + yaml_str = """ +title: Test +name: test +status: test +logsource: + category: test + product: test +detection: + selection: + field|gt: 10 + condition: selection +""" + assert len(self._match(yaml_str, {"field": 15})) == 1 + assert len(self._match(yaml_str, {"field": 5})) == 0 + + def test_condition_and(self): + yaml_str = """ +title: Test +name: test +status: test +logsource: + category: test + product: test +detection: + sel1: + field1: "a" + sel2: + field2: "b" + condition: sel1 and sel2 +""" + assert len(self._match(yaml_str, {"field1": "a", "field2": "b"})) == 1 + assert len(self._match(yaml_str, {"field1": "a", "field2": "c"})) == 0 + + def test_condition_or(self): + yaml_str = """ +title: Test +name: test +status: test +logsource: + category: test + product: test +detection: + sel1: + field1: "a" + sel2: + field2: "b" + condition: sel1 or sel2 +""" + assert len(self._match(yaml_str, {"field1": "a"})) == 1 + assert len(self._match(yaml_str, {"field2": "b"})) == 1 + assert len(self._match(yaml_str, {"field3": "c"})) == 0 + + def test_condition_not(self): + yaml_str = """ +title: Test +name: test +status: test +logsource: + category: test + product: test +detection: + sel: + field: "a" + condition: not sel +""" + assert len(self._match(yaml_str, {"field": "b"})) == 1 + assert len(self._match(yaml_str, {"field": "a"})) == 0 + + def test_keyword(self): + yaml_str = """ +title: Test +name: test +status: test +logsource: + category: test + product: test +detection: + keywords: + - "suspicious" + condition: keywords +""" + assert len(self._match(yaml_str, {"msg": "this is suspicious activity"})) == 1 + assert len(self._match(yaml_str, {"msg": "normal activity"})) == 0 + + +class TestCorrelation: + def test_event_count(self): + yaml_str = """ +title: Base Rule +name: base_rule +status: test +logsource: + category: test + product: test +detection: + selection: + field: "trigger" + condition: selection +--- +title: Correlation +name: correlation_rule +status: test +correlation: + type: event_count + rules: + - base_rule + group-by: + - src_ip + timespan: 5m + condition: + gte: 3 +""" + collection = SigmaCollection.from_yaml(yaml_str) + with SigmaMatcher(collection) as matcher: + now = datetime.now(tz=timezone.utc) + ls = EventLogSource(category="test", product="test") + + # First two events - no correlation match yet + for i in range(2): + matches = matcher.match( + Event( + timestamp=now, logsource=ls, data={"field": "trigger", "src_ip": "1.2.3.4"} + ) + ) + # Should have base rule match but no correlation yet + assert any(m.rule.name == "base_rule" for m in matches) + assert not any(m.rule.name == "correlation_rule" for m in matches) + + # Third event triggers correlation + matches = matcher.match( + Event(timestamp=now, logsource=ls, data={"field": "trigger", "src_ip": "1.2.3.4"}) + ) + assert any(m.rule.name == "correlation_rule" for m in matches) + + def test_value_count(self): + yaml_str = """ +title: Base Rule +name: base_rule +status: test +logsource: + category: test + product: test +detection: + selection: + event: "login" + condition: selection +--- +title: Value Count Correlation +name: vc_corr +status: test +correlation: + type: value_count + rules: + - base_rule + group-by: + - src_ip + timespan: 5m + condition: + gte: 3 + field: user +""" + collection = SigmaCollection.from_yaml(yaml_str) + with SigmaMatcher(collection) as matcher: + now = datetime.now(tz=timezone.utc) + ls = EventLogSource(category="test", product="test") + + # Two distinct users – no correlation yet + for user in ("alice", "bob"): + matches = matcher.match( + Event( + timestamp=now, + logsource=ls, + data={"event": "login", "src_ip": "1.2.3.4", "user": user}, + ) + ) + assert any(m.rule.name == "base_rule" for m in matches) + assert not any(m.rule.name == "vc_corr" for m in matches) + + # Duplicate user does not increase distinct count + matches = matcher.match( + Event( + timestamp=now, + logsource=ls, + data={"event": "login", "src_ip": "1.2.3.4", "user": "alice"}, + ) + ) + assert not any(m.rule.name == "vc_corr" for m in matches) + + # Third distinct user triggers correlation + matches = matcher.match( + Event( + timestamp=now, + logsource=ls, + data={"event": "login", "src_ip": "1.2.3.4", "user": "charlie"}, + ) + ) + assert any(m.rule.name == "vc_corr" for m in matches) + + def test_temporal(self): + yaml_str = """ +title: Login Rule +name: login_rule +status: test +logsource: + category: test + product: test +detection: + selection: + event: "login" + condition: selection +--- +title: Access Rule +name: access_rule +status: test +logsource: + category: test + product: test +detection: + selection: + event: "access" + condition: selection +--- +title: Temporal Correlation +name: temporal_corr +status: test +correlation: + type: temporal + rules: + - login_rule + - access_rule + group-by: + - user + timespan: 5m +""" + collection = SigmaCollection.from_yaml(yaml_str) + with SigmaMatcher(collection) as matcher: + now = datetime.now(tz=timezone.utc) + ls = EventLogSource(category="test", product="test") + + # Only login fires for bob – no temporal match + matches = matcher.match( + Event(timestamp=now, logsource=ls, data={"event": "login", "user": "bob"}) + ) + assert any(m.rule.name == "login_rule" for m in matches) + assert not any(m.rule.name == "temporal_corr" for m in matches) + + # Both rules fire for alice within the window – temporal match fires + matcher.match( + Event(timestamp=now, logsource=ls, data={"event": "login", "user": "alice"}) + ) + matches = matcher.match( + Event(timestamp=now, logsource=ls, data={"event": "access", "user": "alice"}) + ) + assert any(m.rule.name == "access_rule" for m in matches) + assert any(m.rule.name == "temporal_corr" for m in matches) + + def test_temporal_ordered(self): + yaml_str = """ +title: Login Rule +name: login_rule +status: test +logsource: + category: test + product: test +detection: + selection: + event: "login" + condition: selection +--- +title: Access Rule +name: access_rule +status: test +logsource: + category: test + product: test +detection: + selection: + event: "access" + condition: selection +--- +title: Ordered Temporal Correlation +name: ordered_corr +status: test +correlation: + type: temporal_ordered + rules: + - login_rule + - access_rule + group-by: + - user + timespan: 5m +""" + collection = SigmaCollection.from_yaml(yaml_str) + t1 = datetime.now(tz=timezone.utc) + t2 = t1 + timedelta(seconds=30) + ls = EventLogSource(category="test", product="test") + + with SigmaMatcher(collection) as matcher: + # Correct order for alice: login then access + matcher.match( + Event(timestamp=t1, logsource=ls, data={"event": "login", "user": "alice"}) + ) + matches = matcher.match( + Event(timestamp=t2, logsource=ls, data={"event": "access", "user": "alice"}) + ) + assert any(m.rule.name == "ordered_corr" for m in matches) + + with SigmaMatcher(collection) as matcher: + # Wrong order for bob: access then login – should not fire + matcher.match( + Event(timestamp=t1, logsource=ls, data={"event": "access", "user": "bob"}) + ) + matches = matcher.match( + Event(timestamp=t2, logsource=ls, data={"event": "login", "user": "bob"}) + ) + assert not any(m.rule.name == "ordered_corr" for m in matches) + + def test_value_sum(self): + yaml_str = """ +title: Base Rule +name: base_rule +status: test +logsource: + category: test + product: test +detection: + selection: + event: "transfer" + condition: selection +--- +title: Value Sum Correlation +name: vs_corr +status: test +correlation: + type: value_sum + rules: + - base_rule + group-by: + - src_ip + timespan: 5m + condition: + gte: 100 + field: bytes +""" + collection = SigmaCollection.from_yaml(yaml_str) + with SigmaMatcher(collection) as matcher: + now = datetime.now(tz=timezone.utc) + ls = EventLogSource(category="test", product="test") + + for bytes_val in (40, 40): + matches = matcher.match( + Event( + timestamp=now, + logsource=ls, + data={"event": "transfer", "src_ip": "1.2.3.4", "bytes": bytes_val}, + ) + ) + assert not any(m.rule.name == "vs_corr" for m in matches) + + # Sum reaches 110 >= 100 + matches = matcher.match( + Event( + timestamp=now, + logsource=ls, + data={"event": "transfer", "src_ip": "1.2.3.4", "bytes": 30}, + ) + ) + assert any(m.rule.name == "vs_corr" for m in matches) + + def test_value_avg(self): + yaml_str = """ +title: Base Rule +name: base_rule +status: test +logsource: + category: test + product: test +detection: + selection: + event: "metric" + condition: selection +--- +title: Value Avg Correlation +name: avg_corr +status: test +correlation: + type: value_avg + rules: + - base_rule + group-by: + - host + timespan: 5m + condition: + gte: 20 + field: value +""" + collection = SigmaCollection.from_yaml(yaml_str) + with SigmaMatcher(collection) as matcher: + now = datetime.now(tz=timezone.utc) + ls = EventLogSource(category="test", product="test") + + for val in (5, 5): + matches = matcher.match( + Event( + timestamp=now, + logsource=ls, + data={"event": "metric", "host": "srv1", "value": val}, + ) + ) + assert not any(m.rule.name == "avg_corr" for m in matches) + + # Average becomes (5+5+50)/3 = 20 >= 20 + matches = matcher.match( + Event( + timestamp=now, + logsource=ls, + data={"event": "metric", "host": "srv1", "value": 50}, + ) + ) + assert any(m.rule.name == "avg_corr" for m in matches) + + def test_value_median(self): + yaml_str = """ +title: Base Rule +name: base_rule +status: test +logsource: + category: test + product: test +detection: + selection: + event: "metric" + condition: selection +--- +title: Value Median Correlation +name: med_corr +status: test +correlation: + type: value_median + rules: + - base_rule + group-by: + - host + timespan: 5m + condition: + gte: 30 + field: value +""" + collection = SigmaCollection.from_yaml(yaml_str) + with SigmaMatcher(collection) as matcher: + now = datetime.now(tz=timezone.utc) + ls = EventLogSource(category="test", product="test") + + for val in (10, 40): + matches = matcher.match( + Event( + timestamp=now, + logsource=ls, + data={"event": "metric", "host": "srv1", "value": val}, + ) + ) + assert not any(m.rule.name == "med_corr" for m in matches) + + # Median of [10, 30, 40] = 30 >= 30 + matches = matcher.match( + Event( + timestamp=now, + logsource=ls, + data={"event": "metric", "host": "srv1", "value": 30}, + ) + ) + assert any(m.rule.name == "med_corr" for m in matches) + + def test_value_percentile(self): + yaml_str = """ +title: Base Rule +name: base_rule +status: test +logsource: + category: test + product: test +detection: + selection: + event: "metric" + condition: selection +--- +title: Value Percentile Correlation +name: pct_corr +status: test +correlation: + type: value_percentile + rules: + - base_rule + group-by: + - host + timespan: 5m + condition: + gte: 40 + field: value + percentile: 90 +""" + collection = SigmaCollection.from_yaml(yaml_str) + with SigmaMatcher(collection) as matcher: + now = datetime.now(tz=timezone.utc) + ls = EventLogSource(category="test", product="test") + + for val in (10, 20, 30): + matches = matcher.match( + Event( + timestamp=now, + logsource=ls, + data={"event": "metric", "host": "srv1", "value": val}, + ) + ) + assert not any(m.rule.name == "pct_corr" for m in matches) + + # 90th percentile of [10,20,30,40]: idx=int(4*0.9)=3, val=40 >= 40 + matches = matcher.match( + Event( + timestamp=now, + logsource=ls, + data={"event": "metric", "host": "srv1", "value": 40}, + ) + ) + assert any(m.rule.name == "pct_corr" for m in matches) + + +class TestExtendedCorrelation: + """Test extended correlation conditions with boolean expressions.""" + + _BASE_RULES = """ +title: Failed Logon +name: failed_logon +status: test +logsource: + category: test + product: test +detection: + selection: + event: "failed_logon" + condition: selection +--- +title: Successful Logon +name: successful_logon +status: test +logsource: + category: test + product: test +detection: + selection: + event: "successful_logon" + condition: selection +--- +title: Suspicious Activity +name: suspicious_activity +status: test +logsource: + category: test + product: test +detection: + selection: + event: "suspicious" + condition: selection +""" + + def _make_collection(self, correlation_yaml: str) -> SigmaCollection: + return SigmaCollection.from_yaml(self._BASE_RULES + "---\n" + correlation_yaml) + + def _fire( + self, + matcher: SigmaMatcher, + event_type: str, + user: str = "alice", + ts: datetime | None = None, + ) -> list[SigmaMatch]: + return matcher.match( + Event( + timestamp=ts or datetime.now(tz=timezone.utc), + logsource=EventLogSource(category="test", product="test"), + data={"event": event_type, "user": user}, + ) + ) + + def test_and_both_fire(self): + """Extended AND: both rules must fire to produce a correlation match.""" + yaml = """ +title: Extended AND Correlation +name: ext_and_corr +status: test +correlation: + type: temporal + condition: failed_logon and successful_logon + timespan: 5m + group-by: + - user +""" + collection = self._make_collection(yaml) + with SigmaMatcher(collection) as matcher: + now = datetime.now(tz=timezone.utc) + self._fire(matcher, "failed_logon", ts=now) + matches = self._fire(matcher, "successful_logon", ts=now) + assert any(m.rule.name == "ext_and_corr" for m in matches) + + def test_and_only_one_fires(self): + """Extended AND: only one rule fires — no correlation match.""" + yaml = """ +title: Extended AND Correlation +name: ext_and_corr +status: test +correlation: + type: temporal + condition: failed_logon and successful_logon + timespan: 5m + group-by: + - user +""" + collection = self._make_collection(yaml) + with SigmaMatcher(collection) as matcher: + now = datetime.now(tz=timezone.utc) + matches = self._fire(matcher, "failed_logon", ts=now) + assert not any(m.rule.name == "ext_and_corr" for m in matches) + + def test_or_first_fires(self): + """Extended OR: firing any referenced rule triggers a correlation match.""" + yaml = """ +title: Extended OR Correlation +name: ext_or_corr +status: test +correlation: + type: temporal + condition: failed_logon or suspicious_activity + timespan: 5m + group-by: + - user +""" + collection = self._make_collection(yaml) + with SigmaMatcher(collection) as matcher: + now = datetime.now(tz=timezone.utc) + matches = self._fire(matcher, "failed_logon", ts=now) + assert any(m.rule.name == "ext_or_corr" for m in matches) + + def test_or_second_fires(self): + """Extended OR: the second alternative also triggers a match.""" + yaml = """ +title: Extended OR Correlation +name: ext_or_corr +status: test +correlation: + type: temporal + condition: failed_logon or suspicious_activity + timespan: 5m + group-by: + - user +""" + collection = self._make_collection(yaml) + with SigmaMatcher(collection) as matcher: + now = datetime.now(tz=timezone.utc) + matches = self._fire(matcher, "suspicious", ts=now) + assert any(m.rule.name == "ext_or_corr" for m in matches) + + def test_or_neither_fires(self): + """Extended OR: no matching event → no correlation match.""" + yaml = """ +title: Extended OR Correlation +name: ext_or_corr +status: test +correlation: + type: temporal + condition: failed_logon or suspicious_activity + timespan: 5m + group-by: + - user +""" + collection = self._make_collection(yaml) + with SigmaMatcher(collection) as matcher: + now = datetime.now(tz=timezone.utc) + matches = self._fire(matcher, "successful_logon", ts=now) + assert not any(m.rule.name == "ext_or_corr" for m in matches) + + def test_not_excluded_rule_absent(self): + """Extended NOT: trigger rule fires without the excluded rule → match.""" + yaml = """ +title: Extended NOT Correlation +name: ext_not_corr +status: test +correlation: + type: temporal + condition: failed_logon and not successful_logon + timespan: 5m + group-by: + - user +""" + collection = self._make_collection(yaml) + with SigmaMatcher(collection) as matcher: + now = datetime.now(tz=timezone.utc) + matches = self._fire(matcher, "failed_logon", ts=now) + assert any(m.rule.name == "ext_not_corr" for m in matches) + + def test_not_excluded_rule_present(self): + """Extended NOT: excluded rule fires within window → no match.""" + yaml = """ +title: Extended NOT Correlation +name: ext_not_corr +status: test +correlation: + type: temporal + condition: failed_logon and not successful_logon + timespan: 5m + group-by: + - user +""" + collection = self._make_collection(yaml) + with SigmaMatcher(collection) as matcher: + now = datetime.now(tz=timezone.utc) + self._fire(matcher, "failed_logon", ts=now) + matches = self._fire(matcher, "successful_logon", ts=now) + assert not any(m.rule.name == "ext_not_corr" for m in matches) + + def test_ordered_correct_order(self): + """Extended temporal_ordered AND: rules in correct order → match.""" + yaml = """ +title: Extended Ordered Correlation +name: ext_ordered_corr +status: test +correlation: + type: temporal_ordered + condition: failed_logon and successful_logon + timespan: 5m + group-by: + - user +""" + collection = self._make_collection(yaml) + t1 = datetime.now(tz=timezone.utc) + t2 = t1 + timedelta(seconds=30) + with SigmaMatcher(collection) as matcher: + self._fire(matcher, "failed_logon", ts=t1) + matches = self._fire(matcher, "successful_logon", ts=t2) + assert any(m.rule.name == "ext_ordered_corr" for m in matches) + + def test_ordered_wrong_order(self): + """Extended temporal_ordered AND: rules in wrong order → no match.""" + yaml = """ +title: Extended Ordered Correlation +name: ext_ordered_corr +status: test +correlation: + type: temporal_ordered + condition: failed_logon and successful_logon + timespan: 5m + group-by: + - user +""" + collection = self._make_collection(yaml) + t1 = datetime.now(tz=timezone.utc) + t2 = t1 + timedelta(seconds=30) + with SigmaMatcher(collection) as matcher: + self._fire(matcher, "successful_logon", ts=t1) + matches = self._fire(matcher, "failed_logon", ts=t2) + assert not any(m.rule.name == "ext_ordered_corr" for m in matches) + + def test_ordered_three_rules_correct_order(self): + """Extended temporal_ordered: three rules all fire in correct sequence.""" + yaml = """ +title: Extended Ordered 3 Correlation +name: ext_ordered3_corr +status: test +correlation: + type: temporal_ordered + condition: failed_logon and successful_logon and suspicious_activity + timespan: 5m + group-by: + - user +""" + collection = self._make_collection(yaml) + t1 = datetime.now(tz=timezone.utc) + t2 = t1 + timedelta(seconds=10) + t3 = t2 + timedelta(seconds=10) + with SigmaMatcher(collection) as matcher: + self._fire(matcher, "failed_logon", ts=t1) + self._fire(matcher, "successful_logon", ts=t2) + matches = self._fire(matcher, "suspicious", ts=t3) + assert any(m.rule.name == "ext_ordered3_corr" for m in matches) + + def test_ordered_three_rules_wrong_order(self): + """Extended temporal_ordered: middle rule fires last → no match.""" + yaml = """ +title: Extended Ordered 3 Correlation +name: ext_ordered3_corr +status: test +correlation: + type: temporal_ordered + condition: failed_logon and successful_logon and suspicious_activity + timespan: 5m + group-by: + - user +""" + collection = self._make_collection(yaml) + t1 = datetime.now(tz=timezone.utc) + t2 = t1 + timedelta(seconds=10) + t3 = t2 + timedelta(seconds=10) + with SigmaMatcher(collection) as matcher: + self._fire(matcher, "failed_logon", ts=t1) + self._fire(matcher, "suspicious", ts=t2) + matches = self._fire(matcher, "successful_logon", ts=t3) + assert not any(m.rule.name == "ext_ordered3_corr" for m in matches) + + def test_group_by_isolation(self): + """Extended AND: group-by keeps user windows separate.""" + yaml = """ +title: Extended AND Correlation +name: ext_and_corr +status: test +correlation: + type: temporal + condition: failed_logon and successful_logon + timespan: 5m + group-by: + - user +""" + collection = self._make_collection(yaml) + with SigmaMatcher(collection) as matcher: + now = datetime.now(tz=timezone.utc) + # failed_logon for alice, successful_logon for bob — different groups + self._fire(matcher, "failed_logon", user="alice", ts=now) + matches = self._fire(matcher, "successful_logon", user="bob", ts=now) + assert not any(m.rule.name == "ext_and_corr" for m in matches) + + +class TestModifiers: + """Test that all Sigma modifiers produce correct matching behavior end-to-end.""" + + def _match(self, yaml_str: str, data: dict, logsource: EventLogSource | None = None): + collection = SigmaCollection.from_yaml(yaml_str) + ls = logsource or EventLogSource(category="test", product="test") + with SigmaMatcher(collection) as matcher: + event = Event( + timestamp=datetime.now(tz=timezone.utc), + logsource=ls, + data=data, + ) + return matcher.match(event) + + def test_startswith(self): + yaml_str = """ +title: Test +name: test +status: test +logsource: + category: test + product: test +detection: + selection: + field|startswith: "cmd" + condition: selection +""" + assert len(self._match(yaml_str, {"field": "cmd.exe /c dir"})) == 1 + assert len(self._match(yaml_str, {"field": "CMD.exe"})) == 1 # case-insensitive + assert len(self._match(yaml_str, {"field": "not cmd"})) == 0 + + def test_endswith(self): + yaml_str = """ +title: Test +name: test +status: test +logsource: + category: test + product: test +detection: + selection: + field|endswith: ".exe" + condition: selection +""" + assert len(self._match(yaml_str, {"field": "cmd.exe"})) == 1 + assert len(self._match(yaml_str, {"field": "CMD.EXE"})) == 1 # case-insensitive + assert len(self._match(yaml_str, {"field": "cmd.bat"})) == 0 + + def test_base64(self): + # base64("foobar") == "Zm9vYmFy" + yaml_str = """ +title: Test +name: test +status: test +logsource: + category: test + product: test +detection: + selection: + field|base64: "foobar" + condition: selection +""" + assert len(self._match(yaml_str, {"field": "Zm9vYmFy"})) == 1 + assert len(self._match(yaml_str, {"field": "zm9vymfy"})) == 1 # case-insensitive + assert len(self._match(yaml_str, {"field": "foobar"})) == 0 # raw value does not match + assert len(self._match(yaml_str, {"field": "other"})) == 0 + + def test_base64offset(self): + # base64offset("foobar") == ["Zm9vYmFy", "Zvb2Jhc", "mb29iYX"] + yaml_str = """ +title: Test +name: test +status: test +logsource: + category: test + product: test +detection: + selection: + field|base64offset: "foobar" + condition: selection +""" + assert len(self._match(yaml_str, {"field": "Zm9vYmFy"})) == 1 # offset 0 + assert len(self._match(yaml_str, {"field": "Zvb2Jhc"})) == 1 # offset 1 + assert len(self._match(yaml_str, {"field": "mb29iYX"})) == 1 # offset 2 + assert len(self._match(yaml_str, {"field": "foobar"})) == 0 + assert len(self._match(yaml_str, {"field": "other"})) == 0 + + def test_wide(self): + # wide("cmd") encodes each ASCII char as two bytes (UTF-16LE), resulting in "c\x00m\x00d\x00" + yaml_str = """ +title: Test +name: test +status: test +logsource: + category: test + product: test +detection: + selection: + field|wide: "cmd" + condition: selection +""" + assert len(self._match(yaml_str, {"field": "c\x00m\x00d\x00"})) == 1 + assert len(self._match(yaml_str, {"field": "cmd"})) == 0 # not wide-encoded + assert len(self._match(yaml_str, {"field": "other"})) == 0 + + def test_utf16be(self): + # utf16be("ab") == "\x00a\x00b" (big-endian null-prefixed pairs) + yaml_str = """ +title: Test +name: test +status: test +logsource: + category: test + product: test +detection: + selection: + field|utf16be: "ab" + condition: selection +""" + assert len(self._match(yaml_str, {"field": "\x00a\x00b"})) == 1 + assert len(self._match(yaml_str, {"field": "ab"})) == 0 + assert len(self._match(yaml_str, {"field": "other"})) == 0 + + def test_utf16(self): + # utf16("ab") == "\ufeffa\x00b\x00" (BOM + UTF-16LE pairs) + yaml_str = """ +title: Test +name: test +status: test +logsource: + category: test + product: test +detection: + selection: + field|utf16: "ab" + condition: selection +""" + assert len(self._match(yaml_str, {"field": "\ufeffa\x00b\x00"})) == 1 + assert len(self._match(yaml_str, {"field": "ab"})) == 0 + assert len(self._match(yaml_str, {"field": "other"})) == 0 + + def test_windash(self): + # windash expands the leading "-" to "-", "/", en-dash, em-dash, horizontal bar + yaml_str = """ +title: Test +name: test +status: test +logsource: + category: test + product: test +detection: + selection: + field|windash: "-cmd" + condition: selection +""" + assert len(self._match(yaml_str, {"field": "-cmd"})) == 1 + assert len(self._match(yaml_str, {"field": "/cmd"})) == 1 # slash variant + assert len(self._match(yaml_str, {"field": "\u2013cmd"})) == 1 # en-dash variant + assert len(self._match(yaml_str, {"field": "other"})) == 0 + + def test_all_modifier(self): + yaml_str = """ +title: Test +name: test +status: test +logsource: + category: test + product: test +detection: + selection: + field|contains|all: + - "word1" + - "word2" + condition: selection +""" + assert len(self._match(yaml_str, {"field": "word1 and word2 here"})) == 1 + assert len(self._match(yaml_str, {"field": "only word1"})) == 0 + assert len(self._match(yaml_str, {"field": "only word2"})) == 0 + assert len(self._match(yaml_str, {"field": "neither"})) == 0 + + def test_negate_modifier(self): + # |neq wraps the detection item in NOT + yaml_str = """ +title: Test +name: test +status: test +logsource: + category: test + product: test +detection: + selection: + field|neq: "blocked" + condition: selection +""" + assert len(self._match(yaml_str, {"field": "allowed"})) == 1 + assert len(self._match(yaml_str, {"field": "blocked"})) == 0 + + def test_compare_lt(self): + yaml_str = """ +title: Test +name: test +status: test +logsource: + category: test + product: test +detection: + selection: + field|lt: 10 + condition: selection +""" + assert len(self._match(yaml_str, {"field": 5})) == 1 + assert len(self._match(yaml_str, {"field": 10})) == 0 + assert len(self._match(yaml_str, {"field": 15})) == 0 + + def test_compare_lte(self): + yaml_str = """ +title: Test +name: test +status: test +logsource: + category: test + product: test +detection: + selection: + field|lte: 10 + condition: selection +""" + assert len(self._match(yaml_str, {"field": 5})) == 1 + assert len(self._match(yaml_str, {"field": 10})) == 1 + assert len(self._match(yaml_str, {"field": 15})) == 0 + + def test_compare_gte(self): + yaml_str = """ +title: Test +name: test +status: test +logsource: + category: test + product: test +detection: + selection: + field|gte: 10 + condition: selection +""" + assert len(self._match(yaml_str, {"field": 15})) == 1 + assert len(self._match(yaml_str, {"field": 10})) == 1 + assert len(self._match(yaml_str, {"field": 5})) == 0 + + def test_cased(self): + yaml_str = """ +title: Test +name: test +status: test +logsource: + category: test + product: test +detection: + selection: + field|cased: "CasedValue" + condition: selection +""" + assert len(self._match(yaml_str, {"field": "CasedValue"})) == 1 + assert len(self._match(yaml_str, {"field": "casedvalue"})) == 0 # case-sensitive mismatch + assert len(self._match(yaml_str, {"field": "CASEDVALUE"})) == 0 + + def test_exists_true(self): + yaml_str = """ +title: Test +name: test +status: test +logsource: + category: test + product: test +detection: + selection: + field|exists: true + condition: selection +""" + assert len(self._match(yaml_str, {"field": "anything"})) == 1 + assert len(self._match(yaml_str, {"field": None})) == 1 # key present, value null + assert len(self._match(yaml_str, {"other": "val"})) == 0 # key absent + + def test_exists_false(self): + yaml_str = """ +title: Test +name: test +status: test +logsource: + category: test + product: test +detection: + selection: + field|exists: false + condition: selection +""" + assert len(self._match(yaml_str, {"other": "val"})) == 1 # field absent + assert len(self._match(yaml_str, {"field": "anything"})) == 0 # field present + + def test_fieldref(self): + yaml_str = """ +title: Test +name: test +status: test +logsource: + category: test + product: test +detection: + selection: + field1|fieldref: "field2" + condition: selection +""" + assert len(self._match(yaml_str, {"field1": "abc", "field2": "abc"})) == 1 + assert len(self._match(yaml_str, {"field1": "abc", "field2": "def"})) == 0 + + def test_fieldref_contains(self): + # field1 must contain the value of field2 + yaml_str = """ +title: Test +name: test +status: test +logsource: + category: test + product: test +detection: + selection: + field1|fieldref|contains: "field2" + condition: selection +""" + assert len(self._match(yaml_str, {"field1": "abcdef", "field2": "abc"})) == 1 + assert len(self._match(yaml_str, {"field1": "abc", "field2": "abcdef"})) == 0 + + def test_re_ignorecase(self): + yaml_str = """ +title: Test +name: test +status: test +logsource: + category: test + product: test +detection: + selection: + field|re|i: "^pattern$" + condition: selection +""" + assert len(self._match(yaml_str, {"field": "pattern"})) == 1 + assert len(self._match(yaml_str, {"field": "PATTERN"})) == 1 # case-insensitive + assert len(self._match(yaml_str, {"field": "other"})) == 0 + + def test_re_multiline(self): + yaml_str = """ +title: Test +name: test +status: test +logsource: + category: test + product: test +detection: + selection: + field|re|m: "^test$" + condition: selection +""" + assert len(self._match(yaml_str, {"field": "prefix\ntest\nsuffix"})) == 1 + assert len(self._match(yaml_str, {"field": "prefix\nsuffix"})) == 0 + + def test_re_dotall(self): + yaml_str = """ +title: Test +name: test +status: test +logsource: + category: test + product: test +detection: + selection: + field|re|s: "test.data" + condition: selection +""" + assert len(self._match(yaml_str, {"field": "test\ndata"})) == 1 # dot matches newline + assert len(self._match(yaml_str, {"field": "testXdata"})) == 1 + assert len(self._match(yaml_str, {"field": "no match"})) == 0