54 lines
1.6 KiB
Python
54 lines
1.6 KiB
Python
"""Shared bearer-token verification for HTTP frontends.
|
|
|
|
Extracted from ``AnthropicMessagesFrontend`` so the markdown frontend
|
|
(and any future bearer-protected frontend) can reuse one canonical
|
|
verifier instead of copy-pasting the header-parsing dance.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import TYPE_CHECKING
|
|
|
|
from fastapi import HTTPException, status
|
|
|
|
if TYPE_CHECKING:
|
|
from fastapi import Request
|
|
|
|
from beaver_gateway.frontends.base import GatewayRuntime
|
|
|
|
|
|
__all__ = ["require_token"]
|
|
|
|
|
|
async def require_token(
|
|
request: Request, runtime: GatewayRuntime, *, scope: str
|
|
) -> str:
|
|
"""Verify the request's bearer + scope, return the token's audit name.
|
|
|
|
Accepts both ``X-Api-Key: <token>`` (Anthropic SDK / LibreChat) and
|
|
``Authorization: Bearer <token>`` (curl, Cursor). 401 on missing /
|
|
unknown token; 403 on a known token whose scope doesn't cover
|
|
``scope``. Bootstrap tokens implicitly carry ``"*"`` and pass every
|
|
scope check.
|
|
"""
|
|
api_key = request.headers.get("x-api-key")
|
|
identity = (
|
|
await runtime.token_store.verify(api_key)
|
|
if api_key
|
|
else await runtime.token_store.verify_bearer(
|
|
request.headers.get("authorization")
|
|
)
|
|
)
|
|
if identity is None:
|
|
raise HTTPException(
|
|
status.HTTP_401_UNAUTHORIZED,
|
|
"invalid or missing bearer token",
|
|
headers={"WWW-Authenticate": "Bearer"},
|
|
)
|
|
if not identity.allows(scope):
|
|
raise HTTPException(
|
|
status.HTTP_403_FORBIDDEN,
|
|
f"token scope {identity.scope!r} does not cover {scope!r}",
|
|
)
|
|
return identity.name
|