115 lines
4.4 KiB
Python
115 lines
4.4 KiB
Python
"""邮箱验证码登录路由（FR-ACC-14 v2 变更：短信通道由邮箱通道替代）。
|
||
|
||
POST /auth/email/code、POST /auth/email/login；与 auth_sms.py 同构：
|
||
登录成功只依赖 identity 域已有设施（get_or_create_by_external_identity +
|
||
issue_session + TokenPairResponse）。设备维度输入取请求头 ``X-Device-Key``，
|
||
缺省以客户端 IP 兜底（见 packages/identity/email_login.py docstring）。
|
||
|
||
``_access_token``/``_token_pair`` 各持一份小拷贝：私有助手不跨路由复用，
|
||
与 auth_sms.py 的既定做法保持一致。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from typing import Annotated
|
||
|
||
from fastapi import APIRouter, Depends, Request
|
||
|
||
from apps.api.deps import IdentityServiceDep, SettingsDep
|
||
from packages.identity.email_login import (
|
||
EmailAuthService,
|
||
EmailLoginRequest,
|
||
EmailSendRequest,
|
||
build_email_auth_service,
|
||
)
|
||
from packages.identity.schemas import TokenPairResponse
|
||
from packages.identity.service import SessionIssued
|
||
from packages.shared.config import Settings, get_settings
|
||
from packages.shared.redis_client import get_redis
|
||
from packages.shared.response import ok
|
||
from packages.shared.security import create_access_token
|
||
|
||
router = APIRouter(prefix="/auth", tags=["auth-email"])
|
||
|
||
#: 设备维度请求头：缺省时该维度以客户端 IP 兜底参与频控（见模块 docstring）。
|
||
DEVICE_KEY_HEADER = "X-Device-Key"
|
||
|
||
|
||
def email_auth_dep(request: Request) -> EmailAuthService:
|
||
"""惰性装配并缓存到 ``app.state``（与 auth_sms.sms_auth_dep 同一模式）。
|
||
|
||
缓存到 app.state 同时是集成测试的钩子：测试在发请求前把捕获型 fake
|
||
sender 的 service 放进 ``app.state.email_auth`` 即可拿到验证码——生产
|
||
代码不为测试留任何后门。
|
||
"""
|
||
service: EmailAuthService | None = getattr(request.app.state, "email_auth", None)
|
||
if service is None:
|
||
service = build_email_auth_service(get_settings(), get_redis())
|
||
request.app.state.email_auth = service
|
||
return service
|
||
|
||
|
||
EmailAuthDep = Annotated[EmailAuthService, Depends(email_auth_dep)]
|
||
|
||
|
||
def _client_ip(request: Request) -> str:
|
||
"""直连对端 IP（ASGI scope）。反向代理场景的字段解析属网关层职责，V1 不做。"""
|
||
return request.client.host if request.client is not None else "unknown"
|
||
|
||
|
||
def _device_key(request: Request) -> str:
|
||
"""设备维度：X-Device-Key 缺省用客户端 IP（邮箱通道不做「不传不拦」）。"""
|
||
header = request.headers.get(DEVICE_KEY_HEADER)
|
||
return header.strip() if header and header.strip() else _client_ip(request)
|
||
|
||
|
||
def _access_token(settings: Settings, user_id: int) -> str:
|
||
return create_access_token(
|
||
user_id=user_id,
|
||
secret=settings.jwt_secret,
|
||
algorithm=settings.jwt_algorithm,
|
||
ttl_seconds=settings.jwt_access_ttl_seconds,
|
||
issuer=settings.jwt_issuer,
|
||
)
|
||
|
||
|
||
def _token_pair(
|
||
settings: Settings, access_token: str, issued: SessionIssued
|
||
) -> dict[str, str | int]:
|
||
"""与 auth.py 的 dev-token/refresh 完全一致的双令牌响应形态。"""
|
||
return TokenPairResponse(
|
||
access_token=access_token,
|
||
refresh_token=issued.refresh_token,
|
||
expires_in=settings.jwt_access_ttl_seconds,
|
||
refresh_expires_in=settings.refresh_token_ttl_seconds,
|
||
user_id=str(issued.user_id),
|
||
).model_dump()
|
||
|
||
|
||
@router.post("/email/code", summary="发送邮箱登录验证码")
|
||
async def send_email_code(
|
||
payload: EmailSendRequest,
|
||
request: Request,
|
||
email: EmailAuthDep,
|
||
settings: SettingsDep,
|
||
) -> dict[str, object]:
|
||
# 限流/格式错误（42912/42913/40000/61001）由统一异常处理器透传给客户端
|
||
await email.request_code(payload.email, ip=_client_ip(request), device_id=_device_key(request))
|
||
return ok({"sent": True, "cooldown_seconds": settings.email_resend_cooldown_seconds})
|
||
|
||
|
||
@router.post("/email/login", summary="校验邮箱验证码并登录（双令牌）")
|
||
async def email_login(
|
||
payload: EmailLoginRequest,
|
||
request: Request,
|
||
email: EmailAuthDep,
|
||
settings: SettingsDep,
|
||
identity: IdentityServiceDep,
|
||
) -> dict[str, object]:
|
||
union_id = await email.verify(payload.email, payload.code)
|
||
user = await identity.get_or_create_by_external_identity(
|
||
provider="email", union_id=union_id, register_source="email"
|
||
)
|
||
issued = await identity.issue_session(user.user_id)
|
||
return ok(_token_pair(settings, _access_token(settings, user.user_id), issued))
|