浅念
cb57e67179
TDD：7 测试先红（ModuleNotFoundError）后绿。与计划的一处偏差： 敏感键扫描移至属性白名单之前——白名单内无任何键命中敏感模式， 原顺序下 sensitive_field 不可达，test_sensitive_property_rejected （user_password 非白名单键）将得到 property_not_allowed 而失败。 Co-Authored-By: Claude Code <noreply@anthropic.com>
69 lines
2.9 KiB
Python
69 lines
2.9 KiB
Python
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
|