fix: 品牌下发审查修复(esc统一/语料中性化/穿越加固/反向断言动态化)

Co-Authored-By: Claude Code <noreply@anthropic.com>
This commit is contained in:
浅念
2026-10-12 22:52:08 +08:00
parent bbc7bc520e
commit 7697673a3b
8 changed files with 137 additions and 34 deletions
+62 -7
View File
@@ -8,6 +8,7 @@
import asyncio
import json
from pathlib import Path
from urllib.parse import urlparse
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
@@ -72,15 +73,47 @@ def test_brand_values_live_in_seed_profiles():
assert ab["topics"] == ["Get Quotation", "Distributor Cooperation"]
def _brand_identity_values():
"""复审动态化: 从全部 seed_profiles/*/tenant.json(租户默认门面 + 各站点
覆盖)收集品牌身份值作禁词表——第三租户落地即自动纳管,无需改本测试。
只收身份字段(title/whatsapp/email/privacy_url/welcome/topics);
color 与 avatar 文件名是中性 UI 默认(widget.js 的 DEFAULT_COLOR=
#066fc0 恰等于 jiancai 品牌色、DEFAULT_AVATAR_PATH 兜底文件同名),
收进来会误伤合法兜底。privacy_url 额外拆主机名(含去 www. 裸域名),
whatsapp 额外收去 + 变体(客户端硬编码裸号也算残留)。"""
values = set()
for tenant_file in sorted(SEED.glob("*/tenant.json")):
cfg = json.loads(tenant_file.read_text(encoding="utf-8"))
brands = [cfg.get("config", {}).get("brand") or {}]
brands += [s.get("brand") or {} for s in cfg.get("sites", [])]
for b in brands:
for key in ("title", "whatsapp", "email", "privacy_url", "welcome"):
v = str(b.get(key) or "").strip()
if v:
values.add(v)
values.add(v.lstrip("+"))
for t in b.get("topics") or []:
if str(t).strip():
values.add(str(t).strip())
host = urlparse(str(b.get("privacy_url") or "")).hostname or ""
if host:
values.add(host)
if host.startswith("www."):
values.add(host[4:])
assert values, "禁词表收集为空——seed_profiles 结构变了?"
return values
def test_no_brand_copy_left_in_client_files():
"""等价迁移锚:三份客户端文件不再内嵌任何租户品牌值(联系号/邮箱/品牌名/
隐私 URL),品牌只经 /api/public/brand 下发。"""
for rel in ("widget.js", "js/chat.js", "chat.html"):
"""等价迁移锚:四份客户端文件不再内嵌任何租户品牌值(联系号/邮箱/品牌名/
隐私 URL/welcome/topics),品牌只经 /api/public/brand 下发。禁词表动态
收集自 seed_profiles(见 _brand_identity_values)。"""
forbidden = _brand_identity_values()
for rel in ("widget.js", "js/chat.js", "chat.html", "widget-test.html"):
src = _read(rel)
assert "HESUPIPE" not in src, f"{rel} 残留品牌名"
assert "8615937108317" not in src and "8619037179250" not in src, f"{rel} 残留联系号"
assert "hesuu.com" not in src and "borderlesscar.com" not in src, f"{rel} 残留邮箱域名"
assert "hesupipe.com" not in src, f"{rel} 残留隐私 URL 域名"
for v in forbidden:
assert v not in src, f"{rel} 残留品牌值: {v!r}"
# ---- 服务端配套:会话打站点标 ---------------------------------------------------
@@ -125,3 +158,25 @@ def test_rest_session_site_query_recorded(app_client, tmp_path):
def test_rest_session_without_site_no_tag(app_client, tmp_path):
sid = app_client.post("/api/chat/session", json={}).json()["session_id"]
assert "site" not in _read_client_info(tmp_path / "jiancai.db", sid)
# ---- 复审: 专用 site 通道优先级(WS 与 REST 对齐) -------------------------------
def test_ws_dedicated_site_wins_over_self_reported(app_client, tmp_path):
"""WS init 同时带专用 site 字段与 client_info 自报 site 时,专用字段赢
(与 REST ?site= 覆盖 req.client_info 的优先级对齐——专用通道是嵌入
代码显式声明,可信度高于任意自报 dict)。"""
with app_client.websocket_connect("/ws/chat") as ws:
ws.send_text(json.dumps({"source": "widget", "site": "es",
"client_info": {"site": "main"}}))
sid = ws.receive_json()["session_id"]
assert _read_client_info(tmp_path / "jiancai.db", sid).get("site") == "es"
def test_rest_site_query_wins_over_body_client_info(app_client, tmp_path):
# 同上语义的 REST 钉子: ?site= 覆盖请求体 client_info 自报的 site
sid = app_client.post(
"/api/chat/session?site=es", json={"client_info": {"site": "main"}}
).json()["session_id"]
assert _read_client_info(tmp_path / "jiancai.db", sid).get("site") == "es"