Compare commits

...
Author SHA1 Message Date
9qeklajc ccab5e4216 collapse when focus change 2026-06-19 11:53:30 +02:00
9qeklajcandGitHub d4318dcc07 Merge pull request #556 from jeroenubbink/fix/trusted-mint-input-fee
fix: deduct mint NUT-02 input fee when crediting trusted-mint topups
2026-06-18 21:48:24 +02:00
Jeroen UbbinkandClaude Opus 4.8 75ed865a5f fix: deduct input fee in swap_to_primary_mint same-mint shortcut
The same-mint shortcut in swap_to_primary_mint did a same-mint
split(include_fees=True) — which burns the mint's NUT-02 per-proof input
fee — but returned the full token amount, over-crediting the user (the
same bug already fixed for the trusted-mint receive path). It also
skipped DLEQ verification that the trusted path performs.

Extract the shared same-mint redeem into _redeem_same_mint (load mint,
verify DLEQ, split, credit amount - input_fees) and delegate from both
recieve_token and the shortcut, so the two paths can't drift again. This
shortcut is reachable when PRIMARY_MINT_URL is set outside CASHU_MINTS.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-18 15:48:49 +02:00
Jeroen UbbinkandClaude Opus 4.8 7969a8da55 test: clarify mocked input-fee comment in trusted-mint test
The comment described "21 proofs @ 100 ppk" arithmetic, but the mocked
token has one proof and get_fees_for_proofs is hard-mocked to 3, so the
math wasn't exercised. Describe what the mock actually does.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-18 12:26:44 +02:00
9qeklajcandGitHub 830c299130 Merge pull request #557 from Routstr/rollback-token-generation-when-request-failed
rollback when token generation failed
2026-06-17 22:17:55 +02:00
9qeklajc e9dced8148 rollback when token generation failed 2026-06-17 17:27:08 +02:00
Jeroen UbbinkandClaude Opus 4.8 ecd46975b4 fix: deduct mint NUT-02 input fee when crediting trusted-mint topups
recieve_token swaps the incoming proofs at the same mint with
include_fees=True (paying the mint's NUT-02 per-proof input fee) but credited
the full face value. On every topup from a fee-charging trusted mint, routstr
over-credited the user by the fee and its own wallet drifted toward insolvency.

Subtract get_fees_for_proofs(proofs) from the credited amount, mirroring the
foreign-mint swap path which already accounts for it. Adds a fee-charging
trusted-mint unit test (credited == face - input_fee).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-17 17:10:37 +02:00
9qeklajcandGitHub bf05c96dfc Merge pull request #552 from Routstr/fix-reset-reserve-balance
Fix reset reserve balance
2026-06-12 21:52:39 +02:00
9qeklajc ed6e0c0189 Merge main into fix-reset-reserve-balance
Resolve pay_for_request conflict: keep main's atomic child-key
balance_limit guard and post-rowcount-check ordering, and stamp
reserved_at on both billing and child reservations.
2026-06-12 21:08:53 +02:00
9qeklajcandGitHub 412bfe479c Merge pull request #547 from jeroenubbink/fix/child-key-balance-limit-atomic
fix: enforce balance_limit atomically on child key requests
2026-06-12 20:23:25 +02:00
9qeklajc 861fda21b7 resolve review comments 2026-06-12 20:20:38 +02:00
9qeklajc 919dcf5535 make key reservation reset predictable 2026-06-12 19:45:39 +02:00
Jeroen UbbinkandClaude Sonnet 4.6 28c4008892 fix: enforce balance_limit atomically on child key requests
The Python pre-check on balance_limit provided no concurrency guarantee —
two concurrent requests could both pass the check on stale in-memory state
and both proceed to reserve, exceeding the limit.

Add the balance_limit guard to the child key UPDATE's WHERE clause so the
enforcement is atomic. The HTTP 402 is raised without an explicit rollback:
since session.commit() was never called, the uncommitted billing key update
is discarded when the session context manager closes on exception exit.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-06-09 11:37:39 +02:00
9qeklajcandGitHub a16bc1220c Merge pull request #546 from Routstr/fix-parent-key-leak-to-child-key
do not leak parent key to child key
2026-06-07 15:27:32 +02:00
9qeklajcandGitHub 5d3e687876 Merge pull request #545 from Routstr/secure-test-endpoint
secure endpoint
2026-06-07 15:25:53 +02:00
9qeklajc 68a7cd1dfb do not leak parent key to child key 2026-06-07 15:25:13 +02:00
14 changed files with 1015 additions and 48 deletions
@@ -0,0 +1,25 @@
"""add reserved_at to api_keys
Revision ID: b5e7c9d1f3a2
Revises: a2b3c4d5e6f7
Create Date: 2026-06-12 00:00:00.000000
"""
import sqlalchemy as sa
from alembic import op
# revision identifiers, used by Alembic.
revision = "b5e7c9d1f3a2"
down_revision = "a2b3c4d5e6f7"
branch_labels = None
depends_on = None
def upgrade() -> None:
# existing keys keep NULL
# New reservations populate it via pay_for_request.
op.add_column("api_keys", sa.Column("reserved_at", sa.Integer(), nullable=True))
def downgrade() -> None:
op.drop_column("api_keys", "reserved_at")
+70 -15
View File
@@ -364,6 +364,7 @@ async def validate_bearer_key(
except HTTPException:
raise
except Exception as e:
await session.rollback()
logger.error(
"Cashu token redemption failed",
extra={
@@ -534,31 +535,19 @@ async def pay_for_request(
)
# Charge the base cost for the request atomically to avoid race conditions
reserved_at_now = int(time.time())
stmt = (
update(ApiKey)
.where(col(ApiKey.hashed_key) == billing_key.hashed_key)
.where(col(ApiKey.balance) - col(ApiKey.reserved_balance) >= cost_per_request)
.values(
reserved_balance=col(ApiKey.reserved_balance) + cost_per_request,
reserved_at=reserved_at_now,
total_requests=col(ApiKey.total_requests) + 1,
)
)
result = await session.exec(stmt) # type: ignore[call-overload]
# Also increment total_requests and reserved_balance on the child key if it's different
if billing_key.hashed_key != key.hashed_key:
child_stmt = (
update(ApiKey)
.where(col(ApiKey.hashed_key) == key.hashed_key)
.values(
total_requests=col(ApiKey.total_requests) + 1,
reserved_balance=col(ApiKey.reserved_balance) + cost_per_request,
)
)
await session.exec(child_stmt) # type: ignore[call-overload]
await session.commit()
if result.rowcount == 0:
logger.error(
"Concurrent request depleted balance",
@@ -570,7 +559,6 @@ async def pay_for_request(
},
)
# Another concurrent request spent the balance first
raise HTTPException(
status_code=402,
detail={
@@ -582,6 +570,44 @@ async def pay_for_request(
},
)
# Also increment total_requests and reserved_balance on the child key if it's different.
# The balance_limit guard is enforced atomically here — the Python pre-check above
# is a fast-path rejection only and provides no concurrency guarantee.
if billing_key.hashed_key != key.hashed_key:
child_stmt = (
update(ApiKey)
.where(col(ApiKey.hashed_key) == key.hashed_key)
.where(
(col(ApiKey.balance_limit).is_(None))
| (
col(ApiKey.total_spent)
+ col(ApiKey.reserved_balance)
+ cost_per_request
<= col(ApiKey.balance_limit)
)
)
.values(
total_requests=col(ApiKey.total_requests) + 1,
reserved_balance=col(ApiKey.reserved_balance) + cost_per_request,
reserved_at=reserved_at_now,
)
)
child_result = await session.exec(child_stmt) # type: ignore[call-overload]
if child_result.rowcount == 0:
raise HTTPException(
status_code=402,
detail={
"error": {
"message": f"Balance limit exceeded: {key.balance_limit} mSats limit. {key.total_spent} already spent ({key.reserved_balance} reserved), {cost_per_request} required for this request.",
"type": "insufficient_quota",
"code": "balance_limit_exceeded",
}
},
)
await session.commit()
await session.refresh(billing_key)
if billing_key.hashed_key != key.hashed_key:
await session.refresh(key)
@@ -620,12 +646,19 @@ async def revert_pay_for_request(
False if the reservation was already released (prevents negative reserved_balance)."""
billing_key = await get_billing_key(key, session)
# Keep reserved_at while other reservations remain
cleared_reserved_at = case(
(col(ApiKey.reserved_balance) - cost_per_request > 0, col(ApiKey.reserved_at)),
else_=None,
)
stmt = (
update(ApiKey)
.where(col(ApiKey.hashed_key) == billing_key.hashed_key)
.where(col(ApiKey.reserved_balance) >= cost_per_request)
.values(
reserved_balance=col(ApiKey.reserved_balance) - cost_per_request,
reserved_at=cleared_reserved_at,
total_requests=col(ApiKey.total_requests) - 1,
)
)
@@ -641,6 +674,7 @@ async def revert_pay_for_request(
.values(
total_requests=col(ApiKey.total_requests) - 1,
reserved_balance=col(ApiKey.reserved_balance) - cost_per_request,
reserved_at=cleared_reserved_at,
)
)
await session.exec(child_stmt) # type: ignore[call-overload]
@@ -1263,3 +1297,24 @@ async def periodic_key_reset() -> None:
break
except Exception as e:
logger.error(f"Error in periodic_key_reset: {e}")
STALE_RESERVATION_SWEEP_INTERVAL_SECONDS: int = 60
async def periodic_stale_reservation_sweep() -> None:
"""Background task that releases reservations leaked by client disconnects,
crashes or abandoned streams.
"""
from .core.db import create_session, release_stale_reservations
while True:
try:
async with create_session() as session:
await release_stale_reservations(
session, settings.stale_reservation_timeout_seconds
)
except Exception:
logger.exception("Error in periodic_stale_reservation_sweep")
await asyncio.sleep(STALE_RESERVATION_SWEEP_INTERVAL_SECONDS)
+35 -7
View File
@@ -7,7 +7,7 @@ from typing import Annotated, NoReturn
from fastapi import APIRouter, Depends, Header, HTTPException
from fastapi.responses import JSONResponse
from pydantic import BaseModel
from sqlmodel import col, select, update
from sqlmodel import col, or_, select, update
from .auth import get_billing_key, validate_bearer_key
from .core.db import (
@@ -48,7 +48,6 @@ async def get_balance_info(key: ApiKey, session: AsyncSession) -> dict:
"balance": billing_key.total_balance,
"reserved": billing_key.reserved_balance,
"is_child": key.parent_key_hash is not None,
"parent_key": "sk-" + key.parent_key_hash if key.parent_key_hash else None,
"total_requests": key.total_requests,
"total_spent": key.total_spent,
"balance_limit": key.balance_limit,
@@ -56,7 +55,9 @@ async def get_balance_info(key: ApiKey, session: AsyncSession) -> dict:
"validity_date": key.validity_date,
}
if not key.parent_key_hash:
if key.parent_key_hash:
info["parent_key_preview"] = key.parent_key_hash[:8] + "..."
else:
# Fetch child keys if this is a parent key
statement = select(ApiKey).where(ApiKey.parent_key_hash == key.hashed_key)
results = await session.exec(statement)
@@ -312,9 +313,36 @@ async def refund_wallet_endpoint(
)
if key.reserved_balance > 0:
raise HTTPException(
status_code=400,
detail="Cannot refund key. There are ongoing requests for this api key.",
# Release the reservation if it is stale
cutoff = int(time.time()) - settings.stale_reservation_timeout_seconds
stale_release_stmt = (
update(ApiKey)
.where(col(ApiKey.hashed_key) == key.hashed_key)
.where(col(ApiKey.reserved_balance) > 0)
.where(
or_(
col(ApiKey.reserved_at).is_(None),
col(ApiKey.reserved_at) < cutoff,
)
)
.values(reserved_balance=0, reserved_at=None)
)
stale_result = await session.exec(stale_release_stmt) # type: ignore[call-overload]
await session.commit()
if stale_result.rowcount == 0:
raise HTTPException(
status_code=400,
detail="Cannot refund key. There are ongoing requests for this api key.",
)
await session.refresh(key)
logger.warning(
"refund_wallet_endpoint: released stale reservation before refund",
extra={
"hashed_key": key.hashed_key,
"stale_timeout_seconds": settings.stale_reservation_timeout_seconds,
},
)
remaining_balance_msats: int = key.total_balance
@@ -341,7 +369,7 @@ async def refund_wallet_endpoint(
.where(col(ApiKey.hashed_key) == key.hashed_key)
.where(col(ApiKey.balance) == pre_debit_balance)
.where(col(ApiKey.reserved_balance) == pre_debit_reserved)
.values(balance=0, reserved_balance=0)
.values(balance=0, reserved_balance=0, reserved_at=None)
)
debit_result = await session.exec(debit_stmt) # type: ignore[call-overload]
await session.commit()
+33 -1
View File
@@ -33,6 +33,14 @@ class ApiKey(SQLModel, table=True): # type: ignore
reserved_balance: int = Field(
default=0, description="Reserved balance in millisatoshis (msats)"
)
reserved_at: int | None = Field(
default=None,
description=(
"Unix timestamp of the most recent balance reservation. Used to "
"detect and release stale reservations (e.g. after client "
"disconnects). NULL when no reservation has been made yet."
),
)
refund_address: str | None = Field(
default=None,
description="Lightning address to refund remaining balance after key expires",
@@ -87,12 +95,36 @@ class ApiKey(SQLModel, table=True): # type: ignore
async def reset_all_reserved_balances(session: AsyncSession) -> None:
stmt = update(ApiKey).values(reserved_balance=0)
stmt = update(ApiKey).values(reserved_balance=0, reserved_at=None)
await session.exec(stmt) # type: ignore[call-overload]
await session.commit()
logger.info("Reset reserved balances on startup")
async def release_stale_reservations(
session: AsyncSession, max_age_seconds: int
) -> int:
"""Release reservations whose last reserve is older than max_age_seconds.
"""
cutoff = int(time.time()) - max_age_seconds
stmt = (
update(ApiKey)
.where(col(ApiKey.reserved_balance) > 0)
.where(col(ApiKey.reserved_at).is_not(None))
.where(col(ApiKey.reserved_at) < cutoff)
.values(reserved_balance=0, reserved_at=None)
)
result = await session.exec(stmt) # type: ignore[call-overload]
await session.commit()
released = int(result.rowcount or 0)
if released:
logger.warning(
"Released stale balance reservations",
extra={"released_keys": released, "max_age_seconds": max_age_seconds},
)
return released
class ModelRow(SQLModel, table=True): # type: ignore
__tablename__ = "models"
id: str = Field(primary_key=True)
+9 -1
View File
@@ -11,7 +11,7 @@ from starlette.exceptions import HTTPException
from starlette.responses import Response as StarletteResponse
from starlette.types import Scope
from ..auth import periodic_key_reset
from ..auth import periodic_key_reset, periodic_stale_reservation_sweep
from ..balance import balance_router, deprecated_wallet_router
from ..lightning import lightning_router, periodic_invoice_watcher
from ..nostr import (
@@ -54,6 +54,7 @@ async def lifespan(_: FastAPI) -> AsyncGenerator[None, None]:
models_refresh_task = None
model_maps_refresh_task = None
key_reset_task = None
stale_reservation_task = None
auto_topup_task = None
refund_sweep_task = None
routstr_fee_task = None
@@ -122,6 +123,9 @@ async def lifespan(_: FastAPI) -> AsyncGenerator[None, None]:
if global_settings.providers_refresh_interval_seconds > 0:
providers_task = asyncio.create_task(providers_cache_refresher())
key_reset_task = asyncio.create_task(periodic_key_reset())
stale_reservation_task = asyncio.create_task(
periodic_stale_reservation_sweep()
)
auto_topup_task = asyncio.create_task(periodic_auto_topup())
refund_sweep_task = asyncio.create_task(periodic_refund_sweep())
routstr_fee_task = asyncio.create_task(periodic_routstr_fee_payout())
@@ -159,6 +163,8 @@ async def lifespan(_: FastAPI) -> AsyncGenerator[None, None]:
model_maps_refresh_task.cancel()
if key_reset_task is not None:
key_reset_task.cancel()
if stale_reservation_task is not None:
stale_reservation_task.cancel()
if auto_topup_task is not None:
auto_topup_task.cancel()
if refund_sweep_task is not None:
@@ -188,6 +194,8 @@ async def lifespan(_: FastAPI) -> AsyncGenerator[None, None]:
tasks_to_wait.append(model_maps_refresh_task)
if key_reset_task is not None:
tasks_to_wait.append(key_reset_task)
if stale_reservation_task is not None:
tasks_to_wait.append(stale_reservation_task)
if auto_topup_task is not None:
tasks_to_wait.append(auto_topup_task)
if refund_sweep_task is not None:
+6
View File
@@ -67,6 +67,12 @@ class Settings(BaseSettings):
reset_reserved_balance_on_startup: bool = Field(
default=True, env="RESET_RESERVED_BALANCE_ON_STARTUP"
) # deactivate in horizontal scaling setups
# Reservations older than this are considered leaked (client disconnect,
# crash, abandoned stream) and released by the background sweeper and the
# refund endpoint.
stale_reservation_timeout_seconds: int = Field(
default=300, env="STALE_RESERVATION_TIMEOUT_SECONDS"
)
# Network
cors_origins: list[str] = Field(default_factory=lambda: ["*"], env="CORS_ORIGINS")
+16
View File
@@ -1,3 +1,4 @@
import asyncio
import json
from typing import Any
@@ -492,6 +493,21 @@ async def proxy(
return response
except asyncio.CancelledError:
logger.warning(
"Client disconnected mid-request, reverting reservation",
extra={
"path": path,
"model": model_id,
"key_hash": key.hashed_key[:8] + "...",
"max_cost_for_model": max_cost_for_model,
},
)
await asyncio.shield(
revert_pay_for_request(key, session, max_cost_for_model)
)
raise
except UpstreamError as e:
logger.warning(
"Upstream %s failed for model=%s: %s",
+25 -12
View File
@@ -35,6 +35,24 @@ async def get_balance(unit: str) -> int:
return wallet.available_balance.amount
async def _redeem_same_mint(
wallet: Wallet, token_obj: Token
) -> tuple[int, str, str]: # amount, unit, mint_url
"""Redeem proofs at their own issuing mint (no cross-mint swap).
split() re-mints the incoming proofs into fresh ones we own so the sender
can't double-spend them. With include_fees=True the mint deducts its NUT-02
per-proof input fee, so we end up holding only `amount - input_fees`. Credit
that, not the face value, or routstr over-credits the user and its wallet
drifts insolvent.
"""
await wallet.load_mint(keyset_id=token_obj.keysets[0])
wallet.verify_proofs_dleq(token_obj.proofs)
input_fees = wallet.get_fees_for_proofs(token_obj.proofs)
await wallet.split(proofs=token_obj.proofs, amount=0, include_fees=True)
return int(token_obj.amount) - input_fees, token_obj.unit, token_obj.mint
async def recieve_token(
token: str,
) -> tuple[int, str, str]: # amount, unit, mint_url
@@ -48,12 +66,7 @@ async def recieve_token(
if token_obj.mint not in settings.cashu_mints:
return await swap_to_primary_mint(token_obj, wallet)
await wallet.load_mint(keyset_id=token_obj.keysets[0])
wallet.verify_proofs_dleq(token_obj.proofs)
await wallet.split(proofs=token_obj.proofs, amount=0, include_fees=True)
return token_obj.amount, token_obj.unit, token_obj.mint
return await _redeem_same_mint(wallet, token_obj)
async def send(amount: int, unit: str, mint_url: str | None = None) -> tuple[int, str]:
@@ -213,10 +226,9 @@ async def swap_to_primary_mint(
amount_msat = token_amount
else:
raise ValueError("Invalid unit")
primary_wallet = await get_wallet(settings.primary_mint, settings.primary_mint_unit)
# If the token is already from the primary mint, we don't need to swap
# and we definitely don't want to calculate or pay fees.
# If the token is already from the primary mint, we don't need a cross-mint
# swap — redeem it same-mint. There's no melt/Lightning fee, but the mint's
# NUT-02 input fee still applies; _redeem_same_mint accounts for it.
if token_obj.mint == settings.primary_mint:
logger.info(
"swap_to_primary_mint: token already on primary mint, skipping swap",
@@ -226,8 +238,9 @@ async def swap_to_primary_mint(
"unit": token_obj.unit,
},
)
await token_wallet.split(proofs=token_obj.proofs, amount=0, include_fees=True)
return token_amount, token_obj.unit, token_obj.mint
return await _redeem_same_mint(token_wallet, token_obj)
primary_wallet = await get_wallet(settings.primary_mint, settings.primary_mint_unit)
minted_amount = await _calculate_swap_amount(
amount_msat,
+5
View File
@@ -69,9 +69,14 @@ async def test_wallet_info_child_key_no_child_keys(
info_response = await integration_client.get("/v1/wallet/info")
assert info_response.status_code == 200
info_data = info_response.json()
parent_key = authenticated_client._test_api_key # type: ignore[attr-defined]
parent_key_hash = parent_key.removeprefix("sk-")
assert info_data["is_child"] is True
assert "child_keys" not in info_data
assert "parent_key" not in info_data
assert info_data["parent_key_preview"] == parent_key_hash[:8] + "..."
assert info_data["parent_key_preview"] not in {parent_key, parent_key_hash}
@pytest.mark.integration
+267 -1
View File
@@ -1,12 +1,14 @@
import asyncio
import time
from datetime import datetime, timedelta
import pytest
from fastapi import HTTPException
from sqlmodel import select
from sqlmodel.ext.asyncio.session import AsyncSession
from routstr.auth import pay_for_request
from routstr.core.db import ApiKey
from routstr.core.db import ApiKey, create_session
@pytest.mark.asyncio
@@ -120,6 +122,270 @@ async def test_periodic_key_reset_job(integration_session: AsyncSession) -> None
assert key2.total_spent == 0
@pytest.mark.asyncio
async def test_balance_limit_enforced_atomically_under_concurrency(
patched_db_engine: None,
) -> None:
parent_hash = "parent_limit_atomic"
child_hash = "child_limit_atomic"
cost = 300
async with create_session() as session:
parent = ApiKey(hashed_key=parent_hash, balance=10000)
child = ApiKey(
hashed_key=child_hash,
balance=0,
parent_key_hash=parent_hash,
balance_limit=cost,
total_spent=0,
)
session.add(parent)
session.add(child)
await session.commit()
results: list[str] = []
async def attempt() -> None:
async with create_session() as session:
fresh_child = await session.get(ApiKey, child_hash)
assert fresh_child is not None
try:
await pay_for_request(fresh_child, cost, session)
results.append("success")
except HTTPException as exc:
assert exc.status_code == 402
results.append("blocked")
await asyncio.gather(attempt(), attempt())
assert sorted(results) == ["blocked", "success"], (
f"Expected exactly one success and one 402, got: {results}"
)
async with create_session() as session:
final_child = await session.get(ApiKey, child_hash)
assert final_child is not None
assert final_child.reserved_balance == cost, (
f"Child reserved_balance should equal one reservation, "
f"got {final_child.reserved_balance}"
)
@pytest.mark.asyncio
async def test_parallel_payments_with_parent_and_child_key(
patched_db_engine: None,
) -> None:
parent_hash = "parent_parallel_mixed"
child_hash = "child_parallel_mixed"
cost = 300
async with create_session() as session:
parent = ApiKey(hashed_key=parent_hash, balance=10000)
child = ApiKey(
hashed_key=child_hash,
balance=0,
parent_key_hash=parent_hash,
balance_limit=2 * cost,
)
session.add(parent)
session.add(child)
await session.commit()
async def attempt(key_hash: str) -> str:
async with create_session() as session:
fresh_key = await session.get(ApiKey, key_hash)
assert fresh_key is not None
try:
await pay_for_request(fresh_key, cost, session)
return "success"
except HTTPException as exc:
assert exc.status_code == 402
return "blocked"
results = await asyncio.gather(attempt(parent_hash), attempt(child_hash))
assert results == ["success", "success"], (
f"Both parent and child payments should succeed, got: {results}"
)
async with create_session() as session:
final_parent = await session.get(ApiKey, parent_hash)
final_child = await session.get(ApiKey, child_hash)
assert final_parent is not None
assert final_child is not None
# Both requests bill the parent; only the child request reserves on the child.
assert final_parent.reserved_balance == 2 * cost
assert final_parent.total_requests == 2
assert final_child.reserved_balance == cost
assert final_child.total_requests == 1
@pytest.mark.asyncio
async def test_balance_limit_with_existing_total_spent_under_concurrency(
patched_db_engine: None,
) -> None:
parent_hash = "parent_total_spent"
child_hash = "child_total_spent"
cost = 300
async with create_session() as session:
parent = ApiKey(hashed_key=parent_hash, balance=10000)
# 700 already spent against a 1000 limit: only one more 300 request fits.
child = ApiKey(
hashed_key=child_hash,
balance=0,
parent_key_hash=parent_hash,
balance_limit=1000,
total_spent=700,
)
session.add(parent)
session.add(child)
await session.commit()
results: list[str] = []
async def attempt() -> None:
async with create_session() as session:
fresh_child = await session.get(ApiKey, child_hash)
assert fresh_child is not None
try:
await pay_for_request(fresh_child, cost, session)
results.append("success")
except HTTPException as exc:
assert exc.status_code == 402
results.append("blocked")
await asyncio.gather(attempt(), attempt())
assert sorted(results) == ["blocked", "success"], (
f"Expected exactly one success and one 402, got: {results}"
)
async with create_session() as session:
final_child = await session.get(ApiKey, child_hash)
assert final_child is not None
assert final_child.reserved_balance == cost
assert final_child.total_spent == 700
@pytest.mark.asyncio
async def test_balance_limit_with_existing_reserved_balance(
patched_db_engine: None,
) -> None:
parent_hash = "parent_reserved_set"
blocked_hash = "child_reserved_blocked"
allowed_hash = "child_reserved_allowed"
cost = 300
async with create_session() as session:
parent = ApiKey(hashed_key=parent_hash, balance=10000)
# 800 already reserved against a 1000 limit: another 300 must be rejected.
blocked_child = ApiKey(
hashed_key=blocked_hash,
balance=0,
parent_key_hash=parent_hash,
balance_limit=1000,
reserved_balance=800,
)
# 500 reserved against a 1000 limit: another 300 still fits.
allowed_child = ApiKey(
hashed_key=allowed_hash,
balance=0,
parent_key_hash=parent_hash,
balance_limit=1000,
reserved_balance=500,
)
session.add(parent)
session.add(blocked_child)
session.add(allowed_child)
await session.commit()
async with create_session() as session:
fresh_blocked = await session.get(ApiKey, blocked_hash)
assert fresh_blocked is not None
with pytest.raises(HTTPException) as exc_info:
await pay_for_request(fresh_blocked, cost, session)
assert exc_info.value.status_code == 402
async with create_session() as session:
fresh_allowed = await session.get(ApiKey, allowed_hash)
assert fresh_allowed is not None
await pay_for_request(fresh_allowed, cost, session)
async with create_session() as session:
final_blocked = await session.get(ApiKey, blocked_hash)
final_allowed = await session.get(ApiKey, allowed_hash)
final_parent = await session.get(ApiKey, parent_hash)
assert final_blocked is not None
assert final_allowed is not None
assert final_parent is not None
assert final_blocked.reserved_balance == 800, "Rejected request must not reserve"
assert final_blocked.total_requests == 0
assert final_allowed.reserved_balance == 500 + cost
assert final_allowed.total_requests == 1
# Only the allowed request should have billed the parent.
assert final_parent.reserved_balance == cost
assert final_parent.total_requests == 1
@pytest.mark.asyncio
async def test_child_reservation_discarded_when_parent_balance_depleted(
patched_db_engine: None,
) -> None:
parent_hash = "parent_depleted"
child_hash = "child_depleted"
cost = 300
async with create_session() as session:
# Parent can only afford one request; child has no balance_limit.
parent = ApiKey(hashed_key=parent_hash, balance=cost)
child = ApiKey(
hashed_key=child_hash,
balance=0,
parent_key_hash=parent_hash,
)
session.add(parent)
session.add(child)
await session.commit()
results: list[str] = []
async def attempt() -> None:
async with create_session() as session:
fresh_child = await session.get(ApiKey, child_hash)
assert fresh_child is not None
try:
await pay_for_request(fresh_child, cost, session)
results.append("success")
except HTTPException as exc:
assert exc.status_code == 402
results.append("blocked")
await asyncio.gather(attempt(), attempt())
assert sorted(results) == ["blocked", "success"], (
f"Expected exactly one success and one 402, got: {results}"
)
async with create_session() as session:
final_parent = await session.get(ApiKey, parent_hash)
final_child = await session.get(ApiKey, child_hash)
assert final_parent is not None
assert final_child is not None
assert final_parent.reserved_balance == cost
# The failed request must not leave a committed reservation on the child.
assert final_child.reserved_balance == cost, (
f"Child reserved_balance should reflect only the successful request, "
f"got {final_child.reserved_balance}"
)
assert final_child.total_requests == 1
@pytest.mark.asyncio
async def test_refund_does_not_delete_key(integration_session: AsyncSession) -> None:
# This requires mocking the router call or testing the logic in balance.py
+59
View File
@@ -0,0 +1,59 @@
import hashlib
from types import SimpleNamespace
from typing import AsyncGenerator
from unittest.mock import AsyncMock, patch
import pytest
from fastapi import HTTPException
from sqlalchemy.ext.asyncio import AsyncEngine, create_async_engine
from sqlalchemy.pool import StaticPool
from sqlmodel import SQLModel
from sqlmodel.ext.asyncio.session import AsyncSession
from routstr.auth import validate_bearer_key
from routstr.core.db import ApiKey
def _make_engine() -> AsyncEngine:
return create_async_engine(
"sqlite+aiosqlite://",
poolclass=StaticPool,
connect_args={"check_same_thread": False},
)
@pytest.fixture
async def session() -> AsyncGenerator[AsyncSession, None]:
engine = _make_engine()
async with engine.begin() as conn:
await conn.run_sync(SQLModel.metadata.create_all)
db_session = AsyncSession(engine, expire_on_commit=False)
try:
yield db_session
finally:
await db_session.close()
await engine.dispose()
@pytest.mark.asyncio
async def test_failed_first_cashu_redemption_rolls_back_empty_api_key(
session: AsyncSession,
) -> None:
token = "cashuAfirst_seen_but_redemption_fails"
hashed_key = hashlib.sha256(token.encode()).hexdigest()
token_obj = SimpleNamespace(mint="http://mint:3338", unit="sat")
from routstr.core.settings import settings
with (
patch.object(settings, "cashu_mints", ["http://mint:3338"]),
patch("routstr.auth.deserialize_token_from_string", return_value=token_obj),
patch(
"routstr.auth.credit_balance",
new=AsyncMock(side_effect=ValueError("token already spent")),
),
):
with pytest.raises(HTTPException):
await validate_bearer_key(token, session)
assert await session.get(ApiKey, hashed_key) is None
+378
View File
@@ -0,0 +1,378 @@
"""Tests for stale reserved_balance handling (issue #551).
Covers:
- pay_for_request stamping reserved_at on billing and child keys
- release_stale_reservations sweeper semantics
- reset_all_reserved_balances clearing reserved_at
- refund endpoint self-healing stale/legacy reservations
- proxy reverting the reservation when the client disconnects (CancelledError)
"""
import asyncio
import time
from typing import AsyncGenerator
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from sqlalchemy.ext.asyncio import AsyncEngine, create_async_engine
from sqlalchemy.pool import StaticPool
from sqlmodel import SQLModel
from sqlmodel.ext.asyncio.session import AsyncSession
from routstr.auth import pay_for_request
from routstr.balance import refund_wallet_endpoint
from routstr.core.db import (
ApiKey,
release_stale_reservations,
reset_all_reserved_balances,
)
def _make_engine() -> AsyncEngine:
return create_async_engine(
"sqlite+aiosqlite://",
poolclass=StaticPool,
connect_args={"check_same_thread": False},
)
@pytest.fixture
async def session() -> "AsyncGenerator[AsyncSession, None]":
engine = _make_engine()
async with engine.begin() as conn:
await conn.run_sync(SQLModel.metadata.create_all)
db_session = AsyncSession(engine, expire_on_commit=False)
try:
yield db_session
finally:
await db_session.close()
await engine.dispose()
# ---------------------------------------------------------------------------
# pay_for_request stamps reserved_at
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_pay_for_request_sets_reserved_at(session: AsyncSession) -> None:
key = ApiKey(hashed_key="paykey", balance=10_000)
session.add(key)
await session.commit()
before = int(time.time())
await pay_for_request(key, 1_000, session)
await session.refresh(key)
assert key.reserved_balance == 1_000
assert key.reserved_at is not None
assert key.reserved_at >= before
@pytest.mark.asyncio
async def test_pay_for_request_sets_reserved_at_on_child_key(session: AsyncSession) -> None:
parent = ApiKey(hashed_key="parentkey", balance=10_000)
child = ApiKey(hashed_key="childkey", balance=0, parent_key_hash="parentkey")
session.add(parent)
session.add(child)
await session.commit()
await pay_for_request(child, 1_000, session)
await session.refresh(parent)
await session.refresh(child)
assert parent.reserved_balance == 1_000
assert parent.reserved_at is not None
assert child.reserved_balance == 1_000
assert child.reserved_at is not None
@pytest.mark.asyncio
async def test_revert_clears_reserved_at_when_fully_released(
session: AsyncSession,
) -> None:
from routstr.auth import revert_pay_for_request
key = ApiKey(hashed_key="revertkey", balance=10_000)
session.add(key)
await session.commit()
await pay_for_request(key, 1_000, session)
reverted = await revert_pay_for_request(key, session, 1_000)
assert reverted is True
await session.refresh(key)
assert key.reserved_balance == 0
assert key.reserved_at is None
@pytest.mark.asyncio
async def test_revert_keeps_reserved_at_while_other_reservations_remain(
session: AsyncSession,
) -> None:
from routstr.auth import revert_pay_for_request
key = ApiKey(hashed_key="partialrevert", balance=10_000)
session.add(key)
await session.commit()
await pay_for_request(key, 1_000, session)
await pay_for_request(key, 1_000, session)
reverted = await revert_pay_for_request(key, session, 1_000)
assert reverted is True
await session.refresh(key)
assert key.reserved_balance == 1_000
assert key.reserved_at is not None
# ---------------------------------------------------------------------------
# release_stale_reservations sweeper
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_release_stale_reservations_releases_old(session: AsyncSession) -> None:
key = ApiKey(
hashed_key="stalekey",
balance=5_000,
reserved_balance=1_000,
reserved_at=int(time.time()) - 1_000,
)
session.add(key)
await session.commit()
released = await release_stale_reservations(session, max_age_seconds=300)
assert released == 1
await session.refresh(key)
assert key.reserved_balance == 0
assert key.reserved_at is None
@pytest.mark.asyncio
async def test_release_stale_reservations_keeps_fresh(session: AsyncSession) -> None:
key = ApiKey(
hashed_key="freshkey",
balance=5_000,
reserved_balance=1_000,
reserved_at=int(time.time()),
)
session.add(key)
await session.commit()
released = await release_stale_reservations(session, max_age_seconds=300)
assert released == 0
await session.refresh(key)
assert key.reserved_balance == 1_000
assert key.reserved_at is not None
@pytest.mark.asyncio
async def test_release_stale_reservations_skips_null_reserved_at(session: AsyncSession) -> None:
# Reservations without a timestamp may belong to instances running older
# code (rolling deploy) — the background sweeper must not touch them.
key = ApiKey(
hashed_key="legacykey",
balance=5_000,
reserved_balance=1_000,
reserved_at=None,
)
session.add(key)
await session.commit()
released = await release_stale_reservations(session, max_age_seconds=300)
assert released == 0
await session.refresh(key)
assert key.reserved_balance == 1_000
@pytest.mark.asyncio
async def test_reset_all_reserved_balances_clears_reserved_at(session: AsyncSession) -> None:
key = ApiKey(
hashed_key="resetkey",
balance=5_000,
reserved_balance=1_000,
reserved_at=int(time.time()),
)
session.add(key)
await session.commit()
await reset_all_reserved_balances(session)
await session.refresh(key)
assert key.reserved_balance == 0
assert key.reserved_at is None
# ---------------------------------------------------------------------------
# Refund endpoint self-healing
# ---------------------------------------------------------------------------
def _refund_patches(refund_token: str = "cashuArefund"): # type: ignore[no-untyped-def]
return (
patch("routstr.balance.send_token", AsyncMock(return_value=refund_token)),
patch("routstr.balance.store_cashu_transaction", AsyncMock()),
patch("routstr.balance._refund_cache_get", AsyncMock(return_value=None)),
patch("routstr.balance._refund_cache_set", AsyncMock()),
)
async def _add_key(session: AsyncSession, **kwargs) -> ApiKey: # type: ignore[no-untyped-def]
key = ApiKey(refund_currency="sat", **kwargs)
session.add(key)
await session.commit()
return key
@pytest.mark.asyncio
async def test_refund_self_heals_stale_reservation(session: AsyncSession) -> None:
key = await _add_key(
session,
hashed_key="stalerefund",
balance=5_000,
reserved_balance=2_000,
reserved_at=int(time.time()) - 10_000,
)
p1, p2, p3, p4 = _refund_patches()
with p1, p2, p3, p4:
result = await refund_wallet_endpoint(
authorization="Bearer sk-stalerefund",
x_cashu=None,
session=session,
)
assert isinstance(result, dict)
assert result["token"] == "cashuArefund"
# Full balance refunded (5000 msats -> 5 sats), reservation healed
assert result["sats"] == "5"
await session.refresh(key)
assert key.balance == 0
assert key.reserved_balance == 0
assert key.reserved_at is None
@pytest.mark.asyncio
async def test_refund_self_heals_legacy_null_reserved_at(session: AsyncSession) -> None:
# Keys stuck from before reserved_at existed must be refundable.
key = await _add_key(
session,
hashed_key="legacyrefund",
balance=5_000,
reserved_balance=2_000,
reserved_at=None,
)
p1, p2, p3, p4 = _refund_patches()
with p1, p2, p3, p4:
result = await refund_wallet_endpoint(
authorization="Bearer sk-legacyrefund",
x_cashu=None,
session=session,
)
assert isinstance(result, dict)
assert result["token"] == "cashuArefund"
await session.refresh(key)
assert key.balance == 0
assert key.reserved_balance == 0
@pytest.mark.asyncio
async def test_refund_rejects_recent_reservation(session: AsyncSession) -> None:
from fastapi import HTTPException
await _add_key(
session,
hashed_key="activerefund",
balance=5_000,
reserved_balance=2_000,
reserved_at=int(time.time()),
)
p1, p2, p3, p4 = _refund_patches()
with p1, p2, p3, p4:
with pytest.raises(HTTPException) as exc_info:
await refund_wallet_endpoint(
authorization="Bearer sk-activerefund",
x_cashu=None,
session=session,
)
assert exc_info.value.status_code == 400
assert "ongoing requests" in exc_info.value.detail
@pytest.mark.asyncio
async def test_refund_without_reservation_still_works(session: AsyncSession) -> None:
key = await _add_key(
session,
hashed_key="plainrefund",
balance=5_000,
reserved_balance=0,
)
p1, p2, p3, p4 = _refund_patches()
with p1, p2, p3, p4:
result = await refund_wallet_endpoint(
authorization="Bearer sk-plainrefund",
x_cashu=None,
session=session,
)
assert isinstance(result, dict)
assert result["token"] == "cashuArefund"
await session.refresh(key)
assert key.balance == 0
# ---------------------------------------------------------------------------
# Proxy reverts reservation on client disconnect
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_proxy_reverts_reservation_on_client_disconnect() -> None:
from routstr import proxy as proxy_module
key = ApiKey(hashed_key="cancelkey", balance=10_000)
request = MagicMock()
request.method = "POST"
request.headers = {"authorization": "Bearer sk-cancelkey"}
request.body = AsyncMock(return_value=b'{"model": "test-model"}')
upstream = MagicMock()
upstream.provider_type = "test"
upstream.prepare_headers = MagicMock(side_effect=lambda h: h)
upstream.forward_request = AsyncMock(side_effect=asyncio.CancelledError())
session = MagicMock()
revert_mock = AsyncMock(return_value=True)
with (
patch.object(proxy_module, "get_model_instance", return_value=MagicMock()),
patch.object(proxy_module, "get_provider_for_model", return_value=[upstream]),
patch.object(
proxy_module, "get_max_cost_for_model", AsyncMock(return_value=1_000)
),
patch.object(
proxy_module,
"calculate_discounted_max_cost",
AsyncMock(return_value=1_000),
),
patch.object(proxy_module, "check_token_balance", MagicMock()),
patch.object(
proxy_module, "get_bearer_token_key", AsyncMock(return_value=key)
),
patch.object(proxy_module, "pay_for_request", AsyncMock(return_value=1_000)),
patch.object(proxy_module, "revert_pay_for_request", revert_mock),
):
with pytest.raises(asyncio.CancelledError):
await proxy_module.proxy(request, "v1/chat/completions", session=session)
revert_mock.assert_awaited_once_with(key, session, 1_000)
+82 -3
View File
@@ -39,6 +39,8 @@ async def test_recieve_token_valid() -> None:
mock_wallet = Mock()
mock_wallet.split = AsyncMock()
# Fee-free trusted mint (e.g. Minibits): nothing deducted.
mock_wallet.get_fees_for_proofs = Mock(return_value=0)
from routstr.core.settings import settings
@@ -61,6 +63,69 @@ async def test_recieve_token_valid() -> None:
assert mint == "http://mint:3338"
@pytest.mark.asyncio
async def test_recieve_token_trusted_mint_deducts_input_fee() -> None:
"""A trusted mint that charges NUT-02 input fees.
The same-mint receive (`wallet.split(..., include_fees=True)`, a NUT-03 swap
at the same mint — not swap_to_primary_mint) pays the mint's per-proof fee,
so routstr only ends up with `face - input_fee` in fresh proofs. The credited
amount must reflect that, otherwise routstr over-credits the user and its own
wallet drifts toward insolvency.
"""
token_data = {
"token": [
{
"mint": "http://mint:3338",
"proofs": [
{"amount": 1000, "id": "test", "secret": "secret", "C": "curve"}
],
}
],
"unit": "sat",
}
token_json = json.dumps(token_data)
token_b64 = base64.urlsafe_b64encode(token_json.encode()).decode()
token_str = f"cashuA{token_b64}"
mock_wallet = Mock()
mock_wallet.split = AsyncMock()
# Mock a 3-sat input fee from the Cashu wallet API.
mock_wallet.get_fees_for_proofs = Mock(return_value=3)
from routstr.core.settings import settings
with patch.object(settings, "cashu_mints", ["http://mint:3338"]):
with patch("routstr.wallet.deserialize_token_from_string") as mock_deserialize:
mock_token = Mock()
mock_token.keysets = ["keyset1"]
mock_token.mint = "http://mint:3338"
mock_token.unit = "sat"
mock_token.amount = 1000
mock_token.proofs = [{"amount": 1000}]
mock_deserialize.return_value = mock_token
mock_wallet.load_mint = AsyncMock()
mock_wallet.load_proofs = AsyncMock()
# Patch get_wallet directly so the module-level `_wallets` cache
# (keyed by mint URL) can't hand back a wallet from another test.
with patch(
"routstr.wallet.get_wallet",
AsyncMock(return_value=mock_wallet),
):
amount, unit, mint = await recieve_token(token_str)
assert amount == 997 # 1000 face - 3 sat input fee paid on swap
assert unit == "sat"
assert mint == "http://mint:3338"
mock_wallet.get_fees_for_proofs.assert_called_once_with(
mock_token.proofs
)
# DLEQ is verified before re-minting the incoming proofs.
mock_wallet.verify_proofs_dleq.assert_called_once_with(
mock_token.proofs
)
@pytest.mark.asyncio
async def test_send_token() -> None:
mock_wallet = Mock()
@@ -254,19 +319,31 @@ async def test_recieve_token_untrusted_mint() -> None:
assert mint == "http://mint:3338"
@pytest.mark.asyncio
@pytest.mark.asyncio
async def test_swap_to_primary_mint_already_on_primary() -> None:
"""Same-mint shortcut: the token is already on the primary mint.
No cross-mint swap (no melt/mint), but the same-mint split(include_fees=True)
still burns the mint's NUT-02 input fee, so the credited amount must be face
minus the input fee — not full face value (the over-credit bug). DLEQ is
verified too, matching the trusted same-mint receive path.
"""
from routstr.core.settings import settings
from routstr.wallet import swap_to_primary_mint
mock_token = Mock()
mock_token.mint = settings.primary_mint
mock_token.keysets = ["keyset1"]
mock_token.amount = 1000
mock_token.unit = "sat"
mock_token.proofs = []
mock_token.proofs = [{"amount": 1000}]
mock_token_wallet = Mock()
mock_token_wallet.load_mint = AsyncMock()
mock_token_wallet.load_proofs = AsyncMock()
mock_token_wallet.verify_proofs_dleq = Mock()
# Mock a 3-sat input fee from the Cashu wallet API.
mock_token_wallet.get_fees_for_proofs = Mock(return_value=3)
mock_token_wallet.split = AsyncMock(return_value=None)
mock_token_wallet.request_mint = AsyncMock()
mock_token_wallet.melt_quote = AsyncMock()
@@ -274,9 +351,11 @@ async def test_swap_to_primary_mint_already_on_primary() -> None:
with patch("routstr.wallet.get_wallet", AsyncMock(return_value=mock_token_wallet)):
amount, unit, mint = await swap_to_primary_mint(mock_token, mock_token_wallet)
assert amount == 1000
assert amount == 997 # 1000 face - 3 sat input fee
assert unit == "sat"
assert mint == settings.primary_mint
mock_token_wallet.verify_proofs_dleq.assert_called_once_with(mock_token.proofs)
mock_token_wallet.get_fees_for_proofs.assert_called_once_with(mock_token.proofs)
mock_token_wallet.split.assert_called_once()
mock_token_wallet.request_mint.assert_not_called()
mock_token_wallet.melt_quote.assert_not_called()
+5 -8
View File
@@ -297,16 +297,13 @@ export default function ProvidersPage() {
};
const toggleProviderExpansion = (providerId: number) => {
const newExpanded = new Set(expandedProviders);
if (newExpanded.has(providerId)) {
newExpanded.delete(providerId);
} else {
newExpanded.add(providerId);
}
setExpandedProviders(newExpanded);
if (!newExpanded.has(providerId)) {
if (expandedProviders.has(providerId)) {
setExpandedProviders(new Set());
setViewingModels(null);
} else {
// Accordion: only one provider open at a time so switching to another
// provider's models auto-collapses the previously expanded one.
setExpandedProviders(new Set([providerId]));
setViewingModels(providerId);
}
};