From 4cc9aef61f5821598bf61135d5eced340398013a Mon Sep 17 00:00:00 2001 From: 9qeklajc Date: Thu, 30 Jul 2026 02:57:49 +0200 Subject: [PATCH] fix: make payouts and proxy sessions safe --- routstr/auth.py | 24 ++ routstr/core/db.py | 6 + routstr/lightning.py | 12 +- routstr/proxy.py | 12 + routstr/wallet.py | 210 +++++++++++------- .../test_periodic_payout_safety.py | 174 +++++++++++++++ .../test_proxy_session_lifecycle.py | 65 ++++++ tests/unit/test_periodic_payout.py | 6 +- 8 files changed, 421 insertions(+), 88 deletions(-) create mode 100644 tests/integration/test_periodic_payout_safety.py create mode 100644 tests/integration/test_proxy_session_lifecycle.py diff --git a/routstr/auth.py b/routstr/auth.py index 42c7eed2..74165920 100644 --- a/routstr/auth.py +++ b/routstr/auth.py @@ -33,6 +33,7 @@ from .wallet import ( classify_redemption_error, credit_balance, deserialize_token_from_string, + wallet_operation_guard, ) if TYPE_CHECKING: @@ -153,6 +154,29 @@ async def validate_bearer_key( refund_address: Optional[str] = None, key_expiry_time: Optional[int] = None, min_cost: int = 0, +) -> ApiKey: + if bearer_key.startswith("cashu"): + # Acquire before the first lookup/flush so concurrent token creation + # cannot hold SQLite write transactions while waiting to mutate proofs. + async with wallet_operation_guard(): + return await _validate_bearer_key_locked( + bearer_key, + session, + refund_address, + key_expiry_time, + min_cost, + ) + return await _validate_bearer_key_locked( + bearer_key, session, refund_address, key_expiry_time, min_cost + ) + + +async def _validate_bearer_key_locked( + bearer_key: str, + session: AsyncSession, + refund_address: Optional[str] = None, + key_expiry_time: Optional[int] = None, + min_cost: int = 0, ) -> ApiKey: """ Validates the provided API key using SQLModel. diff --git a/routstr/core/db.py b/routstr/core/db.py index 12dcc659..1fc70400 100644 --- a/routstr/core/db.py +++ b/routstr/core/db.py @@ -784,6 +784,12 @@ async def complete_routstr_fee_payout( return result.rowcount == 1 +async def total_user_liability(db_session: AsyncSession) -> int: + """Return all outstanding API-key balances in millisatoshis.""" + result = await db_session.exec(select(func.sum(ApiKey.balance))) + return int(result.one() or 0) + + async def balance_for_mint_and_unit( db_session: AsyncSession, mint_url: str, unit: str ) -> int: diff --git a/routstr/lightning.py b/routstr/lightning.py index d696403a..d75a8c2c 100644 --- a/routstr/lightning.py +++ b/routstr/lightning.py @@ -12,7 +12,7 @@ from sqlmodel.ext.asyncio.session import AsyncSession from .core.db import ApiKey, LightningInvoice, create_session, get_session from .core.logging import get_logger from .core.settings import settings -from .wallet import get_wallet +from .wallet import get_wallet, wallet_operation_guard logger = get_logger(__name__) @@ -222,6 +222,16 @@ async def recover_invoice( async def check_invoice_payment( invoice: LightningInvoice, session: AsyncSession +) -> None: + # Minting makes proofs visible before database finalization. Share the + # cross-process wallet guard with owner payout so that visibility and the + # corresponding liability commit are observed atomically by the payout loop. + async with wallet_operation_guard(): + await _check_invoice_payment_locked(invoice, session) + + +async def _check_invoice_payment_locked( + invoice: LightningInvoice, session: AsyncSession ) -> None: minted = False invoice_id = invoice.id diff --git a/routstr/proxy.py b/routstr/proxy.py index 46596077..887eb38c 100644 --- a/routstr/proxy.py +++ b/routstr/proxy.py @@ -50,6 +50,13 @@ _provider_map: dict[ _unique_models: dict[str, Model] = {} # Unique model.id -> Model (no duplicates) +async def _finish_read_transaction(session: AsyncSession) -> None: + """Release a read transaction without assuming a particular session mock.""" + commit_result = session.commit() + if inspect.isawaitable(commit_result): + await commit_result + + async def initialize_upstreams() -> None: """Initialize upstream providers from database during application startup.""" global _upstreams @@ -465,6 +472,9 @@ async def _proxy( if is_ehbp or request_body_dict: await pay_for_request(key, max_cost_for_model, session) reservation_snapshot = await get_reservation_snapshot(key, session) + # Snapshot validation performs SELECTs after pay_for_request commits. + # End that read transaction before waiting on upstream response headers. + await _finish_read_transaction(session) # Tracks request params already removed in response to upstream rejections, # shared across providers so a stripped param stays stripped on failover and @@ -496,8 +506,10 @@ async def _proxy( raise await pay_for_request(key, max_cost_for_model, session) reservation_snapshot = await get_reservation_snapshot(key, session) + await _finish_read_transaction(session) continue reservation_snapshot = await get_reservation_snapshot(key, session) + await _finish_read_transaction(session) max_cost_for_model = candidate_max headers = upstream.prepare_headers(dict(request.headers)) diff --git a/routstr/wallet.py b/routstr/wallet.py index aed99f92..c6859bf2 100644 --- a/routstr/wallet.py +++ b/routstr/wallet.py @@ -1,9 +1,14 @@ import asyncio +import fcntl +import os import re import socket import time import typing -from typing import TypedDict +from contextlib import asynccontextmanager +from contextvars import ContextVar +from pathlib import Path +from typing import AsyncGenerator, TypedDict import httpx from cashu.core.base import Proof, Token @@ -32,6 +37,46 @@ _CashuMintInfo.model_rebuild(force=True) logger = get_logger(__name__) +# Preserve the real scheduler yield even when payout tests patch asyncio.sleep. +_scheduler_sleep = asyncio.sleep +_WALLET_OPERATION_LOCK = Path(".wallet") / ".routstr-operation.lock" +_wallet_operation_depth: ContextVar[int] = ContextVar( + "wallet_operation_depth", default=0 +) + + +@asynccontextmanager +async def wallet_operation_guard() -> AsyncGenerator[None, None]: + """Serialize proof mutation and owner payout across local worker processes.""" + depth = _wallet_operation_depth.get() + if depth: + token = _wallet_operation_depth.set(depth + 1) + try: + yield + finally: + _wallet_operation_depth.reset(token) + return + + _WALLET_OPERATION_LOCK.parent.mkdir(parents=True, exist_ok=True) + fd = os.open(_WALLET_OPERATION_LOCK, os.O_CREAT | os.O_RDWR, 0o600) + acquired = False + depth_token = None + try: + while not acquired: + try: + fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB) + acquired = True + except BlockingIOError: + await _scheduler_sleep(0.05) + depth_token = _wallet_operation_depth.set(1) + yield + finally: + if depth_token is not None: + _wallet_operation_depth.reset(depth_token) + if acquired: + fcntl.flock(fd, fcntl.LOCK_UN) + os.close(fd) + def _sats_to_msats(amount: int) -> int: return amount * 1000 @@ -686,6 +731,13 @@ async def swap_to_primary_mint( async def credit_balance( cashu_token: str, key: db.ApiKey, session: db.AsyncSession +) -> int: + async with wallet_operation_guard(): + return await _credit_balance_locked(cashu_token, key, session) + + +async def _credit_balance_locked( + cashu_token: str, key: db.ApiKey, session: db.AsyncSession ) -> int: logger.info( "credit_balance: Starting token redemption", @@ -747,6 +799,9 @@ async def credit_balance( ) await session.commit() await session.refresh(key) + # refresh() starts a read transaction; release it before the + # transaction-history write opens its own session below. + await session.commit() except TokenConsumedError: raise except Exception as db_error: @@ -945,6 +1000,70 @@ async def fetch_all_balances( ) +async def _payout_mint_and_unit(mint_url: str, unit: str) -> None: + """Send only conservatively proven owner funds for one wallet.""" + try: + wallet = await get_wallet(mint_url, unit) + proofs = get_proofs_per_mint_and_unit( + wallet, mint_url, unit, not_reserved=True + ) + proofs = await slow_filter_spend_proofs(proofs, wallet) + await asyncio.sleep(5) + except Exception as e: + logger.error( + f"Error sending payout: {type(e).__name__}", + extra={"error": str(e), "mint_url": mint_url, "unit": unit}, + ) + return + + try: + async with db.create_session() as session: + # ApiKey stores a refund preference, not funding provenance. Until + # liabilities have a durable per-credit ledger, subtract the total + # liability from every wallet rather than risk calling customer + # funds owner profit on the wrong mint. + user_balance = await db.total_user_liability(session) + except Exception as e: + logger.error( + f"Error in periodic payout cycle: {type(e).__name__}", + extra={"error": str(e), "mint_url": mint_url, "unit": unit}, + ) + return + + try: + if unit == "sat": + user_balance = _msats_to_sats(user_balance) + proofs_balance = sum(proof.amount for proof in proofs) + available_balance = proofs_balance - user_balance + min_amount = ( + settings.min_payout_sat + if unit == "sat" + else _sats_to_msats(settings.min_payout_sat) + ) + if available_balance > min_amount: + amount_received = await raw_send_to_lnurl( + wallet, + proofs, + settings.receive_ln_address, + unit, + amount=available_balance, + ) + logger.info( + "Payout sent successfully", + extra={ + "mint_url": mint_url, + "unit": unit, + "balance": available_balance, + "amount_received": amount_received, + }, + ) + except Exception as e: + logger.error( + f"Error sending payout: {type(e).__name__}", + extra={"error": str(e), "mint_url": mint_url, "unit": unit}, + ) + + async def periodic_payout() -> None: while True: await asyncio.sleep(settings.payout_interval_seconds) @@ -952,90 +1071,13 @@ async def periodic_payout() -> None: if not settings.receive_ln_address: continue - # Include the primary mint even if it is not listed in cashu_mints, - # matching fetch_all_balances(); otherwise primary-mint funds never - # auto-payout. - mint_urls = _mints_to_inspect() - - for mint_url in mint_urls: + for mint_url in _mints_to_inspect(): for unit in ["sat", "msat"]: - # Isolate failures per mint/unit so one slow or failing - # mint does not abort payout for every other mint/unit. - try: - wallet = await get_wallet(mint_url, unit) - proofs = get_proofs_per_mint_and_unit( - wallet, mint_url, unit, not_reserved=True - ) - proofs = await slow_filter_spend_proofs(proofs, wallet) - await asyncio.sleep(5) - except Exception as e: - logger.error( - f"Error sending payout: {type(e).__name__}", - extra={ - "error": str(e), - "mint_url": mint_url, - "unit": unit, - }, - ) - continue - - # Fetch the liability AFTER the proofs snapshot and the - # settle delay: a concurrent top-up then only inflates the - # liability, shrinking the payout — never sending - # customer-backed funds as profit. - try: - async with db.create_session() as session: - user_balance = await db.balance_for_mint_and_unit( - session, mint_url, unit - ) - except Exception as e: - logger.error( - f"Error in periodic payout cycle: {type(e).__name__}", - extra={ - "error": str(e), - "mint_url": mint_url, - "unit": unit, - }, - ) - continue - - try: - if unit == "sat": - user_balance = _msats_to_sats(user_balance) - proofs_balance = sum(proof.amount for proof in proofs) - available_balance = proofs_balance - user_balance - # Threshold is configured in sats; convert for msat wallets. - min_amount = ( - settings.min_payout_sat - if unit == "sat" - else _sats_to_msats(settings.min_payout_sat) - ) - if available_balance > min_amount: - amount_received = await raw_send_to_lnurl( - wallet, - proofs, - settings.receive_ln_address, - unit, - amount=available_balance, - ) - logger.info( - "Payout sent successfully", - extra={ - "mint_url": mint_url, - "unit": unit, - "balance": available_balance, - "amount_received": amount_received, - }, - ) - except Exception as e: - logger.error( - f"Error sending payout: {type(e).__name__}", - extra={ - "error": str(e), - "mint_url": mint_url, - "unit": unit, - }, - ) + # Proof mutation, liability observation, and sending are one + # cross-process critical section. A credit takes the same + # lock from before redemption through its liability commit. + async with wallet_operation_guard(): + await _payout_mint_and_unit(mint_url, unit) except Exception as e: logger.error( f"Error in periodic payout cycle: {type(e).__name__}", diff --git a/tests/integration/test_periodic_payout_safety.py b/tests/integration/test_periodic_payout_safety.py new file mode 100644 index 00000000..0f2769fc --- /dev/null +++ b/tests/integration/test_periodic_payout_safety.py @@ -0,0 +1,174 @@ +"""Money-safety regression coverage for automatic wallet payouts.""" + +from __future__ import annotations + +import asyncio +from collections.abc import Callable, Coroutine +from typing import Any +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +from sqlalchemy.ext.asyncio import AsyncEngine +from sqlmodel.ext.asyncio.session import AsyncSession + +from routstr.core import db +from routstr.core.db import ApiKey +from routstr.core.settings import settings +from routstr.wallet import credit_balance, periodic_payout + +PRIMARY_MINT = "http://primary:3338" +REFUND_MINT = "http://refund:3338" +PAYOUT_INTERVAL = 987 + + +class _LoopBreak(Exception): + """Stop the otherwise-infinite payout loop after one cycle.""" + + +def _one_payout_cycle() -> Callable[[float], Coroutine[Any, Any, None]]: + intervals_seen = 0 + + async def sleep(seconds: float) -> None: + nonlocal intervals_seen + if seconds == PAYOUT_INTERVAL: + intervals_seen += 1 + if intervals_seen == 2: + raise _LoopBreak() + + return sleep + + +@pytest.mark.asyncio +async def test_cross_mint_liability_is_not_paid_as_owner_profit( + integration_engine: AsyncEngine, + patched_db_engine: None, +) -> None: + """Refund preferences must not make primary-mint customer funds payable.""" + async with AsyncSession(integration_engine, expire_on_commit=False) as setup: + setup.add( + ApiKey( + hashed_key="cross-mint-key", + balance=50_000, + refund_mint_url=REFUND_MINT, + refund_currency="sat", + ) + ) + await setup.commit() + + primary_proof = MagicMock(amount=50) + raw_send = AsyncMock(return_value=50) + + def proofs_for_mint( + _wallet: object, mint_url: str, unit: str, **_kwargs: object + ) -> list[MagicMock]: + if mint_url == PRIMARY_MINT and unit == "sat": + return [primary_proof] + return [] + + with ( + patch.object(settings, "cashu_mints", [REFUND_MINT]), + patch.object(settings, "primary_mint", PRIMARY_MINT), + patch.object(settings, "receive_ln_address", "owner@ln.test"), + patch.object(settings, "payout_interval_seconds", PAYOUT_INTERVAL), + patch.object(settings, "min_payout_sat", 10), + patch("routstr.wallet.asyncio.sleep", _one_payout_cycle()), + patch("routstr.wallet.get_wallet", AsyncMock(return_value=MagicMock())), + patch( + "routstr.wallet.get_proofs_per_mint_and_unit", + MagicMock(side_effect=proofs_for_mint), + ), + patch( + "routstr.wallet.slow_filter_spend_proofs", + AsyncMock(side_effect=lambda proofs, _wallet: proofs), + ), + patch("routstr.wallet.raw_send_to_lnurl", raw_send), + ): + with pytest.raises(_LoopBreak): + await periodic_payout() + + raw_send.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_payout_does_not_send_proofs_whose_liability_commit_is_in_flight( + integration_engine: AsyncEngine, + patched_db_engine: None, +) -> None: + """Proof visibility before liability commit must not expose customer funds.""" + key = ApiKey( + hashed_key="in-flight-topup-key", + balance=0, + refund_mint_url=PRIMARY_MINT, + refund_currency="sat", + ) + async with AsyncSession(integration_engine, expire_on_commit=False) as setup: + setup.add(key) + await setup.commit() + + proofs: list[MagicMock] = [] + proof_visible = asyncio.Event() + finish_redemption = asyncio.Event() + liability_read = asyncio.Event() + + async def redeem_token(token: str) -> tuple[int, str, str]: + proofs.append(MagicMock(amount=200)) + proof_visible.set() + await finish_redemption.wait() + return 200, "sat", PRIMARY_MINT + + real_total_liability = db.total_user_liability + + async def read_liability(_session: AsyncSession) -> int: + async with db.create_session() as snapshot_session: + value = await real_total_liability(snapshot_session) + liability_read.set() + return value + + raw_send = AsyncMock(return_value=200) + + with ( + patch.object(settings, "cashu_mints", []), + patch.object(settings, "primary_mint", PRIMARY_MINT), + patch.object(settings, "receive_ln_address", "owner@ln.test"), + patch.object(settings, "payout_interval_seconds", PAYOUT_INTERVAL), + patch.object(settings, "min_payout_sat", 10), + patch("routstr.wallet.asyncio.sleep", _one_payout_cycle()), + patch("routstr.wallet.recieve_token", AsyncMock(side_effect=redeem_token)), + patch("routstr.wallet.get_wallet", AsyncMock(return_value=MagicMock())), + patch( + "routstr.wallet.get_proofs_per_mint_and_unit", + MagicMock(side_effect=lambda *_args, **_kwargs: list(proofs)), + ), + patch( + "routstr.wallet.slow_filter_spend_proofs", + AsyncMock(side_effect=lambda visible, _wallet: visible), + ), + patch( + "routstr.wallet.db.total_user_liability", + AsyncMock(side_effect=read_liability), + ), + patch("routstr.wallet.raw_send_to_lnurl", raw_send), + ): + async with AsyncSession(integration_engine, expire_on_commit=False) as credit_session: + stored_key = await credit_session.get(ApiKey, key.hashed_key) + assert stored_key is not None + credit_task = asyncio.create_task( + credit_balance("cashu-token", stored_key, credit_session) + ) + await asyncio.wait_for(proof_visible.wait(), timeout=2) + + payout_task = asyncio.create_task(periodic_payout()) + try: + await asyncio.wait_for(liability_read.wait(), timeout=0.1) + liability_was_read_while_crediting = True + except TimeoutError: + liability_was_read_while_crediting = False + + finish_redemption.set() + await asyncio.wait_for(credit_task, timeout=2) + + with pytest.raises(_LoopBreak): + await asyncio.wait_for(payout_task, timeout=2) + + assert liability_was_read_while_crediting is False + raw_send.assert_not_awaited() diff --git a/tests/integration/test_proxy_session_lifecycle.py b/tests/integration/test_proxy_session_lifecycle.py new file mode 100644 index 00000000..9a70f294 --- /dev/null +++ b/tests/integration/test_proxy_session_lifecycle.py @@ -0,0 +1,65 @@ +"""Integration coverage for proxy database-session lifetime.""" + +from __future__ import annotations + +import json +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +from fastapi import Response +from sqlalchemy.ext.asyncio import AsyncEngine +from sqlmodel.ext.asyncio.session import AsyncSession + +from routstr import proxy as proxy_module +from routstr.core.db import ApiKey + + +@pytest.mark.asyncio +async def test_authenticated_proxy_releases_db_connection_before_upstream_headers( + integration_engine: AsyncEngine, + integration_session: AsyncSession, + patched_db_engine: None, +) -> None: + """Slow upstream header waits must not retain a checked-out DB connection.""" + key = ApiKey( + hashed_key="proxy-pool-key", + balance=1_000_000, + refund_mint_url="http://primary:3338", + refund_currency="sat", + ) + integration_session.add(key) + await integration_session.commit() + + request = MagicMock() + request.method = "POST" + request.headers = {"authorization": "Bearer test-key"} + request.body = AsyncMock(return_value=json.dumps({"model": "test-model"}).encode()) + request.url.path = "/v1/chat/completions" + request.state.request_id = "pool-hold-regression" + + model = MagicMock() + upstream = MagicMock() + upstream.provider_type = "test" + upstream.prepare_headers.return_value = {} + + async def wait_for_headers(*args: object, **kwargs: object) -> Response: + assert integration_engine.pool.checkedout() == 0 # type: ignore[attr-defined] + return Response(status_code=200) + + upstream.forward_request = AsyncMock(side_effect=wait_for_headers) + + with ( + patch("routstr.proxy.get_candidates", return_value=[(model, upstream)]), + patch("routstr.proxy.get_max_cost_for_model", AsyncMock(return_value=100)), + patch( + "routstr.proxy.calculate_discounted_max_cost", + AsyncMock(return_value=100), + ), + patch("routstr.proxy.check_token_balance"), + patch("routstr.proxy.get_bearer_token_key", AsyncMock(return_value=key)), + ): + response = await proxy_module._proxy( + request, "v1/chat/completions", integration_session + ) + + assert response.status_code == 200 diff --git a/tests/unit/test_periodic_payout.py b/tests/unit/test_periodic_payout.py index 953f06cb..2b4a29fd 100644 --- a/tests/unit/test_periodic_payout.py +++ b/tests/unit/test_periodic_payout.py @@ -77,7 +77,7 @@ async def test_periodic_payout_includes_primary_mint_not_in_cashu_mints() -> Non AsyncMock(side_effect=lambda proofs, wallet: proofs), ), patch( - "routstr.wallet.db.balance_for_mint_and_unit", + "routstr.wallet.db.total_user_liability", AsyncMock(return_value=0), ), patch("routstr.wallet.raw_send_to_lnurl", raw_send), @@ -131,7 +131,7 @@ async def test_periodic_payout_releases_session_before_slow_mint_send() -> None: AsyncMock(side_effect=lambda proofs, wallet: proofs), ), patch( - "routstr.wallet.db.balance_for_mint_and_unit", + "routstr.wallet.db.total_user_liability", AsyncMock(return_value=0), ), patch("routstr.wallet.raw_send_to_lnurl", AsyncMock(side_effect=raw_send)), @@ -173,7 +173,7 @@ async def test_periodic_payout_isolates_failing_mint() -> None: AsyncMock(side_effect=lambda proofs, wallet: proofs), ), patch( - "routstr.wallet.db.balance_for_mint_and_unit", + "routstr.wallet.db.total_user_liability", AsyncMock(return_value=0), ), patch("routstr.wallet.raw_send_to_lnurl", raw_send),