feat(server): 业务事件契约——白名单+6事件与枚举校验

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

server/tracker/constants.py

@@ -1,6 +1,8 @@
 import re
 
-EVENT_NAMES = {"page_view", "page_stay", "sdk_loaded", "session_resume"}
+BIZ_EVENT_NAMES = {"contact_click", "form_start", "form_error", "form_submit",
+                   "form_submit_result", "product_view"}  # 阶段B业务事件（spec §1）
+EVENT_NAMES = {"page_view", "page_stay", "sdk_loaded", "session_resume"} | BIZ_EVENT_NAMES
 SCHEMA_VERSION = 2  # 批级信缺省 = v1（旧SDK）
 MAX_EVENTS_PER_BATCH = 50
 MAX_PROP_KEYS = 32
@@ -25,6 +27,13 @@ EVENT_PROP_WHITELIST = {          # spec §9.1：未知字段拒绝。免费文
     "page_stay": {"active_ms", "path", "active_from", "active_to", "flush_reason", "measurement_version"},
     "sdk_loaded": {"sdk_version", "build_id", "schema_version", "config_version"},
     "session_resume": {"prev_session_id", "reason"},
+    # 业务事件属性键白名单（spec §2）——值级枚举/格式校验在 validation._biz_props_ok
+    "contact_click": {"channel"},
+    "form_start": {"form_id"},
+    "form_submit": {"form_id", "submission_id"},
+    "form_error": {"form_id", "submission_id", "error_code", "field_count"},
+    "form_submit_result": {"form_id", "submission_id", "result_code"},
+    "product_view": {"product_id", "inferred", "rule_version"},
 }
 MAX_PROP_DEPTH = 2
 MAX_PROP_JSON_BYTES = 4096

server/tracker/services/validation.py

@@ -8,6 +8,39 @@ 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)
 
+# 业务事件属性值域（spec §2 枚举/格式表；全部非自由文本）
+_BIZ_CHANNEL = {"whatsapp", "phone", "email"}
+_BIZ_RESULT = {"mail_sent", "mail_failed"}
+_BIZ_ERROR = {"invalid", "spam", "other"}
+_SUBMIT_RE = re.compile(r"^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$|^$")  # uuid4，可空
+_FORM_RE = re.compile(r"^cf7-\d{1,10}$")     # CF7 表单 DOM id（f{n}-p{m} → 服务端规范化为 cf7-{form_id})
+_PROD_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,99}$")  # slug ≤100
+
+def _biz_props_ok(name, props) -> bool:
+    """业务事件属性逐键校验（spec §2 枚举/格式表；全部非自由文本）。
+    键级白名单已在调用方过完，这里管值：类型 + 枚举 + 格式。"""
+    def s(k): return isinstance(props.get(k), str)
+    def form_ok(): return s("form_id") and bool(_FORM_RE.match(props["form_id"]))
+    def sid_ok():
+        sid = props.get("submission_id", "")
+        return isinstance(sid, str) and bool(_SUBMIT_RE.match(sid))
+    if name == "contact_click":
+        return s("channel") and props["channel"] in _BIZ_CHANNEL
+    if name == "form_start":
+        return form_ok()
+    if name in ("form_submit", "form_submit_result"):
+        return form_ok() and sid_ok() and (
+            name == "form_submit" or (s("result_code") and props["result_code"] in _BIZ_RESULT))
+    if name == "form_error":
+        fc = props.get("field_count")
+        return (form_ok() and sid_ok() and s("error_code") and props["error_code"] in _BIZ_ERROR
+                and isinstance(fc, int) and 0 <= fc <= 99 and not isinstance(fc, bool))
+    if name == "product_view":
+        return (s("product_id") and bool(_PROD_RE.match(props["product_id"]))
+                and isinstance(props.get("inferred"), bool)
+                and s("rule_version") and len(props["rule_version"]) <= 16)
+    return True
+
 def _utf8_ok(s: str) -> bool:
     """孤代理（如 "\ud800"）在下游 encode/JSON 序列化时必然 500，入口处即拒。"""
     try:
@@ -140,6 +173,8 @@ def validate_event(ev, *, v2=False, batch_site_id="", batch_visitor_id="") -> st
         return "sensitive_field"
     if not set(props) <= C.EVENT_PROP_WHITELIST.get(ev["event_name"], set()):
         return "property_not_allowed"
+    if ev["event_name"] in C.BIZ_EVENT_NAMES and not _biz_props_ok(ev["event_name"], props):
+        return "property_not_allowed"
     if not depth_ok:
         return "properties_depth"
     # 体积按 UTF-8 字节数守门（汉字多字节，按字符数会被绕过）；孤代理 encode 必炸，按 invalid 拒；

server/tracker/tests/test_validation.py

@@ -1,5 +1,6 @@
 import json
 import pathlib
+import pytest
 from datetime import datetime, timedelta
 from django.utils import timezone
 from tracker import constants as C
@@ -96,3 +97,33 @@ def test_sanitize_url_vectors():
 
 def test_sanitize_rule_version_exported():
     assert C.SANITIZE_RULE_VERSION == VECTORS["rule_version"]
+
+
+# ---------------------------------------------------------------------------
+# 业务事件契约（spec §1/§2）：+6 事件白名单与属性枚举/格式校验
+# ---------------------------------------------------------------------------
+SITE_ID = "site_test"  # validate_event 不查 DB，本地定义即可
+
+def _biz(name, **props):
+    return {"event_id": f"biz-{name}", "event_name": name, "occurred_at": timezone.now(),
+            "path": "/p", "title": "t", "site_id": SITE_ID, "visitor_id": "v1",
+            "session_id": "s1", "page_id": "p1", "seq": 1, "properties": props}
+
+@pytest.mark.parametrize("ev,expect_reason", [
+    (_biz("contact_click", channel="whatsapp"), None),
+    (_biz("contact_click", channel="im"), "property_not_allowed"),
+    (_biz("form_start", form_id="cf7-156"), None),
+    (_biz("form_start", form_id="wpcf7-f156-p20"), "property_not_allowed"),
+    (_biz("form_submit", form_id="cf7-156", submission_id="9c8b7a6d-1111-4222-8333-444455556666"), None),
+    (_biz("form_submit", form_id="cf7-156", submission_id=""), None),
+    (_biz("form_submit", form_id="cf7-156", submission_id="not-uuid"), "property_not_allowed"),
+    (_biz("form_error", form_id="cf7-156", submission_id="", error_code="invalid", field_count=2), None),
+    (_biz("form_error", form_id="cf7-156", submission_id="", error_code="bogus"), "property_not_allowed"),
+    (_biz("form_error", form_id="cf7-156", submission_id="", error_code="invalid", field_count=-1), "property_not_allowed"),
+    (_biz("form_submit_result", form_id="cf7-156", submission_id="", result_code="mail_sent"), None),
+    (_biz("form_submit_result", form_id="cf7-156", submission_id="", result_code="ok"), "property_not_allowed"),
+    (_biz("product_view", product_id="byd-equation-leopard-5", inferred=True, rule_version="v1"), None),
+    (_biz("product_view", product_id="x"*101, inferred=True, rule_version="v1"), "property_not_allowed"),
+])
+def test_biz_event_validation(ev, expect_reason):
+    assert validate_event(ev, v2=True, batch_site_id=SITE_ID, batch_visitor_id="v1") == expect_reason