Support arbitrary data in basic auth credentials.

Author: aaugustin

src/websockets/headers.py

@@ -530,6 +530,8 @@ def parse_authorization_basic(header: str) -> tuple[str, str]:
 
     Return a ``(username, password)`` tuple.
 
+    Non-UTF-8 data is represented with surrogate code points.
+
     Args:
         header: Value of the ``Authorization`` header.
 
@@ -555,12 +557,13 @@ def parse_authorization_basic(header: str) -> tuple[str, str]:
     parse_end(header, pos, "Authorization")
 
     try:
-        user_pass = base64.b64decode(basic_credentials.encode()).decode()
+        user_pass_bytes = base64.b64decode(basic_credentials.encode())
     except binascii.Error:
         raise InvalidHeaderValue(
             "Authorization",
             "expected base64-encoded credentials",
         ) from None
+    user_pass = user_pass_bytes.decode(errors="surrogateescape")
     try:
         username, password = user_pass.split(":", 1)
     except ValueError:
@@ -578,9 +581,12 @@ def build_authorization_basic(username: str, password: str) -> str:
 
     This is the reverse of :func:`parse_authorization_basic`.
 
+    Non-UTF-8 data can be represented with surrogate code points.
+
     """
     # https://datatracker.ietf.org/doc/html/rfc7617#section-2
     assert ":" not in username
     user_pass = f"{username}:{password}"
-    basic_credentials = base64.b64encode(user_pass.encode()).decode()
+    user_pass_bytes = user_pass.encode(errors="surrogateescape")
+    basic_credentials = base64.b64encode(user_pass_bytes).decode()
     return "Basic " + basic_credentials

tests/asyncio/test_server.py

@@ -852,6 +852,19 @@ async def test_valid_authorization_non_ascii_password(self):
             ) as client:
                 await self.assertEval(client, "ws.username", "test")
 
+    async def test_valid_authorization_non_utf8_password(self):
+        """basic_auth authenticates client with non-UTF-8 password."""
+        password = b"\xca\xfe".decode(errors="surrogateescape")
+        async with serve(
+            *args,
+            process_request=basic_auth(credentials=("drink", password)),
+        ) as server:
+            async with connect(
+                get_uri(server),
+                additional_headers={"Authorization": "Basic ZHJpbms6yv4="},
+            ) as client:
+                await self.assertEval(client, "ws.username", "drink")
+
     async def test_missing_authorization(self):
         """basic_auth rejects client without credentials."""
         async with serve(

tests/sync/test_server.py

@@ -635,6 +635,18 @@ def test_valid_authorization_non_ascii_password(self):
             ) as client:
                 self.assertEval(client, "ws.username", "test")
 
+    def test_valid_authorization_non_utf8_password(self):
+        """basic_auth authenticates client with non-UTF-8 password."""
+        password = b"\xca\xfe".decode(errors="surrogateescape")
+        with run_server(
+            process_request=basic_auth(credentials=("drink", password)),
+        ) as server:
+            with connect(
+                get_uri(server),
+                additional_headers={"Authorization": "Basic ZHJpbms6yv4="},
+            ) as client:
+                self.assertEval(client, "ws.username", "drink")
+
     def test_missing_authorization(self):
         """basic_auth rejects client without credentials."""
         with run_server(