Add specific exceptions for authentication issues

Author: Pierre-Louis Peeters

aiohttp/__init__.py

@@ -27,7 +27,10 @@
     ConnectionTimeoutError,
     ContentTypeError,
     Fingerprint,
+    InvalidAuthClientError,
+    InvalidRedirectUrlAuthClientError,
     InvalidURL,
+    InvalidUrlAuthClientError,
     InvalidUrlClientError,
     InvalidUrlRedirectClientError,
     NamedPipeConnector,
@@ -137,7 +140,10 @@
     "ConnectionTimeoutError",
     "ContentTypeError",
     "Fingerprint",
+    "InvalidAuthClientError",
+    "InvalidRedirectUrlAuthClientError",
     "InvalidURL",
+    "InvalidUrlAuthClientError",
     "InvalidUrlClientError",
     "InvalidUrlRedirectClientError",
     "NonHttpUrlClientError",

aiohttp/client.py

@@ -57,7 +57,10 @@
     ClientSSLError,
     ConnectionTimeoutError,
     ContentTypeError,
+    InvalidAuthClientError,
+    InvalidRedirectUrlAuthClientError,
     InvalidURL,
+    InvalidUrlAuthClientError,
     InvalidUrlClientError,
     InvalidUrlRedirectClientError,
     NonHttpUrlClientError,
@@ -124,7 +127,10 @@
     "ClientSSLError",
     "ConnectionTimeoutError",
     "ContentTypeError",
+    "InvalidAuthClientError",
+    "InvalidRedirectUrlAuthClientError",
     "InvalidURL",
+    "InvalidUrlAuthClientError",
     "InvalidUrlClientError",
     "RedirectClientError",
     "NonHttpUrlClientError",
@@ -576,7 +582,19 @@ async def _request(
 
                     # Override the auth with the one from the URL only if we
                     # have no auth, or if we got an auth from a redirect URL
-                    if auth is None or (history and auth_from_url is not None):
+                    if (auth is None or history) and auth_from_url is not None:
+                        # Pre-check the credentials can be encoded to
+                        # avoid crashing down the line
+                        try:
+                            auth_from_url.encode()
+                        except UnicodeEncodeError as e:
+                            auth_err_exc_cls = (
+                                InvalidRedirectUrlAuthClientError
+                                if redirects
+                                else InvalidUrlAuthClientError
+                            )
+                            raise auth_err_exc_cls(url, str(e))
+
                         auth = auth_from_url
 
                     if (

aiohttp/client_exceptions.py

@@ -48,9 +48,12 @@
     "ContentTypeError",
     "ClientPayloadError",
     "InvalidURL",
+    "InvalidAuthClientError",
+    "InvalidUrlAuthClientError",
     "InvalidUrlClientError",
     "RedirectClientError",
     "NonHttpUrlClientError",
+    "InvalidRedirectUrlAuthClientError",
     "InvalidUrlRedirectClientError",
     "NonHttpUrlRedirectClientError",
     "WSMessageTypeError",
@@ -306,6 +309,14 @@ class InvalidUrlClientError(InvalidURL):
     """Invalid URL client error."""
 
 
+class InvalidAuthClientError(ClientError):
+    """Invalid auth client error."""
+
+
+class InvalidUrlAuthClientError(InvalidURL):
+    """Invalid URL auth client error."""
+
+
 class RedirectClientError(ClientError):
     """Client redirect error."""
 
@@ -318,6 +329,10 @@ class InvalidUrlRedirectClientError(InvalidUrlClientError, RedirectClientError):
     """Invalid URL redirect client error."""
 
 
+class InvalidRedirectUrlAuthClientError(InvalidUrlRedirectClientError):
+    """Invalid redirect URL auth client error."""
+
+
 class NonHttpUrlRedirectClientError(NonHttpUrlClientError, RedirectClientError):
     """Non http URL redirect client error."""
 

aiohttp/client_reqrep.py

@@ -35,7 +35,9 @@
     ClientOSError,
     ClientResponseError,
     ContentTypeError,
+    InvalidAuthClientError,
     InvalidURL,
+    InvalidUrlAuthClientError,
     ServerFingerprintMismatch,
 )
 from .compression_utils import HAS_BROTLI
@@ -497,7 +499,11 @@ def update_transfer_encoding(self) -> None:
     def update_auth(self, auth: Optional[BasicAuth], trust_env: bool = False) -> None:
         """Set basic auth."""
         if auth is None:
+            auth_from_url = True
             auth = self.auth
+        else:
+            auth_from_url = False
+
         if auth is None and trust_env and self.url.host is not None:
             netrc_obj = netrc_from_env()
             with contextlib.suppress(LookupError):
@@ -508,7 +514,13 @@ def update_auth(self, auth: Optional[BasicAuth], trust_env: bool = False) -> Non
         if not isinstance(auth, helpers.BasicAuth):
             raise TypeError("BasicAuth() tuple is required instead")
 
-        self.headers[hdrs.AUTHORIZATION] = auth.encode()
+        try:
+            self.headers[hdrs.AUTHORIZATION] = auth.encode()
+        except UnicodeEncodeError as e:
+            auth_err_exc_cls = (
+                InvalidUrlAuthClientError if auth_from_url else InvalidAuthClientError
+            )
+            raise auth_err_exc_cls(self.url, str(e))
 
     def update_body_from_data(self, body: Any) -> None:
         if body is None:

tests/test_client_functional.py

@@ -2683,6 +2683,10 @@ async def handler_redirect(request: web.Request) -> web.Response:
     ("http:/", "http:///"),
     ("http:/example.com", "http:///example.com"),
     ("http:///example.com", "http:///example.com"),
+    (
+        "http://badchar username@example.com",
+        "http://example.com - 'latin-1' codec can't encode",
+    ),
 )
 
 NON_HTTP_URL_WITH_ERROR_MESSAGE = (

tests/test_client_request.py

@@ -378,6 +378,11 @@ def test_invalid_url(make_request: _RequestMaker) -> None:
         make_request("get", "hiwpefhipowhefopw")
 
 
+def test_invalid_auth(make_request: _RequestMaker) -> None:
+    with pytest.raises(aiohttp.InvalidUrlAuthClientError):
+        make_request("get", "http://badchar username@example.com")
+
+
 def test_no_path(make_request: _RequestMaker) -> None:
     req = make_request("get", "http://python.org")
     assert "/" == req.url.path