From c5f900f3bd0556f0d61b8010e6979636fc1da302 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B5=85=E5=BF=B5?= Date: Tue, 15 Sep 2026 12:10:56 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E6=A0=A1=E9=AA=8C=E5=AE=88=E9=97=A8?= =?UTF-8?q?=E6=8C=89=E5=AD=97=E8=8A=82+=E9=80=92=E5=BD=92=E9=95=BF?= =?UTF-8?q?=E5=BA=A6+=E9=9D=9Edict=E6=8B=92=E7=BB=9D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 体积守门改 UTF-8 字节数（原按字符数，"汉"*1500 约 4.5KB 可绕过 4096 上限） - 字符串长度检查并入 _walk_props 递归，嵌套字符串同样受 MAX_STR_LEN 约束 （顺序：体积先行，故同一值既超长又超体积时报 properties_too_large） - properties 传了但非 dict（[]/""/0/None）不再被 or {} 静默放行，缺省键仍放行 Co-Authored-By: Claude Code --- server/tracker/services/validation.py | 39 ++++++++++++++----------- server/tracker/tests/test_validation.py | 10 +++++++ 2 files changed, 32 insertions(+), 17 deletions(-) diff --git a/server/tracker/services/validation.py b/server/tracker/services/validation.py index 97a3024..8234891 100644 --- a/server/tracker/services/validation.py +++ b/server/tracker/services/validation.py @@ -16,27 +16,32 @@ def sanitize_url(url: str) -> str: return p._replace(query=urlencode(kept)).geturl() def _walk_props(v, depth=0): - """单次递归同时做深度检查与敏感键扫描（dict 所有层级的键都过 _SENSITIVE）。 - 返回 (depth_ok, sensitive_hit)。深度超限即停止下探，防恶意深嵌套递归爆栈。""" + """单次递归同时做深度检查、敏感键扫描与字符串超长检查。 + dict 所有层级的键都过 _SENSITIVE，所有层级的字符串都受 MAX_STR_LEN 约束。 + 返回 (depth_ok, sensitive_hit, str_too_long)。深度超限即停止下探，防恶意深嵌套递归爆栈。""" if depth > C.MAX_PROP_DEPTH: - return False, False + return False, False, False if isinstance(v, dict): - depth_ok, sens = True, False + depth_ok, sens, too_long = True, False, False for k, x in v.items(): if _SENSITIVE.search(str(k)): sens = True - d, s = _walk_props(x, depth + 1) + d, s, l = _walk_props(x, depth + 1) depth_ok = depth_ok and d sens = sens or s - return depth_ok, sens + too_long = too_long or l + return depth_ok, sens, too_long if isinstance(v, list): - depth_ok, sens = True, False + depth_ok, sens, too_long = True, False, False for x in v: - d, s = _walk_props(x, depth + 1) + d, s, l = _walk_props(x, depth + 1) depth_ok = depth_ok and d sens = sens or s - return depth_ok, sens - return True, False + too_long = too_long or l + return depth_ok, sens, too_long + if isinstance(v, str) and len(v) > C.MAX_STR_LEN: + return True, False, True + return True, False, False def validate_event(ev) -> str | None: """返回 None 表示通过；否则返回拒绝原因码。类型/白名单/体积/深度全在此守门。""" @@ -58,23 +63,23 @@ def validate_event(ev) -> str | None: skew = (timezone.now() - occ).total_seconds() if abs(skew) > C.CLOCK_SKEW_REJECT_SEC: return "occurred_at_out_of_window" - props = ev.get("properties") or {} + props = ev.get("properties", {}) if not isinstance(props, dict) or len(props) > C.MAX_PROP_KEYS: - return "properties_invalid" + return "properties_invalid" # 传了但非 dict（[]/""/0/None）一律拒 # 敏感键扫描（含嵌套层级）先于属性白名单：白名单内无任何键命中敏感模式， # 若放在白名单之后，sensitive_field 将不可达（见 test_sensitive_property_rejected）。 - depth_ok, sensitive_hit = _walk_props(props) + depth_ok, sensitive_hit, str_too_long = _walk_props(props) if sensitive_hit: return "sensitive_field" if not set(props) <= C.EVENT_PROP_WHITELIST.get(ev["event_name"], set()): return "property_not_allowed" if not depth_ok: return "properties_depth" - if len(json.dumps(props, ensure_ascii=False, default=str)) > C.MAX_PROP_JSON_BYTES: + # 体积按 UTF-8 字节数守门（汉字多字节，按字符数会被绕过） + if len(json.dumps(props, ensure_ascii=False, default=str).encode("utf-8")) > C.MAX_PROP_JSON_BYTES: return "properties_too_large" - for v in props.values(): - if isinstance(v, str) and len(v) > C.MAX_STR_LEN: - return "properties_too_long" + if str_too_long: + return "properties_too_long" 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: diff --git a/server/tracker/tests/test_validation.py b/server/tracker/tests/test_validation.py index dd07651..34e6ee1 100644 --- a/server/tracker/tests/test_validation.py +++ b/server/tracker/tests/test_validation.py @@ -47,3 +47,13 @@ def test_nested_sensitive_key_rejected(): def test_naive_datetime_rejected(): assert validate_event(_ev(occurred_at=NOW.replace(tzinfo=None))) == "invalid_occurred_at" + +def test_prop_size_gate_counts_bytes(): + r = validate_event(_ev(properties={"utm": {"a": "汉" * 1500}})) + assert r == "properties_too_large" # 字符数约 2K 未超，UTF-8 约 4.5KB 超 4096，须按字节算 + +def test_nested_string_length_capped(): + assert validate_event(_ev(properties={"utm": {"a": "x" * 600}})) == "properties_too_long" + +def test_falsy_non_dict_properties_rejected(): + assert validate_event(_ev(properties=[])) == "properties_invalid"