Merge "Add transparent PBKDF2-SHA512 password rehashing on login"

doc/source/admin/configuration.rst

@@ -13,6 +13,7 @@ complete keystone configuration documentation and sample config files.
 .. include:: integrate-with-ldap.inc
 .. include:: caching-layer.inc
 .. include:: security-compliance.inc
+.. include:: password-hashing.inc
 .. include:: performance.inc
 .. include:: url-safe-naming.inc
 .. include:: limit-list-size.inc

doc/source/admin/password-hashing.inc

@@ -0,0 +1,63 @@
+.. -*- rst -*-
+
+..
+      Licensed under the Apache License, Version 2.0 (the "License"); you may
+      not use this file except in compliance with the License. You may obtain
+      a copy of the License at
+
+      http://www.apache.org/licenses/LICENSE-2.0
+
+      Unless required by applicable law or agreed to in writing, software
+      distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+      WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+      License for the specific language governing permissions and limitations
+      under the License.
+
+.. _password_hashing:
+
+Password hashing
+================
+
+Keystone stores local user passwords as hashes in the SQL identity backend.
+The algorithm is selected with ``[identity] password_hash_algorithm``. Supported
+values are ``bcrypt`` (the default), ``bcrypt_sha256``, ``scrypt``, and
+``pbkdf2_sha512``. Work-factor options such as
+``[identity] password_hash_rounds``, ``[identity] scrypt_block_size``, and
+``[identity] scrypt_parallelism`` apply to newly created hashes.
+
+Transparent rehash on authentication
+------------------------------------
+
+When a user authenticates successfully against the SQL identity backend,
+Keystone upgrades the stored hash in place if either of the following is true:
+
+* The stored hash uses a different algorithm than
+  ``[identity] password_hash_algorithm``.
+* The stored hash uses a weaker work factor than the currently configured
+  value for that algorithm (for example fewer bcrypt rounds, a lower scrypt
+  ``n``/``r``/``p``, or fewer PBKDF2 iterations).
+
+The rehash updates only the hash string. Password metadata (``created_at``,
+``expires_at``, and password history) is left unchanged. The write is skipped
+if another session changed the stored hash after the authentication read, so
+an administrative password reset cannot be overwritten.
+
+This applies to every supported hasher, not only PBKDF2-SHA512. Operators can
+raise ``password_hash_rounds`` or switch ``password_hash_algorithm`` without
+forcing users to reset their passwords. Existing hashes are upgraded on the
+next successful login. Lowering the work factor does not downgrade hashes that
+already use a stronger value.
+
+LDAP and other non-SQL identity backends are not affected; they keep their own
+password storage.
+
+Example
+-------
+
+Raise the bcrypt cost and let existing users pick up the new value on login:
+
+.. code-block:: ini
+
+   [identity]
+   password_hash_algorithm = bcrypt
+   password_hash_rounds = 12

keystone/common/password_hashers/__init__.py

@@ -43,3 +43,13 @@ class PasswordHasher(abc.ABC):
 
     name: str
     ident: ty.Optional[str]
+
+    @staticmethod
+    def needs_rehash(hashed: str, **params) -> bool:
+        """Return True if the stored hash should be upgraded.
+
+        Subclasses compare ``hashed`` against the current work-factor
+        parameters. The default returns False: only an algorithm mismatch
+        (handled by the caller) would trigger a rehash.
+        """
+        return False

keystone/common/password_hashers/bcrypt.py

@@ -40,6 +40,22 @@ class Bcrypt(password_hashers.PasswordHasher):
         digest: bytes = bcrypt.hashpw(password, salt)
         return digest.decode("ascii")
 
