From f2a5f7611a60aec3f138560e7d60325458336342 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B5=85=E5=BF=B5?= Date: Tue, 15 Sep 2026 12:04:40 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E6=A0=A1=E9=AA=8C=E5=8A=A0=E5=9B=BA?= =?UTF-8?q?=EF=BC=88=E5=B5=8C=E5=A5=97=E6=95=8F=E6=84=9F=E9=94=AE/naive?= =?UTF-8?q?=E6=97=B6=E9=97=B4=E9=98=B2=E5=BE=A1=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - _depth_ok 与敏感键扫描合并为单次递归 _walk_props，dict 所有层级键均过 _SENSITIVE（此前嵌套键不扫描，utm 白名单内可夹带敏感键） - occurred_at 为 naive datetime 时返回 invalid_occurred_at（原为 TypeError 500） - 删除未使用的 timedelta 导入 - 深度超限即停止下探，10 万层嵌套不爆栈 Co-Authored-By: Claude Code --- server/tracker/services/validation.py | 38 ++++++++++++++++++------- server/tracker/tests/test_validation.py | 7 +++++ 2 files changed, 34 insertions(+), 11 deletions(-) diff --git a/server/tracker/services/validation.py b/server/tracker/services/validation.py index 607c737..97a3024 100644 --- a/server/tracker/services/validation.py +++ b/server/tracker/services/validation.py @@ -1,6 +1,6 @@ import json import re -from datetime import datetime, timedelta +from datetime import datetime from urllib.parse import parse_qsl, urlencode, urlparse from django.utils import timezone from tracker import constants as C @@ -15,14 +15,28 @@ def sanitize_url(url: str) -> str: if not _URL_DROP_KEY.fullmatch(k)] return p._replace(query=urlencode(kept)).geturl() -def _depth_ok(v, depth=0) -> bool: +def _walk_props(v, depth=0): + """单次递归同时做深度检查与敏感键扫描（dict 所有层级的键都过 _SENSITIVE）。 + 返回 (depth_ok, sensitive_hit)。深度超限即停止下探，防恶意深嵌套递归爆栈。""" if depth > C.MAX_PROP_DEPTH: - return False + return False, False if isinstance(v, dict): - return all(_depth_ok(x, depth + 1) for x in v.values()) + depth_ok, sens = True, False + for k, x in v.items(): + if _SENSITIVE.search(str(k)): + sens = True + d, s = _walk_props(x, depth + 1) + depth_ok = depth_ok and d + sens = sens or s + return depth_ok, sens if isinstance(v, list): - return all(_depth_ok(x, depth + 1) for x in v) - return True + depth_ok, sens = True, False + for x in v: + d, s = _walk_props(x, depth + 1) + depth_ok = depth_ok and d + sens = sens or s + return depth_ok, sens + return True, False def validate_event(ev) -> str | None: """返回 None 表示通过；否则返回拒绝原因码。类型/白名单/体积/深度全在此守门。""" @@ -39,20 +53,22 @@ def validate_event(ev) -> str | None: occ = ev.get("occurred_at") if not isinstance(occ, datetime): return "invalid_occurred_at" + if timezone.is_naive(occ): + return "invalid_occurred_at" 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 {} if not isinstance(props, dict) or len(props) > C.MAX_PROP_KEYS: return "properties_invalid" - # 敏感键扫描必须先于属性白名单：白名单内无任何键命中敏感模式， + # 敏感键扫描（含嵌套层级）先于属性白名单：白名单内无任何键命中敏感模式， # 若放在白名单之后，sensitive_field 将不可达（见 test_sensitive_property_rejected）。 - for k in props: - if _SENSITIVE.search(str(k)): - return "sensitive_field" + depth_ok, sensitive_hit = _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(props): + if not depth_ok: return "properties_depth" if len(json.dumps(props, ensure_ascii=False, default=str)) > C.MAX_PROP_JSON_BYTES: return "properties_too_large" diff --git a/server/tracker/tests/test_validation.py b/server/tracker/tests/test_validation.py index 9957e96..dd07651 100644 --- a/server/tracker/tests/test_validation.py +++ b/server/tracker/tests/test_validation.py @@ -40,3 +40,10 @@ def test_sanitize_url_drops_sensitive_params(): u = sanitize_url("/p?email=a@b.com&token=x&id=3&utm_source=google&monkey=1") assert "email=" not in u and "token=" not in u and "utm_source=google" in u assert "monkey=1" in u # fullmatch：含 key 子串的正常参数不误伤 + +def test_nested_sensitive_key_rejected(): + r = validate_event(_ev(properties={"utm": {"utm_a": 1, "user_password": "x"}})) + assert "sensitive" in r # utm 在 page_view 白名单内，嵌套敏感键仍须命中 + +def test_naive_datetime_rejected(): + assert validate_event(_ev(occurred_at=NOW.replace(tzinfo=None))) == "invalid_occurred_at"