Release Core v0.1.33 with redirect-sensitive headers

src/govoplan_core/security/http_fetch.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 import urllib.parse
 import urllib.request
+from collections.abc import Iterable
 from dataclasses import dataclass
 from typing import Mapping
 
@@ -13,6 +14,9 @@ from govoplan_core.security.outbound_http import (
 
 
 MAX_OUTBOUND_HTTP_REQUEST_BODY_BYTES = 1_000_000
+_STANDARD_REDIRECT_SENSITIVE_HEADERS = frozenset(
+    {"authorization", "proxy-authorization", "cookie", "cookie2"}
+)
 
 
 @dataclass(frozen=True, slots=True)
@@ -51,6 +55,7 @@ def fetch_http(
     headers: Mapping[str, str] | None = None,
     body: bytes | None = None,
     max_bytes: int | None = None,
+    redirect_sensitive_headers: Iterable[str] = (),
 ) -> HttpFetchResponse:
     if body is not None and len(body) > MAX_OUTBOUND_HTTP_REQUEST_BODY_BYTES:
         raise ValueError(
@@ -63,7 +68,12 @@ def fetch_http(
         headers=dict(headers or {}),
         method=method,
     )
-    opener = build_outbound_http_opener(_PolicyRedirectHandler(label=label))
+    opener = build_outbound_http_opener(
+        _PolicyRedirectHandler(
+            label=label,
+            sensitive_headers=redirect_sensitive_headers,
+        )
+    )
     with opener.open(request, timeout=timeout) as response:  # noqa: S310 - URL and every redirect are policy-validated.  # nosec B310  # nosemgrep: python.lang.security.audit.dynamic-urllib-use-detected.dynamic-urllib-use-detected
         response_headers = dict(response.headers.items())
         return HttpFetchResponse(
@@ -88,6 +98,7 @@ def fetch_http_text(
     body: bytes | None = None,
     encoding: str = "utf-8",
     max_bytes: int | None = None,
+    redirect_sensitive_headers: Iterable[str] = (),
 ) -> str:
     return fetch_http(
         url,
@@ -97,13 +108,22 @@ def fetch_http_text(
         headers=headers,
         body=body,
         max_bytes=max_bytes,
+        redirect_sensitive_headers=redirect_sensitive_headers,
     ).text(encoding)
 
 
 class _PolicyRedirectHandler(urllib.request.HTTPRedirectHandler):
-    def __init__(self, *, label: str) -> None:
+    def __init__(
+        self,
+        *,
+        label: str,
+        sensitive_headers: Iterable[str] = (),
+    ) -> None:
         super().__init__()
         self._label = label
+        self._sensitive_headers = _STANDARD_REDIRECT_SENSITIVE_HEADERS | {
+            value.strip().lower() for value in sensitive_headers if value.strip()
+        }
 
     def redirect_request(self, req, fp, code, msg, headers, newurl):  # type: ignore[no-untyped-def]
         candidate = validate_outbound_http_url(newurl, label=f"{self._label} redirect")
@@ -113,8 +133,9 @@ class _PolicyRedirectHandler(urllib.request.HTTPRedirectHandler):
             return None
         new_request = super().redirect_request(req, fp, code, msg, headers, candidate)
         if new_request is not None and _http_origin(previous) != _http_origin(redirected):
-            for header in ("Authorization", "Proxy-Authorization", "Cookie", "Cookie2"):
-                new_request.remove_header(header)
+            for header in tuple(new_request.headers) + tuple(new_request.unredirected_hdrs):
+                if header.lower() in self._sensitive_headers:
+                    new_request.remove_header(header)
         return new_request