"""SmsAuthService 单测（packages/identity/sms_login.py）——无 IO、无网络。 存储用自写内存 dict 实现 Protocol 化的 SmsStore 接口（参考 InMemoryCostCounter 的做法，不引 fakeredis）；TTL 用可注入时钟模拟过期。 频控计数语义按「请求」计（见被测模块 docstring），错误码矩阵对应 FR-ACC-14 的三重频控。 """ from __future__ import annotations import re import time from collections.abc import Callable from types import SimpleNamespace from typing import Any import pytest from packages.identity.sms_login import ( FakeSmsSender, SmsAuthService, TencentSmsSender, build_sms_auth_service, generate_code, hash_code, masked_phone, normalize_phone, phone_hash_key, phone_union_id, ) from packages.shared.config import Settings from packages.shared.errors import AppError, ErrorCode PHONE = "13812341234" MASKED = "138****1234" 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 InMemorySmsStore: """dict 实现的 SmsStore：键 → (值, 过期时刻)；过期惰性清除。""" 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, phone: str, masked: str, code: str) -> None: self.sent.append((phone, masked, code)) class FailingSender: async def send_code(self, phone: 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, "sms_resend_cooldown_seconds": 0} values.update(overrides) return Settings(**values) def _service( settings: Settings, sender: Any | None = None ) -> tuple[SmsAuthService, InMemorySmsStore, CapturingSender, FakeClock]: clock = FakeClock() store = InMemorySmsStore(clock) capturing = CapturingSender() service = SmsAuthService(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"), [ ("13812341234", "13812341234"), ("+8613812341234", "13812341234"), ("+86 138-1234 1234", "13812341234"), ("008613812341234", "13812341234"), ("8613812341234", "13812341234"), (" 13812341234 ", "13812341234"), ], ) def test_normalize_phone_accepts_common_writings(raw: str, expected: str) -> None: assert normalize_phone(raw) == expected @pytest.mark.parametrize( "bad", ["12345", "12345678901", "23812341234", "10812341234", "abc-def", "1381234123456"] ) def test_normalize_phone_rejects_invalid_formats(bad: str) -> None: # 非法格式是 400 类错误，且不消耗频控额度（不碰存储） with pytest.raises(AppError) as excinfo: normalize_phone(bad) assert excinfo.value.code == ErrorCode.BAD_REQUEST def test_masked_phone_hides_middle_digits() -> None: assert masked_phone(PHONE) == MASKED assert masked_phone(PHONE)[3:7] == "****" # 中间 4 位不出现 def test_phone_hash_key_is_16_hex_chars() -> None: digest = phone_hash_key(PHONE) assert len(digest) == 16 assert re.fullmatch(r"[0-9a-f]{16}", digest) def test_phone_union_id_is_stable_and_hashed() -> None: uid = phone_union_id("+8613812341234") assert uid.startswith("phone:") # 规范化后同号同 ID：+86 写法与裸号映射到同一用户 assert uid == phone_union_id(PHONE) assert len(uid) == len("phone:") + 16 assert PHONE not in uid # --- 发送：码形态与存储纪律 --------------------------------------------------- async def test_request_code_sends_six_digit_code() -> None: service, _, sender, _ = _service(_settings()) await service.request_code(PHONE, ip="1.2.3.4", device_id="dev-1") assert len(sender.sent) == 1 phone, masked, code = sender.sent[0] assert phone == PHONE assert masked == MASKED assert re.fullmatch(r"\d{6}", code) async def test_code_stored_as_sha256_never_plaintext() -> None: # FR-ACC-14 掩码纪律：Redis（此处为 store）里不得出现明文号码/验证码/IP service, store, sender, _ = _service(_settings()) await service.request_code(PHONE, ip="203.0.113.7", device_id="dev-1") code = _last_code(sender) entry = store._data.get(f"sms:code:{phone_hash_key(PHONE)}") assert entry is not None and entry[0] == hash_code(code) for key, (value, _) in store._data.items(): assert PHONE not in key and PHONE 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 def test_generate_code_is_zero_padded_digits() -> None: assert re.fullmatch(r"\d{6}", generate_code(6)) assert len(generate_code(6)) == 6 def test_hash_code_is_sha256_hex() -> None: assert hash_code("123456") == hash_code("123456") assert len(hash_code("123456")) == 64 assert "123456" not in hash_code("123456") # --- 频控矩阵（FR-ACC-14）----------------------------------------------------- async def test_hourly_limit_sixth_request_rejected() -> None: # 同号 1h ≤ 5：第 6 次 42910 service, _, sender, _ = _service(_settings(sms_hourly_limit=5)) for _ in range(5): await service.request_code(PHONE, ip="1.1.1.1") with pytest.raises(AppError) as excinfo: await service.request_code(PHONE, ip="1.1.1.1") assert excinfo.value.code == ErrorCode.SMS_SEND_TOO_FREQUENT assert len(sender.sent) == 5 async def test_daily_limit_eleventh_request_rejected() -> None: # 同号 24h ≤ 10：第 11 次 42911（小时上限抬高以单独验证日窗） service, _, sender, _ = _service(_settings(sms_hourly_limit=99, sms_daily_limit=10)) for _ in range(10): await service.request_code(PHONE, ip="1.1.1.1") with pytest.raises(AppError) as excinfo: await service.request_code(PHONE, ip="1.1.1.1") assert excinfo.value.code == ErrorCode.SMS_DAILY_LIMIT assert len(sender.sent) == 10 async def test_ip_dimension_twenty_first_rejected() -> None: # 同 IP 24h ≤ 20：每次换新号码（同号额度不干扰），第 21 次 42911 service, _, _, _ = _service(_settings(sms_ip_daily_limit=20)) for i in range(20): await service.request_code(f"138{i:08d}", ip="9.9.9.9") with pytest.raises(AppError) as excinfo: await service.request_code("13899999999", ip="9.9.9.9") assert excinfo.value.code == ErrorCode.SMS_DAILY_LIMIT async def test_device_dimension_eleventh_rejected() -> None: # 同设备 24h ≤ 10：第 11 次 42911（号码与 IP 都逐次更换以隔离维度） service, _, _, _ = _service(_settings(sms_device_daily_limit=10)) for i in range(10): await service.request_code(f"138{i:08d}", ip=f"10.0.0.{i}", device_id="device-a") with pytest.raises(AppError) as excinfo: await service.request_code("13899999999", ip="10.0.0.99", device_id="device-a") assert excinfo.value.code == ErrorCode.SMS_DAILY_LIMIT async def test_cooldown_blocks_resend_within_60s_and_allows_after() -> None: # 同号 60s 冷却：期内二次 42910，过窗放行（小时上限抬高排除干扰） service, _, sender, clock = _service( _settings(sms_resend_cooldown_seconds=60, sms_hourly_limit=99) ) await service.request_code(PHONE, ip="1.1.1.1") with pytest.raises(AppError) as excinfo: await service.request_code(PHONE, ip="1.1.1.1") assert excinfo.value.code == ErrorCode.SMS_SEND_TOO_FREQUENT clock.advance(61) await service.request_code(PHONE, ip="1.1.1.1") assert len(sender.sent) == 2 async def test_sender_failure_maps_to_provider_failed() -> None: # 通道故障 → 61000（外部依赖错误，不与内部 5xxxx 混淆） service = SmsAuthService(InMemorySmsStore(), FailingSender(), settings=_settings()) with pytest.raises(AppError) as excinfo: await service.request_code(PHONE, ip="1.1.1.1") assert excinfo.value.code == ErrorCode.SMS_PROVIDER_FAILED # --- 校验：FR-ACC-04 四要素 --------------------------------------------------- async def test_verify_success_is_one_time() -> None: # 一次性：成功即删，同一码二次使用按「已过期」回应 service, _, sender, _ = _service(_settings()) await service.request_code(PHONE, ip="1.1.1.1") code = _last_code(sender) assert await service.verify(PHONE, code) == phone_union_id(PHONE) with pytest.raises(AppError) as excinfo: await service.verify(PHONE, code) assert excinfo.value.code == ErrorCode.SMS_CODE_EXPIRED async def test_verify_expired_after_ttl() -> None: # 5 分钟 TTL：过窗后校验 40111 service, _, sender, clock = _service(_settings(sms_code_ttl_seconds=300)) await service.request_code(PHONE, ip="1.1.1.1") clock.advance(301) with pytest.raises(AppError) as excinfo: await service.verify(PHONE, _last_code(sender)) assert excinfo.value.code == ErrorCode.SMS_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(PHONE, ip="1.1.1.1") wrong = _wrong_code(_last_code(sender)) for _ in range(4): with pytest.raises(AppError) as excinfo: await service.verify(PHONE, wrong) assert excinfo.value.code == ErrorCode.SMS_CODE_INVALID assert await service.verify(PHONE, _last_code(sender)) == phone_union_id(PHONE) async def test_fifth_wrong_attempt_invalidates_code() -> None: # 第 5 次错误：返回 40110 且码已删——之后连正确码都只剩 40111 service, _, sender, _ = _service(_settings()) await service.request_code(PHONE, ip="1.1.1.1") correct = _last_code(sender) for _ in range(5): with pytest.raises(AppError) as excinfo: await service.verify(PHONE, _wrong_code(correct)) assert excinfo.value.code == ErrorCode.SMS_CODE_INVALID with pytest.raises(AppError) as excinfo: await service.verify(PHONE, correct) assert excinfo.value.code == ErrorCode.SMS_CODE_EXPIRED async def test_new_code_resets_attempt_counter() -> None: # 「失败 5 次」按码计不按号计：换新码后失败计数清零 service, _, sender, _ = _service(_settings(sms_hourly_limit=99)) await service.request_code(PHONE, ip="1.1.1.1") for _ in range(4): with pytest.raises(AppError): await service.verify(PHONE, _wrong_code(_last_code(sender))) await service.request_code(PHONE, ip="1.1.1.1") # cooldown=0 可立即重发 assert await service.verify(PHONE, _last_code(sender)) == phone_union_id(PHONE) async def test_verify_unknown_phone_is_expired() -> None: # 从未发过码的号：40111（不区分「无码」细节，防状态探测） service, _, _, _ = _service(_settings()) with pytest.raises(AppError) as excinfo: await service.verify(PHONE, "123456") assert excinfo.value.code == ErrorCode.SMS_CODE_EXPIRED # --- 通道装配与腾讯实现（不连网：桩替 SDK 客户端）----------------------------- def test_build_service_fake_provider() -> None: service = build_sms_auth_service(_settings(sms_provider="fake"), object()) assert isinstance(service._sender, FakeSmsSender) def test_build_service_unknown_provider_is_config_error() -> None: with pytest.raises(AppError) as excinfo: build_sms_auth_service(_settings(sms_provider="pigeon"), object()) assert excinfo.value.code == ErrorCode.CONFIG_ERROR def test_build_service_tencent_without_credentials_maps_61000() -> None: # 构造失败（凭证缺失）→ 61000：装配期就炸，不带病上岗 with pytest.raises(AppError) as excinfo: build_sms_auth_service(_settings(sms_provider="tencent"), object()) assert excinfo.value.code == ErrorCode.SMS_PROVIDER_FAILED _TENCENT_SETTINGS: dict[str, Any] = { "sms_provider": "tencent", "tencent_sms_secret_id": "secret-id", "tencent_sms_secret_key": "secret-key", "tencent_sms_sdk_app_id": "1400006666", "tencent_sms_sign_name": "美搭", "tencent_sms_template_id": "1234560", } class _StubClient: """替身 SDK 客户端：记录请求并返回固定 SendStatusSet。""" def __init__(self, code: str) -> None: self._code = code self.requests: list[Any] = [] async def SendSms(self, request: Any) -> Any: # noqa: N802 - 对齐 SDK 的 API 名 self.requests.append(request) status = SimpleNamespace(Code=self._code, Message="stub") return SimpleNamespace(SendStatusSet=[status], RequestId="stub-rid") class _StubbedTencentSender(TencentSmsSender): def __init__(self, code: str) -> None: super().__init__(_settings(**_TENCENT_SETTINGS)) self.stub = _StubClient(code) def _ensure_client(self) -> Any: return self.stub class _ExplodingTencentSender(TencentSmsSender): def __init__(self) -> None: super().__init__(_settings(**_TENCENT_SETTINGS)) def _ensure_client(self) -> Any: async def send(_: Any) -> Any: raise TimeoutError("connect timeout") return SimpleNamespace(SendSms=send) async def test_tencent_sender_builds_e164_request_and_accepts_ok() -> None: sender = _StubbedTencentSender("Ok") await sender.send_code(PHONE, MASKED, "123456") request = sender.stub.requests[0] assert request.PhoneNumberSet == ["+8613812341234"] assert request.TemplateParamSet == ["123456"] assert request.SmsSdkAppId == "1400006666" async def test_tencent_sender_rejected_status_maps_61000() -> None: sender = _StubbedTencentSender("LimitExceeded.PhoneNumberDailyLimit") with pytest.raises(AppError) as excinfo: await sender.send_code(PHONE, MASKED, "123456") assert excinfo.value.code == ErrorCode.SMS_PROVIDER_FAILED async def test_tencent_sender_network_error_maps_61000() -> None: with pytest.raises(AppError) as excinfo: await _ExplodingTencentSender().send_code(PHONE, MASKED, "123456") assert excinfo.value.code == ErrorCode.SMS_PROVIDER_FAILED