Update for BCrypt wrap bug detected

passlib/handlers/bcrypt.py

@@ -17,10 +17,12 @@
 from importlib.util import find_spec
 from warnings import warn
 
+from packaging.version import parse
+
 import passlib.utils.handlers as uh
 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 +45,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():
@@ -102,7 +104,7 @@ class _BcryptCommon(  # type: ignore[misc]
     # PasswordHash
     # --------------------
     name = "bcrypt"
-    setting_kwds: tuple[str, ...] = ("salt", "rounds", "ident", "truncate_error")
+    setting_kwds: tuple[str, ...] = ("salt", "rounds", "ident", "truncate_error", "truncate_verify_reject")
 
     # --------------------
     # GenericHandler
@@ -147,6 +149,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 +385,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 +431,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 +459,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 +479,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 +586,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 +635,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 +662,26 @@ def _load_backend_mixin(mixin_cls, name, dryrun):
     #     assert result.startswith(eff_ident)
     #     return consteq(result, hash)
 
+    @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 +691,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,12 +733,23 @@ 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.
 
         .. versionadded:: 1.7
 
+    :param bool truncate_verify_reject:
+        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_verify_reject=False`` will maintain backward compatibility by truncating long passwords silently.
+        Setting ``truncate_verify_reject=True`` will cause :meth:`~passlib.ifc.PasswordHash.verify`
+        to raise a :exc:`~passlib.exc.PasswordTruncateError` instead.
+
+        .. versionadded:: 1.10
+
     :type relaxed: bool
     :param relaxed:
         By default, providing an invalid value for one of the other
@@ -761,7 +809,7 @@ class _wrapped_bcrypt(bcrypt):
     """
 
     setting_kwds = tuple(
-        elem for elem in bcrypt.setting_kwds if elem not in ["truncate_error"]
+        elem for elem in bcrypt.setting_kwds if elem not in ["truncate_error", "truncate_verify_reject"]
     )
     truncate_size: int | None = None
 

passlib/utils/handlers.py

@@ -469,24 +469,40 @@ class TruncateMixin(MinimalHandler):
     truncate_verify_reject = False
 
     @classmethod
-    def using(cls, truncate_error=None, **kwds):
+    def using(cls, truncate_error=None, truncate_verify_reject=None, **kwds):
         subcls = super().using(**kwds)
         if truncate_error is not None:
-            truncate_error = as_bool(truncate_error, param="truncate_error")
-            if truncate_error is not None:
-                subcls.truncate_error = truncate_error
+            subcls.truncate_error = as_bool(truncate_error, param="truncate_error")
+        if truncate_verify_reject is not None:
+            subcls.truncate_verify_reject = as_bool(truncate_verify_reject, param="truncate_verify_reject")
         return subcls
 
+    @classmethod
+    def _check_truncate_flag(cls, truncate_flag, secret):
+        """Check if secret exceeds truncate_size when truncate_flag is enabled."""
+        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 _check_truncate_policy(cls, secret):
         """
-        make sure secret won't be truncated.
-        NOTE: this should only be called for .hash(), not for .verify(),
-        which should honor the .truncate_verify_reject policy.
+        Ensure secret won't be truncated during hashing.
+
+        Uses the truncate_error policy to determine whether to raise an error
+        if the secret exceeds the maximum allowed length.
         """
-        assert cls.truncate_size is not None, "truncate_size must be set by subclass"
-        if cls.truncate_error and len(secret) > cls.truncate_size:
-            raise exc.PasswordTruncateError(cls)
+        cls._check_truncate_flag(cls.truncate_error, secret)
+
+    @classmethod
+    def _check_verify_truncate_policy(cls, secret):
+        """
+        Ensure secret won't be truncated during verification.
+
+        Uses the truncate_verify_reject policy to determine whether to raise an error
+        if the secret exceeds the maximum allowed length during verification.
+        """
+        cls._check_truncate_flag(cls.truncate_verify_reject, secret)
 
 
 class GenericHandler(MinimalHandler):

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

@@ -1997,10 +1997,11 @@ def test_truncate_error_setting(self):
         validate 'truncate_error' setting & related attributes
         """
         # If it doesn't have truncate_size set,
-        # it shouldn't support truncate_error
+        # it shouldn't support truncate_error or truncate_verify_reject
         hasher = self.handler
         if hasher.truncate_size is None:
             assert "truncate_error" not in hasher.setting_kwds
+            assert "truncate_verify_reject" not in hasher.setting_kwds
             return
 
         # if hasher defaults to silently truncating,
@@ -2015,17 +2016,26 @@ def test_truncate_error_setting(self):
             assert hasher.truncate_error
             return
 
-        # test value parsing
-        def parse_value(value):
-            return hasher.using(truncate_error=value).truncate_error
+        # helper function to test value parsing for truncate settings
+        def test_truncate_setting_parsing(setting_name, current_value):
+            """Test that a truncate setting correctly parses various input values"""
+            def parse_value(value):
+                return getattr(hasher.using(**{setting_name: value}), setting_name)
+
+            assert parse_value(None) == current_value
+            assert parse_value(True) is True
+            assert parse_value("true") is True
+            assert parse_value(False) is False
+            assert parse_value("false") is False
+            with pytest.raises(ValueError):
+                parse_value("xxx")
 
-        assert parse_value(None) == hasher.truncate_error
-        assert parse_value(True) is True
-        assert parse_value("true") is True
-        assert parse_value(False) is False
-        assert parse_value("false") is False
-        with pytest.raises(ValueError):
-            parse_value("xxx")
+        # test truncate_error value parsing
+        test_truncate_setting_parsing("truncate_error", hasher.truncate_error)
+
+        # test truncate_verify_reject value parsing if supported
+        if "truncate_verify_reject" in hasher.setting_kwds:
+            test_truncate_setting_parsing("truncate_verify_reject", hasher.truncate_verify_reject)
 
     def test_secret_wo_truncate_size(self):
         """
@@ -2075,7 +2085,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)