"""Route-level tests for the public product-event beacon (POST /api/v2/event). The endpoint is unauthenticated and reachable by anything on the internet — on a site where ~60% of inbound traffic is crawlers and scanners — so these tests are mostly about what it REFUSES to do: grow storage, echo anything back, or accept a payload big enough to be worth abusing. """ import json import pytest from fastapi.testclient import TestClient from core import events from web import app as appmod EVENT_URL = "/thermograph/api/v2/event" # Browsers set this on fetch/sendBeacon; a bare scanner POST does not. FIRST_PARTY = {"sec-fetch-site": "same-origin"} @pytest.fixture(autouse=True) def _fresh_rate_limiter(): """TestClient always presents the same client IP, so without this one test's beacon traffic silently rate-limits the next one's.""" from core import metrics metrics.reset_rate_limiter() yield metrics.reset_rate_limiter() @pytest.fixture def client(): return TestClient(appmod.app) @pytest.fixture def recorded(monkeypatch): """Capture what reaches the recorder, with the flag forced on.""" seen = [] monkeypatch.setattr(events, "ENABLED", True) monkeypatch.setattr(events, "_upsert", lambda *a: None) monkeypatch.setattr(events.audit, "log_activity", lambda tag, rec: seen.append(rec)) return seen def test_beacon_always_answers_204_and_returns_no_body(client, recorded): """Counted, rejected, or rate-limited must be indistinguishable to a prober.""" for body in ({"event": "view.open", "props": {"view": "home"}}, {"event": "not-a-real-event"}, {"nonsense": True}, [1, 2, 3], "plain string"): r = client.post(EVENT_URL, content=json.dumps(body), headers=FIRST_PARTY) assert r.status_code == 204 assert r.content == b"" def test_valid_event_is_recorded_with_allowlisted_dimensions(client, recorded): client.post(EVENT_URL, json={"event": "view.control", "props": {"view": "compare", "prop": "unit", "value": "C"}}, headers=FIRST_PARTY) assert recorded == [{"event": "view.control", "view": "compare", "prop": "unit", "value": "C", "referrer": ""}] def test_batched_payload_is_accepted_and_capped(client, recorded): items = [{"event": "share", "props": {"prop": "link"}}] * (events.MAX_BATCH + 5) client.post(EVENT_URL, json={"events": items}, headers=FIRST_PARTY) assert len(recorded) == events.MAX_BATCH def test_a_full_batch_of_real_events_fits_under_the_byte_cap(client, recorded): """Regression: if MAX_BODY_BYTES ever drops below a full MAX_BATCH payload, the byte cap silently eats legitimate batches instead of abusive ones.""" items = [{"event": "view.control", "props": {"view": "compare", "prop": "chart_metric", "value": "precip"}} ] * events.MAX_BATCH assert len(json.dumps({"events": items})) < events.MAX_BODY_BYTES client.post(EVENT_URL, json={"events": items}, headers=FIRST_PARTY) assert len(recorded) == events.MAX_BATCH def test_oversized_body_is_dropped_without_being_parsed(client, recorded): """The endpoint must not become storage: a body past the cap is never read into a record at all.""" fat = {"event": "view.open", "props": {"view": "home"}, "junk": "x" * (events.MAX_BODY_BYTES * 4)} r = client.post(EVENT_URL, content=json.dumps(fat), headers=FIRST_PARTY) assert r.status_code == 204 assert recorded == [] def test_free_text_and_coordinates_never_survive(client, recorded): """Anything a client puts in a dimension slot is allowlisted or bucketed — a leaked query string or lat/lon cannot reach a sink.""" client.post(EVENT_URL, json={"event": "place.pick", "props": {"view": "home", "prop": "47.6062,-122.3321"}}, headers=FIRST_PARTY) client.post(EVENT_URL, json={"event": "view.control", "props": {"view": "home", "prop": "chart_metric", "value": "user@example.com"}}, headers=FIRST_PARTY) stored = {v for rec in recorded for v in rec.values()} assert not any("47.6" in s or "@" in s for s in stored) def test_cross_site_post_is_dropped(client, recorded): r = client.post(EVENT_URL, json={"event": "view.open", "props": {"view": "home"}}, headers={"sec-fetch-site": "cross-site"}) assert r.status_code == 204 assert recorded == [] def test_flag_off_records_nothing(client, monkeypatch): """The whole feature ships inert: with THERMOGRAPH_EVENTS unset the beacon still answers, and nothing reaches the durable sink.""" seen = [] monkeypatch.setattr(events, "ENABLED", False) monkeypatch.setattr(events, "_upsert", lambda *a: seen.append(a)) monkeypatch.setattr(events.audit, "log_activity", lambda tag, rec: seen.append(rec)) r = client.post(EVENT_URL, json={"event": "view.open", "props": {"view": "home"}}, headers=FIRST_PARTY) assert r.status_code == 204 assert seen == [] def test_legacy_single_event_shape_still_counts(client, monkeypatch): """The four home.* aggregate counters already in production keep working — they are unflagged and predate the schema.""" from core import metrics seen = [] monkeypatch.setattr(metrics, "_store", type("S", (), { "record_event": lambda self, *a: seen.append(a)})()) client.post(EVENT_URL, json={"event": "home.locate"}, headers=FIRST_PARTY) assert seen and seen[0][0] == "home.locate" def test_beacon_is_not_counted_as_inbound_traffic(client): """It reports traffic; counting it would double every interaction.""" from core import metrics assert metrics.classify_inbound(EVENT_URL, "/thermograph") == "event" # --- Tier B over the wire ------------------------------------------------------- @pytest.fixture def session_rows(monkeypatch, recorded): rows = [] monkeypatch.setattr(events, "SESSIONS_ENABLED", True) monkeypatch.setattr(events, "_insert_session", lambda *a: rows.append(a)) return rows def test_session_and_seq_travel_with_a_batch(client, session_rows): client.post(EVENT_URL, json={ "session": "D" * 22, "events": [{"event": "view.open", "props": {"view": "home"}, "seq": 1}, {"event": "place.pick", "props": {"view": "home", "prop": "map"}, "seq": 2}], }, headers=FIRST_PARTY) assert [(r[0], r[1], r[2]) for r in session_rows] == [ ("D" * 22, 1, "view.open"), ("D" * 22, 2, "place.pick")] def test_a_forged_session_field_cannot_smuggle_text(client, session_rows, recorded): """The session field is the only client-controlled string that reaches a sink, so a malformed one must lose its payload entirely — while the anonymous aggregate still records.""" for forged in ("'; DROP TABLE ui_event_session; --", "user@example.com", "47.6062,-122.3321", "x" * 500): client.post(EVENT_URL, json={"session": forged, "seq": 1, "event": "view.open", "props": {"view": "home"}}, headers=FIRST_PARTY) assert session_rows == [] assert len(recorded) == 4 # Tier A recorded every one of them def test_sessions_flag_off_records_only_the_aggregate(client, monkeypatch, recorded): rows = [] monkeypatch.setattr(events, "SESSIONS_ENABLED", False) monkeypatch.setattr(events, "_insert_session", lambda *a: rows.append(a)) client.post(EVENT_URL, json={"session": "E" * 22, "seq": 1, "event": "view.open", "props": {"view": "home"}}, headers=FIRST_PARTY) assert rows == [] and len(recorded) == 1 def test_beacon_never_reads_the_account_session_cookie(client, session_rows, recorded): """Even a signed-in visitor's events stay unlinked to the account: the route does not consult the auth cookie, so there is nothing to join on.""" client.cookies.set("thermographauth", "pretend-jwt-value") client.post(EVENT_URL, json={"event": "view.open", "props": {"view": "home"}}, headers=FIRST_PARTY) assert "pretend-jwt" not in str(recorded) + str(session_rows) assert recorded and all("user" not in k for rec in recorded for k in rec)