134 lines
4.3 KiB
Python
134 lines
4.3 KiB
Python
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"
|