fix: enforce allowlist as structured output boundary

Author: igors93

src/logprivacy/safe_data/normalization.py

@@ -649,10 +649,23 @@ def _process_child(
                 category=blocking_field_rule.category,
             )
 
+        # The allowlist is an output boundary. After blocking rules have had a
+        # chance to stop the operation, a path that is not allowed must not be
+        # reintroduced by a mask, truncate, remove, or pseudonymize rule.
+        allowlist_match: _AllowlistPathMatch = "exact"
+        if self._allowlist_active:
+            allowlist_match = self._allowlist.match_path(path)
+            if allowlist_match == "none":
+                counters.not_allowed += 1
+                return _OMIT
+            if allowlist_match == "prefix" and _is_definitely_scalar(value):
+                counters.not_allowed += 1
+                return _OMIT
+
         path_rule = next((rule for rule in path_rules if rule.action != "block"), None)
         if path_rule is not None:
             counters.path_rule_matches += 1
-            return self._apply_action(
+            processed = self._apply_action(
                 path_rule.action,
                 value,
                 depth=depth,
@@ -662,16 +675,11 @@ def _process_child(
                 max_chars=path_rule.max_chars,
                 category=path_rule.category,
             )
-
-        allowlist_match: _AllowlistPathMatch = "exact"
-        if self._allowlist_active:
-            allowlist_match = self._allowlist.match_path(path)
-            if allowlist_match == "none":
-                counters.not_allowed += 1
-                return _OMIT
-            if allowlist_match == "prefix" and _is_definitely_scalar(value):
-                counters.not_allowed += 1
-                return _OMIT
+            return self._omit_non_container_prefix_result(
+                processed,
+                allowlist_match=allowlist_match,
+                counters=counters,
+            )
 
         if field_name is not None:
             field_rule = next((rule for rule in field_rules if rule.action != "block"), None)

tests/__pycache__/test_api.cpython-314-pytest-9.0.3.pyc

@@ -0,0 +1,128 @@
+"""Regression tests for allowlist precedence as an output boundary."""
+
+from __future__ import annotations
+
+import pytest
+
+from logprivacy import (
+    CleanerPolicy,
+    FieldRule,
+    LogBlockedError,
+    PathRule,
+    to_safe_data,
+    to_safe_data_with_result,
+)
+
+
+def test_allowlist_omits_path_even_when_path_rule_would_mask() -> None:
+    policy = (
+        CleanerPolicy.default()
+        .allow_paths("status")
+        .add_path_rules(PathRule.exact("account.balance", action="mask"))
+    )
+
+    result = to_safe_data_with_result(
+        {
+            "status": "ok",
+            "account": {"balance": "10000"},
+        },
+        policy=policy,
+    )
+
+    assert result.cleaned == {"status": "ok"}
+    assert result.stats.not_allowed == 1
+    assert result.stats.path_rule_matches == 0
+
+
+def test_allowlist_omits_path_even_when_path_rule_would_remove() -> None:
+    policy = (
+        CleanerPolicy.default()
+        .allow_paths("status")
+        .add_path_rules(PathRule.exact("debug", action="remove"))
+    )
+
+    safe = to_safe_data(
+        {
+            "status": "ok",
+            "debug": "internal details",
+        },
+        policy=policy,
+    )
+
+    assert safe == {"status": "ok"}
+    assert "debug" not in safe
+
+
+def test_allowlisted_path_still_uses_path_rule() -> None:
+    policy = (
+        CleanerPolicy.default()
+        .allow_paths("account.balance")
+        .add_path_rules(PathRule.exact("account.balance", action="mask"))
+    )
+
+    result = to_safe_data_with_result(
+        {
+            "account": {
+                "balance": "10000",
+                "currency": "USD",
+            }
+        },
+        policy=policy,
+    )
+
+    assert result.cleaned == {"account": {"balance": "[SECRET]"}}
+    assert result.stats.path_rule_matches == 1
+    assert result.stats.not_allowed == 1
+
+
+def test_parent_prefix_cannot_be_reintroduced_as_masked_scalar() -> None:
+    policy = (
+        CleanerPolicy.default()
+        .allow_paths("account.balance")
+        .add_path_rules(PathRule.exact("account", action="mask"))
+    )
+
+    result = to_safe_data_with_result(
+        {"account": {"balance": "10000"}},
+        policy=policy,
+    )
+
+    assert result.cleaned == {}
+    assert result.stats.not_allowed == 1
+
+
+def test_allowlist_runs_before_non_blocking_field_rule() -> None:
+    policy = (
+        CleanerPolicy.default()
+        .allow_paths("status")
+        .add_field_rules(FieldRule.exact("debug", action="mask"))
+    )
+
+    result = to_safe_data_with_result(
+        {
+            "status": "ok",
+            "debug": "internal details",
+        },
+        policy=policy,
+    )
+
+    assert result.cleaned == {"status": "ok"}
+    assert result.stats.field_rule_matches == 0
+    assert result.stats.not_allowed == 1
+
+
+def test_block_still_precedes_allowlist() -> None:
+    policy = (
+        CleanerPolicy.default()
+        .allow_paths("status")
+        .add_path_rules(PathRule.exact("debug", action="block"))
+    )
+
+    with pytest.raises(LogBlockedError):
+        to_safe_data(
+            {
+                "status": "ok",
+                "debug": "internal details",
+            },
+            policy=policy,
+        )