fix(api): keep configured headers on their own origin

tests/test_api_cross_origin_token.py

@@ -20,6 +20,7 @@
 import http.server
 import threading
 import pytest
+from urllib.parse import parse_qs, quote, urlsplit
 
 from tina4_python.api import Api
 
@@ -30,6 +31,7 @@ class _RecordingServer:
     def __init__(self):
         self.last_auth = None
         self.last_headers = {}
+        self.last_path = None
         outer = self
 
         class Handler(http.server.BaseHTTPRequestHandler):
@@ -41,7 +43,15 @@ def log_message(self, *a):
             def do_GET(self):
                 outer.last_auth = self.headers.get("Authorization")
                 outer.last_headers = {k.lower(): v for k, v in self.headers.items()}
+                outer.last_path = self.path
                 self.rfile.read(int(self.headers.get("Content-Length", "0")))
+                if self.path.startswith("/redirect?"):
+                    query = parse_qs(urlsplit(self.path).query)
+                    self.send_response(int(query.get("code", ["302"])[0]))
+                    self.send_header("Location", query["to"][0])
+                    self.send_header("Content-Length", "0")
+                    self.end_headers()
+                    return
                 payload = b'{"ok": true}'
                 self.send_response(200)
                 self.send_header("Content-Type", "application/json")
@@ -104,3 +114,53 @@ def test_final_target_credentials_on_every_http_path(mode, off_origin, tmp_path)
         assert result["http_code"] == 200
         assert bool(target.last_headers.get("authorization")) is (not off_origin)
         assert bool(target.last_headers.get("cookie")) is (not off_origin)
+
+
+# A configured header can carry a credential under any name. It is bound to the
+# base origin exactly like the token: never to an absolute target on another
+# origin, never onto a cross-origin redirect hop.
+
+def _redirect(to, code=302):
+    return f"/redirect?code={code}&to={quote(to, safe='')}"
+
+
+@pytest.mark.parametrize("how", ["ctor", "add_headers"])
+def test_configured_header_stays_off_an_absolute_off_origin_target(how):
+    with _RecordingServer() as base, _RecordingServer() as other:
+        if how == "ctor":
+            api = Api(base.base_url, headers={"X-Api-Key": "synthetic-key"})
+        else:
+            api = Api(base.base_url)
+            api.add_headers({"X-Api-Key": "synthetic-key"})
+        assert api.get(f"{other.base_url}/probe")["http_code"] == 200
+        assert "x-api-key" not in other.last_headers, f"{how} key leaked to an absolute off-origin URL"
+        api.get("/probe")
+        assert base.last_headers.get("x-api-key") == "synthetic-key", f"{how} key lost on its own origin"
+
+
+def test_configured_and_per_call_headers_stay_off_a_cross_origin_redirect(tmp_path):
+    with _RecordingServer() as base, _RecordingServer() as other:
+        api = Api(base.base_url, headers={"X-Api-Key": "synthetic-key", "Accept": "application/json"})
+        assert api.get(_redirect(f"{other.base_url}/landed"))["http_code"] == 200
+        assert "x-api-key" not in other.last_headers, "configured key followed a redirect to another origin"
+        assert other.last_headers.get("accept") == "application/json", "content negotiation must still cross"
+
+        # urllib follows a POST only on 301/302/303 (re-issued as a GET); a 307
+        # POST is not followed in this port.
+        source = tmp_path / "upload.txt"; source.write_text("upload")
+        api.upload(_redirect(f"{other.base_url}/uploaded", 302), file_path=str(source),
+                   headers={"X-Upload-Token": "synthetic-call"})
+        assert other.last_path == "/uploaded", "the redirect hop never reached the other origin"
+        assert "x-upload-token" not in other.last_headers, "per-call header followed a redirect to another origin"
+
+        api.get(_redirect(f"{base.base_url}/landed"))
+        assert base.last_headers.get("x-api-key") == "synthetic-key", "same-origin redirect lost the key"
+
+
+def test_baseless_client_sends_configured_header_only_to_the_url_it_names():
+    with _RecordingServer() as named, _RecordingServer() as other:
+        api = Api(headers={"X-Api-Key": "synthetic-key"})
+        api.get(f"{named.base_url}/probe")
+        assert named.last_headers.get("x-api-key") == "synthetic-key"
+        api.get(named.base_url + _redirect(f"{other.base_url}/landed"))
+        assert "x-api-key" not in other.last_headers, "baseless key followed a redirect to another origin"

