feat(frontends,security,api,ui): a frontend can hold a token scope of its own

tests/test_scopes.py

@@ -0,0 +1,133 @@
+import tempfile
+from pathlib import Path
+
+import httpx
+import pytest
+from fastapi import FastAPI, Request
+
+from beaver_gateway.app import AgentRegistry, McpRegistry
+from beaver_gateway.frontends.api.frontend import build_app as build_api
+from beaver_gateway.frontends.base import Frontend, GatewayRuntime
+from beaver_gateway.frontends.bearer import require_token
+from beaver_gateway.security.auth import BUILTIN_SCOPES, TokenStore, scopes_with
+from beaver_gateway.storage import Database
+
+ROOT = {"Authorization": "Bearer root-key"}
+VOICE = {"Authorization": "Bearer voice-key"}
+OPS = {"Authorization": "Bearer ops-key"}
+
+
+class Speaker(Frontend):
+    """A frontend a setup declares, gated by a scope of its own."""
+
+    name = "voice"
+    path = "/voice"
+    scope = "voice"
+
+    def configure(self, runtime: GatewayRuntime) -> None:
+        self.runtime = runtime
+
+    def app(self) -> FastAPI:
+        app = FastAPI()
+
+        @app.post("/")
+        async def ask(request: Request) -> dict[str, str]:
+            return {"by": await require_token(request, self.runtime, scope=self.scope)}
+
+        return app
+
+
+@pytest.fixture
+async def runtime() -> GatewayRuntime:
+    root = Path(tempfile.mkdtemp(prefix="beaver-scopes-"))
+    db = Database(f"sqlite:///{root / 's.db'}")
+    await db.create_all()
+    speaker = Speaker()
+    rt = GatewayRuntime(
+        agents=AgentRegistry([]),
+        mcps=McpRegistry([]),
+        backends={},
+        token_store=TokenStore(
+            db,
+            bootstrap={"root": "root-key", "voice": "voice-key", "ops": "ops-key"},
+            bootstrap_scopes={"voice": "voice", "ops": "api"},
+        ),
+        db=db,
+        frontends=(speaker,),
+        scopes=scopes_with(fe.scope for fe in (speaker,)),
+    )
+    speaker.configure(rt)
+    yield rt
+    await db.dispose()
+
+
+def api(runtime: GatewayRuntime) -> httpx.AsyncClient:
+    return httpx.AsyncClient(
+        transport=httpx.ASGITransport(app=build_api(runtime)), base_url="http://t"
+    )
+
+
+def frontend(runtime: GatewayRuntime) -> httpx.AsyncClient:
+    fe = runtime.frontends[0]
+    return httpx.AsyncClient(
+        transport=httpx.ASGITransport(app=fe.app()), base_url="http://t"
+    )
+
+
+def test_a_frontend_scope_joins_the_builtin_vocabulary() -> None:
+    assert "voice" not in BUILTIN_SCOPES
+    assert scopes_with(["voice", None, "api"]) == BUILTIN_SCOPES | {"voice"}
+
+
+async def test_the_scope_of_a_frontend_is_listed_with_its_name(
+    runtime: GatewayRuntime,
+) -> None:
+    async with api(runtime) as client:
+        listed = await client.get("/scopes", headers=ROOT)
+    scopes = listed.json()["scopes"]
+    assert {"scope": "voice", "frontend": "voice"} in scopes
+    assert {"scope": "api", "frontend": None} in scopes
+
+
+async def test_tokens_mint_for_a_declared_scope_and_not_for_an_unknown_one(
+    runtime: GatewayRuntime,
+) -> None:
+    async with api(runtime) as client:
+        minted = await client.post(
+            "/tokens", json={"name": "esp32-hall", "scope": "voice"}, headers=ROOT
+        )
+        unknown = await client.post(
+            "/tokens", json={"name": "nope", "scope": "kitchen"}, headers=ROOT
+        )
+    assert minted.status_code == 201
+    assert minted.json()["token"]["scope"] == "voice"
+    assert unknown.status_code == 400
+    assert "kitchen" in unknown.json()["error"]
+
+
+async def test_the_route_opens_for_its_own_scope_only(runtime: GatewayRuntime) -> None:
+    async with frontend(runtime) as client:
+        own = await client.post("/", headers=VOICE)
+        other = await client.post("/", headers=OPS)
+        wildcard = await client.post("/", headers=ROOT)
+        anonymous = await client.post("/")
+    assert own.status_code == 200
+    assert own.json()["by"] == "voice"
+    assert other.status_code == 403
+    assert wildcard.status_code == 200
+    assert anonymous.status_code == 401
+
+
+async def test_a_minted_token_reaches_the_frontend_it_was_minted_for(
+    runtime: GatewayRuntime,
+) -> None:
+    async with api(runtime) as client:
+        plaintext = (
+            await client.post(
+                "/tokens", json={"name": "esp32-hall", "scope": "voice"}, headers=ROOT
+            )
+        ).json()["plaintext"]
+    async with frontend(runtime) as client:
+        opened = await client.post("/", headers={"X-Api-Key": plaintext})
+    assert opened.status_code == 200
+    assert opened.json()["by"] == "esp32-hall"