Update to Argon2

operate/account/user.py

@@ -23,21 +23,23 @@
 from dataclasses import dataclass
 from pathlib import Path
 
+from argon2 import PasswordHasher, Type
+from argon2.exceptions import InvalidHashError, VerificationError
+
 from operate.resource import LocalResource
 
 
-def sha256(string: str) -> str:
-    """Get SHA256 hexdigest of a string."""
-    sh256 = hashlib.sha256()
-    sh256.update(string.encode())
-    return sh256.hexdigest()
+def argon2id(string: str) -> str:
+    """Get Argon2id hexdigest of a string."""
+    ph = PasswordHasher(type=Type.ID)
+    return ph.hash(string)
 
 
 @dataclass
 class UserAccount(LocalResource):
     """User account."""
 
-    password_sha: str
+    password_hash: str
     path: Path
 
     @classmethod
@@ -49,24 +51,38 @@ def load(cls, path: Path) -> "UserAccount":
     def new(cls, password: str, path: Path) -> "UserAccount":
         """Create a new user."""
         user = UserAccount(
-            password_sha=sha256(string=password),
+            password_hash=argon2id(string=password),
             path=path,
         )
         user.store()
         return UserAccount.load(path=path)
 
     def is_valid(self, password: str) -> bool:
         """Check if a password string is valid."""
-        return sha256(string=password) == self.password_sha
+        try:
+            ph = PasswordHasher(type=Type.ID)
+            return ph.verify(self.password_hash, password)
+        except VerificationError:
+            return False
+        except InvalidHashError:
+            # Verifies legacy password hash and updates to Argon2id if valid
+            sha256 = hashlib.sha256()
+            sha256.update(password.encode())
+            if sha256.hexdigest() == self.password_hash:
+                self.password_hash = argon2id(string=password)
+                self.store()
+                return True
+
+            return False
 
     def update(self, old_password: str, new_password: str) -> None:
         """Update current password."""
         if not self.is_valid(password=old_password):
             raise ValueError("Old password is not valid")
-        self.password_sha = sha256(string=new_password)
+        self.password_hash = argon2id(string=new_password)
         self.store()
 
     def force_update(self, new_password: str) -> None:
         """Force update current password."""
-        self.password_sha = sha256(string=new_password)
+        self.password_hash = argon2id(string=new_password)
         self.store()

operate/cli.py

@@ -44,6 +44,7 @@
 from operate.account.user import UserAccount
 from operate.constants import KEY, KEYS, OPERATE_HOME, SERVICES
 from operate.ledger.profiles import DEFAULT_NEW_SAFE_FUNDS_AMOUNT
+from operate.migration import MigrationManager
 from operate.operate_types import Chain, DeploymentStatus, LedgerType
 from operate.quickstart.analyse_logs import analyse_logs
 from operate.quickstart.claim_staking_rewards import claim_staking_rewards
@@ -95,6 +96,9 @@ def __init__(
         )
         self.password: t.Optional[str] = os.environ.get("OPERATE_USER_PASSWORD")
 
+        mm = MigrationManager(self._path, self.logger)
+        mm.migrate_account_user()
+
     def create_user_account(self, password: str) -> UserAccount:
         """Create a user account."""
         self.password = password

operate/migration.py

@@ -0,0 +1,60 @@
+# -*- coding: utf-8 -*-
+# ------------------------------------------------------------------------------
+#
+#   Copyright 2025 Valory AG
+#
+#   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.
+#
+# ------------------------------------------------------------------------------
+
+"""Utilities for format migration"""
+
+
+import json
+import logging
+from pathlib import Path
+
+
+class MigrationManager:
+    """MigrationManager"""
+
+    # TODO Backport here migration for services/config.json, etc.
+
+    def __init__(
+        self,
+        home: Path,
+        logger: logging.Logger,
+    ) -> None:
+        """Initialize object."""
+        super().__init__()
+        self._path = home
+        self.logger = logger
+
+    def migrate_account_user(self) -> None:
+        """Migrates user.json"""
+
+        path = self._path / "user.json"
+        if not path.exists():
+            return
+
+        with open(path, "r", encoding="utf-8") as f:
+            data = json.load(f)
+
+        if "password_sha" not in data:
+            return
+
+        new_data = {"password_hash": data["password_sha"]}
+        with open(path, "w", encoding="utf-8") as f:
+            json.dump(new_data, f, indent=4)
+
+        self.logger.info("[MIGRATION MANAGER] Migrated user.json.")