diff --git a/server/config/urls.py b/server/config/urls.py index 865a87e..fe5268d 100644 --- a/server/config/urls.py +++ b/server/config/urls.py @@ -1 +1,6 @@ -urlpatterns = [] # Task 7 起逐任务填充 +from django.urls import path +from tracker.api.collect import CollectBatchView + +urlpatterns = [ + path("api/collect/batch", CollectBatchView.as_view()), +] diff --git a/server/tracker/api/__init__.py b/server/tracker/api/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/server/tracker/api/collect.py b/server/tracker/api/collect.py new file mode 100644 index 0000000..a221ead --- /dev/null +++ b/server/tracker/api/collect.py @@ -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 diff --git a/server/tracker/tests/test_collect.py b/server/tracker/tests/test_collect.py new file mode 100644 index 0000000..56344af --- /dev/null +++ b/server/tracker/tests/test_collect.py @@ -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