浅念
d1aed8e722
上游 ai-kefu → phase1 吸收(Task 11 unit 12): - 拷入 7 个测试文件,import/夹具按目标库重锚(app.engine.* / conftest async_db+build_jiancai_runtime / seed_profiles 提示词锚点) - 补实现(同步自上游): chat_rest 服务端可信 IP(_server_ip/ _merge_client_info,直连场景自报假 IP 不再原样落库);widget.js 与 chat.js 的 data-placeholder 读取(叠加 Task 10 管道:data-* > brand > 中性兜底,tests/test_placeholder_static.py 守卫) - 断言重锚以 jc 主线为准: ev.contact_info 存裸号+等待尾注(e164 走 _cc 派生)、source 值 fallback/customer_stated 口径、推送标注键 号码补全说明、prompt 静态锚按租户(seed_profiles)双测 - 全量回归 459 passed(基线 421 + 吸收 38) Co-Authored-By: Claude Code <noreply@anthropic.com>
57 lines
2.6 KiB
Python
57 lines
2.6 KiB
Python
"""REST /session 服务端 IP 注入(同步自上游 ai-kefu tests/test_chat_rest_ip.py)。
|
|
|
|
spec 语义: XFF 第一段优先压过客户端自报,堵伪造。两处按目标库重锚:
|
|
- 实现侧: _server_ip/_merge_client_info 已照上游同步进
|
|
app/engine/routes/chat_rest.py( jc 原先只在有 XFF 时覆盖,直连场景自报
|
|
假 IP 原样落库——上游修了 jc 没跟上的 bug,补实现);
|
|
- 读回侧: 上游覆盖 chat_rest.async_session(全局 sessionmaker),jc(Task 5b)
|
|
无全局单例——走 TestClient portal + app.state.registry.resolve,请求与
|
|
读回同一事件循环(engine 连接绑定安全,conftest app_client 夹具纪律)。
|
|
WS 路径(app/engine/routes/chat.py)保持 jc 现状(无 XFF 时保留自报),
|
|
两侧不对称是 jc 主线决策,见该文件注释。
|
|
"""
|
|
from sqlalchemy import select
|
|
|
|
from app.engine.models import Conversation
|
|
from tests.conftest import JIANCAI_HOST
|
|
|
|
|
|
def _create(client, headers=None, body_client_info=None):
|
|
return client.post("/api/chat/session",
|
|
json={"client_info": body_client_info}, headers=headers or {})
|
|
|
|
|
|
def _conv_client_info(client, session_id):
|
|
async def run():
|
|
registry = client.app.state.registry
|
|
rt = await registry.resolve(JIANCAI_HOST)
|
|
async with rt.session_factory() as db:
|
|
conv = (await db.execute(
|
|
select(Conversation).where(Conversation.session_id == session_id)
|
|
)).scalar_one()
|
|
return dict(conv.client_info or {})
|
|
|
|
return client.portal.call(run)
|
|
|
|
|
|
def test_xff_first_segment_overrides_self_reported(client):
|
|
resp = _create(client, headers={"X-Forwarded-For": "41.90.64.1, 10.0.0.1"},
|
|
body_client_info={"ip": "1.2.3.4", "ua": "test"})
|
|
assert resp.status_code == 200 and resp.json()["is_new"] is True
|
|
ci = _conv_client_info(client, resp.json()["session_id"])
|
|
assert ci["ip"] == "41.90.64.1" # 服务端 XFF 第一段压过自报假 IP
|
|
assert ci["ua"] == "test" # 其余自报键保留
|
|
|
|
|
|
def test_no_xff_falls_back_to_direct_host(client):
|
|
resp = _create(client, body_client_info={"ip": "1.2.3.4"})
|
|
ci = _conv_client_info(client, resp.json()["session_id"])
|
|
assert ci["ip"] == "testclient" # TestClient 直连 host 覆盖自报
|
|
|
|
|
|
def test_merge_client_info_empty_server_ip_preserves_self_report():
|
|
from app.engine.routes.chat_rest import _merge_client_info
|
|
assert _merge_client_info({"ip": "9.9.9.9", "ua": "x"}, "") == {"ip": "9.9.9.9", "ua": "x"}
|
|
assert _merge_client_info(None, "") == {}
|
|
assert _merge_client_info(None, "41.90.64.1") == {"ip": "41.90.64.1"}
|