fix(gateway): mask a credential nested inside a free-form settings dict

src/gateway/models/secret_fields.py

@@ -18,12 +18,35 @@ class can take its vendor key as a parameter), and none of them may echo one
 down.
 """
 
-from typing import Any
+from typing import Any, cast
 
 # Substrings (matched case-insensitively against a key name) that a
 # credential-bearing field is expected to contain.
 _SECRET_LOOKING_KEY_SUBSTRINGS = ("key", "secret", "token", "password", "authorization", "credential")
 REDACTED_VALUE = "***"
+# Past this, a nested value is masked wholesale rather than walked. See _redact_node.
+_MAX_NESTING_DEPTH = 16
+
+
+def _looks_secret(key: str) -> bool:
+    return any(marker in key.lower() for marker in _SECRET_LOOKING_KEY_SUBSTRINGS)
+
+
+def _redact_node(node: Any, depth: int) -> Any:
+    """Mask a value in place in the tree; containers recurse, leaves pass through."""
+    if depth >= _MAX_NESTING_DEPTH:
+        # Fail closed. Nothing legitimate nests this deep in a kwargs blob, and
+        # the alternative to masking is either a RecursionError turning a read
+        # into a 500 or a depth the masking never reaches.
+        return REDACTED_VALUE
+    if isinstance(node, dict):
+        return {
+            key: REDACTED_VALUE if _looks_secret(str(key)) else _redact_node(value, depth + 1)
+            for key, value in node.items()
+        }
+    if isinstance(node, list):
+        return [_redact_node(item, depth + 1) for item in node]
+    return node
 
 
 def redact_secret_like_values(values: dict[str, Any] | None) -> dict[str, Any] | None:
@@ -32,13 +55,46 @@ def redact_secret_like_values(values: dict[str, Any] | None) -> dict[str, Any] |
     Substring match, not an exact-name allow-list: an operator can name a
     Bedrock/vertex/custom client kwarg however any-llm expects it, so a fixed
     set of exact names would miss a variant spelling and silently leak it.
+
+    Nested objects and lists are walked, because none of the four columns
+    constrains its shape and a credential one level down was returned in clear
+    (otari#1125). A matching key masks its value WHOLE, dict or list included —
+    the same thing a matching top-level key has always done to a non-scalar, so
+    depth 0 behaves exactly as before.
+
+    Lists are walked but their bare elements are never masked: an element has no
+    key name to match on, and masking one on its value would be a guess. Only a
+    mapping inside a list can carry a masked entry. :func:`restore_redacted_values`
+    depends on that — see its own note.
     """
     if values is None:
         return None
