From 87fb6a4e618f3c1596f95df291059853eeddafd6 Mon Sep 17 00:00:00 2001
From: Christopher <131207452+PetrefiedThunder@users.noreply.github.com>
Date: Sat, 24 Oct 2026 22:35:10 -0700
Subject: [PATCH 1/5] fix(gateway): keep the admin key out of approvals CLI
 errors

Reject API keys with C0, DEL or C1 control characters using a fixed message, and print only fixed messages (at most the exception class name) in every CLI error path, so h11 header errors can no longer echo the key to stderr.

Signed-off-by: Christopher <131207452+PetrefiedThunder@users.noreply.github.com>
---
 gateway/scripts/amw_approvals.py        | 53 ++++++++++----
 gateway/tests/test_amw_approvals_cli.py | 94 +++++++++++++++++++++++++
 2 files changed, 133 insertions(+), 14 deletions(-)

diff --git a/gateway/scripts/amw_approvals.py b/gateway/scripts/amw_approvals.py
index 09e93cd5..3accb043 100644
--- a/gateway/scripts/amw_approvals.py
+++ b/gateway/scripts/amw_approvals.py
@@ -121,11 +121,27 @@ def _refusal_reason(base_url: str) -> str | None:
     return f"non-loopback http host {host!r}"
 
 
+def _has_control_chars(value: str) -> bool:
+    """True when value holds C0, DEL, or C1 controls."""
+    for ch in value:
+        code = ord(ch)
+        if code < 0x20 or code == 0x7F or 0x80 <= code <= 0x9F:
+            return True
+    return False
+
+
 def _read_api_key() -> str:
-    """First key from API_KEY_ENV; empty when unset or blank."""
+    """First key from API_KEY_ENV; empty when unset or blank.
+
+    Raises ValueError with a fixed message when the key holds control
+    characters. The message never includes the value.
+    """
     for entry in os.environ.get(API_KEY_ENV, "").split(","):
         if entry.strip():
-            return entry.strip()
+            candidate = entry.strip()
+            if _has_control_chars(entry) or _has_control_chars(candidate):
+                raise ValueError("invalid API key in VALID_API_KEYS")
+            return candidate
     return ""
 
 
@@ -158,15 +174,16 @@ def cmd_list(args: argparse.Namespace) -> int:
     try:
         response = _client(args.base_url, args.api_key).get("/v1/approvals/pending")
         response.raise_for_status()
-    except ValueError as exc:
-        print(f"error: {exc}", file=sys.stderr)
+    except ValueError:
+        print("error: refusing to send the API key", file=sys.stderr)
         return 1
-    except httpx.InvalidURL as exc:
-        print(f"error: invalid gateway URL: {exc}", file=sys.stderr)
+    except httpx.InvalidURL:
+        print("error: invalid gateway URL", file=sys.stderr)
         return 1
     except httpx.HTTPError as exc:
         print(
-            f"error: gateway not reachable at {args.base_url}: {exc}", file=sys.stderr
+            f"error: gateway request failed ({type(exc).__name__})",
+            file=sys.stderr,
         )
         return 1
     approvals = response.json().get("approvals", [])
@@ -186,15 +203,16 @@ def _decide(args: argparse.Namespace, action: str) -> int:
         response = _client(args.base_url, args.api_key).post(
             f"/v1/approvals/{args.nonce}/{action}"
         )
-    except ValueError as exc:
-        print(f"error: {exc}", file=sys.stderr)
+    except ValueError:
+        print("error: refusing to send the API key", file=sys.stderr)
         return 1
-    except httpx.InvalidURL as exc:
-        print(f"error: invalid gateway URL: {exc}", file=sys.stderr)
+    except httpx.InvalidURL:
+        print("error: invalid gateway URL", file=sys.stderr)
         return 1
     except httpx.HTTPError as exc:
         print(
-            f"error: gateway not reachable at {args.base_url}: {exc}", file=sys.stderr
+            f"error: gateway request failed ({type(exc).__name__})",
+            file=sys.stderr,
         )
         return 1
     if response.status_code == 404:
@@ -210,7 +228,10 @@ def _decide(args: argparse.Namespace, action: str) -> int:
     try:
         response.raise_for_status()
     except httpx.HTTPError as exc:
-        print(f"error: {exc}", file=sys.stderr)
+        print(
+            f"error: gateway request failed ({type(exc).__name__})",
+            file=sys.stderr,
+        )
         return 1
     print(f"{args.nonce}: {response.json().get('status')}")
     return 0
@@ -247,7 +268,11 @@ def main(argv: list[str] | None = None) -> int:
             file=sys.stderr,
         )
         return 1
