Implement license key generation with batch metadata and error handling

.github/workflows/code_test.yml

@@ -2,12 +2,12 @@ name: Tests
 
 on:
   push:
-    branches: [ main ]
+    branches: [main]
   pull_request:
-    branches: [ main ]
+    branches: [main]
 
   schedule:
-    - cron: "0 0 * * 0"  # Runs every Sunday at midnight UTC
+    - cron: "0 0 * * 0" # Runs every Sunday at midnight UTC
 
   workflow_dispatch:
 
@@ -34,6 +34,8 @@ jobs:
   test_backend:
     needs: fetch-python-versions
     runs-on: ubuntu-latest
+    env:
+      UV_NO_DEV: true
 
     permissions:
       id-token: write
@@ -52,7 +54,7 @@ jobs:
           python-version: ${{ matrix.python-version }}
 
       - name: Install dependencies
-        run: uv sync --group dev
+        run: uv sync --group test
 
       - name: Unit tests
         working-directory: backend

.github/workflows/sonarcloud.yml

@@ -4,23 +4,25 @@ permissions:
   contents: read
 on:
   push:
-    branches: [ main ]
+    branches: [main]
   pull_request:
-    branches: [ main ]
+    branches: [main]
 
 jobs:
   sonarqube:
     name: SonarQube
     runs-on: ubuntu-latest
+    env:
+      UV_NO_DEV: true
     steps:
       - uses: actions/checkout@v6
         with:
-          fetch-depth: 0  # Shallow clones should be disabled for a better relevancy of analysis
+          fetch-depth: 0 # Shallow clones should be disabled for a better relevancy of analysis
 
       - uses: astral-sh/setup-uv@v7
 
       - name: Install dependencies
-        run: uv sync --group dev
+        run: uv sync --group test
 
       - name: Run tests and generate coverage
         working-directory: backend

.gitignore

@@ -16,3 +16,6 @@ dist/
 *.log
 coverage.xml
 .coverage.*
+.fastembed_cache
+.database
+.amdb

backend/app/api/routes/auth.py

@@ -17,11 +17,7 @@
 router = APIRouter()
 
 
-@router.post(
-    "/signup",
-    status_code=status.HTTP_201_CREATED,
-    response_model=SuccessResponse[SignupResponse],
-)
+@router.post("/signup", status_code=status.HTTP_201_CREATED)
 def signup(
     body: SignupRequest, cursor: CursorDep, settings: SettingsDep
 ) -> SuccessResponse[SignupResponse]:
@@ -40,7 +36,7 @@ def signup(
     return SuccessResponse(data=result)
 
 
-@router.post("/login", response_model=SuccessResponse[TokenPair])
+@router.post("/login")
 def login(
     body: LoginRequest, cursor: CursorDep, settings: SettingsDep
 ) -> SuccessResponse[TokenPair]:
@@ -59,7 +55,7 @@ def login(
     return SuccessResponse(data=result)
 
 
-@router.post("/refresh", response_model=SuccessResponse[TokenPair])
+@router.post("/refresh")
 def refresh(
     body: RefreshRequest, cursor: CursorDep, settings: SettingsDep
 ) -> SuccessResponse[TokenPair]:

backend/app/core/ed25519.py

@@ -0,0 +1,30 @@
+from __future__ import annotations
+
+from cryptography.hazmat.primitives.asymmetric.ed25519 import (
+    Ed25519PrivateKey,
+    Ed25519PublicKey,
+)
+from cryptography.hazmat.primitives.serialization import (
+    load_pem_private_key,
+    load_pem_public_key,
+)
+from cryptography.utils import Buffer
+
+
+__all__ = ["load_private_ed25519_key", "load_public_ed25519_key"]
+
+
+def load_private_ed25519_key(
+    pem: Buffer, password: bytes | None = None
+) -> Ed25519PrivateKey:
+    key = load_pem_private_key(pem, password=password)
+    if not isinstance(key, Ed25519PrivateKey):
+        raise TypeError("Provided key is not an Ed25519 private key")
+    return key
+
+
+def load_public_ed25519_key(pem: bytes) -> Ed25519PublicKey:
+    key = load_pem_public_key(pem)
+    if not isinstance(key, Ed25519PublicKey):
+        raise TypeError("Provided key is not an Ed25519 public key")
+    return key

backend/app/core/exception_handlers.py

@@ -38,7 +38,7 @@ def api_exception_handler(request: Request, exc: APIException) -> JSONResponse:
             code=exc.error_code,
             message=exc.message,
             http_status=exc.http_status,
-            details=_build_error_details(exc.details),
+            details=build_error_details(exc.details),
             request_id=request_id,
         )
     )
