feat(classify): add regex_fields whole-field AND rule type

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(

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)