feat(middleware): /t/<slug>/ 路径前缀解析(与 Host 模式共存)

- TenantHostMiddleware 增路径前缀分支:raw_path 直判 slug 段(未编码
  [a-z0-9-],%61uto/大写/点号不匹配落 Host 解析);仅带尾斜杠形态剥离;
  path 定长切割(保留前缀尾斜杠为剩余路径领头斜杠),raw_path 按解码
  path 重编码重算(非长度截断),orig_path 留痕;未知 404/ws 1008、
  停用 503/ws 1013 与 Host 模式同语义
- _PREFIX_RE 字符类取自 registry.TENANT_SLUG_RE(掐 ^$ 锚)单一事实源;
  tenant_routes 的 SLUG_RE 换源同常量别名 import(删本地定义与 import re)
- 评审折入:registry 空 sites 分支注释纠正为生产可达(DB 源全站点停用
  合法状态),test_registry 用例 docstring 同步;conftest ADMIN_PASSWORD
  注释补硬编码登录用例警示
- 新增 TestPathPrefixRouting 13 用例(全链/拒绝族/停用/site 覆盖/豁免
  优先/direct-ASGI 改写钉位/WS 双例/路径优先)

Co-Authored-By: Claude Code <noreply@anthropic.com>
This commit is contained in:
浅念
2026-10-19 17:27:01 +08:00
parent 6fc43c1f1c
commit 9b97eafa1c
6 changed files with 248 additions and 10 deletions
+2 -1
View File
@@ -17,7 +17,8 @@ from app.engine.models import Base
# 测试环境封闭(2026-10-19 修复):Settings(BaseSettings) 会读仓库 .env——
# 开发机 .env 的 ADMIN_PASSWORD(如 Docker 测试环境的强密码)会盖过 lifespan
# 引导的出厂值 admin123,测试登录全员 401。真实环境变量优先级本就高于
# .env,这里把测试期望值钉进 os.environ(已显式设置则不覆盖,CI 可另行注入)。
# .env,这里把测试期望值钉进 os.environ(已显式设置则不覆盖,CI 可另行
# 注入;注入非 admin123 值会破坏 ~20 个硬编码登录用例)。
os.environ.setdefault("ADMIN_PASSWORD", "admin123")
# Task 5b 装配切换: app 不再有全局 engine/async_session——旧 conftest 里
+170
View File
@@ -7,12 +7,14 @@
import json
import shutil
from types import SimpleNamespace
import pytest
from starlette.websockets import WebSocketDisconnect
from app.main import app
from app.platform.host_middleware import TenantHostMiddleware
from app.platform.registry import TenantRegistry
from tests.conftest import (
AUTO_HOST,
JIANCAI_HOST,
@@ -150,3 +152,171 @@ def test_ws_routes_to_correct_tenant_db(app_client, tmp_path):
assert r_jc.status_code == 200
assert r_jc.json()["is_new"] is True
assert r_jc.json()["session_id"] != sid
# ---- 路径前缀模式(/t/<slug>/,spec 2026-10-19 §3) ----------------------------
class TestPathPrefixRouting:
"""路径前缀与 Host 模式共存:优先级 平台豁免 → 路径前缀 → Host 解析。
Host 用未知值(NOPE_HOST)——路径命中即证明解析先于 Host;反向用已知
AUTO_HOST(Host 说是 auto 租户)+ /t/jiancai/ 断言路径优先(spec §2)。"""
def test_聊天页_t_slug_根路径(self, app_client):
r = app_client.get("/t/jiancai/", headers={"host": NOPE_HOST})
assert r.status_code == 200
assert "chat" in r.text.lower()
def test_静态与公开API全链(self, app_client):
r = app_client.get("/t/jiancai/static/widget.js",
headers={"host": NOPE_HOST})
assert r.status_code == 200
r = app_client.get("/t/jiancai/api/public/brand",
headers={"host": NOPE_HOST})
assert r.status_code == 200
assert r.json()["slug"] == "jiancai"
def test_无尾斜杠不剥离_走Host解析404(self, app_client):
# 只断言未知 Host 形态:/t/x 在 Host 模式下本就无路由(/ 才是聊天页),
# 若带已知 Host 断言 200 会诱导实现去剥离无斜杠形态——违反 spec §3.2
r = app_client.get("/t/jiancai", headers={"host": NOPE_HOST})
assert r.status_code == 404
# 已知 Host 的正常根路径(Host 模式对照,不受路径模式影响)
assert app_client.get("/", headers={"host": JIANCAI_HOST}).status_code == 200
def test_编码slug拒绝(self, app_client):
# %61uto 解码后即合法 slug auto——解码形态判不出编码,raw_path 直判
# 为准(spec §3.1);不匹配 → 落 Host 解析(未知 Host)→ 404
r = app_client.get("/t/%61uto/", headers={"host": NOPE_HOST})
assert r.status_code == 404
def test_大写与点号拒绝(self, app_client):
# 不能用 /t/../:httpx 客户端侧归一点段到不了中间件——用字面点号
assert app_client.get(
"/t/Jiancai/", headers={"host": NOPE_HOST}).status_code == 404
assert app_client.get(
"/t/jian.cai/", headers={"host": NOPE_HOST}).status_code == 404
def test_未知slug_404(self, app_client):
r = app_client.get("/t/no-such/x", headers={"host": NOPE_HOST})
assert r.status_code == 404
def test_停用租户_503(self, disabled_client):
# 路径模式同样吃 TenantDisabledError → 503 中性暂停页(与 Host 模式
# 同语义;此处 Host 头无关紧要,解析根本不走到 Host)
r = disabled_client.get("/t/jiancai/api/chat/session",
headers={"host": NOPE_HOST})
assert r.status_code == 503
assert "暂停" in r.text
def test_site参数覆盖在路径模式生效(self, app_client):
# jiancai 的 es 站点(空域名站点):默认取首站点 main,?site= 覆盖
# 链路既有(spec §3.4)——brand title 取站点覆盖值即证覆盖真生效
r = app_client.get("/t/jiancai/api/public/brand?site=es",
headers={"host": NOPE_HOST})
assert r.status_code == 200
assert r.json()["title"] == "HESUPIPE Soporte"
def test_平台豁免优先级不受影响(self, app_client):
# /health 豁免不受路径模式影响;/t/x/api/platform/... 剥离后命中
# 平台路由属 by design(spec §3.5)——认证守护兜底,未带凭据
# 401/422/429(POST 空 body → 422)
assert app_client.get("/health", headers={"host": NOPE_HOST}).status_code == 200
r = app_client.post("/t/jiancai/api/platform/auth/login",
headers={"host": NOPE_HOST})
assert r.status_code in (401, 422, 429)
async def test_orig_path与路径改写直接钉位(self, tmp_path):
"""直接构造 ASGI scope 断言改写(不经 httpx 归一化):capture app
记录中间件改写后的 path/raw_path/state["orig_path"]——spec §3.2/
§3.3 的直证。registry 手工装配,用完 dispose_all(Windows 文件锁)。"""
reg = TenantRegistry.load_from_dir(SEED_DIR, data_dir=tmp_path)
captured = {}
async def capture_app(scope, receive, send):
state = scope.get("state") or {}
captured.update(
path=scope.get("path"),
raw_path=scope.get("raw_path"),
orig_path=state.get("orig_path"),
tenant=(state.get("tenant").config.slug
if state.get("tenant") else None),
site=state.get("site").slug if state.get("site") else None,
)
async def noop_receive():
return {"type": "http.request", "body": b"", "more_body": False}
async def noop_send(message):
pass
def make_scope(rtype, path, raw_path):
return {
"type": rtype, "asgi": {"version": "3.0"},
"http_version": "1.1", "method": "GET", "scheme": "http",
"path": path, "raw_path": raw_path, "query_string": b"",
"headers": [(b"host", b"unknown.example")],
"client": ("127.0.0.1", 12345),
"app": SimpleNamespace(state=SimpleNamespace(registry=reg)),
}
mw = TenantHostMiddleware(capture_app)
try:
# http:基本剥离 + orig_path 留痕 + state 写入
await mw(make_scope("http", "/t/jiancai/static/widget.js",
b"/t/jiancai/static/widget.js"),
noop_receive, noop_send)
assert captured["path"] == "/static/widget.js"
assert captured["raw_path"] == b"/static/widget.js"
assert captured["orig_path"] == "/t/jiancai/static/widget.js"
assert captured["tenant"] == "jiancai"
assert captured["site"] == "main"
# raw_path 重编码而非长度截断(spec §3.2):%2F 解码成 "/" 后,
# 截断会得 b"/a%2Fb",重算得 b"/a/b"
await mw(make_scope("http", "/t/jiancai/a/b",
b"/t/jiancai/a%2Fb"),
noop_receive, noop_send)
assert captured["path"] == "/a/b"
assert captured["raw_path"] == b"/a/b"
# 根路径形态:/t/jiancai/ → "/"(剥空兜底)
await mw(make_scope("http", "/t/jiancai/", b"/t/jiancai/"),
noop_receive, noop_send)
assert captured["path"] == "/"
assert captured["raw_path"] == b"/"
# websocket scope 同构改写(不必真握手,capture 收到即证)
await mw(make_scope("websocket", "/t/jiancai/ws/chat",
b"/t/jiancai/ws/chat"),
noop_receive, noop_send)
assert captured["path"] == "/ws/chat"
assert captured["orig_path"] == "/t/jiancai/ws/chat"
finally:
await reg.dispose_all()
def test_WS路径模式握手(self, app_client):
# WS 握手后服务端先等客户端 init 消息(chat.py:48)——先发
# standalone 再收 init;session_id 即证路径模式 WS 全链打通
with app_client.websocket_connect(
"/t/jiancai/ws/chat", headers={"host": NOPE_HOST}) as ws:
ws.send_text(json.dumps({"source": "standalone"}))
data = ws.receive_json()
assert data["type"] == "init"
assert "session_id" in data
def test_WS未知slug拒绝(self, app_client):
with pytest.raises(WebSocketDisconnect) as ei:
with app_client.websocket_connect(
"/t/no-such/ws/chat", headers={"host": NOPE_HOST}):
pass # pragma: no cover — 未 accept 即 close,进不来
assert ei.value.code == 1008
def test_Host模式回归_路径优先(self, app_client):
# 已知 Host(AUTO_HOST 属 auto 租户)+ /t/jiancai/ → 出 jiancai
# 而非 auto,证明路径前缀解析赢过 Host 解析(spec §2 优先级)
r = app_client.get("/t/jiancai/api/public/brand",
headers={"host": AUTO_HOST})
assert r.status_code == 200
assert r.json()["slug"] == "jiancai"
+4 -3
View File
@@ -329,9 +329,10 @@ class TestResolveBySlug:
await reg.dispose_all()
async def test_空站点表按未知处理(self, tmp_path):
"""防御分支(spec §5):两源加载期均拒零站点,直装 sites=[] 的
LoadedTenant 绕过源校验覆盖该分支——正常运行不可达,防将来源
语义漂移。"""
"""空站点表 → 按未知处理(404)。该分支生产可达而非防御:DB 源
全站点停用是合法运营状态(tenant_source 过滤 active 站点后可发出
sites=[])。文件源加载期拒零站点——直装 sites=[] 的 LoadedTenant
绕过源校验直接覆盖此分支。"""
seed = _write_seed(tmp_path, {"jiancai": _tenant("jiancai")})
source = FileTenantSource(seed)
lt = dataclasses.replace(source.read_tenant("jiancai"), sites=[])