+    @staticmethod
+    def needs_rehash(hashed: str, rounds: int = 12, **kwargs) -> bool:
+        """Return True if hashed used a lower cost than ``rounds``.
+
+        :param str hashed: Stored password hash (modular crypt format).
+        :param int rounds: Target bcrypt cost (defaults to 12).
+        :returns: True when the hash should be upgraded on next login.
+        """
+        parts = hashed.split("$")
+        if len(parts) >= 4:
+            try:
+                return int(parts[2]) < rounds
+            except ValueError:
+                pass
+        return True
+
     @staticmethod
     def verify(password: bytes, hashed: str) -> bool:
         """Verify hashing password would be equal to the `hashed` value
@@ -97,6 +113,27 @@ class Bcrypt_sha256(password_hashers.PasswordHasher):
         # Construct `passlib` compatible format of the bcrypt-sha256 hash
         return f"{Bcrypt_sha256.prefix}v=2,t=2b,r={rounds}${salt_str}${digest}"
 
+    @staticmethod
+    def needs_rehash(hashed: str, rounds: int = 12, **kwargs) -> bool:
+        """Return True if hashed used a lower cost than ``rounds``.
+
+        :param str hashed: Stored password hash.
+        :param int rounds: Target bcrypt cost (defaults to 12).
+        :returns: True when the hash should be upgraded on next login.
+        """
+        data = hashed
+        if hashed.startswith(Bcrypt_sha256.prefix):
+            data = hashed[len(Bcrypt_sha256.prefix) :]
+        parts = data.split("$")
+        if parts:
+            for param in parts[0].split(","):
+                if param.startswith("r="):
+                    try:
+                        return int(param[2:]) < rounds
+                    except ValueError:
+                        return True
+        return True
+
     @staticmethod
     def verify(password: bytes, hashed: str) -> bool:
         """Verify hashing password would be equal to the `hashed` value

keystone/common/password_hashers/pbkdf2.py

@@ -20,6 +20,8 @@ from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
 from keystone.common import password_hashers
 from keystone import exception
 
+DEFAULT_ROUNDS: int = 25000
+
 
 class Sha512(password_hashers.PasswordHasher):
     """passlib transition class for PBKDF2 Sha512 password hashing"""
@@ -29,21 +31,22 @@ class Sha512(password_hashers.PasswordHasher):
     hash_algo = hashes.SHA512()
 
     @staticmethod
-    def hash(password: bytes, salt_size: int = 16, rounds: int = 25000) -> str:
-        """Generate password hash string with ident and params
+    def hash(
+        password: bytes, salt_size: int = 16, rounds: int = DEFAULT_ROUNDS
+    ) -> str:
+        """Generate password hash string with ident and params.
 
         https://cryptography.io/en/stable/hazmat/primitives/key-derivation-functions/#pbkdf2
 
         :param bytes password: Password to be hashed.
-        :param bytes salt: Salt.
-        :param int iterations: Iterations count
+        :param int salt_size: Number of random bytes to use as salt.
+        :param int rounds: PBKDF2 iteration count.
 
-        :returns: String in format
-            `$pbkdf2-sha512$ln=logN,r=R,p=P$salt$checksum`
+        :returns: String in format ``$pbkdf2-sha512$<rounds>$<salt>$<digest>``
         """
         salt: bytes = os.urandom(salt_size)
 
-        # Prepave the kdf function with params
+        # Prepare the kdf function with params
         kdf = PBKDF2HMAC(
             algorithm=Sha512.hash_algo, length=64, salt=salt, iterations=rounds
         )
@@ -60,7 +63,27 @@ class Sha512(password_hashers.PasswordHasher):
             binascii.b2a_base64(salt).rstrip(b"=\n").decode("ascii")
         )
 
-        return f"$pbkdf2-sha512${rounds}${salt_str}${digest_str}"
+        return f"{Sha512.ident}{rounds}${salt_str}${digest_str}"
+
+    @staticmethod
+    def needs_rehash(
+        hashed: str, rounds: int = DEFAULT_ROUNDS, **kwargs
+    ) -> bool:
+        """Return True if hashed used fewer rounds than ``rounds``.
+
+        :param str hashed: Stored password hash.
+        :param int rounds: Target iteration count (defaults to
+            DEFAULT_ROUNDS; callers should pass the operator-configured value).
+        :returns: True when the hash should be upgraded on next login.
+        """
+        parts = hashed[1:].split('$')
+        if len(parts) == 4:
+            _, rounds_str, _, _ = parts
+            try:
+                return int(rounds_str) < rounds
+            except ValueError:
+                pass
+        return True
 
     @staticmethod
     def verify(password: bytes, hashed: str) -> bool:
@@ -92,7 +115,7 @@ class Sha512(password_hashers.PasswordHasher):
         else:
             raise exception.PasswordValidationError("malformed password hash")
 
-        # Prepave the kdf function with params
+        # Prepare the kdf function with params
         kdf = PBKDF2HMAC(
             algorithm=Sha512.hash_algo, length=64, salt=salt, iterations=rounds
         )

keystone/common/password_hashers/scrypt.py

@@ -62,6 +62,36 @@ class Scrypt(password_hashers.PasswordHasher):
         )
         return f"$scrypt$ln={n},r={r},p={p}${salt_str}${digest_str}"
 
+    @staticmethod
+    def needs_rehash(
+        hashed: str, n: int = 16, r: int = 8, p: int = 1, **kwargs
+    ) -> bool:
+        """Return True if hashed used a weaker scrypt work factor.
+
+        :param str hashed: Stored password hash.
+        :param int n: Target logN cost factor (defaults to 16).
+        :param int r: Target block size (defaults to 8).
+        :param int p: Target parallelism (defaults to 1).
+        :returns: True when the hash should be upgraded on next login.
+        """
+        parts = hashed[1:].split("$")
+        if len(parts) != 4:
+            return True
+        stored_n = stored_r = stored_p = None
+        try:
+            for param in parts[1].split(","):
+                if param.startswith("ln="):
+                    stored_n = int(param[3:])
+                elif param.startswith("r="):
+                    stored_r = int(param[2:])
+                elif param.startswith("p="):
+                    stored_p = int(param[2:])
+        except ValueError:
+            return True
+        if stored_n is None or stored_r is None or stored_p is None:
+            return True
+        return stored_n < n or stored_r < r or stored_p < p
+
     @staticmethod
     def verify(password: bytes, hashed: str) -> bool:
         """Verify hashing password would be equal to the `hashed` value

