fix: harden allowlist patterns against ReDoS, document match semantics

Review findings: reject quantified groups containing nested quantifiers
at compile time (catastrophic backtracking on attacker-influenced entity
text), cap pattern length at 512 chars, and skip pattern matching for
entities longer than 512 chars (fail-safe: the finding is kept). Match
semantics documented as case-sensitive with no Unicode normalization;
allowlist entries are operator configuration, never end-user input.
Adds regression tests for the rejection heuristic, the smart-engine
path, and the redact(entities=..., allowlist=...) guard. Replaces a
walrus assignment with a plain one in the litellm adapter.

Author: sidmohan0

datafog/engine.py

@@ -281,11 +281,35 @@ def _filter_entity_types(
     return [entity for entity in entities if entity.type in allowed]
 
 
+# Python's re module backtracks; a quantified group containing another
+# quantifier (e.g. ``(a+)+``) can take exponential time on adversarial
+# input, and entity text can be attacker-influenced (LLM messages, tool
+# output). Reject that construct outright rather than matching under it.
+_NESTED_QUANTIFIER = re.compile(
+    r"\((?:[^()\\]|\\.)*(?<!\\)[+*}](?:[^()\\]|\\.)*\)\s*[+*{]"
+)
+MAX_ALLOWLIST_PATTERN_LENGTH = 512
+# Entities longer than this skip pattern matching (fail-safe: the finding
+# is kept, never suppressed) so match time stays bounded.
+MAX_PATTERN_SUBJECT_LENGTH = 512
+
+
 def _compile_allowlist_patterns(
     allowlist_patterns: Optional[list[str]],
 ) -> list["re.Pattern[str]"]:
     compiled = []
     for raw in allowlist_patterns or []:
+        if len(raw) > MAX_ALLOWLIST_PATTERN_LENGTH:
+            raise ValueError(
+                "allowlist_patterns entries must be at most "
+                f"{MAX_ALLOWLIST_PATTERN_LENGTH} characters"
+            )
+        if _NESTED_QUANTIFIER.search(raw):
+            raise ValueError(
+                "allowlist_patterns contains a quantified group with a nested "
+                f"quantifier ({raw!r}), which risks catastrophic backtracking; "
+                "rewrite the pattern without nesting quantifiers"
+            )
         try:
             compiled.append(re.compile(raw))
         except re.error as exc:
@@ -302,8 +326,12 @@ def _apply_allowlist(
 ) -> list[Entity]:
     """Drop entities whose exact text is allowlisted.
 
-    Exact values match the full entity text; patterns must fullmatch it,
-    so a partial match never suppresses a finding.
+    Matching semantics, deliberately strict for a security boundary:
+    exact values are case-sensitive with no Unicode normalization, and
+    patterns must fullmatch the entity text, so a partial match never
+    suppresses a finding. Allowlist entries and patterns are operator
+    configuration; treat them like code and never accept them from end
+    users.
     """
     if not allowlist and not allowlist_patterns:
         return entities
@@ -313,7 +341,11 @@ def _apply_allowlist(
         entity
         for entity in entities
         if entity.text not in exact
-        and not any(pattern.fullmatch(entity.text) for pattern in patterns)
+        and not any(
+            pattern.fullmatch(entity.text)
+            for pattern in patterns
+            if len(entity.text) <= MAX_PATTERN_SUBJECT_LENGTH
+        )
     ]
 
 

datafog/integrations/litellm_guardrail.py

@@ -197,10 +197,9 @@ async def async_pre_call_hook(
                 },
             )
 
-        self._record_guardrail_logging(
-            new_data_final := {**data, "messages": new_messages}, total_counts
-        )
-        return new_data_final
+        new_data = {**data, "messages": new_messages}
+        self._record_guardrail_logging(new_data, total_counts)
+        return new_data
 
     def _record_guardrail_logging(
         self, data: dict, total_counts: dict[str, int]

tests/test_allowlist.py

@@ -23,9 +23,7 @@ def test_allowlisted_value_is_not_reported(self):
         assert [e.text for e in result.entities] == [EMAIL]
 
     def test_allowlist_is_exact_not_substring(self):
-        result = datafog.scan(
-            f"mail {EMAIL}", engine="regex", allowlist=["jane.doe"]
-        )
+        result = datafog.scan(f"mail {EMAIL}", engine="regex", allowlist=["jane.doe"])
         assert [e.text for e in result.entities] == [EMAIL]
 
     def test_empty_allowlist_is_noop(self):
@@ -76,6 +74,47 @@ def test_patterns_and_values_combine(self):
         assert result.entities == []
 
 
+class TestReDoSGuards:
+    def test_catastrophic_pattern_rejected(self):
+        with pytest.raises(ValueError, match="catastrophic backtracking"):
+            datafog.scan("text", engine="regex", allowlist_patterns=[r"(a+)+$"])
+
+    def test_nested_star_rejected(self):
+        with pytest.raises(ValueError, match="catastrophic backtracking"):
+            datafog.scan("text", engine="regex", allowlist_patterns=[r"(.*)*"])
+
+    def test_overlong_pattern_rejected(self):
+        with pytest.raises(ValueError, match="at most"):
+            datafog.scan("text", engine="regex", allowlist_patterns=["a" * 513])
+
+    def test_benign_quantified_group_still_allowed(self):
+        result = datafog.scan(
+            f"mail {EMAIL}",
+            engine="regex",
+            allowlist_patterns=[r"(abc)+", r".*@example\.com"],
+        )
+        assert result.entities == []  # broad pattern suppresses, no rejection
+
+
+class TestEnginePaths:
+    def test_smart_engine_applies_allowlist(self):
+        import warnings as _warnings
+
+        with _warnings.catch_warnings():
+            _warnings.simplefilter("ignore")
+            result = datafog.scan(f"mail {EMAIL}", engine="smart", allowlist=[EMAIL])
+        assert result.entities == []
+
+    def test_redact_rejects_allowlist_with_explicit_entities(self):
+        scanned = datafog.scan(f"mail {EMAIL}", engine="regex")
+        with pytest.raises(ValueError, match="cannot be combined"):
+            datafog.redact(
+                f"mail {EMAIL}",
+                entities=scanned.entities,
+                allowlist=[EMAIL],
+            )
+
+
 class TestPresidioAliases:
     def test_email_address_alias(self):
         result = datafog.scan(
@@ -85,9 +124,7 @@ def test_email_address_alias(self):
 
     def test_us_ssn_alias(self):
         ssn = "856-45-" "6789"
-        result = datafog.scan(
-            f"ssn {ssn}", engine="regex", entity_types=["US_SSN"]
-        )
+        result = datafog.scan(f"ssn {ssn}", engine="regex", entity_types=["US_SSN"])
         assert [e.type for e in result.entities] == ["SSN"]