-    api_key = _read_api_key()
+    try:
+        api_key = _read_api_key()
+    except ValueError:
+        print("error: invalid API key in VALID_API_KEYS", file=sys.stderr)
+        return 1
     if not api_key:
         print(
             "error: VALID_API_KEYS is not set; export the gateway "
diff --git a/gateway/tests/test_amw_approvals_cli.py b/gateway/tests/test_amw_approvals_cli.py
index 3a2f7227..5c7da039 100644
--- a/gateway/tests/test_amw_approvals_cli.py
+++ b/gateway/tests/test_amw_approvals_cli.py
@@ -348,3 +348,97 @@ def test_client_disables_trust_env_and_ignores_proxies(
             proving.close()
     finally:
         client.close()
+
+
+def _assert_no_key_leak(key: str, out: str, err: str) -> None:
+    assert key not in out
+    assert key not in err
+    for start in range(len(key) - 7):
+        part = key[start : start + 8]
+        assert part not in out
+        assert part not in err
+
+
+@pytest.mark.parametrize("injected", ["\n", "\r", "\x85", "\x7f"])
+def test_control_char_key_rejected_without_request(
+    monkeypatch: pytest.MonkeyPatch,
+    capsys: pytest.CaptureFixture[str],
+    injected: str,
+) -> None:
+    token = secrets.token_hex(16)
+    key = f"{token[:16]}{injected}{token[16:]}"
+    monkeypatch.setattr(amw_approvals.httpx, "Client", _ExplodingClient)
+    monkeypatch.setenv("VALID_API_KEYS", key)
+    base_url = f"http://{LOOPBACK_V4}:{HTTP_TEST_PORT}"
+    nonce = "test-nonce"
+    for argv in (["list"], ["approve", nonce], ["deny", nonce]):
+        assert amw_approvals.main(["--base-url", base_url, *argv]) == 1
+        out = capsys.readouterr()
+        assert "Traceback" not in out.err
+        assert "Traceback" not in out.out
+        assert "invalid API key" in out.err
+        _assert_no_key_leak(key, out.out, out.err)
+
+
+def test_http_error_hides_key(
+    monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
+) -> None:
+    key = secrets.token_hex(16)
+
+    class _LeakingClient:
+        def __init__(self, *args: object, **kwargs: object) -> None:
+            pass
+
+        def get(self, path: str) -> _StubResponse:
+            raise amw_approvals.httpx.ConnectError(f"dial boom {key} quoted")
+
+        def post(self, path: str) -> _StubResponse:
+            raise amw_approvals.httpx.ConnectError(f"dial boom {key} quoted")
+
+    monkeypatch.setattr(amw_approvals.httpx, "Client", _LeakingClient)
+    monkeypatch.setenv("VALID_API_KEYS", key)
+    base_url = f"http://{LOOPBACK_V4}:{HTTP_TEST_PORT}"
+    nonce = "test-nonce"
+    for argv in (["list"], ["approve", nonce], ["deny", nonce]):
+        assert amw_approvals.main(["--base-url", base_url, *argv]) == 1
+        out = capsys.readouterr()
+        assert "Traceback" not in out.err
+        assert "Traceback" not in out.out
+        assert "gateway request failed" in out.err
+        assert "ConnectError" in out.err
+        _assert_no_key_leak(key, out.out, out.err)
+
+
+def test_status_error_hides_key(
+    monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
+) -> None:
+    key = secrets.token_hex(16)
+
+    class _StatusLeakResponse:
+        status_code = 500
+
+        def raise_for_status(self) -> None:
+            raise amw_approvals.httpx.ConnectError(f"status boom {key} quoted")
+
+    class _StatusLeakClient:
+        def __init__(self, *args: object, **kwargs: object) -> None:
+            pass
+
+        def get(self, path: str) -> _StatusLeakResponse:
+            return _StatusLeakResponse()
+
+        def post(self, path: str) -> _StatusLeakResponse:
+            return _StatusLeakResponse()
+
+    monkeypatch.setattr(amw_approvals.httpx, "Client", _StatusLeakClient)
+    monkeypatch.setenv("VALID_API_KEYS", key)
+    base_url = f"http://{LOOPBACK_V4}:{HTTP_TEST_PORT}"
+    nonce = "test-nonce"
+    for argv in (["list"], ["approve", nonce], ["deny", nonce]):
+        assert amw_approvals.main(["--base-url", base_url, *argv]) == 1
+        out = capsys.readouterr()
+        assert "Traceback" not in out.err
+        assert "Traceback" not in out.out
+        assert "gateway request failed" in out.err
+        assert "ConnectError" in out.err
+        _assert_no_key_leak(key, out.out, out.err)

From 16d6149872ad63652424b1abb276349d3ba4ca6f Mon Sep 17 00:00:00 2001
From: Christopher <131207452+PetrefiedThunder@users.noreply.github.com>
Date: Sat, 24 Oct 2026 22:49:42 -0700
Subject: [PATCH 2/5] fix(gateway): report HTTP status and accept stripped keys
 in approvals CLI

Print only the numeric status on HTTPStatusError, strip the key before the control-character check, show the validated base_url on connect failure, and handle non-JSON responses with a fixed message.

Signed-off-by: Christopher <131207452+PetrefiedThunder@users.noreply.github.com>
---
 gateway/scripts/amw_approvals.py        |  51 ++++++++++--
 gateway/tests/test_amw_approvals_cli.py | 102 +++++++++++++++++++-----
 2 files changed, 130 insertions(+), 23 deletions(-)

diff --git a/gateway/scripts/amw_approvals.py b/gateway/scripts/amw_approvals.py
index 3accb043..29992b5e 100644
--- a/gateway/scripts/amw_approvals.py
+++ b/gateway/scripts/amw_approvals.py
@@ -137,9 +137,9 @@ def _read_api_key() -> str:
     characters. The message never includes the value.
     """
     for entry in os.environ.get(API_KEY_ENV, "").split(","):
-        if entry.strip():
-            candidate = entry.strip()
-            if _has_control_chars(entry) or _has_control_chars(candidate):
+        candidate = entry.strip()
+        if candidate:
+            if _has_control_chars(candidate):
                 raise ValueError("invalid API key in VALID_API_KEYS")
             return candidate
     return ""
@@ -180,13 +180,30 @@ def cmd_list(args: argparse.Namespace) -> int:
     except httpx.InvalidURL:
         print("error: invalid gateway URL", file=sys.stderr)
         return 1
+    except httpx.HTTPStatusError as exc:
+        print(
+            f"gateway request failed: HTTP {exc.response.status_code}",
+            file=sys.stderr,
+        )
+        return 1
+    except httpx.TransportError as exc:
+        print(
+            f"gateway unreachable at {args.base_url} ({type(exc).__name__})",
+            file=sys.stderr,
+        )
+        return 1
     except httpx.HTTPError as exc:
         print(
             f"error: gateway request failed ({type(exc).__name__})",
             file=sys.stderr,
         )
         return 1
-    approvals = response.json().get("approvals", [])
+    try:
+        approvals = response.json().get("approvals", [])
+    except ValueError:
+        # ValueError covers json.JSONDecodeError from a non-JSON body.
+        print("gateway returned a non-JSON response", file=sys.stderr)
+        return 1
     if not approvals:
         print("no pending approvals")
         return 0
@@ -209,6 +226,18 @@ def _decide(args: argparse.Namespace, action: str) -> int:
     except httpx.InvalidURL:
         print("error: invalid gateway URL", file=sys.stderr)
         return 1
+    except httpx.HTTPStatusError as exc:
+        print(
+            f"gateway request failed: HTTP {exc.response.status_code}",
+            file=sys.stderr,
+        )
+        return 1
+    except httpx.TransportError as exc:
+        print(
+            f"gateway unreachable at {args.base_url} ({type(exc).__name__})",
+            file=sys.stderr,
+        )
+        return 1
     except httpx.HTTPError as exc:
         print(
             f"error: gateway request failed ({type(exc).__name__})",
@@ -227,13 +256,25 @@ def _decide(args: argparse.Namespace, action: str) -> int:
         return 1
     try:
         response.raise_for_status()
+    except httpx.HTTPStatusError as exc:
+        print(
+            f"gateway request failed: HTTP {exc.response.status_code}",
+            file=sys.stderr,
+        )
+        return 1
     except httpx.HTTPError as exc:
         print(
             f"error: gateway request failed ({type(exc).__name__})",
             file=sys.stderr,
         )
         return 1
-    print(f"{args.nonce}: {response.json().get('status')}")
+    try:
+        status = response.json().get("status")
+    except ValueError:
+        # ValueError covers json.JSONDecodeError from a non-JSON body.
+        print("gateway returned a non-JSON response", file=sys.stderr)
+        return 1
+    print(f"{args.nonce}: {status}")
     return 0
 
 
diff --git a/gateway/tests/test_amw_approvals_cli.py b/gateway/tests/test_amw_approvals_cli.py
index 5c7da039..62330df0 100644
--- a/gateway/tests/test_amw_approvals_cli.py
+++ b/gateway/tests/test_amw_approvals_cli.py
@@ -7,8 +7,10 @@
 from __future__ import annotations
 
 import importlib.util
+import json
 import secrets
 from pathlib import Path
+from typing import Any
 
 import pytest
 
@@ -57,8 +59,14 @@ def test_first_comma_separated_entry_wins(
 
 def test_client_attaches_key_header(monkeypatch: pytest.MonkeyPatch) -> None:
     monkeypatch.setenv("VALID_API_KEYS", PLACEHOLDER_KEY)
-    client = amw_approvals._client("http://127.0.0.1:9", amw_approvals._read_api_key())
-    assert client.headers["X-API-Key"] == PLACEHOLDER_KEY
+    # RFC 1122 loopback, RFC 6335 fixed port: allowed http loopback URL.
+    client = amw_approvals._client(
+        f"http://{LOOPBACK_V4}:{HTTP_TEST_PORT}", amw_approvals._read_api_key()
+    )
+    try:
+        assert client.headers["X-API-Key"] == PLACEHOLDER_KEY
+    finally:
+        client.close()
 
 
 class _StubResponse:
@@ -113,11 +121,15 @@ def post(self, path: str) -> _StubResponse:
 def test_unreachable_gateway_error_hides_key(
     monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
 ) -> None:
-    monkeypatch.setenv("VALID_API_KEYS", PLACEHOLDER_KEY)
-    assert amw_approvals.main(["--base-url", "http://127.0.0.1:9", "list"]) == 1
+    key = secrets.token_hex(16)
+    monkeypatch.setenv("VALID_API_KEYS", key)
+    # RFC 1122 loopback; port 9 (discard) is closed, so connect fails.
+    base_url = f"http://{LOOPBACK_V4}:9"
+    assert amw_approvals.main(["--base-url", base_url, "list"]) == 1
     out = capsys.readouterr()
-    assert PLACEHOLDER_KEY not in out.out
-    assert PLACEHOLDER_KEY not in out.err
+    assert "gateway unreachable at" in out.err
+    assert base_url in out.err
+    _assert_no_key_leak(key, out.out, out.err)
 
 
 # Fixed documentation-only fixtures. Every host below is reserved and
@@ -404,8 +416,11 @@ def post(self, path: str) -> _StubResponse:
         out = capsys.readouterr()
         assert "Traceback" not in out.err
         assert "Traceback" not in out.out
-        assert "gateway request failed" in out.err
+        assert "gateway unreachable at" in out.err
+        assert base_url in out.err
         assert "ConnectError" in out.err
+        assert "dial boom" not in out.err
+        assert "dial boom" not in out.out
         _assert_no_key_leak(key, out.out, out.err)
 
 
@@ -414,23 +429,75 @@ def test_status_error_hides_key(
 ) -> None:
     key = secrets.token_hex(16)
 
-    class _StatusLeakResponse:
-        status_code = 500
+    def _unauthorized(
+        request: Any,
+    ) -> Any:
+        assert request.headers["X-API-Key"] == key
+        return amw_approvals.httpx.Response(401, json={"detail": "unauthorized"})
+
+    transport = amw_approvals.httpx.MockTransport(_unauthorized)
+    _real_client = amw_approvals.httpx.Client
+
+    def _mock_client(*args: Any, **kwargs: Any) -> Any:
+        kwargs["transport"] = transport
+        return _real_client(*args, **kwargs)
+
+    monkeypatch.setattr(amw_approvals.httpx, "Client", _mock_client)
+    monkeypatch.setenv("VALID_API_KEYS", key)
+    base_url = f"http://{LOOPBACK_V4}:{HTTP_TEST_PORT}"
+    nonce = "test-nonce"
+    for argv in (["list"], ["approve", nonce], ["deny", nonce]):
+        assert amw_approvals.main(["--base-url", base_url, *argv]) == 1
+        out = capsys.readouterr()
+        assert "Traceback" not in out.err
+        assert "Traceback" not in out.out
+        assert "HTTP 401" in out.err
+        assert "gateway request failed: HTTP 401" in out.err
+        assert base_url not in out.err
+        _assert_no_key_leak(key, out.out, out.err)
+
+
+def test_trailing_newline_key_accepted_and_used(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    key = secrets.token_hex(16)
+    monkeypatch.setenv("VALID_API_KEYS", f"{key}\n")
+    assert amw_approvals._read_api_key() == key
+    # RFC 1122 loopback, RFC 6335 fixed port: allowed http loopback URL.
+    client = amw_approvals._client(
+        f"http://{LOOPBACK_V4}:{HTTP_TEST_PORT}", amw_approvals._read_api_key()
+    )
+    try:
+        assert client.headers["X-API-Key"] == key
+    finally:
+        client.close()
+
+
+def test_non_json_response_prints_fixed_message(
+    monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
+) -> None:
+    key = secrets.token_hex(16)
+
+    class _NonJsonResponse:
+        status_code = 200
 
         def raise_for_status(self) -> None:
-            raise amw_approvals.httpx.ConnectError(f"status boom {key} quoted")
+            return None
 
-    class _StatusLeakClient:
+        def json(self) -> dict[str, object]:
+            raise json.JSONDecodeError("Expecting value", "not json", 0)
+
+    class _NonJsonClient:
         def __init__(self, *args: object, **kwargs: object) -> None:
             pass
 
-        def get(self, path: str) -> _StatusLeakResponse:
-            return _StatusLeakResponse()
+        def get(self, path: str) -> _NonJsonResponse:
+            return _NonJsonResponse()
 
-        def post(self, path: str) -> _StatusLeakResponse:
-            return _StatusLeakResponse()
+        def post(self, path: str) -> _NonJsonResponse:
+            return _NonJsonResponse()
 
-    monkeypatch.setattr(amw_approvals.httpx, "Client", _StatusLeakClient)
+    monkeypatch.setattr(amw_approvals.httpx, "Client", _NonJsonClient)
     monkeypatch.setenv("VALID_API_KEYS", key)
     base_url = f"http://{LOOPBACK_V4}:{HTTP_TEST_PORT}"
     nonce = "test-nonce"
@@ -439,6 +506,5 @@ def post(self, path: str) -> _StatusLeakResponse:
         out = capsys.readouterr()
         assert "Traceback" not in out.err
         assert "Traceback" not in out.out
-        assert "gateway request failed" in out.err
-        assert "ConnectError" in out.err
+        assert "gateway returned a non-JSON response" in out.err
         _assert_no_key_leak(key, out.out, out.err)

From 4c487f5d28bb4f4d75ee16d381a646643131da5a Mon Sep 17 00:00:00 2001
From: Christopher <131207452+PetrefiedThunder@users.noreply.github.com>
Date: Sat, 24 Oct 2026 22:59:02 -0700
Subject: [PATCH 3/5] fix(gateway): reject non-object JSON in approvals CLI

Signed-off-by: Christopher <131207452+PetrefiedThunder@users.noreply.github.com>
---
 gateway/scripts/amw_approvals.py        |  27 ++++++-
 gateway/tests/test_amw_approvals_cli.py | 101 ++++++++++++++++++++++++
 2 files changed, 126 insertions(+), 2 deletions(-)

diff --git a/gateway/scripts/amw_approvals.py b/gateway/scripts/amw_approvals.py
index 29992b5e..b6936cf1 100644
--- a/gateway/scripts/amw_approvals.py
+++ b/gateway/scripts/amw_approvals.py
@@ -170,6 +170,16 @@ def _format_time(value: float) -> str:
     )
 
 
+def _is_pending_item(item: object) -> bool:
+    """True when item has the keys cmd_list prints, with a numeric deadline."""
+    if not isinstance(item, dict):
+        return False
+    if "nonce" not in item or "tool" not in item or "principal" not in item:
+        return False
+    deadline = item.get("deadline")
+    return isinstance(deadline, (int, float)) and not isinstance(deadline, bool)
+
+
 def cmd_list(args: argparse.Namespace) -> int:
     try:
         response = _client(args.base_url, args.api_key).get("/v1/approvals/pending")
@@ -199,11 +209,20 @@ def cmd_list(args: argparse.Namespace) -> int:
         )
         return 1
     try:
-        approvals = response.json().get("approvals", [])
+        payload = response.json()
     except ValueError:
         # ValueError covers json.JSONDecodeError from a non-JSON body.
         print("gateway returned a non-JSON response", file=sys.stderr)
         return 1
+    if not isinstance(payload, dict):
+        print("gateway returned an unexpected response", file=sys.stderr)
+        return 1
+    approvals = payload.get("approvals", [])
+    if not isinstance(approvals, list) or any(
+        not _is_pending_item(item) for item in approvals
+    ):
+        print("gateway returned an unexpected response", file=sys.stderr)
+        return 1
     if not approvals:
         print("no pending approvals")
         return 0
@@ -269,11 +288,15 @@ def _decide(args: argparse.Namespace, action: str) -> int:
         )
         return 1
     try:
-        status = response.json().get("status")
+        payload = response.json()
     except ValueError:
         # ValueError covers json.JSONDecodeError from a non-JSON body.
         print("gateway returned a non-JSON response", file=sys.stderr)
         return 1
+    if not isinstance(payload, dict):
+        print("gateway returned an unexpected response", file=sys.stderr)
+        return 1
+    status = payload.get("status")
     print(f"{args.nonce}: {status}")
     return 0
 
diff --git a/gateway/tests/test_amw_approvals_cli.py b/gateway/tests/test_amw_approvals_cli.py
index 62330df0..98f7d2fb 100644
--- a/gateway/tests/test_amw_approvals_cli.py
+++ b/gateway/tests/test_amw_approvals_cli.py
@@ -508,3 +508,104 @@ def post(self, path: str) -> _NonJsonResponse:
         assert "Traceback" not in out.out
         assert "gateway returned a non-JSON response" in out.err
         _assert_no_key_leak(key, out.out, out.err)
+
+
+@pytest.mark.parametrize("kind", ["list", "null", "string"])
+def test_unexpected_json_shape_prints_fixed_message(
+    monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], kind: str
+) -> None:
+    key = secrets.token_hex(16)
+    payload: Any
+    if kind == "list":
+        payload = []
+    elif kind == "null":
+        payload = None
+    else:
+        payload = f"unexpected-{secrets.token_hex(8)}"
+
+    class _UnexpectedResponse:
+        status_code = 200
+
+        def raise_for_status(self) -> None:
+            return None
+
+        def json(self) -> Any:
+            return payload
+
+    class _UnexpectedClient:
+        def __init__(self, *args: object, **kwargs: object) -> None:
+            pass
+
+        def get(self, path: str) -> _UnexpectedResponse:
+            return _UnexpectedResponse()
+
+        def post(self, path: str) -> _UnexpectedResponse:
+            return _UnexpectedResponse()
+
+    monkeypatch.setattr(amw_approvals.httpx, "Client", _UnexpectedClient)
+    monkeypatch.setenv("VALID_API_KEYS", key)
+    # RFC 1122 loopback, RFC 6335 fixed port: allowed http loopback URL.
+    base_url = f"http://{LOOPBACK_V4}:{HTTP_TEST_PORT}"
+    nonce = "test-nonce"
+    for argv in (["list"], ["approve", nonce], ["deny", nonce]):
+        assert amw_approvals.main(["--base-url", base_url, *argv]) == 1
+        out = capsys.readouterr()
+        assert "Traceback" not in out.err
+        assert "Traceback" not in out.out
+        assert "gateway returned an unexpected response" in out.err
+        if isinstance(payload, str):
+            assert payload not in out.out
+            assert payload not in out.err
+        _assert_no_key_leak(key, out.out, out.err)
+
+
+@pytest.mark.parametrize(
+    "approvals",
+    [
+        "not-a-list",
+        None,
+        ["not-a-dict"],
+        [{"nonce": "test-nonce"}],
+        [
+            {
+                "nonce": "test-nonce",
+                "tool": "test-tool",
+                "principal": "test-principal",
+                "deadline": "soon",
+            }
+        ],
+    ],
+)
+def test_malformed_approvals_list_prints_fixed_message(
+    monkeypatch: pytest.MonkeyPatch,
+    capsys: pytest.CaptureFixture[str],
+    approvals: Any,
+) -> None:
+    key = secrets.token_hex(16)
+
+    class _MalformedResponse:
+        status_code = 200
+
+        def raise_for_status(self) -> None:
+            return None
+
+        def json(self) -> Any:
+            return {"approvals": approvals}
+
+    class _MalformedClient:
+        def __init__(self, *args: object, **kwargs: object) -> None:
+            pass
+
+        def get(self, path: str) -> _MalformedResponse:
+            return _MalformedResponse()
+
+    monkeypatch.setattr(amw_approvals.httpx, "Client", _MalformedClient)
+    monkeypatch.setenv("VALID_API_KEYS", key)
+    # RFC 1122 loopback, RFC 6335 fixed port: allowed http loopback URL.
+    base_url = f"http://{LOOPBACK_V4}:{HTTP_TEST_PORT}"
+    assert amw_approvals.main(["--base-url", base_url, "list"]) == 1
+    out = capsys.readouterr()
+    assert "Traceback" not in out.err
+    assert "Traceback" not in out.out
+    assert "gateway returned an unexpected response" in out.err
+    _assert_no_key_leak(key, out.out, out.err)

From 9cfbe7dbf893475a49bf612c848f077c0a6c81f8 Mon Sep 17 00:00:00 2001
From: Christopher <131207452+PetrefiedThunder@users.noreply.github.com>
Date: Sat, 24 Oct 2026 23:08:12 -0700
Subject: [PATCH 4/5] fix(gateway): validate approvals CLI response fields

Signed-off-by: Christopher <131207452+PetrefiedThunder@users.noreply.github.com>
---
 gateway/scripts/amw_approvals.py        |  27 ++++-
 gateway/tests/test_amw_approvals_cli.py | 136 ++++++++++++++++++++++++
 2 files changed, 159 insertions(+), 4 deletions(-)

diff --git a/gateway/scripts/amw_approvals.py b/gateway/scripts/amw_approvals.py
index b6936cf1..c6c42783 100644
--- a/gateway/scripts/amw_approvals.py
+++ b/gateway/scripts/amw_approvals.py
@@ -165,19 +165,27 @@ def _client(base_url: str, api_key: str) -> httpx.Client:
 
 
 def _format_time(value: float) -> str:
+    if isinstance(value, bool):
+        raise TypeError("bool is not a valid deadline")
     return datetime.fromtimestamp(value, tz=timezone.utc).strftime(
         "%Y-%m-%d %H:%M:%S UTC"
     )
 
 
 def _is_pending_item(item: object) -> bool:
-    """True when item has the keys cmd_list prints, with a numeric deadline."""
+    """True when item has the keys cmd_list prints, with a renderable deadline."""
     if not isinstance(item, dict):
         return False
     if "nonce" not in item or "tool" not in item or "principal" not in item:
         return False
     deadline = item.get("deadline")
-    return isinstance(deadline, (int, float)) and not isinstance(deadline, bool)
+    if not isinstance(deadline, (int, float)):
+        return False
+    try:
+        _format_time(deadline)
+    except (ValueError, OverflowError, OSError, TypeError):
+        return False
+    return True
 
 
 def cmd_list(args: argparse.Namespace) -> int:
@@ -217,7 +225,10 @@ def cmd_list(args: argparse.Namespace) -> int:
     if not isinstance(payload, dict):
         print("gateway returned an unexpected response", file=sys.stderr)
         return 1
-    approvals = payload.get("approvals", [])
+    if "approvals" not in payload:
+        print("gateway returned an unexpected response", file=sys.stderr)
+        return 1
+    approvals = payload["approvals"]
     if not isinstance(approvals, list) or any(
         not _is_pending_item(item) for item in approvals
     ):
@@ -227,9 +238,14 @@ def cmd_list(args: argparse.Namespace) -> int:
         print("no pending approvals")
         return 0
     for item in approvals:
+        try:
+            rendered = _format_time(item["deadline"])
+        except (ValueError, OverflowError, OSError, TypeError):
+            print("gateway returned an unexpected response", file=sys.stderr)
+            return 1
         print(
             f"{item['nonce']} tool={item['tool']} principal={item['principal']} "
-            f"deadline={_format_time(item['deadline'])}"
+            f"deadline={rendered}"
         )
     return 0
 
@@ -297,6 +313,9 @@ def _decide(args: argparse.Namespace, action: str) -> int:
         print("gateway returned an unexpected response", file=sys.stderr)
         return 1
     status = payload.get("status")
+    if not isinstance(status, str) or not status:
+        print("gateway returned an unexpected response", file=sys.stderr)
+        return 1
     print(f"{args.nonce}: {status}")
     return 0
 
diff --git a/gateway/tests/test_amw_approvals_cli.py b/gateway/tests/test_amw_approvals_cli.py
index 98f7d2fb..f17868f9 100644
--- a/gateway/tests/test_amw_approvals_cli.py
+++ b/gateway/tests/test_amw_approvals_cli.py
@@ -609,3 +609,139 @@ def get(self, path: str) -> _MalformedResponse:
     assert "Traceback" not in out.out
     assert "gateway returned an unexpected response" in out.err
     _assert_no_key_leak(key, out.out, out.err)
+
+
+@pytest.mark.parametrize(
+    "deadline",
+    [1e300, float("nan"), float("inf"), float("-inf"), True, False],
+)
+def test_unrenderable_deadline_prints_fixed_message(
+    monkeypatch: pytest.MonkeyPatch,
+    capsys: pytest.CaptureFixture[str],
+    deadline: Any,
+) -> None:
+    key = secrets.token_hex(16)
+    nonce = f"test-nonce-{secrets.token_hex(4)}"
+
+    class _DeadlineResponse:
+        status_code = 200
+
+        def raise_for_status(self) -> None:
+            return None
+
+        def json(self) -> Any:
+            return {
+                "approvals": [
+                    {
+                        "nonce": nonce,
+                        "tool": "test-tool",
+                        "principal": "test-principal",
+                        "deadline": deadline,
+                    }
+                ]
+            }
+
+    class _DeadlineClient:
+        def __init__(self, *args: object, **kwargs: object) -> None:
+            pass
+
+        def get(self, path: str) -> _DeadlineResponse:
+            return _DeadlineResponse()
+
+    monkeypatch.setattr(amw_approvals.httpx, "Client", _DeadlineClient)
+    monkeypatch.setenv("VALID_API_KEYS", key)
+    # RFC 1122 loopback, RFC 6335 fixed port: allowed http loopback URL.
+    base_url = f"http://{LOOPBACK_V4}:{HTTP_TEST_PORT}"
+    assert amw_approvals.main(["--base-url", base_url, "list"]) == 1
+    out = capsys.readouterr()
+    assert "Traceback" not in out.err
+    assert "Traceback" not in out.out
+    assert "gateway returned an unexpected response" in out.err
+    assert out.out == ""
+    assert nonce not in out.err
+    _assert_no_key_leak(key, out.out, out.err)
+
+
+def test_missing_approvals_key_prints_fixed_message(
+    monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
+) -> None:
+    key = secrets.token_hex(16)
+
+    class _MissingKeyResponse:
+        status_code = 200
+
+        def raise_for_status(self) -> None:
+            return None
+
+        def json(self) -> Any:
+            return {}
+
+    class _MissingKeyClient:
+        def __init__(self, *args: object, **kwargs: object) -> None:
+            pass
+
+        def get(self, path: str) -> _MissingKeyResponse:
+            return _MissingKeyResponse()
+
+    monkeypatch.setattr(amw_approvals.httpx, "Client", _MissingKeyClient)
+    monkeypatch.setenv("VALID_API_KEYS", key)
+    # RFC 1122 loopback, RFC 6335 fixed port: allowed http loopback URL.
+    base_url = f"http://{LOOPBACK_V4}:{HTTP_TEST_PORT}"
+    assert amw_approvals.main(["--base-url", base_url, "list"]) == 1
+    out = capsys.readouterr()
+    assert "Traceback" not in out.err
+    assert "Traceback" not in out.out
+    assert "gateway returned an unexpected response" in out.err
+    assert "no pending approvals" not in out.out
+    assert out.out == ""
+    _assert_no_key_leak(key, out.out, out.err)
+
+
+@pytest.mark.parametrize(
+    "payload",
+    [
+        {},
+        {"status": ""},
+        {"status": None},
+        {"status": 123},
+        {"status": True},
+        {"status": []},
+        {"status": {}},
+    ],
+)
+def test_decide_invalid_status_prints_fixed_message(
+    monkeypatch: pytest.MonkeyPatch,
+    capsys: pytest.CaptureFixture[str],
+    payload: Any,
+) -> None:
+    key = secrets.token_hex(16)
+
+    class _StatusResponse:
+        status_code = 200
+
+        def raise_for_status(self) -> None:
+            return None
+
+        def json(self) -> Any:
+            return payload
+
+    class _StatusClient:
+        def __init__(self, *args: object, **kwargs: object) -> None:
+            pass
+
+        def post(self, path: str) -> _StatusResponse:
+            return _StatusResponse()
+
+    monkeypatch.setattr(amw_approvals.httpx, "Client", _StatusClient)
+    monkeypatch.setenv("VALID_API_KEYS", key)
+    # RFC 1122 loopback, RFC 6335 fixed port: allowed http loopback URL.
+    base_url = f"http://{LOOPBACK_V4}:{HTTP_TEST_PORT}"
+    nonce = "test-nonce"
+    for argv in (["approve", nonce], ["deny", nonce]):
+        assert amw_approvals.main(["--base-url", base_url, *argv]) == 1
+        out = capsys.readouterr()
+        assert "Traceback" not in out.err
+        assert "Traceback" not in out.out
+        assert "gateway returned an unexpected response" in out.err
+        assert out.out == ""
+        _assert_no_key_leak(key, out.out, out.err)

From 16af061d33b1ae0dbbac7c0a976a7f7019965871 Mon Sep 17 00:00:00 2001
From: Christopher <131207452+PetrefiedThunder@users.noreply.github.com>
Date: Sat, 24 Oct 2026 23:26:40 -0700
Subject: [PATCH 5/5] test(gateway): mock connect failure in approvals CLI leak
 test

Signed-off-by: Christopher <131207452+PetrefiedThunder@users.noreply.github.com>
---
 gateway/tests/test_amw_approvals_cli.py | 16 ++++++++++++++--
 1 file changed, 14 insertions(+), 2 deletions(-)

diff --git a/gateway/tests/test_amw_approvals_cli.py b/gateway/tests/test_amw_approvals_cli.py
index f17868f9..457c0387 100644
--- a/gateway/tests/test_amw_approvals_cli.py
+++ b/gateway/tests/test_amw_approvals_cli.py
@@ -122,9 +122,21 @@ def test_unreachable_gateway_error_hides_key(
     monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
 ) -> None:
     key = secrets.token_hex(16)
+
+    def _unreachable(request: Any) -> Any:
+        raise amw_approvals.httpx.ConnectError(f"dial boom {key} quoted")
+
+    transport = amw_approvals.httpx.MockTransport(_unreachable)
+    _real_client = amw_approvals.httpx.Client
+
+    def _mock_client(*args: Any, **kwargs: Any) -> Any:
+        kwargs["transport"] = transport
+        return _real_client(*args, **kwargs)
+
+    monkeypatch.setattr(amw_approvals.httpx, "Client", _mock_client)
     monkeypatch.setenv("VALID_API_KEYS", key)
-    # RFC 1122 loopback; port 9 (discard) is closed, so connect fails.
-    base_url = f"http://{LOOPBACK_V4}:9"
+    # RFC 2606 example.com: allowed https URL, never dialed (MockTransport).
+    base_url = f"https://{EXAMPLE_COM}"
     assert amw_approvals.main(["--base-url", base_url, "list"]) == 1
     out = capsys.readouterr()
     assert "gateway unreachable at" in out.err