fix(kyc): a verification session must not lift a control it does not own

app/services/kyc_service.py

@@ -17,16 +17,30 @@
 from uuid import uuid4
 
 import stripe
-from sqlalchemy import select
+from sqlalchemy import and_, or_, select, update as sa_update
 
 from ..core.time import utc_now
 from ..db.database import get_session_factory
 from ..db.models import KYCVerificationModel, WalletModel
 from ..core.config import get_settings
-from ..schemas.billing import KYCStatus
+from ..schemas.billing import KYCStatus, WalletStatus
 from .agent_money import WalletNotFoundError
 
 logger = logging.getLogger(__name__)
+
+# Statuses a KYC session must never overwrite: each was set by a control that
+# KYC does not own -- the velocity anomaly freeze, an operator suspension, or
+# wallet closure.
+_NON_SPENDABLE_WALLET_STATUSES = (
+    WalletStatus.FROZEN.value,
+    WalletStatus.SUSPENDED.value,
+    WalletStatus.CLOSED.value,
+)
+
+# ...with one exception: a suspension KYC itself imposed. handle_rejected and
+# handle_expired below suspend the wallet, so re-verifying is the documented
+# way out of that state and must stay reachable.
+_KYC_OWNED_SUSPENSION_REASONS = ("rejected", "expired")
 settings = get_settings()
 
 stripe.api_key = settings.STRIPE_SECRET_KEY
@@ -158,8 +172,47 @@ async def create_verification_session(
             )
             session.add(verification)
 
-            wallet.status = "pending_kyc"
-            session.add(wallet)
+            # Moving the wallet to pending_kyc must never WIDEN what it is
+            # allowed to do. pending_kyc is a spendable status, so writing it
+            # unconditionally turned a velocity auto-freeze, an operator
+            # suspension, or a closed wallet back into a spending one -- and
+            # the wallet was read before the Stripe network call above, so the
+            # write also clobbered anything committed while that call was in
+            # flight.
+            #
+            # A KYC session may only claim the status it is entitled to: one
+            # that is already spendable, or a suspension KYC itself imposed
+            # (a rejected or expired verification, set below). A freeze or a
+            # closure is not KYC's to lift; those wallets still get their
+            # session, and still get their kyc_status updated by the webhook,
+            # but stay exactly as non-spendable as they were.
+            claimed = await session.execute(
+                sa_update(WalletModel)
+                .where(
+                    cast(Any, WalletModel.wallet_id) == wallet_id,
+                    or_(
+                        cast(Any, WalletModel.status).notin_(
+                            _NON_SPENDABLE_WALLET_STATUSES
+                        ),
+                        and_(
+                            cast(Any, WalletModel.status)
+                            == WalletStatus.SUSPENDED.value,
+                            cast(Any, WalletModel.kyc_status).in_(
+                                _KYC_OWNED_SUSPENSION_REASONS
+                            ),
+                        ),
+                    ),
+                )
+                .values(status="pending_kyc", updated_at=utc_now())
+                .execution_options(synchronize_session=False)
+            )
+            if (cast(Any, claimed).rowcount or 0) != 1:
+                logger.info(
+                    "KYC session %s created for wallet %s without moving it to "
+                    "pending_kyc: its status is not KYC's to change",
+                    verification_id,
+                    wallet_id,
+                )
 
             await session.commit()
 

tests/test_kyc_status_authority.py

