Passwords stored salt and some change is password logic

backend/app/api/user_config.py

@@ -8,7 +8,7 @@ from fastapi import APIRouter, Depends, File, HTTPException, UploadFile
 from pydantic import BaseModel
 
 from backend.app.api.dependencies import get_current_user
-from backend.app.database import AVATARS_DIR, get_connection, hash_password
+from backend.app.database import AVATARS_DIR, get_connection, hash_password, verify_password
 from backend.app.services.link_service import create_label, delete_label, list_user_labels, update_label
 from backend.app.services.otp_service import create_secret, provisioning_uri, verify_code
 
@@ -86,7 +86,7 @@ def update_current_user_profile(
 def update_password(payload: PasswordUpdate, user: dict = Depends(get_current_user)):
     if len(payload.new_password) < 8:
         raise HTTPException(status_code=422, detail='New password must be at least 8 characters')
-    if hash_password(payload.current_password) != user['password_hash']:
+    if not verify_password(payload.current_password, user['password_hash']):
         raise HTTPException(status_code=400, detail='Current password is incorrect')
 
     with get_connection() as conn:

backend/app/database.py

@@ -3,7 +3,8 @@
 
 import sqlite3
 import os
-from hashlib import sha256
+from hashlib import scrypt, sha256
+import hmac
 from pathlib import Path
 from uuid import uuid4
 
@@ -15,7 +16,28 @@ AVATARS_DIR.mkdir(parents=True, exist_ok=True)
 
 
 def hash_password(password: str) -> str:
-    return sha256(password.encode('utf-8')).hexdigest()
+    salt = os.urandom(16)
+    digest = scrypt(password.encode('utf-8'), salt=salt, n=16_384, r=8, p=1, dklen=32)
+    return f'scrypt$16384$8$1${salt.hex()}${digest.hex()}'
+
+
+def verify_password(password: str, stored_hash: str) -> bool:
+    if stored_hash.startswith('scrypt$'):
+        try:
+            algorithm, cost, block_size, parallelism, salt_hex, digest_hex = stored_hash.split('$')
+            if algorithm != 'scrypt':
+                return False
+            digest = scrypt(
+                password.encode('utf-8'), salt=bytes.fromhex(salt_hex),
+                n=int(cost), r=int(block_size), p=int(parallelism), dklen=32,
+            )
+            return hmac.compare_digest(digest.hex(), digest_hex)
+        except (ValueError, TypeError):
+            return False
+    if len(stored_hash) == 64:
+        legacy_digest = sha256(password.encode('utf-8')).hexdigest()
+        return hmac.compare_digest(legacy_digest, stored_hash)
+    return False
 
 DEFAULT_TAGS = ('#Internet', '#Cybersecurity', '#Fediverse', '#Food', '#Photography', '#Music', '#AI')
 

backend/app/services/auth_service.py

@@ -2,23 +2,25 @@
 ## SPDX-License-Identifier: GPL-3.0-or-later
 
 from datetime import datetime, timezone
-from hashlib import sha256
-
-from backend.app.database import get_connection
-
-
-def hash_password(password: str) -> str:
-    return sha256(password.encode('utf-8')).hexdigest()
+from backend.app.database import get_connection, hash_password, verify_password
 
 
 def authenticate_user(username: str, password: str):
-    password_hash = hash_password(password)
     with get_connection() as conn:
-        row = conn.execute(
-            'SELECT * FROM users WHERE username = ? AND password_hash = ?',
-            (username, password_hash),
-        ).fetchone()
-    return dict(row) if row else None
+        row = conn.execute('SELECT * FROM users WHERE username = ?', (username,)).fetchone()
+        if row is None or not verify_password(password, row['password_hash']):
+            return None
+        user = dict(row)
+        if not row['password_hash'].startswith('scrypt$'):
+            conn.execute(
+                'UPDATE users SET password_hash = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ?',
+                (hash_password(password), row['id']),
+            )
+            conn.commit()
+            user['password_hash'] = conn.execute(
+                'SELECT password_hash FROM users WHERE id = ?', (row['id'],)
+            ).fetchone()['password_hash']
+    return user
 
 
 def find_user(username: str):