Update for BCrypt wrap bug detected

passlib/handlers/bcrypt.py

@@ -17,10 +17,13 @@
 from importlib.util import find_spec
 from warnings import warn
 
+from packaging.version import parse
+
 import passlib.utils.handlers as uh
+from passlib import exc
 from passlib._logging import logger
 from passlib.crypto.digest import compile_hmac
-from passlib.exc import PasslibHashWarning, PasslibSecurityError
+from passlib.exc import PasslibHashWarning, PasslibSecurityError, PasswordSizeError, PasswordTruncateError
 from passlib.utils import (
     repeat_string,
     to_unicode,
@@ -43,7 +46,7 @@
 _BNULL = b"\x00"
 
 # reference hash of "test", used in various self-checks
-TEST_HASH_2A = "$2a$04$5BJqKfqMQvV7nS.yUguNcueVirQqDBGaLXSqj.rs.pZPlNR0UX/HK"
+TEST_HASH_2A = f"{IDENT_2A}04$5BJqKfqMQvV7nS.yUguNcueVirQqDBGaLXSqj.rs.pZPlNR0UX/HK"
 
 
 def _detect_pybcrypt():
@@ -147,6 +150,7 @@ class _BcryptCommon(  # type: ignore[misc]
     # NOTE: these are only set on the backend mixin classes
     _workrounds_initialized = False
     _has_2a_wraparound_bug = False
+    _fails_on_long_secrets = False
     _lacks_20_support = False
     _lacks_2y_support = False
     _lacks_2b_support = False
@@ -382,9 +386,11 @@ def detect_wrap_bug(ident):
                 # If we get here, the backend auto-truncates, test for wraparound bug
                 if verify(secret, bug_hash):
                     return True
-            except ValueError:
+            except ValueError as err:
+                if not mixin_cls.is_secret_truncate_err(secret, err):
+                    raise
                 # Backend explicitly will not auto-truncate, truncate the password to 72 characters
-                secret = secret[:72]
+                secret = secret[:mixin_cls.truncate_size]
 
             # Check to make sure that the backend still hashes correctly; if not, we're in a failure case
             # not related to the original wraparound bug or bcrypt >= 5.0.0 input length restriction.
@@ -426,18 +432,18 @@ def assert_lacks_wrap_bug(ident):
         result = safe_verify("test", TEST_HASH_2A)
         if result is NotImplemented:
             # 2a support is required, and should always be present
-            raise RuntimeError(f"{backend} lacks support for $2a$ hashes")
+            raise RuntimeError(f"{backend} lacks support for {IDENT_2A} hashes")
         if not result:
-            raise RuntimeError(f"{backend} incorrectly rejected $2a$ hash")
+            raise RuntimeError(f"{backend} incorrectly rejected {IDENT_2A} hash")
         assert_lacks_8bit_bug(IDENT_2A)
         if detect_wrap_bug(IDENT_2A):
             if backend == "os_crypt":
                 # don't make this a warning for os crypt (e.g. openbsd);
                 # they'll have proper 2b implementation which will be used for new hashes.
                 # so even if we didn't have a workaround, this bug wouldn't be a concern.
                 logger.debug(
-                    "%r backend has $2a$ bsd wraparound bug, enabling workaround",
-                    backend,
+                    "%r backend has %s bsd wraparound bug, enabling workaround",
+                    backend, IDENT_2A
                 )
             else:
                 # installed library has the bug -- want to let users know,
@@ -454,13 +460,13 @@ def assert_lacks_wrap_bug(ident):
         # ----------------------------------------------------------------
         # check for 2y support
         # ----------------------------------------------------------------
-        test_hash_2y = TEST_HASH_2A.replace("2a", "2y")
+        test_hash_2y = TEST_HASH_2A.replace(IDENT_2A, IDENT_2Y)
         result = safe_verify("test", test_hash_2y)
         if result is NotImplemented:
             mixin_cls._lacks_2y_support = True
-            logger.debug("%r backend lacks $2y$ support, enabling workaround", backend)
+            logger.debug("%r backend lacks %s support, enabling workaround", backend, IDENT_2Y)
         elif not result:
-            raise RuntimeError(f"{backend} incorrectly rejected $2y$ hash")
+            raise RuntimeError(f"{backend} incorrectly rejected {IDENT_2Y} hash")
         else:
             # NOTE: Not using this as fallback candidate,
             #       lacks wide enough support across implementations.
@@ -474,13 +480,13 @@ def assert_lacks_wrap_bug(ident):
         # ----------------------------------------------------------------
         # check for 2b support
         # ----------------------------------------------------------------
-        test_hash_2b = TEST_HASH_2A.replace("2a", "2b")
+        test_hash_2b = TEST_HASH_2A.replace(IDENT_2A, IDENT_2B)
         result = safe_verify("test", test_hash_2b)
         if result is NotImplemented:
             mixin_cls._lacks_2b_support = True
-            logger.debug("%r backend lacks $2b$ support, enabling workaround", backend)
+            logger.debug("%r backend lacks %s support, enabling workaround", backend, IDENT_2B)
         elif not result:
-            raise RuntimeError(f"{backend} incorrectly rejected $2b$ hash")
+            raise RuntimeError(f"{backend} incorrectly rejected {IDENT_2B} hash")
         else:
             mixin_cls._fallback_ident = IDENT_2B
             assert_lacks_8bit_bug(IDENT_2B)
@@ -581,13 +587,23 @@ def _norm_digest_args(cls, secret, ident, new=False):
         elif ident == IDENT_2X:
             # NOTE: shouldn't get here.
             # XXX: could check if backend does actually offer 'support'
-            raise RuntimeError("$2x$ hashes not currently supported by passlib")
+            raise RuntimeError(f"{IDENT_2X} hashes not currently supported by passlib")
 
         else:
             raise AssertionError(f"unexpected ident value: {ident!r}")
 
         return secret, ident
 
+    @classmethod
+    def is_secret_truncate_err(cls, secret, err):
+        if isinstance(err, PasswordTruncateError):
+            return True
+        if isinstance(err, PasswordSizeError):
+            return False
+        return (cls._fails_on_long_secrets
+                and "password" in str(err).lower()
+                and len(secret) > cls.truncate_size)
+
 
 class _NoBackend(_BcryptCommon):
     """
@@ -620,6 +636,8 @@ def _load_backend_mixin(mixin_cls, name, dryrun):
             return False
         try:
             version = metadata.version("bcrypt")
+            # From bcrypt >= 5.0.0 is expected a failure on secrets greater than 72 characters
+            mixin_cls._fails_on_long_secrets = parse(version) >= parse("5.0.0")
         except Exception:
             logger.warning("(trapped) error reading bcrypt version", exc_info=True)
             version = "<unknown>"
@@ -645,6 +663,32 @@ def _load_backend_mixin(mixin_cls, name, dryrun):
     #     assert result.startswith(eff_ident)
     #     return consteq(result, hash)
 
+    @classmethod
+    def _check_truncate_flag(cls, truncate_flag, secret):
+        assert cls.truncate_size is not None, "truncate_size must be set by subclass"
+        if truncate_flag and len(secret) > cls.truncate_size:
+            raise exc.PasswordTruncateError(cls)
+
+    @classmethod
+    def _handle_w_truncate(cls, func, truncate_flag, secret, *args, **kwargs):
+        """
+        Helper method to handle ValueError exceptions for passwords > 72 bytes.
+        Truncates password if needed and retries the operation.
+        """
+        try:
+            return func(secret, *args, **kwargs)
+        except ValueError as err:
+            # bcrypt >= 5.0.0 will raise ValueError on passwords > 72 bytes
+            if not cls.is_secret_truncate_err(secret, err):
+                raise
+            cls._check_truncate_flag(truncate_flag, secret)
+            # silently truncate password to truncate_size bytes, and try again
+            return func(secret[:cls.truncate_size], *args, **kwargs)
+
+    @classmethod
+    def verify(cls, secret, hash, **context):
+        return cls._handle_w_truncate(super().verify, cls.truncate_verify_reject, secret, hash, **context)
+
     def _calc_checksum(self, secret):
         # bcrypt behavior:
         #   secret must be bytes
@@ -654,7 +698,7 @@ def _calc_checksum(self, secret):
         config = self._get_config(ident)
         if isinstance(config, str):
             config = config.encode("ascii")
-        hash = _bcrypt.hashpw(secret, config)
+        hash = self._handle_w_truncate(_bcrypt.hashpw, self.truncate_error, secret, config)
         assert isinstance(hash, bytes)
         if not hash.startswith(config) or len(hash) != len(config) + 31:
             raise uh.exc.CryptBackendError(
@@ -696,7 +740,9 @@ class bcrypt(_NoBackend, _BcryptCommon):  # type: ignore[misc]
         * ``"2b"`` - latest revision of the official BCrypt algorithm, current default.
 
     :param bool truncate_error:
-        By default, BCrypt will silently truncate passwords larger than 72 bytes.
+        By default, BCrypt will silently truncate passwords larger than 72 bytes (in bcrypt < 5.0.0)
+        or raise a ValueError (in bcrypt >= 5.0.0).
+        Setting ``truncate_error=False`` will maintain backward compatibility by truncating long passwords silently.
         Setting ``truncate_error=True`` will cause :meth:`~passlib.ifc.PasswordHash.hash`
         to raise a :exc:`~passlib.exc.PasswordTruncateError` instead.
 

tests/test_handlers_bcrypt.py

@@ -220,7 +220,7 @@ def check_bcrypt(secret, hash):
                 hash = IDENT_2B + hash[4:]
             hash = to_bytes(hash)
             try:
-                return bcrypt.hashpw(secret, hash) == hash
+                return self.handler.verify(secret, hash.decode('ascii'))
             except ValueError:
                 raise ValueError(f"bcrypt rejected hash: {hash!r} (secret={secret!r})")
 

tests/utils.py

@@ -2075,7 +2075,7 @@ def test_secret_w_truncate_size(self):
         # setup vars
         # --------------------------------------------------
         # try to get versions w/ and w/o truncate_error set.
-        # set to None if policy isn't configruable
+        # set to None if policy isn't configurable
         size_error_type = exc.PasswordSizeError
         if "truncate_error" in handler.setting_kwds:
             without_error = handler.using(truncate_error=False)