feat(m0): 后端骨架 + Web/安卓前端骨架 + 设计文档
- 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
This commit is contained in:
@@ -0,0 +1,80 @@
|
||||
"""令牌与验证码。
|
||||
|
||||
安全约定:
|
||||
- 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)
|
||||
Reference in New Issue
Block a user