"""简单 Token 鉴权(用户名 + 密码 → HMAC token)。""" from __future__ import annotations import hashlib import hmac import secrets from typing import Annotated from fastapi import Depends, Header, HTTPException, Request, status from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer from packages.config import get_settings _bearer = HTTPBearer(auto_error=False) COOKIE_NAME = "mi_token" def auth_disabled() -> bool: s = get_settings() return (s.auth_secret or "").strip().lower() in ("", "disabled", "off", "none") def _token_for_credentials(username: str, password: str, secret: str) -> str: payload = f"{username}:{password}" return hmac.new( secret.encode("utf-8"), payload.encode("utf-8"), hashlib.sha256, ).hexdigest() def verify_credentials(username: str, password: str) -> bool: s = get_settings() u = (username or "").strip() if not u: return False return secrets.compare_digest(u, s.admin_username) and secrets.compare_digest( password, s.admin_password ) def expected_token() -> str: s = get_settings() return _token_for_credentials(s.admin_username, s.admin_password, s.auth_secret) def issue_token(username: str, password: str) -> str | None: s = get_settings() if not verify_credentials(username, password): return None return _token_for_credentials(username, password, s.auth_secret) def _extract_token( request: Request, authorization: str | None, x_mi_token: str | None, creds: HTTPAuthorizationCredentials | None, ) -> str | None: if creds and creds.credentials: return creds.credentials.strip() if authorization and authorization.lower().startswith("bearer "): return authorization[7:].strip() if x_mi_token: return x_mi_token.strip() q = request.query_params.get("token") if q: return q.strip() cookie = request.cookies.get(COOKIE_NAME) if cookie: return cookie.strip() return None def require_auth( request: Request, authorization: Annotated[str | None, Header()] = None, x_mi_token: Annotated[str | None, Header(alias="X-MI-Token")] = None, creds: Annotated[HTTPAuthorizationCredentials | None, Depends(_bearer)] = None, ) -> None: if auth_disabled(): return token = _extract_token(request, authorization, x_mi_token, creds) if not token or not secrets.compare_digest(token, expected_token()): raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="unauthorized", headers={"WWW-Authenticate": "Bearer"}, ) AuthDep = Depends(require_auth)