dekun
92 lines
2.9 KiB
Python
92 lines
2.9 KiB
Python
"""简单 HMAC Token 鉴权（无 JWT 依赖）。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import hashlib
|
|
import hmac
|
|
import json
|
|
import time
|
|
from typing import Annotated
|
|
|
|
from fastapi import Depends, HTTPException, status
|
|
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
|
|
from pydantic import BaseModel, Field
|
|
|
|
from ..config import Settings, get_settings
|
|
|
|
_bearer = HTTPBearer(auto_error=False)
|
|
|
|
|
|
class LoginRequest(BaseModel):
|
|
username: str = Field(min_length=1)
|
|
password: str = Field(min_length=1)
|
|
|
|
|
|
class LoginResponse(BaseModel):
|
|
token: str
|
|
username: str
|
|
expires_in: int
|
|
env_name: str
|
|
mode: str
|
|
|
|
|
|
def _b64url(data: bytes) -> str:
|
|
return base64.urlsafe_b64encode(data).decode("ascii").rstrip("=")
|
|
|
|
|
|
def _b64url_decode(s: str) -> bytes:
|
|
pad = "=" * (-len(s) % 4)
|
|
return base64.urlsafe_b64decode(s + pad)
|
|
|
|
|
|
def issue_token(username: str, settings: Settings) -> tuple[str, int]:
|
|
exp = int(time.time()) + int(settings.auth_token_ttl_sec)
|
|
payload = {
|
|
"u": username,
|
|
"exp": exp,
|
|
"v": int(settings.auth_token_version),
|
|
}
|
|
raw = _b64url(json.dumps(payload, separators=(",", ":")).encode("utf-8"))
|
|
sig = hmac.new(
|
|
settings.auth_secret.encode("utf-8"),
|
|
raw.encode("ascii"),
|
|
hashlib.sha256,
|
|
).hexdigest()
|
|
return f"{raw}.{sig}", settings.auth_token_ttl_sec
|
|
|
|
|
|
def verify_token(token: str, settings: Settings) -> str:
|
|
try:
|
|
raw, sig = token.rsplit(".", 1)
|
|
except ValueError as e:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="invalid token") from e
|
|
expect = hmac.new(
|
|
settings.auth_secret.encode("utf-8"),
|
|
raw.encode("ascii"),
|
|
hashlib.sha256,
|
|
).hexdigest()
|
|
if not hmac.compare_digest(expect, sig):
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="invalid token")
|
|
try:
|
|
payload = json.loads(_b64url_decode(raw))
|
|
except Exception as e:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="invalid token") from e
|
|
if int(payload.get("exp") or 0) < int(time.time()):
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="token expired")
|
|
if int(payload.get("v") or 0) != int(settings.auth_token_version):
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="token revoked")
|
|
username = str(payload.get("u") or "")
|
|
if not username:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="invalid token")
|
|
return username
|
|
|
|
|
|
def require_user(
|
|
creds: Annotated[HTTPAuthorizationCredentials | None, Depends(_bearer)],
|
|
settings: Annotated[Settings, Depends(get_settings)],
|
|
) -> str:
|
|
if creds is None or not creds.credentials:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="login required")
|
|
return verify_token(creds.credentials, settings)
|