From cb57e67179bf73f7deac401e1a274a2e4d21c3d1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B5=85=E5=BF=B5?= Date: Tue, 15 Sep 2026 12:01:45 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E4=BA=8B=E4=BB=B6=E6=A0=A1=E9=AA=8C?= =?UTF-8?q?=E6=9C=8D=E5=8A=A1=EF=BC=88=E7=99=BD=E5=90=8D=E5=8D=95/?= =?UTF-8?q?=E6=95=8F=E6=84=9F=E9=94=AE/=E6=97=B6=E9=97=B4=E7=AA=97/URL?= =?UTF-8?q?=E6=B8=85=E6=B4=97=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit TDD：7 测试先红（ModuleNotFoundError）后绿。与计划的一处偏差： 敏感键扫描移至属性白名单之前——白名单内无任何键命中敏感模式， 原顺序下 sensitive_field 不可达，test_sensitive_property_rejected （user_password 非白名单键）将得到 property_not_allowed 而失败。 Co-Authored-By: Claude Code --- server/tracker/services/__init__.py | 0 server/tracker/services/validation.py | 68 +++++++++++++++++++++++++ server/tracker/tests/test_validation.py | 42 +++++++++++++++ 3 files changed, 110 insertions(+) create mode 100644 server/tracker/services/__init__.py create mode 100644 server/tracker/services/validation.py create mode 100644 server/tracker/tests/test_validation.py diff --git a/server/tracker/services/__init__.py b/server/tracker/services/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/server/tracker/services/validation.py b/server/tracker/services/validation.py new file mode 100644 index 0000000..607c737 --- /dev/null +++ b/server/tracker/services/validation.py @@ -0,0 +1,68 @@ +import json +import re +from datetime import datetime, timedelta +from urllib.parse import parse_qsl, urlencode, urlparse +from django.utils import timezone +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 sanitize_url(url: str) -> str: + """按参数名整键删除（fullmatch 防止误伤 monkey 这类含 key 子串的参数）。""" + p = urlparse(url) + kept = [(k, v) for k, v in parse_qsl(p.query, keep_blank_values=True) + if not _URL_DROP_KEY.fullmatch(k)] + return p._replace(query=urlencode(kept)).geturl() + +def _depth_ok(v, depth=0) -> bool: + if depth > C.MAX_PROP_DEPTH: + return False + if isinstance(v, dict): + return all(_depth_ok(x, depth + 1) for x in v.values()) + if isinstance(v, list): + return all(_depth_ok(x, depth + 1) for x in v) + return True + +def validate_event(ev) -> str | None: + """返回 None 表示通过；否则返回拒绝原因码。类型/白名单/体积/深度全在此守门。""" + 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: + return "invalid_event_id" + for k in ("path", "title"): + 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): + 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" + if not set(props) <= C.EVENT_PROP_WHITELIST.get(ev["event_name"], set()): + return "property_not_allowed" + if not _depth_ok(props): + return "properties_depth" + if len(json.dumps(props, ensure_ascii=False, default=str)) > 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 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 diff --git a/server/tracker/tests/test_validation.py b/server/tracker/tests/test_validation.py new file mode 100644 index 0000000..9957e96 --- /dev/null +++ b/server/tracker/tests/test_validation.py @@ -0,0 +1,42 @@ +from datetime import datetime, timedelta +from django.utils import timezone +from tracker.services.validation import validate_event, sanitize_url + +NOW = timezone.now() + +def _ev(**kw): + base = dict(event_id="e1", event_name="page_view", + occurred_at=NOW, path="/p", title="t", properties={}) + base.update(kw) + return base + +def test_ok_event(): + assert validate_event(_ev()) is None # None = 通过 + +def test_unknown_event_name(): + assert validate_event(_ev(event_name="xxx")) == "event_name_not_allowed" + +def test_sensitive_property_rejected(): + r = validate_event(_ev(properties={"user_password": "1"})) + assert "sensitive" in r + +def test_shape_type_and_whitelist(): + assert validate_event(["x"]) == "event_invalid_shape" + assert validate_event(_ev(event_id=123)) == "invalid_event_id" + assert validate_event(_ev(properties={"foo": "bar"})) == "property_not_allowed" + +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={"referrer": "x" * 600})) + assert "properties" in r + +def test_future_and_stale_time(): + r = validate_event(_ev(occurred_at=NOW + timedelta(days=2))) + assert "occurred_at" in r # 超窗拒绝 + assert validate_event(_ev(occurred_at=NOW - timedelta(hours=2))) is None # 窗内迟到可收 + +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 子串的正常参数不误伤