fix: 孤代理三入口堵漏+精确断言+截断清洗换序
- _utf8_ok 助手:event_id/path/title 含孤代理("\ud800")在入口即拒 (原会流到下游 encode 必 500);props 字节守门 try/except 归入 properties_invalid - 补零覆盖原因码测试:path=123 → invalid_field_type;utm 三层嵌套 → properties_depth - 子串断言("sensitive" in r 等)全改精确等值断言,码即 API - path 改为先 sanitize_url 再截 500:中文查询参数 urlencode 膨胀最多 3 倍, 先截后洗会破坏出门 ≤MAX_PATH_LEN 不变量(实测汉*200 查询洗后截回 500) - docstring 写明就地写回契约(Task 6 ingest 依赖) Co-Authored-By: Claude Code <noreply@anthropic.com>
This commit is contained in:
@@ -8,6 +8,14 @@ from tracker import constants as C
|
||||
_SENSITIVE = re.compile("|".join(C.SENSITIVE_KEY_PATTERNS), re.I)
|
||||
_URL_DROP_KEY = re.compile("|".join(C.URL_PARAM_DROP_PATTERNS), re.I)
|
||||
|
||||
def _utf8_ok(s: str) -> bool:
|
||||
"""孤代理(如 "\ud800")在下游 encode/JSON 序列化时必然 500,入口处即拒。"""
|
||||
try:
|
||||
s.encode("utf-8")
|
||||
return True
|
||||
except UnicodeEncodeError:
|
||||
return False
|
||||
|
||||
def sanitize_url(url: str) -> str:
|
||||
"""按参数名整键删除(fullmatch 防止误伤 monkey 这类含 key 子串的参数)。"""
|
||||
p = urlparse(url)
|
||||
@@ -44,16 +52,19 @@ def _walk_props(v, depth=0):
|
||||
return True, False, False
|
||||
|
||||
def validate_event(ev) -> str | None:
|
||||
"""返回 None 表示通过;否则返回拒绝原因码。类型/白名单/体积/深度全在此守门。"""
|
||||
"""返回 None 表示通过;否则返回拒绝原因码。类型/白名单/体积/深度全在此守门。
|
||||
通过时会就地截断/清洗并写回 ev(path 先 sanitize_url 再截 MAX_PATH_LEN,title 截
|
||||
MAX_TITLE_LEN)——Task 6 的 ingest 依赖此写回契约。"""
|
||||
if not isinstance(ev, dict):
|
||||
return "event_invalid_shape"
|
||||
if ev.get("event_name") not in C.EVENT_NAMES:
|
||||
return "event_name_not_allowed"
|
||||
eid = ev.get("event_id")
|
||||
if not isinstance(eid, str) or not eid or len(eid) > 64:
|
||||
if not isinstance(eid, str) or not eid or len(eid) > 64 or not _utf8_ok(eid):
|
||||
return "invalid_event_id"
|
||||
for k in ("path", "title"):
|
||||
if ev.get(k) is not None and not isinstance(ev[k], str):
|
||||
v = ev.get(k)
|
||||
if v is not None and (not isinstance(v, str) or not _utf8_ok(v)):
|
||||
return "invalid_field_type"
|
||||
occ = ev.get("occurred_at")
|
||||
if not isinstance(occ, datetime):
|
||||
@@ -75,15 +86,20 @@ def validate_event(ev) -> str | None:
|
||||
return "property_not_allowed"
|
||||
if not depth_ok:
|
||||
return "properties_depth"
|
||||
# 体积按 UTF-8 字节数守门(汉字多字节,按字符数会被绕过)
|
||||
if len(json.dumps(props, ensure_ascii=False, default=str).encode("utf-8")) > C.MAX_PROP_JSON_BYTES:
|
||||
# 体积按 UTF-8 字节数守门(汉字多字节,按字符数会被绕过);孤代理 encode 必炸,按 invalid 拒
|
||||
try:
|
||||
prop_bytes = len(json.dumps(props, ensure_ascii=False, default=str).encode("utf-8"))
|
||||
except UnicodeEncodeError:
|
||||
return "properties_invalid"
|
||||
if prop_bytes > C.MAX_PROP_JSON_BYTES:
|
||||
return "properties_too_large"
|
||||
if str_too_long:
|
||||
return "properties_too_long"
|
||||
# 先清洗后截断:urlencode 对中文查询参数最多膨胀 3 倍,先截后洗会破坏"出门 ≤MAX_PATH_LEN"不变量
|
||||
if ev.get("path"):
|
||||
ev["path"] = sanitize_url(ev["path"])
|
||||
if len(ev.get("path") or "") > C.MAX_PATH_LEN:
|
||||
ev["path"] = ev["path"][:C.MAX_PATH_LEN]
|
||||
if len(ev.get("title") or "") > C.MAX_TITLE_LEN:
|
||||
ev["title"] = ev["title"][:C.MAX_TITLE_LEN]
|
||||
if ev.get("path"):
|
||||
ev["path"] = sanitize_url(ev["path"])
|
||||
return None
|
||||
|
||||
Reference in New Issue
Block a user