From df460eeea841d51958371535500050a32ae56664 Mon Sep 17 00:00:00 2001 From: xsl Date: Tue, 15 Sep 2026 00:02:55 +0800 Subject: [PATCH] =?UTF-8?q?feat(auth):=20=E9=82=AE=E7=AE=B1=E9=AA=8C?= =?UTF-8?q?=E8=AF=81=E7=A0=81=E7=99=BB=E5=BD=95=E9=80=9A=E9=81=93=EF=BC=88?= =?UTF-8?q?=E6=AD=A3=E5=BC=8F=E5=85=9C=E5=BA=95=EF=BC=8C=E6=9B=BF=E4=BB=A3?= =?UTF-8?q?=E4=B8=8D=E5=8F=AF=E7=94=A8=E7=9A=84=E7=9F=AD=E4=BF=A1=EF=BC=89?= =?UTF-8?q?=E2=80=94=E2=80=94/auth/email/code=20+=20/auth/email/login?= =?UTF-8?q?=EF=BC=8CSmtpEmailSender=EF=BC=88stdlib=20smtplib=EF=BC=8C?= =?UTF-8?q?=E6=97=A0=E6=96=B0=E4=BE=9D=E8=B5=96=EF=BC=89+=20FakeEmailSende?= =?UTF-8?q?r=EF=BC=88=E4=BB=85=E6=9C=AC=E5=9C=B0=EF=BC=89=EF=BC=9B?= =?UTF-8?q?=E9=A2=91=E6=8E=A7=E5=AF=B9=E9=BD=90=E7=9F=AD=E4=BF=A1=E5=8F=A3?= =?UTF-8?q?=E5=BE=84=EF=BC=8860s=20=E5=86=B7=E5=8D=B4/=E9=82=AE=E7=AE=B1?= =?UTF-8?q?=201h=E2=89=A45=C2=B724h=E2=89=A410/IP=E2=89=A420/=E8=AE=BE?= =?UTF-8?q?=E5=A4=87=E2=89=A410=EF=BC=89=EF=BC=9B=E6=96=B0=E5=A2=9E=20EMAI?= =?UTF-8?q?L=5F*/SMTP=5F*=20=E9=85=8D=E7=BD=AE=E4=B8=8E=2040112/40113/4291?= =?UTF-8?q?2/42913/61001=20=E9=94=99=E8=AF=AF=E7=A0=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- apps/api/main.py | 2 + apps/api/routers/auth.py | 8 +- apps/api/routers/auth_email.py | 114 ++++++++ packages/identity/email_login.py | 384 ++++++++++++++++++++++++++ packages/shared/config.py | 30 ++ packages/shared/errors.py | 13 +- tests/integration/test_auth_email.py | 273 +++++++++++++++++++ tests/unit/test_email_login.py | 391 +++++++++++++++++++++++++++ 8 files changed, 1212 insertions(+), 3 deletions(-) create mode 100644 apps/api/routers/auth_email.py create mode 100644 packages/identity/email_login.py create mode 100644 tests/integration/test_auth_email.py create mode 100644 tests/unit/test_email_login.py diff --git a/apps/api/main.py b/apps/api/main.py index bebd51b..c7f3380 100644 --- a/apps/api/main.py +++ b/apps/api/main.py @@ -15,6 +15,7 @@ from apps.api.deps import build_llm_gateway from apps.api.routers import ( agent, auth, + auth_email, auth_sms, events, health, @@ -84,6 +85,7 @@ def create_app(settings: Settings | None = None) -> FastAPI: routers = ( auth.router, auth_sms.router, + auth_email.router, agent.router, quota.router, tasks.router, diff --git a/apps/api/routers/auth.py b/apps/api/routers/auth.py index 20698e9..5873b67 100755 --- a/apps/api/routers/auth.py +++ b/apps/api/routers/auth.py @@ -8,8 +8,12 @@ 华为账号登录（T-P1-12）已实现（``/auth/huawei``）：Account Kit id_token 的 OIDC 验签与 UnionID 映射在 :mod:`packages.identity.huawei`，代码先行——真实 联调待 AGC 配置（见该模块 docstring）。明确不做（防止后人误判「漏了」）： -手机验证码（T-P1-12b，等短信通道）、游客/合并/注销。dev-token 端点不是正式 -鉴权设计，仅为纵向切片与联调服务：``DEV_AUTH_ENABLED`` + 本地环境双开关才放行。 +游客/合并/注销。dev-token 端点不是正式鉴权设计，仅为纵向切片与联调服务： +``DEV_AUTH_ENABLED`` + 本地环境双开关才放行。 + +短信验证码通道已由邮箱验证码通道替代（FR-ACC-14 v2 变更，TD-32 短信实现 +仅作参照保留）：正式兜底登录走 ``auth_email`` 路由的 +``/auth/email/code`` + ``/auth/email/login``。 """ from __future__ import annotations diff --git a/apps/api/routers/auth_email.py b/apps/api/routers/auth_email.py new file mode 100644 index 0000000..881defb --- /dev/null +++ b/apps/api/routers/auth_email.py @@ -0,0 +1,114 @@ +"""邮箱验证码登录路由（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)) diff --git a/packages/identity/email_login.py b/packages/identity/email_login.py new file mode 100644 index 0000000..89abd7f --- /dev/null +++ b/packages/identity/email_login.py @@ -0,0 +1,384 @@ +"""邮箱验证码登录（FR-ACC-14 v2 变更：正式登录的兜底通道，替代短信）。 + +短信通道（``sms_login.py``）因签名/资质上不了线，产品决策改用邮箱验证码 +承接「无第三方账号用户的正式登录」。本模块与 sms_login **同构**：验证码 +生命周期与三重频控口径逐条对齐（6 位数字 / TTL 300s / 校验最多 5 次 / +重发冷却 60s / 单邮箱 1h≤5 / 24h≤10 / 单 IP 24h≤20 / 单设备 24h≤10）， +Redis 键前缀换成 ``email:``，差异只在与通道相关的地方： + +- **union_id 是明文邮箱**（``email:<小写邮箱>``，产品契约，非哈希）： + 邮箱本身是登录账号而非手机号那样的敏感标识，掩码纪律落在「日志与 + Redis 键」两处——日志只打掩码（``ab***@example.com``），Redis 键用 + ``sha256(规范化邮箱)`` 前 16 位，不存明文。 +- **设备维度缺省用 IP**：请求头 ``X-Device-Key`` 缺省时路由层以客户端 + IP 兜底传入，保证「单设备 24h≤10」对所有请求生效（sms 通道允许不传 + 设备号跳过该维度，邮箱通道不做这个让步——攻击面更大，宁可误伤不可漏拦）。 +- **发送通道**：``SmtpEmailSender`` 用 stdlib ``smtplib`` 同步发送、 + ``anyio.to_thread`` 包成异步（不引 aiosmtplib 依赖）；``FakeEmailSender`` + 验证码明文进日志，**仅限本地开发**（``EMAIL_PROVIDER=fake`` 默认值）。 +- 频控计数同样按「请求」计而非「成功发送」计（理由见 sms_login docstring）。 +""" + +from __future__ import annotations + +import hashlib +import hmac +import re +import smtplib +from datetime import datetime, timedelta, timezone +from email.message import EmailMessage +from typing import Protocol, runtime_checkable + +import anyio.to_thread +from pydantic import BaseModel, Field + +from packages.identity.sms_login import generate_code, hash_code +from packages.shared.config import Settings +from packages.shared.errors import AppError, ErrorCode +from packages.shared.logging import get_logger + +logger = get_logger(__name__) + +#: 宽松邮箱正则（前端也校验）：非空白 @ 非空白 . 非空白，长度上限交给 Pydantic。 +_EMAIL_RE = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+$") + +#: 自然窗口的时区：与 sms/billing 的日预算桶一致（FR-BILL-01 纪律），UTC+8。 +_CN_TZ = timezone(timedelta(hours=8)) + +_HOUR_BUCKET_TTL = 2 * 60 * 60 # 小时桶键活 2h，覆盖桶期 + 少量时钟偏移 +_DAY_BUCKET_TTL = 26 * 60 * 60 # 日桶键活 26h，过期由桶名（日期）保证语义正确 + +#: union_id 的 provider 前缀（id_user_identity.provider 固定 "email"）。 +_EMAIL_PROVIDER = "email" + + +# --- 邮箱规范化与掩码 ---------------------------------------------------------- + + +def normalize_email(raw: str) -> str: + """规范化邮箱：去首尾空白 + 小写；宽松正则不过即 40000（不消耗频控额度）。""" + email = raw.strip().lower() + if not _EMAIL_RE.fullmatch(email) or len(email) > 254: + raise AppError(ErrorCode.BAD_REQUEST, message="邮箱格式不正确", hint="请输入正确的邮箱地址") + return email + + +def masked_email(email: str) -> str: + """日志掩码：``ab***@example.com``（输入应为规范化后的邮箱）。 + + 本地部分只留前 2 位；不足 2 位按 1 位留。域名不打码（域名不含 PII， + 且运营排查需要它区分邮箱服务商）。 + """ + local, _, domain = email.partition("@") + head = local[:2] if len(local) >= 2 else local[:1] + return f"{head}***@{domain}" + + +def _hash_key(value: str) -> str: + """任意维度标识（IP/设备号/邮箱）进 Redis 键前的哈希：键不落明文。""" + return hashlib.sha256(value.encode()).hexdigest()[:16] + + +def email_hash_key(email: str) -> str: + """邮箱在 Redis 键中的形态：sha256 前 16 位（内部先规范化，同箱恒同键）。""" + return _hash_key(normalize_email(email)) + + +def email_union_id(email: str) -> str: + """邮箱映射到 ``id_user_identity.union_id``（产品契约：明文小写邮箱）。""" + return f"{_EMAIL_PROVIDER}:{normalize_email(email)}" + + +# --- wire 模型 ----------------------------------------------------------------- + + +class EmailSendRequest(BaseModel): + """POST /auth/email/code 请求体。""" + + email: str = Field(min_length=5, max_length=254, description="邮箱地址") + + +class EmailLoginRequest(BaseModel): + """POST /auth/email/login 请求体。""" + + email: str = Field(min_length=5, max_length=254) + code: str = Field(min_length=4, max_length=8, description="6 位数字验证码") + + +# --- 存储接口（Redis 生产 / dict 单测）------------------------------------------ + + +@runtime_checkable +class EmailStore(Protocol): + """验证码与频控计数的最小存储接口（与 SmsStore 同形）。""" + + async def get(self, key: str) -> str | None: ... + + async def set(self, key: str, value: str, ttl_seconds: int) -> None: ... + + async def delete(self, key: str) -> None: ... + + async def incr(self, key: str, ttl_seconds: int) -> int: ... + + +class RedisEmailStore: + """生产实现：包一层 ``redis.asyncio.Redis`` 适配到 :class:`EmailStore`。 + + 持有 ``object`` 而非 ``Redis`` 具体类型并局部 ignore——与 + ``RedisSmsStore`` 相同的做法：identity 包只认接口，不 import redis 包。 + """ + + def __init__(self, redis: object) -> None: + self._redis = redis + + async def get(self, key: str) -> str | None: + value = await self._redis.get(key) # type: ignore[attr-defined] + return str(value) if value is not None else None + + async def set(self, key: str, value: str, ttl_seconds: int) -> None: + await self._redis.set(key, value, ex=ttl_seconds) # type: ignore[attr-defined] + + async def delete(self, key: str) -> None: + await self._redis.delete(key) # type: ignore[attr-defined] + + async def incr(self, key: str, ttl_seconds: int) -> int: + # INCR + EXPIRE 走 pipeline：计数与过期原子下发（同 RedisSmsStore）。 + pipe = self._redis.pipeline() # type: ignore[attr-defined] + pipe.incr(key) + pipe.expire(key, ttl_seconds) + results = await pipe.execute() + return int(results[0]) + + +# --- 发送通道（Protocol 隔离）--------------------------------------------------- + + +@runtime_checkable +class EmailSender(Protocol): + """验证码发送通道。fake 打日志（dev），smtp 真实下发。 + + 实现层约定：日志/异常只允许 ``masked``，不得记录邮箱明文； + ``code`` 明文仅 dev 专用实现可为联调打印。 + """ + + async def send_code(self, email: str, masked: str, code: str) -> None: ... + + +class FakeEmailSender: + """dev 专用（``EMAIL_PROVIDER=fake``，默认值）：验证码打日志。 + + 生产必须切 smtp——这里把 code 明文打进日志只是本地联调需要， + 任何非 local 环境都不该用它。 + """ + + async def send_code(self, email: str, masked: str, code: str) -> None: + logger.info("email_code_issued", masked=masked, code=code) + + +class SmtpEmailSender: + """SMTP 真实通道（stdlib smtplib 同步发送，anyio.to_thread 包成异步）。 + + 构造时校验配置完整性（缺 SMTP_HOST/SMTP_USER/SMTP_PASS 即抛 61001 + EMAIL_PROVIDER_FAILED，装配期炸，不带病上岗）；发送失败/超时同样映射 + 61001——通道故障是外部依赖问题，不能被当成 5xxxx 内部错误稀释告警。 + """ + + def __init__(self, settings: Settings) -> None: + missing = [ + name + for name, value in ( + ("SMTP_HOST", settings.smtp_host), + ("SMTP_USER", settings.smtp_user), + ("SMTP_PASS", settings.smtp_pass), + ) + if not value + ] + if missing: + # 只回显配置项名字，绝不回显已配置的值（secret 可能就在其中） + raise AppError( + ErrorCode.EMAIL_PROVIDER_FAILED, + message="邮件通道配置不完整", + data={"missing": missing}, + ) + self._settings = settings + + def _build_message(self, email: str, code: str) -> EmailMessage: + settings = self._settings + sender = settings.mail_from or settings.smtp_user + message = EmailMessage() + message["From"] = sender + message["To"] = email + message["Subject"] = "【美搭】登录验证码" + message.set_content( + f"您的登录验证码是：{code}\n\n" + f"{settings.email_code_ttl_seconds // 60} 分钟内有效，请勿泄露给他人。\n" + "若非本人操作，请忽略本邮件。" + ) + return message + + def _send_sync(self, email: str, code: str) -> None: + """同步发送本体（在 worker 线程里跑）：smtplib 全程不触碰事件循环。""" + settings = self._settings + message = self._build_message(email, code) + if settings.smtp_use_ssl: + with smtplib.SMTP_SSL(settings.smtp_host, settings.smtp_port, timeout=15) as client: + client.login(settings.smtp_user, settings.smtp_pass) + client.send_message(message) + else: + with smtplib.SMTP(settings.smtp_host, settings.smtp_port, timeout=15) as client: + client.starttls() + client.login(settings.smtp_user, settings.smtp_pass) + client.send_message(message) + + async def send_code(self, email: str, masked: str, code: str) -> None: + try: + await anyio.to_thread.run_sync(self._send_sync, email, code) + except AppError: + raise + except Exception as exc: + # 网络超时/认证失败/被对方拒收都视为通道故障；细节只留类型名 + logger.warning("email_provider_error", masked=masked, error=type(exc).__name__) + raise AppError(ErrorCode.EMAIL_PROVIDER_FAILED) from exc + + +# --- 服务：验证码生命周期 + 三重频控 ------------------------------------------ + + +class EmailAuthService: + """验证码生命周期（6 位 / 300s / 一次性 / 失败 5 次作废）+ 三重频控。 + + 频控维度与 sms_login 对齐：同邮箱（60s 冷却 / 1h≤5 / 24h≤10）、同 IP + （24h≤20）、同设备（24h≤10，设备键由路由层以 ``X-Device-Key`` 或客户端 + IP 兜底，永远参与计数——与 sms 通道「不传不拦」的差异见模块 docstring）。 + """ + + def __init__(self, store: EmailStore, sender: EmailSender, *, settings: Settings) -> None: + self._store = store + self._sender = sender + self._settings = settings + + async def request_code(self, email: str, *, ip: str, device_id: str | None = None) -> None: + """发送验证码：先频控后发送，全过才生成并存储随机码。""" + normalized = normalize_email(email) + ekey = email_hash_key(normalized) + masked = masked_email(normalized) + cfg = self._settings + now = datetime.now(_CN_TZ) + day = f"{now:%Y%m%d}" + hour = f"{now:%Y%m%d%H}" + cooldown_key = f"email:cooldown:{ekey}" + + # 计数按请求计（含被拒请求）：与 sms_login 同一论证。 + hourly = await self._store.incr(f"email:cnt:h:{ekey}:{hour}", ttl_seconds=_HOUR_BUCKET_TTL) + daily = await self._store.incr(f"email:cnt:d:{ekey}:{day}", ttl_seconds=_DAY_BUCKET_TTL) + ip_count = await self._store.incr( + f"email:cnt:ip:{_hash_key(ip)}:{day}", ttl_seconds=_DAY_BUCKET_TTL + ) + # 设备维度永远参与：路由层已用 X-Device-Key（缺省 IP）兜底传入。 + device_count = await self._store.incr( + f"email:cnt:dev:{_hash_key(device_id or ip)}:{day}", ttl_seconds=_DAY_BUCKET_TTL + ) + + # 检查顺序与 sms_login 一致：先报「今日已达上限」，再小时/冷却，最后维度限额。 + if daily > cfg.email_daily_limit: + logger.info("email_rate_limited", masked=masked, dimension="email_daily", count=daily) + raise AppError(ErrorCode.EMAIL_DAILY_LIMIT) + if hourly > cfg.email_hourly_limit: + logger.info("email_rate_limited", masked=masked, dimension="email_hourly", count=hourly) + raise AppError(ErrorCode.EMAIL_SEND_TOO_FREQUENT) + if ( + cfg.email_resend_cooldown_seconds > 0 + and await self._store.get(cooldown_key) is not None + ): + logger.info("email_rate_limited", masked=masked, dimension="cooldown") + raise AppError(ErrorCode.EMAIL_SEND_TOO_FREQUENT) + if ip_count > cfg.email_ip_daily_limit: + logger.info("email_rate_limited", masked=masked, dimension="ip_daily", count=ip_count) + raise AppError(ErrorCode.EMAIL_DAILY_LIMIT) + if device_count > cfg.email_device_daily_limit: + logger.info( + "email_rate_limited", masked=masked, dimension="device_daily", count=device_count + ) + raise AppError(ErrorCode.EMAIL_DAILY_LIMIT) + + code = generate_code(cfg.email_code_length) + await self._store.set( + f"email:code:{ekey}", hash_code(code), ttl_seconds=cfg.email_code_ttl_seconds + ) + # 新码作废旧码的全部失败计数：「失败 5 次」按码计，不按邮箱计 + await self._store.delete(f"email:attempts:{ekey}") + if cfg.email_resend_cooldown_seconds > 0: + await self._store.set(cooldown_key, "1", ttl_seconds=cfg.email_resend_cooldown_seconds) + logger.info("email_code_dispatched", masked=masked, ttl=cfg.email_code_ttl_seconds) + try: + await self._sender.send_code(normalized, masked, code) + except AppError: + raise + except Exception as exc: + # 兜底：sender 实现漏映射异常时，通道故障也不许变成 5xxxx 内部错误 + logger.warning("email_sender_unexpected_error", masked=masked, error=type(exc).__name__) + raise AppError(ErrorCode.EMAIL_PROVIDER_FAILED) from exc + + async def verify( + self, email: str, code: str, *, ip: str = "", device_id: str | None = None + ) -> str: + """校验验证码；成功返回该邮箱的 ``union_id``，由调用方建户/查户。 + + ``ip``/``device_id`` 为风控审计预留（与 sms_login.verify 同口径， + 校验侧的防护是「失败 5 次作废」+ 一次性删除）。 + """ + normalized = normalize_email(email) + ekey = email_hash_key(normalized) + masked = masked_email(normalized) + stored = await self._store.get(f"email:code:{ekey}") + if stored is None: + # 无码 = 过期 / 从未发送 / 已用尽作废 / 已消费——统一按过期回应， + # 不区分细节，避免给攻击者探测状态的机会 + raise AppError(ErrorCode.EMAIL_CODE_EXPIRED) + if not hmac.compare_digest(stored, hash_code(code)): + attempts = await self._store.incr( + f"email:attempts:{ekey}", ttl_seconds=self._settings.email_code_ttl_seconds + ) + if attempts >= self._settings.email_verify_max_attempts: + await self._store.delete(f"email:code:{ekey}") + await self._store.delete(f"email:attempts:{ekey}") + logger.warning("email_code_invalidated", masked=masked, attempts=attempts) + else: + logger.info("email_code_rejected", masked=masked, attempts=attempts) + raise AppError(ErrorCode.EMAIL_CODE_INVALID) + # 一次性：成功即删（连同失败计数），二次使用按「已过期」回应 + await self._store.delete(f"email:code:{ekey}") + await self._store.delete(f"email:attempts:{ekey}") + logger.info("email_code_verified", masked=masked) + return email_union_id(normalized) + + +def build_email_auth_service(settings: Settings, redis: object) -> EmailAuthService: + """按 ``EMAIL_PROVIDER`` 装配（apps/api 层调用；单测直接构造 service）。""" + sender: EmailSender + if settings.email_provider == "fake": + sender = FakeEmailSender() + elif settings.email_provider == "smtp": + sender = SmtpEmailSender(settings) # 配置缺失在这里就抛 61001（构造失败） + else: + raise AppError( + ErrorCode.CONFIG_ERROR, message=f"未知 EMAIL_PROVIDER: {settings.email_provider}" + ) + return EmailAuthService(RedisEmailStore(redis), sender, settings=settings) + + +__all__ = [ + "EmailAuthService", + "EmailLoginRequest", + "EmailSendRequest", + "EmailSender", + "EmailStore", + "FakeEmailSender", + "RedisEmailStore", + "SmtpEmailSender", + "build_email_auth_service", + "email_hash_key", + "email_union_id", + "masked_email", + "normalize_email", +] diff --git a/packages/shared/config.py b/packages/shared/config.py index 8fbda97..2fc6971 100644 --- a/packages/shared/config.py +++ b/packages/shared/config.py @@ -207,6 +207,25 @@ class Settings(BaseSettings): sms_ip_daily_limit: int = 20 sms_device_daily_limit: int = 10 + # --- 邮箱验证码登录（FR-ACC-14 v2 变更：短信通道由邮箱通道替代）------------ + # fake = 验证码明文进日志（仅限本地开发，生产必须切 smtp）； + # smtp = stdlib smtplib 经 anyio.to_thread 同步发送（不引 aiosmtplib 依赖）。 + email_provider: str = "fake" + smtp_host: str = "" # 空 = smtp provider 不可装配（构造即 61001，不带病上岗） + smtp_port: int = 465 + smtp_user: str = "" + smtp_pass: str = "" + smtp_use_ssl: bool = True + mail_from: str = "" # 空 = 发件人回退用 SMTP_USER + email_code_ttl_seconds: int = 300 + email_code_length: int = 6 + email_verify_max_attempts: int = 5 + email_resend_cooldown_seconds: int = 60 + email_hourly_limit: int = 5 + email_daily_limit: int = 10 + email_ip_daily_limit: int = 20 + email_device_daily_limit: int = 10 + # --- 极简运营后台（T-P2-09）---------------------------------------------- # 空字符串 = 后台全部端点 404（等价于未部署）。V1 极简鉴权：共享令牌， # 后续接运营账号体系时替换。令牌只进 .env / SSM，绝不进代码。 @@ -230,6 +249,16 @@ class Settings(BaseSettings): ], alias="REPORT_PIPELINE_STAGES_JSON", ) + # 深度诊断（tier=full）流水线：无视觉阶段（复用 basic 报告的 derived_features）， + # 诊断装配复用，文案/校验换 full 专用阶段（四大块结构化产出）。 + report_full_pipeline_stages: list[str] = Field( + default_factory=lambda: [ + "packages.report.diagnosis_stage:DiagnosisStage", + "packages.report.full_copy_stage:FullCopyStage", + "packages.report.full_validation_stage:FullValidationStage", + ], + alias="REPORT_FULL_PIPELINE_STAGES_JSON", + ) # 视觉提供者：deterministic = V1 骨架的确定性假实现（真实 RTMPose/BiRefNet # 是 Sprint 1 结论落地后的替换项，接口位见 packages/report/vision.py）。 report_vision_provider: str = "deterministic" @@ -254,6 +283,7 @@ class Settings(BaseSettings): "upload_scenes", "retrieval_rerank_weights", "report_pipeline_stages", + "report_full_pipeline_stages", mode="before", ) @classmethod diff --git a/packages/shared/errors.py b/packages/shared/errors.py index a1abda9..607f40c 100644 --- a/packages/shared/errors.py +++ b/packages/shared/errors.py @@ -34,6 +34,8 @@ class ErrorCode(IntEnum): SESSION_REVOKED = 40105 SMS_CODE_INVALID = 40110 SMS_CODE_EXPIRED = 40111 + EMAIL_CODE_INVALID = 40112 + EMAIL_CODE_EXPIRED = 40113 FORBIDDEN = 40300 QUOTA_EXHAUSTED = 40301 CONSENT_REQUIRED = 40302 @@ -46,10 +48,12 @@ class ErrorCode(IntEnum): REPORT_EXPORT_NOT_READY = 40421 STATE_CONFLICT = 40900 TASK_BONUS_ALREADY_CLAIMED = 40910 - REPORT_ALREADY_EXISTS = 40920 + REPORT_ALREADY_EXISTS = 40920 # 已下线，仅历史兼容：重复生成改由 report_basic 配额兜底 RATE_LIMITED = 42900 SMS_SEND_TOO_FREQUENT = 42910 SMS_DAILY_LIMIT = 42911 + EMAIL_SEND_TOO_FREQUENT = 42912 + EMAIL_DAILY_LIMIT = 42913 # --- 5xxxx server ----------------------------------------------------- INTERNAL_ERROR = 50000 @@ -67,6 +71,7 @@ class ErrorCode(IntEnum): MODERATION_UNAVAILABLE = 60005 LLM_SCENE_DEGRADED = 60006 SMS_PROVIDER_FAILED = 61000 + EMAIL_PROVIDER_FAILED = 61001 @dataclass(frozen=True, slots=True) @@ -95,6 +100,8 @@ ERROR_META: dict[ErrorCode, ErrorMeta] = { ErrorCode.SESSION_REVOKED: ErrorMeta(401, "该会话已登出", "请重新登录"), ErrorCode.SMS_CODE_INVALID: ErrorMeta(401, "验证码错误", "请检查后重试"), ErrorCode.SMS_CODE_EXPIRED: ErrorMeta(401, "验证码已过期", "请重新获取"), + ErrorCode.EMAIL_CODE_INVALID: ErrorMeta(401, "验证码错误", "请检查后重试"), + ErrorCode.EMAIL_CODE_EXPIRED: ErrorMeta(401, "验证码已过期", "请重新获取"), ErrorCode.FORBIDDEN: ErrorMeta(403, "没有访问权限"), ErrorCode.QUOTA_EXHAUSTED: ErrorMeta(403, "今日额度已用完", "明天恢复，或完成任务多换几次"), ErrorCode.CONSENT_REQUIRED: ErrorMeta(403, "需要先完成授权", "请阅读并同意后继续"), @@ -111,10 +118,13 @@ ERROR_META: dict[ErrorCode, ErrorMeta] = { ErrorCode.TASK_BONUS_ALREADY_CLAIMED: ErrorMeta( 409, "这个周期的任务奖励已领取", "下个周期再来" ), + # 40920 已无任何 raise 方（service 终身资格硬校验已删），保留仅为兼容旧端缓存的错误文案。 ErrorCode.REPORT_ALREADY_EXISTS: ErrorMeta(409, "本期报告已生成过", "每 30 天可重新生成一次"), ErrorCode.RATE_LIMITED: ErrorMeta(429, "操作太频繁了", "请稍后再试"), ErrorCode.SMS_SEND_TOO_FREQUENT: ErrorMeta(429, "发送太频繁了", "请稍后再试"), ErrorCode.SMS_DAILY_LIMIT: ErrorMeta(429, "今日发送次数已达上限", "明天再试"), + ErrorCode.EMAIL_SEND_TOO_FREQUENT: ErrorMeta(429, "发送太频繁了", "请稍后再试"), + ErrorCode.EMAIL_DAILY_LIMIT: ErrorMeta(429, "今日发送次数已达上限", "明天再试"), ErrorCode.INTERNAL_ERROR: ErrorMeta(500, "服务开小差了", "请稍后重试"), ErrorCode.DB_ERROR: ErrorMeta(500, "服务开小差了", "请稍后重试"), ErrorCode.TASK_STATE_INVALID: ErrorMeta(500, "任务状态异常", "请稍后在首页「报告」中查看"), @@ -138,6 +148,7 @@ ERROR_META: dict[ErrorCode, ErrorMeta] = { 503, "该功能今日暂时关闭", "AI 服务今日已达服务上限，明天恢复" ), ErrorCode.SMS_PROVIDER_FAILED: ErrorMeta(502, "短信服务暂时不可用", "请稍后重试"), + ErrorCode.EMAIL_PROVIDER_FAILED: ErrorMeta(502, "邮件服务暂时不可用", "请稍后重试"), } diff --git a/tests/integration/test_auth_email.py b/tests/integration/test_auth_email.py new file mode 100644 index 0000000..080a717 --- /dev/null +++ b/tests/integration/test_auth_email.py @@ -0,0 +1,273 @@ +"""邮箱登录集成（FR-ACC-14 v2 变更）—— 真实 PostgreSQL + Redis，EMAIL_PROVIDER=fake。 + +测试钩子：sender 经 ``app.state.email_auth`` 注入为捕获型 fake（路由的惰性装配 +优先读 app.state），验证码从 fixture 拿而不是去挖生产日志——生产代码不为 +测试留任何后门。每用例用随机邮箱/IP/设备：频控计数按 UTC+8 自然日分桶， +固定值会跨用例、跨次运行、跨天污染 Redis 计数。 + +``email_resend_cooldown_seconds`` 覆写为 0：同邮箱二次登录（重复发送）与 +刷量循环不必等 60s 真实冷却；冷却语义在单测里用注入时钟验证。 +""" + +from __future__ import annotations + +import os +import uuid +from collections.abc import AsyncIterator +from contextlib import asynccontextmanager +from typing import Any + +import httpx +import pytest + +from packages.identity.email_login import EmailAuthService, RedisEmailStore + + +def _infra_available() -> bool: + return bool(os.getenv("MEIDA_RUN_INTEGRATION")) + + +pytestmark = [ + pytest.mark.integration, + pytest.mark.skipif( + not _infra_available(), + reason="set MEIDA_RUN_INTEGRATION=1 with PostgreSQL and Redis running", + ), +] + + +class CapturingSender: + """捕获型 fake sender：把发出去的码留在内存里供 login 使用。""" + + def __init__(self) -> None: + self.codes: dict[str, str] = {} + + async def send_code(self, email: str, masked: str, code: str) -> None: + self.codes[email] = code + + +def _rand_email() -> str: + # 随机 local 部分：跨运行撞桶概率可忽略 + return f"it-{uuid.uuid4().hex[:12]}@example.com" + + +def _rand_ip() -> str: + # 随机源 IP（文档保留网段 198.51.100.0/24）：让 IP 维度计数互不污染 + return f"198.51.100.{uuid.uuid4().int % 254 + 1}" + + +@pytest.fixture(autouse=True) +async def _clean_email_counters() -> AsyncIterator[None]: + """每用例前清空 email:* 计数键（理由同 test_auth_sms 的同名 fixture）。""" + if not _infra_available(): + yield + return + from packages.shared.redis_client import get_redis + + redis = await get_redis() + keys = [k async for k in redis.scan_iter(match="email:*")] + if keys: + await redis.delete(*keys) + yield + + +@pytest.fixture +async def env() -> AsyncIterator[tuple[Any, CapturingSender]]: + from apps.api.main import create_app + from packages.shared.config import get_settings + from packages.shared.redis_client import get_redis + + settings = get_settings().model_copy(update={"email_resend_cooldown_seconds": 0}) + app = create_app(settings) + sender = CapturingSender() + app.state.email_auth = EmailAuthService(RedisEmailStore(get_redis()), sender, settings=settings) + async with app.router.lifespan_context(app): + yield app, sender + + +@asynccontextmanager +async def _client(app: Any, ip: str | None = None) -> AsyncIterator[httpx.AsyncClient]: + """ASGI 客户端；``client=(ip, port)`` 决定 Request.client.host（IP 维度输入）。""" + transport = httpx.ASGITransport(app=app, client=(ip or _rand_ip(), 54321)) + async with httpx.AsyncClient(transport=transport, base_url="http://test") as http: + yield http + + +async def _code( + http: httpx.AsyncClient, email: str, device_key: str | None = None +) -> httpx.Response: + headers = {"X-Device-Key": device_key} if device_key else {} + return await http.post("/api/v1/auth/email/code", json={"email": email}, headers=headers) + + +async def test_send_then_login_returns_working_token_pair( + env: tuple[Any, CapturingSender], +) -> None: + app, sender = env + email = _rand_email() + async with _client(app) as http: + sent = await _code(http, email, device_key="dev-a") + assert sent.status_code == 200 + body = sent.json() + assert body["code"] == 0 + assert body["data"]["sent"] is True + # cooldown_seconds 回报的是进程配置值（全局 Settings，不受 fixture 覆写影响） + assert body["data"]["cooldown_seconds"] == 60 + + login = await http.post( + "/api/v1/auth/email/login", + json={"email": email, "code": sender.codes[email]}, + ) + assert login.status_code == 200 + data = login.json()["data"] + assert data["access_token"] + assert data["refresh_token"] + assert data["expires_in"] > 0 + assert data["refresh_expires_in"] > 0 + assert data["user_id"] + + # 新签发的 access 立即可用（双令牌链路闭环） + quota = await http.get( + "/api/v1/quota", headers={"Authorization": f"Bearer {data['access_token']}"} + ) + assert quota.status_code == 200 + assert quota.json()["code"] == 0 + + +async def test_same_email_relogin_maps_to_same_user(env: tuple[Any, CapturingSender]) -> None: + """幂等：同邮箱重复登录返回同一 user_id（provider=email 映射唯一）。""" + app, sender = env + email = _rand_email() + async with _client(app) as http: + user_ids = [] + for _ in range(2): + await _code(http, email) # cooldown=0，可立即重发取新码 + login = await http.post( + "/api/v1/auth/email/login", + json={"email": email, "code": sender.codes[email]}, + ) + assert login.status_code == 200 + user_ids.append(login.json()["data"]["user_id"]) + assert user_ids[0] == user_ids[1] + + +async def test_email_case_insensitive_login(env: tuple[Any, CapturingSender]) -> None: + """大小写归一：注册用小写，登录用大写混搭，映射到同一 user_id。""" + app, sender = env + email = _rand_email() + async with _client(app) as http: + await _code(http, email) + login = await http.post( + "/api/v1/auth/email/login", + json={"email": email.upper(), "code": sender.codes[email]}, + ) + assert login.status_code == 200 + first = login.json()["data"]["user_id"] + + await _code(http, email) + relogin = await http.post( + "/api/v1/auth/email/login", + json={"email": email, "code": sender.codes[email]}, + ) + assert relogin.status_code == 200 + assert relogin.json()["data"]["user_id"] == first + + +async def test_scripted_flood_eleventh_send_hits_daily_limit( + env: tuple[Any, CapturingSender], +) -> None: + """FR-ACC-14 口径的邮箱版验收：同邮箱连发 11 次。 + + 前 5 次放行；第 6~10 次先撞 1h≤5 的小时窗（42912）；第 11 次撞 24h≤10 + 的日窗（42913）。设备逐次更换以隔离设备维度。 + """ + app, _ = env + email = _rand_email() + async with _client(app) as http: + for i in range(11): + resp = await _code(http, email, device_key=f"flood-dev-{i}") + body = resp.json() + if i < 5: + assert resp.status_code == 200, body + elif i < 10: + assert resp.status_code == 429 and body["code"] == 42912, body + else: + assert resp.status_code == 429 and body["code"] == 42913, body + + +async def test_ip_dimension_blocks_twenty_first_send(env: tuple[Any, CapturingSender]) -> None: + """IP 维度刷量：同一 IP 打 21 个不同邮箱，第 21 次 42913。""" + app, _ = env + ip = _rand_ip() + async with _client(app, ip=ip) as http: + for _ in range(20): + resp = await _code(http, _rand_email(), device_key=f"ipdev-{uuid.uuid4().hex[:8]}") + assert resp.status_code == 200, resp.text + blocked = await _code(http, _rand_email(), device_key=f"ipdev-{uuid.uuid4().hex[:8]}") + assert blocked.status_code == 429 + assert blocked.json()["code"] == 42913 + + +async def test_device_dimension_blocks_eleventh_send(env: tuple[Any, CapturingSender]) -> None: + """设备维度刷量：同一 X-Device-Key 换邮箱换 IP 打 11 次，第 11 次 42913。""" + app, _ = env + device = f"dev-{uuid.uuid4().hex[:12]}" + async with _client(app) as http: # 本用例客户端 IP 固定但仅 11 次发送 < IP 上限 20 + for _ in range(10): + resp = await _code(http, _rand_email(), device_key=device) + assert resp.status_code == 200, resp.text + blocked = await _code(http, _rand_email(), device_key=device) + assert blocked.status_code == 429 + assert blocked.json()["code"] == 42913 + + +async def test_device_dimension_falls_back_to_ip(env: tuple[Any, CapturingSender]) -> None: + """缺 X-Device-Key 时设备维度以 IP 兜底：同 IP 打 11 个邮箱，第 11 次 42913。""" + app, _ = env + ip = _rand_ip() + async with _client(app, ip=ip) as http: + for _ in range(10): + resp = await _code(http, _rand_email()) + assert resp.status_code == 200, resp.text + blocked = await _code(http, _rand_email()) + assert blocked.status_code == 429 + assert blocked.json()["code"] == 42913 + + +async def test_five_wrong_logins_invalidate_the_code(env: tuple[Any, CapturingSender]) -> None: + """验证码错 5 次作废——第 5 次仍报 40112，之后正确码也只剩 40113。""" + app, sender = env + email = _rand_email() + async with _client(app) as http: + await _code(http, email) + code = sender.codes[email] + wrong = code[:-1] + ("0" if code[-1] != "0" else "1") # 保证 != 真码 + for _ in range(5): + resp = await http.post("/api/v1/auth/email/login", json={"email": email, "code": wrong}) + assert resp.status_code == 401 + assert resp.json()["code"] == 40112 + # 码已作废：即使拿到正确码也只能 40113（一次性 + 作废都不留活口） + after = await http.post("/api/v1/auth/email/login", json={"email": email, "code": code}) + assert after.status_code == 401 + assert after.json()["code"] == 40113 + + +async def test_login_without_code_is_expired(env: tuple[Any, CapturingSender]) -> None: + """从未发过验证码的邮箱直接登录：40113（端点活着、校验链路通）。""" + app, _ = env + async with _client(app) as http: + resp = await http.post( + "/api/v1/auth/email/login", + json={"email": _rand_email(), "code": "000000"}, + ) + assert resp.status_code == 401 + assert resp.json()["code"] == 40113 + + +async def test_code_rejects_malformed_email(env: tuple[Any, CapturingSender]) -> None: + """服务层规范化兜底：能过 Pydantic 长度但非邮箱 → 40000。""" + app, _ = env + async with _client(app) as http: + resp = await _code(http, "not-an-email") + assert resp.status_code == 400 + assert resp.json()["code"] == 40000 diff --git a/tests/unit/test_email_login.py b/tests/unit/test_email_login.py new file mode 100644 index 0000000..3a9ebe0 --- /dev/null +++ b/tests/unit/test_email_login.py @@ -0,0 +1,391 @@ +"""EmailAuthService 单测（packages/identity/email_login.py）——无 IO、无网络。 + +存储用自写内存 dict 实现 Protocol 化的 EmailStore 接口（与 test_sms_login +同一做法，不引 fakeredis）；TTL 用可注入时钟模拟过期。频控计数语义按 +「请求」计（见被测模块 docstring），错误码矩阵对应 FR-ACC-14 v2 的 +三重频控（邮箱版）。 +""" + +from __future__ import annotations + +import re +import time +from collections.abc import Callable +from typing import Any + +import pytest + +from packages.identity.email_login import ( + EmailAuthService, + FakeEmailSender, + SmtpEmailSender, + build_email_auth_service, + email_hash_key, + email_union_id, + masked_email, + normalize_email, +) +from packages.identity.sms_login import hash_code +from packages.shared.config import Settings +from packages.shared.errors import AppError, ErrorCode + +EMAIL = "someone@example.com" +MASKED = "so***@example.com" + + +class FakeClock: + """单调时钟桩：advance 推进 TTL 判定，单测无需真实等待。""" + + def __init__(self) -> None: + self._now = 1000.0 + + def __call__(self) -> float: + return self._now + + def advance(self, seconds: float) -> None: + self._now += seconds + + +class InMemoryEmailStore: + """dict 实现的 EmailStore：键 → (值, 过期时刻)；过期惰性清除。""" + + def __init__(self, clock: Callable[[], float] | None = None) -> None: + self._data: dict[str, tuple[str, float]] = {} + self._clock = clock or time.monotonic + + async def get(self, key: str) -> str | None: + entry = self._data.get(key) + if entry is None: + return None + value, expires_at = entry + if self._clock() >= expires_at: + self._data.pop(key, None) + return None + return value + + async def set(self, key: str, value: str, ttl_seconds: int) -> None: + self._data[key] = (value, self._clock() + ttl_seconds) + + async def delete(self, key: str) -> None: + self._data.pop(key, None) + + async def incr(self, key: str, ttl_seconds: int) -> int: + current = await self.get(key) # 复用过期语义，与 Redis 行为对齐 + nxt = int(current) + 1 if current is not None else 1 + self._data[key] = (str(nxt), self._clock() + ttl_seconds) + return nxt + + +class CapturingSender: + """捕获型 fake：记下发出的 (邮箱, 掩码, 验证码)，不打日志。""" + + def __init__(self) -> None: + self.sent: list[tuple[str, str, str]] = [] + + async def send_code(self, email: str, masked: str, code: str) -> None: + self.sent.append((email, masked, code)) + + +class FailingSender: + async def send_code(self, email: str, masked: str, code: str) -> None: + raise RuntimeError("provider down") + + +def _settings(**overrides: Any) -> Settings: + """单测配置：不吃 worktree 的 .env（_env_file=None），保持确定性。 + + 默认 cooldown=0：让频控矩阵只受被测的那一个维度约束（冷却与限额分开测）。 + """ + values: dict[str, Any] = {"_env_file": None, "email_resend_cooldown_seconds": 0} + values.update(overrides) + return Settings(**values) + + +def _service( + settings: Settings, sender: Any | None = None +) -> tuple[EmailAuthService, InMemoryEmailStore, CapturingSender, FakeClock]: + clock = FakeClock() + store = InMemoryEmailStore(clock) + capturing = CapturingSender() + service = EmailAuthService(store, sender or capturing, settings=settings) + return service, store, capturing, clock + + +def _wrong_code(code: str) -> str: + """保证与真码不同的 6 位错码（尾位翻转）。""" + return code[:-1] + ("0" if code[-1] != "0" else "1") + + +def _last_code(sender: CapturingSender) -> str: + return sender.sent[-1][2] + + +# --- 邮箱规范化与掩码 --------------------------------------------------------- + + +@pytest.mark.parametrize( + ("raw", "expected"), + [ + ("someone@example.com", "someone@example.com"), + (" Someone@Example.COM ", "someone@example.com"), + ("a.b+c@example.com", "a.b+c@example.com"), + ], +) +def test_normalize_email_lowercases_and_strips(raw: str, expected: str) -> None: + assert normalize_email(raw) == expected + + +@pytest.mark.parametrize("bad", ["not-an-email", "a@b", "a b@c.com", "plainaddress", "@x.com"]) +def test_normalize_email_rejects_invalid_formats(bad: str) -> None: + # 非法格式是 400 类错误，且不消耗频控额度（不碰存储） + with pytest.raises(AppError) as excinfo: + normalize_email(bad) + assert excinfo.value.code == ErrorCode.BAD_REQUEST + + +def test_masked_email_hides_local_part() -> None: + assert masked_email(EMAIL) == MASKED + assert "someone" not in masked_email(EMAIL) + + +def test_email_hash_key_is_16_hex_chars() -> None: + digest = email_hash_key(EMAIL) + assert len(digest) == 16 + assert re.fullmatch(r"[0-9a-f]{16}", digest) + + +def test_email_union_id_is_plaintext_lowercase_email() -> None: + # 产品契约（与 phone 通道哈希 union_id 不同）：email union_id 即明文小写邮箱 + uid = email_union_id(" Someone@Example.COM ") + assert uid == "email:someone@example.com" + + +# --- 发送：码形态与存储纪律 --------------------------------------------------- + + +async def test_request_code_sends_six_digit_code() -> None: + service, _, sender, _ = _service(_settings()) + await service.request_code(EMAIL, ip="1.2.3.4", device_id="dev-1") + assert len(sender.sent) == 1 + email, masked, code = sender.sent[0] + assert email == EMAIL + assert masked == MASKED + assert re.fullmatch(r"\d{6}", code) + + +async def test_code_stored_as_sha256_never_plaintext() -> None: + service, store, sender, _ = _service(_settings()) + await service.request_code(EMAIL, ip="203.0.113.7", device_id="dev-1") + code = _last_code(sender) + entry = store._data.get(f"email:code:{email_hash_key(EMAIL)}") + assert entry is not None and entry[0] == hash_code(code) + for key, (value, _) in store._data.items(): + assert EMAIL not in key and EMAIL not in value + assert code not in key and code not in value + assert "203.0.113.7" not in key and "203.0.113.7" not in value + assert "dev-1" not in key + + +# --- 频控矩阵（FR-ACC-14 v2）--------------------------------------------------- + + +async def test_hourly_limit_sixth_request_rejected() -> None: + # 同邮箱 1h ≤ 5：第 6 次 42912 + service, _, sender, _ = _service(_settings(email_hourly_limit=5)) + for _ in range(5): + await service.request_code(EMAIL, ip="1.1.1.1") + with pytest.raises(AppError) as excinfo: + await service.request_code(EMAIL, ip="1.1.1.1") + assert excinfo.value.code == ErrorCode.EMAIL_SEND_TOO_FREQUENT + assert len(sender.sent) == 5 + + +async def test_daily_limit_eleventh_request_rejected() -> None: + # 同邮箱 24h ≤ 10：第 11 次 42913（小时上限抬高以单独验证日窗） + service, _, sender, _ = _service(_settings(email_hourly_limit=99, email_daily_limit=10)) + for _ in range(10): + await service.request_code(EMAIL, ip="1.1.1.1") + with pytest.raises(AppError) as excinfo: + await service.request_code(EMAIL, ip="1.1.1.1") + assert excinfo.value.code == ErrorCode.EMAIL_DAILY_LIMIT + assert len(sender.sent) == 10 + + +async def test_ip_dimension_twenty_first_rejected() -> None: + # 同 IP 24h ≤ 20：每次换新邮箱 + 新设备键（设备维度缺省回退 IP，不隔离会 + # 先撞设备上限 10——这正是「缺省用 IP」的语义），第 21 次 42913 + service, _, _, _ = _service(_settings(email_ip_daily_limit=20)) + for i in range(20): + await service.request_code(f"user{i}@example.com", ip="9.9.9.9", device_id=f"dev-{i}") + with pytest.raises(AppError) as excinfo: + await service.request_code("other@example.com", ip="9.9.9.9", device_id="dev-x") + assert excinfo.value.code == ErrorCode.EMAIL_DAILY_LIMIT + + +async def test_device_dimension_eleventh_rejected() -> None: + # 同设备 24h ≤ 10：第 11 次 42913（邮箱与 IP 都逐次更换以隔离维度） + service, _, _, _ = _service(_settings(email_device_daily_limit=10)) + for i in range(10): + await service.request_code(f"user{i}@example.com", ip=f"10.0.0.{i}", device_id="device-a") + with pytest.raises(AppError) as excinfo: + await service.request_code("other@example.com", ip="10.0.0.99", device_id="device-a") + assert excinfo.value.code == ErrorCode.EMAIL_DAILY_LIMIT + + +async def test_device_dimension_falls_back_to_ip() -> None: + # 不传 device_id 时该维度以 IP 兜底计数：同 IP 打 11 个邮箱，第 11 次 42913 + service, _, _, _ = _service(_settings(email_device_daily_limit=10, email_ip_daily_limit=99)) + for i in range(10): + await service.request_code(f"user{i}@example.com", ip="10.0.0.1") + with pytest.raises(AppError) as excinfo: + await service.request_code("other@example.com", ip="10.0.0.1") + assert excinfo.value.code == ErrorCode.EMAIL_DAILY_LIMIT + + +async def test_cooldown_blocks_resend_within_window_and_allows_after() -> None: + # 同邮箱 60s 冷却：期内二次 42912，过窗放行（小时上限抬高排除干扰） + service, _, sender, clock = _service( + _settings(email_resend_cooldown_seconds=60, email_hourly_limit=99) + ) + await service.request_code(EMAIL, ip="1.1.1.1") + with pytest.raises(AppError) as excinfo: + await service.request_code(EMAIL, ip="1.1.1.1") + assert excinfo.value.code == ErrorCode.EMAIL_SEND_TOO_FREQUENT + clock.advance(61) + await service.request_code(EMAIL, ip="1.1.1.1") + assert len(sender.sent) == 2 + + +async def test_sender_failure_maps_to_provider_failed() -> None: + # 通道故障 → 61001（外部依赖错误，不与内部 5xxxx 混淆） + service = EmailAuthService(InMemoryEmailStore(), FailingSender(), settings=_settings()) + with pytest.raises(AppError) as excinfo: + await service.request_code(EMAIL, ip="1.1.1.1") + assert excinfo.value.code == ErrorCode.EMAIL_PROVIDER_FAILED + + +# --- 校验：验证码生命周期 ----------------------------------------------------- + + +async def test_verify_success_is_one_time() -> None: + # 一次性：成功即删，同一码二次使用按「已过期」回应 + service, _, sender, _ = _service(_settings()) + await service.request_code(EMAIL, ip="1.1.1.1") + code = _last_code(sender) + assert await service.verify(EMAIL, code) == email_union_id(EMAIL) + with pytest.raises(AppError) as excinfo: + await service.verify(EMAIL, code) + assert excinfo.value.code == ErrorCode.EMAIL_CODE_EXPIRED + + +async def test_verify_expired_after_ttl() -> None: + # 5 分钟 TTL：过窗后校验 40113 + service, _, sender, clock = _service(_settings(email_code_ttl_seconds=300)) + await service.request_code(EMAIL, ip="1.1.1.1") + clock.advance(301) + with pytest.raises(AppError) as excinfo: + await service.verify(EMAIL, _last_code(sender)) + assert excinfo.value.code == ErrorCode.EMAIL_CODE_EXPIRED + + +async def test_four_wrong_attempts_still_allowed_then_correct_code_passes() -> None: + # 失败 5 次作废：错 4 次仍可再试，第 5 次尝试用对码可以通过 + service, _, sender, _ = _service(_settings()) + await service.request_code(EMAIL, ip="1.1.1.1") + wrong = _wrong_code(_last_code(sender)) + for _ in range(4): + with pytest.raises(AppError) as excinfo: + await service.verify(EMAIL, wrong) + assert excinfo.value.code == ErrorCode.EMAIL_CODE_INVALID + assert await service.verify(EMAIL, _last_code(sender)) == email_union_id(EMAIL) + + +async def test_fifth_wrong_attempt_invalidates_code() -> None: + # 第 5 次错误：返回 40112 且码已删——之后连正确码都只剩 40113 + service, _, sender, _ = _service(_settings()) + await service.request_code(EMAIL, ip="1.1.1.1") + correct = _last_code(sender) + for _ in range(5): + with pytest.raises(AppError) as excinfo: + await service.verify(EMAIL, _wrong_code(correct)) + assert excinfo.value.code == ErrorCode.EMAIL_CODE_INVALID + with pytest.raises(AppError) as excinfo: + await service.verify(EMAIL, correct) + assert excinfo.value.code == ErrorCode.EMAIL_CODE_EXPIRED + + +async def test_new_code_resets_attempt_counter() -> None: + # 「失败 5 次」按码计不按邮箱计：换新码后失败计数清零 + service, _, sender, _ = _service(_settings(email_hourly_limit=99)) + await service.request_code(EMAIL, ip="1.1.1.1") + for _ in range(4): + with pytest.raises(AppError): + await service.verify(EMAIL, _wrong_code(_last_code(sender))) + await service.request_code(EMAIL, ip="1.1.1.1") # cooldown=0 可立即重发 + assert await service.verify(EMAIL, _last_code(sender)) == email_union_id(EMAIL) + + +async def test_verify_unknown_email_is_expired() -> None: + # 从未发过码的邮箱：40113（不区分「无码」细节，防状态探测） + service, _, _, _ = _service(_settings()) + with pytest.raises(AppError) as excinfo: + await service.verify(EMAIL, "123456") + assert excinfo.value.code == ErrorCode.EMAIL_CODE_EXPIRED + + +# --- 通道装配与 SMTP 实现（不连网：桩替真实连接）------------------------------- + + +def test_build_service_fake_provider() -> None: + service = build_email_auth_service(_settings(email_provider="fake"), object()) + assert isinstance(service._sender, FakeEmailSender) + + +def test_build_service_unknown_provider_is_config_error() -> None: + with pytest.raises(AppError) as excinfo: + build_email_auth_service(_settings(email_provider="pigeon"), object()) + assert excinfo.value.code == ErrorCode.CONFIG_ERROR + + +def test_build_service_smtp_without_config_maps_61001() -> None: + # 构造失败（SMTP 配置缺失）→ 61001：装配期就炸，不带病上岗 + with pytest.raises(AppError) as excinfo: + build_email_auth_service(_settings(email_provider="smtp"), object()) + assert excinfo.value.code == ErrorCode.EMAIL_PROVIDER_FAILED + + +_SMTP_SETTINGS: dict[str, Any] = { + "email_provider": "smtp", + "smtp_host": "smtp.example.com", + "smtp_user": "noreply@example.com", + "smtp_pass": "secret", +} + + +class _RecordingSmtpSender(SmtpEmailSender): + """替换 _send_sync 为记录调用（不碰网络）的桩实现。""" + + def __init__(self, settings: Settings) -> None: + super().__init__(settings) + self.calls: list[tuple[str, str]] = [] + + def _send_sync(self, email: str, code: str) -> None: + self.calls.append((email, code)) + + +async def test_smtp_sender_builds_message_and_accepts_success() -> None: + sender = _RecordingSmtpSender(_settings(**_SMTP_SETTINGS)) + await sender.send_code(EMAIL, MASKED, "123456") + assert sender.calls == [(EMAIL, "123456")] + + +class _ExplodingSmtpSender(SmtpEmailSender): + def _send_sync(self, email: str, code: str) -> None: + raise TimeoutError("connect timeout") + + +async def test_smtp_sender_network_error_maps_61001() -> None: + sender = _ExplodingSmtpSender(_settings(**_SMTP_SETTINGS)) + with pytest.raises(AppError) as excinfo: + await sender.send_code(EMAIL, MASKED, "123456") + assert excinfo.value.code == ErrorCode.EMAIL_PROVIDER_FAILED