fangke/server/tracker/services/validation.py

import json
import re
from datetime import datetime, timezone as dt_tz
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)

# 业务事件属性值域（spec §2 枚举/格式表；全部非自由文本）
_BIZ_CHANNEL = {"whatsapp", "phone", "email"}
_BIZ_RESULT = {"mail_sent", "mail_failed"}
_BIZ_ERROR = {"invalid", "spam", "other"}
# 锚点一律 \Z 而非 $：Python `$` 匹配串尾换行前，"cf7-156\n" 等脏值会被放行（collect 端点公开）
_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}\Z|\Z")  # uuid4，可空
_FORM_RE = re.compile(r"^cf7-\d{1,10}\Z")     # 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}\Z")  # 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:
        s.encode("utf-8")
        return True
    except UnicodeEncodeError:
        return False


def parse_iso_dt(v):
    """datetime 原样返回；ISO 字符串解析为 aware datetime（"Z"→+00:00，naive 视为 UTC
    ——Django≥5 已删除 django.utils.timezone.utc）。解析失败抛 ValueError/TypeError，
    错误策略由调用方自定：collect 视图整批 400（invalid_occurred_at），ingest v2 逐事件拒。"""
    if not isinstance(v, str):
        return v
    d = datetime.fromisoformat(v.replace("Z", "+00:00"))
    return d if d.tzinfo else d.replace(tzinfo=dt_tz.utc)

def sanitize_url(url: str) -> str:
    """三轨制清洗（spec §2.4）：黑名单参数名删 + 归因白名单提 + 其余按值判别。
    另：userinfo 剥离、fragment 删除。规则版本 SANITIZE_RULE_VERSION。"""
    p = urlparse(url)
    kept = []
    for k, v in parse_qsl(p.query, keep_blank_values=True):
        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):
    """单次递归同时做深度检查、敏感键扫描与字符串超长检查。
    dict 所有层级的键都过 _SENSITIVE，所有层级的字符串都受 MAX_STR_LEN 约束。
    返回 (depth_ok, sensitive_hit, str_too_long)。深度超限即停止下探，防恶意深嵌套递归爆栈。"""
    if depth > C.MAX_PROP_DEPTH:
        return False, False, False
    if isinstance(v, dict):
        depth_ok, sens, too_long = True, False, False
        for k, x in v.items():
            if _SENSITIVE.search(str(k)):
                sens = True
            d, s, l = _walk_props(x, depth + 1)
            depth_ok = depth_ok and d
            sens = sens or s
            too_long = too_long or l
        return depth_ok, sens, too_long
    if isinstance(v, list):
        depth_ok, sens, too_long = True, False, False
        for x in v:
            d, s, l = _walk_props(x, depth + 1)
            depth_ok = depth_ok and d
            sens = sens or s
            too_long = too_long or l
        return depth_ok, sens, too_long
    if isinstance(v, str) and len(v) > C.MAX_STR_LEN:
        return True, False, True
    return True, False, False

def sanitize_payload(obj):
    """原始批次快照脱敏：递归遍历 dict/list，键名命中 SENSITIVE_KEY_PATTERNS（大小写不敏感）
    → 值替换为字符串 "[FILTERED]"（结构保留，便于排查时仍看清字段形态）。
    返回新对象（容器全部重建，叶值原样共享）——调用方在 ingest 之前对 payload 深拷贝后传入，
    与 ingest/validate_event 的就地改写互不影响。"""
    if isinstance(obj, dict):
        return {k: ("[FILTERED]" if _SENSITIVE.search(str(k)) else sanitize_payload(v))
                for k, v in obj.items()}
    if isinstance(obj, list):
        return [sanitize_payload(x) for x in obj]
    return obj


