From fff4f27e98035aa6b7742361bff74d3247be9094 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B5=85=E5=BF=B5?= Date: Wed, 14 Oct 2026 10:17:52 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=B5=8B=E8=AF=95=E7=AB=AF=E7=82=B9?= =?UTF-8?q?=E9=99=90=E6=B5=81=EF=BC=88=E6=AF=8F=E7=A7=9F=E6=88=B75?= =?UTF-8?q?=E6=AC=A1/15=E5=88=86=E9=92=9F=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - email/test 与 push/test 复用 LoginLimiter 滑动窗模式: (kind, tenant_slug) 键每租户 15 分钟 5 次真实出站,超限 429——guard 在端点最前(鉴权后), 被限流的租户连 400 预检也不触发;record 只落在通过全部预检真正出站 的那次(校验失败/未配置不消耗预算,两端口预算按 kind 互不挤占) - conftest autouse 复位限流器(跨测试隔离);新增三测: 第 6 次 429+ 不同租户独立+email/push 互不挤占、校验失败不耗预算 - 回归: 后端 630 passed（627+3） Co-Authored-By: Claude Code --- app/platform/routes/config_routes.py | 81 +++++++++++++++++++++++++++- tests/conftest.py | 11 ++++ tests/test_email_test_endpoint.py | 50 +++++++++++++++++ tests/test_push_test_endpoint.py | 31 +++++++++++ 4 files changed, 171 insertions(+), 2 deletions(-) diff --git a/app/platform/routes/config_routes.py b/app/platform/routes/config_routes.py index ebc869a..da70756 100644 --- a/app/platform/routes/config_routes.py +++ b/app/platform/routes/config_routes.py @@ -42,6 +42,8 @@ email.smtp.password 及平台设置的密钥名值,所有携带 config/value 的 import copy import json import socket +import time +from collections import defaultdict import httpx from fastapi import APIRouter, Depends, HTTPException, Request @@ -78,6 +80,75 @@ settings_router = APIRouter( prefix="/api/platform/settings", tags=["platform-settings"]) +# ---- 测试端点限流(Phase 4 Task 5B) ------------------------------------------------- + +TEST_WINDOW_SECONDS = 15 * 60 +TEST_MAX_CALLS = 5 + + +class SlidingWindowCounter: + """(kind, tenant_slug) 键滑动窗计数: 窗内第 5 次发起后拒绝后续,窗过 + 恢复。LoginLimiter 同款风格/单进程内存态,但语义是"计数一切发起的 + 调用"——email/push 测试端点每次受理都可能触发出站,不分成败(登录 + 限流只计失败,这里没有"成功即清零"的正路)。计数只落在通过全部 + 预检(参数/租户/配置可用)真正出站的那次——校验失败的调用不消耗 + 预算,但超限后连预检一并拒绝(allow 守卫在端点最前)。""" + + def __init__( + self, + window_seconds: float = TEST_WINDOW_SECONDS, + max_calls: int = TEST_MAX_CALLS) -> None: + self.window_seconds = window_seconds + self.max_calls = max_calls + self._calls: dict[tuple[str, str], list[float]] = defaultdict(list) + + def allow(self, key: tuple[str, str]) -> bool: + """窗内计数是否仍低于上限(不记录——记录走 record)。""" + self._prune(key, time.time()) + return len(self._calls.get(key, ())) < self.max_calls + + def record(self, key: tuple[str, str]) -> None: + now = time.time() + self._prune(key, now) + self._calls[key].append(now) + + def _prune(self, key: tuple[str, str], now: float) -> None: + calls = self._calls.get(key) + if calls is None: + return + kept = [t for t in calls if t > now - self.window_seconds] + if kept: + self._calls[key] = kept + else: + self._calls.pop(key, None) + + +# 模块级单例: email/push 两测试端点共用(键含 kind,两预算互不挤占) +_test_limiter = SlidingWindowCounter() + + +def _reset_test_limiter_for_test() -> None: + """单例复位(测试隔离),conftest autouse 前后各跑一次(同 auth 纪律)。""" + _test_limiter._calls.clear() + _test_limiter.window_seconds = TEST_WINDOW_SECONDS + _test_limiter.max_calls = TEST_MAX_CALLS + + +def _test_rate_guard(kind: str, slug: str) -> None: + """测试端点的统一限流守卫: 超限 → 429(先于一切业务逻辑——被限流的 + 租户连 400 预检也不触发,更不许出站)。""" + if not _test_limiter.allow((kind, slug)): + raise HTTPException( + status_code=429, + detail=f"测试端点调用过于频繁(每租户 15 分钟 {TEST_MAX_CALLS} 次)" + "——请稍后再试") + + +def _record_test_call(kind: str, slug: str) -> None: + """通过全部预检、即将出站的那次调用计数。""" + _test_limiter.record((kind, slug)) + + # ---- 助手 ----------------------------------------------------------------------- @@ -513,7 +584,9 @@ async def test_email(slug: str, request: Request, 优先/空槽回退 env SMTP_*)真发一封到 body.to——主题/正文契约见 email_service.send_test_email。SMTP 未配置(回退后 host/user 仍空) → 400;发送失败不 5xx,{sent: false, error} 如实返回。不落库不 bump - 版本;密钥绝不进响应。""" + 版本;密钥绝不进响应。 + 限流(Phase 4 Task 5B): 每租户 15 分钟 5 次真实出站,超限 429。""" + _test_rate_guard("email", slug) body = await _json_body(request) unknown = set(body) - _EMAIL_TEST_KEYS if unknown: @@ -536,6 +609,7 @@ async def test_email(slug: str, request: Request, status_code=400, detail="SMTP 未配置(服务器/账号在租户配置与环境回退后均为空)" "——请先在「配置中心 → 凭据」保存邮件 SMTP") + _record_test_call("email", slug) sent, error = await send_test_email(to.strip(), tenant.name, snapshot) return {"slug": slug, "sent": sent, "error": error} @@ -590,7 +664,9 @@ async def test_push(slug: str, request: Request, """推送连通性探测: 只验 URL 可达 + 鉴权通过,不发真数据(空 messages 批量,见 _probe_push)。URL 回退后仍空 → 400;其余如实返回——不可达 (含 dns/timeout/connect 原因分类)/可达但鉴权失败(401/403)/正常(2xx); - 其余状态码可达但鉴权不定(auth_ok=null)。不落库不 bump 版本。""" + 其余状态码可达但鉴权不定(auth_ok=null)。不落库不 bump 版本。 + 限流(Phase 4 Task 5B): 每租户 15 分钟 5 次探测,超限 429。""" + _test_rate_guard("push", slug) engine = request.app.state.platform_engine async with AsyncSession(engine, expire_on_commit=False) as session: tenant = await _get_tenant_or_404(session, slug) @@ -601,6 +677,7 @@ async def test_push(slug: str, request: Request, status_code=400, detail="推送 URL 未配置(租户配置与环境回退后均为空)" "——请先在「配置中心 → 凭据」保存推送 URL") + _record_test_call("push", slug) return {"slug": slug, **await _probe_push(cfg)} diff --git a/tests/conftest.py b/tests/conftest.py index 0ab16e1..fd9dbe6 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -97,6 +97,17 @@ def _reset_platform_settings_snapshot(): _reset_for_test() +@pytest.fixture(autouse=True) +def _reset_test_endpoint_limiter(): + """Phase 4 Task 5B: email/push 测试端点限流器前后复位——同测试内 + 连发会被窗限流(这是被测行为),跨测试不隔离则前一脚打满的预算顶掉 + 后脚的断言;注入的窗长/上限一并还原默认。""" + from app.platform.routes.config_routes import _reset_test_limiter_for_test + _reset_test_limiter_for_test() + yield + _reset_test_limiter_for_test() + + # ---- Phase 2 Task 10: FakeAI 公共确定性桩(清理单 #1) -------------------------- # 收敛自四份拷贝(test_lead_email_trigger / test_cc_integration / # test_channel_hygiene / test_lead_push_wiring,接口取超集: 前两份全量, diff --git a/tests/test_email_test_endpoint.py b/tests/test_email_test_endpoint.py index f709dd5..8820a21 100644 --- a/tests/test_email_test_endpoint.py +++ b/tests/test_email_test_endpoint.py @@ -157,6 +157,56 @@ def test_email_test_unknown_tenant_404(super_client, monkeypatch): assert r.status_code == 404 +def test_email_test_rate_limited_after_5(super_client, monkeypatch): + """限流(Phase 4 Task 5B): 每租户 15 分钟 5 次真实出站——前 5 次 200, + 第 6 次 429;不同租户预算独立;400 校验失败/未配置不消耗预算。""" + _create(super_client, slug="demo") + _create(super_client, slug="other", profile="auto") + assert _put_email_group(super_client, "demo", SMTP_GROUP).status_code == 200 + sends = [] + + async def counting_send(msg, **kwargs): + sends.append(msg) + return msg + + monkeypatch.setattr(email_service.aiosmtplib, "send", counting_send) + + for i in range(5): + r = _post_test(super_client, "demo", {"to": "checker@example.com"}) + assert r.status_code == 200, f"第 {i + 1} 次不应限流: {r.text}" + # 第 6 次 → 429,且不再出站(真实发送仍 5 次) + r = _post_test(super_client, "demo", {"to": "checker@example.com"}) + assert r.status_code == 429 + assert "频繁" in r.json()["detail"] + assert len(sends) == 5 + # 429 优先于参数校验: 超限后坏 body 也 429(不许预检出站侧通道) + r = _post_test(super_client, "demo", {"to": "not-an-email"}) + assert r.status_code == 429 + + # 不同租户独立预算: other(未配置 SMTP)不受 demo 拉满影响 + _clear_smtp_env(monkeypatch) + r = _post_test(super_client, "other", {"to": "a@b.com"}) + assert r.status_code == 400 # 未配置业务分支照常,非 429 + + +def test_email_test_validation_failures_do_not_consume_quota( + super_client, monkeypatch): + """校验失败(to 非法/未配置 400)不计数——只有真实出站消耗预算 + (管理操作误点不该被坏请求打满)。""" + _clear_smtp_env(monkeypatch) + _create(super_client, slug="demo") + for _ in range(7): + assert _post_test( + super_client, "demo", {"to": "bad"}).status_code == 400 + assert _post_test( + super_client, "demo", {"to": "a@b.com"}).status_code == 400 + # 15 分钟预算原封: 配好 SMTP 后第 1 次仍放行 + assert _put_email_group(super_client, "demo", SMTP_GROUP).status_code == 200 + _capture_smtp(monkeypatch) + assert _post_test( + super_client, "demo", {"to": "a@b.com"}).status_code == 200 + + def test_email_test_auth_matrix(super_client, monkeypatch): """require_tenant(slug): 本租户管理员放行(400 是业务分支,鉴权先于 业务);跨租户 403;未认证 401。""" diff --git a/tests/test_push_test_endpoint.py b/tests/test_push_test_endpoint.py index 2019a51..62d8c94 100644 --- a/tests/test_push_test_endpoint.py +++ b/tests/test_push_test_endpoint.py @@ -225,6 +225,37 @@ def test_push_test_unknown_tenant_404(super_client, monkeypatch): assert r.status_code == 404 +def test_push_test_rate_limited_after_5(super_client, monkeypatch): + """限流(Phase 4 Task 5B): 每租户 15 分钟 5 次探测——前 5 次 200, + 第 6 次 429;email 与 push 预算互不挤占;不同租户独立。""" + _create(super_client, slug="demo") + _create(super_client, slug="other", profile="auto") + assert _put_push_group(super_client, "demo", PUSH_GROUP).status_code == 200 + calls = _install_fake_client(monkeypatch, _ok) + + for i in range(5): + r = _post_test(super_client, "demo") + assert r.status_code == 200, f"第 {i + 1} 次不应限流: {r.text}" + r = _post_test(super_client, "demo") + assert r.status_code == 429 + assert "频繁" in r.json()["detail"] + assert len(_post_calls(calls)) == 5 # 第 6 次不再出站 + + # email 预算独立(键含 kind): 同租户 email/test 不受 push 拉满影响 + # (不配置 SMTP → 400 业务分支,证明 guard 放行) + monkeypatch.setattr(settings, "SMTP_HOST", "") + monkeypatch.setattr(settings, "SMTP_USER", "") + r = super_client.post(f"/api/platform/tenants/demo/email/test", + headers=_auth(super_client), json={"to": "a@b.com"}) + assert r.status_code == 400 + + # 不同租户独立预算 + assert _put_push_group(super_client, "other", { + **PUSH_GROUP, "prefix": "other-p"}).status_code == 200 + r = _post_test(super_client, "other") + assert r.status_code == 200 + + def test_push_test_auth_matrix(super_client, monkeypatch): """require_tenant(slug): 本租户管理员放行;跨租户 403;未认证 401。""" _create(super_client, slug="jc2", admin="jc2-admin", password="jc2-pass-1")