feat(auth): per-agent registry-identity auth for the A2A bus read proxy (#138 Phase 1)

README.md

@@ -305,6 +305,10 @@ Create shared file spaces for agents, groups, and departments. The design team s
 ### Authentication
 Password-protected dashboard with persistent sessions. Per-agent API keys. Exempt paths for cluster workers and health checks.
 
+**Agents authenticate with their own identity, not the owner password.** Each registered agent has an Ed25519 registry identity (canonical id + signed JWT). The owner password is human-only and is never handed to an agent. An agent calls scoped endpoints by presenting `Authorization: Bearer <registry-jwt>`; the route verifies the signature against the registry public key and checks the agent is active and holds the required scope grant. Today this covers the registry feed endpoints (scope `registry_feeds_read`) and the read-only A2A bus proxy `/api/a2a/bus/channels` + `/api/a2a/bus/messages` (scope `a2a_receive`). The Bearer allowlist is exact: a registry JWT authenticates only those agent paths, never an arbitrary route.
+
+Onboarding an internal driver agent: an admin mints its identity once with `taosctl agents mint --handle @taOS-dev --slug taos-dev --scopes a2a_send,a2a_receive` (or `taosctl agents seed-internal` to mint the four built-in driver agents idempotently). Minting prints the registry JWT; store it on the agent host in a gitignored per-host file (for example `~/.config/taos/agent-token`) and have the agent send it as `Authorization: Bearer <jwt>`. Re-running mint/seed for an existing handle reuses the same canonical id and re-asserts the grants, so it is safe to run again.
+
 ### Model Conversion
 Convert models between formats (GGUF→RKLLM, HF→GGUF, GGUF→MLX). Capability-gated, "Convert for NPU" button appears when an x86 worker joins the cluster.
 

docs/design/external-agent-onboarding.md

@@ -59,6 +59,29 @@ taOSmd's `has_grant` (post-merge fix `fcc0fd7`) **fails closed on a non-numeric
 - Anti-spoof: the enforced `project_id`/`user_id` come from the verified token claim, never the request body.
 - Reattach is append-only, so the governance audit log (#730) shows every shelf change.
 
+## Phase 1 shipped: bus read with a registry JWT
+
+The first enforcement slice is live on the controller. An agent reads the
+read-only A2A bus proxy with its OWN registry identity, never the owner
+password:
+
+- Endpoints: `GET /api/agents/registry/grants` + `/revoked` (scope
+  `registry_feeds_read`) and `GET /api/a2a/bus/channels` + `/api/a2a/bus/messages`
+  (scope `a2a_receive`). These are the only paths that accept a registry JWT in
+  place of the admin session; the Bearer allowlist is exact (no skeleton key).
+- The agent presents `Authorization: Bearer <registry-jwt>`. The route verifies
+  the Ed25519 signature against the registry public key, requires the agent be
+  `active`, and requires an active (non-expired) grant for the scope. Malformed
+  or wrong-key tokens get 401; valid-but-unauthorized get 403. Fail closed.
+- Token storage: the operator stores the minted JWT on the agent host in a
+  gitignored per-host file (for example `~/.config/taos/agent-token`); the agent
+  reads it from there. Tokens are never written into the repo.
+- Minting the four internal driver agents (@taOS-dev, @taOS-website-dev,
+  @taOSmd-dev, @Hermes), idempotently by handle:
+  `taosctl agents seed-internal` (or `taosctl agents mint --handle @X --slug x
+  --scopes a2a_send,a2a_receive`). Re-running reuses the existing canonical id
+  and re-asserts the grants.
+
 ## Open questions
 
 - Default scope set for a coding agent: `memory_read` + `memory_write` + `a2a_send` + `a2a_receive`; `files_*` / `tools_execute` gated tighter (probably off by default).

tests/test_a2a_bus_agent_auth.py

@@ -0,0 +1,239 @@
+"""Agent-token authentication for the read-only A2A bus proxy.
+
+An agent reads /api/a2a/bus/* with its OWN Ed25519 registry JWT (scope
+a2a_receive), never the owner session/password.  These tests cover the full
+edge-case matrix Jay flagged plus the admin/local-token regressions and the
+skeleton-key guard (the agent token must not authenticate any other route).
+"""
+from __future__ import annotations
+
+import tempfile
+from pathlib import Path
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+import pytest_asyncio
+from httpx import ASGITransport, AsyncClient
+
+from tinyagentos.agent_registry_store import (
+    load_or_create_signing_keypair,
+    mint_registry_token,
+)
+
+_BUS_PATCH = "tinyagentos.routes.a2a_bus.httpx.AsyncClient"
+
+
+def _mock_bus_client(json_payload: dict):
+    """A mock httpx.AsyncClient whose async context manager yields a client whose
+    .get().json() returns *json_payload* (so no real network call is made)."""
+    mock_resp = MagicMock()
+    mock_resp.raise_for_status = MagicMock()
+    mock_resp.json.return_value = json_payload
+
+    mock_client = AsyncMock()
+    mock_client.get.return_value = mock_resp
+
+    mock_ctx = AsyncMock()
+    mock_ctx.__aenter__ = AsyncMock(return_value=mock_client)
+    mock_ctx.__aexit__ = AsyncMock(return_value=False)
+    return mock_ctx
+
+
+@pytest_asyncio.fixture
+async def bus_client(app, tmp_data_dir):
+    """Admin client with the registry + grants stores initialised.
+
+    Exposes ._app so tests can register agents / mint tokens directly against the
+    stores and drive bare (cookieless) requests with a Bearer header.
+    """
+    for attr in ("agent_registry", "agent_grants", "metrics"):
+        store = getattr(app.state, attr)
+        if store._db is None:
+            await store.init()
+
+    app.state.auth.setup_user("admin", "Test Admin", "", "testpass")
+    record = app.state.auth.find_user("admin")
+    uid = record["id"] if record else ""
+    token = app.state.auth.create_session(user_id=uid, long_lived=True)
+    app.state._startup_complete = True
+
+    transport = ASGITransport(app=app)
+    async with AsyncClient(
+        transport=transport,
+        base_url="http://test",
+        cookies={"taos_session": token},
+    ) as c:
+        c._app = app
+        c._test_admin_uid = uid
+        yield c
+
+    for attr in ("agent_registry", "agent_grants", "metrics"):
+        store = getattr(app.state, attr)
+        if store._db is not None:
+            await store.close()
+
+
+async def _make_agent_token(app, *, scopes=("a2a_receive",), origin="taos-deployed"):
+    """Register an agent, add its grants, and return (canonical_id, signed JWT)."""
+    registry = app.state.agent_registry
+    grants = app.state.agent_grants
+    priv, _pub = app.state.agent_registry_keypair
+
+    rec = await registry.register(
+        framework="taosmd",
+        display_name="Bus Reader",
+        origin=origin,
+        handle="@bus-reader",
+    )
+    cid = rec["canonical_id"]
+    for scope in scopes:
+        await grants.add_grant(cid, scope)
+    token = mint_registry_token(cid, priv, user_id="u", framework="taosmd")
+    return cid, token
+
+
+def _bare(app):
+    return AsyncClient(transport=ASGITransport(app=app), base_url="http://test")
+
+
+@pytest.mark.asyncio
+class TestBusAgentAuth:
+
+    async def test_agent_with_receive_scope_reads_channels(self, bus_client):
+        _cid, token = await _make_agent_token(bus_client._app)
+        with patch(_BUS_PATCH, return_value=_mock_bus_client({"channels": []})):
+            async with _bare(bus_client._app) as bare:
+                resp = await bare.get(
+                    "/api/a2a/bus/channels",
+                    headers={"Authorization": f"Bearer {token}"},
+                )
+        assert resp.status_code == 200
+        assert resp.json()["available"] is True
+
+    async def test_agent_with_receive_scope_reads_messages(self, bus_client):
+        _cid, token = await _make_agent_token(bus_client._app)
+        with patch(_BUS_PATCH, return_value=_mock_bus_client({"messages": []})):
+            async with _bare(bus_client._app) as bare:
+                resp = await bare.get(
+                    "/api/a2a/bus/messages",
+                    params={"channel": "general"},
+                    headers={"Authorization": f"Bearer {token}"},
+                )
+        assert resp.status_code == 200
+        assert resp.json()["available"] is True
+
+    async def test_agent_without_receive_scope_gets_403(self, bus_client):
+        _cid, token = await _make_agent_token(bus_client._app, scopes=("a2a_send",))
+        async with _bare(bus_client._app) as bare:
+            resp = await bare.get(
+                "/api/a2a/bus/channels",
+                headers={"Authorization": f"Bearer {token}"},
+            )
+        assert resp.status_code == 403
+
+    async def test_suspended_agent_gets_403(self, bus_client):
+        cid, token = await _make_agent_token(bus_client._app)
+        await bus_client._app.state.agent_registry.set_status(cid, "suspended")
+        async with _bare(bus_client._app) as bare:
+            resp = await bare.get(
+                "/api/a2a/bus/channels",
+                headers={"Authorization": f"Bearer {token}"},
+            )
+        assert resp.status_code == 403
+
+    async def test_revoked_agent_gets_403(self, bus_client):
+        cid, token = await _make_agent_token(bus_client._app)
+        await bus_client._app.state.agent_registry.revoke(cid)
+        async with _bare(bus_client._app) as bare:
+            resp = await bare.get(
+                "/api/a2a/bus/messages",
+                params={"channel": "general"},
+                headers={"Authorization": f"Bearer {token}"},
+            )
+        assert resp.status_code == 403
+
+    async def test_malformed_token_gets_401(self, bus_client):
+        async with _bare(bus_client._app) as bare:
+            resp = await bare.get(
+                "/api/a2a/bus/channels",
+                headers={"Authorization": "Bearer not-a-valid-token"},
+            )
+        assert resp.status_code == 401
+
+    async def test_token_signed_by_different_key_gets_401(self, bus_client):
+        # Register the agent so the sub exists, but sign with a foreign key.
+        registry = bus_client._app.state.agent_registry
+        grants = bus_client._app.state.agent_grants
+        rec = await registry.register(framework="taosmd", display_name="Foreign", handle="@foreign")
+        cid = rec["canonical_id"]
+        await grants.add_grant(cid, "a2a_receive")
+        with tempfile.TemporaryDirectory() as d:
+            foreign_priv, _foreign_pub = load_or_create_signing_keypair(Path(d))
+        bad_token = mint_registry_token(cid, foreign_priv, user_id="u", framework="taosmd")
+        async with _bare(bus_client._app) as bare:
+            resp = await bare.get(
+                "/api/a2a/bus/channels",
+                headers={"Authorization": f"Bearer {bad_token}"},
+            )
+        assert resp.status_code == 401
+
+    async def test_valid_sig_unknown_sub_gets_403(self, bus_client):
+        # Properly signed by the app key but the sub is not in the registry.
+        priv, _pub = bus_client._app.state.agent_registry_keypair
+        token = mint_registry_token("ghost-20260101-000000", priv, user_id="u", framework="taosmd")
+        async with _bare(bus_client._app) as bare:
+            resp = await bare.get(
+                "/api/a2a/bus/channels",
+                headers={"Authorization": f"Bearer {token}"},
+            )
+        assert resp.status_code == 403
+
+    async def test_expired_grant_gets_403(self, bus_client):
+        from datetime import datetime, timezone, timedelta
+
+        registry = bus_client._app.state.agent_registry
+        grants = bus_client._app.state.agent_grants
+        rec = await registry.register(framework="taosmd", display_name="Expired", handle="@expired")
+        cid = rec["canonical_id"]
+        past = (datetime.now(timezone.utc) - timedelta(seconds=1)).isoformat()
+        await grants.add_grant(cid, "a2a_receive", expires_at=past)
+        priv, _pub = bus_client._app.state.agent_registry_keypair
+        token = mint_registry_token(cid, priv, user_id="u", framework="taosmd")
+        async with _bare(bus_client._app) as bare:
+            resp = await bare.get(
+                "/api/a2a/bus/channels",
+                headers={"Authorization": f"Bearer {token}"},
+            )
+        assert resp.status_code == 403
+
+    async def test_admin_session_still_reads_bus(self, bus_client):
+        """Regression: an admin cookie session reads the bus unchanged."""
+        with patch(_BUS_PATCH, return_value=_mock_bus_client({"channels": []})):
+            resp = await bus_client.get("/api/a2a/bus/channels")
+        assert resp.status_code == 200
+        assert resp.json()["available"] is True
+
+    async def test_local_token_reads_bus_as_admin(self, bus_client):
+        """Regression: a Bearer local token authenticates as admin and is NOT
+        parsed as a registry JWT."""
+        local_token = bus_client._app.state.auth.get_local_token()
+        with patch(_BUS_PATCH, return_value=_mock_bus_client({"channels": []})):
+            async with _bare(bus_client._app) as bare:
+                resp = await bare.get(
+                    "/api/a2a/bus/channels",
+                    headers={"Authorization": f"Bearer {local_token}"},
+                )
+        assert resp.status_code == 200
+        assert resp.json()["available"] is True
+
+    async def test_skeleton_key_guard_agent_token_rejected_off_allowlist(self, bus_client):
+        """A valid agent JWT (with a2a_receive) must NOT authenticate a
+        non-allowlisted protected route -- the token is not a skeleton key."""
+        _cid, token = await _make_agent_token(bus_client._app)
+        async with _bare(bus_client._app) as bare:
+            resp = await bare.get(
+                "/api/agents/registry",
+                headers={"Authorization": f"Bearer {token}"},
+            )
+        # Falls through middleware to the session gate: API request -> 401.
+        assert resp.status_code == 401