dekun
94 lines
2.7 KiB
Python
94 lines
2.7 KiB
Python
"""简单 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)
|