tina4_python/api/__init__.py

@@ -55,9 +55,20 @@
 # yields whatever the transport hands back per read; this bounds the buffer.
 _STREAM_CHUNK_SIZE = 8 * 1024
 
-# Headers dropped when a redirect crosses to a different origin — a bearer token
-# or a session cookie must never be handed to a host you didn't authenticate to.
-_STRIP_ON_CROSS_ORIGIN = frozenset({"authorization", "cookie"})
+# The only headers carried onto a different origin. A caller-set header can hold
+# a credential under any name (X-Api-Key, X-Auth-Token, ...), so a denylist
+# cannot contain it: the headers configured on the client (and any per-call
+# header, on a redirect) are bound to the origin they were meant for, and only
+# these content-negotiation headers cross.
+_KEEP_ON_CROSS_ORIGIN = frozenset({
+    "user-agent", "accept", "accept-encoding", "accept-language",
+    "content-type", "content-length",
+})
+
+
+def _keep_cross_origin_headers(headers: dict) -> dict:
+    """Only the headers allowed onto a different origin (_KEEP_ON_CROSS_ORIGIN)."""
+    return {k: v for k, v in headers.items() if k.lower() in _KEEP_ON_CROSS_ORIGIN}
 
 
 class ApiTimeoutError(TimeoutError):
@@ -115,8 +126,9 @@ def _same_origin(url_a: str, url_b: str) -> bool:
 
 
 class _AuthStripRedirectHandler(HTTPRedirectHandler):
-    """Follow redirects, but drop the Authorization (and Cookie) header on a
-    cross-origin hop.
+    """Follow redirects, but carry only content-negotiation headers
+    (``_KEEP_ON_CROSS_ORIGIN``) onto a cross-origin hop -- no Authorization,
+    Cookie, or any other caller-set header that may hold a credential.
 
     Plain urllib forwards the Authorization header to ANY redirect target,
     including a different host — so an ``api.get("/login")`` that 302s to
@@ -141,14 +153,9 @@ def redirect_request(self, req, fp, code, msg, headers, newurl):
         guard_url(newurl, self._allow_hosts)
         new_req = super().redirect_request(req, fp, code, msg, headers, newurl)
         if new_req is not None and not _same_origin(req.full_url, newurl):
-            new_req.headers = {
-                k: v for k, v in new_req.headers.items()
-                if k.lower() not in _STRIP_ON_CROSS_ORIGIN
-            }
-            new_req.unredirected_hdrs = {
-                k: v for k, v in getattr(new_req, "unredirected_hdrs", {}).items()
-                if k.lower() not in _STRIP_ON_CROSS_ORIGIN
-            }
+            new_req.headers = _keep_cross_origin_headers(new_req.headers)
+            new_req.unredirected_hdrs = _keep_cross_origin_headers(
+                getattr(new_req, "unredirected_hdrs", {}))
         return new_req
 
 
@@ -800,7 +807,14 @@ def _build_request(self, method: str, url: str, body, content_type: str,
         # ``add_headers()``, or a per-call ``extra_headers``) always wins --
         # this is a default, never a clobber.
         headers = {"User-Agent": f"Tina4/{__version__}"}
-        headers.update(self._headers)
+        # Headers configured on the client belong to its base origin, like the
+        # token below: an absolute target on another origin gets only the
+        # cross-origin-safe ones. A client with no base has no origin to bind
+        # them to, and sends them to the URL each call names.
+        if self.base_url and not _same_origin(url, self.base_url):
+            headers.update(_keep_cross_origin_headers(self._headers))
+        else:
+            headers.update(self._headers)
         # Attach the configured Authorization / Cookie ONLY when the request
         # target is same-origin as the configured base. A path that is itself an
         # absolute off-origin URL (e.g. get("http://evil/x")) otherwise leaks the