diff --git a/aw_transform/classify.py b/aw_transform/classify.py index 35a33b8..e3902b1 100644 --- a/aw_transform/classify.py +++ b/aw_transform/classify.py @@ -23,36 +23,105 @@ 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") + 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 + + 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" + ) + 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 + 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..4fcd2fc 100644 --- a/tests/test_transforms.py +++ b/tests/test_transforms.py @@ -828,3 +828,160 @@ 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 with embedded newline in the field value.""" + e = _make_event({"title": "first\nsecond"}) + + # "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" + + # [\s\S]* spans lines and also matches. + assert Rule( + {"type": "regex_fields", "fields": {"title": r"first[\s\S]*second"}} + ).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_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 + + 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)