diff --git a/server/tracker/api/collect.py b/server/tracker/api/collect.py index 2ec0cf4..e9ec22d 100644 --- a/server/tracker/api/collect.py +++ b/server/tracker/api/collect.py @@ -1,5 +1,4 @@ import json -from datetime import datetime, timezone as dt_tz from urllib.parse import urlsplit from uuid import uuid4 from django.utils import timezone as dj_timezone @@ -11,7 +10,7 @@ from rest_framework.parsers import BaseParser from rest_framework.settings import api_settings from tracker.models import RawBatch, Site, Visitor from tracker.services.ingest import ingest_batch -from tracker.services.validation import sanitize_payload +from tracker.services.validation import parse_iso_dt, sanitize_payload class PlainJSONParser(BaseParser): media_type = "text/plain" @@ -30,13 +29,6 @@ class CollectThrottle(AnonRateThrottle): # override_settings 改 DEFAULT_THROTTLE_RATES 对它不生效——改读实时 api_settings（DRF 收 setting_changed 会 reload） return api_settings.DEFAULT_THROTTLE_RATES[self.scope] -def _iso(dt): - if isinstance(dt, str): - s = dt.replace("Z", "+00:00") - d = datetime.fromisoformat(s) - return d if d.tzinfo else d.replace(tzinfo=dt_tz.utc) # Django≥5 已删除 django.utils.timezone.utc - return dt - class CollectBatchView(APIView): permission_classes = [AllowAny] parser_classes = [PlainJSONParser] @@ -66,7 +58,7 @@ class CollectBatchView(APIView): for ev in payload.get("events") or []: if isinstance(ev, dict) and "occurred_at" in ev: try: - ev["occurred_at"] = _iso(ev["occurred_at"]) + ev["occurred_at"] = parse_iso_dt(ev["occurred_at"]) # 共享解析（validation），视图错误策略=整批 400 except (ValueError, TypeError): return self._deny(400, "invalid_occurred_at", request_id) if not payload.get("visitor_id") or not payload.get("session_id"): diff --git a/server/tracker/services/ingest.py b/server/tracker/services/ingest.py index 702f246..afd1752 100644 --- a/server/tracker/services/ingest.py +++ b/server/tracker/services/ingest.py @@ -23,7 +23,8 @@ page_id），会话按事件级 session_id 解析（v1 信封不再是唯一依 （v2 附注：stay_ms 重算对"并发同会话双批"同样存在同类窗口——后提交者的 SELECT 可能 看不到前者未提交事件而少算，与 TOCTOU 同级、只影响 stay_ms 单字段。） """ -from datetime import datetime, timedelta, timezone as dt_tz +from datetime import timedelta +import math from django.db import transaction, IntegrityError from django.db.models import F, Q @@ -32,31 +33,25 @@ from django.utils import timezone from tracker import constants as C from tracker.models import SourceClassification, Visitor, VisitSession, Event -from tracker.services.validation import validate_event +from tracker.services.validation import validate_event, parse_iso_dt from tracker.services.attribution import classify from tracker.services.geo import resolver as _geo # 模块级单例（勿实例化，11MB/份） _FIELD_LEN = {"device_type": 16, "os": 64, "browser": 64, "language": 32, "screen": 32} _CLASS_FIELDS = ("channel", "platform", "variant", "search_term", "evidence_kind", "resolution", "detail") - - -def _occ_dt(v): - """v2 事件 occurred_at 允许 ISO 字符串（SDK 线格式；collect 视图已预转 datetime， - 直连调用方可能传原文）。解析失败抛 ValueError/TypeError——调用方放行原值， - 由 validate_event 统一拒 invalid_occurred_at。naive 视为 UTC（与 collect._iso 同口径）。""" - if isinstance(v, str): - d = datetime.fromisoformat(v.replace("Z", "+00:00")) - return d if d.tzinfo else d.replace(tzinfo=dt_tz.utc) - return v +_BIGINT_MAX = 2 ** 63 - 1 # BigIntegerField 8 字节量纲上限——超范围值炸列，按 None 降级 def _ms(v): - """v2 区间字段宽松 int 化：bool/非数值 → None（脏数据防御——recompute 会过滤 None 区间， - 该事件不参与并集但不拒批）。""" + """v2 区间字段宽松 int 化，任何脏值一律 None（recompute 过滤 None 区间，不拒批）： + bool/非数值；NaN/Infinity（json.loads 默认收这两个字面量，int() 抛 ValueError/ + OverflowError）；超 BigIntegerField 量纲的有限值（DB 写入炸列）。""" if isinstance(v, bool) or not isinstance(v, (int, float)): return None - return int(v) + if not math.isfinite(v): + return None + return int(v) if -_BIGINT_MAX <= v <= _BIGINT_MAX else None def _time_quality(now, occ) -> str: @@ -284,7 +279,7 @@ def ingest_batch(site, payload: dict, ip: str | None = None, ua: str = "") -> di for i, ev in enumerate(events): if v2 and isinstance(ev, dict) and isinstance(ev.get("occurred_at"), str): try: # v2 事件 occurred_at 允许 ISO 字符串（SDK 线格式）；解析失败放行原值， - ev["occurred_at"] = _occ_dt(ev["occurred_at"]) # validate_event 统一拒 invalid_occurred_at + ev["occurred_at"] = parse_iso_dt(ev["occurred_at"]) # validate_event 统一拒 invalid_occurred_at except (ValueError, TypeError): pass reason = validate_event(ev, v2=v2, batch_site_id=site.site_id, diff --git a/server/tracker/services/validation.py b/server/tracker/services/validation.py index fcf4c28..2161c1e 100644 --- a/server/tracker/services/validation.py +++ b/server/tracker/services/validation.py @@ -1,6 +1,6 @@ import json import re -from datetime import datetime +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 @@ -16,6 +16,16 @@ def _utf8_ok(s: str) -> bool: 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。""" @@ -96,9 +106,15 @@ def validate_event(ev, *, v2=False, batch_site_id="", batch_visitor_id="") -> st or not isinstance(ev.get("session_id"), str) or not ev["session_id"]): return "attribution_mismatch" pid = ev.get("page_id") - if ev["event_name"] != "session_resume" and ( - not isinstance(pid, str) or not pid or len(pid) > 64 or not _utf8_ok(pid)): - return "invalid_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" @@ -126,10 +142,13 @@ def validate_event(ev, *, v2=False, batch_site_id="", batch_visitor_id="") -> st return "property_not_allowed" if not depth_ok: return "properties_depth" - # 体积按 UTF-8 字节数守门（汉字多字节，按字符数会被绕过）；孤代理 encode 必炸，按 invalid 拒 + # 体积按 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).encode("utf-8")) - except UnicodeEncodeError: + 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" diff --git a/server/tracker/tests/test_ingest.py b/server/tracker/tests/test_ingest.py index 08078d3..6b0b35d 100644 --- a/server/tracker/tests/test_ingest.py +++ b/server/tracker/tests/test_ingest.py @@ -350,3 +350,37 @@ def test_dirty_intervals_soft_skipped(db): e2 = Event.objects.get(event_id="e-page_stay-2") assert e2.active_from_ms is None # 非数值 int 化为 None assert VisitSession.objects.get(session_id="s1").stay_ms == 10000 # 仅 p1 有效区间 + +# 27. NaN/Infinity/超量纲区间值：NaN/Inf 是非法 JSON 数值（json.loads 默认收，但 SQLite +# JSON 有效性校验拒 + bulk_create(ignore_conflicts) 静默丢行而 accepted 已计数——幽灵 +# 事件），入口 properties_invalid 拒；有限超量纲值（1e308）合法入库但区间列降级 None +def test_nonfinite_interval_values_rejected_or_degraded(db): + site = Site.objects.create(name="s", domains=["a.com"], site_id=SITE_ID) + evs = [_v2_event("page_stay", "p1", "s1", seq=1, active_from=0, active_to=10000, flush_reason="periodic"), + _v2_event("page_stay", "p2", "s1", seq=2, + active_from=float("nan"), active_to=float("inf"), flush_reason="periodic"), + _v2_event("page_stay", "p3", "s1", seq=3, + active_from=-1e308, active_to=1e308, flush_reason="periodic")] + r = ingest_batch(site, _v2_batch(evs, sid="s1")) + assert (r["accepted"], r["duplicated"]) == (2, 0) + assert r["rejected"] == [{"index": 1, "event_id": "e-page_stay-2", + "reason": "properties_invalid"}] + e3 = Event.objects.get(event_id="e-page_stay-3") + assert e3.active_from_ms is None and e3.active_to_ms is None # 超 ±2^63 量纲 → None + assert VisitSession.objects.get(session_id="s1").stay_ms == 10000 + +# 28. page_id 校验（invalid_page_id）：session_resume 仅豁免"空值"，非空值一律完整校验 +# （孤代理 "\ud800" 是合规 JSON，跳过 _utf8_ok 会在存储时炸整批） +def test_v2_page_id_validation(db): + site = Site.objects.create(name="s", domains=["a.com"], site_id=SITE_ID) + bad_surrogate = _v2_event("session_resume", "", "s1", seq=1, + prev_session_id="s0", reason="idle_timeout") + bad_surrogate["page_id"] = "p\ud800" # session_resume 的非空 page_id 也不豁免孤代理 + bad_long = _v2_event("page_view", "L" * 65, "s1", seq=2) + bad_type = _v2_event("page_view", 123, "s1", seq=3) + ok_resume = _v2_event("session_resume", "", "s1", seq=4, + prev_session_id="s0", reason="idle_timeout") + r = ingest_batch(site, _v2_batch([bad_surrogate, bad_long, bad_type, ok_resume], sid="s1")) + assert r["accepted"] == 1 # 空 page_id 的 session_resume 通过 + assert [x["reason"] for x in r["rejected"]] == ["invalid_page_id"] * 3 + assert {x["index"] for x in r["rejected"]} == {0, 1, 2}