"""令牌与验证码。 安全约定: - access token 短期(默认 30 分钟),refresh token 长期(14 天),两者 type 不同,不可互换。 - 令牌里不放手机号等 PII,只放 sub=user_id。 - 验证码只在服务端比对,**不写日志**;开发环境返回固定码仅用于本地联调。 """ from __future__ import annotations import base64 import hashlib import hmac import secrets import time import uuid from dataclasses import dataclass from typing import Any, Literal import jwt from .config import get_settings from .errors import AppError, ErrorCode settings = get_settings() TokenType = Literal["access", "refresh"] @dataclass(frozen=True) class TokenPair: access_token: str refresh_token: str expires_in: int def _new_token(user_id: uuid.UUID, token_type: TokenType, ttl_seconds: int) -> str: now = int(time.time()) payload: dict[str, Any] = { "sub": str(user_id), "typ": token_type, "iat": now, "exp": now + ttl_seconds, "jti": secrets.token_urlsafe(8), } return jwt.encode(payload, settings.jwt_secret, algorithm=settings.jwt_algorithm) def create_token_pair(user_id: uuid.UUID) -> TokenPair: access_ttl = settings.access_token_minutes * 60 return TokenPair( access_token=_new_token(user_id, "access", access_ttl), refresh_token=_new_token(user_id, "refresh", settings.refresh_token_days * 86400), expires_in=access_ttl, ) def decode_token(token: str, expect: TokenType) -> uuid.UUID: try: payload = jwt.decode(token, settings.jwt_secret, algorithms=[settings.jwt_algorithm]) except jwt.ExpiredSignatureError as exc: raise AppError(ErrorCode.TOKEN_EXPIRED, "登录已过期,请重新登录") from exc except jwt.PyJWTError as exc: raise AppError(ErrorCode.TOKEN_INVALID, "令牌无效") from exc if payload.get("typ") != expect: # 用 refresh 当 access 用是常见攻击面,必须拦 raise AppError(ErrorCode.TOKEN_INVALID, "令牌类型不匹配") try: return uuid.UUID(str(payload["sub"])) except (KeyError, ValueError) as exc: raise AppError(ErrorCode.TOKEN_INVALID, "令牌内容不合法") from exc def new_invite_code() -> str: # 去掉容易看错的 0/O/1/I alphabet = "23456789ABCDEFGHJKLMNPQRSTUVWXYZ" return "".join(secrets.choice(alphabet) for _ in range(8)) def invite_code_matches(provided: str, expected: str) -> bool: return hmac.compare_digest(provided.strip().upper(), expected) # —— 登录密码 —— # 用标准库 scrypt,不引入 passlib/bcrypt:少一个依赖,也就少一处装不上的风险。 # 存储格式 `scrypt$n$r$p$salt_b64$hash_b64`,参数写进串里,将来调强度不用改代码。 _SCRYPT_N = 2**14 _SCRYPT_R = 8 _SCRYPT_P = 1 _PASSWORD_MIN = 8 _PASSWORD_MAX = 64 def hash_password(password: str) -> str: salt = secrets.token_bytes(16) digest = hashlib.scrypt( password.encode("utf-8"), salt=salt, n=_SCRYPT_N, r=_SCRYPT_R, p=_SCRYPT_P, dklen=32 ) return "$".join( [ "scrypt", str(_SCRYPT_N), str(_SCRYPT_R), str(_SCRYPT_P), base64.b64encode(salt).decode(), base64.b64encode(digest).decode(), ] ) def verify_password(password: str, stored: str | None) -> bool: """stored 为空(没设过密码)一律返回 False,别把"没密码"当成"密码对"。""" if not stored: return False try: scheme, n, r, p, salt_b64, hash_b64 = stored.split("$") if scheme != "scrypt": return False digest = hashlib.scrypt( password.encode("utf-8"), salt=base64.b64decode(salt_b64), n=int(n), r=int(r), p=int(p), dklen=len(base64.b64decode(hash_b64)), ) except (ValueError, TypeError): return False return hmac.compare_digest(digest, base64.b64decode(hash_b64)) def password_problem(password: str) -> str | None: """返回不合格的原因;合格返回 None。规则写在一处,设置与修改都用它。""" if len(password) < _PASSWORD_MIN: return f"密码至少 {_PASSWORD_MIN} 位" if len(password) > _PASSWORD_MAX: return f"密码最多 {_PASSWORD_MAX} 位" has_letter = any(c.isalpha() for c in password) has_digit = any(c.isdigit() for c in password) if not (has_letter and has_digit): return "密码要同时有字母和数字" return None