MyECLPay: unique constraints and for update lock to prevent race conditions (#727)

Add unique constraint for ed25519_public_key, refund transaction_id and
confirmation_token and Request transaction_id
Lock `for update` when selecting wallet and transactions to prevent race
conditions

Require #729 to ensure there is a unique db *transaction* per endpoint
which may either be committed or rollback

Author: armanddidierjean

app/core/myeclpay/cruds_myeclpay.py

@@ -4,10 +4,11 @@
 
 from sqlalchemy import and_, delete, or_, select, update
 from sqlalchemy.ext.asyncio import AsyncSession
-from sqlalchemy.orm import selectinload
+from sqlalchemy.orm import noload, selectinload
 
 from app.core.memberships import schemas_memberships
 from app.core.myeclpay import models_myeclpay, schemas_myeclpay
+from app.core.myeclpay.exceptions_myeclpay import WalletNotFoundOnUpdateError
 from app.core.myeclpay.types_myeclpay import (
     TransactionStatus,
     WalletDeviceStatus,
@@ -441,11 +442,16 @@ async def get_wallet(
     wallet_id: UUID,
     db: AsyncSession,
 ) -> models_myeclpay.Wallet | None:
-    result = await db.execute(
-        select(models_myeclpay.Wallet).where(
+    # We lock the wallet `for update` to prevent race conditions
+    request = (
+        select(models_myeclpay.Wallet)
+        .where(
             models_myeclpay.Wallet.id == wallet_id,
-        ),
+        )
+        .with_for_update(of=models_myeclpay.Wallet)
     )
+
+    result = await db.execute(request)
     return result.scalars().first()
 
 
@@ -512,11 +518,23 @@ async def increment_wallet_balance(
     """
     Append `amount` to the wallet balance.
     """
-    await db.execute(
-        update(models_myeclpay.Wallet)
+    # Prevent a race condition by locking the wallet row
+    # as we don't want the balance to be modified between the select and the update.
+    request = (
+        select(models_myeclpay.Wallet)
         .where(models_myeclpay.Wallet.id == wallet_id)
-        .values(balance=models_myeclpay.Wallet.balance + amount),
+        .options(
+            noload(models_myeclpay.Wallet.store),
+            noload(models_myeclpay.Wallet.user),
+        )
+        .with_for_update()
     )
+    result = await db.execute(request)
+    wallet = result.scalars().first()
+
+    if wallet is None:
+        raise WalletNotFoundOnUpdateError(wallet_id=wallet_id)
+    wallet.balance += amount
 
 
 async def create_user_payment(
@@ -600,12 +618,16 @@ async def get_transaction(
     transaction_id: UUID,
     db: AsyncSession,
 ) -> schemas_myeclpay.Transaction | None:
+    # We lock the transaction `for update` to prevent
+    # race conditions
     result = (
         (
             await db.execute(
-                select(models_myeclpay.Transaction).where(
+                select(models_myeclpay.Transaction)
+                .where(
                     models_myeclpay.Transaction.id == transaction_id,
-                ),
+                )
+                .with_for_update(of=models_myeclpay.Transaction),
             )
         )
         .scalars()

app/core/myeclpay/exceptions_myeclpay.py

@@ -0,0 +1,11 @@
+from uuid import UUID
+
+
+class WalletNotFoundOnUpdateError(Exception):
+    """
+    Exception raised when a wallet is not found during an update operation.
+    This should lead to an internal server error response and a rollback of the transaction.
+    """
+
+    def __init__(self, wallet_id: UUID):
+        super().__init__(f"Wallet {wallet_id} not found when updating")

app/core/myeclpay/models_myeclpay.py

@@ -38,7 +38,7 @@ class WalletDevice(Base):
     id: Mapped[PrimaryKey]
     name: Mapped[str]
     wallet_id: Mapped[UUID] = mapped_column(ForeignKey("myeclpay_wallet.id"))
-    ed25519_public_key: Mapped[bytes]
+    ed25519_public_key: Mapped[bytes] = mapped_column(unique=True)
     creation: Mapped[datetime]
     status: Mapped[WalletDeviceStatus]
     activation_token: Mapped[str] = mapped_column(unique=True)
@@ -87,7 +87,10 @@ class Refund(Base):
     __tablename__ = "myeclpay_refund"
 
     id: Mapped[PrimaryKey]
-    transaction_id: Mapped[UUID] = mapped_column(ForeignKey("myeclpay_transaction.id"))
+    transaction_id: Mapped[UUID] = mapped_column(
+        ForeignKey("myeclpay_transaction.id"),
+        unique=True,
+    )
     debited_wallet_id: Mapped[UUID] = mapped_column(ForeignKey("myeclpay_wallet.id"))
     credited_wallet_id: Mapped[UUID] = mapped_column(ForeignKey("myeclpay_wallet.id"))
     total: Mapped[int]  # Stored in cents
@@ -137,7 +140,7 @@ class StructureManagerTransfert(Base):
         primary_key=True,
     )
     user_id: Mapped[str] = mapped_column(ForeignKey("core_user.id"))
-    confirmation_token: Mapped[str]
+    confirmation_token: Mapped[str] = mapped_column(unique=True)
     valid_until: Mapped[datetime]
 
 
@@ -170,6 +173,7 @@ class Request(Base):
     status: Mapped[RequestStatus]
     transaction_id: Mapped[UUID | None] = mapped_column(
         ForeignKey("myeclpay_transaction.id"),
+        unique=True,
     )
 
 

migrations/versions/33-MyECLPay-unique-constraints.py

@@ -0,0 +1,79 @@
+"""empty message
+
+Create Date: 2025-05-29 17:32:45.619972
+"""
+
+from collections.abc import Sequence
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from pytest_alembic import MigrationContext
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "ea26eebe3f8d"
+down_revision: str | None = "e16b58cc6084"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    op.create_unique_constraint(
+        "myeclpay_refund_transaction_id_key",
+        "myeclpay_refund",
+        ["transaction_id"],
+    )
+    op.create_unique_constraint(
+        "myeclpay_request_transaction_id_key",
+        "myeclpay_request",
+        ["transaction_id"],
+    )
+    op.create_unique_constraint(
+        "myeclpay_structure_manager_transfer_confirmation_token_key",
+        "myeclpay_structure_manager_transfer",
+        ["confirmation_token"],
+    )
+    op.create_unique_constraint(
+        "myeclpay_wallet_device_ed25519_public_key_key",
+        "myeclpay_wallet_device",
+        ["ed25519_public_key"],
+    )
+
+
+def downgrade() -> None:
+    op.drop_constraint(
+        constraint_name="myeclpay_refund_transaction_id_key",
+        table_name="myeclpay_refund",
+        type_="unique",
+    )
+    op.drop_constraint(
+        constraint_name="myeclpay_request_transaction_id_key",
+        table_name="myeclpay_request",
+        type_="unique",
+    )
+    op.drop_constraint(
+        constraint_name="myeclpay_structure_manager_transfer_confirmation_token_key",
+        table_name="myeclpay_structure_manager_transfer",
+        type_="unique",
+    )
+    op.drop_constraint(
+        constraint_name="myeclpay_wallet_device_ed25519_public_key_key",
+        table_name="myeclpay_wallet_device",
+        type_="unique",
+    )
+
+
+def pre_test_upgrade(
+    alembic_runner: "MigrationContext",
+    alembic_connection: sa.Connection,
+) -> None:
+    pass
+
+
+def test_upgrade(
+    alembic_runner: "MigrationContext",
+    alembic_connection: sa.Connection,
+) -> None:
+    pass

tests/test_myeclpay.py

@@ -1518,13 +1518,16 @@ async def test_create_and_activate_user_device(
         return_value=UNIQUE_TOKEN,
     )
 
+    private_key = Ed25519PrivateKey.generate()
+    public_key = private_key.public_key()
+
     response = client.post(
         "/myeclpay/users/me/wallet/devices",
         headers={"Authorization": f"Bearer {ecl_user_access_token}"},
         json={
             "name": "MySuperDevice",
             "ed25519_public_key": base64.b64encode(
-                ecl_user_wallet_device_public_key.public_bytes(
+                public_key.public_bytes(
                     encoding=serialization.Encoding.Raw,
                     format=serialization.PublicFormat.Raw,
                 ),
@@ -1540,12 +1543,9 @@ async def test_create_and_activate_user_device(
             response.json()["id"],
         )
         assert wallet_device is not None
-        assert (
-            wallet_device.ed25519_public_key
-            == ecl_user_wallet_device_public_key.public_bytes(
-                encoding=serialization.Encoding.Raw,
-                format=serialization.PublicFormat.Raw,
-            )
+        assert wallet_device.ed25519_public_key == public_key.public_bytes(
+            encoding=serialization.Encoding.Raw,
+            format=serialization.PublicFormat.Raw,
         )
 
     response = client.get(
@@ -1628,7 +1628,7 @@ async def test_revoke_user_device(
         id=uuid4(),
         name="Will revoke device",
         wallet_id=ecl_user_wallet.id,
-        ed25519_public_key=b"key",
+        ed25519_public_key=b"keytest_revoke_user_device",
         creation=datetime.now(UTC),
         status=WalletDeviceStatus.ACTIVE,
         activation_token="will_revoke_activation_token",