-    return {
-        key: REDACTED_VALUE if any(marker in key.lower() for marker in _SECRET_LOOKING_KEY_SUBSTRINGS) else value
-        for key, value in values.items()
-    }
+    # The walker returns whatever shape it was given, and it was given a dict.
+    # The depth bound cannot fire at the root, so this is not a `dict | str`.
+    return cast(dict[str, Any], _redact_node(values, 0))
+
+
+def _restore_node(incoming: Any, stored: Any, depth: int) -> Any:
+    """Prefer the stored value wherever the caller echoed the mask back, at any depth."""
+    if depth >= _MAX_NESTING_DEPTH:
+        return incoming
+    if isinstance(incoming, dict):
+        stored_map = stored if isinstance(stored, dict) else {}
+        out: dict[Any, Any] = {}
+        for key, value in incoming.items():
+            if value == REDACTED_VALUE and key in stored_map:
+                out[key] = stored_map[key]
+            else:
+                out[key] = _restore_node(value, stored_map.get(key), depth + 1)
+        return out
+    if isinstance(incoming, list):
+        # Positional, and only when the shapes still line up: a list the caller
+        # resized is a rewrite, and pairing it off by index would splice stored
+        # values into positions that no longer mean the same thing.
+        if isinstance(stored, list) and len(stored) == len(incoming):
+            return [_restore_node(item, stored[index], depth + 1) for index, item in enumerate(incoming)]
+        return list(incoming)
+    return incoming
 
 
 def restore_redacted_values(
@@ -59,11 +115,19 @@ def restore_redacted_values(
     so, since the two are the same bytes on the wire. Clearing the entry and
     setting it again is the way out, and losing that beats overwriting a
     credential with a placeholder.
+
+    Walks nested objects and lists IN STEP with the masking, and that pairing is
+    the whole point rather than a detail: the moment the mask reaches a nested
+    entry, a restore that still walks one level writes ``***`` into the database
+    where a credential used to be on the next PATCH. Worse than the leak it was
+    fixing (otari#1125).
+
+    A bare ``***`` inside a LIST is taken literally, because masking never puts
+    one there — list elements have no key to match on — so an element that looks
+    like the mask came from the caller and means itself.
     """
     if incoming is None:
         return None
     if not stored:
         return dict(incoming)
-    return {
-        key: stored[key] if value == REDACTED_VALUE and key in stored else value for key, value in incoming.items()
-    }
+    return cast(dict[str, Any], _restore_node(incoming, stored, 0))

tests/unit/test_secret_fields.py

@@ -55,6 +55,94 @@ def test_none_stays_none(self) -> None:
         assert redact_secret_like_values(None) is None
 
 
+class TestNestedRedaction:
+    """otari#1125: the walk stopped at depth one, so a credential one level down
+    was returned in clear. Every cell here pairs the new masking with the restore
+    that has to move with it — a mask that reaches a nested entry while the
+    restore still walks one level writes ``***`` over the credential on the next
+    PATCH, which is worse than the leak it was fixing."""
+
+    def test_a_nested_credential_is_masked(self) -> None:
+        assert redact_secret_like_values({"headers": {"api_key": "secret"}, "region_name": "eu-west-1"}) == {
+            "headers": {"api_key": REDACTED_VALUE},
+            "region_name": "eu-west-1",
+        }
+
+    def test_a_credential_inside_a_list_of_objects_is_masked(self) -> None:
+        assert redact_secret_like_values({"extra_headers": [{"name": "x", "token": "live"}, {"name": "y"}]}) == {
+            "extra_headers": [{"name": "x", "token": REDACTED_VALUE}, {"name": "y"}]
+        }
+
+    def test_a_matching_key_masks_its_whole_subtree(self) -> None:
+        # Same thing a matching top-level key has always done to a non-scalar:
+        # the name is the signal, so nothing under it is shown either.
+        assert redact_secret_like_values({"credentials": {"user": "bob", "passphrase": "p"}}) == {
+            "credentials": REDACTED_VALUE
+        }
+
+    def test_a_bare_mask_in_a_list_is_never_produced(self) -> None:
+        # A list element has no key to match on, so masking one would be a guess.
+        # `restore` leans on this: an element that looks like the mask came from
+        # the caller and means itself.
+        assert redact_secret_like_values({"values": ["***", "plain"]}) == {"values": ["***", "plain"]}
+
+    def test_deep_nesting_is_masked_rather_than_walked_forever(self) -> None:
+        # Fail closed past the bound: the alternatives are a RecursionError
+        # turning a read into a 500, or a depth the masking never reaches.
+        deep: dict[str, object] = {"leaf": "visible"}
+        for _ in range(40):
+            deep = {"nest": deep}
+
+        assert redact_secret_like_values(deep) != deep
+        assert REDACTED_VALUE in str(redact_secret_like_values(deep))
+
+
+class TestNestedRoundTrip:
+    def test_a_nested_credential_survives_an_edit_of_its_sibling(self) -> None:
+        # The failure this prevents: the dashboard loads a row, changes one
+        # visible field, and saves the whole object back.
+        stored = {"headers": {"api_key": "live-secret", "trace": "off"}}
+        echoed = redact_secret_like_values(stored)
+        assert echoed is not None
+        echoed["headers"]["trace"] = "on"
+
+        assert restore_redacted_values(echoed, stored) == {
+            "headers": {"api_key": "live-secret", "trace": "on"}
+        }
+
+    def test_a_masked_subtree_is_restored_whole(self) -> None:
+        stored = {"credentials": {"user": "bob", "passphrase": "p"}}
+        echoed = redact_secret_like_values(stored)
+
+        assert restore_redacted_values(echoed, stored) == stored
+
+    def test_a_credential_inside_a_list_survives_the_round_trip(self) -> None:
+        stored = {"extra_headers": [{"name": "x", "token": "live"}, {"name": "y"}]}
+        echoed = redact_secret_like_values(stored)
+
+        assert restore_redacted_values(echoed, stored) == stored
+
+    def test_a_resized_list_is_a_rewrite_and_not_paired_off_by_index(self) -> None:
+        # Splicing stored values into positions that no longer mean the same
+        # thing would put a credential under a different header.
+        stored = {"extra_headers": [{"token": "live-a"}, {"token": "live-b"}]}
+        submitted = {"extra_headers": [{"token": REDACTED_VALUE}]}
+
+        assert restore_redacted_values(submitted, stored) == {"extra_headers": [{"token": REDACTED_VALUE}]}
+
+    def test_a_real_nested_value_still_replaces_the_stored_one(self) -> None:
+        # The control for the whole pairing: restoring must not mean "the caller
+        # can never change a nested credential".
+        stored = {"headers": {"api_key": "old"}}
+
+        assert restore_redacted_values({"headers": {"api_key": "new"}}, stored) == {"headers": {"api_key": "new"}}
+
+    def test_a_nested_entry_the_caller_dropped_stays_dropped(self) -> None:
+        stored = {"headers": {"api_key": "live", "trace": "on"}}
+
+        assert restore_redacted_values({"headers": {"trace": "on"}}, stored) == {"headers": {"trace": "on"}}
+
+
 class TestRestoreRedactedValues:
     def test_the_mask_echoed_back_keeps_the_stored_value(self) -> None:
         stored = {"region_name": "us-east-1", "aws_secret_access_key": "wJalrXUtnFEMI"}