55 lines
2.0 KiB
Python
55 lines
2.0 KiB
Python
from datetime import datetime, timedelta, timezone
|
||
import importlib.util
|
||
|
||
from fastapi import Depends, HTTPException, status
|
||
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
|
||
|
||
_jose_spec = importlib.util.find_spec("jose")
|
||
_jose_origin = _jose_spec.origin if _jose_spec else None
|
||
|
||
_incompatible_jose = bool(_jose_origin and _jose_origin.endswith("site-packages/jose.py"))
|
||
|
||
try:
|
||
import jwt
|
||
from jwt import InvalidTokenError as JWTError
|
||
except Exception as exc:
|
||
if not _incompatible_jose:
|
||
try:
|
||
from jose import JWTError, jwt
|
||
except Exception as fallback_exc:
|
||
raise RuntimeError(
|
||
"JWT 依赖不可用，请安装 python-jose[cryptography] 或 PyJWT"
|
||
) from fallback_exc
|
||
else:
|
||
raise RuntimeError(
|
||
"检测到不兼容依赖 jose，请卸载 jose 后安装 python-jose[cryptography] 或安装 PyJWT"
|
||
) from exc
|
||
|
||
from backend.config import settings
|
||
|
||
security = HTTPBearer()
|
||
|
||
|
||
def create_access_token(subject: str) -> str:
|
||
expire = datetime.now(timezone.utc) + timedelta(minutes=settings.access_token_expire_minutes)
|
||
payload = {"sub": subject, "exp": expire}
|
||
return jwt.encode(payload, settings.jwt_secret_key, algorithm=settings.jwt_algorithm)
|
||
|
||
|
||
def verify_password(username: str, password: str) -> bool:
|
||
return username == settings.admin_username and password == settings.admin_password
|
||
|
||
|
||
def get_current_user(
|
||
credentials: HTTPAuthorizationCredentials = Depends(security),
|
||
) -> str:
|
||
token = credentials.credentials
|
||
try:
|
||
payload = jwt.decode(token, settings.jwt_secret_key, algorithms=[settings.jwt_algorithm])
|
||
username: str | None = payload.get("sub")
|
||
if not username:
|
||
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="无效Token")
|
||
return username
|
||
except JWTError as exc:
|
||
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="登录已过期") from exc
|