diff --git a/server/tracker/constants.py b/server/tracker/constants.py index 1794098..8af2034 100644 --- a/server/tracker/constants.py +++ b/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 diff --git a/server/tracker/services/validation.py b/server/tracker/services/validation.py index 2161c1e..cffdff5 100644 --- a/server/tracker/services/validation.py +++ b/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 拒; diff --git a/server/tracker/tests/test_validation.py b/server/tracker/tests/test_validation.py index c0b800b..86ae9bc 100644 --- a/server/tracker/tests/test_validation.py +++ b/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