fix!: follow capture redirects only to the configured origin

.sampo/changesets/capture-same-origin-redirects.md

@@ -0,0 +1,5 @@
+---
+pypi/posthog: major
+---
+
+Capture follows a `307` or `308` redirect only to the origin of `host`, at most 5 times, for the sync and async clients. Any other redirect fails the batch and reaches `on_error`. In 7.x the sync client followed redirects to any origin and resent the event batch there.

docs/migration-7.x-to-8.0.md

@@ -12,6 +12,7 @@ You need to change code if your app does any of these:
 - sets `capture_mode`, `POSTHOG_CAPTURE_MODE` or `gzip`
 - imports `CaptureV1Error`, `posthog.capture_v1`, `request.batch_post`, `EVENTS_ENDPOINT` or `AI_EVENTS_ENDPOINT`
 - sends events to a self-hosted PostHog that does not serve the capture v1 endpoints
+- sends events through a proxy that redirects capture requests to another host
 - reuses one event `uuid` for more than one event
 - sets `$process_person_profile` to turn person processing on for events without a distinct ID
 - passes strings such as `"true"` for `$cookieless_mode`, `$ignore_sent_at` or `$process_person_profile`
@@ -31,6 +32,11 @@ You need to change code if your app does any of these:
 If you send events to a self-hosted PostHog, check that it serves both endpoints before you upgrade.
 An endpoint that is not served drops every event sent to it.
 
+The SDK follows a `307` or `308` redirect only to the origin of `host` (same scheme, host and port), at most 5 times.
+Any other redirect fails the batch and reaches `on_error`.
+In 7.x the sync client followed redirects to any origin.
+If a proxy redirects capture requests to another host, point `host` at the final host.
+
 ## SDK identity
 
 PostHog sets `$lib` and `$lib_version` on every event from the `PostHog-Sdk-Info` request header, which is always `posthog-python/<version>`.

posthog/capture_send.py

@@ -29,6 +29,7 @@
 from gzip import GzipFile
 from io import BytesIO
 from typing import TYPE_CHECKING, Optional
+from urllib.parse import urljoin, urlsplit
 from uuid import UUID
 
 from posthog.capture_compression import CaptureCompression, _zstandard
@@ -72,6 +73,9 @@
 # HTTP status classification. 429 is terminal in v1 (unlike v0, where it is
 # retried) — the backend signals overload via retryable 5xx + Retry-After.
 _RETRYABLE_STATUSES = frozenset({408, 500, 502, 503, 504})
+# Only these keep the method and body, so a resent batch is unchanged.
+_REDIRECT_STATUSES = frozenset({307, 308})
+_MAX_REDIRECTS = 5
 _TERMINAL_STATUSES = frozenset({400, 401, 402, 413, 415, 429})
 
 # Single ceiling (seconds) for the retry backoff: caps the exponential schedule
