diff --git a/scrapy/downloadermiddlewares/auth.py b/scrapy/downloadermiddlewares/auth.py index 211d2ecfc..b1242c69d 100644 --- a/scrapy/downloadermiddlewares/auth.py +++ b/scrapy/downloadermiddlewares/auth.py @@ -1,19 +1,38 @@ """ -HTTP basic auth downloader middleware +HTTP/FTP Authorization downloader middleware See documentation in docs/topics/downloader-middleware.rst """ +from six.moves.urllib.parse import urlunparse from w3lib.http import basic_auth_header from scrapy import signals +from scrapy.utils.httpobj import urlparse_cached -from six.moves.urllib.parse import urlparse + +def credstrip_url(parsed_url): + """Strip username and password from an urlparse'd URL""" + return urlunparse(( + parsed_url.scheme, + parsed_url.netloc.split('@')[-1], + parsed_url.path, + parsed_url.params, + parsed_url.query, + parsed_url.fragment)) class AuthMiddleware(object): - """Set Basic HTTP Authorization header - (http_user and http_pass spider class attributes)""" + """ + Populate authorization credentials for HTTP and FTP requests. + + For http(s):// requests, set Basic HTTP Authorization header, + either from http_user and http_pass spider attributes, + or from URL netloc parsing. + + Also handle FTP credentials from ftp://user:password@... URLs, + populating request's meta accordingly. + """ @classmethod def from_crawler(cls, crawler): @@ -28,19 +47,33 @@ class AuthMiddleware(object): self.auth = basic_auth_header(usr, pwd) def process_request(self, request, spider): - auth = getattr(self, 'auth', None) - if auth and 'Authorization' not in request.headers: - request.headers['Authorization'] = auth + url = urlparse_cached(request) + if url.scheme.startswith('http'): + # do not override Auth header set priorly + if 'Authorization' in request.headers: + return - # credentials from url are supposed to override spider settings - url = urlparse(request.url) - if url.username and url.password: - if url.scheme.startswith('ftp'): - request.meta['ftp_user'] = url.username - request.meta['ftp_password'] = url.password - elif url.scheme.startswith('http'): - request.headers['Authorization'] = basic_auth_header(url.username, url.password) + new_url = None + # credentials from URL override spider attributes + if url.username or url.password: + auth = basic_auth_header(url.username, url.password) - # no credentials in new url - new_url = url.scheme + '://' + request.url.split('@')[-1] - return request.replace(url=new_url) + # no credentials in new url + new_url = credstrip_url(url) + + else: + auth = getattr(self, 'auth', None) + + if auth: + request.headers['Authorization'] = auth + if new_url: + return request.replace(url=new_url) + + elif url.scheme.startswith('ftp'): + if url.username or url.password: + # priorly set credentials take precedence + request.meta.setdefault('ftp_user', url.username) + request.meta.setdefault('ftp_password', url.password) + + # no credentials in new url + return request.replace(url=credstrip_url(url)) diff --git a/tests/test_downloadermiddleware_auth.py b/tests/test_downloadermiddleware_auth.py index d5b5b4020..c58eabcb6 100644 --- a/tests/test_downloadermiddleware_auth.py +++ b/tests/test_downloadermiddleware_auth.py @@ -10,6 +10,32 @@ class TestSpider(Spider): http_pass = 'bar' +class NoAuthTestSpider(Spider): + """A test spider that does not set http auth atttributes""" + + +class AuthMiddlewareNoAuthTest(unittest.TestCase): + + def setUp(self): + self.mw = AuthMiddleware() + self.spider = NoAuthTestSpider('bar') + self.mw.spider_opened(self.spider) + + def tearDown(self): + del self.mw + + def test_no_auth_http(self): + req = Request('http://scrapytest.org/') + assert self.mw.process_request(req, self.spider) is None + self.assertNotIn('Authorization', req.headers) + + def test_no_auth_ftp(self): + req = Request('ftp://scrapytest.org/') + assert self.mw.process_request(req, self.spider) is None + self.assertNotIn('ftp_user', req.meta) + self.assertNotIn('ftp_password', req.meta) + + class AuthMiddlewareTest(unittest.TestCase): def setUp(self): @@ -20,6 +46,9 @@ class AuthMiddlewareTest(unittest.TestCase): def tearDown(self): del self.mw + +class AuthMiddlewareHttpAuthTest(AuthMiddlewareTest): + def test_auth(self): req = Request('http://scrapytest.org/') assert self.mw.process_request(req, self.spider) is None @@ -38,10 +67,104 @@ class AuthMiddlewareTest(unittest.TestCase): self.assertEquals(new_req.headers['Authorization'], b'Basic dXNlcm5hbWU6cGFzc3dvcmQ=') self.assertEquals(new_req.url, 'http://scrapytest.org/') + def test_auth_from_https_url(self): + req = Request('https://username:password@scrapytest.org/') + new_req = self.mw.process_request(req, self.spider) + assert new_req is not None + self.assertEquals(new_req.headers['Authorization'], b'Basic dXNlcm5hbWU6cGFzc3dvcmQ=') + self.assertEquals(new_req.url, 'https://scrapytest.org/') + + def test_auth_from_http_url_no_spider_attrs(self): + class AnotherTestSpider(Spider): + pass + req = Request('http://username:password@scrapytest.org/') + new_req = self.mw.process_request(req, AnotherTestSpider('bar')) + assert new_req is not None + self.assertEquals(new_req.headers['Authorization'], b'Basic dXNlcm5hbWU6cGFzc3dvcmQ=') + self.assertEquals(new_req.url, 'http://scrapytest.org/') + + def test_auth_from_https_url_no_spider_attrs(self): + class AnotherTestSpider(Spider): + pass + req = Request('https://username:password@scrapytest.org/') + new_req = self.mw.process_request(req, AnotherTestSpider('bar')) + assert new_req is not None + self.assertEquals(new_req.headers['Authorization'], b'Basic dXNlcm5hbWU6cGFzc3dvcmQ=') + self.assertEquals(new_req.url, 'https://scrapytest.org/') + + def test_auth_from_http_url_empty_pass(self): + req = Request('http://username:@scrapytest.org/') + new_req = self.mw.process_request(req, self.spider) + assert new_req is not None + self.assertEquals(new_req.headers['Authorization'], b'Basic dXNlcm5hbWU6') + self.assertEquals(new_req.url, 'http://scrapytest.org/') + + def test_auth_from_http_url_pass_none(self): + req = Request('http://username@scrapytest.org/') + new_req = self.mw.process_request(req, self.spider) + assert new_req is not None + self.assertEquals(new_req.headers['Authorization'], b'Basic dXNlcm5hbWU6Tm9uZQ==') + self.assertEquals(new_req.url, 'http://scrapytest.org/') + + def test_auth_from_http_url_empty_user(self): + req = Request('http://:password@scrapytest.org/') + new_req = self.mw.process_request(req, self.spider) + assert new_req is not None + self.assertEquals(new_req.headers['Authorization'], b'Basic OnBhc3N3b3Jk') + self.assertEquals(new_req.url, 'http://scrapytest.org/') + + +class AuthMiddlewareFtpAuthTest(AuthMiddlewareTest): + + def test_no_auth_from_ftp_url_meta_unchanged(self): + usr, pwd = 'u', 'p' + req = Request('ftp://scrapytest.org/', + meta={"ftp_user": usr, "ftp_password": pwd}) + assert self.mw.process_request(req, self.spider) is None + self.assertEquals(req.meta['ftp_user'], usr) + self.assertEquals(req.meta['ftp_password'], pwd) + + def test_auth_from_ftp_url_meta_unchanged(self): + """Request's meta credentials are kept as-is, + but URL is stripped from credentials + """ + usr, pwd = 'u', 'p' + req = Request('ftp://username:password@scrapytest.org/', + meta={"ftp_user": usr, "ftp_password": pwd}) + new_req = self.mw.process_request(req, self.spider) + assert new_req is not None + self.assertEquals(new_req.meta['ftp_user'], usr) + self.assertEquals(new_req.meta['ftp_password'], pwd) + self.assertEquals(new_req.url, 'ftp://scrapytest.org/') + def test_auth_from_ftp_url(self): req = Request('ftp://username:password@scrapytest.org/') new_req = self.mw.process_request(req, self.spider) assert new_req is not None - self.assertIn('ftp_user', new_req.meta) - self.assertIn('ftp_password', new_req.meta) - self.assertNotIn('@', new_req.url) + self.assertEquals(req.meta['ftp_user'], 'username') + self.assertEquals(req.meta['ftp_password'], 'password') + self.assertEquals(new_req.url, 'ftp://scrapytest.org/') + + def test_auth_from_ftp_url_empty_user(self): + req = Request('ftp://:password@scrapytest.org/') + new_req = self.mw.process_request(req, self.spider) + assert new_req is not None + self.assertEquals(req.meta['ftp_user'], '') + self.assertEquals(req.meta['ftp_password'], 'password') + self.assertEquals(new_req.url, 'ftp://scrapytest.org/') + + def test_auth_from_ftp_url_empty_pass(self): + req = Request('ftp://username:@scrapytest.org/') + new_req = self.mw.process_request(req, self.spider) + assert new_req is not None + self.assertEquals(req.meta['ftp_user'], 'username') + self.assertEquals(req.meta['ftp_password'], '') + self.assertEquals(new_req.url, 'ftp://scrapytest.org/') + + def test_auth_from_ftp_url_pass_none(self): + req = Request('ftp://username@scrapytest.org/') + new_req = self.mw.process_request(req, self.spider) + assert new_req is not None + self.assertEquals(req.meta['ftp_user'], 'username') + self.assertEquals(req.meta['ftp_password'], None) + self.assertEquals(new_req.url, 'ftp://scrapytest.org/')