@@ -0,0 +1,171 @@
+"""A KYC session must not widen what a wallet is allowed to do.
+
+``create_verification_session`` moves a wallet to ``pending_kyc``, and
+``pending_kyc`` is a *spendable* status -- it is absent from the non-spendable
+set the debit paths check. Writing it unconditionally therefore let an ordinary
+API call undo three separate controls: the velocity anomaly auto-freeze, an
+operator suspension, and wallet closure. No race was required.
+
+The wallet is also read *before* the Stripe network call and written after, so
+the same write clobbered anything committed while that call was in flight.
+
+These tests pin both halves of the rule: KYC may claim a status that is already
+spendable, or a suspension KYC itself imposed, and nothing else.
+"""
+
+from __future__ import annotations
+
+import uuid
+from decimal import Decimal
+from unittest.mock import MagicMock, patch
+
+import pytest
+import pytest_asyncio
+
+from app.db.database import get_session_factory
+from app.db.models import WalletModel
+from app.services.kyc_service import KYCService
+
+pytestmark = pytest.mark.anyio
+
+
+def _stripe_session() -> MagicMock:
+    """A stand-in for the Stripe Identity session object."""
+    fake = MagicMock()
+    fake.id = f"vs_{uuid.uuid4().hex[:12]}"
+    fake.url = "https://verify.stripe.test/session"
+    return fake
+
+
+async def _seed_wallet(status: str, kyc_status: str = "pending") -> str:
+    """Persist a sponsor wallet in an exact status, returning its id."""
+    wallet_id = f"spn-kyc-{uuid.uuid4().hex[:12]}"
+    factory = get_session_factory()
+    async with factory() as session:
+        async with session.begin():
+            session.add(
+                WalletModel(
+                    wallet_id=wallet_id,
+                    wallet_type="sponsor",
+                    owner_name="KYC authority probe",
+                    email=f"{wallet_id}@example.com",
+                    balance=Decimal("100"),
+                    status=status,
+                    kyc_status=kyc_status,
+                )
+            )
+    return wallet_id
+
+
+async def _status(wallet_id: str) -> str:
+    factory = get_session_factory()
+    async with factory() as session:
+        wallet = await session.get(WalletModel, wallet_id)
+        assert wallet is not None
+        return wallet.status
+
+
+async def _start_kyc(wallet_id: str) -> None:
+    with patch(
+        "stripe.identity.VerificationSession.create",
+        return_value=_stripe_session(),
+    ):
+        await KYCService().create_verification_session(
+            wallet_id, "https://example.test/return"
+        )
+
+
+@pytest_asyncio.fixture(autouse=True)
+async def _clean(clean_database):
+    """Every case starts from an empty wallets table."""
+    yield
+
+
+@pytest.mark.parametrize(
+    ("status", "kyc_status"),
+    [
+        ("frozen", "pending"),
+        ("suspended", "pending"),
+        ("closed", "pending"),
+    ],
+)
+async def test_kyc_session_does_not_lift_a_control_it_does_not_own(
+    status: str, kyc_status: str
+) -> None:
+    """A frozen, operator-suspended, or closed wallet stays that way.
+
+    The freeze is the sharpest of the three: it is the anomaly control that
+    fires on a concurrent spend burst, and lifting it by starting KYC would
+    hand the burst its wallet back.
+    """
+    wallet_id = await _seed_wallet(status, kyc_status)
+
+    await _start_kyc(wallet_id)
+
+    assert await _status(wallet_id) == status
+
+
+@pytest.mark.parametrize("kyc_status", ["rejected", "expired"])
+async def test_kyc_session_may_reclaim_a_suspension_it_imposed(
+    kyc_status: str,
+) -> None:
+    """Re-verifying after a rejection or expiry must stay reachable.
+
+    ``handle_rejected`` and ``handle_expired`` suspend the wallet themselves,
+    so this suspension IS KYC's to lift -- and the success path only
+    reactivates from ``pending_kyc``. Refusing the transition here would strand
+    every rejected sponsor in a state they could never verify out of.
+    """
+    wallet_id = await _seed_wallet("suspended", kyc_status)
+
+    await _start_kyc(wallet_id)
+
+    assert await _status(wallet_id) == "pending_kyc"
+
+
+async def test_kyc_session_claims_a_spendable_wallet_normally() -> None:
+    """The ordinary path is unchanged."""
+    wallet_id = await _seed_wallet("active", "pending")
+
+    await _start_kyc(wallet_id)
+
+    assert await _status(wallet_id) == "pending_kyc"
+
+
+async def test_a_freeze_landing_during_the_stripe_call_survives() -> None:
+    """A freeze committed while Stripe is being called must not be clobbered.
+
+    The wallet is read before the network call and written after, so this is
+    the window the guarded write closes: the status is decided by the row at
+    write time, not by the copy this request read.
+    """
+    wallet_id = await _seed_wallet("active", "pending")
+
+    async def _freeze() -> None:
+        factory = get_session_factory()
+        async with factory() as session:
+            async with session.begin():
+                wallet = await session.get(WalletModel, wallet_id)
+                assert wallet is not None
+                wallet.status = "frozen"
+
+    freezing_session = _stripe_session()
+
+    class _FreezeOnCreate:
+        """Commit the freeze at the moment Stripe would be answering."""
+
+        def __call__(self, *args: object, **kwargs: object) -> MagicMock:
+            import anyio.from_thread
+
+            with anyio.from_thread.start_blocking_portal() as portal:
+                portal.call(_freeze)
+            return freezing_session
+
+    with patch(
+        "stripe.identity.VerificationSession.create", side_effect=_FreezeOnCreate()
+    ):
+        await KYCService().create_verification_session(
+            wallet_id, "https://example.test/return"
+        )
+
+    assert await _status(wallet_id) == "frozen"