From 3331e2aecdbf575dd60abef4df79c52d78610a83 Mon Sep 17 00:00:00 2001
From: Bruce Merry <bmerry@ska.ac.za>
Date: Thu, 28 Jun 2018 16:38:42 +0200
Subject: [PATCH 1/2] Strip Authorization header whenever root URL changes
Previously the header was stripped only if the hostname changed, but in
an https -> http redirect that can leak the credentials on the wire
(#4716). Based on with RFC 7235 section 2.2, the header is now stripped
if the "canonical root URL" (scheme+authority) has changed, by checking
scheme, hostname and port.
---
requests/sessions.py | 4 +++-
tests/test_requests.py | 12 +++++++++++-
2 files changed, 14 insertions(+), 2 deletions(-)
diff --git a/requests/sessions.py b/requests/sessions.py
index dd525e2ac..702cd73ec 100644
--- a/requests/sessions.py
+++ b/requests/sessions.py
@@ -242,7 +242,9 @@ def rebuild_auth(self, prepared_request, response):
original_parsed = urlparse(response.request.url)
redirect_parsed = urlparse(url)
- if (original_parsed.hostname != redirect_parsed.hostname):
+ if (original_parsed.hostname != redirect_parsed.hostname
+ or original_parsed.port != redirect_parsed.port
+ or original_parsed.scheme != redirect_parsed.scheme):
del headers['Authorization']
# .netrc might have more auth for us on our new host.
diff --git a/tests/test_requests.py b/tests/test_requests.py
index fd04ad270..b05d8ebb2 100644
--- a/tests/test_requests.py
+++ b/tests/test_requests.py
@@ -1581,7 +1581,17 @@ def test_auth_is_stripped_on_redirect_off_host(self, httpbin):
auth=('user', 'pass'),
)
assert r.history[0].request.headers['Authorization']
- assert not r.request.headers.get('Authorization', '')
+ assert 'Authorization' not in r.request.headers
+
+ def test_auth_is_stripped_on_scheme_redirect(self, httpbin, httpbin_secure, httpbin_ca_bundle):
+ r = requests.get(
+ httpbin_secure('redirect-to'),
+ params={'url': httpbin('get')},
+ auth=('user', 'pass'),
+ verify=httpbin_ca_bundle
+ )
+ assert r.history[0].request.headers['Authorization']
+ assert 'Authorization' not in r.request.headers
def test_auth_is_retained_for_redirect_on_host(self, httpbin):
r = requests.get(httpbin('redirect/1'), auth=('user', 'pass'))
From 857e9b7ac20c3accf4cc328f594aecb8b6a644a6 Mon Sep 17 00:00:00 2001
From: Bruce Merry <bmerry@ska.ac.za>
Date: Tue, 14 Aug 2018 13:30:43 +0200
Subject: [PATCH 2/2] Rework authorization stripping logic as discussed
The exception for http->https upgrade now requires the standard HTTP(S)
ports to be used, either implicitly (no port specified) or explicitly.
---
requests/sessions.py | 26 ++++++++++++++++++--------
tests/test_requests.py | 33 ++++++++++++++++++++++-----------
2 files changed, 40 insertions(+), 19 deletions(-)
diff --git a/requests/sessions.py b/requests/sessions.py
index 702cd73ec..27d0e9717 100644
--- a/requests/sessions.py
+++ b/requests/sessions.py
@@ -115,6 +115,22 @@ def get_redirect_target(self, resp):
return to_native_string(location, 'utf8')
return None
+ def should_strip_auth(self, old_url, new_url):
+ """Decide whether Authorization header should be removed when redirecting"""
+ old_parsed = urlparse(old_url)
+ new_parsed = urlparse(new_url)
+ if old_parsed.hostname != new_parsed.hostname:
+ return True
+ # Special case: allow http -> https redirect when using the standard
+ # ports. This isn't specified by RFC 7235, but is kept to avoid
+ # breaking backwards compatibility with older versions of requests
+ # that allowed any redirects on the same host.
+ if (old_parsed.scheme == 'http' and old_parsed.port in (80, None)
+ and new_parsed.scheme == 'https' and new_parsed.port in (443, None)):
+ return False
+ # Standard case: root URI must match
+ return old_parsed.port != new_parsed.port or old_parsed.scheme != new_parsed.scheme
+
def resolve_redirects(self, resp, req, stream=False, timeout=None,
verify=True, cert=None, proxies=None, yield_requests=False, **adapter_kwargs):
"""Receives a Response. Returns a generator of Responses or Requests."""
@@ -236,16 +252,10 @@ def rebuild_auth(self, prepared_request, response):
headers = prepared_request.headers
url = prepared_request.url
- if 'Authorization' in headers:
+ if 'Authorization' in headers and self.should_strip_auth(response.request.url, url):
# If we get redirected to a new host, we should strip out any
# authentication headers.
- original_parsed = urlparse(response.request.url)
- redirect_parsed = urlparse(url)
-
- if (original_parsed.hostname != redirect_parsed.hostname
- or original_parsed.port != redirect_parsed.port
- or original_parsed.scheme != redirect_parsed.scheme):
- del headers['Authorization']
+ del headers['Authorization']
# .netrc might have more auth for us on our new host.
new_auth = get_netrc_auth(url) if self.trust_env else None
diff --git a/tests/test_requests.py b/tests/test_requests.py
index b05d8ebb2..660437988 100644
--- a/tests/test_requests.py
+++ b/tests/test_requests.py
@@ -1573,17 +1573,7 @@ def test_nonhttp_schemes_dont_check_URLs(self):
preq = req.prepare()
assert test_url == preq.url
- @pytest.mark.xfail(raises=ConnectionError)
- def test_auth_is_stripped_on_redirect_off_host(self, httpbin):
- r = requests.get(
- httpbin('redirect-to'),
- params={'url': 'http://www.google.co.uk'},
- auth=('user', 'pass'),
- )
- assert r.history[0].request.headers['Authorization']
- assert 'Authorization' not in r.request.headers
-
- def test_auth_is_stripped_on_scheme_redirect(self, httpbin, httpbin_secure, httpbin_ca_bundle):
+ def test_auth_is_stripped_on_http_downgrade(self, httpbin, httpbin_secure, httpbin_ca_bundle):
r = requests.get(
httpbin_secure('redirect-to'),
params={'url': httpbin('get')},
@@ -1600,6 +1590,27 @@ def test_auth_is_retained_for_redirect_on_host(self, httpbin):
assert h1 == h2
+ def test_should_strip_auth_host_change(self):
+ s = requests.Session()
+ assert s.should_strip_auth('http://example.com/foo', 'http://another.example.com/')
+
+ def test_should_strip_auth_http_downgrade(self):
+ s = requests.Session()
+ assert s.should_strip_auth('https://example.com/foo', 'http://example.com/bar')
+
+ def test_should_strip_auth_https_upgrade(self):
+ s = requests.Session()
+ assert not s.should_strip_auth('http://example.com/foo', 'https://example.com/bar')
+ assert not s.should_strip_auth('http://example.com:80/foo', 'https://example.com/bar')
+ assert not s.should_strip_auth('http://example.com/foo', 'https://example.com:443/bar')
+ # Non-standard ports should trigger stripping
+ assert s.should_strip_auth('http://example.com:8080/foo', 'https://example.com/bar')
+ assert s.should_strip_auth('http://example.com/foo', 'https://example.com:8443/bar')
+
+ def test_should_strip_auth_port_change(self):
+ s = requests.Session()
+ assert s.should_strip_auth('http://example.com:1234/foo', 'https://example.com:4321/bar')
+
def test_manual_redirect_with_partial_body_read(self, httpbin):
s = requests.Session()
r1 = s.get(httpbin('redirect/2'), allow_redirects=False, stream=True)