feat: 采集接口（域名白名单/限流/textplain解析/幂等）

POST /api/collect/batch：text/plain JSON 解析（ParseError 400 防 500）、域名白名单/站点停用/未知站点统一 403 not_allowed 防枚举、 Anon 限流（collect scope，get_rate 读实时 api_settings 修 import 固化）、 occurred_at ISO 归一（naive 补 UTC）、XFF 末段取 IP + ::ffff: 归一、成功/拒绝分支均回显 ACAO 供 SDK 读响应。 Co-Authored-By: Claude Code <noreply@anthropic.com>
2026-09-15 13:07:56 +08:00
parent acb180e363
commit cf04db794b
4 changed files with 170 additions and 1 deletions
@@ -0,0 +1,101 @@
+import json
+from datetime import datetime, timezone as dt_tz
+from rest_framework.views import APIView
+from rest_framework.response import Response
+from rest_framework.permissions import AllowAny
+from rest_framework.throttling import AnonRateThrottle
+from rest_framework.parsers import BaseParser
+from rest_framework.settings import api_settings
+from tracker.models import Site
+from tracker.services.ingest import ingest_batch
+
+class PlainJSONParser(BaseParser):
+    media_type = "text/plain"
+    def parse(self, stream, *args, **kwargs):
+        try:
+            return json.loads(stream.read().decode("utf-8") or "{}")
+        except (ValueError, UnicodeDecodeError):
+            from rest_framework.exceptions import ParseError
+            raise ParseError("invalid_json")   # DRF 对 parser 异常是 re-raise 不是包装，不自包就是 500
+
+class CollectThrottle(AnonRateThrottle):
+    scope = "collect"
+
+    def get_rate(self):
+        # SimpleRateThrottle.THROTTLE_RATES 在 import 时固化为类属性，
+        # 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]
+    throttle_classes = [CollectThrottle]
+    authentication_classes = []
+
+    def post(self, request):
+        origin = request.headers.get("Origin") or request.headers.get("Referer") or ""
+        if int(request.headers.get("Content-Length") or 0) > 262144:
+            return self._deny(400, "payload_too_large")   # 应用层兜底（nginx 另设 client_max_body_size）
+        site = self._match_site(request)
+        if site is None or not site.is_active:
+            return self._deny(403, "not_allowed")         # 三种拒绝统一 reason，防 site_id 枚举探测
+        host = (origin.split("//")[-1].split("/")[0].split(":")[0].lower()
+                if origin else "")
+        if host and host not in [d.lower() for d in site.domains]:
+            return self._deny(403, "not_allowed")
+        if not host and not site.allow_no_origin:
+            return self._deny(403, "not_allowed")
+        payload = request.data if isinstance(request.data, dict) else {}
+        for ev in payload.get("events") or []:
+            if isinstance(ev, dict) and "occurred_at" in ev:
+                try:
+                    ev["occurred_at"] = _iso(ev["occurred_at"])
+                except (ValueError, TypeError):
+                    return self._deny(400, "invalid_occurred_at")
+        if not payload.get("visitor_id") or not payload.get("session_id"):
+            return self._deny(400, "invalid_payload")
+        ip = self._client_ip(request)
+        r = ingest_batch(site, payload, ip, request.headers.get("User-Agent", ""))
+        resp = Response({"code": 200,
+            "data": {"accepted": r["accepted"], "duplicated": r["duplicated"]},
+            "message": "ok"})
+        if origin:   # 成功分支回显 ACAO（SDK fetch 需读响应）
+            resp["Access-Control-Allow-Origin"] = origin
+            resp["Access-Control-Allow-Headers"] = "content-type"
+        return resp
+
+    def options(self, request):   # 预检兜底（SDK text/plain 简单请求通常不触发）
+        origin = request.headers.get("Origin") or ""
+        resp = Response(status=200)
+        if origin:
+            resp["Access-Control-Allow-Origin"] = origin
+            resp["Access-Control-Allow-Methods"] = "POST, OPTIONS"
+            resp["Access-Control-Allow-Headers"] = "content-type"
+        return resp
+
+    def _match_site(self, request):
+        sid = (request.data or {}).get("site_id") if isinstance(request.data, dict) else None
+        return Site.objects.filter(site_id=sid).first() if sid else None
+
+    def _client_ip(self, request):
+        """nginx 前置时取 X-Forwarded-For 最后一段（与 DRF NUM_PROXIES=1 一致；套 CDN 时按层数前移）。"""
+        xff = request.META.get("HTTP_X_FORWARDED_FOR", "")
+        ip = xff.split(",")[-1].strip() if xff else request.META.get("REMOTE_ADDR")
+        # 归一 IPv4 映射地址（::ffff:1.2.3.4 → 1.2.3.4），防地理解析丢失
+        if ip and ip.lower().startswith("::ffff:"):
+            ip = ip[7:]
+        return ip
+
+    def _deny(self, code, reason):
+        resp = Response({"code": code, "data": None, "message": reason}, status=code)
+        origin = self.request.headers.get("Origin")
+        if origin:   # 拒绝也要回 ACAO，否则浏览器读不到错误体、SDK 误判为网络错误走重试
+            resp["Access-Control-Allow-Origin"] = origin
+        return resp
@@ -0,0 +1,63 @@
+import json
+import pytest
+from django.core.cache import cache
+from django.utils import timezone
+from tracker.models import Site
+
+NOW = timezone.now()
+
+@pytest.fixture(autouse=True)
+def _clear_cache():
+    cache.clear()   # locmem 限流历史跨测试残留，会让 429 用例假红
+
+def _post(client, body, origin="https://www.a.com", extra=None):
+    return client.post("/api/collect/batch", data=json.dumps(body),
+        content_type="text/plain", HTTP_ORIGIN=origin, **(extra or {}))
+
+def _body(n=1, sid="s1", vid="v1"):
+    return {"site_id": "site_x", "visitor_id": vid, "session_id": sid,
+            "context": {"referrer_domain": "", "path": "/", "device_type": "PC"},
+            "events": [{"event_id": f"e{i}", "event_name": "page_view",
+                        "occurred_at": NOW.isoformat(), "path": "/", "title": "t",
+                        "properties": {}} for i in range(n)]}
+
+def test_collect_ok_creates_events(db, client):
+    Site.objects.create(name="s", domains=["www.a.com"], site_id="site_x")
+    r = _post(client, _body(2))
+    assert r.status_code == 200 and r.json() == {"code": 200, "data": {"accepted": 2, "duplicated": 0}, "message": "ok"}
+    assert r["Access-Control-Allow-Origin"] == "https://www.a.com"
+
+def test_origin_not_whitelisted_rejected(db, client):
+    Site.objects.create(name="s", domains=["www.a.com"], site_id="site_x")
+    r = _post(client, _body(1), origin="https://evil.com")
+    assert r.status_code == 403 and r.json()["code"] == 403
+    assert r.json()["message"] == "not_allowed"     # 与未知站点同 message，防枚举
+
+def test_inactive_site_rejected(db, client):
+    Site.objects.create(name="s", domains=["www.a.com"], site_id="site_x", is_active=False)
+    assert _post(client, _body(1)).status_code == 403
+
+def test_unknown_site_403_not_404(db, client):   # 不泄露站点是否存在
+    r = _post(client, _body(1))
+    assert r.status_code == 403
+    assert r.json()["message"] == "not_allowed"   # message 与上例一致
+
+def test_replay_duplicated(db, client):
+    Site.objects.create(name="s", domains=["www.a.com"], site_id="site_x")
+    _post(client, _body(1))
+    r = _post(client, _body(1))
+    assert r.json()["data"]["duplicated"] == 1
+
+def test_invalid_json_400_not_500(db, client):
+    Site.objects.create(name="s", domains=["www.a.com"], site_id="site_x")
+    r = client.post("/api/collect/batch", data="{bad json",
+                    content_type="text/plain", HTTP_ORIGIN="https://www.a.com")
+    assert r.status_code == 400
+
+def test_throttle_429(db, client, settings):
+    settings.REST_FRAMEWORK = {**settings.REST_FRAMEWORK,
+        "DEFAULT_THROTTLE_RATES": {"collect": "3/min"}}
+    Site.objects.create(name="s", domains=["www.a.com"], site_id="site_x")
+    for _ in range(3):
+        assert _post(client, _body(1)).status_code == 200
+    assert _post(client, _body(1)).status_code == 429