keystone/common/password_hashing.py

@@ -120,6 +120,51 @@ def verify_length_and_trunc_password(password) -> bytes:
         raise exception.ValidationError(attribute='string', target='password')
 
 
+def _hasher_work_params(hasher) -> dict:
+    """Return the work-factor kwargs currently configured for hasher."""
+    if hasher is scrypt.Scrypt:
+        return {
+            "n": CONF.identity.password_hash_rounds or 16,
+            "r": CONF.identity.scrypt_block_size or 8,
+            "p": CONF.identity.scrypt_parallelism or 1,
+        }
+    if hasher is pbkdf2.Sha512:
+        return {
+            "rounds": (
+                CONF.identity.password_hash_rounds or pbkdf2.DEFAULT_ROUNDS
+            )
+        }
+    if hasher in (bcrypt.Bcrypt, bcrypt.Bcrypt_sha256):
+        return {"rounds": CONF.identity.password_hash_rounds or 12}
+    return {}
+
+
+def needs_rehash(hashed: str) -> bool:
+    """Return True if the stored password hash should be upgraded.
+
+    A rehash is needed when the stored hash uses a different algorithm
+    than the currently configured one, or when the stored hash was
+    computed with a weaker work factor than the currently configured
+    (or default) value. This applies to bcrypt, bcrypt_sha256, scrypt,
+    and PBKDF2-SHA512.
+
+    Callers should rehash transparently on the next successful
+    authentication.
+    """
+    if not hashed:
+        return False
+    try:
+        hasher = _get_hasher_from_ident(hashed)
+    except ValueError:
+        return True
+
+    conf_hasher = _HASHER_NAME_MAP.get(CONF.identity.password_hash_algorithm)
+    if hasher is not conf_hasher:
+        return True
+
+    return hasher.needs_rehash(hashed, **_hasher_work_params(hasher))
+
+
 def check_password(password: str, hashed: str) -> bool:
     """Check that a plaintext password matches hashed.
 
@@ -159,7 +204,8 @@ def hash_password(password: str) -> str:
     if CONF.identity.password_hash_rounds:
         params['rounds'] = CONF.identity.password_hash_rounds
     if hasher is scrypt.Scrypt:
-        params["n"] = 16
+        # password_hash_rounds is logN for scrypt (conf default 16).
+        params["n"] = params.pop("rounds", 16)
         if CONF.identity.scrypt_block_size:
             params["r"] = CONF.identity.scrypt_block_size
         if CONF.identity.scrypt_parallelism:

keystone/conf/identity.py

@@ -144,6 +144,8 @@ password_hash_algorithm = cfg.StrOpt(
     help=utils.fmt(
         """
 The password hashing algorithm to use for passwords stored within keystone.
