fix(sanitizer): the depth cap fails closed — nested secrets no longer escape redaction (#3312)

docker/base-image/agent_server/utils/credential_sanitizer.py

@@ -581,10 +581,12 @@ def sanitize_dict(data: Dict[str, Any], depth: int = 0, max_depth: int = 10) ->
         max_depth: Maximum recursion depth to prevent infinite loops
 
     Returns:
-        Sanitized dictionary with credentials replaced
+        Sanitized dictionary with credentials replaced. Past ``max_depth`` the
+        subtree is replaced with ``REDACTION_PLACEHOLDER``, not walked (fail
+        closed, #3312) — the root itself is never cut.
     """
     if depth > max_depth:
-        return data
+        return REDACTION_PLACEHOLDER
 
     result = {}
     for key, value in data.items():
@@ -609,10 +611,12 @@ def sanitize_list(data: List[Any], depth: int = 0, max_depth: int = 10) -> List[
         max_depth: Maximum recursion depth
 
     Returns:
-        Sanitized list with credentials replaced
+        Sanitized list with credentials replaced. Past ``max_depth`` the
+        subtree is replaced with ``REDACTION_PLACEHOLDER``, not walked (fail
+        closed, #3312) — the root itself is never cut.
     """
     if depth > max_depth:
-        return data
+        return REDACTION_PLACEHOLDER
 
     result = []
     for item in data:
@@ -649,8 +653,9 @@ def sanitize_json_string(json_str: str) -> str:
         else:
             return sanitize_text(json_str)
         return json.dumps(sanitized)
-    except json.JSONDecodeError:
-        # If not valid JSON, sanitize as plain text
+    except (json.JSONDecodeError, RecursionError):
+        # Not valid JSON, or nested too deep for json.loads (#3312):
+        # sanitize as plain text
         return sanitize_text(json_str)
 
 
@@ -700,7 +705,8 @@ def sanitize_subprocess_line(line: str) -> str:
             else:
                 return sanitize_text(line)
             return json.dumps(sanitized)
-        except json.JSONDecodeError:
+        except (json.JSONDecodeError, RecursionError):
+            # RecursionError: too deep for json.loads (#3312) — text pass below
             pass
 
     # Fall back to text sanitization

src/backend/utils/credential_sanitizer.py

@@ -446,9 +446,13 @@ def sanitize_text(text: str) -> str:
 
 
 def sanitize_dict(data: Dict[str, Any], depth: int = 0, max_depth: int = 10) -> Dict[str, Any]:
-    """Recursively sanitize sensitive values in a dictionary."""
+    """Recursively sanitize sensitive values in a dictionary.
+
+    Past ``max_depth`` the subtree is replaced with ``REDACTION_PLACEHOLDER``,
+    not walked (fail closed, #3312) — the root itself is never cut.
+    """
     if depth > max_depth:
-        return data
+        return REDACTION_PLACEHOLDER
 
     result = {}
     for key, value in data.items():
@@ -464,9 +468,13 @@ def sanitize_dict(data: Dict[str, Any], depth: int = 0, max_depth: int = 10) ->
 
 
 def sanitize_list(data: List[Any], depth: int = 0, max_depth: int = 10) -> List[Any]:
-    """Recursively sanitize sensitive values in a list."""
+    """Recursively sanitize sensitive values in a list.
+
+    Past ``max_depth`` the subtree is replaced with ``REDACTION_PLACEHOLDER``,
+    not walked (fail closed, #3312) — the root itself is never cut.
+    """
     if depth > max_depth:
-        return data
+        return REDACTION_PLACEHOLDER
 
     result = []
     for item in data:
@@ -503,8 +511,9 @@ def sanitize_json_string(json_str: str) -> str:
         else:
             return sanitize_text(json_str)
         return json.dumps(sanitized)
-    except json.JSONDecodeError:
-        # If not valid JSON, sanitize as plain text
+    except (json.JSONDecodeError, RecursionError):
+        # Not valid JSON, or nested too deep for json.loads (#3312):
+        # sanitize as plain text
         return sanitize_text(json_str)
 
 

tests/unit/test_credential_sanitizer_agent.py

@@ -330,9 +330,11 @@ def test_max_depth_protection(self):
             current = current["nested"]
         current["secret"] = "sk-1234567890abcdefghij1234567890ghij1234567890"
 
-        # Should not crash, should stop at max depth
+        # Should not crash, and must not hand the uncut subtree back raw (#3312)
         result = sanitize_dict(data, max_depth=10)
         assert isinstance(result, dict)
+        assert "sk-1234567890abcdefghij1234567890ghij1234567890" not in json.dumps(result)
+        assert REDACTION_PLACEHOLDER in json.dumps(result)
 
 
 @pytest.mark.unit
@@ -438,6 +440,43 @@ def test_json_array_line(self):
         assert parsed[1] == "normal"
 
 
+@pytest.mark.unit
+class TestDepthFailsClosed:
+    """#3312 — the agent copy carries the same depth cap; past it the subtree
+    is replaced, and a JSON string too deep for `json.loads` falls back to the
+    linear text pass instead of raising RecursionError."""
+
+    SECRET = "ghp_" + "a" * 36
+
+    def test_dict_list_mix_past_max_depth_is_redacted(self):
+        leaf = "key " + self.SECRET
+        for i in range(14):
+            leaf = {"n": leaf} if i % 2 else [leaf]
+        result = sanitize_dict({"root": leaf})
+        assert self.SECRET not in json.dumps(result)
+        assert REDACTION_PLACEHOLDER in json.dumps(result)
+
+    def test_very_deep_object_does_not_recurse(self):
+        leaf = "key " + self.SECRET
+        for _ in range(2000):
+            leaf = [leaf]
+        result = sanitize_list([leaf])
+        assert self.SECRET not in json.dumps(result)
+
+    def _deep_json(self) -> str:
+        return "[" * 200_000 + json.dumps("key " + self.SECRET) + "]" * 200_000
+
+    def test_deep_json_string_does_not_raise(self):
+        out = sanitize_json_string(self._deep_json())
+        assert isinstance(out, str)
+        assert self.SECRET not in out
+
+    def test_deep_subprocess_line_does_not_raise(self):
+        out = sanitize_subprocess_line(self._deep_json())
+        assert isinstance(out, str)
+        assert self.SECRET not in out
+
+
 @pytest.mark.unit
 class TestCredentialValueCache:
     """Test credential value caching from environment."""

tests/unit/test_credential_sanitizer_backend.py

@@ -338,9 +338,11 @@ def test_max_depth_protection(self):
             current = current["nested"]
         current["secret"] = "sk-1234567890abcdefghij1234567890ghij1234567890"
 
-        # Should not crash, should stop at max depth
+        # Should not crash, and must not hand the uncut subtree back raw (#3312)
         result = sanitize_dict(data, max_depth=10)
         assert isinstance(result, dict)
+        assert "sk-1234567890abcdefghij1234567890ghij1234567890" not in json.dumps(result)
+        assert REDACTION_PLACEHOLDER in json.dumps(result)
 
 
 @pytest.mark.unit

tests/unit/test_ec_input_hardening_edges.py

@@ -701,16 +701,12 @@ def test_secret_at_the_last_sanitized_depth_is_redacted(self):
         out = CS.sanitize_dict(data)
         assert GHP not in json.dumps(out)
 
-    @pytest.mark.xfail(strict=True, reason=(
-        "BUG: sanitize_dict/sanitize_list return the subtree UNSANITIZED past "
-        "max_depth (fail-open) — #3312"))
     @pytest.mark.parametrize("shape", ["log", "lists"], ids=["S12-dict-log", "S12-list-nest"])
     def test_secret_below_max_depth_is_still_redacted(self, shape):
-        """S12. `if depth > max_depth: return data` hands back the raw subtree,
-        so any secret nested 12+ containers deep is persisted verbatim. The cap
-        exists to bound recursion (existing test only asserts 'does not crash'),
-        but the safe failure for a REDACTOR is to redact what it cannot walk
-        (e.g. `sanitize_text(json.dumps(subtree))`), not to pass it through.
+        """S12 (#3312). The cap bounds recursion; past it the subtree is
+        replaced with the placeholder (fail closed), never handed back raw —
+        `if depth > max_depth: return data` persisted any secret nested 12+
+        containers deep verbatim.
 
         Reachability: `chat_execution_service.py:434` →
         `sanitize_execution_log` → `sanitize_json_string` → `sanitize_list`.
@@ -726,6 +722,29 @@ def test_secret_below_max_depth_is_still_redacted(self, shape):
         out = CS.sanitize_execution_log(json.dumps(log))
         assert GHP not in out
 
+    @pytest.mark.parametrize("walker", ["dict", "list"])
+    def test_very_deep_object_is_redacted_without_recursing(self, walker):
+        """#3312 — 2000 levels (built iteratively, past the interpreter's
+        recursion limit): no RecursionError, and the secret does not survive.
+        The capped result is shallow, so `json.dumps` over it is safe."""
+        leaf = "key " + GHP
+        for i in range(2000):
+            leaf = {"n": leaf} if (walker == "dict") == (i % 2 == 0) else [leaf]
+        root = {"r": leaf} if walker == "dict" else [leaf]
+        fn = CS.sanitize_dict if walker == "dict" else CS.sanitize_list
+        out = fn(root)
+        assert GHP not in json.dumps(out)
+        assert RED in json.dumps(out)
+
+    def test_pathologically_deep_json_does_not_raise(self):
+        """#3312 AC2 — `json.loads` itself raises RecursionError on a ~200k-deep
+        array; `sanitize_execution_log` must fall back to the linear text pass
+        instead of letting it escape the log-persist path."""
+        raw = "[" * 200_000 + json.dumps("key " + GHP) + "]" * 200_000
+        out = CS.sanitize_execution_log(raw)
+        assert isinstance(out, str)
+        assert GHP not in out
+
     def test_list_and_dict_depth_count_together(self):
         """S23 — lists consume depth exactly like dicts."""
         data = {"l": [[[[{"s": GHP}]]]]}

tests/unit/test_ec_input_hardening_properties.py

@@ -26,7 +26,8 @@
   * IDEMPOTENCE — ``sanitize_text(sanitize_text(x)) == sanitize_text(x)``.
   * NO SECRET SURVIVES — every token family, embedded at arbitrary boundaries
     in arbitrary unicode text, is absent from the output; the same for
-    ``sanitize_dict`` within ``max_depth`` and for ``scrub_secret``.
+    ``sanitize_dict`` at any depth (past ``max_depth`` the subtree is
+    replaced, #3312) and for ``scrub_secret``.
 """
 
 from __future__ import annotations
@@ -323,14 +324,6 @@ def test_sanitize_text_is_idempotent_on_a_quoted_value_with_a_glued_suffix():
 )
 
 
-def _depth(x) -> int:
-    if isinstance(x, dict):
-        return 1 + max((_depth(v) for v in x.values()), default=0)
-    if isinstance(x, list):
-        return 1 + max((_depth(v) for v in x), default=0)
-    return 0
-
-
 def _strings(x):
     if isinstance(x, str):
         yield x
@@ -344,8 +337,7 @@ def _strings(x):
 
 @PROPS
 @given(st.dictionaries(st.sampled_from(["x", "y", "z"]), _kids, max_size=3))
-def test_no_token_survives_sanitize_dict_within_max_depth(data):
-    assume(_depth(data) <= 11)  # the >max_depth fail-open is the edges S12 xfail
+def test_no_token_survives_sanitize_dict_at_any_depth(data):
     out = json.dumps(CS.sanitize_dict(data))
     for s in _strings(data):
         if any(p.fullmatch(s) for p in CS._secret_value_re):