56 lines
1.9 KiB
Python
56 lines
1.9 KiB
Python
# sanguo_api/auth.py
|
|
"""JWT 单用户认证。secret/用户名/密码 hash 来自 config/backtest.yaml。"""
|
|
import os
|
|
from datetime import datetime, timedelta, timezone
|
|
import jwt
|
|
import bcrypt
|
|
from fastapi import HTTPException, status
|
|
|
|
_CONFIG = {"secret": "change-me", "expire_minutes": 60, "algorithm": "HS256"}
|
|
|
|
|
|
def set_jwt_config(secret: str, expire_minutes: int, algorithm: str = "HS256"):
|
|
_CONFIG.update(secret=secret, expire_minutes=expire_minutes, algorithm=algorithm)
|
|
|
|
|
|
def hash_password(password: str) -> str:
|
|
return bcrypt.hashpw(password.encode("utf-8"), bcrypt.gensalt()).decode("utf-8")
|
|
|
|
|
|
def verify_password(password: str, password_hash: str) -> bool:
|
|
try:
|
|
return bcrypt.checkpw(password.encode("utf-8"), password_hash.encode("utf-8"))
|
|
except (ValueError, TypeError):
|
|
return False
|
|
|
|
|
|
def create_token(username: str) -> str:
|
|
payload = {
|
|
"sub": username,
|
|
"exp": datetime.now(timezone.utc) + timedelta(minutes=_CONFIG["expire_minutes"]),
|
|
}
|
|
return jwt.encode(payload, _CONFIG["secret"], algorithm=_CONFIG["algorithm"])
|
|
|
|
|
|
def verify_token(token: str) -> str:
|
|
try:
|
|
payload = jwt.decode(token, _CONFIG["secret"], algorithms=[_CONFIG["algorithm"]])
|
|
return payload["sub"]
|
|
except jwt.PyJWTError:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="无效 token")
|
|
|
|
|
|
def token_expires_in() -> int:
|
|
"""当前配置的 token 有效期(秒),登录/刷新响应的 expires_in。"""
|
|
return int(_CONFIG["expire_minutes"] * 60)
|
|
|
|
|
|
def get_token_exp(token: str) -> int | None:
|
|
"""解析 token 的 exp(unix 秒),不验签(供拦截器判断剩余时间);失败 → None。"""
|
|
try:
|
|
payload = jwt.decode(token, options={"verify_signature": False})
|
|
exp = payload.get("exp")
|
|
return int(exp) if exp is not None else None
|
|
except (jwt.PyJWTError, ValueError, TypeError):
|
|
return None
|