+Stored hashes that use a different algorithm are upgraded transparently on
+the next successful authentication (SQL identity backend).
 """
     ),
 )
@@ -153,10 +155,10 @@ password_hash_rounds = cfg.IntOpt(
     help=utils.fmt(
         """
 This option represents a trade off between security and performance. Higher
-values lead to slower performance, but higher security. Changing this option
-will only affect newly created passwords as existing password hashes already
-have a fixed number of rounds applied, so it is safe to tune this option in a
-running cluster.
+values lead to slower performance, but higher security. Existing password
+hashes that use a weaker work factor are upgraded transparently on the next
+successful authentication (SQL identity backend), so it is safe to raise this
+option in a running cluster. Lowering it does not downgrade existing hashes.
 
 The default for bcrypt is 12, must be between 4 and 31, inclusive.
 

keystone/identity/backends/sql.py

@@ -79,8 +79,34 @@ class Identity(base.IdentityDriverBase):
         # successful auth, reset failed count if present
         if user_ref.local_user.failed_auth_count:
             self._reset_failed_auth(user_id)
+        # Transparently upgrade hashes that use outdated parameters so
+        # that existing passwords become more secure over time without
+        # requiring a forced password reset.
+        if password_hashing.needs_rehash(user_ref.password):
+            self._rehash_password(user_id, password, user_ref.password)
         return user_dict
 
