feat: bring password hashing inline with industry best practices (#881)

* feat: update iteration count and add support for automatic hash migration

* Resolve lint failures

* resolve mypy error

Mypy complains because iterations was a string, but we change it to an int

* add some comments

* add unit test

---------

Co-authored-by: Daniel Townsend <[email protected]>

piccolo/apps/user/tables.py

@@ -49,6 +49,9 @@ class BaseUser(Table, tablename="piccolo_user"):
 
     _min_password_length = 6
     _max_password_length = 128
+    # The number of hash iterations recommended by OWASP:
+    # https://cheatsheetseries.owasp.org/cheatsheets/Password_Storage_Cheat_Sheet.html#pbkdf2
+    _pbkdf2_iteration_count = 600_000
 
     def __init__(self, **kwargs):
         # Generating passwords upfront is expensive, so might need reworking.
@@ -131,7 +134,7 @@ async def update_password(cls, user: t.Union[str, int], password: str):
 
     @classmethod
     def hash_password(
-        cls, password: str, salt: str = "", iterations: int = 10000
+        cls, password: str, salt: str = "", iterations: t.Optional[int] = None
     ) -> str:
         """
         Hashes the password, ready for storage, and for comparing during
@@ -147,6 +150,10 @@ def hash_password(
 
         if not salt:
             salt = cls.get_salt()
+
+        if iterations is None:
+            iterations = cls._pbkdf2_iteration_count
+
         hashed = hashlib.pbkdf2_hmac(
             "sha256",
             bytes(password, encoding="utf-8"),
@@ -210,14 +217,18 @@ async def login(cls, username: str, password: str) -> t.Optional[int]:
 
         stored_password = response["password"]
 
-        algorithm, iterations, salt, hashed = cls.split_stored_password(
+        algorithm, iterations_, salt, hashed = cls.split_stored_password(
             stored_password
         )
+        iterations = int(iterations_)
+
+        if cls.hash_password(password, salt, iterations) == stored_password:
+            # If the password was hashed in an earlier Piccolo version, update
+            # it so it's hashed with the currently recommended number of
+            # iterations:
+            if iterations != cls._pbkdf2_iteration_count:
+                await cls.update_password(username, password)
 
-        if (
-            cls.hash_password(password, salt, int(iterations))
-            == stored_password
-        ):
             await cls.update({cls.last_login: datetime.datetime.now()}).where(
                 cls.username == username
             )

tests/apps/user/test_tables.py

@@ -234,3 +234,58 @@ def test_no_password_error(self):
         self.assertEqual(
             manager.exception.__str__(), "A password must be provided."
         )
+
+
+class TestAutoHashingUpdate(TestCase):
+    """
+    Make sure that users with passwords which were hashed in earlier Piccolo
+    versions are automatically re-hashed, meeting current best practices with
+    the number of hashing iterations.
+    """
+
+    def setUp(self):
+        BaseUser.create_table().run_sync()
+
+    def tearDown(self):
+        BaseUser.alter().drop_table().run_sync()
+
+    def test_hash_update(self):
+        # Create a user
+        username = "bob"
+        password = "abc123"
+        user = BaseUser.create_user_sync(username=username, password=password)
+
+        # Update their password, so it uses less than the recommended number
+        # of hashing iterations.
+        BaseUser.update(
+            {
+                BaseUser.password: BaseUser.hash_password(
+                    password=password,
+                    iterations=int(BaseUser._pbkdf2_iteration_count / 2),
+                )
+            }
+        ).where(BaseUser.id == user.id).run_sync()
+
+        # Login the user - Piccolo should detect their password needs rehashing
+        # and update it.
+        self.assertIsNotNone(
+            BaseUser.login_sync(username=username, password=password)
+        )
+
+        hashed_password = (
+            BaseUser.select(BaseUser.password)
+            .where(BaseUser.id == user.id)
+            .first()
+            .run_sync()["password"]
+        )
+
+        algorithm, iterations_, salt, hashed = BaseUser.split_stored_password(
+            hashed_password
+        )
+
+        self.assertEqual(int(iterations_), BaseUser._pbkdf2_iteration_count)
+
+        # Make sure subsequent logins work as expected
+        self.assertIsNotNone(
+            BaseUser.login_sync(username=username, password=password)
+        )