Co-Authored-By: Claude Code <noreply@anthropic.com>
34 KiB
单域名多租户(路径前缀路由)+ 上线收口项 实施计划
For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (
- [ ]) syntax for tracking.
Goal: 按 spec(docs/superpowers/specs/2026-10-19-shared-domain-routing-design.md)实现 /t/<slug>/ 路径前缀路由(与 Host 模式共存)、静态面前缀自推导、嵌入代码双形态,并收口登录限流 XFF 末段与 limiter 周期清扫。
Architecture: 纯 ASGI 中间件层加二级解析(路径前缀优先于 Host,剥离后下游路由零改动);widget.js/chat.html 从自身地址推导基础路径(Host 模式空串零变化);registry 增 slug 解析公开方法;登录限流 IP 取 XFF 末段(反代追加语义,不可伪造);lifespan 注册 300s 限流器清扫任务。
Tech Stack: FastAPI/Starlette 纯 ASGI、原生 JS(widget/chat 静态面,无构建)、Vue3+TS admin、pytest(PYTHONUTF8=1)/vitest。
执行环境注意(Windows):
- 仓库根
D:\refactor\all-kefu;后端测试PYTHONUTF8=1 .venv/Scripts/python -m pytest tests/test_xxx.py -q(全量PYTHONUTF8=1 .venv/Scripts/python -m pytest -q,基线 632+2skip) - 前端:
cd admin && pnpm test(基线 182);pnpm build;dts 行尾纪律照旧(git diff --ignore-cr-at-eol为空才 checkout 还原) - 分支:
feature/shared-domain(从 main 切出) - 本地路径模式验证技巧:路径前缀优先于 Host——
curl http://127.0.0.1:8018/t/jiancai/(Host 是 IP 未知也没关系,路径解析兜住) - 单进程铁律、日志/内存上限铁律(本计划 Task 7 为其进程内面)不变
拼接安全性四防线(用户 2026-10-19 要求,实现与测试双重钉住):
- 后端 slug 只查表不拼输出——
TENANT_SLUG_RE白名单([a-z0-9-])后仅作 dict lookup key;404/503 响应为静态常量串,任何用户输入(slug/路径段)都不进响应体 → 无服务端注入/反射面; - 前端 base 值白名单推导——chat.html 引导脚本的 base 来自
location.pathname.match(/^\/t\/[a-z0-9-]+\/$/)(match 命中即字符集白名单),widget.js 的 basePath 来自自身 script.src 定位切分;两者只做 URL 属性赋值(l.href/s.src/img.src/fetch/WebSocket),绝不进任何 innerHTML/字符串拼 HTML; - 既有 innerHTML 纪律不破——widget.js 现有插值全走
esc()五字符转义(已核:panel/invite/toggle 三处);本计划新增代码零 innerHTML(静态测试断言钉住); - 嵌入代码生成侧域名校验——
normalizeSharedDomain与域名形态生成前做主机名字符集校验(^[a-z0-9]([a-z0-9.-]*[a-z0-9])?$,拒绝引号/尖括号/空格/协议),非法返回 null/空串不产出<script src>串(防配置侧注入嵌入代码)。
Batch A:路由核心(后端)(✅ 2026-10-19 完成——e08a675+6fc43c1/9b97eaf/3862577;全任务过双审;.venv 丢失已重建 + conftest 钉 ADMIN_PASSWORD 封闭本地 .env)
Task 1: registry.resolve_by_slug(TDD)
Files:
-
Modify:
app/platform/registry.py(顶部正则常量 + 解析区新增方法) -
Test:
tests/test_registry.py(追加) -
Step 1: 写失败测试——test_registry.py 追加(沿用该文件既有夹具的建租户方式;看文件内现有用例怎么建 LoadedTenant/装配 registry,照抄同款):
class TestResolveBySlug:
"""路径前缀模式解析(spec 2026-10-19 §5)。"""
async def test_已知_slug_返回租户与首站点(self, ...既有夹具...):
rt, site = await registry.resolve_by_slug("jiancai")
assert rt.config.slug == "jiancai"
assert site.slug == rt.config.sites[0].slug # 首站点为默认
async def test_未知_slug_UnknownTenantError(self, ...):
with pytest.raises(UnknownTenantError):
await registry.resolve_by_slug("no-such")
async def test_停用租户_TenantDisabledError(self, ...):
# 夹具建 active 租户后置 status != "active" 再 resolve
...
async def test_空站点表按未知处理(self, ...):
# 夹具建 sites=[] 的租户 → UnknownTenantError
...
- Step 2: 跑测试确认失败(方法不存在)
- Step 3: 实现——registry.py:
顶部(import re 后):
# 租户 slug 形态(创建侧 tenant_routes 原 SLUG_RE 的单一事实源——路径前缀
# 解析与创建校验共用,永不漂移;spec 2026-10-19 §3.1)
TENANT_SLUG_RE = re.compile(r"^[a-z0-9-]+$")
解析区(resolve_site 之后):
async def resolve_by_slug(self, slug: str) -> tuple[TenantRuntime, Site]:
"""路径前缀模式(/t/<slug>/,spec 2026-10-19 §5):slug → (runtime,
首站点)。语义与 resolve_site 同款:未知 slug / 空站点表 →
UnknownTenantError;停用 → TenantDisabledError;懒建 engine、
reload 竞态取 _ensure_ready 返回值(与 resolve_site 同理)。"""
rt = self._by_slug.get(slug)
if rt is None:
raise UnknownTenantError(f"未知租户 slug: {slug!r}")
if rt.config.status != "active":
raise TenantDisabledError(f"租户已停用: {slug!r}")
rt = await self._ensure_ready(rt)
if not rt.config.sites:
# 创建向导恒建 main 站点,此分支为防御(spec §5)
raise UnknownTenantError(
f"租户 {slug!r} 无站点——路径模式无从选站点")
return rt, rt.config.sites[0]
- Step 4: 全文件回归——
PYTHONUTF8=1 .venv/Scripts/python -m pytest tests/test_registry.py -q全绿 - Step 5: Commit——
feat(registry): resolve_by_slug 路径模式解析 + TENANT_SLUG_RE 常量上提
Task 2: 中间件路径前缀解析(TDD)
Files:
-
Modify:
app/platform/host_middleware.py、app/platform/routes/tenant_routes.py(SLUG_RE 改引 registry) -
Test:
tests/test_host_routing.py(追加;该文件既有夹具/Client 用法照抄) -
Step 1: tenant_routes 的 SLUG_RE 换源——
SLUG_RE = re.compile(r"^[a-z0-9-]+$")(:65)改为from app.platform.registry import TENANT_SLUG_RE as SLUG_RE(删本地定义与现在不再用的import re——该文件 re 仅 :65 一处;grepSLUG_RE确认只有本文件用) -
Step 2: 写失败测试——test_host_routing.py 追加(照既有用例的 app/Client 夹具;Host 一律用未知域名验证"路径优先于 Host"):
class TestPathPrefixRouting:
"""路径前缀模式(spec 2026-10-19 §3)。Host 用未知值——路径命中证明优先级。"""
def test_聊天页_t_slug_根路径(self, client):
r = client.get("/t/jiancai/", headers={"Host": "unknown.example"})
assert r.status_code == 200
assert "chat" in r.text.lower()
def test_静态与公开API全链(self, client):
assert client.get("/t/jiancai/static/widget.js", headers={"Host": "unknown.example"}).status_code == 200
r = client.get("/t/jiancai/api/public/brand", headers={"Host": "unknown.example"})
assert r.status_code == 200 and r.json()["slug"] == "jiancai"
def test_orig_path留痕(self, client):
# 借 brand 端点无法直接看 state——用响应正常 + 行为断言替代:
# state 写入由后续 WS 用例与全链用例间接钉住;此处断言剥离后路由可达即证改写成立
...
def test_未知slug_404(self, client):
assert client.get("/t/no-such/x", headers={"Host": "unknown.example"}).status_code == 404
def test_停用租户_503(self, client):
... # 既有停用夹具同款
def test_无尾斜杠不剥离_走Host解析404(self, client):
# 注意只断言未知 Host 形态:/t/x 在 Host 模式下本就无路由(/ 才是聊天页),
# 若带已知 Host 断言 200 会诱导实现去剥离无斜杠形态——违反 spec §3.2
assert client.get("/t/jiancai", headers={"Host": "unknown.example"}).status_code == 404
# 已知 Host 的正常根路径(Host 模式对照,证明 404 不是 Host 解析坏了)
assert client.get("/", headers={"Host": "chat2.agent.hnxinyouhe.com"}).status_code == 200
def test_编码slug拒绝(self, client):
# %61uto 解码即 auto——raw 直判必须拒(spec §3.1)
assert client.get("/t/%61uto/", headers={"Host": "unknown.example"}).status_code == 404
def test_大写与点号拒绝(self, client):
# 注意不能用 /t/../:httpx 客户端侧归一点段,到不了中间件——用字面点号
assert client.get("/t/Jiancai/", headers={"Host": "unknown.example"}).status_code == 404
assert client.get("/t/jian.cai/", headers={"Host": "unknown.example"}).status_code == 404
def test_site参数覆盖在路径模式生效(self, client):
# spec §5:?site= 覆盖链路既有——路径模式下同样生效
r = client.get("/t/jiancai/api/public/brand?site=es", headers={"Host": "unknown.example"})
assert r.status_code == 200 # jiancai 的 es 站点(空域名站点)brand 通路
def test_orig_path与路径改写直接钉位(self):
# 直接构造 ASGI scope 断言改写(不经 httpx 归一化):capture app 记录
# scope["path"]/raw_path/state["orig_path"]——spec §3.2/§3.3 的直证
captured = {}
async def capture_app(scope, receive, send):
captured.update(path=scope["path"], raw_path=scope.get("raw_path"),
orig=scope.get("state", {}).get("orig_path"))
await PlainTextResponse("ok")(scope, receive, send)
# 用真实 app 的中间件包 capture_app(照 test_host_routing 既有注入手法;
# 若无现成手法,直接实例化 TenantHostMiddleware(capture_app) 并注入
# scope["app"].state.registry 测试夹具)
...
def test_平台豁免优先级不受影响(self, client):
assert client.get("/health", headers={"Host": "unknown.example"}).status_code == 200
# /t/x/api/platform/... 剥离后命中平台路由(by design,认证守护)
assert client.post("/t/jiancai/api/platform/auth/login", headers={"Host": "unknown.example"}).status_code in (401, 422, 429)
def test_WS路径模式握手(self, client):
with client.websocket_connect("/t/jiancai/ws/chat", headers={"Host": "unknown.example"}) as ws:
data = ws.receive_json()
assert "session_id" in data
def test_WS未知slug拒绝(self, client):
with pytest.raises(Exception):
with client.websocket_connect("/t/no-such/ws/chat", headers={"Host": "unknown.example"}):
pass
def test_Host模式回归(self, client):
# 既有用例全绿即回归;此处补一条:已知 Host + /t/其他租户/ → 路径优先(spec 优先级)
r = client.get("/t/jiancai/api/public/brand", headers={"Host": "chat2.agent.hnxinyouhe.com"})
assert r.status_code == 200 and r.json()["slug"] == "jiancai"
- Step 3: 跑测试确认失败
- Step 4: 实现中间件——host_middleware.py:
模块顶部 import 增加 import re、import urllib.parse、from app.platform.registry import TENANT_SLUG_RE(并入既有 registry import 块;registry.py 顶部同样需补 import re——该文件当前没有);类内新增:
# 路径前缀形态(spec 2026-10-19 §3):raw_path 直判——slug 段必须
# 未编码(%61uto 解码后即 auto,解码形态判不出编码;raw bytes 上
# [a-z0-9-] 天然排除 % 与大写)。仅带尾斜杠形态剥离(/t/x 不剥)。
# 安全注记:slug 只作 registry 的 dict lookup key,绝不插入任何
# 输出(404/503 均静态响应)——无注入面。
_PREFIX_RE = re.compile(rb"^/t/([a-z0-9-]+)(/.*)?$", re.ASCII)
__call__ 内,平台豁免 return 之后插入(registry 取值上提一次供路径/Host 两分支共用——删除原 Host 分支里的重复 getattr):
registry = getattr(scope["app"].state, "registry", None)
if registry is None:
raise RuntimeError(
"app.state.registry 未挂载——TenantHostMiddleware 需要 "
"lifespan(或测试注入缝)先加载 TenantRegistry")
# 路径前缀模式(spec 2026-10-19 §3):/t/<slug>/ 前缀优先于 Host 解析。
raw = scope.get("raw_path") or b""
m = self._PREFIX_RE.match(raw)
if m and m.group(2) is not None: # group(2) None = /t/x 无尾斜杠 → 不剥离
slug = m.group(1).decode("ascii")
try:
runtime, site = await registry.resolve_by_slug(slug)
except UnknownTenantError:
if scope["type"] == "websocket":
await _close_ws(receive, send, 1008, "unknown tenant")
else:
await PlainTextResponse("Not Found", status_code=404)(scope, receive, send)
return
except TenantDisabledError:
if scope["type"] == "websocket":
await _close_ws(receive, send, 1013, "service paused")
else:
await HTMLResponse(PAUSE_PAGE, status_code=503)(scope, receive, send)
return
state = scope.setdefault("state", {})
state["orig_path"] = scope.get("path") or "" # 排障留痕(§3.3)
# 剥离量按解码 path 对应段计:slug 段在 raw 上未编码 → 解码/原始
# 同长,前缀 len("/t/"+slug+"/") 在两形态一致
scope["path"] = scope["path"][len("/t/") + len(slug) + 1:] or "/"
scope["raw_path"] = urllib.parse.quote(
scope["path"], safe="/:@!$&'()*+,;=-._~").encode("ascii")
state["tenant"] = runtime
state["site"] = site
await self.app(scope, receive, send)
return
(其后原有 Host 解析逻辑不动——其 try 块起于 try: runtime, site = await registry.resolve_site(...)。)
- Step 5: 跑 Task 2 测试 + 全文件 + test_host_routing 既有用例全绿
- Step 6: Commit——
feat(middleware): /t/<slug>/ 路径前缀解析(与 Host 模式共存)
Task 3: 全链集成测试(chat REST + 限流隔离)
Files:
-
Test:
tests/test_host_routing.py(TestPathPrefixRouting 内追加) -
Step 1: 追加用例(照抄既有 chat REST 用例的 payload/头形态,test_channel_hygiene / test_chat_rest_ip 里有样板):
def test_chat_REST_路径模式建会话与消息(self, client):
# POST /t/<slug>/api/chat/session → 200 会话;同前缀 GET messages 通路
...
def test_限流隔离_同IP双租户独立计数(self, client, monkeypatch):
# 同 client(同 IP)分别打 /t/a/... 与 /t/b/... 的会话接口至各自阈值,
# 断言 a 触发 429 时 b 仍可用(键含 tenant——spec §11.2)
...
- Step 2: 跑绿(实现已在 Task 2 完成,此为验收钉位;若红按信号修中间件)
- Step 3: Commit——
test: 路径模式全链与限流隔离钉位
Batch B:静态面前缀自推导(✅ 0baa9ab/a1dc1ff;双模式页面逐字节一致验证)
Task 4: widget.js basePath(先写静态守卫测试)
Files:
-
Modify:
static/widget.js(:26 附近加 basePath;:132/:164-166/:683-685 四处端点构造加前缀) -
Create:
tests/test_widget_static.py(照 test_placeholder_static.py 的读文件断言风格) -
Step 1: 写失败测试:
"""widget.js 静态守卫(spec 2026-10-19 §4):脚本地址自推导 basePath,
三类端点构造(brand/avatar/WS)统一带前缀;Host 模式前缀为空串零变化。"""
def test_有_basepath_推导():
src = read("static/widget.js")
assert "indexOf(\"/static/widget.js\")" in src # 从 script.src 剥尾推导
def test_端点构造全部走_basePath():
src = read("static/widget.js")
# 不允许再出现裸 wsBase + "/api/ 或 wsHost + "/ws/(必须经 basePath)
assert 'wsBase + "/api/' not in src
assert 'wsHost + "/ws/' not in src
assert 'wsBase + basePath' in src
def test_默认头像路径带前缀():
src = read("static/widget.js")
assert 'basePath + DEFAULT_AVATAR_PATH' in src or "basePath + '" in src
- Step 2: RED 确认 → Step 3: 实现——widget.js :26 后加:
// spec 2026-10-19 §4: 脚本地址自携带租户路径(/t/<slug>/static/widget.js)
// → 剥掉 /static/widget.js 起的前缀得 basePath("" = Host 模式零变化)。
// script.src 永远指向本文件,indexOf 命中即安全;query(?v=)不影响 pathname。
var basePath = (function () {
var p = new URL(script.src).pathname;
var i = p.indexOf("/static/widget.js");
return i > 0 ? p.slice(0, i) : "";
})();
:132 brand → wsBase + basePath + "/api/public/brand";:163-166 avatar → wsBase + basePath + "/" + avatarAttr... / wsBase + basePath + (b.avatar...) / wsBase + basePath + DEFAULT_AVATAR_PATH;:683-685 WS → wWs = new WebSocket(wsHost + basePath + "/ws/chat")。
- Step 4: GREEN + 全量静态测试(
PYTHONUTF8=1 .venv/Scripts/python -m pytest tests/test_widget_static.py tests/test_placeholder_static.py tests/test_theme_static.py tests/test_invite_static.py -q——后三个是回归,widget.js 改动不能碰主题块) - Step 5: Commit——
feat(widget): 脚本地址自推导 basePath(路径模式端点前缀)
Task 5: chat.html 引导脚本 + chat.js BASE_PATH
Files:
-
Modify:
static/chat.html(head 顶部内联引导;删静态 link/script 标签改为动态装载;avatar img 去 src 改 DOMContentLoaded 设置) -
Modify:
static/js/chat.js(:5 brand fetch、:132 WS、文件内其余/api/、/static/、/ws/绝对引用统一加(window.__BASE_PATH__ || "")前缀——动手前 grep 全列) -
Create:
tests/test_chat_static.py -
Step 1: 写失败测试:
"""chat 静态面前缀守卫(spec 2026-10-19 §4)。"""
def test_引导脚本存在且定义_BASE_PATH():
src = read("static/chat.html")
assert "__BASE_PATH__" in src
# 代码锚(非注释):base 推导必须走正则白名单 match
assert "location.pathname.match(" in src
def test_不再有静态绝对引用标签():
src = read("static/chat.html")
assert 'href="/static/' not in src
assert 'src="/static/js/' not in src
assert 'src="/api/public/avatar"' not in src
def test_引导脚本零innerHTML_拼接安全防线():
# 安全防线 §2/§3:base 只做 URL 属性赋值,绝不拼 HTML
boot = src_between(src, "<script>", "</script>") # 引导块(首个内联 script)
assert "innerHTML" not in boot
def test_chatjs_端点全部走_BASE_PATH():
src = read("static/js/chat.js")
assert '__BASE_PATH__' in src
# 裸绝对端点不允许(fetch("/api/... 或 + "/ws/ 直接拼接形态)
assert 'fetch("/api/' not in src
- Step 2: RED → Step 3: 实现——chat.html
<head>内 charset/title 之后、原 link/script 位置替换为:
<script>
(function () {
/* spec 2026-10-19 §4:路径前缀模式(/t/<slug>/)按当前地址推导基础路径;
Host 模式(根路径 "/")推导为空串——全部引用零变化。 */
var m = location.pathname.match(/^\/t\/[a-z0-9-]+\/$/);
var base = m ? location.pathname.slice(0, -1) : "";
window.__BASE_PATH__ = base;
function css(href) {
var l = document.createElement("link");
l.rel = "stylesheet"; l.href = base + href;
document.head.appendChild(l);
}
function js(src) {
var s = document.createElement("script");
s.src = base + src; s.async = false; /* 保序:theme 先于 chat */
document.head.appendChild(s);
}
css("/static/css/style.css");
document.addEventListener("DOMContentLoaded", function () {
/* js 装载放 DOMContentLoaded:head 动态插入的 async=false 脚本与
解析进度无形式化顺序保证,且 chat.js 顶层有同步 DOM 查询——
DOM 就绪后装载与原 body 末尾形态等价且保序(theme 先于 chat) */
js("/static/js/theme.js");
js("/static/js/chat.js");
var img = document.querySelector(".avatar-img");
if (img) img.src = base + "/api/public/avatar";
});
})();
</script>
(body 末尾原 <script src="/static/js/chat.js"> 删除;<img ... src="/api/public/avatar"> 的 src 属性删除只留空 alt 占位。)chat.js 内所有绝对端点改 var B = window.__BASE_PATH__ || ""; 后 B + "/api/public/brand" 等(含 WS 构造行)。
- Step 4: GREEN + 手动双模式目检——
PYTHONUTF8=1 .venv/Scripts/python -m uvicorn app.main:app --port 8018 --workers 1后 curl 两种形态(Host 模式需 hosts/Host 头;路径模式curl http://127.0.0.1:8018/t/jiancai/返回页含引导脚本) - Step 5: Commit——
feat(chat): 页面级 basePath 引导(路径模式静态面)
Batch C:上线收口项(✅ 4f58398/f938118;XFF 末段 RED 判别实证 + 清扫窗沿互补钉位)
Task 6: 登录限流 XFF 末段(TDD)
Files:
-
Modify:
app/platform/routes/auth_routes.py:48-49(_client_ip单点) -
Test:
tests/test_auth.py(追加;该文件既有 lockout 用例照抄夹具) -
Step 1: 写失败测试:
class TestClientIpXffLast:
"""spec 2026-10-19 §6:限流 IP 取 XFF 末段(反代追加语义,不可伪造)。"""
def test_末段入键_伪造首段轮换不绕过(self, client):
# 伪造首段轮换 + 固定真末段 → 同键计数,第 5 次失败后第 6 次 429
for i in range(5):
client.post("/api/platform/auth/login",
headers={"X-Forwarded-For": f"6.6.6.{i}, 1.2.3.4"},
json={"username": "admin", "password": "wrong"})
r = client.post("/api/platform/auth/login",
headers={"X-Forwarded-For": "9.9.9.9, 1.2.3.4"},
json={"username": "admin", "password": "wrong"})
assert r.status_code == 429
def test_末段不同即独立键_判别用例(self, client):
# 这是 RED 判别用例:现状(无 XFF 解析)所有请求同键(client.host
# 恒 testclient)→ 第二段也会 429;新行为下 8.8.8.8 是独立键 → 401
for i in range(5):
client.post("/api/platform/auth/login",
headers={"X-Forwarded-For": f"6.6.6.{i}, 1.2.3.4"},
json={"username": "admin", "password": "wrong"})
r = client.post("/api/platform/auth/login",
headers={"X-Forwarded-For": "7.7.7.7, 8.8.8.8"},
json={"username": "admin", "password": "wrong"})
assert r.status_code == 401 # 独立键:未锁,密码错 → 401(现状此断言为 429 → RED)
def test_无XFF回退client_host(self, ...): ...
def test_空串XFF回退(self, client):
r = client.post(..., headers={"X-Forwarded-For": " "}, ...)
assert r.status_code in (200, 401) # 正常走通,键为 client.host
def test_单段直连形态(self, client):
# XFF "1.2.3.4"(无反代追加)→ 末段即唯一段,入键
...
(注意夹具间 login_limiter 状态隔离——既有 lockout 用例已有重置手法,照抄。)
- Step 2: RED → Step 3: 实现——auth_routes.py 替换
_client_ip:
def _client_ip(request: Request) -> str:
"""限流键 IP(spec 2026-10-19 §6):XFF **末段**。
反代模板 proxy_add_x_forwarded_for 语义 = 客户端自带 XFF 在前、反代把
真实来源 $remote_addr 追加在尾——末段不可伪造(单跳拓扑);取首段会被
轮换伪造绕过限流(比现状更糟)。将来加 CDN 时末段退化为 CDN IP →
键退化为按用户名 = 现状,故障安全。无 XFF/空串/空白回退 client.host。
与引擎 _server_ip(取首段,"CDN 在前"拓扑假设,数据质量用途)刻意
不对称——两者用途与威胁模型不同,不许互相"统一"(spec §6 双向注记)。
"""
xff = request.headers.get("x-forwarded-for", "")
if xff.strip():
last = xff.split(",")[-1].strip()
if last:
return last
return request.client.host if request.client else ""
- Step 4: GREEN + test_auth 全文件回归 → Step 5: Commit——
feat(auth): 登录限流 IP 取 XFF 末段(防首段轮换伪造)
Task 7: limiter 周期清扫(TDD)
Files:
-
Modify:
app/platform/auth.py(LoginLimiter.sweep)、app/engine/services/rate_limit.py(sweep)、app/main.py(lifespan 注册任务) -
Test:
tests/test_auth.py或新建tests/test_limiter_sweep.py(含 lifespan 启停断言——用既有 app 夹具的 with 生命周期手法) -
Step 1: 写失败测试:
class TestLimiterSweep:
"""spec 2026-10-19 §7:过期键周期清扫(日志/内存上限铁律)。"""
def test_LoginLimiter_sweep_清过期保活跃(self):
from app.platform.auth import LoginLimiter, login_limiter # 实例名照实文件
limiter = LoginLimiter(window_seconds=900.0, max_fails=5) # 参数名照 __init__ 实参
limiter.record_fail("u", "1.1.1.1")
limiter.record_fail("u", "2.2.2.2")
# 时间推进手法照 test_auth.py 既有 lockout 用例(:66-69):直接改
# _fails 内时间戳(record_fail 不收 now 参数)
limiter._fails[("u", "1.1.1.1")] = [0.0] # 远古 → 待清
n = limiter.sweep(now=time.time())
assert n == 1 and ("u", "1.1.1.1") not in limiter._fails
assert ("u", "2.2.2.2") in limiter._fails
def test_engine_rate_limit_sweep同款(self):
# app.engine.services.rate_limit 是模块级 check() + _buckets(非类)
from app.engine.services import rate_limit
rate_limit._buckets[("a", "1.1.1.1", "")] = [0.0]
rate_limit._buckets[("b", "2.2.2.2", "")] = [time.time()]
n = rate_limit.sweep(now=time.time())
assert n == 1 and ("a", "1.1.1.1", "") not in rate_limit._buckets
def test_lifespan注册与取消(self):
# monkeypatch asyncio.create_task 记录调用(返回真实 task 语义的
# stub);TestClient with 生命周期 → 断言 sweep loop 任务创建过;
# 退出后断言 cancel 被调(不真等 300s——loop 里 sleep 在前)
...
- Step 2: RED → Step 3: 实现——
LoginLimiter加sweep(self, now=None) -> int(清"全部时间戳均过期"的键);enginerate_limit.py加模块级sweep(now=None) -> int(同语义,清_buckets);main.py lifespan(:108 起):import asyncio后在yield(:199)前注册任务、yield后既有 finally/清理段里cancel():
# 铁律(2026-10-19)进程内面:登录/聊天限流器过期键周期清扫(§7)
async def _limiter_sweep_loop() -> None:
while True:
await asyncio.sleep(300)
login_limiter.sweep()
chat_rate_limiter.sweep() # 实例名以 chat_rest/runtime 实际引用为准
_sweep_task = asyncio.create_task(_limiter_sweep_loop())
yield 之后取消(_sweep_task.cancel(),与文件内既有启停代码同段)。限流器实例定位:grep -rn "login_limiter\|RateLimiter(" app/ 找模块级单例(auth 侧在 platform,chat 侧在 engine services/chat_rest——以实际为准,sweep 加在实例所在模块)。
- Step 4: GREEN + 全量回归 → Step 5: Commit——
feat(limits): 限流器过期键周期清扫(内存上限铁律)
Batch D:管理面与部署工件(✅ 5813fc9/e1eae00/9315dba + 终审 Merge-ready 0C/0I/3M——M1 空站点引擎懒建次序/M3 data-site 前端校验注记 已记档;M2 陈旧注释已修;走查实测:路径模式 E2E 对话/嵌入双形态切换生成)
Task 8: 共享域名端点 + 设置键 + 嵌入代码双形态
Files:
-
Modify:
app/platform/routes/config_routes.py(settings_router 所在文件——新增 GET /api/platform/widget-domain:platform_user依赖(两种角色可读),读 settings_store 快照的shared_widget_domain,返回{"widget_domain": str | None};测试加在 tests/test_config_api.py 既有 SETTINGS 区(~:394)) -
Modify:
admin/src/api/settings.ts(settings API 模块)加getWidgetDomain() -
Modify:
admin/src/views/settings/platform/model.ts(推荐键表加shared_widget_domain:文本键、说明「共享挂件域名(嵌入代码 /t// 形态的基地址,不带 https:// 前缀)」——注意 SETTING_FIELD_GROUPS 的组 id 联合类型需扩一个组(如 'site'),表驱动渲染即吃上) -
Modify:
admin/src/views/config/center/model.ts(新增纯函数)+index.vue嵌入卡(形态切换 UI) -
Test:
admin/src/views/config/center/__tests__/model.test.ts(追加) -
Step 1: model.ts 纯函数 + 失败测试:
/** 共享域名规范化(安全防线 §4):去协议/尾斜杠/转小写/去空白后须过
* 主机名字符集 ^[a-z0-9]([a-z0-9.-]*[a-z0-9])?$ ——拒绝引号/尖括号/空格等
* (防配置侧把恶意字符注入生成的 <script src> 串);空或非法 → null */
export function normalizeSharedDomain(v: string): string | null
/** 嵌入代码双形态(spec §5):
* domain 形态 = <script src="https://<域名>/static/widget.js"[ data-site]></script>
* shared 形态 = <script src="https://<共享域名>/t/<slug>/static/widget.js"[ data-site]></script>
* 生成前域名须过同款主机名校验(tenantSlug 由创建侧 SLUG_RE 保证字符集);
* 不可用(缺域名/共享域名/校验不过)返回 '' */
export type EmbedForm = 'domain' | 'shared'
export function buildEmbedCode(form: EmbedForm, opts: {
domain: string; sharedDomain: string | null; tenantSlug: string; site: string | null
}): string
vitest:normalize 各形态 + 恶意字符样例(kefu.com" onload="x、<script>、带空格 → null)、双形态代码串、缺料/非法域名空串、data-site 可选。
- Step 2: RED → 实现 model.ts → GREEN
- Step 3: 后端端点(小:读 settings_store 快照返回;测试加在既有 settings 测试文件——两种角色 200 / 未认证 401)
- Step 4: index.vue 嵌入卡改造——加载数据时请求 widget-domain;形态 radio(域名形态/共享域名形态——按可用性禁用);embedCode computed 换用 buildEmbedCode;无域名租户仅共享形态;都无显示引导文案(去平台设置配共享域名)。设置页推荐键渲染无需改(表驱动)
- Step 5:
pnpm test+pnpm build全绿;dts 纪律 - Step 6: Commit——
feat(admin): 嵌入代码双形态 + shared_widget_domain 平台键
Task 9: 创建向导域名可选
Files:
-
Modify:
admin/src/views/tenants/list/index.vue(向导步骤②「绑定域名」改为可跳过——校验放宽、步骤文案「可留空 = 使用共享域名路径模式」;创建摘要显示「未绑定域名(路径模式)」。注:该目录无 model.ts,向导校验是 index.vue 内联逻辑——纯模板/校验放宽,无 vitest 新用例,走手动走查记录) -
Step 1: 定位向导校验(grep 域名/必填/required)→ 放宽 + 文案
-
Step 2: 走查——dev server 建一个零域名租户 → 详情正常 → 配置中心品牌页出共享形态嵌入代码(依赖 Task 8 的 shared_widget_domain 已配)
-
Step 3:
pnpm test+ build → Commit——feat(admin): 创建向导域名可跳过(纯路径模式租户)
Task 10: 部署工件
Files:
-
Modify:
deploy/openresty-all-kefu.conf.template(增共享域名 server 块:同 upstream、WS 升级头单套、超时 3600s;XFF 头已有$proxy_add_x_forwarded_for确认覆盖新块并注记「登录限流取末段依赖此语义,勿改成 $remote_addr 覆盖式」) -
Modify:
docs/deploy-production.md(增「共享域名模式」一节:/t/<slug>/地址形态、无尾斜杠 404 语义、shared_widget_domain 设置、新租户零 DNS 接入流程、嵌入代码形态示例) -
Step 1: 模板增块 + 手册增节 → Step 2: Commit——
docs(deploy): 共享域名模式部署工件
Task 11: 全局回归 + 走查收口
- Step 1: 后端全量——
PYTHONUTF8=1 .venv/Scripts/python -m pytest -q(≥ 基线 632+2skip + 新增) - Step 2: 前端全量 + 构建——
cd admin && pnpm test && pnpm build(≥182 + 新增;dts 纪律) - Step 3: 双模式手动走查——本地 uvicorn:Host 模式(hosts 指向 chat2...)聊天页/挂件原样;路径模式
curl http://127.0.0.1:8018/t/jiancai/+ 浏览器开/t/jiancai/全链(页面样式/WS 对话/头像/品牌色);嵌入代码两形态各贴到 widget-test.html 验证 - Step 4: 铁律检查项——compose
mem_limit/日志封顶在位;清扫任务在 lifespan 启停(Task 7 测试已钉) - Step 5: 收尾——工作区干净、临时产物清空、
git commit(如有杂项)
依赖与并行性
- Task 1 → 2 → 3 顺序;Task 4/5 依赖 2(联调)但代码可先行(测试独立)——建议仍顺序
- Task 6/7 独立于 A/B(可穿插);Task 8 依赖 1-5 完成(嵌入代码端到端验证);Task 9 依赖 8;Task 10 独立;Task 11 收口
- 分支:
feature/shared-domain