fix: 采集接口 origin urlsplit/finalize_response 信封/CL 防御

审查三项必修:
1. Origin host 提取改 urlsplit(封堵 https://evil.com/?x=https://www.a.com
   类查询串绕过);"null" Origin 视为无 host 走 allow_no_origin;
   Referer 回显死逻辑一并清理,ACAO 只回显真正的 Origin 头。
2. finalize_response 统一钩子:429/ParseError 等 DRF 异常响应补 ACAO +
   {code,data,message} 信封(浏览器 SDK 需读错误体,误判网络错误重试会
   打爆限流器);删 post/_deny/options 三处临时加头代码。
3. Content-Length isdecimal 防御("abc" → 400 非 500;isdigit 放过 "²"
   而 int("²") 仍崩,isdecimal 与 int() 可解析集精确对齐)。

锁定测试 ×4:429 信封+ACAO / 坏 JSON 信封+ACAO / 查询串绕过 403 / 畸形 CL 400。

Co-Authored-By: Claude Code <noreply@anthropic.com>
This commit is contained in:
浅念
2026-09-15 13:24:31 +08:00
parent cf04db794b
commit 678d215b31
2 changed files with 65 additions and 21 deletions
+36
View File
@@ -61,3 +61,39 @@ def test_throttle_429(db, client, settings):
for _ in range(3):
assert _post(client, _body(1)).status_code == 200
assert _post(client, _body(1)).status_code == 429
# ---- 审查修复锁定测试 ----
def test_throttle_429_envelope_and_acao(db, client, settings):
"""429 走 finalize_response:信封 + ACAO——否则浏览器 SDK 读不到错误体误判网络错误重试,反而打爆限流器。"""
settings.REST_FRAMEWORK = {**settings.REST_FRAMEWORK,
"DEFAULT_THROTTLE_RATES": {"collect": "3/min"}}
Site.objects.create(name="s", domains=["www.a.com"], site_id="site_x")
for _ in range(3):
assert _post(client, _body(1)).status_code == 200
r = _post(client, _body(1))
assert r.status_code == 429
assert r["Access-Control-Allow-Origin"] == "https://www.a.com"
body = r.json()
assert body["code"] == 429 and body["data"] is None and body["message"]
def test_invalid_json_envelope_and_acao(db, client):
"""ParseError 400 由 DRF 默认异常处理器产出 {detail}——finalize_response 统一转信封并补 ACAO。"""
Site.objects.create(name="s", domains=["www.a.com"], site_id="site_x")
r = client.post("/api/collect/batch", data="{bad json",
content_type="text/plain", HTTP_ORIGIN="https://www.a.com")
assert r.status_code == 400
assert r.json() == {"code": 400, "data": None, "message": "invalid_json"}
assert r["Access-Control-Allow-Origin"] == "https://www.a.com"
def test_origin_url_in_query_bypass_blocked(db, client):
"""split("//")[-1] 提取 host 会被查询串里的白名单域名绕过——urlsplit 后必 403。"""
Site.objects.create(name="s", domains=["www.a.com"], site_id="site_x")
r = _post(client, _body(1), origin="https://evil.com/?x=https://www.a.com")
assert r.status_code == 403 and r.json()["message"] == "not_allowed"
def test_malformed_content_length_400_not_500(db, client):
"""Content-Length 非数字("abc"):isdigit/isdecimal 防御,拒 400 而非 int() 崩 500。"""
Site.objects.create(name="s", domains=["www.a.com"], site_id="site_x")
r = _post(client, _body(1), extra={"CONTENT_LENGTH": "abc"})
assert r.status_code == 400 and r.json()["message"] == "payload_too_large"