CVE-2021-33880: Observable Discrepancy (CWE-203)
value: The aaugustin websockets library before 9.1 for Python has an Observable Timing Discrepancy on servers when HTTP Basic Authentication is enabled with basic_auth_protocol_factory(credentials=...). An attacker may be able to guess a password via a timing attack.
CWE description: The product behaves differently or sends different responses under different circumstances in a way that is observable to an unauthorized actor, which exposes security-relevant information about the state of the product, such as whether a particular operation was successful or not.
CVSS v3 base score: 5.9
Published: 2021-06-06T15:15Z
Repository: https://github.com/aaugustin/websockets
Language: Python
Files changed: docs/changelog.rst, src/websockets/legacy/auth.py, tests/legacy/test_auth.py
Commit date: 2021-05-27 13:37:45 +0200
Commit message: Use constant-time comparison for passwords.

Backport of c91b4c2a and dfecbd03.

Diff: diff --git a/docs/changelog.rst b/docs/changelog.rst
index 1064af7..f3e1acf 100644
--- a/docs/changelog.rst
+++ b/docs/changelog.rst
@@ -28,10 +28,16 @@ They may change at any time.
 9.1
 ...
 
 *In development*
 
+.. note::
+
+    **Version 9.1 fixes a security issue introduced in version 8.0.**
+
+    Version 8.0 was vulnerable to timing attacks on HTTP Basic Auth passwords.
+
 9.0.2
 .....
 
 *May 15, 2021*
 
diff --git a/src/websockets/legacy/auth.py b/src/websockets/legacy/auth.py
index e0beede..80ceff2 100644
--- a/src/websockets/legacy/auth.py
+++ b/src/websockets/legacy/auth.py
@@ -4,10 +4,11 @@
 
 """
 
 
 import functools
+import hmac
 import http
 from typing import Any, Awaitable, Callable, Iterable, Optional, Tuple, Union, cast
 
 from ..datastructures import Headers
 from ..exceptions import InvalidHeader
@@ -130,33 +131,34 @@ def basic_auth_protocol_factory(
     if (credentials is None) == (check_credentials is None):
         raise TypeError("provide either credentials or check_credentials")
 
     if credentials is not None:
         if is_credentials(credentials):
-
-            async def check_credentials(username: str, password: str) -> bool:
-                return (username, password) == credentials
-
+            credentials_list = [cast(Credentials, credentials)]
         elif isinstance(credentials, Iterable):
             credentials_list = list(credentials)
-            if all(is_credentials(item) for item in credentials_list):
-                credentials_dict = dict(credentials_list)
-
-                async def check_credentials(username: str, password: str) -> bool:
-                    return credentials_dict.get(username) == password
-
-            else:
+            if not all(is_credentials(item) for item in credentials_list):
                 raise TypeError(f"invalid credentials argument: {credentials}")
-
         else:
             raise TypeError(f"invalid credentials argument: {credentials}")
 
+        credentials_dict = dict(credentials_list)
+
+        async def check_credentials(username: str, password: str) -> bool:
+            try:
+                expected_password = credentials_dict[username]
+            except KeyError:
+                return False
+            return hmac.compare_digest(expected_password, password)
+
     if create_protocol is None:
         # Not sure why mypy cannot figure this out.
         create_protocol = cast(
             Callable[[Any], BasicAuthWebSocketServerProtocol],
             BasicAuthWebSocketServerProtocol,
         )
 
     return functools.partial(
-        create_protocol, realm=realm, check_credentials=check_credentials
+        create_protocol,
+        realm=realm,
+        check_credentials=check_credentials,
     )
diff --git a/tests/legacy/test_auth.py b/tests/legacy/test_auth.py
index bb8c6a6..3d8eb90 100644
--- a/tests/legacy/test_auth.py
+++ b/tests/legacy/test_auth.py
@@ -1,5 +1,6 @@
+import hmac
 import unittest
 import urllib.error
 
 from websockets.exceptions import InvalidStatusCode
 from websockets.headers import build_authorization_basic
@@ -74,11 +75,11 @@ class AuthClientServerTests(ClientServerTestsMixin, AsyncioTestCase):
             str(raised.exception),
             "invalid credentials argument: [('hello', 'iloveyou'), 42]",
         )
 
     async def check_credentials(username, password):
-        return password == "iloveyou"
+        return hmac.compare_digest(password, "iloveyou")
 
     create_protocol_check_credentials = basic_auth_protocol_factory(
         realm="auth-tests",
         check_credentials=check_credentials,
     )
@@ -138,11 +139,17 @@ class AuthClientServerTests(ClientServerTestsMixin, AsyncioTestCase):
             'Basic realm="auth-tests", charset="UTF-8"',
         )
         self.assertEqual(raised.exception.read().decode(), "Unsupported credentials\n")
 
     @with_server(create_protocol=create_protocol)
-    def test_basic_auth_invalid_credentials(self):
+    def test_basic_auth_invalid_username(self):
+        with self.assertRaises(InvalidStatusCode) as raised:
+            self.start_client(user_info=("goodbye", "iloveyou"))
+        self.assertEqual(raised.exception.status_code, 401)
+
+    @with_server(create_protocol=create_protocol)
+    def test_basic_auth_invalid_password(self):
         with self.assertRaises(InvalidStatusCode) as raised:
             self.start_client(user_info=("hello", "ihateyou"))
         self.assertEqual(raised.exception.status_code, 401)
 
     @with_server(create_protocol=create_protocol)

Vulnerable code: 
            async def check_credentials(username: str, password: str) -> bool:
                return (username, password) == credentials

            if all(is_credentials(item) for item in credentials_list):
                credentials_dict = dict(credentials_list)

                async def check_credentials(username: str, password: str) -> bool:
                    return credentials_dict.get(username) == password

            else:

        create_protocol, realm=realm, check_credentials=check_credentials
        return password == "iloveyou"
    def test_basic_auth_invalid_credentials(self):
Fixed code: .. note::

    **Version 9.1 fixes a security issue introduced in version 8.0.**

    Version 8.0 was vulnerable to timing attacks on HTTP Basic Auth passwords.

import hmac
            credentials_list = [cast(Credentials, credentials)]
            if not all(is_credentials(item) for item in credentials_list):
        credentials_dict = dict(credentials_list)

        async def check_credentials(username: str, password: str) -> bool:
            try:
                expected_password = credentials_dict[username]
            except KeyError:
                return False
            return hmac.compare_digest(expected_password, password)

        create_protocol,
        realm=realm,
        check_credentials=check_credentials,
import hmac
        return hmac.compare_digest(password, "iloveyou")
    def test_basic_auth_invalid_username(self):
        with self.assertRaises(InvalidStatusCode) as raised:
            self.start_client(user_info=("goodbye", "iloveyou"))
        self.assertEqual(raised.exception.status_code, 401)

    @with_server(create_protocol=create_protocol)
    def test_basic_auth_invalid_password(self):