def validate_event(ev, *, v2=False, batch_site_id="", batch_visitor_id="") -> str | None:
    """返回 None 表示通过；否则返回拒绝原因码。类型/白名单/体积/深度全在此守门。
    通过时会就地截断/清洗并写回 ev（path 先 sanitize_url 再截 MAX_PATH_LEN，title 截
    MAX_TITLE_LEN）——Task 6 的 ingest 依赖此写回契约。

    v2=True（批级 schema_version==2）：额外校验事件级归属四元组——site_id/visitor_id
    必须与批级一致、session_id 非空 str（归属依据从 v1 信封转移到事件级）；
    page_id 必须为非空 ≤64 的合法 UTF-8（session_resume 可空——它归属会话而非页面）。
    不一致 → attribution_mismatch；page_id 非法 → invalid_page_id。"""
    if not isinstance(ev, dict):
        return "event_invalid_shape"
    if ev.get("event_name") not in C.EVENT_NAMES:
        return "event_name_not_allowed"
    if v2:
        if (ev.get("site_id") != batch_site_id
                or ev.get("visitor_id") != batch_visitor_id
                or not isinstance(ev.get("session_id"), str) or not ev["session_id"]):
            return "attribution_mismatch"
        pid = ev.get("page_id")
        if isinstance(pid, str) and pid:
            # 非空 page_id 全事件统一完整校验——session_resume 也不豁免：合规 JSON 可携带
            # 孤代理（json.loads 接受 "\ud800"），漏检 _utf8_ok 会在存储时炸整批事务
            if len(pid) > 64 or not _utf8_ok(pid):
                return "invalid_page_id"
        elif pid not in ("", None):
            return "invalid_page_id"              # 非 str 类型（123/[] 等）
        elif ev["event_name"] != "session_resume":
            return "invalid_page_id"              # 空/缺失：仅 session_resume 豁免（归属会话而非页面）
    eid = ev.get("event_id")
    if not isinstance(eid, str) or not eid or len(eid) > 64 or not _utf8_ok(eid):
        return "invalid_event_id"
    for k in ("path", "title"):
        v = ev.get(k)
        if v is not None and (not isinstance(v, str) or not _utf8_ok(v)):
            return "invalid_field_type"
    occ = ev.get("occurred_at")
    if not isinstance(occ, datetime):
        return "invalid_occurred_at"
    if timezone.is_naive(occ):
        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", {})
    if not isinstance(props, dict) or len(props) > C.MAX_PROP_KEYS:
        return "properties_invalid"          # 传了但非 dict（[]/""/0/None）一律拒
    # 敏感键扫描（含嵌套层级）先于属性白名单：白名单内无任何键命中敏感模式，
    # 若放在白名单之后，sensitive_field 将不可达（见 test_sensitive_property_rejected）。
    depth_ok, sensitive_hit, str_too_long = _walk_props(props)
    if sensitive_hit:
        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 拒；
    # NaN/Infinity 同拒（allow_nan=False）——json.loads 默认收这两个字面量，落库时 SQLite 的
    # JSON 有效性校验拒 + bulk_create(ignore_conflicts) 静默丢行，而 accepted 已计数（幽灵事件）
    try:
        prop_bytes = len(json.dumps(props, ensure_ascii=False, default=str,
                                    allow_nan=False).encode("utf-8"))
    except (UnicodeEncodeError, ValueError):
        return "properties_invalid"
    if prop_bytes > C.MAX_PROP_JSON_BYTES:
        return "properties_too_large"
    if str_too_long:
        return "properties_too_long"
    # 先清洗后截断：urlencode 对中文查询参数最多膨胀 3 倍，先截后洗会破坏"出门 ≤MAX_PATH_LEN"不变量
    if ev.get("path"):
        ev["path"] = sanitize_url(ev["path"])
    if len(ev.get("path") or "") > C.MAX_PATH_LEN:
        ev["path"] = ev["path"][:C.MAX_PATH_LEN]
    # referrer 是完整 URL：同样过参数清洗（email/token 等整键剔除）再截 MAX_STR_LEN。
    # 放在 str_too_long 检查之后——超长 referrer 仍按 properties_too_long 拒，不静默截半。
    if ev["event_name"] == "page_view" and isinstance(props.get("referrer"), str) and props["referrer"]:
        props["referrer"] = sanitize_url(props["referrer"])[:C.MAX_STR_LEN]
    if len(ev.get("title") or "") > C.MAX_TITLE_LEN:
        ev["title"] = ev["title"][:C.MAX_TITLE_LEN]
    return None