From 9700ed9d6b8715e9a87f7842e4db84af0cf682fe Mon Sep 17 00:00:00 2001 From: Adrian Date: Fri, 27 Feb 2026 15:49:07 -0600 Subject: [PATCH 1/5] Adding header removal handler --- .../options/redirect_handler_option.py | 49 +++- .../kiota_http/middleware/redirect_handler.py | 33 +-- .../middleware_tests/test_redirect_handler.py | 215 +++++++++++++++++- 3 files changed, 259 insertions(+), 38 deletions(-) diff --git a/packages/http/httpx/kiota_http/middleware/options/redirect_handler_option.py b/packages/http/httpx/kiota_http/middleware/options/redirect_handler_option.py index 9f0c43a3..9bc10064 100644 --- a/packages/http/httpx/kiota_http/middleware/options/redirect_handler_option.py +++ b/packages/http/httpx/kiota_http/middleware/options/redirect_handler_option.py @@ -1,5 +1,40 @@ +from typing import Callable, Optional from kiota_abstractions.request_option import RequestOption +import httpx + +# Type alias for the scrub sensitive headers callback +ScrubSensitiveHeadersCallback = Callable[[httpx.Headers, httpx.URL, httpx.URL], None] + + +def default_scrub_sensitive_headers( + headers: httpx.Headers, original_url: httpx.URL, new_url: httpx.URL +) -> None: + """ + The default implementation for scrubbing sensitive headers during redirects. + This method removes Authorization and Cookie headers when the host or scheme changes. + Args: + headers: The headers object to modify + original_url: The original request URL + new_url: The new redirect URL + """ + if not headers or not original_url or not new_url: + return + + # Remove Authorization and Cookie headers if the request's scheme or host changes + is_different_host_or_scheme = ( + original_url.host != new_url.host or original_url.scheme != new_url.scheme + ) + + if is_different_host_or_scheme: + headers.pop("Authorization", None) + headers.pop("Cookie", None) + + # Note: Proxy-Authorization is not handled here as proxy configuration in httpx + # is managed at the transport level and not accessible to middleware. + # In environments where this matters, the proxy configuration should be managed + # at the HTTP client level. + class RedirectHandlerOption(RequestOption): @@ -15,7 +50,8 @@ def __init__( self, max_redirect: int = DEFAULT_MAX_REDIRECT, should_redirect: bool = True, - allow_redirect_on_scheme_change: bool = False + allow_redirect_on_scheme_change: bool = False, + scrub_sensitive_headers: Optional[ScrubSensitiveHeadersCallback] = None ) -> None: if max_redirect > self.MAX_MAX_REDIRECT: @@ -28,6 +64,7 @@ def __init__( self._max_redirect = max_redirect self._should_redirect = should_redirect self._allow_redirect_on_scheme_change = allow_redirect_on_scheme_change + self._scrub_sensitive_headers = scrub_sensitive_headers or default_scrub_sensitive_headers @property def max_redirect(self): @@ -59,6 +96,16 @@ def allow_redirect_on_scheme_change(self): def allow_redirect_on_scheme_change(self, value: bool): self._allow_redirect_on_scheme_change = value + @property + def scrub_sensitive_headers(self) -> ScrubSensitiveHeadersCallback: + """The callback for scrubbing sensitive headers during redirects. + Defaults to default_scrub_sensitive_headers.""" + return self._scrub_sensitive_headers + + @scrub_sensitive_headers.setter + def scrub_sensitive_headers(self, value: ScrubSensitiveHeadersCallback): + self._scrub_sensitive_headers = value + @staticmethod def get_key() -> str: return RedirectHandlerOption.REDIRECT_HANDLER_OPTION_KEY diff --git a/packages/http/httpx/kiota_http/middleware/redirect_handler.py b/packages/http/httpx/kiota_http/middleware/redirect_handler.py index e7a1197f..bcc30954 100644 --- a/packages/http/httpx/kiota_http/middleware/redirect_handler.py +++ b/packages/http/httpx/kiota_http/middleware/redirect_handler.py @@ -26,7 +26,6 @@ class RedirectHandler(BaseMiddleware): } STATUS_CODE_SEE_OTHER: int = 303 LOCATION_HEADER: str = "Location" - AUTHORIZATION_HEADER: str = "Authorization" def __init__(self, options: RedirectHandlerOption = RedirectHandlerOption()) -> None: super().__init__() @@ -125,7 +124,7 @@ def _build_redirect_request( """ method = self._redirect_method(request, response) url = self._redirect_url(request, response, options) - headers = self._redirect_headers(request, url, method) + headers = self._redirect_headers(request, url, method, options) stream = self._redirect_stream(request, method) new_request = httpx.Request( method=method, @@ -202,20 +201,18 @@ def _redirect_url( return url def _redirect_headers( - self, request: httpx.Request, url: httpx.URL, method: str + self, request: httpx.Request, url: httpx.URL, method: str, options: RedirectHandlerOption ) -> httpx.Headers: """ Return the headers that should be used for the redirect request. """ headers = httpx.Headers(request.headers) - if not self._same_origin(url, request.url): - if not self.is_https_redirect(request.url, url): - # Strip Authorization headers when responses are redirected - # away from the origin. (Except for direct HTTP to HTTPS redirects.) - headers.pop("Authorization", None) + # Scrub sensitive headers before following the redirect + options.scrub_sensitive_headers(headers, request.url, url) - # Update the Host header. + # Update the Host header if not same origin + if not self._same_origin(url, request.url): headers["Host"] = url.netloc.decode("ascii") if method != request.method and method == "GET": @@ -224,10 +221,6 @@ def _redirect_headers( headers.pop("Content-Length", None) headers.pop("Transfer-Encoding", None) - # We should use the client cookie store to determine any cookie header, - # rather than whatever was on the original outgoing request. - headers.pop("Cookie", None) - return headers def _redirect_stream( @@ -246,17 +239,3 @@ def _same_origin(self, url: httpx.URL, other: httpx.URL) -> bool: Return 'True' if the given URLs share the same origin. """ return (url.scheme == other.scheme and url.host == other.host) - - def port_or_default(self, url: httpx.URL) -> typing.Optional[int]: - if url.port is not None: - return url.port - return {"http": 80, "https": 443}.get(url.scheme) - - def is_https_redirect(self, url: httpx.URL, location: httpx.URL) -> bool: - """ - Return 'True' if 'location' is a HTTPS upgrade of 'url' - """ - if url.host != location.host: - return False - - return (url.scheme == "http" and location.scheme == "https") diff --git a/packages/http/httpx/tests/middleware_tests/test_redirect_handler.py b/packages/http/httpx/tests/middleware_tests/test_redirect_handler.py index 1f7308b1..77fd4ffe 100644 --- a/packages/http/httpx/tests/middleware_tests/test_redirect_handler.py +++ b/packages/http/httpx/tests/middleware_tests/test_redirect_handler.py @@ -69,16 +69,6 @@ def test_not_same_origin(mock_redirect_handler): assert not mock_redirect_handler._same_origin(origin1, origin2) -def test_is_https_redirect(mock_redirect_handler): - url = httpx.URL("http://example.com") - location = httpx.URL(BASE_URL) - assert mock_redirect_handler.is_https_redirect(url, location) - - -def test_is_not_https_redirect(mock_redirect_handler): - url = httpx.URL(BASE_URL) - location = httpx.URL("http://www.example.com") - assert not mock_redirect_handler.is_https_redirect(url, location) @pytest.mark.asyncio @@ -281,3 +271,208 @@ def request_handler(request: httpx.Request): with pytest.raises(Exception) as e: await handler.send(request, mock_transport) assert "Too many redirects" in str(e.value) + + +@pytest.mark.asyncio +async def test_redirect_cross_host_removes_auth_and_cookie(): + """Test that cross-host redirects remove both Authorization and Cookie headers""" + + def request_handler(request: httpx.Request): + if request.url == "https://other.example.com/api": + return httpx.Response(200, ) + return httpx.Response( + MOVED_PERMANENTLY, + headers={LOCATION_HEADER: "https://other.example.com/api"}, + ) + + handler = RedirectHandler() + request = httpx.Request( + 'GET', + BASE_URL, + headers={ + AUTHORIZATION_HEADER: "Bearer token", + "Cookie": "session=SECRET" + }, + ) + mock_transport = httpx.MockTransport(request_handler) + resp = await handler.send(request, mock_transport) + assert resp.status_code == 200 + assert AUTHORIZATION_HEADER not in resp.request.headers + assert "Cookie" not in resp.request.headers + + +@pytest.mark.asyncio +async def test_redirect_scheme_change_removes_auth_and_cookie(): + """Test that scheme changes remove both Authorization and Cookie headers""" + + def request_handler(request: httpx.Request): + if request.url == "http://example.com/api": + return httpx.Response(200, ) + return httpx.Response( + MOVED_PERMANENTLY, + headers={LOCATION_HEADER: "http://example.com/api"}, + ) + + handler = RedirectHandler(RedirectHandlerOption(allow_redirect_on_scheme_change=True)) + request = httpx.Request( + 'GET', + BASE_URL, + headers={ + AUTHORIZATION_HEADER: "Bearer token", + "Cookie": "session=SECRET" + }, + ) + mock_transport = httpx.MockTransport(request_handler) + resp = await handler.send(request, mock_transport) + assert resp.status_code == 200 + assert AUTHORIZATION_HEADER not in resp.request.headers + assert "Cookie" not in resp.request.headers + + +@pytest.mark.asyncio +async def test_redirect_same_host_and_scheme_keeps_all_headers(): + """Test that same-host and same-scheme redirects keep Authorization and Cookie headers""" + + def request_handler(request: httpx.Request): + if request.url == f"{BASE_URL}/v2/api": + return httpx.Response(200, ) + return httpx.Response( + MOVED_PERMANENTLY, + headers={LOCATION_HEADER: f"{BASE_URL}/v2/api"}, + ) + + handler = RedirectHandler() + request = httpx.Request( + 'GET', + f"{BASE_URL}/v1/api", + headers={ + AUTHORIZATION_HEADER: "Bearer token", + "Cookie": "session=SECRET" + }, + ) + mock_transport = httpx.MockTransport(request_handler) + resp = await handler.send(request, mock_transport) + assert resp.status_code == 200 + assert AUTHORIZATION_HEADER in resp.request.headers + assert resp.request.headers[AUTHORIZATION_HEADER] == "Bearer token" + assert "Cookie" in resp.request.headers + assert resp.request.headers["Cookie"] == "session=SECRET" + + +@pytest.mark.asyncio +async def test_redirect_with_custom_scrubber(): + """Test that custom scrubber can be provided and is used""" + + def custom_scrubber(headers, original_url, new_url): + # Custom logic: never remove headers + pass + + def request_handler(request: httpx.Request): + if request.url == "https://evil.attacker.com/steal": + return httpx.Response(200, ) + return httpx.Response( + MOVED_PERMANENTLY, + headers={LOCATION_HEADER: "https://evil.attacker.com/steal"}, + ) + + options = RedirectHandlerOption(scrub_sensitive_headers=custom_scrubber) + handler = RedirectHandler(options) + request = httpx.Request( + 'GET', + BASE_URL, + headers={ + AUTHORIZATION_HEADER: "Bearer token", + "Cookie": "session=SECRET" + }, + ) + mock_transport = httpx.MockTransport(request_handler) + resp = await handler.send(request, mock_transport) + assert resp.status_code == 200 + # Headers should be kept because custom scrubber doesn't remove them + assert AUTHORIZATION_HEADER in resp.request.headers + assert "Cookie" in resp.request.headers + + +def test_default_scrub_sensitive_headers_removes_on_host_change(): + """Test that default scrubber removes Authorization and Cookie when host changes""" + from kiota_http.middleware.options.redirect_handler_option import default_scrub_sensitive_headers + + headers = httpx.Headers({ + AUTHORIZATION_HEADER: "Bearer token", + "Cookie": "session=SECRET", + "Content-Type": "application/json" + }) + original_url = httpx.URL("https://example.com/v1/api") + new_url = httpx.URL("https://other.com/api") + + default_scrub_sensitive_headers(headers, original_url, new_url) + + assert AUTHORIZATION_HEADER not in headers + assert "Cookie" not in headers + assert "Content-Type" in headers # Other headers should remain + + +def test_default_scrub_sensitive_headers_removes_on_scheme_change(): + """Test that default scrubber removes Authorization and Cookie when scheme changes""" + from kiota_http.middleware.options.redirect_handler_option import default_scrub_sensitive_headers + + headers = httpx.Headers({ + AUTHORIZATION_HEADER: "Bearer token", + "Cookie": "session=SECRET", + "Content-Type": "application/json" + }) + original_url = httpx.URL("https://example.com/v1/api") + new_url = httpx.URL("http://example.com/v1/api") + + default_scrub_sensitive_headers(headers, original_url, new_url) + + assert AUTHORIZATION_HEADER not in headers + assert "Cookie" not in headers + assert "Content-Type" in headers + + +def test_default_scrub_sensitive_headers_keeps_on_same_origin(): + """Test that default scrubber keeps headers when host and scheme are the same""" + from kiota_http.middleware.options.redirect_handler_option import default_scrub_sensitive_headers + + headers = httpx.Headers({ + AUTHORIZATION_HEADER: "Bearer token", + "Cookie": "session=SECRET", + "Content-Type": "application/json" + }) + original_url = httpx.URL("https://example.com/v1/api") + new_url = httpx.URL("https://example.com/v2/api") + + default_scrub_sensitive_headers(headers, original_url, new_url) + + assert AUTHORIZATION_HEADER in headers + assert "Cookie" in headers + assert "Content-Type" in headers + + +def test_default_scrub_sensitive_headers_handles_none_gracefully(): + """Test that default scrubber handles None/empty inputs gracefully""" + from kiota_http.middleware.options.redirect_handler_option import default_scrub_sensitive_headers + + # Should not raise exceptions + default_scrub_sensitive_headers(None, httpx.URL(BASE_URL), httpx.URL(BASE_URL)) + default_scrub_sensitive_headers(httpx.Headers(), None, httpx.URL(BASE_URL)) + default_scrub_sensitive_headers(httpx.Headers(), httpx.URL(BASE_URL), None) + + +def test_custom_scrub_sensitive_headers(): + """Test that custom scrubber can be set on options""" + def custom_scrubber(headers, original_url, new_url): + # Custom logic + pass + + options = RedirectHandlerOption(scrub_sensitive_headers=custom_scrubber) + assert options.scrub_sensitive_headers == custom_scrubber + + +def test_default_options_uses_default_scrubber(): + """Test that default options use the default scrubber""" + from kiota_http.middleware.options.redirect_handler_option import default_scrub_sensitive_headers + + options = RedirectHandlerOption() + assert options.scrub_sensitive_headers == default_scrub_sensitive_headers From 648d31b4fb1848330da58641eb8de9c14c8f7206 Mon Sep 17 00:00:00 2001 From: Adrian Date: Mon, 2 Mar 2026 10:41:16 -0600 Subject: [PATCH 2/5] adding port condition --- .../options/redirect_handler_option.py | 12 ++- .../middleware_tests/test_redirect_handler.py | 98 +++++++++++++++++++ 2 files changed, 105 insertions(+), 5 deletions(-) diff --git a/packages/http/httpx/kiota_http/middleware/options/redirect_handler_option.py b/packages/http/httpx/kiota_http/middleware/options/redirect_handler_option.py index 9bc10064..132685b7 100644 --- a/packages/http/httpx/kiota_http/middleware/options/redirect_handler_option.py +++ b/packages/http/httpx/kiota_http/middleware/options/redirect_handler_option.py @@ -12,7 +12,7 @@ def default_scrub_sensitive_headers( ) -> None: """ The default implementation for scrubbing sensitive headers during redirects. - This method removes Authorization and Cookie headers when the host or scheme changes. + This method removes Authorization and Cookie headers when the host, scheme, or port changes. Args: headers: The headers object to modify original_url: The original request URL @@ -21,12 +21,14 @@ def default_scrub_sensitive_headers( if not headers or not original_url or not new_url: return - # Remove Authorization and Cookie headers if the request's scheme or host changes - is_different_host_or_scheme = ( - original_url.host != new_url.host or original_url.scheme != new_url.scheme + # Remove Authorization and Cookie headers if the request's scheme, host, or port changes + is_different_origin = ( + original_url.host != new_url.host + or original_url.scheme != new_url.scheme + or original_url.port != new_url.port ) - if is_different_host_or_scheme: + if is_different_origin: headers.pop("Authorization", None) headers.pop("Cookie", None) diff --git a/packages/http/httpx/tests/middleware_tests/test_redirect_handler.py b/packages/http/httpx/tests/middleware_tests/test_redirect_handler.py index 77fd4ffe..643365df 100644 --- a/packages/http/httpx/tests/middleware_tests/test_redirect_handler.py +++ b/packages/http/httpx/tests/middleware_tests/test_redirect_handler.py @@ -359,6 +359,66 @@ def request_handler(request: httpx.Request): assert resp.request.headers["Cookie"] == "session=SECRET" +@pytest.mark.asyncio +async def test_redirect_with_different_port_removes_auth_and_cookie(): + """Test that redirects to a different port remove Authorization and Cookie headers""" + + def request_handler(request: httpx.Request): + if request.url == "http://example.org:9090/bar": + return httpx.Response(200, ) + return httpx.Response( + MOVED_PERMANENTLY, + headers={LOCATION_HEADER: "http://example.org:9090/bar"}, + ) + + handler = RedirectHandler() + request = httpx.Request( + 'GET', + "http://example.org:8080/foo", + headers={ + AUTHORIZATION_HEADER: "Bearer token", + "Cookie": "session=SECRET" + }, + ) + mock_transport = httpx.MockTransport(request_handler) + resp = await handler.send(request, mock_transport) + assert resp.status_code == 200 + assert resp.request != request + assert AUTHORIZATION_HEADER not in resp.request.headers + assert "Cookie" not in resp.request.headers + + +@pytest.mark.asyncio +async def test_redirect_with_same_port_keeps_auth_and_cookie(): + """Test that redirects to the same port keep Authorization and Cookie headers""" + + def request_handler(request: httpx.Request): + if request.url == "http://example.org:8080/bar": + return httpx.Response(200, ) + return httpx.Response( + FOUND, + headers={LOCATION_HEADER: "http://example.org:8080/bar"}, + ) + + handler = RedirectHandler() + request = httpx.Request( + 'GET', + "http://example.org:8080/foo", + headers={ + AUTHORIZATION_HEADER: "Bearer token", + "Cookie": "session=SECRET" + }, + ) + mock_transport = httpx.MockTransport(request_handler) + resp = await handler.send(request, mock_transport) + assert resp.status_code == 200 + assert resp.request != request + assert AUTHORIZATION_HEADER in resp.request.headers + assert resp.request.headers[AUTHORIZATION_HEADER] == "Bearer token" + assert "Cookie" in resp.request.headers + assert resp.request.headers["Cookie"] == "session=SECRET" + + @pytest.mark.asyncio async def test_redirect_with_custom_scrubber(): """Test that custom scrubber can be provided and is used""" @@ -450,6 +510,44 @@ def test_default_scrub_sensitive_headers_keeps_on_same_origin(): assert "Content-Type" in headers +def test_default_scrub_sensitive_headers_removes_on_port_change(): + """Test that default scrubber removes Authorization and Cookie when port changes""" + from kiota_http.middleware.options.redirect_handler_option import default_scrub_sensitive_headers + + headers = httpx.Headers({ + AUTHORIZATION_HEADER: "Bearer token", + "Cookie": "session=SECRET", + "Content-Type": "application/json" + }) + original_url = httpx.URL("http://example.org:8080/foo") + new_url = httpx.URL("http://example.org:9090/bar") + + default_scrub_sensitive_headers(headers, original_url, new_url) + + assert AUTHORIZATION_HEADER not in headers + assert "Cookie" not in headers + assert "Content-Type" in headers # Other headers should remain + + +def test_default_scrub_sensitive_headers_keeps_on_same_port(): + """Test that default scrubber keeps headers when port is the same""" + from kiota_http.middleware.options.redirect_handler_option import default_scrub_sensitive_headers + + headers = httpx.Headers({ + AUTHORIZATION_HEADER: "Bearer token", + "Cookie": "session=SECRET", + "Content-Type": "application/json" + }) + original_url = httpx.URL("http://example.org:8080/foo") + new_url = httpx.URL("http://example.org:8080/bar") + + default_scrub_sensitive_headers(headers, original_url, new_url) + + assert AUTHORIZATION_HEADER in headers + assert "Cookie" in headers + assert "Content-Type" in headers + + def test_default_scrub_sensitive_headers_handles_none_gracefully(): """Test that default scrubber handles None/empty inputs gracefully""" from kiota_http.middleware.options.redirect_handler_option import default_scrub_sensitive_headers From b64ec6e584baee40fa2f99a1dc4984fc9b98a56b Mon Sep 17 00:00:00 2001 From: Adrian Date: Mon, 2 Mar 2026 11:39:17 -0600 Subject: [PATCH 3/5] updating func to only use 2 instead of 3 params --- .../options/redirect_handler_option.py | 17 ++- .../kiota_http/middleware/redirect_handler.py | 42 +++--- .../middleware_tests/test_redirect_handler.py | 124 ++++++++++-------- 3 files changed, 97 insertions(+), 86 deletions(-) diff --git a/packages/http/httpx/kiota_http/middleware/options/redirect_handler_option.py b/packages/http/httpx/kiota_http/middleware/options/redirect_handler_option.py index 132685b7..1c9e91b5 100644 --- a/packages/http/httpx/kiota_http/middleware/options/redirect_handler_option.py +++ b/packages/http/httpx/kiota_http/middleware/options/redirect_handler_option.py @@ -4,21 +4,24 @@ import httpx # Type alias for the scrub sensitive headers callback -ScrubSensitiveHeadersCallback = Callable[[httpx.Headers, httpx.URL, httpx.URL], None] +ScrubSensitiveHeadersCallback = Callable[[httpx.Request, httpx.URL], None] def default_scrub_sensitive_headers( - headers: httpx.Headers, original_url: httpx.URL, new_url: httpx.URL + new_request: httpx.Request, original_url: httpx.URL ) -> None: """ The default implementation for scrubbing sensitive headers during redirects. This method removes Authorization and Cookie headers when the host, scheme, or port changes. Args: - headers: The headers object to modify + new_request: The new redirect request to modify original_url: The original request URL - new_url: The new redirect URL """ - if not headers or not original_url or not new_url: + if not new_request or not original_url: + return + + new_url = new_request.url + if not new_url: return # Remove Authorization and Cookie headers if the request's scheme, host, or port changes @@ -29,8 +32,8 @@ def default_scrub_sensitive_headers( ) if is_different_origin: - headers.pop("Authorization", None) - headers.pop("Cookie", None) + new_request.headers.pop("Authorization", None) + new_request.headers.pop("Cookie", None) # Note: Proxy-Authorization is not handled here as proxy configuration in httpx # is managed at the transport level and not accessible to middleware. diff --git a/packages/http/httpx/kiota_http/middleware/redirect_handler.py b/packages/http/httpx/kiota_http/middleware/redirect_handler.py index bcc30954..4e575fc4 100644 --- a/packages/http/httpx/kiota_http/middleware/redirect_handler.py +++ b/packages/http/httpx/kiota_http/middleware/redirect_handler.py @@ -124,15 +124,31 @@ def _build_redirect_request( """ method = self._redirect_method(request, response) url = self._redirect_url(request, response, options) - headers = self._redirect_headers(request, url, method, options) stream = self._redirect_stream(request, method) + + # Create the new request with the redirect URL and original headers new_request = httpx.Request( method=method, url=url, - headers=headers, + headers=request.headers.copy(), stream=stream, extensions=request.extensions, ) + + # Scrub sensitive headers before following the redirect + options.scrub_sensitive_headers(new_request, request.url) + + # Update the Host header if not same origin + if not self._same_origin(url, request.url): + new_request.headers["Host"] = url.netloc.decode("ascii") + + # Handle 303 See Other and other method changes + if method != request.method and method == "GET": + # If we've switched to a 'GET' request, strip any headers which + # are only relevant to the request body. + new_request.headers.pop("Content-Length", None) + new_request.headers.pop("Transfer-Encoding", None) + if hasattr(request, "context"): new_request.context = request.context #type: ignore new_request.options = {} #type: ignore @@ -200,28 +216,6 @@ def _redirect_url( return url - def _redirect_headers( - self, request: httpx.Request, url: httpx.URL, method: str, options: RedirectHandlerOption - ) -> httpx.Headers: - """ - Return the headers that should be used for the redirect request. - """ - headers = httpx.Headers(request.headers) - - # Scrub sensitive headers before following the redirect - options.scrub_sensitive_headers(headers, request.url, url) - - # Update the Host header if not same origin - if not self._same_origin(url, request.url): - headers["Host"] = url.netloc.decode("ascii") - - if method != request.method and method == "GET": - # If we've switch to a 'GET' request, then strip any headers which - # are only relevant to the request body. - headers.pop("Content-Length", None) - headers.pop("Transfer-Encoding", None) - - return headers def _redirect_stream( self, request: httpx.Request, method: str diff --git a/packages/http/httpx/tests/middleware_tests/test_redirect_handler.py b/packages/http/httpx/tests/middleware_tests/test_redirect_handler.py index 643365df..e34c0101 100644 --- a/packages/http/httpx/tests/middleware_tests/test_redirect_handler.py +++ b/packages/http/httpx/tests/middleware_tests/test_redirect_handler.py @@ -423,7 +423,7 @@ def request_handler(request: httpx.Request): async def test_redirect_with_custom_scrubber(): """Test that custom scrubber can be provided and is used""" - def custom_scrubber(headers, original_url, new_url): + def custom_scrubber(new_request, original_url): # Custom logic: never remove headers pass @@ -457,95 +457,110 @@ def test_default_scrub_sensitive_headers_removes_on_host_change(): """Test that default scrubber removes Authorization and Cookie when host changes""" from kiota_http.middleware.options.redirect_handler_option import default_scrub_sensitive_headers - headers = httpx.Headers({ - AUTHORIZATION_HEADER: "Bearer token", - "Cookie": "session=SECRET", - "Content-Type": "application/json" - }) original_url = httpx.URL("https://example.com/v1/api") - new_url = httpx.URL("https://other.com/api") + new_request = httpx.Request( + "GET", + "https://other.com/api", + headers={ + AUTHORIZATION_HEADER: "Bearer token", + "Cookie": "session=SECRET", + "Content-Type": "application/json" + } + ) - default_scrub_sensitive_headers(headers, original_url, new_url) + default_scrub_sensitive_headers(new_request, original_url) - assert AUTHORIZATION_HEADER not in headers - assert "Cookie" not in headers - assert "Content-Type" in headers # Other headers should remain + assert AUTHORIZATION_HEADER not in new_request.headers + assert "Cookie" not in new_request.headers + assert "Content-Type" in new_request.headers # Other headers should remain def test_default_scrub_sensitive_headers_removes_on_scheme_change(): """Test that default scrubber removes Authorization and Cookie when scheme changes""" from kiota_http.middleware.options.redirect_handler_option import default_scrub_sensitive_headers - headers = httpx.Headers({ - AUTHORIZATION_HEADER: "Bearer token", - "Cookie": "session=SECRET", - "Content-Type": "application/json" - }) original_url = httpx.URL("https://example.com/v1/api") - new_url = httpx.URL("http://example.com/v1/api") + new_request = httpx.Request( + "GET", + "http://example.com/v1/api", + headers={ + AUTHORIZATION_HEADER: "Bearer token", + "Cookie": "session=SECRET", + "Content-Type": "application/json" + } + ) - default_scrub_sensitive_headers(headers, original_url, new_url) + default_scrub_sensitive_headers(new_request, original_url) - assert AUTHORIZATION_HEADER not in headers - assert "Cookie" not in headers - assert "Content-Type" in headers + assert AUTHORIZATION_HEADER not in new_request.headers + assert "Cookie" not in new_request.headers + assert "Content-Type" in new_request.headers def test_default_scrub_sensitive_headers_keeps_on_same_origin(): """Test that default scrubber keeps headers when host and scheme are the same""" from kiota_http.middleware.options.redirect_handler_option import default_scrub_sensitive_headers - headers = httpx.Headers({ - AUTHORIZATION_HEADER: "Bearer token", - "Cookie": "session=SECRET", - "Content-Type": "application/json" - }) original_url = httpx.URL("https://example.com/v1/api") - new_url = httpx.URL("https://example.com/v2/api") + new_request = httpx.Request( + "GET", + "https://example.com/v2/api", + headers={ + AUTHORIZATION_HEADER: "Bearer token", + "Cookie": "session=SECRET", + "Content-Type": "application/json" + } + ) - default_scrub_sensitive_headers(headers, original_url, new_url) + default_scrub_sensitive_headers(new_request, original_url) - assert AUTHORIZATION_HEADER in headers - assert "Cookie" in headers - assert "Content-Type" in headers + assert AUTHORIZATION_HEADER in new_request.headers + assert "Cookie" in new_request.headers + assert "Content-Type" in new_request.headers def test_default_scrub_sensitive_headers_removes_on_port_change(): """Test that default scrubber removes Authorization and Cookie when port changes""" from kiota_http.middleware.options.redirect_handler_option import default_scrub_sensitive_headers - headers = httpx.Headers({ - AUTHORIZATION_HEADER: "Bearer token", - "Cookie": "session=SECRET", - "Content-Type": "application/json" - }) original_url = httpx.URL("http://example.org:8080/foo") - new_url = httpx.URL("http://example.org:9090/bar") + new_request = httpx.Request( + "GET", + "http://example.org:9090/bar", + headers={ + AUTHORIZATION_HEADER: "Bearer token", + "Cookie": "session=SECRET", + "Content-Type": "application/json" + } + ) - default_scrub_sensitive_headers(headers, original_url, new_url) + default_scrub_sensitive_headers(new_request, original_url) - assert AUTHORIZATION_HEADER not in headers - assert "Cookie" not in headers - assert "Content-Type" in headers # Other headers should remain + assert AUTHORIZATION_HEADER not in new_request.headers + assert "Cookie" not in new_request.headers + assert "Content-Type" in new_request.headers # Other headers should remain def test_default_scrub_sensitive_headers_keeps_on_same_port(): """Test that default scrubber keeps headers when port is the same""" from kiota_http.middleware.options.redirect_handler_option import default_scrub_sensitive_headers - headers = httpx.Headers({ - AUTHORIZATION_HEADER: "Bearer token", - "Cookie": "session=SECRET", - "Content-Type": "application/json" - }) original_url = httpx.URL("http://example.org:8080/foo") - new_url = httpx.URL("http://example.org:8080/bar") + new_request = httpx.Request( + "GET", + "http://example.org:8080/bar", + headers={ + AUTHORIZATION_HEADER: "Bearer token", + "Cookie": "session=SECRET", + "Content-Type": "application/json" + } + ) - default_scrub_sensitive_headers(headers, original_url, new_url) + default_scrub_sensitive_headers(new_request, original_url) - assert AUTHORIZATION_HEADER in headers - assert "Cookie" in headers - assert "Content-Type" in headers + assert AUTHORIZATION_HEADER in new_request.headers + assert "Cookie" in new_request.headers + assert "Content-Type" in new_request.headers def test_default_scrub_sensitive_headers_handles_none_gracefully(): @@ -553,14 +568,13 @@ def test_default_scrub_sensitive_headers_handles_none_gracefully(): from kiota_http.middleware.options.redirect_handler_option import default_scrub_sensitive_headers # Should not raise exceptions - default_scrub_sensitive_headers(None, httpx.URL(BASE_URL), httpx.URL(BASE_URL)) - default_scrub_sensitive_headers(httpx.Headers(), None, httpx.URL(BASE_URL)) - default_scrub_sensitive_headers(httpx.Headers(), httpx.URL(BASE_URL), None) + default_scrub_sensitive_headers(None, httpx.URL(BASE_URL)) + default_scrub_sensitive_headers(httpx.Request("GET", BASE_URL), None) def test_custom_scrub_sensitive_headers(): """Test that custom scrubber can be set on options""" - def custom_scrubber(headers, original_url, new_url): + def custom_scrubber(new_request, original_url): # Custom logic pass From 44e8d940bb6d6284b2eee67a01b5bdd31f7e9e5c Mon Sep 17 00:00:00 2001 From: Adrian Date: Mon, 2 Mar 2026 12:21:00 -0600 Subject: [PATCH 4/5] fix linter issues --- .../middleware/options/redirect_handler_option.py | 7 ++----- .../http/httpx/kiota_http/middleware/redirect_handler.py | 1 - 2 files changed, 2 insertions(+), 6 deletions(-) diff --git a/packages/http/httpx/kiota_http/middleware/options/redirect_handler_option.py b/packages/http/httpx/kiota_http/middleware/options/redirect_handler_option.py index 1c9e91b5..3d619b19 100644 --- a/packages/http/httpx/kiota_http/middleware/options/redirect_handler_option.py +++ b/packages/http/httpx/kiota_http/middleware/options/redirect_handler_option.py @@ -7,9 +7,7 @@ ScrubSensitiveHeadersCallback = Callable[[httpx.Request, httpx.URL], None] -def default_scrub_sensitive_headers( - new_request: httpx.Request, original_url: httpx.URL -) -> None: +def default_scrub_sensitive_headers(new_request: httpx.Request, original_url: httpx.URL) -> None: """ The default implementation for scrubbing sensitive headers during redirects. This method removes Authorization and Cookie headers when the host, scheme, or port changes. @@ -26,8 +24,7 @@ def default_scrub_sensitive_headers( # Remove Authorization and Cookie headers if the request's scheme, host, or port changes is_different_origin = ( - original_url.host != new_url.host - or original_url.scheme != new_url.scheme + original_url.host != new_url.host or original_url.scheme != new_url.scheme or original_url.port != new_url.port ) diff --git a/packages/http/httpx/kiota_http/middleware/redirect_handler.py b/packages/http/httpx/kiota_http/middleware/redirect_handler.py index 4e575fc4..f0f8a930 100644 --- a/packages/http/httpx/kiota_http/middleware/redirect_handler.py +++ b/packages/http/httpx/kiota_http/middleware/redirect_handler.py @@ -216,7 +216,6 @@ def _redirect_url( return url - def _redirect_stream( self, request: httpx.Request, method: str ) -> typing.Optional[typing.Union[httpx.SyncByteStream, httpx.AsyncByteStream]]: From 67bb784b0794072b7a97d40bc1454c5a8ee3e414 Mon Sep 17 00:00:00 2001 From: Adrian Date: Mon, 2 Mar 2026 14:33:51 -0600 Subject: [PATCH 5/5] no sonar warnings added --- .../middleware_tests/test_redirect_handler.py | 26 +++++++++---------- 1 file changed, 13 insertions(+), 13 deletions(-) diff --git a/packages/http/httpx/tests/middleware_tests/test_redirect_handler.py b/packages/http/httpx/tests/middleware_tests/test_redirect_handler.py index e34c0101..4963af5e 100644 --- a/packages/http/httpx/tests/middleware_tests/test_redirect_handler.py +++ b/packages/http/httpx/tests/middleware_tests/test_redirect_handler.py @@ -306,11 +306,11 @@ async def test_redirect_scheme_change_removes_auth_and_cookie(): """Test that scheme changes remove both Authorization and Cookie headers""" def request_handler(request: httpx.Request): - if request.url == "http://example.com/api": + if request.url == "http://example.com/api": # NOSONAR return httpx.Response(200, ) return httpx.Response( MOVED_PERMANENTLY, - headers={LOCATION_HEADER: "http://example.com/api"}, + headers={LOCATION_HEADER: "http://example.com/api"}, # NOSONAR ) handler = RedirectHandler(RedirectHandlerOption(allow_redirect_on_scheme_change=True)) @@ -364,17 +364,17 @@ async def test_redirect_with_different_port_removes_auth_and_cookie(): """Test that redirects to a different port remove Authorization and Cookie headers""" def request_handler(request: httpx.Request): - if request.url == "http://example.org:9090/bar": + if request.url == "http://example.org:9090/bar": # NOSONAR return httpx.Response(200, ) return httpx.Response( MOVED_PERMANENTLY, - headers={LOCATION_HEADER: "http://example.org:9090/bar"}, + headers={LOCATION_HEADER: "http://example.org:9090/bar"}, # NOSONAR ) handler = RedirectHandler() request = httpx.Request( 'GET', - "http://example.org:8080/foo", + "http://example.org:8080/foo", # NOSONAR headers={ AUTHORIZATION_HEADER: "Bearer token", "Cookie": "session=SECRET" @@ -393,17 +393,17 @@ async def test_redirect_with_same_port_keeps_auth_and_cookie(): """Test that redirects to the same port keep Authorization and Cookie headers""" def request_handler(request: httpx.Request): - if request.url == "http://example.org:8080/bar": + if request.url == "http://example.org:8080/bar": # NOSONAR return httpx.Response(200, ) return httpx.Response( FOUND, - headers={LOCATION_HEADER: "http://example.org:8080/bar"}, + headers={LOCATION_HEADER: "http://example.org:8080/bar"}, # NOSONAR ) handler = RedirectHandler() request = httpx.Request( 'GET', - "http://example.org:8080/foo", + "http://example.org:8080/foo", # NOSONAR headers={ AUTHORIZATION_HEADER: "Bearer token", "Cookie": "session=SECRET" @@ -482,7 +482,7 @@ def test_default_scrub_sensitive_headers_removes_on_scheme_change(): original_url = httpx.URL("https://example.com/v1/api") new_request = httpx.Request( "GET", - "http://example.com/v1/api", + "http://example.com/v1/api", # NOSONAR headers={ AUTHORIZATION_HEADER: "Bearer token", "Cookie": "session=SECRET", @@ -523,10 +523,10 @@ def test_default_scrub_sensitive_headers_removes_on_port_change(): """Test that default scrubber removes Authorization and Cookie when port changes""" from kiota_http.middleware.options.redirect_handler_option import default_scrub_sensitive_headers - original_url = httpx.URL("http://example.org:8080/foo") + original_url = httpx.URL("http://example.org:8080/foo") # NOSONAR new_request = httpx.Request( "GET", - "http://example.org:9090/bar", + "http://example.org:9090/bar", # NOSONAR headers={ AUTHORIZATION_HEADER: "Bearer token", "Cookie": "session=SECRET", @@ -545,10 +545,10 @@ def test_default_scrub_sensitive_headers_keeps_on_same_port(): """Test that default scrubber keeps headers when port is the same""" from kiota_http.middleware.options.redirect_handler_option import default_scrub_sensitive_headers - original_url = httpx.URL("http://example.org:8080/foo") + original_url = httpx.URL("http://example.org:8080/foo") # NOSONAR new_request = httpx.Request( "GET", - "http://example.org:8080/bar", + "http://example.org:8080/bar", # NOSONAR headers={ AUTHORIZATION_HEADER: "Bearer token", "Cookie": "session=SECRET",