From 5dd7311cd48e676147138746229d7ab2b429b8a9 Mon Sep 17 00:00:00 2001 From: Paul Tremberth Date: Wed, 19 Oct 2016 14:45:33 +0200 Subject: [PATCH] Move URL credentials stripping to a helper function --- scrapy/spidermiddlewares/referer.py | 43 +++----------- scrapy/utils/url.py | 34 ++++++++++- tests/test_utils_url.py | 92 ++++++++++++++++++++++++++++- 3 files changed, 133 insertions(+), 36 deletions(-) diff --git a/scrapy/spidermiddlewares/referer.py b/scrapy/spidermiddlewares/referer.py index 01f1fdf85..e40e798b8 100644 --- a/scrapy/spidermiddlewares/referer.py +++ b/scrapy/spidermiddlewares/referer.py @@ -2,14 +2,13 @@ RefererMiddleware: populates Request referer field, based on the Response which originated it. """ -from six.moves.urllib.parse import ParseResult, urlunparse - from scrapy.http import Request, Response from scrapy.exceptions import NotConfigured from scrapy import signals from scrapy.utils.python import to_native_str from scrapy.utils.httpobj import urlparse_cached from scrapy.utils.misc import load_object +from scrapy.utils.url import strip_url_credentials LOCAL_SCHEMES = ('about', 'blob', 'data', 'filesystem',) @@ -31,14 +30,10 @@ class ReferrerPolicy(object): raise NotImplementedError() def stripped_referrer(self, req_or_resp): - stripped = self.strip_url(req_or_resp) - if stripped is not None: - return urlunparse(stripped) + return self.strip_url(req_or_resp) def origin_referrer(self, req_or_resp): - stripped = self.strip_url(req_or_resp, origin_only=True) - if stripped is not None: - return urlunparse(stripped) + return self.strip_url(req_or_resp, origin_only=True) def strip_url(self, req_or_resp, origin_only=False): """ @@ -56,33 +51,13 @@ class ReferrerPolicy(object): """ if req_or_resp.url is None or not req_or_resp.url: return None - parsed = urlparse_cached(req_or_resp) - - if parsed.scheme in self.NOREFERRER_SCHEMES: - return None - - netloc = parsed.netloc - # strip username and password if present - if parsed.username or parsed.password: - netloc = netloc.replace('{p.username}:{p.password}@'.format(p=parsed), '') - - # strip standard protocol numbers - # Note: strictly speaking, standard port numbers should only be - # stripped when comparing origins - if parsed.port: - if (parsed.scheme, parsed.port) in (('http', 80), ('https', 443)): - netloc = netloc.replace(':{p.port}'.format(p=parsed), '') - - return ParseResult(parsed.scheme, - netloc, - '/' if origin_only else parsed.path, - '' if origin_only else parsed.params, - '' if origin_only else parsed.query, - '') + parsed_url = urlparse_cached(req_or_resp) + if parsed_url.scheme not in self.NOREFERRER_SCHEMES: + return strip_url_credentials(parsed_url, origin_only=origin_only) def origin(self, req_or_resp): - """Return (scheme, host, path) tuple for a request or response URL.""" - return tuple(self.strip_url(req_or_resp, origin_only=True)[:3]) + """Return serialized origin (scheme, host, path) for a request or response URL.""" + return self.strip_url(req_or_resp, origin_only=True) class NoReferrerPolicy(ReferrerPolicy): @@ -178,7 +153,7 @@ class OriginWhenCrossOriginPolicy(ReferrerPolicy): if origin == self.origin(request): return self.stripped_referrer(response) else: - return urlunparse(origin + ('', '', '')) + return origin class UnsafeUrlPolicy(ReferrerPolicy): diff --git a/scrapy/utils/url.py b/scrapy/utils/url.py index dc1cce4ac..f3ccfb0e8 100644 --- a/scrapy/utils/url.py +++ b/scrapy/utils/url.py @@ -7,7 +7,7 @@ to the w3lib.url module. Always import those from there instead. """ import posixpath import re -from six.moves.urllib.parse import (ParseResult, urldefrag, urlparse) +from six.moves.urllib.parse import (ParseResult, urldefrag, urlparse, urlunparse) # scrapy.utils.url was moved to w3lib.url and import * ensures this # move doesn't break old code @@ -103,3 +103,35 @@ def guess_scheme(url): return any_to_uri(url) else: return add_http_if_no_scheme(url) + + +def strip_url_credentials(url, origin_only=False, keep_fragments=False): + + if url is None: + return None + + if not isinstance(url, ParseResult): + parsed_url = urlparse(url) + else: + parsed_url = url + + netloc = parsed_url.netloc + # strip username and password if present + if parsed_url.username or parsed_url.password: + netloc = netloc.split('@')[-1] + + # strip standard protocol numbers + # Note: strictly speaking, standard port numbers should only be + # stripped when comparing origins + if parsed_url.port: + if (parsed_url.scheme, parsed_url.port) in (('http', 80), ('https', 443)): + netloc = netloc.replace(':{p.port}'.format(p=parsed_url), '') + + return urlunparse(( + parsed_url.scheme, + netloc, + '/' if origin_only else parsed_url.path, + '' if origin_only else parsed_url.params, + '' if origin_only else parsed_url.query, + '' if not keep_fragments else parsed_url.fragment + )) diff --git a/tests/test_utils_url.py b/tests/test_utils_url.py index f46d1d927..f1a5c3196 100644 --- a/tests/test_utils_url.py +++ b/tests/test_utils_url.py @@ -6,7 +6,8 @@ from six.moves.urllib.parse import urlparse from scrapy.spiders import Spider from scrapy.utils.url import (url_is_from_any_domain, url_is_from_spider, - add_http_if_no_scheme, guess_scheme, parse_url) + add_http_if_no_scheme, guess_scheme, + parse_url, strip_url_credentials) __doctests__ = ['scrapy.utils.url'] @@ -241,5 +242,94 @@ for k, args in enumerate ([ setattr (GuessSchemeTest, t_method.__name__, t_method) +class StripUrlCredentials(unittest.TestCase): + + def test_noop(self): + self.assertEqual(strip_url_credentials( + 'http://www.example.com/index.html'), + 'http://www.example.com/index.html') + + def test_noop_query_string(self): + self.assertEqual(strip_url_credentials( + 'http://www.example.com/index.html?somekey=somevalue'), + 'http://www.example.com/index.html?somekey=somevalue') + + def test_fragments(self): + self.assertEqual(strip_url_credentials( + 'http://www.example.com/index.html?somekey=somevalue#section', keep_fragments=True), + 'http://www.example.com/index.html?somekey=somevalue#section') + + def test_noop_trailing_path(self): + self.assertEqual(strip_url_credentials( + 'http://www.example.com/'), + 'http://www.example.com/') + + def test_noop_trailing_path2(self): + self.assertEqual(strip_url_credentials( + 'http://www.example.com'), + 'http://www.example.com') + + def test_trailing_path_origin(self): + self.assertEqual(strip_url_credentials( + 'http://www.example.com', origin_only=True), + 'http://www.example.com/') + + def test_username(self): + # username is stripped (and fragment too) + self.assertEqual(strip_url_credentials( + 'http://username@www.example.com/index.html?somekey=somevalue#section'), + 'http://www.example.com/index.html?somekey=somevalue') + + def test_username_empty_pass(self): + # same as above + self.assertEqual(strip_url_credentials( + 'https://username:@www.example.com/index.html?somekey=somevalue#section'), + 'https://www.example.com/index.html?somekey=somevalue') + + def test_username_password(self): + self.assertEqual(strip_url_credentials( + 'ftp://username:password@www.example.com/index.html?somekey=somevalue#section'), + 'ftp://www.example.com/index.html?somekey=somevalue') + + def test_default_http_port(self): + self.assertEqual(strip_url_credentials( + 'http://username:password@www.example.com:80/index.html'), + 'http://www.example.com/index.html') + + def test_non_default_http_port(self): + self.assertEqual(strip_url_credentials( + 'http://username:password@www.example.com:8080/index.html'), + 'http://www.example.com:8080/index.html') + + def test_default_https_port(self): + self.assertEqual(strip_url_credentials( + 'https://username:password@www.example.com:443/index.html'), + 'https://www.example.com/index.html') + + def test_non_default_https_port(self): + self.assertEqual(strip_url_credentials( + 'https://username:password@www.example.com:442/index.html'), + 'https://www.example.com:442/index.html') + + def test_origin_only(self): + self.assertEqual(strip_url_credentials( + 'http://username:password@www.example.com/index.html', origin_only=True), + 'http://www.example.com/') + + def test_default_http_port_origin_only(self): + self.assertEqual(strip_url_credentials( + 'http://username:password@www.example.com:80/index.html', origin_only=True), + 'http://www.example.com/') + + def test_non_default_http_port_origin_only(self): + self.assertEqual(strip_url_credentials( + 'http://username:password@www.example.com:8008/index.html', origin_only=True), + 'http://www.example.com:8008/') + + def test_default_https_port_origin_only(self): + self.assertEqual(strip_url_credentials( + 'https://username:password@www.example.com:443/index.html', origin_only=True), + 'https://www.example.com/') + if __name__ == "__main__": unittest.main()