@@ -64,7 +64,7 @@ def validation_exception_handler(
     details = []
     for error in exc.errors():
         field_path = ".".join(str(loc) for loc in error["loc"][1:])
-        field = field_path if field_path else None
+        field = field_path or None
         details.append({"field": field, "message": error["msg"]})
 
     logger.warning(
@@ -84,7 +84,7 @@ def validation_exception_handler(
             code=ErrorCode.VALIDATION_FAILED,
             message="Validation error",
             http_status=http_422_unprocessable_content,
-            details=_build_error_details(details),
+            details=build_error_details(details),
             request_id=request_id,
         )
     )
@@ -105,7 +105,7 @@ def general_exception_handler(request: Request, exc: Exception) -> JSONResponse:
     request_id = getattr(request.state, "request_id", str(uuid.uuid4()))
 
     # Log the full traceback server-side
-    logger.exception(
+    logger.exception(  # noqa: LOG004 - this will be used by the exceptions
         "Unexpected error: %s", exc, extra={"request_id": request_id}
     )
 
@@ -127,7 +127,7 @@ def general_exception_handler(request: Request, exc: Exception) -> JSONResponse:
     )
 
 
-def _build_error_details(
+def build_error_details(
     details: list[dict | ErrorDetail] | dict | ErrorDetail | None,
 ) -> list[ErrorDetail]:
     """

backend/app/core/exceptions.py

@@ -157,3 +157,16 @@ def __init__(self, message: str = "License has expired") -> None:
             message=message,
             http_status=status.HTTP_409_CONFLICT,
         )
+
+
+class LicenseKeyGenerationError(APIException):
+    """License key generation failed after exhausting retries."""
+
+    def __init__(
+        self, message: str = "Failed to generate a unique license key"
+    ) -> None:
+        super().__init__(
+            error_code=ErrorCode.LICENSE_KEY_GENERATION_ERROR,
+            message=message,
+            http_status=status.HTTP_500_INTERNAL_SERVER_ERROR,
+        )

backend/app/core/security.py

@@ -1,24 +1,24 @@
 from datetime import datetime, timedelta, timezone
 from typing import Any
+from uuid import UUID
 
 import jwt
 from pwdlib import PasswordHash
-from pwdlib.hashers.argon2 import Argon2Hasher
-from pwdlib.hashers.bcrypt import BcryptHasher
+from pwdlib.hashers import argon2, bcrypt
 
 from app.core.config import Settings
 
 
 # BcryptHasher is listed first so new passwords are hashed with bcrypt.
 # Argon2Hasher is kept for verification of legacy hashes.
-password_hash = PasswordHash((BcryptHasher(), Argon2Hasher()))
+PASSWORD_HASH = PasswordHash((bcrypt.BcryptHasher(), argon2.Argon2Hasher()))
 
 
-ALGORITHM = "HS256"
+JWT_ALGORITHM: str = "HS256"
 
 
 def create_access_token(
-    vendor_id: str,
+    vendor_id: str | UUID,
     settings: Settings,
     *,
     expires_delta: timedelta | None = None,
@@ -36,11 +36,11 @@ def create_access_token(
         "exp": expire,
         "token_type": "access",
     }
-    return jwt.encode(to_encode, settings.SECRET_KEY, algorithm=ALGORITHM)
+    return jwt.encode(to_encode, settings.SECRET_KEY, algorithm=JWT_ALGORITHM)
 
 
 def create_refresh_token(
-    vendor_id: str,
+    vendor_id: str | UUID,
     settings: Settings,
     *,
     expires_delta: timedelta | None = None,
@@ -58,7 +58,7 @@ def create_refresh_token(
         "exp": expire,
         "token_type": "refresh",
     }
-    return jwt.encode(to_encode, settings.SECRET_KEY, algorithm=ALGORITHM)
+    return jwt.encode(to_encode, settings.SECRET_KEY, algorithm=JWT_ALGORITHM)
 
 
 def decode_token(token: str, settings: Settings) -> dict[str, Any]:
@@ -67,14 +67,14 @@ def decode_token(token: str, settings: Settings) -> dict[str, Any]:
     Returns:
         dict[str, Any]: The decoded token payload.
     """
-    return jwt.decode(token, settings.SECRET_KEY, algorithms=[ALGORITHM])
+    return jwt.decode(token, settings.SECRET_KEY, algorithms=[JWT_ALGORITHM])
 
 
 def verify_password(
     plain_password: str, hashed_password: str
 ) -> tuple[bool, str | None]:
-    return password_hash.verify_and_update(plain_password, hashed_password)
+    return PASSWORD_HASH.verify_and_update(plain_password, hashed_password)
 
 
 def get_password_hash(password: str) -> str:
-    return password_hash.hash(password)
+    return PASSWORD_HASH.hash(password)

backend/app/crud/vendor.py

@@ -13,11 +13,13 @@ def get_vendor_by_email(cursor: Cursor, email: str) -> dict[str, Any] | None:
     Returns:
         dict[str, Any] | None: The vendor row or None.
     """
-    cursor.execute(
-        'SELECT "id", "email", "password_hash" '
-        'FROM app."vendors" '
-        'WHERE LOWER("email") = LOWER(%s) '
-        'AND "deleted_at" IS NULL',
+    cursor.execute(  # SQL
+        """
+        SELECT "id", "email", "password_hash"
+        FROM app."vendors"
+        WHERE LOWER("email") = LOWER(%s)
+        AND "deleted_at" IS NULL
+        """,
         (email,),
     )
     row = cursor.fetchone()
@@ -32,10 +34,12 @@ def get_vendor_by_id(cursor: Cursor, vendor_id: str) -> dict[str, Any] | None:
     Returns:
         dict[str, Any] | None: The vendor row or None.
     """
-    cursor.execute(
-        'SELECT "id", "email" '
-        'FROM app."vendors" '
-        'WHERE "id" = %s AND "deleted_at" IS NULL',
+    cursor.execute(  # SQL
+        """
+        SELECT "id", "email"
+        FROM app."vendors"
+        WHERE "id" = %s AND "deleted_at" IS NULL
+        """,
         (vendor_id,),
     )
     row = cursor.fetchone()
@@ -56,11 +60,13 @@ def create_vendor(
     Returns:
         dict[str, Any] | None: The created vendor row, or None on conflict.
     """
-    cursor.execute(
-        'INSERT INTO app."vendors" ("email", "password_hash") '
-        "VALUES (%s, %s) "
-        'ON CONFLICT ((LOWER("email"))) DO NOTHING '
-        'RETURNING "id", "email"',
+    cursor.execute(  # SQL
+        """
+        INSERT INTO app."vendors" ("email", "password_hash")
+        VALUES (%s, %s)
+        ON CONFLICT (LOWER("email")) DO NOTHING
+        RETURNING "id", "email"
+        """,
         (email, password_hash),
     )
     row = cursor.fetchone()

backend/app/domain/activation.py

@@ -0,0 +1,60 @@
+from __future__ import annotations
+
+from typing import ClassVar
+from uuid import UUID
+
+import uuid6
+from pydantic import BaseModel, ConfigDict, computed_field, field_validator
+
+from app.internal import base32_crockford
+
+
+__all__ = ["ActivationCode"]
+
+
+class ActivationCode(BaseModel):
+    model_config = ConfigDict(validate_assignment=True)
+
+    LENGTH: ClassVar[int] = 30
+    GROUP: ClassVar[int] = 5
+
+    code: str
+
+    @field_validator("code")
+    @classmethod
+    def validate_code(cls, v: str) -> str:
+        # TODO: need the proper error handling here.
+        normalized = base32_crockford.normalize(v)
+
+        if len(normalized) != cls.LENGTH:
+            raise ValueError("Activation code must contain 30 symbols")
+
+        base32_crockford.decode(normalized, checksum=True)
+
+        return "-".join(
+            normalized[i : i + cls.GROUP]
+            for i in range(0, cls.LENGTH, cls.GROUP)
+        )
+
+    @computed_field
+    @property
+    def uuid(self) -> UUID:
+        flat = base32_crockford.normalize(self.code)
+        n = base32_crockford.decode(flat, checksum=True)
+        return UUID(int=n)
+
+    @classmethod
+    def generate(cls, uuid: UUID | None = None) -> ActivationCode:
+        if uuid is None:
+            uuid = uuid6.uuid7()
+        if not isinstance(uuid, UUID):
+            raise TypeError(f"uuid cannot be of type {uuid.__class__.__name__}")
+        if uuid.version != 7:  # noqa: PLR2004
+            raise TypeError(
+                f"uuid must be a UUID version 7, got {uuid.version}"
+            )
+
+        encoded = base32_crockford.encode(uuid.int, checksum=True)
+        encoded = encoded.rjust(cls.LENGTH, "0")
+
+        return cls(code=encoded)