Steven
0c53aa7062
- backend: FastAPI 应用工厂、统一错误信封、JWT、短信登录、工作区/成员/邀请、幂等中间件 - alembic 0001: workspace/member/invitation/audit_log/app_user + RLS 策略 + 双角色 GRANT - tests: 认证/工作区/租户隔离/幂等 四组用例（尚未在本机执行） - web: React18+Vite 骨架（登录、工作区、邀请落地页） - android: Compose 骨架（登录、工作区列表） - 2.Design: PRODUCT-PLAN / ROADMAP-M0-M1 / NAMING / DEVELOPMENT
81 lines
2.5 KiB
Python
81 lines
2.5 KiB
Python
"""令牌与验证码。
|
|
|
|
安全约定：
|
|
- access token 短期（默认 30 分钟），refresh token 长期（14 天），两者 type 不同，不可互换。
|
|
- 令牌里不放手机号等 PII，只放 sub=user_id。
|
|
- 验证码只在服务端比对，**不写日志**；开发环境返回固定码仅用于本地联调。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
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)
|