fix-forward #2633: the dead-credential 404 fires on existing off-allowlist routes and the garbage-credential control is inert (tsk-sonaie)

changelog.d/tsk-hbzm7l-fix-route-enumeration.md

@@ -0,0 +1,2 @@
+### Fixed
+- Auth middleware now returns 404 directly for authenticated callers on paths no route serves, instead of 401. This fixes a bug where a wrong URL was indistinguishable from dead credentials. The auth middleware runs before routing, so previously routing failures were reported as auth failures. The fix consults the router and, only when nothing matches, returns 404 from the middleware itself without calling routing, while unauthenticated callers still get 401 for anti-enumeration (issue #tsk-hbzm7l).

changelog.d/tsk-iqk2bn-dead-credential-404.md

@@ -0,0 +1,2 @@
+### Fixed
+- Auth middleware now checks registry record status and token rotation cutoff before returning 404 for a path no route serves, so revoked or rotated registry JWTs receive 401 instead of being misreported as a wrong URL.

changelog.d/tsk-sonaie-404-only-for-unrouted-paths.md

@@ -0,0 +1,2 @@
+### Fixed
+- The auth middleware's wrong-URL 404 is now limited to paths the router cannot serve. A registry JWT presented against a route that exists but is deliberately off the agent-token allowlist -- `GET /api/agents/registry`, or the scope-request approve/deny routes -- again receives 401 from the session gate instead of being told the URL does not exist. The route handler is still never reached in either case.

changelog.d/tsk-vylg2y-registry-jwt-unknown-route-404.md

@@ -0,0 +1,3 @@
+### Fixed
+
+- Auth middleware now returns 404 (not 401) for a valid registry JWT presented against a path that no route serves. This makes a wrong URL distinguishable from dead credentials while keeping the agent-token allowlist closed and preserving the anonymous 401. A route that exists but is off the allowlist is not a wrong URL, so it keeps its 401/403 from the session gate.

docs/agent-coordination.md

@@ -322,6 +322,25 @@ A registered external agent authenticates with its registry JWT
 (`Authorization: Bearer`) and reaches exactly the routes its granted SCOPES
 allow, nothing else: the middleware allowlist is a closed set, no skeleton key.
 
+**What a refused request looks like.** Off the allowlist, the request never
+reaches a route handler, and the status code says WHICH thing was wrong:
+
+- **The URL does not exist** (no route in the app matches the path, whatever
+  the method) and the token is a live registry JWT -> **404 `{"error": "Not
+  Found"}`**, straight from the middleware. Routing is never invoked, so this
+  is not a way to reach an unlisted route -- it only tells a correctly
+  credentialled agent that it typed the wrong path.
+- **The URL exists but the token is not authorised for it** -- e.g.
+  `GET /api/agents/registry`, or `POST .../scope-requests/{id}/(approve|deny)`,
+  which are deliberately owner/admin session-only -- -> **401 `{"error":
+  "Authentication required"}`** from the session gate. A correct URL is never
+  reported as missing.
+- **The credential is dead** (revoked, or rotated so its `iat` predates the
+  identity's `token_min_iat`) -> **401**, on any path. A dead credential is
+  never dressed up as a wrong URL.
+- **No credential at all** -> **401** on every path, existing or not, so an
+  anonymous caller cannot enumerate routes by status code.
+
 A SEPARATE credential class exists for the Agent-as-a-Model surface:
 `GET /v1/models` and `POST /v1/chat/completions` are reachable without a
 session using a CONSENT KEY (`Authorization: Bearer sk-taosagent-...`, minted

tests/test_auth_middleware.py

@@ -1,11 +1,15 @@
 """Unit tests for auth_middleware allow/deny logic."""
 from __future__ import annotations
 
+import time
 from unittest.mock import AsyncMock, MagicMock
 
 import pytest
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+from cryptography.hazmat.primitives.serialization import Encoding, NoEncryption, PrivateFormat, PublicFormat
 from fastapi.responses import JSONResponse
 from starlette.responses import RedirectResponse
+from starlette.routing import Route
 
 from tinyagentos.auth_middleware import (
     AuthMiddleware,
@@ -703,3 +707,288 @@ def test_allowlist_does_not_cover_unrelated_paths(self):
         assert _is_device_bearer_path("POST", "/api/decisions/a/b/answer") is False
         assert _is_device_bearer_path("GET", "/api/settings") is False
         assert _is_device_bearer_path("POST", "/api/projects") is False
+
+
+def _registry_keypair() -> tuple[bytes, bytes]:
+    """Return (private_pem, public_pem) for a fresh Ed25519 keypair."""
+    private = Ed25519PrivateKey.generate()
+    private_pem = private.private_bytes(
+        encoding=Encoding.PEM,
+        format=PrivateFormat.PKCS8,
+        encryption_algorithm=NoEncryption(),
+    )
+    public_pem = private.public_key().public_bytes(
+        encoding=Encoding.PEM,
+        format=PublicFormat.SubjectPublicKeyInfo,
+    )
+    return private_pem, public_pem
+
+
+def _signed_registry_token(private_pem: bytes) -> str:
+    """Mint a registry JWT signed by *private_pem*."""
+    from tinyagentos.agent_registry_store import mint_registry_token
+    return mint_registry_token(
+        canonical_id="test-agent",
+        private_key_pem=private_pem,
+        user_id="",
+        framework="test",
+        project_id=None,
+    )
+
+
+async def _never_called(_request):  # pragma: no cover - routing is never reached
+    return JSONResponse({})
+
+
+def _router_routes() -> list[Route]:
+    """A stand-in for the app's real route table.
+
+    The middleware consults the router before answering 404, so these tests
+    must expose one: a static GET route off the agent-token allowlist, and a
+    POST route with path parameters (scope-request approve).  Neither is
+    reachable with an agent token, so both must answer 401 -- not 404 -- while
+    a path absent from this list is a genuinely wrong URL.
+    """
+    return [
+        Route("/api/system", _never_called, methods=["GET"]),
+        Route(
+            "/api/agents/registry/{cid}/scope-requests/{rid}/approve",
+            _never_called,
+            methods=["POST"],
+        ),
+    ]
+
+
+class TestRegistryJwtUnknownRouteDispatch:
+    """Valid cred on a wrong URL, absent cred, garbage cred, dead creds, and
+    the off-allowlist controls: a route that exists keeps its 401 (tsk-vylg2y,
+    tsk-iqk2bn, tsk-sonaie)."""
+
+    UNKNOWN_PATH = "/api/nonexistent/unknown-route"
+
+    def _app_state(self, *, public_pem: bytes | None = None) -> MagicMock:
+        state = MagicMock()
+        if public_pem is not None:
+            state.agent_registry_keypair = (b"private", public_pem)
+        else:
+            state.agent_registry_keypair = None
+        return state
+
+    @pytest.mark.asyncio
+    async def test_valid_registry_jwt_unknown_path_returns_404(self):
+        """RED: a real credential on a path no route serves must return 404
+        from the middleware directly -- routing must NOT be reached."""
+        private_pem, public_pem = _registry_keypair()
+        token = _signed_registry_token(private_pem)
+
+        middleware = AuthMiddleware(app=MagicMock())
+        req = _request(
+            path=self.UNKNOWN_PATH,
+            headers={"authorization": f"Bearer {token}"},
+        )
+        req.app.routes = _router_routes()
+        state = self._app_state(public_pem=public_pem)
+        state.agent_registry = MagicMock()
+        state.agent_registry.get = AsyncMock(return_value={"status": "active"})
+        req.app.state = state
+        req.app.state.auth = _default_auth_mgr()
+        call_next = AsyncMock(return_value=JSONResponse({"ok": True}))
+
+        resp = await middleware.dispatch(req, call_next)
+
+        assert resp.status_code == 404
+        assert resp.body == b'{"error":"Not Found"}'
+        call_next.assert_not_awaited()
+
+    @pytest.mark.asyncio
+    async def test_no_credential_unknown_path_returns_401(self):
+        """Control B: no credential on the same unknown path still returns
+        401.  A blanket 404 from the middleware would fail this test."""
+        middleware = AuthMiddleware(app=MagicMock())
+        req = _request(path=self.UNKNOWN_PATH)
+        req.app.routes = _router_routes()
+        req.app.state = self._app_state()
+        req.app.state.auth = _default_auth_mgr()
+        call_next = AsyncMock()
+
+        resp = await middleware.dispatch(req, call_next)
+
+        assert resp.status_code == 401
+        assert resp.body == b'{"error":"Authentication required"}'
+        call_next.assert_not_awaited()
+
+    @pytest.mark.asyncio
+    async def test_garbage_bearer_unknown_path_returns_401(self):
+        """Control C: a forged/garbage bearer on the same unknown path still
+        returns 401.  Verifies the fix keys off a real credential, not merely
+        the presence of an Authorization header."""
+        _priv, _pub = _registry_keypair()
+        middleware = AuthMiddleware(app=MagicMock())
+        req = _request(
+            path=self.UNKNOWN_PATH,
+            headers={"authorization": "Bearer garbage-not-a-jwt"},
+        )
+        req.app.routes = _router_routes()
+        # The registry must be a working AsyncMock like every other arm here.
+        # Left as a bare MagicMock, `await registry.get(...)` raises TypeError
+        # and the liveness check returns False for a reason that has nothing to
+        # do with the signature -- which made this control blind to a missing
+        # signature check.
+        _state = self._app_state(public_pem=_pub)
+        _state.agent_registry = MagicMock()
+        _state.agent_registry.get = AsyncMock(return_value={"status": "active"})
+        req.app.state = _state
+        req.app.state.auth = _default_auth_mgr()
+        call_next = AsyncMock()
+
+        resp = await middleware.dispatch(req, call_next)
+
+        assert resp.status_code == 401
+        assert resp.body == b'{"error":"Authentication required"}'
+        call_next.assert_not_awaited()
+
+    @pytest.mark.asyncio
+    async def test_revoked_registry_jwt_unknown_path_returns_401(self):
+        """A revoked registry JWT on an unlisted route must return 401, not
+        404.  The auth middleware must distinguish dead credentials from wrong
+        URLs."""
+        private_pem, public_pem = _registry_keypair()
+        token = _signed_registry_token(private_pem)
+
+        middleware = AuthMiddleware(app=MagicMock())
+        req = _request(
+            path=self.UNKNOWN_PATH,
+            headers={"authorization": f"Bearer {token}"},
+        )
+        req.app.routes = _router_routes()
+        state = self._app_state(public_pem=public_pem)
+        state.agent_registry = MagicMock()
+        state.agent_registry.get = AsyncMock(
+            return_value={"status": "revoked"}
+        )
+        req.app.state = state
+        req.app.state.auth = _default_auth_mgr()
+        call_next = AsyncMock()
+
+        resp = await middleware.dispatch(req, call_next)
+
+        assert resp.status_code == 401
+        assert resp.body == b'{"error":"Authentication required"}'
+        call_next.assert_not_awaited()
+
+    @pytest.mark.asyncio
+    async def test_rotated_registry_jwt_unknown_path_returns_401(self):
+        """A rotated registry JWT (iat < token_min_iat) on an unlisted route
+        must return 401, not 404.  The auth middleware must distinguish dead
+        credentials from wrong URLs."""
+        private_pem, public_pem = _registry_keypair()
+        token = _signed_registry_token(private_pem)
+
+        middleware = AuthMiddleware(app=MagicMock())
+        req = _request(
+            path=self.UNKNOWN_PATH,
+            headers={"authorization": f"Bearer {token}"},
+        )
+        req.app.routes = _router_routes()
+        state = self._app_state(public_pem=public_pem)
+        state.agent_registry = MagicMock()
+        state.agent_registry.get = AsyncMock(
+            return_value={
+                "status": "active",
+                "token_min_iat": int(time.time()) + 3600,
+            }
+        )
+        req.app.state = state
+        req.app.state.auth = _default_auth_mgr()
+        call_next = AsyncMock()
+
+        resp = await middleware.dispatch(req, call_next)
+
+        assert resp.status_code == 401
+        assert resp.body == b'{"error":"Authentication required"}'
+        call_next.assert_not_awaited()
+
+    @pytest.mark.asyncio
+    async def test_skeleton_key_registry_jwt_existing_non_allowlisted_route(self):
+        """Skeleton-key control: a valid registry JWT on a route that exists
+        but is NOT in the agent-token allowlist must NOT be routed -- the
+        handler must never run.  The URL is correct and the credential is
+        simply not authorised for it, so the answer is the session gate's 401,
+        never the wrong-URL 404."""
+        private_pem, public_pem = _registry_keypair()
+        token = _signed_registry_token(private_pem)
+
+        middleware = AuthMiddleware(app=MagicMock())
+        req = _request(
+            path="/api/system",  # exists but is NOT an _AGENT_TOKEN_PATHS entry
+            headers={"authorization": f"Bearer {token}"},
+        )
+        req.app.routes = _router_routes()
+        state = self._app_state(public_pem=public_pem)
+        state.agent_registry = MagicMock()
+        state.agent_registry.get = AsyncMock(return_value={"status": "active"})
+        req.app.state = state
+        req.app.state.auth = _default_auth_mgr()
+        call_next = AsyncMock(return_value=JSONResponse({"system": "ok"}))
+
+        resp = await middleware.dispatch(req, call_next)
+
+        assert resp.status_code == 401
+        assert resp.body == b'{"error":"Authentication required"}'
+        call_next.assert_not_awaited()
+
+    @pytest.mark.asyncio
+    async def test_registry_jwt_off_allowlist_param_route_returns_401(self):
+        """The scope-request approve route exists with path parameters and is
+        deliberately owner/admin session-only.  A live registry JWT there must
+        get the gate's 401, not a 404 claiming the URL is wrong."""
+        private_pem, public_pem = _registry_keypair()
+        token = _signed_registry_token(private_pem)
+
+        middleware = AuthMiddleware(app=MagicMock())
+        req = _request(
+            method="POST",
+            path="/api/agents/registry/agent-1/scope-requests/req-1/approve",
+            headers={"authorization": f"Bearer {token}"},
+        )
+        req.app.routes = _router_routes()
+        state = self._app_state(public_pem=public_pem)
+        state.agent_registry = MagicMock()
+        state.agent_registry.get = AsyncMock(return_value={"status": "active"})
+        req.app.state = state
+        req.app.state.auth = _default_auth_mgr()
+        call_next = AsyncMock(return_value=JSONResponse({"ok": True}))
+
+        resp = await middleware.dispatch(req, call_next)
+
+        assert resp.status_code == 401
+        assert resp.body == b'{"error":"Authentication required"}'
+        call_next.assert_not_awaited()
+
+    @pytest.mark.asyncio
+    async def test_registry_jwt_wrong_method_on_existing_path_returns_401(self):
+        """A verb the route does not accept is still a real URL: the path
+        resolves, so the caller gets the gate's 401 rather than being told the
+        path does not exist."""
+        private_pem, public_pem = _registry_keypair()
+        token = _signed_registry_token(private_pem)
+
+        middleware = AuthMiddleware(app=MagicMock())
+        req = _request(
+            method="DELETE",  # /api/system is GET-only
+            path="/api/system",
+            headers={"authorization": f"Bearer {token}"},
+        )
+        req.app.routes = _router_routes()
+        state = self._app_state(public_pem=public_pem)
+        state.agent_registry = MagicMock()
+        state.agent_registry.get = AsyncMock(return_value={"status": "active"})
+        req.app.state = state
+        req.app.state.auth = _default_auth_mgr()
+        call_next = AsyncMock(return_value=JSONResponse({"ok": True}))
+
+        resp = await middleware.dispatch(req, call_next)
+
+        assert resp.status_code == 401
+        assert resp.body == b'{"error":"Authentication required"}'
+        call_next.assert_not_awaited()