feat(server): URL清洗三轨制+共享测试向量（D08）

Co-Authored-By: Claude Code <noreply@anthropic.com>

server/tracker/constants.py

@@ -1,3 +1,5 @@
+import re
+
 EVENT_NAMES = {"page_view", "page_stay", "sdk_loaded"}
 MAX_EVENTS_PER_BATCH = 50
 MAX_PROP_KEYS = 32
@@ -7,7 +9,14 @@ MAX_TITLE_LEN = 200
 CLOCK_SKEW_LATE_SEC = 120        # 超过即标 late
 CLOCK_SKEW_REJECT_SEC = 86400    # 超过即拒绝
 SENSITIVE_KEY_PATTERNS = ("password", "passwd", "token", "secret", "card", "verif", "otp", "idcard")
-URL_PARAM_DROP_PATTERNS = ("email", "phone", "mobile", "token", "key", "password", "verif")
+SANITIZE_RULE_VERSION = "v2"
+# 三轨制（spec §2.4，slug 安全线=32）：①参数名黑名单整键删；②归因白名单提取保留；
+# ③其余按值判别：纯数字≤10位或 slug≤32 保留，其余删。
+URL_PARAM_DROP_PATTERNS = ("email", "phone", "mobile", "tel", "name", "user",
+                           "token", "key", "password", "verif", "passwd", "idcard")
+ATTRIBUTION_PARAM_WHITELIST = {"gclid", "gbraid", "wbraid", "fbclid", "msclkid",
+                               "yclid", "gad_source", "gad_campaignid", "dclid"}
+_SAFE_VALUE = re.compile(r"^(?:\d{1,10}|[A-Za-z0-9][A-Za-z0-9._-]{0,30})$")  # slug≤32：32+连续字母数字视为token删
 SESSION_IDLE_MS = 30 * 60 * 1000
 BOT_UA_KEYWORDS = ("bot", "spider", "crawl", "slurp", "bingpreview", "headless", "lighthouse")
 EVENT_PROP_WHITELIST = {          # spec §9.1：未知字段拒绝。免费文本/任意属性不得入库

server/tracker/services/validation.py

@@ -17,11 +17,24 @@ def _utf8_ok(s: str) -> bool:
         return False
 
 def sanitize_url(url: str) -> str:
-    """按参数名整键删除（fullmatch 防止误伤 monkey 这类含 key 子串的参数）。"""
+    """三轨制清洗（spec §2.4）：黑名单参数名删 + 归因白名单提 + 其余按值判别。
+    另：userinfo 剥离、fragment 删除。规则版本 SANITIZE_RULE_VERSION。"""
     p = urlparse(url)
-    kept = [(k, v) for k, v in parse_qsl(p.query, keep_blank_values=True)
+    kept = []
-            if not _URL_DROP_KEY.fullmatch(k)]
+    for k, v in parse_qsl(p.query, keep_blank_values=True):
-    return p._replace(query=urlencode(kept)).geturl()
+        if _URL_DROP_KEY.fullmatch(k):
+            continue
+        if k.lower() in C.ATTRIBUTION_PARAM_WHITELIST or k.lower().startswith("utm_"):
+            kept.append((k, v)); continue
+        if C._SAFE_VALUE.match(v or ""):
+            kept.append((k, v))
+    netloc = p.hostname or ""
+    try:
+        if p.port:
+            netloc = f"{netloc}:{p.port}"
+    except ValueError:
+        pass  # 非法端口（如 :abc）——剥离端口降级，不让入口清洗抛异常
+    return p._replace(netloc=netloc, query=urlencode(kept), fragment="").geturl()
 
 def _walk_props(v, depth=0):
     """单次递归同时做深度检查、敏感键扫描与字符串超长检查。

server/tracker/tests/fixtures/sanitize_vectors.json

@@ -0,0 +1,19 @@
+{
+  "rule_version": "v2",
+  "vectors": [
+    {"in": "/product?id=456", "out": "/product?id=456"},
+    {"in": "/product?id=456&email=a@b.com", "out": "/product?id=456"},
+    {"in": "/page?phone=13800138000", "out": "/page"},
+    {"in": "/lp?utm_source=meta&utm_campaign=x", "out": "/lp?utm_source=meta&utm_campaign=x"},
+    {"in": "/lp?gclid=EAIaIQ&fbclid=IwAR2x", "out": "/lp?gclid=EAIaIQ&fbclid=IwAR2x"},
+    {"in": "/p?token=abcdef0123456789abcdef0123456789x", "out": "/p"},
+    {"in": "/p?ref=abcdefghijklmnopqrstuvwxyz012345", "out": "/p"},
+    {"in": "/p?slug=abcdefghijklmnopqrstuvwxyz0123456", "out": "/p"},
+    {"in": "/p?cat=sedans&sort=price", "out": "/p?cat=sedans&sort=price"},
+    {"in": "/p?q=%E5%BC%A0%E4%B8%89%E7%9A%84%E9%82%AE%E7%AE%B1a%40b.com", "out": "/p"},
+    {"in": "/a/b?name=zhangsan", "out": "/a/b"},
+    {"in": "/view?v=abc123DEF-_456", "out": "/view?v=abc123DEF-_456"},
+    {"in": "/long?v=aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", "out": "/long"},
+    {"in": "https://user:pass@example.com/p?x=1&email=c@d.com#frag", "out": "https://example.com/p?x=1"}
+  ]
+}

server/tracker/tests/test_validation.py

@@ -1,7 +1,12 @@
+import json
+import pathlib
 from datetime import datetime, timedelta
 from django.utils import timezone
+from tracker import constants as C
 from tracker.services.validation import validate_event, sanitize_url
 
+VECTORS = json.loads((pathlib.Path(__file__).parent / "fixtures" / "sanitize_vectors.json").read_text(encoding="utf-8"))
+
 NOW = timezone.now()
 
 def _ev(**kw):
@@ -46,11 +51,12 @@ def test_pageview_referrer_url_sanitized():
 
 
 def test_pageview_referrer_truncated_after_reencode():
-    """清洗重编码（中文 → %XX 九倍膨胀）可能超出原长：出门统一截 MAX_STR_LEN=500。"""
+    """清洗重编码（中文 → %XX 九倍膨胀）可能超出原长：出门统一截 MAX_STR_LEN=500。
-    ev = _ev(properties={"referrer": "https://www.bing.com/search?q=" + "词" * 200})
+    用 utm 值驱动：utm 参数三轨制下无条件保留（不走值判别），膨胀路径仍可达。"""
-    assert validate_event(ev) is None              # 原文 227 字符过长度门
+    ev = _ev(properties={"referrer": "https://www.bing.com/search?utm_term=" + "词" * 200})
+    assert validate_event(ev) is None              # 原文 237 字符过长度门
     ref = ev["properties"]["referrer"]
-    assert len(ref) == 500 and ref.startswith("https://www.bing.com/search?q=%E8%AF%8D")
+    assert len(ref) == 500 and ref.startswith("https://www.bing.com/search?utm_term=%E8%AF%8D")
 
 def test_nested_sensitive_key_rejected():
     assert validate_event(_ev(properties={"utm": {"utm_a": 1, "user_password": "x"}})) == "sensitive_field"  # utm 在白名单内，嵌套敏感键仍须命中
@@ -83,3 +89,10 @@ def test_non_str_path_rejected():
 
 def test_prop_depth_exceeded():
     assert validate_event(_ev(properties={"utm": {"a": {"b": 1}}})) == "properties_depth"
+
+def test_sanitize_url_vectors():
+    for v in VECTORS["vectors"]:
+        assert sanitize_url(v["in"]) == v["out"], v["in"]
+
+def test_sanitize_rule_version_exported():
+    assert C.SANITIZE_RULE_VERSION == VECTORS["rule_version"]