diff --git a/docs/superpowers/plans/2026-09-15-m1-data-collection-foundation.md b/docs/superpowers/plans/2026-09-15-m1-data-collection-foundation.md index 88d7de2..5d67a8a 100644 --- a/docs/superpowers/plans/2026-09-15-m1-data-collection-foundation.md +++ b/docs/superpowers/plans/2026-09-15-m1-data-collection-foundation.md @@ -498,9 +498,14 @@ def test_shape_type_and_whitelist(): def test_too_many_props_and_long_string(): r = validate_event(_ev(properties={f"k{i}": "v" for i in range(40)})) assert "properties" in r - r = validate_event(_ev(properties={"a": "x" * 600})) + r = validate_event(_ev(properties={"referrer": "x" * 600})) # 白名单内的键才能测到长度分支 assert "properties" in r +def test_nested_sensitive_key_and_naive_datetime(): + r = validate_event(_ev(properties={"utm": {"utm_a": 1, "user_password": "x"}})) + assert r == "sensitive_field" # 嵌套层级的键也要扫 + assert validate_event(_ev(occurred_at=NOW.replace(tzinfo=None))) == "invalid_occurred_at" + def test_future_and_stale_time(): r = validate_event(_ev(occurred_at=NOW + timedelta(days=2))) assert "occurred_at" in r # 超窗拒绝 @@ -519,7 +524,7 @@ def test_sanitize_url_drops_sensitive_params(): ```python 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 @@ -534,14 +539,29 @@ 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): + """一次递归同时做深度合规与敏感键扫描(任意层级)。返回 (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()) + for k, x in v.items(): + if _SENSITIVE.search(str(k)): + return True, True + ok, hit = _walk_props(x, depth + 1) + if hit: + return True, True + if not ok: + return False, False + return True, False if isinstance(v, list): - return all(_depth_ok(x, depth + 1) for x in v) - return True + for x in v: + ok, hit = _walk_props(x, depth + 1) + if hit: + return True, True + if not ok: + return False, False + return True, False + return True, False def validate_event(ev) -> str | None: """返回 None 表示通过;否则返回拒绝原因码。类型/白名单/体积/深度全在此守门。""" @@ -556,7 +576,7 @@ def validate_event(ev) -> str | None: if ev.get(k) is not None and not isinstance(ev[k], str): return "invalid_field_type" occ = ev.get("occurred_at") - if not isinstance(occ, datetime): + if not isinstance(occ, datetime) or timezone.is_naive(occ): return "invalid_occurred_at" skew = (timezone.now() - occ).total_seconds() if abs(skew) > C.CLOCK_SKEW_REJECT_SEC: @@ -564,15 +584,16 @@ def validate_event(ev) -> str | None: props = ev.get("properties") or {} if not isinstance(props, dict) or len(props) > C.MAX_PROP_KEYS: return "properties_invalid" + depth_ok, sensitive_hit = _walk_props(props) # 敏感键(任意层级)与深度先于白名单检查, + if sensitive_hit: # 否则白名单会把敏感码短路成 not_allowed + 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" - for k, v in props.items(): - if _SENSITIVE.search(str(k)): - return "sensitive_field" + for v in props.values(): if isinstance(v, str) and len(v) > C.MAX_STR_LEN: return "properties_too_long" if len(ev.get("path") or "") > C.MAX_PATH_LEN: