From cfb72c30a0a2afb7bd47c8bba8395256bed2d878 Mon Sep 17 00:00:00 2001 From: Bob Date: Thu, 10 Sep 2026 05:44:22 +0000 Subject: [PATCH 1/3] feat(classify): add regex_fields whole-field AND rule type Extend Rule to support a 'regex_fields' variant where every named field must match its own pattern in its entirety (re.fullmatch, logical AND). Existing 'regex' rules are fully backward-compatible. Motivation: users need separate categories when two apps (e.g. mstsc.exe vs winbox.exe) show identical titles. Title-only regexes cannot tell them apart; only an app-AND-title rule can. Contract: - type: 'regex_fields', fields: {: , ...} - Optional ignore_case: bool (default False) - All named fields must exist as strings in event.data - Each pattern is tested with re.fullmatch (whole-field anchoring) - 'regex' and 'select_keys' members raise ValueError on this variant, guarding against the rollout hazard where old Python reads a stray 'regex' member and produces false positives - Empty fields dict raises ValueError 11 new tests added to tests/test_transforms.py; 37 existing tests pass. Git-Session-Id: 3f81 --- aw_transform/classify.py | 98 ++++++++++++++++++++++----- tests/test_transforms.py | 142 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 222 insertions(+), 18 deletions(-) diff --git a/aw_transform/classify.py b/aw_transform/classify.py index 35a33b8..2355ded 100644 --- a/aw_transform/classify.py +++ b/aw_transform/classify.py @@ -23,36 +23,98 @@ def _parse_optional_priority(rules: Dict[str, Any]) -> Optional[int]: class Rule: + """Category/tag rule. + + Supports two variants, selected by the optional ``type`` field: + + * ``"regex"`` (default when ``type`` is absent): the existing per-field + ``search`` rule. A single ``regex`` pattern is tested against every event + value (or the ``select_keys`` subset). Backward-compatible. + + * ``"regex_fields"``: a new whole-field AND rule. Every key in ``fields`` + must exist in the event data, be a string, and its value must satisfy the + corresponding pattern in its entirety (``re.fullmatch``). ``regex`` and + ``select_keys`` are forbidden on this variant. + """ + regex: Optional[Pattern] select_keys: Optional[List[str]] ignore_case: bool priority: Optional[int] + _rule_type: str + # Only populated for regex_fields rules. + _field_patterns: Optional[Dict[str, Pattern]] def __init__(self, rules: Dict[str, Any]) -> None: - self.select_keys = rules.get("select_keys", None) + self._rule_type = rules.get("type", "regex") self.ignore_case = rules.get("ignore_case", False) self.priority = _parse_optional_priority(rules) + flags = (re.IGNORECASE if self.ignore_case else 0) | re.UNICODE + + if self._rule_type == "regex_fields": + # Guard against the identified rollout hazard: old Python silently + # reads a stale `regex` member and produces false matches. Reject + # both forbidden members explicitly. + if "regex" in rules: + raise ValueError( + "regex_fields rule must not contain a 'regex' member " + "(use 'fields' instead)" + ) + if "select_keys" in rules: + raise ValueError( + "regex_fields rule must not contain 'select_keys' " + "(use 'fields' instead)" + ) + raw_fields: Any = rules.get("fields") + if not isinstance(raw_fields, dict) or not raw_fields: + raise ValueError("regex_fields rule requires a non-empty 'fields' dict") + self._field_patterns = {} + for field, pattern in raw_fields.items(): + if not isinstance(field, str) or not field: + raise ValueError( + "regex_fields: field names must be non-empty strings" + ) + if not isinstance(pattern, str) or not pattern: + raise ValueError( + f"regex_fields: pattern for field '{field}' must be a non-empty string" + ) + self._field_patterns[field] = re.compile(pattern, flags) + # Legacy attributes unused for this variant. + self.regex = None + self.select_keys = None + else: + # Legacy "regex" variant (also the default when type is absent). + self._field_patterns = None + self.select_keys = rules.get("select_keys", None) - # NOTE: Also checks that the regex isn't an empty string (which would erroneously match everything) - regex_str = rules.get("regex", None) - self.regex = ( - re.compile( - regex_str, (re.IGNORECASE if self.ignore_case else 0) | re.UNICODE - ) - if regex_str - else None - ) + # NOTE: Also checks that the regex isn't an empty string (which would erroneously match everything) + regex_str = rules.get("regex", None) + self.regex = re.compile(regex_str, flags) if regex_str else None def match(self, e: Event) -> bool: - if self.select_keys: - values = [e.data.get(key, None) for key in self.select_keys] + if self._rule_type == "regex_fields": + # ALL named fields must exist, be strings, and fully satisfy their pattern. + assert self._field_patterns is not None + for field, pattern in self._field_patterns.items(): + value = e.data.get(field) + if not isinstance(value, str): + return False + if not pattern.fullmatch(value): + return False + return bool( + self._field_patterns + ) # empty map never matches (guarded at init) else: - values = list(e.data.values()) - if self.regex: - for val in values: - if isinstance(val, str) and self.regex.search(val): - return True - return False + # Legacy regex variant. + if self.select_keys: + values = [e.data.get(key, None) for key in self.select_keys] + else: + values = list(e.data.values()) + if self.regex: + for val in values: + if isinstance(val, str) and self.regex.search(val): + return True + return False def categorize( diff --git a/tests/test_transforms.py b/tests/test_transforms.py index c30e97d..26612ab 100644 --- a/tests/test_transforms.py +++ b/tests/test_transforms.py @@ -828,3 +828,145 @@ def test_merge_subwatcher_fields_invalid_conflict(): with pytest.raises(ValueError, match="conflict must be"): merge_subwatcher_fields(base, sub, ["project"], conflict="invalid") + + +# --------------------------------------------------------------------------- +# regex_fields rule tests +# --------------------------------------------------------------------------- + + +def _make_event(data: dict) -> Event: + from datetime import timedelta + + return Event( + timestamp=datetime(2024, 1, 1, 12, tzinfo=timezone.utc), + duration=timedelta(seconds=1), + data=data, + ) + + +def test_regex_fields_and_semantics(): + """All named fields must match for the rule to fire.""" + rdp_match = _make_event({"app": "mstsc.exe", "title": "office.example.com"}) + rdp_no_title = _make_event({"app": "mstsc.exe", "title": "other.example.com"}) + winbox = _make_event({"app": "winbox.exe", "title": "office.example.com"}) + + rule = Rule( + { + "type": "regex_fields", + "fields": {"app": r"mstsc\.exe", "title": r"office\.example\.com"}, + } + ) + assert rule.match(rdp_match), "both fields matching must match" + assert not rule.match(rdp_no_title), "title miss must not match" + assert not rule.match(winbox), "wrong app must not match" + + +def test_regex_fields_whole_field_anchoring(): + """Pattern 'mstsc' (no wildcards) must not match 'mstsc.exe'.""" + e = _make_event({"app": "mstsc.exe", "title": "office.example.com"}) + rule = Rule( + { + "type": "regex_fields", + "fields": {"app": "mstsc", "title": r"office\.example\.com"}, + } + ) + assert not rule.match(e), "partial pattern must not match full field value" + + +def test_regex_fields_ignore_case(): + e_upper = _make_event({"app": "MSTSC.EXE", "title": "Office.Example.Com"}) + rule_case = Rule( + { + "type": "regex_fields", + "fields": {"app": r"mstsc\.exe", "title": r"office\.example\.com"}, + } + ) + rule_nocase = Rule( + { + "type": "regex_fields", + "fields": {"app": r"mstsc\.exe", "title": r"office\.example\.com"}, + "ignore_case": True, + } + ) + assert not rule_case.match(e_upper) + assert rule_nocase.match(e_upper) + + +def test_regex_fields_missing_field_no_match(): + """Missing required field must cause rule to not match.""" + e = _make_event({"app": "mstsc.exe"}) # no 'title' + rule = Rule( + { + "type": "regex_fields", + "fields": {"app": r"mstsc\.exe", "title": r".*"}, + } + ) + assert not rule.match(e) + + +def test_regex_fields_embedded_newline(): + """Whole-field anchoring: 'first' must not match 'first\\nsecond'.""" + e = _make_event({"title": "first\nsecond"}) + rule_partial = Rule({"type": "regex_fields", "fields": {"title": "first"}}) + assert not rule_partial.match(e) + + # Note: literal r"first\nsecond" does NOT match the actual embedded newline; + # the user must use [\s\S] or the actual newline character for cross-line patterns. + assert not Rule( + {"type": "regex_fields", "fields": {"title": r"first\nsecond"}} + ).match(e), "literal \\n pattern must not match an actual embedded newline" + + rule_dotall = Rule( + {"type": "regex_fields", "fields": {"title": r"first[\s\S]*second"}} + ) + assert rule_dotall.match(e), r"[\s\S]* should match across embedded newline" + + +def test_regex_fields_rejects_legacy_regex_member(): + """Stale 'regex' member on a regex_fields rule must raise ValueError.""" + import pytest + + with pytest.raises(ValueError, match="'regex' member"): + Rule({"type": "regex_fields", "fields": {"app": "mstsc"}, "regex": "office"}) + + +def test_regex_fields_rejects_select_keys(): + import pytest + + with pytest.raises(ValueError, match="select_keys"): + Rule( + {"type": "regex_fields", "fields": {"app": "mstsc"}, "select_keys": ["app"]} + ) + + +def test_regex_fields_empty_fields_raises(): + import pytest + + with pytest.raises(ValueError): + Rule({"type": "regex_fields", "fields": {}}) + + +def test_regex_fields_categorize_integration(): + """Integration: regex_fields rules work inside categorize().""" + rdp = _make_event({"app": "mstsc.exe", "title": "office.example.com"}) + winbox = _make_event({"app": "winbox.exe", "title": "office.example.com"}) + other = _make_event({"app": "firefox.exe", "title": "office.example.com"}) + + classes = [ + (["RDP"], Rule({"type": "regex_fields", "fields": {"app": r"mstsc\.exe"}})), + (["Winbox"], Rule({"type": "regex_fields", "fields": {"app": r"winbox\.exe"}})), + ] + events = categorize([rdp, winbox, other], classes) + assert events[0].data["$category"] == ["RDP"] + assert events[1].data["$category"] == ["Winbox"] + assert events[2].data["$category"] == ["Uncategorized"] + + +def test_legacy_regex_rule_still_works(): + """Existing 'regex' rules must be unaffected by this change.""" + e = _make_event({"app": "terminal", "title": "just a test"}) + rule = Rule({"regex": "test"}) + assert rule.match(e) + rule_no_match = Rule({"regex": "nonono"}) + assert not rule_no_match.match(e) From 7f27ac2c96027c3ffc65d4ed3fffe85e7a007ad4 Mon Sep 17 00:00:00 2001 From: Bob Date: Thu, 10 Sep 2026 05:44:48 +0000 Subject: [PATCH 2/3] test(classify): fix embedded-newline test assertions \n in a regex pattern is a newline metacharacter (matches actual newline), so the previous assertion was wrong. Rewrote the test to document the actual engine behavior: partial patterns don't match full-field values, \n metacharacter matches embedded newlines, bare dot does not. Git-Session-Id: 3f81 --- tests/test_transforms.py | 27 ++++++++++++++++----------- 1 file changed, 16 insertions(+), 11 deletions(-) diff --git a/tests/test_transforms.py b/tests/test_transforms.py index 26612ab..0159645 100644 --- a/tests/test_transforms.py +++ b/tests/test_transforms.py @@ -906,21 +906,26 @@ def test_regex_fields_missing_field_no_match(): def test_regex_fields_embedded_newline(): - """Whole-field anchoring: 'first' must not match 'first\\nsecond'.""" + """Whole-field anchoring with embedded newline in the field value.""" e = _make_event({"title": "first\nsecond"}) - rule_partial = Rule({"type": "regex_fields", "fields": {"title": "first"}}) - assert not rule_partial.match(e) - # Note: literal r"first\nsecond" does NOT match the actual embedded newline; - # the user must use [\s\S] or the actual newline character for cross-line patterns. - assert not Rule( - {"type": "regex_fields", "fields": {"title": r"first\nsecond"}} - ).match(e), "literal \\n pattern must not match an actual embedded newline" + # "first" alone does not match "first\nsecond" (whole-field). + assert not Rule({"type": "regex_fields", "fields": {"title": "first"}}).match(e) + + # r"first\nsecond" uses \n as a regex newline metacharacter, so it matches. + assert Rule({"type": "regex_fields", "fields": {"title": r"first\nsecond"}}).match( + e + ), r"\n regex metacharacter must match an actual embedded newline" - rule_dotall = Rule( + # [\s\S]* spans lines and also matches. + assert Rule( {"type": "regex_fields", "fields": {"title": r"first[\s\S]*second"}} - ) - assert rule_dotall.match(e), r"[\s\S]* should match across embedded newline" + ).match(e), r"[\s\S]* should match across embedded newline" + + # A dot-only pattern without re.DOTALL does NOT span the newline. + assert not Rule( + {"type": "regex_fields", "fields": {"title": "first.second"}} + ).match(e), "bare dot must not cross embedded newline" def test_regex_fields_rejects_legacy_regex_member(): From a9beedf00ee01fdbf6d5f4f120b8ee9d4eb5021d Mon Sep 17 00:00:00 2001 From: Bob Date: Thu, 10 Sep 2026 06:01:00 +0000 Subject: [PATCH 3/3] fix(classify): reject invalid rule configurations Git-Session-Id: 5673c0d5-27b3-5309-a5c9-0229d1db2bd2 --- aw_transform/classify.py | 9 ++++++++- tests/test_transforms.py | 10 ++++++++++ 2 files changed, 18 insertions(+), 1 deletion(-) diff --git a/aw_transform/classify.py b/aw_transform/classify.py index 2355ded..e3902b1 100644 --- a/aw_transform/classify.py +++ b/aw_transform/classify.py @@ -47,6 +47,8 @@ class Rule: def __init__(self, rules: Dict[str, Any]) -> None: self._rule_type = rules.get("type", "regex") + if self._rule_type not in ("regex", "regex_fields"): + raise ValueError(f"unsupported rule type: {self._rule_type!r}") self.ignore_case = rules.get("ignore_case", False) self.priority = _parse_optional_priority(rules) flags = (re.IGNORECASE if self.ignore_case else 0) | re.UNICODE @@ -78,7 +80,12 @@ def __init__(self, rules: Dict[str, Any]) -> None: raise ValueError( f"regex_fields: pattern for field '{field}' must be a non-empty string" ) - self._field_patterns[field] = re.compile(pattern, flags) + try: + self._field_patterns[field] = re.compile(pattern, flags) + except re.error as exc: + raise ValueError( + f"regex_fields: invalid pattern for field '{field}': {exc}" + ) from exc # Legacy attributes unused for this variant. self.regex = None self.select_keys = None diff --git a/tests/test_transforms.py b/tests/test_transforms.py index 0159645..4fcd2fc 100644 --- a/tests/test_transforms.py +++ b/tests/test_transforms.py @@ -928,6 +928,16 @@ def test_regex_fields_embedded_newline(): ).match(e), "bare dot must not cross embedded newline" +def test_regex_fields_rejects_unknown_rule_type(): + with pytest.raises(ValueError, match="unsupported rule type"): + Rule({"type": "regex_field", "regex": "office"}) + + +def test_regex_fields_invalid_pattern_raises_value_error(): + with pytest.raises(ValueError, match="invalid pattern for field 'app'"): + Rule({"type": "regex_fields", "fields": {"app": "["}}) + + def test_regex_fields_rejects_legacy_regex_member(): """Stale 'regex' member on a regex_fields rule must raise ValueError.""" import pytest