dekun
88 lines
2.6 KiB
Python
88 lines
2.6 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_password(password: str, secret: str) -> str:
|
|
return hmac.new(
|
|
secret.encode("utf-8"),
|
|
password.encode("utf-8"),
|
|
hashlib.sha256,
|
|
).hexdigest()
|
|
|
|
|
|
def expected_token() -> str:
|
|
s = get_settings()
|
|
return _token_for_password(s.admin_password, s.auth_secret)
|
|
|
|
|
|
def issue_token(password: str) -> str | None:
|
|
s = get_settings()
|
|
if not secrets.compare_digest(password, s.admin_password):
|
|
return None
|
|
return _token_for_password(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()
|
|
# 查询参数兜底（方便内网脚本；生产建议只用 Header）
|
|
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:
|
|
"""
|
|
AUTH_SECRET=disabled 时跳过。
|
|
否则需要 Bearer / X-MI-Token / Cookie / ?token=。
|
|
"""
|
|
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)
|