feat: 测试端点限流(每租户5次/15分钟)

- 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 <noreply@anthropic.com>
This commit is contained in:
浅念
2026-10-14 10:17:52 +08:00
parent d734ed6d2a
commit fff4f27e98
4 changed files with 171 additions and 2 deletions
+79 -2
View File
@@ -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)}