Existence-hiding 404 across ALL agent-registry routes (issue #2106, reported by hognek)

changelog.d/2356-registry-existence-hiding.md

@@ -0,0 +1,7 @@
+### Security
+
+- All owner-gated agent-registry routes are now existence-hiding: a caller who
+  does not own an agent gets the same 404 as a nonexistent id, on the
+  scope-request create/approve/deny routes and on registry PATCH, revoke,
+  rotate-tokens, and org update. Previously a 403-vs-404 difference disclosed
+  whether an agent id existed (issue #2106, reported by hognek) (#2356).

docs/agent-coordination.md

@@ -396,6 +396,15 @@ that SAME canonical_id instead:
 - `POST /api/agents/registry/{canonical_id}/scope-requests/{req_id}/deny`:
   owner/admin only.
 
+All owner-gated registry routes are existence-hiding (#2106): an authenticated
+caller who is not the owner gets the same 404 body as a nonexistent
+`canonical_id`, on the scope-request create/approve/deny routes above and on
+registry PATCH, DELETE (revoke), rotate-tokens, and `PUT /api/agents/{id}/org`.
+Agents must not treat a 404 from these routes as proof an id does not exist,
+and must not expect a 403 to distinguish "exists, not yours". Admin-only
+lifecycle routes (approve/reject/suspend/reactivate) still 403 non-admins
+before any lookup, which discloses nothing.
+
 Requested scopes are validated against the same closed `VALID_SCOPES` vocabulary
 as the consent flow. `project_tasks` and the canvas scopes still require an
 explicit `project_id`; `decisions_read` / `decisions_write` (and the other global

tests/test_agent_registry.py

@@ -829,3 +829,88 @@ def test_allowed_scopes_includes_project_doc_review():
 def test_allowed_scopes_includes_observatory_control():
     """observatory_control must be in the mint allowlist so internal agents can be granted it."""
     assert "observatory_control" in _ALLOWED_SCOPES
+
+
+# ---------------------------------------------------------------------------
+# Existence-hiding: non-owner vs non-existent must be byte-identical
+# (same contract the scope-request routes assert in
+# tests/test_agent_scope_requests.py; GET already had it, these four
+# write routes gained it in the same pass)
+# ---------------------------------------------------------------------------
+
+@pytest.mark.asyncio
+class TestRegistryWriteExistenceHiding:
+    async def _non_owner_client(self, app):
+        """Session client for a real NON-admin user who owns nothing."""
+        code = app.state.auth.add_user_invite("mallory", "admin")
+        app.state.auth.complete_invite("mallory", code, "Mallory", "", "malpass123")
+        record = app.state.auth.find_user("mallory")
+        session = app.state.auth.create_session(user_id=record["id"], long_lived=True)
+        return AsyncClient(
+            transport=ASGITransport(app=app),
+            base_url="http://test",
+            cookies={"taos_session": session},
+        )
+
+    async def _register(self, registry_client):
+        resp = await registry_client.post(
+            "/api/agents/registry/register",
+            json={"framework": "openclaw", "display_name": "Hidden Agent"},
+        )
+        assert resp.status_code == 200
+        return resp.json()["canonical_id"]
+
+    async def test_patch_non_owner_and_nonexistent_identical(self, app, registry_client):
+        cid = await self._register(registry_client)
+        async with await self._non_owner_client(app) as mallory:
+            resp_owned = await mallory.patch(
+                f"/api/agents/registry/{cid}", json={"display_name": "Stolen"}
+            )
+            resp_missing = await mallory.patch(
+                "/api/agents/registry/does-not-exist", json={"display_name": "Stolen"}
+            )
+        assert resp_owned.status_code == resp_missing.status_code == 404
+        assert resp_owned.content == resp_missing.content
+        # And the record was not modified.
+        check = await registry_client.get(f"/api/agents/registry/{cid}")
+        assert check.json()["display_name"] == "Hidden Agent"
+
+    async def test_delete_non_owner_and_nonexistent_identical(self, app, registry_client):
+        cid = await self._register(registry_client)
+        async with await self._non_owner_client(app) as mallory:
+            resp_owned = await mallory.delete(f"/api/agents/registry/{cid}")
+            resp_missing = await mallory.delete("/api/agents/registry/does-not-exist")
+        assert resp_owned.status_code == resp_missing.status_code == 404
+        assert resp_owned.content == resp_missing.content
+        # And the record was not revoked.
+        check = await registry_client.get(f"/api/agents/registry/{cid}")
+        assert check.status_code == 200
+        assert not check.json().get("revoked_at")
+
+    async def test_rotate_tokens_non_owner_and_nonexistent_identical(
+        self, app, registry_client
+    ):
+        cid = await self._register(registry_client)
+        async with await self._non_owner_client(app) as mallory:
+            resp_owned = await mallory.post(
+                f"/api/agents/registry/{cid}/rotate-tokens"
+            )
+            resp_missing = await mallory.post(
+                "/api/agents/registry/does-not-exist/rotate-tokens"
+            )
+        assert resp_owned.status_code == resp_missing.status_code == 404
+        assert resp_owned.content == resp_missing.content
+
+    async def test_org_put_non_owner_and_nonexistent_identical(
+        self, app, registry_client
+    ):
+        cid = await self._register(registry_client)
+        async with await self._non_owner_client(app) as mallory:
+            resp_owned = await mallory.put(
+                f"/api/agents/{cid}/org", json={"role": "usurper"}
+            )
+            resp_missing = await mallory.put(
+                "/api/agents/does-not-exist/org", json={"role": "usurper"}
+            )
+        assert resp_owned.status_code == resp_missing.status_code == 404
+        assert resp_owned.content == resp_missing.content

tests/test_agent_scope_requests.py

@@ -157,7 +157,7 @@ async def test_agent_cannot_request_for_another_identity(client, monkeypatch, tm
                 headers={"Authorization": f"Bearer {token_a}"},
                 json={"requested_scopes": ["a2a_send"]},
             )
-        assert resp.status_code == 403, resp.text
+        assert resp.status_code == 404, resp.text
         assert await env.scope_store.count_pending_for(cid_b) == 0
     finally:
         await env.close()
@@ -294,7 +294,7 @@ async def test_non_owner_cannot_approve(client, monkeypatch, tmp_path):
                 f"/api/agents/registry/{cid}/scope-requests/{rec['id']}/approve",
                 json={"granted_scopes": ["memory_read"]},
             )
-        assert resp.status_code == 403, resp.text
+        assert resp.status_code == 404, resp.text
         assert await env.grants.list_grants(cid) == []
     finally:
         await env.close()
@@ -459,7 +459,7 @@ async def test_create_authorizes_before_scope_vocab(client, monkeypatch, tmp_pat
                 json={"requested_scopes": ["not_a_real_scope"]},
             )
         # Authz runs first -> 403, NOT a 400 vocab error confirming the bad scope.
-        assert resp.status_code == 403, resp.text
+        assert resp.status_code == 404, resp.text
         assert "not_a_real_scope" not in resp.text
     finally:
         await env.close()
@@ -598,3 +598,160 @@ def test_every_project_bound_scope_is_a_valid_scope():
     from tinyagentos.routes.agent_auth_requests import VALID_SCOPES, _PROJECT_SCOPES
 
     assert _PROJECT_SCOPES <= set(VALID_SCOPES)
+
+
+# ---------------------------------------------------------------------------
+# Existence-hiding: non-owner vs non-existent must be byte-identical
+# ---------------------------------------------------------------------------
+
+@pytest.mark.asyncio
+async def test_create_scope_request_non_owner_and_nonexistent_identical(
+    client, monkeypatch, tmp_path
+):
+    """An authenticated non-owner and a nonexistent canonical_id must produce
+    byte-identical responses on create_scope_request (status + body)."""
+    app = client._transport.app
+    code = app.state.auth.add_user_invite("carol", "admin")
+    app.state.auth.complete_invite("carol", code, "Carol", "", "carpass123")
+    carol = app.state.auth.find_user("carol")
+    carol_session = app.state.auth.create_session(user_id=carol["id"], long_lived=True)
+
+    env = await _wire(client, monkeypatch, tmp_path)
+    try:
+        cid = await _register_active(env)  # owned by admin, not carol
+
+        async with AsyncClient(
+            transport=ASGITransport(app=app),
+            base_url="http://test",
+            cookies={"taos_session": carol_session},
+        ) as carol_client:
+            resp_owner = await carol_client.post(
+                f"/api/agents/registry/{cid}/scope-requests",
+                json={"requested_scopes": ["memory_read"]},
+            )
+            # Same caller for the nonexistent probe: an admin would 404 on a
+            # missing id too, but the contract under test is what ONE
+            # unprivileged caller can distinguish.
+            resp_nonexistent = await carol_client.post(
+                "/api/agents/registry/does-not-exist/scope-requests",
+                json={"requested_scopes": ["memory_read"]},
+            )
+
+        assert resp_owner.status_code == resp_nonexistent.status_code == 404
+        assert resp_owner.content == resp_nonexistent.content
+    finally:
+        await env.close()
+
+
+@pytest.mark.asyncio
+async def test_approve_scope_request_non_owner_and_nonexistent_identical(
+    client, monkeypatch, tmp_path
+):
+    """An authenticated non-owner and a nonexistent canonical_id must produce
+    byte-identical responses on approve_scope_request (status + body)."""
+    app = client._transport.app
+    code = app.state.auth.add_user_invite("carol", "admin")
+    app.state.auth.complete_invite("carol", code, "Carol", "", "carpass123")
+    carol = app.state.auth.find_user("carol")
+    carol_session = app.state.auth.create_session(user_id=carol["id"], long_lived=True)
+
+    env = await _wire(client, monkeypatch, tmp_path)
+    try:
+        cid = await _register_active(env)  # owned by admin, not carol
+        rec = await env.scope_store.create(
+            canonical_id=cid, requested_scopes=["memory_read"]
+        )
+
+        async with AsyncClient(
+            transport=ASGITransport(app=app),
+            base_url="http://test",
+            cookies={"taos_session": carol_session},
+        ) as carol_client:
+            resp_owner = await carol_client.post(
+                f"/api/agents/registry/{cid}/scope-requests/{rec['id']}/approve",
+                json={"granted_scopes": ["memory_read"]},
+            )
+            resp_nonexistent = await carol_client.post(
+                "/api/agents/registry/does-not-exist/scope-requests/does-not-exist/approve",
+                json={"granted_scopes": ["memory_read"]},
+            )
+
+        assert resp_owner.status_code == resp_nonexistent.status_code == 404
+        assert resp_owner.content == resp_nonexistent.content
+    finally:
+        await env.close()
+
+
+@pytest.mark.asyncio
+async def test_deny_scope_request_non_owner_and_nonexistent_identical(
+    client, monkeypatch, tmp_path
+):
+    """An authenticated non-owner and a nonexistent canonical_id must produce
+    byte-identical responses on deny_scope_request (status + body)."""
+    app = client._transport.app
+    code = app.state.auth.add_user_invite("carol", "admin")
+    app.state.auth.complete_invite("carol", code, "Carol", "", "carpass123")
+    carol = app.state.auth.find_user("carol")
+    carol_session = app.state.auth.create_session(user_id=carol["id"], long_lived=True)
+
+    env = await _wire(client, monkeypatch, tmp_path)
+    try:
+        cid = await _register_active(env)  # owned by admin, not carol
+        rec = await env.scope_store.create(
+            canonical_id=cid, requested_scopes=["memory_read"]
+        )
+
+        async with AsyncClient(
+            transport=ASGITransport(app=app),
+            base_url="http://test",
+            cookies={"taos_session": carol_session},
+        ) as carol_client:
+            resp_owner = await carol_client.post(
+                f"/api/agents/registry/{cid}/scope-requests/{rec['id']}/deny",
+            )
+            resp_nonexistent = await carol_client.post(
+                "/api/agents/registry/does-not-exist/scope-requests/does-not-exist/deny",
+            )
+
+        assert resp_owner.status_code == resp_nonexistent.status_code == 404
+        assert resp_owner.content == resp_nonexistent.content
+    finally:
+        await env.close()
+
+
+@pytest.mark.asyncio
+async def test_create_scope_request_inactive_token_no_existence_oracle(
+    client, monkeypatch, tmp_path
+):
+    """A suspended agent's (validly signed) token must get the SAME response
+    for an existing target as for a nonexistent one. Before the fix,
+    check_agent_identity's 403 surfaced only when the target existed (an
+    unknown target 404s first), disclosing existence through the
+    credential-error path."""
+    env = await _wire(client, monkeypatch, tmp_path)
+    try:
+        cid_target = await _register_active(env, handle="@target", display="target")
+        cid_b = await _register_active(env, handle="@suspended", display="suspended")
+        token_b = env.agent_token(cid_b)
+        await env.registry.set_status(cid_b, "suspended", actor="test")
+
+        app = client._transport.app
+        async with AsyncClient(
+            transport=ASGITransport(app=app), base_url="http://test"
+        ) as bare:
+            resp_existing = await bare.post(
+                f"/api/agents/registry/{cid_target}/scope-requests",
+                headers={"Authorization": f"Bearer {token_b}"},
+                json={"requested_scopes": ["a2a_send"]},
+            )
+            resp_missing = await bare.post(
+                "/api/agents/registry/does-not-exist/scope-requests",
+                headers={"Authorization": f"Bearer {token_b}"},
+                json={"requested_scopes": ["a2a_send"]},
+            )
+
+        assert resp_existing.status_code == resp_missing.status_code == 404
+        assert resp_existing.content == resp_missing.content
+        assert await env.scope_store.count_pending_for(cid_target) == 0
+    finally:
+        await env.close()

tests/test_registry_governance_lifecycle.py

@@ -846,12 +846,12 @@ async def test_patch_by_non_owner_member_returns_403(
             user_id="other-user-uid",
         )
         cid = rec["canonical_id"]
-        # member tries to patch another user's entry → 403
+        # member tries to patch another user's entry → 404 (existence-hiding)
         resp = await gov_member_client.patch(
             f"/api/agents/registry/{cid}",
             json={"display_name": "Hijacked"},
         )
-        assert resp.status_code == 403
+        assert resp.status_code == 404
 
     async def test_patch_empty_body_is_noop(self, gov_client, tmp_data_dir):
         client, _ = gov_client

tests/test_routes_agent_org.py

@@ -258,4 +258,4 @@ async def test_member_cannot_update_others_entry(self, org_client, app):
             )
         finally:
             await member_client.aclose()
-        assert resp.status_code == 403
+        assert resp.status_code == 404

tests/test_token_rotation.py

@@ -241,7 +241,7 @@ async def test_nonadmin_owner_can_rotate_own_identity(self, agent_app, app):
 
     @pytest.mark.asyncio
     async def test_nonadmin_cannot_rotate_others_agent(self, agent_app, app):
-        """A non-admin, non-owner rotating someone ELSE'S agent → 403."""
+        """A non-admin, non-owner rotating someone ELSE'S agent → 404 (existence-hiding)."""
         # First user owns an agent.
         owner_client, owner_uid = _make_nonadmin_client(
             app, app.state.auth, username="owner2", full_name="Owner Two",
@@ -260,22 +260,22 @@ async def test_nonadmin_cannot_rotate_others_agent(self, agent_app, app):
             resp = await intruder_client.post(
                 f"/api/agents/registry/{cid}/rotate-tokens"
             )
-            assert resp.status_code == 403
+            assert resp.status_code == 404
 
     @pytest.mark.asyncio
     async def test_empty_userid_agent_is_admin_only(self, agent_app, app):
         """An agent_registry row with user_id='' can ONLY be rotated by admin.
 
-        Non-admin sessions get 403 because require_owner_or_admin compares
-        the session's user_id against an empty string (owner match fails)
-        and the session is not admin.
+        Non-admin sessions get 404 (existence-hiding): the owner match
+        against an empty string fails, the session is not admin, and the
+        route answers exactly as it would for an unknown id.
         """
         # Register an agent with the default user_id="" (admin-only).
         cid, _token = await _register_and_mint(app, user_id="admin")
         r = await app.state.agent_registry.get(cid)
         assert r["user_id"] == ""
 
-        # A non-admin session trying to rotate it must get 403.
+        # A non-admin session trying to rotate it must get the not-found 404.
         nonadmin_client, _uid = _make_nonadmin_client(
             app, app.state.auth, username="randouser", full_name="Rando",
             password="password123",
@@ -284,7 +284,7 @@ async def test_empty_userid_agent_is_admin_only(self, agent_app, app):
             resp = await nonadmin_client.post(
                 f"/api/agents/registry/{cid}/rotate-tokens"
             )
-            assert resp.status_code == 403
+            assert resp.status_code == 404
 
     @pytest.mark.asyncio
     async def test_rotate_nonexistent_returns_404(self, agent_app):