+    def _rehash_password(self, user_id, password, old_hash):
+        """Rehash the stored password using the current algorithm and params.
+
+        Updates the password_hash field of the current Password record in
+        place, deliberately bypassing the User.password setter so that
+        password-history, expiry, and created_at metadata are left untouched.
+
+        Uses compare-and-swap: the write is skipped if the stored hash
+        no longer matches old_hash, which means another session (e.g. an
+        admin password reset) changed it after the authentication read.
+        """
+        new_hash = password_hashing.hash_password(password)
+        with sql.session_for_write() as session:
+            user_ref = session.get(model.User, user_id)
+            if (
+                user_ref is not None
+                and user_ref.password_ref is not None
+                and user_ref.password_ref.password_hash == old_hash
+            ):
+                user_ref.password_ref.password_hash = new_hash
+
     def _is_account_locked(self, user_id, user_ref):
         """Check if the user account is locked.
 

keystone/tests/unit/common/test_password_hashing.py

@@ -16,6 +16,9 @@ import string
 
 from oslo_config import fixture as config_fixture
 
+from keystone.common.password_hashers import bcrypt
+from keystone.common.password_hashers import pbkdf2
+from keystone.common.password_hashers import scrypt
 from keystone.common import password_hashing
 import keystone.conf
 from keystone.tests import unit
@@ -91,6 +94,124 @@ class TestPasswordHashing(unit.BaseTestCase):
             hashed = password_hashing.hash_password(password)
             self.assertTrue(password_hashing.check_password(password, hashed))
 
+    def test_pbkdf2_sha512_needs_rehash_low_iterations(self):
+        old_hash = pbkdf2.Sha512.hash(b"password", rounds=10000)
+        self.assertTrue(pbkdf2.Sha512.needs_rehash(old_hash))
+
+    def test_pbkdf2_sha512_needs_rehash_current_iterations(self):
+        current_hash = pbkdf2.Sha512.hash(
+            b"password", rounds=pbkdf2.DEFAULT_ROUNDS
+        )
+        self.assertFalse(pbkdf2.Sha512.needs_rehash(current_hash))
+
+    def test_needs_rehash_pbkdf2_low_iterations(self):
+        self.config_fixture.config(
+            group="identity", password_hash_algorithm="pbkdf2_sha512"
+        )
+        old_hash = pbkdf2.Sha512.hash(b"password", rounds=10000)
+        self.assertTrue(password_hashing.needs_rehash(old_hash))
+
+    def test_needs_rehash_pbkdf2_current_iterations(self):
+        self.config_fixture.config(
+            group="identity", password_hash_algorithm="pbkdf2_sha512"
+        )
+        current_hash = pbkdf2.Sha512.hash(
+            b"password", rounds=pbkdf2.DEFAULT_ROUNDS
+        )
+        self.assertFalse(password_hashing.needs_rehash(current_hash))
+
+    def test_needs_rehash_algorithm_mismatch(self):
+        self.config_fixture.config(group="identity", max_password_length="72")
+        self.config_fixture.config(
+            group="identity", password_hash_algorithm="bcrypt"
+        )
+        bcrypt_hash = password_hashing.hash_password("testpassword")
+        self.config_fixture.config(
+            group="identity", password_hash_algorithm="pbkdf2_sha512"
+        )
+        self.assertTrue(password_hashing.needs_rehash(bcrypt_hash))
+
+    def test_needs_rehash_none_returns_false(self):
+        self.assertFalse(password_hashing.needs_rehash(None))
+
+    def test_needs_rehash_empty_string_returns_false(self):
+        self.assertFalse(password_hashing.needs_rehash(""))
+
+    def test_bcrypt_needs_rehash_low_rounds(self):
+        old_hash = bcrypt.Bcrypt.hash(b"password", rounds=4)
+        self.assertTrue(bcrypt.Bcrypt.needs_rehash(old_hash, rounds=12))
+
+    def test_bcrypt_needs_rehash_current_rounds(self):
+        current_hash = bcrypt.Bcrypt.hash(b"password", rounds=12)
+        self.assertFalse(bcrypt.Bcrypt.needs_rehash(current_hash, rounds=12))
+
+    def test_bcrypt_needs_rehash_malformed_returns_true(self):
+        self.assertTrue(bcrypt.Bcrypt.needs_rehash("not-a-bcrypt-hash"))
+
+    def test_needs_rehash_bcrypt_low_rounds(self):
+        self.config_fixture.config(
+            group="identity", password_hash_algorithm="bcrypt"
+        )
+        self.config_fixture.config(group="identity", password_hash_rounds=12)
+        old_hash = bcrypt.Bcrypt.hash(b"password", rounds=4)
+        self.assertTrue(password_hashing.needs_rehash(old_hash))
+
+    def test_needs_rehash_bcrypt_current_rounds(self):
+        self.config_fixture.config(
+            group="identity", password_hash_algorithm="bcrypt"
+        )
+        self.config_fixture.config(group="identity", password_hash_rounds=12)
+        current_hash = bcrypt.Bcrypt.hash(b"password", rounds=12)
+        self.assertFalse(password_hashing.needs_rehash(current_hash))
+
+    def test_bcrypt_sha256_needs_rehash_low_rounds(self):
+        old_hash = bcrypt.Bcrypt_sha256.hash(b"password", rounds=4)
+        self.assertTrue(bcrypt.Bcrypt_sha256.needs_rehash(old_hash, rounds=12))
+
+    def test_bcrypt_sha256_needs_rehash_current_rounds(self):
+        current_hash = bcrypt.Bcrypt_sha256.hash(b"password", rounds=12)
+        self.assertFalse(
+            bcrypt.Bcrypt_sha256.needs_rehash(current_hash, rounds=12)
+        )
+
+    def test_needs_rehash_bcrypt_sha256_low_rounds(self):
+        self.config_fixture.config(
+            group="identity", password_hash_algorithm="bcrypt_sha256"
+        )
+        self.config_fixture.config(group="identity", password_hash_rounds=12)
+        old_hash = bcrypt.Bcrypt_sha256.hash(b"password", rounds=4)
+        self.assertTrue(password_hashing.needs_rehash(old_hash))
+
+    def test_scrypt_needs_rehash_low_cost(self):
+        old_hash = scrypt.Scrypt.hash(b"password", n=8, r=8, p=1)
+        self.assertTrue(scrypt.Scrypt.needs_rehash(old_hash, n=16, r=8, p=1))
+
+    def test_scrypt_needs_rehash_low_block_size(self):
+        old_hash = scrypt.Scrypt.hash(b"password", n=16, r=4, p=1)
+        self.assertTrue(scrypt.Scrypt.needs_rehash(old_hash, n=16, r=8, p=1))
+
+    def test_scrypt_needs_rehash_current_params(self):
+        current_hash = scrypt.Scrypt.hash(b"password", n=16, r=8, p=1)
+        self.assertFalse(
+            scrypt.Scrypt.needs_rehash(current_hash, n=16, r=8, p=1)
+        )
+
+    def test_needs_rehash_scrypt_low_cost(self):
+        self.config_fixture.config(
+            group="identity", password_hash_algorithm="scrypt"
+        )
+        self.config_fixture.config(group="identity", password_hash_rounds=16)
+        old_hash = scrypt.Scrypt.hash(b"password", n=8, r=8, p=1)
+        self.assertTrue(password_hashing.needs_rehash(old_hash))
+
+    def test_needs_rehash_scrypt_current_params(self):
+        self.config_fixture.config(
+            group="identity", password_hash_algorithm="scrypt"
+        )
+        self.config_fixture.config(group="identity", password_hash_rounds=16)
+        current_hash = scrypt.Scrypt.hash(b"password", n=16, r=8, p=1)
+        self.assertFalse(password_hashing.needs_rehash(current_hash))
+
 
 class TestGeneratePartialPasswordHash(unit.BaseTestCase):
     def setUp(self):

keystone/tests/unit/identity/test_backend_sql.py

@@ -17,6 +17,7 @@ import freezegun
 from oslo_utils import timeutils
 
 from keystone.common import password_hashers
+from keystone.common.password_hashers import pbkdf2
 from keystone.common import password_hashing
 from keystone.common import provider_api
 from keystone.common import resource_options
@@ -1083,3 +1084,180 @@ class ChangePasswordRequiredAfterFirstUse(test_backend_sql.SqlTests):
         user['password'] = admin_password
         PROVIDERS.identity_api.update_user(user['id'], user)
         self.assertPasswordIsExpired(user['id'], admin_password)
+
+
+class Pbkdf2Sha512RehashTests(test_backend_sql.SqlTests):
+    """Integration tests for transparent PBKDF2-SHA512 password rehashing."""
+
+    def config_overrides(self):
+        super().config_overrides()
+        self.config_fixture.config(
+            group='identity', password_hash_algorithm='pbkdf2_sha512'
+        )
+
+    def _create_pbkdf2_user(self, rounds):
+        """Create a user whose password is hashed with a specific round count."""
+        self.config_fixture.config(
+            group='identity', password_hash_rounds=rounds
+        )
+        password = uuid.uuid4().hex
+        user = PROVIDERS.identity_api.create_user(
+            {
+                'name': uuid.uuid4().hex,
+                'domain_id': CONF.identity.default_domain_id,
+                'enabled': True,
+                'password': password,
+            }
+        )
+        return user, password
+
+    def _get_stored_rounds(self, user_id):
+        with sql.session_for_read() as session:
+            user_ref = PROVIDERS.identity_api._get_user(session, user_id)
+            return int(user_ref.password.split('$')[2])
+
+    def test_authenticate_transparently_rehashes_low_iteration_password(self):
+        user, password = self._create_pbkdf2_user(rounds=10000)
+        self.assertEqual(10000, self._get_stored_rounds(user['id']))
+
+        self.config_fixture.config(
+            group='identity', password_hash_rounds=pbkdf2.DEFAULT_ROUNDS
+        )
+        with self.make_request():
+            PROVIDERS.identity_api.authenticate(
+                user_id=user['id'], password=password
+            )
+
+        self.assertEqual(
+            pbkdf2.DEFAULT_ROUNDS, self._get_stored_rounds(user['id'])
+        )
+
+    def test_authenticate_does_not_rehash_current_iteration_password(self):
+        user, password = self._create_pbkdf2_user(rounds=pbkdf2.DEFAULT_ROUNDS)
+
+        with sql.session_for_read() as session:
+            user_ref = PROVIDERS.identity_api._get_user(session, user['id'])
+            original_hash = user_ref.password
+
+        with self.make_request():
+            PROVIDERS.identity_api.authenticate(
+                user_id=user['id'], password=password
+            )
+
+        with sql.session_for_read() as session:
+            user_ref = PROVIDERS.identity_api._get_user(session, user['id'])
+            self.assertEqual(original_hash, user_ref.password)
+
+    def test_rehash_does_not_alter_password_metadata(self):
+        user, password = self._create_pbkdf2_user(rounds=10000)
+
+        with sql.session_for_read() as session:
+            user_ref = PROVIDERS.identity_api._get_user(session, user['id'])
+            created_at_before = user_ref.password_ref.created_at
+            expires_at_before = user_ref.password_ref.expires_at
+
+        self.config_fixture.config(
+            group='identity', password_hash_rounds=pbkdf2.DEFAULT_ROUNDS
+        )
+        with self.make_request():
+            PROVIDERS.identity_api.authenticate(
+                user_id=user['id'], password=password
+            )
+
+        with sql.session_for_read() as session:
+            user_ref = PROVIDERS.identity_api._get_user(session, user['id'])
+            self.assertEqual(1, len(user_ref.local_user.passwords))
+            self.assertEqual(
+                created_at_before, user_ref.password_ref.created_at
+            )
+            self.assertEqual(
+                expires_at_before, user_ref.password_ref.expires_at
+            )
+
+    def test_rehash_skipped_when_hash_changed_concurrently(self):
+        user, password = self._create_pbkdf2_user(rounds=10000)
+
+        with sql.session_for_read() as session:
+            user_ref = PROVIDERS.identity_api._get_user(session, user['id'])
+            original_hash = user_ref.password
+
+        admin_password = uuid.uuid4().hex
+        PROVIDERS.identity_api.update_user(
+            user['id'], {'password': admin_password}
+        )
+
+        with sql.session_for_read() as session:
+            user_ref = PROVIDERS.identity_api._get_user(session, user['id'])
+            admin_hash = user_ref.password
+
+        self.assertNotEqual(original_hash, admin_hash)
+
+        driver = PROVIDERS.identity_api.driver
+        driver._rehash_password(user['id'], password, original_hash)
+
+        with sql.session_for_read() as session:
+            user_ref = PROVIDERS.identity_api._get_user(session, user['id'])
+            self.assertEqual(admin_hash, user_ref.password)
+
+
+class BcryptRehashTests(test_backend_sql.SqlTests):
+    """Integration tests for transparent bcrypt password rehashing."""
+
+    def config_overrides(self):
+        super().config_overrides()
+        self.config_fixture.config(
+            group='identity', password_hash_algorithm='bcrypt'
+        )
+
+    def _create_bcrypt_user(self, rounds):
+        self.config_fixture.config(
+            group='identity', password_hash_rounds=rounds
+        )
+        password = uuid.uuid4().hex
+        user = PROVIDERS.identity_api.create_user(
+            {
+                'name': uuid.uuid4().hex,
+                'domain_id': CONF.identity.default_domain_id,
+                'enabled': True,
+                'password': password,
+            }
+        )
+        return user, password
+
+    def _get_stored_rounds(self, user_id):
+        with sql.session_for_read() as session:
+            user_ref = PROVIDERS.identity_api._get_user(session, user_id)
+            return int(user_ref.password.split('$')[2])
+
+    def test_authenticate_transparently_rehashes_low_cost_password(self):
+        user, password = self._create_bcrypt_user(rounds=4)
+        self.assertEqual(4, self._get_stored_rounds(user['id']))
+
+        self.config_fixture.config(group='identity', password_hash_rounds=12)
+        with self.make_request():
+            PROVIDERS.identity_api.authenticate(
+                user_id=user['id'], password=password
+            )
+
+        self.assertEqual(12, self._get_stored_rounds(user['id']))
+
+    def test_authenticate_migrates_hash_when_algorithm_changes(self):
+        user, password = self._create_bcrypt_user(rounds=4)
+        with sql.session_for_read() as session:
+            user_ref = PROVIDERS.identity_api._get_user(session, user['id'])
+            self.assertTrue(user_ref.password.startswith('$2'))
+
+        self.config_fixture.config(
+            group='identity', password_hash_algorithm='pbkdf2_sha512'
+        )
+        self.config_fixture.config(
+            group='identity', password_hash_rounds=pbkdf2.DEFAULT_ROUNDS
+        )
+        with self.make_request():
+            PROVIDERS.identity_api.authenticate(
+                user_id=user['id'], password=password
+            )
+
+        with sql.session_for_read() as session:
+            user_ref = PROVIDERS.identity_api._get_user(session, user['id'])
+            self.assertTrue(user_ref.password.startswith('$pbkdf2-sha512$'))

releasenotes/notes/pbkdf2-transparent-rehash-a1b2c3d4e5f6g7h8.yaml

@@ -0,0 +1,12 @@
+---
+features:
+  - |
+    Keystone now transparently upgrades stored password hashes on successful
+    authentication when the stored hash uses a different algorithm than
+    ``[identity] password_hash_algorithm``, or a weaker work factor than the
+    currently configured value. This applies to bcrypt, bcrypt_sha256, scrypt,
+    and PBKDF2-SHA512. Operators can raise ``password_hash_rounds`` or change
+    algorithm without requiring users to reset their passwords. The rehash
+    preserves password metadata (created_at, expires_at, history) and only
+    updates the hash itself. See the administrator guide on password hashing
+    for details.