62 lines
2.1 KiB
Python
62 lines
2.1 KiB
Python
"""安全组件：bcrypt 密码哈希 + JWT 双 Token（对齐《02-技术文档》4.2）"""
|
||
import uuid
|
||
from datetime import datetime, timedelta, timezone
|
||
|
||
import bcrypt
|
||
import jwt
|
||
|
||
from app.config import settings
|
||
from app.core import errors
|
||
from app.core.response import BizException
|
||
|
||
|
||
def hash_password(plain: str) -> str:
|
||
return bcrypt.hashpw(plain.encode(), bcrypt.gensalt()).decode()
|
||
|
||
|
||
def verify_password(plain: str, hashed: str) -> bool:
|
||
try:
|
||
return bcrypt.checkpw(plain.encode(), hashed.encode())
|
||
except ValueError:
|
||
return False
|
||
|
||
|
||
def _create_token(subject: int, token_type: str, ttl: int, epoch: int = 0) -> tuple[str, str, int]:
|
||
"""生成 JWT，返回 (token, jti, 过期时间戳)"""
|
||
now = datetime.now(timezone.utc)
|
||
jti = uuid.uuid4().hex
|
||
exp = now + timedelta(seconds=ttl)
|
||
payload = {
|
||
"sub": str(subject),
|
||
"jti": jti,
|
||
"type": token_type, # access / refresh / admin:{role}
|
||
"ep": epoch, # 令牌纪元：改密/注销后递增使旧 Token 全端失效
|
||
"iat": int(now.timestamp()),
|
||
"exp": exp,
|
||
}
|
||
token = jwt.encode(payload, settings.jwt_secret, algorithm=settings.jwt_algorithm)
|
||
return token, jti, int(exp.timestamp())
|
||
|
||
|
||
def create_access_token(user_id: int, epoch: int = 0) -> tuple[str, str, int]:
|
||
return _create_token(user_id, "access", settings.access_token_ttl, epoch)
|
||
|
||
|
||
def create_refresh_token(user_id: int, epoch: int = 0) -> tuple[str, str, int]:
|
||
return _create_token(user_id, "refresh", settings.refresh_token_ttl, epoch)
|
||
|
||
|
||
def create_admin_token(admin_id: int, role: str) -> tuple[str, str, int]:
|
||
"""管理员 Token，payload 带角色，有效期 12h"""
|
||
token, jti, exp = _create_token(admin_id, f"admin:{role}", settings.admin_token_ttl)
|
||
return token, jti, exp
|
||
|
||
|
||
def decode_token(token: str) -> dict:
|
||
"""解码并校验 Token，失败抛 1002"""
|
||
try:
|
||
payload = jwt.decode(token, settings.jwt_secret, algorithms=[settings.jwt_algorithm])
|
||
except jwt.PyJWTError:
|
||
raise BizException(errors.UNAUTHORIZED, http_status=401)
|
||
return payload
|