@@ -275,6 +279,34 @@ def _compress_v1(
     return data, None
 
 
+def _origin(url: str) -> Optional[tuple[str, str, int]]:
+    parsed = urlsplit(url)
+    try:
+        port = parsed.port
+    except ValueError:
+        return None
+    if port is None:
+        port = 443 if parsed.scheme.lower() == "https" else 80
+    return parsed.scheme.lower(), (parsed.hostname or "").lower(), port
+
+
+def _same_origin_redirect_url(
+    base_url: str, current_url: str, location: Optional[str]
+) -> Optional[str]:
+    """Return the redirect target on ``base_url``'s origin, or ``None``."""
+    if not location:
+        return None
+    target = urlsplit(urljoin(current_url, location))
+    base_origin = _origin(base_url)
+    if base_origin is None or _origin(target.geturl()) != base_origin:
+        return None
+    return (
+        urlsplit(base_url)
+        ._replace(path=target.path or "/", query=target.query, fragment="")
+        .geturl()
+    )
+
+
 def _post_v1(
     api_key: str,
     host: Optional[str],
@@ -296,6 +328,11 @@ def _post_v1(
     retries. The body is compressed per ``compression`` (advertised via
     ``Content-Encoding``). Returns the raw response; classification is left to
     the caller.
+
+    Follows only 307/308 redirects to the origin of ``host``, at most
+    :data:`_MAX_REDIRECTS` times, resending the same body and headers. Any other
+    redirect comes back as the response, which the caller treats as a terminal
+    failure, so a batch never goes to another origin.
     """
     trimmed_host = remove_trailing_slash(normalize_host(host))
     url = trimmed_host + path
@@ -314,9 +351,25 @@ def _post_v1(
         headers["Content-Encoding"] = encoding
 
     log.debug("capture v1 POST %s attempt=%s request_id=%s", url, attempt, request_id)
-    return (session or _get_session()).post(
-        url, data=body, headers=headers, timeout=timeout
-    )
+    http = session or _get_session()
+    for _ in range(_MAX_REDIRECTS + 1):
+        res = http.post(
+            url, data=body, headers=headers, timeout=timeout, allow_redirects=False
+        )
+        if res.status_code not in _REDIRECT_STATUSES:
+            return res
+        target = _same_origin_redirect_url(
+            trimmed_host, url, res.headers.get("Location")
+        )
+        if target is None:
+            log.warning(
+                "capture v1 did not follow a %s redirect to another origin",
+                res.status_code,
+            )
+            return res
+        url = target
+    log.warning("capture v1 stopped after %d redirects", _MAX_REDIRECTS)
+    return res
 
 
 def _parse_v1_response(res: "requests.Response") -> _V1ParsedResponse:

posthog/test/test_capture_send.py

@@ -49,17 +49,23 @@ def json(self):
 
 
 class _RecordingSession:
-    """Captures the args of a single ``.post`` and returns a canned response."""
+    """Records each ``.post`` and returns canned responses, repeating the last."""
 
-    def __init__(self, response):
-        self._response = response
+    def __init__(self, *responses):
+        self._responses = list(responses)
         self.calls = []
 
-    def post(self, url, data=None, headers=None, timeout=None):
+    def post(self, url, data=None, headers=None, timeout=None, allow_redirects=True):
         self.calls.append(
-            {"url": url, "data": data, "headers": headers, "timeout": timeout}
+            {
+                "url": url,
+                "data": data,
+                "headers": headers,
+                "timeout": timeout,
+                "allow_redirects": allow_redirects,
+            }
         )
-        return self._response
+        return self._responses[min(len(self.calls), len(self._responses)) - 1]
 
 
 class _PostV1Stub:
@@ -212,6 +218,90 @@ def test_zstd_without_package_raises_actionable_error(self) -> None:
                 self._post(_results_response({}), compression=CaptureCompression.ZSTD)
         self.assertIn("posthog[zstd]", str(ctx.exception))
 
+    @parameterized.expand(
+        [
+            (
+                "relative_location",
+                "https://us.i.posthog.com",
+                "/i/v1/analytics/events?retry=1",
+                "https://us.i.posthog.com/i/v1/analytics/events?retry=1",
+            ),
+            (
+                "host_path_prefix",
+                "https://example.com/ingest",
+                "https://example.com/ingest/i/v1/analytics/events",
+                "https://example.com/ingest/i/v1/analytics/events",
+            ),
+            (
+                "explicit_default_port",
+                "https://us.i.posthog.com",
+                "https://us.i.posthog.com:443/other",
+                "https://us.i.posthog.com/other",
+            ),
+        ]
+    )
+    def test_follows_same_origin_redirect_with_same_body(
+        self, _name, host, location, expected_url
+    ) -> None:
+        final = _results_response({})
+        session = _RecordingSession(
+            _FakeResponse(307, headers={"Location": location}), final
+        )
+        body = _build_v1_batch_body([_to_v1_event(_msg("u-1"))])
+        res = _post_v1(
+            "phc_key", host, body, attempt=1, request_id="r", session=session
+        )
+
+        self.assertIs(res, final)
+        self.assertEqual([c["url"] for c in session.calls][1], expected_url)
+        self.assertEqual(session.calls[0]["data"], session.calls[1]["data"])
+        self.assertEqual(session.calls[0]["headers"], session.calls[1]["headers"])
+        self.assertTrue(all(c["allow_redirects"] is False for c in session.calls))
+
+    @parameterized.expand(
+        [
+            ("other_host", 307, "https://attacker.example.com/collect"),
+            ("loopback", 308, "http://127.0.0.1:8080/collect"),
+            ("https_to_http", 307, "http://us.i.posthog.com/i/v1/analytics/events"),
+            ("other_port", 308, "https://us.i.posthog.com:8443/i/v1/analytics/events"),
+            ("missing_location", 307, None),
+            ("not_307_or_308", 302, "/i/v1/analytics/events"),
+        ]
+    )
+    def test_does_not_follow_other_redirects(self, _name, status, location) -> None:
+        redirect = _FakeResponse(
+            status, headers={"Location": location} if location else {}
+        )
+        session = _RecordingSession(redirect, _results_response({}))
+        body = _build_v1_batch_body([_to_v1_event(_msg("u-1"))])
+        res = _post_v1(
+            "phc_key",
+            "https://us.i.posthog.com",
+            body,
+            attempt=1,
+            request_id="r",
+            session=session,
+        )
+
+        self.assertIs(res, redirect)
+        self.assertEqual(len(session.calls), 1)
+
+    def test_stops_after_max_redirects(self) -> None:
+        loop = _FakeResponse(307, headers={"Location": "/i/v1/analytics/events"})
+        session = _RecordingSession(loop)
+        body = _build_v1_batch_body([_to_v1_event(_msg("u-1"))])
+        res = _post_v1(
+            "phc_key",
+            "https://us.i.posthog.com",
+            body,
+            attempt=1,
+            request_id="r",
+            session=session,
+        )
+
+        self.assertIs(res, loop)
+        self.assertEqual(len(session.calls), 6)
+
 
 class TestParseV1Response(unittest.TestCase):
     def test_success_parses_results_with_details(self) -> None:
@@ -445,7 +535,14 @@ def test_malformed_2xx_is_terminal(self) -> None:
         self.assertEqual(len(stub.calls), 1)
         self.assertEqual(exc.status, 200)
 
-    @parameterized.expand([("bad_request", 400), ("rate_limited", 429)])
+    @parameterized.expand(
+        [
+            ("bad_request", 400),
+            ("rate_limited", 429),
+            ("unfollowed_redirect", 307),
+            ("unfollowed_permanent_redirect", 308),
+        ]
+    )
     def test_terminal_status_raises_immediately(self, _name, status) -> None:
         stub, exc = self._run_expecting_error(
             [_msg("u-1")],

typings/requests/__init__.pyi

@@ -36,6 +36,7 @@ class Session:
         headers: dict[str, str],
         timeout: float,
         stream: bool = ...,
+        allow_redirects: bool = ...,
     ) -> Response: ...
     def get(
         self,