fixes

Author: igorbenav

.github/workflows/type-checking.yml

@@ -26,7 +26,7 @@ jobs:
           SECRET_KEY: test-secret-key-for-testing-only
 
       - name: Mypy (cli)
-        run: cd cli && uv run --no-sync mypy src/cli
+        run: cd cli && uv run --no-sync mypy -p cli
         env:
           ENVIRONMENT: local
           SECRET_KEY: test-secret-key-for-testing-only

backend/src/modules/api_keys/service.py

@@ -1,19 +1,22 @@
 """API key management service for developer-facing products."""
 
+import base64
+import binascii
 import hashlib
 import hmac
 import secrets
 from datetime import UTC, datetime, timedelta
 from typing import Any
 
 from fastcrud.types import GetMultiResponseDict
+from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from ...infrastructure.config.settings import get_settings
 from ...infrastructure.logging import get_logger
 from ..common.exceptions import PermissionDeniedError, ResourceNotFoundError
 from .crud import crud_api_keys, crud_key_permissions, crud_key_usage
 from .enums import KeyPermissionAction, KeyPermissionResource
+from .models import APIKey
 from .schemas import (
     APIKeyCreate,
     APIKeyCreateInternal,
@@ -33,7 +36,11 @@
 )
 
 logger = get_logger()
-settings = get_settings()
+
+_SCRYPT_N = 2**14
+_SCRYPT_R = 8
+_SCRYPT_P = 1
+_SCRYPT_DKLEN = 32
 
 
 class APIKeyService:
@@ -62,17 +69,49 @@ def _generate_api_key(self) -> tuple[str, str, str]:
         return api_key, prefix, key_hash
 
     def _hash_api_key(self, api_key: str) -> str:
-        """Hash an API key for storage or comparison.
+        """Hash an API key for storage using scrypt with a per-row salt.
 
-        HMAC-SHA256 with SECRET_KEY as the pepper. Deterministic so DB lookup
-        by `key_hash` stays O(1); the server-side secret means a stolen
-        `key_hash` column alone cannot be used to forge or verify keys offline.
+        Stored format: ``scrypt$N$r$p$salt_b64$derived_b64``. Non-deterministic;
+        DB lookup uses ``key_prefix`` (already indexed) instead of ``key_hash``.
         """
-        return hmac.new(
-            settings.SECRET_KEY.encode("utf-8"),
+        salt = secrets.token_bytes(16)
+        derived = hashlib.scrypt(
             api_key.encode("utf-8"),
-            hashlib.sha256,
-        ).hexdigest()
+            salt=salt,
+            n=_SCRYPT_N,
+            r=_SCRYPT_R,
+            p=_SCRYPT_P,
+            dklen=_SCRYPT_DKLEN,
+        )
+        salt_b64 = base64.b64encode(salt).decode("ascii")
+        derived_b64 = base64.b64encode(derived).decode("ascii")
+        return f"scrypt${_SCRYPT_N}${_SCRYPT_R}${_SCRYPT_P}${salt_b64}${derived_b64}"
+
+    def _verify_api_key(self, api_key: str, stored_hash: str) -> bool:
+        """Verify a candidate ``api_key`` against a stored scrypt hash."""
+        try:
+            scheme, n_str, r_str, p_str, salt_b64, derived_b64 = stored_hash.split("$", 5)
+        except ValueError:
+            return False
+        if scheme != "scrypt":
+            return False
+        try:
+            n = int(n_str)
+            r = int(r_str)
+            p = int(p_str)
+            salt = base64.b64decode(salt_b64)
+            expected = base64.b64decode(derived_b64)
+        except (ValueError, binascii.Error):
+            return False
+        actual = hashlib.scrypt(
+            api_key.encode("utf-8"),
+            salt=salt,
+            n=n,
+            r=r,
+            p=p,
+            dklen=len(expected),
+        )
+        return hmac.compare_digest(actual, expected)
 
     async def create_api_key(
         self,
@@ -263,15 +302,31 @@ async def validate_api_key(
         Returns:
             Validation response with key details and permissions
         """
-        key_hash = self._hash_api_key(api_key)
-        key = await crud_api_keys.get(db=db, key_hash=key_hash, schema_to_select=APIKeyRead)
+        parts = api_key.split("_", 2)
+        if len(parts) != 3 or parts[0] != "fai":
+            return APIKeyValidationResponse(
+                is_valid=False,
+                error_message="Invalid API key",
+            )
+        prefix = parts[1]
 
-        if not key:
+        result = await db.execute(select(APIKey).where(APIKey.key_prefix == prefix).execution_options(populate_existing=True))
+        candidates = result.scalars().all()
+
+        matched: APIKey | None = None
+        for candidate in candidates:
+            if self._verify_api_key(api_key, candidate.key_hash):
+                matched = candidate
+                break
+
+        if matched is None:
             return APIKeyValidationResponse(
                 is_valid=False,
                 error_message="Invalid API key",
             )
 
+        key = APIKeyRead.model_validate(matched).model_dump()
+
         if not key["is_active"]:
             return APIKeyValidationResponse(
                 is_valid=False,

backend/tests/conftest.py

@@ -19,7 +19,7 @@
 
 from src.infrastructure.auth.session.backends.memory import MemorySessionStorage
 from src.infrastructure.auth.session.dependencies import get_current_superuser, get_current_user
-from src.infrastructure.auth.session.schemas import CSRFToken
+from src.infrastructure.auth.session.schemas import CSRFToken, SessionData
 from src.infrastructure.auth.utils import get_password_hash
 from src.infrastructure.config.settings import Settings, get_settings
 from src.infrastructure.database.session import Base, async_session
@@ -258,8 +258,8 @@ async def override_get_current_superuser():
 @pytest.fixture(autouse=True)
 def mock_session_backend(monkeypatch):
     """Use in-memory session backend instead of Redis during tests."""
-    memory_storage = MemorySessionStorage(prefix="session:", expiration=1800)
-    memory_csrf_storage = MemorySessionStorage(prefix="csrf:", expiration=1800)
+    memory_storage: MemorySessionStorage[SessionData] = MemorySessionStorage(prefix="session:", expiration=1800)
+    memory_csrf_storage: MemorySessionStorage[CSRFToken] = MemorySessionStorage(prefix="csrf:", expiration=1800)
 
     def override_session_dependency(backend, model_type, **kwargs):
         if model_type == CSRFToken:

backend/tests/unit/modules/api_keys/test_service.py

@@ -370,15 +370,19 @@ async def test_get_user_summary(api_key_service, db_session: AsyncSession, test_
 
 
 @pytest.mark.asyncio
-async def test_api_key_hash_consistency(api_key_service):
-    """Test that API key hashing is consistent."""
+async def test_api_key_hash_roundtrip(api_key_service):
+    """Hashing produces a fresh salt each call; verifying must still succeed."""
     test_key = "fai_test_key_12345"
 
     hash1 = api_key_service._hash_api_key(test_key)
     hash2 = api_key_service._hash_api_key(test_key)
 
-    assert hash1 == hash2
-    assert len(hash1) == 64  # SHA256 hex digest length
+    assert hash1 != hash2
+    assert hash1.startswith("scrypt$")
+    assert hash2.startswith("scrypt$")
+    assert api_key_service._verify_api_key(test_key, hash1)
+    assert api_key_service._verify_api_key(test_key, hash2)
+    assert not api_key_service._verify_api_key("fai_wrong_key", hash1)
 
 
 @pytest.mark.asyncio