wuyang
71 lines
2.4 KiB
Python
71 lines
2.4 KiB
Python
from __future__ import annotations
|
|
|
|
import hmac
|
|
from dataclasses import dataclass
|
|
from typing import Annotated
|
|
|
|
import jwt
|
|
from fastapi import Header, HTTPException, status
|
|
|
|
from agent_platform.config import get_settings
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class UserIdentity:
|
|
user_id: str
|
|
email: str
|
|
name: str
|
|
role: str
|
|
|
|
|
|
def verify_service_bearer(authorization: str | None, expected: str) -> None:
|
|
if not expected:
|
|
raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="Service secret is not configured")
|
|
scheme, _, token = (authorization or "").partition(" ")
|
|
if scheme.lower() != "bearer" or not hmac.compare_digest(token, expected):
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid service credentials")
|
|
|
|
|
|
def decode_openwebui_identity(token: str | None, secret: str) -> UserIdentity:
|
|
if not secret:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
|
|
detail="Identity verification is unavailable",
|
|
)
|
|
if not token:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Missing signed user identity")
|
|
try:
|
|
claims = jwt.decode(
|
|
token,
|
|
secret,
|
|
algorithms=["HS256"],
|
|
issuer="open-webui",
|
|
options={"require": ["sub", "iss", "iat", "exp"]},
|
|
)
|
|
except jwt.PyJWTError as exc:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid signed user identity") from exc
|
|
user_id = str(claims.get("sub", "")).strip()
|
|
if not user_id:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Signed identity has no subject")
|
|
return UserIdentity(
|
|
user_id=user_id,
|
|
email=str(claims.get("email", "")),
|
|
name=str(claims.get("name", "")),
|
|
role=str(claims.get("role", "user")),
|
|
)
|
|
|
|
|
|
def require_gateway_identity(
|
|
authorization: Annotated[
|
|
str | None,
|
|
Header(alias="Authorization", include_in_schema=False),
|
|
] = None,
|
|
user_jwt: Annotated[
|
|
str | None,
|
|
Header(alias="X-OpenWebUI-User-Jwt", include_in_schema=False),
|
|
] = None,
|
|
) -> UserIdentity:
|
|
settings = get_settings()
|
|
verify_service_bearer(authorization, settings.internal_gateway_key)
|
|
return decode_openwebui_identity(user_jwt, settings.openwebui_forward_jwt_secret)
|