diff --git a/docs/topics/downloader-middleware.rst b/docs/topics/downloader-middleware.rst index dec9904d2..731e6baeb 100644 --- a/docs/topics/downloader-middleware.rst +++ b/docs/topics/downloader-middleware.rst @@ -289,7 +289,10 @@ HttpAuthMiddleware "HTTPAUTH_DOMAIN": "intranet.example.com", } - # .. rest of the spider code omitted ... + You can also let + :class:`~scrapy.downloadermiddlewares.uriuserinfo.UriUserInfoMiddleware` + fill :reqmeta:`http_user` and :reqmeta:`http_pass` from the credentials in + the request URL. Example using per-request meta: @@ -1206,6 +1209,16 @@ DownloaderStats To use this middleware you must enable the :setting:`DOWNLOADER_STATS` setting. + +UriUserInfoMiddleware +--------------------- + +.. module:: scrapy.downloadermiddlewares.uriuserinfo + :synopsis: URI Userinfo Middleware + +.. autoclass:: UriUserInfoMiddleware + + UserAgentMiddleware ------------------- diff --git a/docs/topics/request-response.rst b/docs/topics/request-response.rst index 1d97e39b6..8a0487929 100644 --- a/docs/topics/request-response.rst +++ b/docs/topics/request-response.rst @@ -839,8 +839,8 @@ Those are: * :reqmeta:`download_slot` * :reqmeta:`download_warnsize` * :reqmeta:`download_timeout` -* ``ftp_password`` (See :setting:`FTP_PASSWORD` for more info) -* ``ftp_user`` (See :setting:`FTP_USER` for more info) +* :reqmeta:`ftp_password` +* :reqmeta:`ftp_user` * :reqmeta:`give_up_log_level` * :reqmeta:`handle_httpstatus_all` * :reqmeta:`handle_httpstatus_list` @@ -940,6 +940,10 @@ http_auth_domain Overrides :setting:`HTTPAUTH_DOMAIN` for this request. +If this key is not set, :reqmeta:`http_user` and :reqmeta:`http_pass` are only +sent to the origin (scheme, host and port) of the request where they were first +used, e.g. they are not sent after a cross-origin redirect. + .. reqmeta:: http_pass http_pass diff --git a/docs/topics/settings.rst b/docs/topics/settings.rst index e58815fad..987fd3c98 100644 --- a/docs/topics/settings.rst +++ b/docs/topics/settings.rst @@ -910,6 +910,7 @@ Default: { "scrapy.downloadermiddlewares.offsite.OffsiteMiddleware": 50, "scrapy.downloadermiddlewares.robotstxt.RobotsTxtMiddleware": 100, + "scrapy.downloadermiddlewares.uriuserinfo.UriUserInfoMiddleware": 200, "scrapy.downloadermiddlewares.httpauth.HttpAuthMiddleware": 300, "scrapy.downloadermiddlewares.downloadtimeout.DownloadTimeoutMiddleware": 350, "scrapy.downloadermiddlewares.defaultheaders.DefaultHeadersMiddleware": 400, @@ -1459,6 +1460,13 @@ Default: ``"guest"`` The password to use for FTP connections when there is no ``"ftp_password"`` in ``Request`` meta. +It can be overriden in a request in any of the following ways: + +- Specifying ``ftp_password`` in :attr:`Request.meta ` + +- Specifying the password in :attr:`Request.url ` + (see :class:`~scrapy.downloadermiddlewares.uriuserinfo.UriUserInfoMiddleware`) + .. note:: Paraphrasing `RFC 1635`_, although it is common to use either the password "guest" or one's e-mail address for anonymous FTP, @@ -1481,8 +1489,14 @@ FTP_USER Default: ``"anonymous"`` -The username to use for FTP connections when there is no ``"ftp_user"`` -in ``Request`` meta. +The default username to use for FTP connections. + +It can be overriden in a request in any of the following ways: + +- Specifying ``ftp_user`` in :attr:`Request.meta ` + +- Specifying the username in :attr:`Request.url ` + (see :class:`~scrapy.downloadermiddlewares.uriuserinfo.UriUserInfoMiddleware`) .. note:: diff --git a/scrapy/downloadermiddlewares/httpauth.py b/scrapy/downloadermiddlewares/httpauth.py index b09abdadd..79b3beac7 100644 --- a/scrapy/downloadermiddlewares/httpauth.py +++ b/scrapy/downloadermiddlewares/httpauth.py @@ -15,6 +15,7 @@ from scrapy import Request, Spider, signals from scrapy.exceptions import ScrapyDeprecationWarning from scrapy.settings import SETTINGS_PRIORITIES from scrapy.utils.decorators import _warn_spider_arg +from scrapy.utils.httpobj import urlparse_cached from scrapy.utils.url import url_is_from_any_domain if TYPE_CHECKING: @@ -25,6 +26,32 @@ if TYPE_CHECKING: from scrapy.http import Response +_DEFAULT_PORTS = { + "http": 80, + "https": 443, +} + + +def _origin(request: Request) -> str: + parsed_url = urlparse_cached(request) + scheme = parsed_url.scheme + netloc = ( + parsed_url.netloc + if parsed_url.port != _DEFAULT_PORTS[scheme] + else parsed_url.hostname + ) + return f"{scheme}://{netloc}" + + +def _setdefault_auth_origin(request: Request) -> str: + origin: str | None = request.meta.get("auth_origin") + if origin: + return origin + origin = _origin(request) + request.meta["auth_origin"] = origin + return origin + + class HttpAuthMiddleware: """Set Basic HTTP Authorization header.""" @@ -70,14 +97,22 @@ class HttpAuthMiddleware: def process_request( self, request: Request, spider: Spider | None = None ) -> Request | Response | None: - if b"Authorization" in request.headers: + if ( + b"Authorization" in request.headers + or urlparse_cached(request).scheme not in _DEFAULT_PORTS + ): return None # Per-request meta overrides usr = request.meta.get("http_user", "") pwd = request.meta.get("http_pass", "") if usr or pwd: domain = request.meta.get("http_auth_domain") - if not domain or url_is_from_any_domain(request.url, [domain]): + allowed = ( + url_is_from_any_domain(request.url, [domain]) + if domain + else _setdefault_auth_origin(request) == _origin(request) + ) + if allowed: request.headers[b"Authorization"] = basic_auth_header(usr, pwd) return None # Middleware-level auth diff --git a/scrapy/downloadermiddlewares/uriuserinfo.py b/scrapy/downloadermiddlewares/uriuserinfo.py new file mode 100644 index 000000000..c7870caa5 --- /dev/null +++ b/scrapy/downloadermiddlewares/uriuserinfo.py @@ -0,0 +1,60 @@ +from __future__ import annotations + +from typing import TYPE_CHECKING +from urllib.parse import unquote, urlunparse + +from scrapy.utils.httpobj import urlparse_cached + +if TYPE_CHECKING: + from scrapy import Request, Spider + from scrapy.http import Response + + +class UriUserInfoMiddleware: + """Downloader middleware that replaces `URI userinfo`_ data (user + credentials for HTTP or FTP specified in the request URL) with the + corresponding meta keys for later middlewares or download handlers to use + them for authentication. + + It sets: + + - :reqmeta:`ftp_user` and :reqmeta:`ftp_password` for FTP requests + + - :reqmeta:`http_user` and :reqmeta:`http_pass` for HTTP and HTTPS + requests + + .. _URI userinfo: https://tools.ietf.org/html/rfc2396.html#section-3.2.2 + """ + + def process_request( + self, request: Request, spider: Spider | None = None + ) -> Request | Response | None: + parsed_url = urlparse_cached(request) + if parsed_url.username is None and parsed_url.password is None: + return None + + if parsed_url.scheme.startswith("http"): + username_field, password_field = "http_user", "http_pass" + elif parsed_url.scheme.startswith("ftp"): + username_field, password_field = "ftp_user", "ftp_password" + else: + return None + + for key, value in ( + (username_field, parsed_url.username), + (password_field, parsed_url.password), + ): + if value is not None: + request.meta.setdefault(key, unquote(value)) + + userinfoless_url = urlunparse( + ( + parsed_url.scheme, + parsed_url.netloc.split("@")[-1], + parsed_url.path, + parsed_url.params, + parsed_url.query, + parsed_url.fragment, + ) + ) + return request.replace(url=userinfoless_url) diff --git a/scrapy/settings/default_settings.py b/scrapy/settings/default_settings.py index b97844103..d2eaaf397 100644 --- a/scrapy/settings/default_settings.py +++ b/scrapy/settings/default_settings.py @@ -318,6 +318,7 @@ DOWNLOADER_MIDDLEWARES_BASE = { # Engine side "scrapy.downloadermiddlewares.offsite.OffsiteMiddleware": 50, "scrapy.downloadermiddlewares.robotstxt.RobotsTxtMiddleware": 100, + "scrapy.downloadermiddlewares.uriuserinfo.UriUserInfoMiddleware": 200, "scrapy.downloadermiddlewares.httpauth.HttpAuthMiddleware": 300, "scrapy.downloadermiddlewares.downloadtimeout.DownloadTimeoutMiddleware": 350, "scrapy.downloadermiddlewares.defaultheaders.DefaultHeadersMiddleware": 400, diff --git a/tests/test_downloadermiddleware_httpauth.py b/tests/test_downloadermiddleware_httpauth.py index e4b414f45..002e049c4 100644 --- a/tests/test_downloadermiddleware_httpauth.py +++ b/tests/test_downloadermiddleware_httpauth.py @@ -1,7 +1,8 @@ +from __future__ import annotations + from typing import Any import pytest -from w3lib.http import basic_auth_header from scrapy.downloadermiddlewares.httpauth import HttpAuthMiddleware from scrapy.exceptions import ScrapyDeprecationWarning @@ -10,170 +11,281 @@ from scrapy.spiders import Spider from scrapy.utils.misc import build_from_crawler from scrapy.utils.test import get_crawler -_DOMAIN_NOT_SET = object() + +def _build_mw(settings: dict[str, Any], spider: Spider) -> HttpAuthMiddleware: + mw = build_from_crawler(HttpAuthMiddleware, get_crawler(settings_dict=settings)) + if getattr(spider, "http_user", "") or getattr(spider, "http_pass", ""): + with pytest.warns(ScrapyDeprecationWarning, match="spider attributes"): + mw.spider_opened(spider) + else: + mw.spider_opened(spider) + return mw -def make_mw( - user: str = "", passwd: str = "", domain: str | object = _DOMAIN_NOT_SET -) -> HttpAuthMiddleware: - settings: dict[str, Any] = { - "HTTPAUTH_USER": user, - "HTTPAUTH_PASS": passwd, - } - if domain is not _DOMAIN_NOT_SET: - settings["HTTPAUTH_DOMAIN"] = domain - return build_from_crawler(HttpAuthMiddleware, get_crawler(settings_dict=settings)) - - -# --- Spider attribute tests (deprecated) --- - - -class LegacySpider(Spider): - http_user = "foo" - http_pass = "bar" - - -class DomainSpider(Spider): - http_user = "foo" - http_pass = "bar" - http_auth_domain = "example.com" - - -class AnyDomainSpider(Spider): - http_user = "foo" - http_pass = "bar" - http_auth_domain = None - - -class TestHttpAuthMiddlewareLegacySpiderAttr: - def test_missing_domain_raises(self): - mw = HttpAuthMiddleware() - with pytest.warns(ScrapyDeprecationWarning), pytest.raises(AttributeError): - mw.spider_opened(LegacySpider("foo")) - - def test_domain_spider(self): - mw = HttpAuthMiddleware() - with pytest.warns(ScrapyDeprecationWarning): - mw.spider_opened(DomainSpider("foo")) - req = Request("http://example.com/") - mw.process_request(req) - assert req.headers["Authorization"] == basic_auth_header("foo", "bar") - - def test_no_auth_wrong_domain(self): - mw = HttpAuthMiddleware() - with pytest.warns(ScrapyDeprecationWarning): - mw.spider_opened(DomainSpider("foo")) - req = Request("http://other.com/") - mw.process_request(req) - assert "Authorization" not in req.headers - - def test_any_domain_spider(self): - mw = HttpAuthMiddleware() - with pytest.warns(ScrapyDeprecationWarning): - mw.spider_opened(AnyDomainSpider("foo")) - req = Request("http://anywhere.com/") - mw.process_request(req) - assert req.headers["Authorization"] == basic_auth_header("foo", "bar") - - -# --- Settings-based tests --- - - -class TestHttpAuthMiddlewareSettings: - def test_no_auth(self): - mw = make_mw() - req = Request("http://example.com/") - mw.process_request(req) - assert "Authorization" not in req.headers - - def test_auth_without_domain_raises(self): - with pytest.raises(ValueError, match="HTTPAUTH_DOMAIN"): - make_mw(user="foo", passwd="bar") - - def test_auth_all_domains(self): - mw = make_mw(user="foo", passwd="bar", domain=None) - req = Request("http://example.com/") - mw.process_request(req) - assert req.headers["Authorization"] == basic_auth_header("foo", "bar") - - def test_auth_domain_match(self): - mw = make_mw(user="foo", passwd="bar", domain="example.com") - req = Request("http://example.com/") - mw.process_request(req) - assert req.headers["Authorization"] == basic_auth_header("foo", "bar") - - def test_auth_subdomain(self): - mw = make_mw(user="foo", passwd="bar", domain="example.com") - req = Request("http://sub.example.com/") - mw.process_request(req) - assert req.headers["Authorization"] == basic_auth_header("foo", "bar") - - def test_no_auth_wrong_domain(self): - mw = make_mw(user="foo", passwd="bar", domain="example.com") - req = Request("http://other.com/") - mw.process_request(req) - assert "Authorization" not in req.headers - - def test_auth_already_set(self): - mw = make_mw(user="foo", passwd="bar", domain="example.com") - req = Request("http://example.com/", headers={"Authorization": "Digest 123"}) - mw.process_request(req) - assert req.headers["Authorization"] == b"Digest 123" - - -# --- Per-request meta tests --- - - -class TestHttpAuthMiddlewareMeta: - def test_meta_auth_no_domain(self): - mw = make_mw() - req = Request("http://example.com/", meta={"http_user": "u", "http_pass": "p"}) - mw.process_request(req) - assert req.headers["Authorization"] == basic_auth_header("u", "p") - - def test_meta_auth_domain_match(self): - mw = make_mw() - req = Request( - "http://example.com/", - meta={ - "http_user": "u", - "http_pass": "p", - "http_auth_domain": "example.com", +@pytest.mark.parametrize( + ("config", "expected"), + [ + # Baseline + ({}, None), + # Settings. + # HTTPAUTH_DOMAIN=None allows any domain. + ( + {"settings": {"HTTPAUTH_USER": "cu", "HTTPAUTH_DOMAIN": None}}, + b"Basic Y3U6", + ), + ( + {"settings": {"HTTPAUTH_PASS": "cp", "HTTPAUTH_DOMAIN": None}}, + b"Basic OmNw", + ), + ( + { + "settings": { + "HTTPAUTH_USER": "cu", + "HTTPAUTH_PASS": "cp", + "HTTPAUTH_DOMAIN": None, + } }, - ) - mw.process_request(req) - assert req.headers["Authorization"] == basic_auth_header("u", "p") - - def test_meta_auth_domain_no_match(self): - mw = make_mw() - req = Request( - "http://other.com/", - meta={ - "http_user": "u", - "http_pass": "p", - "http_auth_domain": "example.com", + b"Basic Y3U6Y3A=", + ), + # HTTPAUTH_DOMAIN=domain allows only that domain and subdomains. + ( + {"settings": {"HTTPAUTH_USER": "cu", "HTTPAUTH_DOMAIN": "a.example"}}, + b"Basic Y3U6", + ), + ( + { + "url": "https://s.a.example/a", + "settings": {"HTTPAUTH_USER": "cu", "HTTPAUTH_DOMAIN": "a.example"}, }, - ) - mw.process_request(req) - assert "Authorization" not in req.headers + b"Basic Y3U6", + ), + ( + {"settings": {"HTTPAUTH_USER": "cu", "HTTPAUTH_DOMAIN": "b.example"}}, + None, + ), + # HTTPAUTH_DOMAIN must be set if HTTPAUTH_USER or HTTPAUTH_PASS are. + ({"settings": {"HTTPAUTH_USER": "cu"}}, ValueError), + ({"settings": {"HTTPAUTH_PASS": "cp"}}, ValueError), + # Spider attributes. + # http_auth_domain=None allows any domain. + ( + {"spider_attributes": {"http_user": "su", "http_auth_domain": None}}, + b"Basic c3U6", + ), + ( + {"spider_attributes": {"http_pass": "sp", "http_auth_domain": None}}, + b"Basic OnNw", + ), + ( + { + "spider_attributes": { + "http_user": "su", + "http_pass": "sp", + "http_auth_domain": None, + } + }, + b"Basic c3U6c3A=", + ), + # http_auth_domain=domain allows only that domain and subdomains. + ( + {"spider_attributes": {"http_user": "su", "http_auth_domain": "a.example"}}, + b"Basic c3U6", + ), + ( + { + "url": "https://s.a.example/a", + "spider_attributes": { + "http_user": "su", + "http_auth_domain": "a.example", + }, + }, + b"Basic c3U6", + ), + ( + {"spider_attributes": {"http_user": "su", "http_auth_domain": "b.example"}}, + None, + ), + # http_auth_domain must be defined if http_user or http_pass are. + ({"spider_attributes": {"http_user": "su"}}, AttributeError), + # Spider attributes take priority over settings. + ( + { + "settings": { + "HTTPAUTH_USER": "cu", + "HTTPAUTH_PASS": "cp", + "HTTPAUTH_DOMAIN": None, + }, + "spider_attributes": {"http_user": "su", "http_auth_domain": None}, + }, + b"Basic c3U6", + ), + # Request.meta. + ({"meta": {"http_user": "mu"}}, b"Basic bXU6"), + ({"meta": {"http_pass": "mp"}}, b"Basic Om1w"), + ({"meta": {"http_user": "mu", "http_pass": "mp"}}, b"Basic bXU6bXA="), + # Request.meta["http_auth_domain"]=domain allows only that domain and + # its subdomains. + ( + {"meta": {"http_user": "mu", "http_auth_domain": "a.example"}}, + b"Basic bXU6", + ), + ( + { + "url": "https://s.a.example/a", + "meta": {"http_user": "mu", "http_auth_domain": "a.example"}, + }, + b"Basic bXU6", + ), + ({"meta": {"http_user": "mu", "http_auth_domain": "b.example"}}, None), + # Without http_auth_domain, credentials from Request.meta are limited + # to the origin of the request where they were first seen. + # + # Note: auth_origin is not meant to be set by users, it is set the + # first time a request is processed by the middleware. See + # test_origin_setdefault. + ( + {"meta": {"auth_origin": "https://a.example", "http_user": "mu"}}, + b"Basic bXU6", + ), + ( + { + "url": "https://a.example:443", + "meta": {"auth_origin": "https://a.example", "http_user": "mu"}, + }, + b"Basic bXU6", + ), + ( + { + "url": "https://s.a.example", + "meta": {"auth_origin": "https://a.example", "http_user": "mu"}, + }, + None, + ), + ({"meta": {"auth_origin": "http://a.example", "http_user": "mu"}}, None), + ({"meta": {"auth_origin": "https://a.example:1", "http_user": "mu"}}, None), + ({"meta": {"auth_origin": "https://b.example", "http_user": "mu"}}, None), + # http_auth_domain takes priority over auth_origin. + ( + { + "meta": { + "auth_origin": "https://b.example", + "http_user": "mu", + "http_auth_domain": "a.example", + } + }, + b"Basic bXU6", + ), + # Takes priority over settings and spider attributes. + ( + { + "meta": {"http_user": "mu"}, + "settings": { + "HTTPAUTH_USER": "cu", + "HTTPAUTH_PASS": "cp", + "HTTPAUTH_DOMAIN": None, + }, + }, + b"Basic bXU6", + ), + ( + { + "meta": {"http_user": "mu"}, + "spider_attributes": { + "http_user": "su", + "http_pass": "sp", + "http_auth_domain": None, + }, + }, + b"Basic bXU6", + ), + # If the Authorization header is set, it is not modified. + ( + { + "headers": {"Authorization": "a"}, + "settings": {"HTTPAUTH_USER": "cu", "HTTPAUTH_DOMAIN": None}, + }, + b"a", + ), + ( + { + "headers": {"Authorization": "a"}, + "spider_attributes": {"http_user": "su", "http_auth_domain": None}, + }, + b"a", + ), + ({"headers": {"Authorization": "a"}, "meta": {"http_user": "mu"}}, b"a"), + # If a non-HTTP request is received, nothing is done. + ( + { + "url": "ftp://example.com", + "spider_attributes": {"http_user": "su", "http_auth_domain": None}, + }, + None, + ), + ({"url": "s3://example.com", "meta": {"http_user": "mu"}}, None), + ], +) +def test_main(config: dict[str, Any], expected: bytes | type[Exception] | None) -> None: + url = config.get("url", "https://a.example") + headers = config.get("headers", {}) + meta = config.get("meta", {}) + settings = config.get("settings", {}) + spider_attributes = config.get("spider_attributes", {}) - def test_meta_overrides_middleware(self): - mw = make_mw(user="mw_user", passwd="mw_pass", domain="example.com") - req = Request( - "http://example.com/", - meta={"http_user": "meta_user", "http_pass": "meta_pass"}, - ) - mw.process_request(req) - assert req.headers["Authorization"] == basic_auth_header( - "meta_user", "meta_pass" + class TestSpider(Spider): + pass + + for k, v in spider_attributes.items(): + setattr(TestSpider, k, v) + + spider = TestSpider("foo") + + if isinstance(expected, type) and issubclass(expected, Exception): + with pytest.raises(expected): + _build_mw(settings, spider) + return + + mw = _build_mw(settings, spider) + request = Request(url, headers=headers, meta=meta) + assert mw.process_request(request) is None + if expected is None: + assert "Authorization" not in request.headers + else: + assert request.headers["Authorization"] == expected, repr( + request.headers["Authorization"] ) - def test_meta_already_set(self): - mw = make_mw() - req = Request( - "http://example.com/", - headers={"Authorization": "Digest 123"}, - meta={"http_user": "u", "http_pass": "p"}, - ) - mw.process_request(req) - assert req.headers["Authorization"] == b"Digest 123" + +@pytest.mark.parametrize( + ("meta", "url", "output_value"), + [ + ({}, "https://example.com/a", None), + ({"http_user": "a", "auth_origin": "foo"}, "https://example.com/a", "foo"), + ({"http_user": "a"}, "https://example.com/a", "https://example.com"), + ({"http_user": "a"}, "http://example.com/a", "http://example.com"), + ({"http_user": "a"}, "https://example.com:443/a", "https://example.com"), + ({"http_user": "a"}, "http://example.com:80/a", "http://example.com"), + ({"http_user": "a"}, "https://example.com:80/a", "https://example.com:80"), + ({"http_user": "a"}, "http://example.com:443/a", "http://example.com:443"), + ({"http_user": "a"}, "https://example.com:1234/a", "https://example.com:1234"), + ({"http_user": "a"}, "http://example.com:1234/a", "http://example.com:1234"), + # No origin is tracked when the allowed domain is explicit. + ( + {"http_user": "a", "http_auth_domain": "example.com"}, + "https://example.com/a", + None, + ), + ], +) +def test_origin_setdefault( + meta: dict[str, Any], url: str, output_value: str | None +) -> None: + """When request.meta is used for authorization, an auth_origin meta key is + defined on the request if not defined already.""" + mw = build_from_crawler(HttpAuthMiddleware, get_crawler()) + request = Request(url, meta=meta) + assert mw.process_request(request) is None + if output_value is None: + assert "auth_origin" not in request.meta + else: + assert request.meta["auth_origin"] == output_value diff --git a/tests/test_downloadermiddleware_uriuserinfo.py b/tests/test_downloadermiddleware_uriuserinfo.py new file mode 100644 index 000000000..24c3b950b --- /dev/null +++ b/tests/test_downloadermiddleware_uriuserinfo.py @@ -0,0 +1,73 @@ +from __future__ import annotations + +import pytest + +from scrapy.downloadermiddlewares.uriuserinfo import UriUserInfoMiddleware +from scrapy.http import Request +from scrapy.utils.misc import build_from_crawler +from scrapy.utils.test import get_crawler + + +def _meta_fields(protocol: str) -> tuple[str, str]: + if protocol == "ftp": + return "ftp_user", "ftp_password" + return "http_user", "http_pass" + + +@pytest.mark.parametrize("protocol", ["ftp", "http", "https"]) +@pytest.mark.parametrize( + ("userinfo", "user", "password"), + [ + ("foo:bar@", "foo", "bar"), + ("foo:@", "foo", ""), + # No password in the URL means no password meta key. + ("foo@", "foo", None), + (":bar@", "", "bar"), + # Percent-encoded delimiters are unquoted. + ("foo%3A:b%40r@", "foo:", "b@r"), + ], +) +def test_userinfo( + protocol: str, userinfo: str, user: str, password: str | None +) -> None: + user_field, password_field = _meta_fields(protocol) + mw = build_from_crawler(UriUserInfoMiddleware, get_crawler()) + request = Request(f"{protocol}://{userinfo}example.com/") + processed_request = mw.process_request(request) + assert isinstance(processed_request, Request) + assert processed_request.url == f"{protocol}://example.com/" + assert request.meta[user_field] == user + if password is None: + assert password_field not in request.meta + else: + assert request.meta[password_field] == password + + +@pytest.mark.parametrize("protocol", ["ftp", "http", "https"]) +def test_no_userinfo(protocol: str) -> None: + mw = build_from_crawler(UriUserInfoMiddleware, get_crawler()) + request = Request(f"{protocol}://example.com/") + assert mw.process_request(request) is None + assert not request.meta + + +@pytest.mark.parametrize("protocol", ["ftp", "http", "https"]) +def test_meta_takes_precedence(protocol: str) -> None: + user_field, password_field = _meta_fields(protocol) + mw = build_from_crawler(UriUserInfoMiddleware, get_crawler()) + request = Request( + f"{protocol}://foo:bar@example.com/", + meta={user_field: "baz", password_field: "qux"}, + ) + processed_request = mw.process_request(request) + assert isinstance(processed_request, Request) + assert processed_request.url == f"{protocol}://example.com/" + assert request.meta[user_field] == "baz" + assert request.meta[password_field] == "qux" + + +def test_unhandled_protocol() -> None: + mw = build_from_crawler(UriUserInfoMiddleware, get_crawler()) + request = Request("s3://foo:bar@example.com/") + assert mw.process_request(request) is None + assert not request.meta