fix(keys): block restricted API keys from minting or managing other keys

app/routes/keys.py

@@ -22,11 +22,33 @@ class CreateKey(BaseModel):
     name: str
 
 
+def require_unrestricted(kc: KeyContext) -> None:
+    """Key management is reserved for unrestricted keys.
+
+    A key that carries any restriction (`model_allowlist` or
+    `budget_limit_cents`) must not be able to mint, list, or revoke other
+    keys — otherwise it could create a sibling with no restrictions and
+    trivially bypass its own allowlist/budget. Unrestricted keys already
+    hold the maximum privilege this single-workspace edition exposes
+    (same trust level as PUT /v1/providers/*), so denying restricted keys
+    here grants nothing to anyone; it only closes the escalation path.
+    """
+    if kc.model_allowlist is not None or kc.budget_limit_cents is not None:
+        raise HTTPException(
+            status_code=403,
+            detail=(
+                "Restricted API keys cannot manage keys. "
+                "Use an unrestricted key."
+            ),
+        )
+
+
 @router.get("")
 async def list_keys(
-    _kc: KeyContext = Depends(get_key_context),
+    kc: KeyContext = Depends(get_key_context),
     db: AsyncSession = Depends(get_db),
 ) -> dict:
+    require_unrestricted(kc)
     rows = (
         await db.execute(
             select(ApiKey).where(ApiKey.is_deleted == 0).order_by(ApiKey.created_at)
@@ -39,6 +61,8 @@ async def list_keys(
                 "name": r.name,
                 "key_prefix": r.key_prefix,
                 "is_active": r.is_active,
+                "model_allowlist": r.model_allowlist,
+                "budget_limit_cents": r.budget_limit_cents,
                 "last_used_at": iso_utc(r.last_used_at),
                 "revoked_at": iso_utc(r.revoked_at),
                 "created_at": iso_utc(r.created_at),
@@ -54,6 +78,7 @@ async def create_key(
     kc: KeyContext = Depends(get_key_context),
     db: AsyncSession = Depends(get_db),
 ) -> dict:
+    require_unrestricted(kc)
     full_key, key_hash, key_prefix = generate_api_key()
     row = ApiKey(
         workspace_id=kc.workspace_id,
@@ -76,9 +101,10 @@ async def create_key(
 @router.delete("/{key_id}", status_code=204)
 async def revoke_key(
     key_id: str,
-    _kc: KeyContext = Depends(get_key_context),
+    kc: KeyContext = Depends(get_key_context),
     db: AsyncSession = Depends(get_db),
 ) -> Response:
+    require_unrestricted(kc)
     row = (
         await db.execute(
             select(ApiKey).where(ApiKey.id == key_id, ApiKey.is_deleted == 0)

app/routes/providers.py

@@ -46,6 +46,7 @@
 from app.config import get_settings
 from app.deps import get_db, get_key_context
 from app.router_cache import usable_providers_from_db
+from app.routes.keys import require_unrestricted
 from packages.auth.encryption import encrypt_credential
 from packages.auth.types import KeyContext
 from packages.db.models.provider_key import ProviderKey
@@ -143,6 +144,10 @@ async def set_provider_key(
     _kc: KeyContext = Depends(get_key_context),
     db: AsyncSession = Depends(get_db),
 ) -> dict:
+    # A restricted key (model allowlist or budget cap) must not be able to
+    # overwrite operator credentials or delete providers — that would let it
+    # escape its restrictions and redirect the whole workspace's traffic.
+    require_unrestricted(_kc)
     if not body.api_key.strip():
         raise HTTPException(status_code=422, detail="api_key cannot be empty")
 
@@ -200,6 +205,7 @@ async def delete_provider_key(
     _kc: KeyContext = Depends(get_key_context),
     db: AsyncSession = Depends(get_db),
 ) -> Response:
+    require_unrestricted(_kc)
     """Hard-delete the DB row for this provider. After this, runtime
     resolution falls back to the env value (if `.env` has one) or
     treats the provider as unconfigured.

app/routes/quality.py

@@ -33,6 +33,7 @@
 )
 from app.cache_invalidation_bus import broadcast_metrics_cache_invalidation
 from app.deps import get_db, get_key_context
+from app.routes.keys import require_unrestricted
 from app.quality_scores import (
     fetch_aa_index,
     load_overrides,
@@ -391,6 +392,7 @@ async def upsert_override(
     kc: KeyContext = Depends(get_key_context),
     db: AsyncSession = Depends(get_db),
 ) -> dict:
+    require_unrestricted(kc)
     """Create or update an override. Idempotent on (workspace, model_id).
 
     `session.merge()` is select-then-write under the hood, so two
@@ -463,6 +465,7 @@ async def delete_override(
     kc: KeyContext = Depends(get_key_context),
     db: AsyncSession = Depends(get_db),
 ) -> None:
+    require_unrestricted(kc)
     """Remove an override. Returns 204 even if the override didn't exist —
     the operator's intent ('no override on this model') is satisfied either
     way, no need to surface the distinction."""

app/routes/routing.py

@@ -9,6 +9,7 @@
 
 from app import cache_invalidation_bus
 from app.deps import get_db, get_key_context
+from app.routes.keys import require_unrestricted
 from app.seed import DEFAULT_WORKSPACE_ID
 from packages.auth.types import KeyContext
 from packages.db.models.routing_config import RoutingConfig
@@ -57,6 +58,7 @@ async def update_routing(
     _kc: KeyContext = Depends(get_key_context),
     db: AsyncSession = Depends(get_db),
 ) -> dict:
+    require_unrestricted(_kc)
     row = (
         await db.execute(
             select(RoutingConfig).where(

tests/integration/test_keys_authz.py

@@ -0,0 +1,205 @@
+"""Key-management authorization tests.
+
+A restricted key (model_allowlist or budget_limit_cents set) must never be
+able to mint, list, or revoke API keys — otherwise it could mint an
+unrestricted sibling and bypass its own restrictions entirely.
+See issue: restricted-key privilege escalation via POST /v1/keys.
+"""
+
+import pytest
+
+
+@pytest.fixture
+async def seeded_keys(db_session):
+    """Seed the workspace root key plus one restricted and one budgeted key.
+
+    Returns (root_full_key, restricted_full_key, budgeted_full_key).
+    """
+    from app.seed import seed_initial_state
+    from packages.auth.hashing import generate_api_key
+    from packages.db.models.api_key import ApiKey
+
+    seed = await seed_initial_state(db_session)
+    assert seed.api_key is not None
+
+    def _make(**kwargs) -> str:
+        full_key, key_hash, key_prefix = generate_api_key()
+        row = ApiKey(
+            workspace_id="default",
+            name=kwargs.pop("name", "test"),
+            key_hash=key_hash,
+            key_prefix=key_prefix,
+            **kwargs,
+        )
+        db_session.add(row)
+        return full_key
+
+    # flush once so all rows land before any request reads them
+    restricted = _make(name="restricted", model_allowlist=["gpt-4o-mini"])
+    budgeted = _make(name="budgeted", budget_limit_cents=500)
+    await db_session.commit()
+    return seed.api_key, restricted, budgeted
+
+
+@pytest.fixture
+async def keys_app(db_session, monkeypatch):
+    """FastAPI app with auth middleware and only the /v1/keys routes mounted."""
+    monkeypatch.setenv("DATABASE_URL", str(db_session.bind.url))
+    from fastapi import FastAPI
+
+    from app.middleware.auth import AuthMiddleware
+    from packages.db import session as session_mod
+
+    class _PassthroughFactory:
+        async def __aenter__(self):
+            return db_session
+
+        async def __aexit__(self, *exc):
+            return False  # propagate, don't close — fixture owns the session
+
+    monkeypatch.setattr(session_mod, "_session_factory", lambda: _PassthroughFactory())
+
+    from app.routes.keys import router as keys_router
+
+    app = FastAPI()
+    app.add_middleware(AuthMiddleware)
+    app.include_router(keys_router)
+    return app
+
+
+async def _client(app):
+    from httpx import ASGITransport, AsyncClient
+
+    return AsyncClient(transport=ASGITransport(app=app), base_url="http://t")
+
+
+@pytest.mark.parametrize("which", [1, 2], ids=["allowlist-restricted", "budget-restricted"])
+async def test_restricted_key_cannot_create_keys(keys_app, seeded_keys, db_session, which):
+    keys, restricted, budgeted = seeded_keys
+    caller = (restricted, budgeted)[which - 1]
+    async with await _client(keys_app) as c:
+        r = await c.post(
+            "/v1/keys",
+            json={"name": "escalated"},
+            headers={"Authorization": f"Bearer {caller}"},
+        )
+    assert r.status_code == 403
+    # The escalation must not have persisted anything.
+    from sqlalchemy import func, select
+
+    from packages.db.models.api_key import ApiKey
+
+    count = (
+        await db_session.execute(select(func.count()).select_from(ApiKey))
+    ).scalar_one()
+    assert count == 3  # root + restricted + budgeted, nothing new
+
+
+@pytest.mark.parametrize("which", [1, 2], ids=["allowlist-restricted", "budget-restricted"])
+async def test_restricted_key_cannot_list_keys(keys_app, seeded_keys, which):
+    keys, restricted, budgeted = seeded_keys
+    caller = (restricted, budgeted)[which - 1]
+    async with await _client(keys_app) as c:
+        r = await c.get("/v1/keys", headers={"Authorization": f"Bearer {caller}"})
+    assert r.status_code == 403
+
+
+async def test_restricted_key_cannot_revoke_keys(keys_app, seeded_keys, db_session):
+    _, restricted, _budgeted = seeded_keys
+    from sqlalchemy import select
+
+    from packages.db.models.api_key import ApiKey
+
+    rows = (await db_session.execute(select(ApiKey))).scalars().all()
+    target_id = next(r.id for r in rows if r.name == "default")
+    async with await _client(keys_app) as c:
+        r = await c.delete(
+            f"/v1/keys/{target_id}",
+            headers={"Authorization": f"Bearer {restricted}"},
+        )
+    assert r.status_code == 403
+    target = next(r for r in rows if r.name == "default")
+    assert target.is_active  # untouched
+
+
+async def test_unrestricted_key_retains_full_management(keys_app, seeded_keys):
+    root, _restricted, _budgeted = seeded_keys
+    h = {"Authorization": f"Bearer {root}"}
+    async with await _client(keys_app) as c:
+        listed = await c.get("/v1/keys", headers=h)
+        assert listed.status_code == 200
+
+        created = await c.post("/v1/keys", json={"name": "child"}, headers=h)
+        assert created.status_code == 201
+        child_id = created.json()["id"]
+
+        revoked = await c.delete(f"/v1/keys/{child_id}", headers=h)
+        assert revoked.status_code == 204
+
+
+@pytest.fixture
+async def mgmt_app(db_session, monkeypatch):
+    """App mounting the privilege-sensitive write routes alongside keys."""
+    monkeypatch.setenv("DATABASE_URL", str(db_session.bind.url))
+    from fastapi import FastAPI
+
+    from app.middleware.auth import AuthMiddleware
+    from app.routes.keys import router as keys_router
+    from app.routes.providers import router as providers_router
+    from app.routes.quality import router as quality_router
+    from app.routes.routing import router as routing_router
+    from packages.db import session as session_mod
+
+    class _PassthroughFactory:
+        async def __aenter__(self):
+            return db_session
+
+        async def __aexit__(self, *exc):
+            return False
+
+    monkeypatch.setattr(session_mod, "_session_factory", lambda: _PassthroughFactory())
+
+    app = FastAPI()
+    app.add_middleware(AuthMiddleware)
+    app.include_router(keys_router)
+    app.include_router(providers_router)
+    app.include_router(routing_router)
+    app.include_router(quality_router)
+    return app
+
+
+@pytest.mark.parametrize("which", [1, 2], ids=["allowlist-restricted", "budget-restricted"])
+async def test_restricted_key_cannot_write_provider_credentials(
+    mgmt_app, seeded_keys, which
+):
+    _keys, restricted, budgeted = seeded_keys
+    caller = (restricted, budgeted)[which - 1]
+    h = {"Authorization": f"Bearer {caller}"}
+    async with await _client(mgmt_app) as c:
+        r = await c.put(
+            "/v1/providers/openai",
+            json={"api_key": "sk-test-123"},
+            headers=h,
+        )
+    assert r.status_code == 403
+
+
+async def test_restricted_key_cannot_write_routing(mgmt_app, seeded_keys):
+    _keys, restricted, _budgeted = seeded_keys
+    h = {"Authorization": f"Bearer {restricted}"}
+    async with await _client(mgmt_app) as c:
+        r = await c.put("/v1/routing", json={"strategy": "balanced"}, headers=h)
+    assert r.status_code == 403
+
+
+async def test_unrestricted_key_can_write_provider_and_routing(mgmt_app, seeded_keys):
+    root, _restricted, _budgeted = seeded_keys
+    h = {"Authorization": f"Bearer {root}"}
+    async with await _client(mgmt_app) as c:
+        # Gate only — we assert the restricted-key block doesn't apply to root.
+        p = await c.put(
+            "/v1/providers/openai", json={"api_key": "sk-test-123"}, headers=h
+        )
+        assert p.status_code != 403
+        r = await c.put("/v1/routing", json={"strategy": "balanced"}, headers=h)
+        assert r.status_code != 403