davide
e8fdea0389
detect_external_spends marked a UTXO spent_txid='external-spend' irreversibly the moment it was missing from one listunspent reply, on one server, with no way to undo it. A rotated-to server that's broken or behind, or an empty reply, could zero a user's balance permanently. Split into three functions in deposits/service.py: find_utxos_missing_ from (read-only candidate detection, and refuses to flag anything at all when listunspent comes back entirely empty for a funded address - that reads as a broken response, not a full sweep), mark_utxos_spent_ externally (persistence only, once a candidate is already confirmed), and reinstate_reappeared_utxos (undoes the mark if the outpoint reappears as unspent later). ElectrumListener gains corroborate_utxo_spent, sharing the same majority-quorum logic corroborate_header already uses for B-28: before a candidate is marked, the other configured servers are asked whether they also see it as spent. _refresh_user now reads candidates, then corroborates each one with no DB session held open across those network calls (same shape as B-18/B-25), then persists. Suite grows from 165 to 176 tests. BUGS.md moves B-29 to Previously fixed.
187 lines
8.2 KiB
Python
187 lines
8.2 KiB
Python
import pytest
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
|
|
|
|
from app.db.base import Base
|
|
from app.db.models import AuditLog, User, UtxoEvent
|
|
from app.deposits.service import (
|
|
credit_confirmed_utxos,
|
|
find_utxos_missing_from,
|
|
mark_utxos_spent_externally,
|
|
reinstate_reappeared_utxos,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
async def session_factory():
|
|
engine = create_async_engine("sqlite+aiosqlite:///:memory:")
|
|
async with engine.begin() as conn:
|
|
await conn.run_sync(Base.metadata.create_all)
|
|
yield async_sessionmaker(engine, expire_on_commit=False)
|
|
await engine.dispose()
|
|
|
|
|
|
@pytest.fixture
|
|
async def user_id(session_factory):
|
|
async with session_factory() as session:
|
|
user = User(username="alice", password_hash="x", derivation_index=0, address="plm1qxxx")
|
|
session.add(user)
|
|
await session.commit()
|
|
return user.id
|
|
|
|
|
|
async def test_credits_confirmed_utxo_and_updates_balance(session_factory, user_id):
|
|
entries = [{"tx_hash": "aa" * 32, "tx_pos": 0, "height": 100, "value": 10_000_000}]
|
|
async with session_factory() as session:
|
|
credited = await credit_confirmed_utxos(session, user_id, entries)
|
|
assert credited == 1
|
|
user = await session.get(User, user_id)
|
|
assert user.cached_balance_sats == 10_000_000
|
|
|
|
|
|
async def test_unconfirmed_entry_is_ignored(session_factory, user_id):
|
|
entries = [{"tx_hash": "bb" * 32, "tx_pos": 0, "height": 0, "value": 5_000_000}]
|
|
async with session_factory() as session:
|
|
credited = await credit_confirmed_utxos(session, user_id, entries)
|
|
assert credited == 0
|
|
user = await session.get(User, user_id)
|
|
assert user.cached_balance_sats == 0
|
|
|
|
|
|
async def test_idempotent_on_repeated_notification(session_factory, user_id):
|
|
entries = [{"tx_hash": "cc" * 32, "tx_pos": 0, "height": 100, "value": 7_000_000}]
|
|
async with session_factory() as session:
|
|
first = await credit_confirmed_utxos(session, user_id, entries)
|
|
async with session_factory() as session:
|
|
second = await credit_confirmed_utxos(session, user_id, entries)
|
|
user = await session.get(User, user_id)
|
|
assert first == 1
|
|
assert second == 0
|
|
assert user.cached_balance_sats == 7_000_000
|
|
|
|
|
|
# --- B-29: detecting a UTXO spent outside the platform is now a three-step,
|
|
# corroborate-before-you-mark process, split across find_utxos_missing_from
|
|
# (read-only candidate detection), the caller's own corroboration against other
|
|
# servers (electrum/listener.py, not exercised here), and mark_utxos_spent_
|
|
# externally (persistence only, once a candidate is already confirmed). ---------
|
|
|
|
|
|
async def test_find_utxos_missing_from_returns_the_missing_candidate(session_factory, user_id):
|
|
async with session_factory() as session:
|
|
await credit_confirmed_utxos(
|
|
session, user_id, [{"tx_hash": "dd" * 32, "tx_pos": 0, "height": 100, "value": 20_000_000}]
|
|
)
|
|
|
|
# A different outpoint present in this refresh — our own tracked one is
|
|
# genuinely absent from it, not just from an entirely empty reply.
|
|
other_entries = [{"tx_hash": "99" * 32, "tx_pos": 0, "height": 100, "value": 1_000_000}]
|
|
async with session_factory() as session:
|
|
candidates = await find_utxos_missing_from(session, user_id, other_entries)
|
|
assert len(candidates) == 1
|
|
assert candidates[0].txid == "dd" * 32
|
|
assert candidates[0].spent_txid is None # read-only: nothing is marked yet
|
|
|
|
|
|
async def test_find_utxos_missing_from_returns_nothing_when_present(session_factory, user_id):
|
|
entries = [{"tx_hash": "ee" * 32, "tx_pos": 0, "height": 100, "value": 3_000_000}]
|
|
async with session_factory() as session:
|
|
await credit_confirmed_utxos(session, user_id, entries)
|
|
|
|
async with session_factory() as session:
|
|
candidates = await find_utxos_missing_from(session, user_id, entries)
|
|
assert candidates == []
|
|
user = await session.get(User, user_id)
|
|
assert user.cached_balance_sats == 3_000_000
|
|
|
|
|
|
async def test_find_utxos_missing_from_skips_a_totally_empty_response(session_factory, user_id):
|
|
"""B-29: an entirely empty listunspent for a funded address reads as an
|
|
incomplete/broken response, not proof of a full external sweep — it would
|
|
otherwise flag every UTXO of this user as missing from one bad reply."""
|
|
async with session_factory() as session:
|
|
await credit_confirmed_utxos(
|
|
session, user_id, [{"tx_hash": "dd" * 32, "tx_pos": 0, "height": 100, "value": 20_000_000}]
|
|
)
|
|
|
|
async with session_factory() as session:
|
|
candidates = await find_utxos_missing_from(session, user_id, [])
|
|
assert candidates == []
|
|
|
|
|
|
async def test_mark_utxos_spent_externally_marks_and_corrects_balance(session_factory, user_id):
|
|
async with session_factory() as session:
|
|
await credit_confirmed_utxos(
|
|
session, user_id, [{"tx_hash": "dd" * 32, "tx_pos": 0, "height": 100, "value": 20_000_000}]
|
|
)
|
|
|
|
async with session_factory() as session:
|
|
utxo = (await session.scalars(select(UtxoEvent).where(UtxoEvent.user_id == user_id))).one()
|
|
marked = await mark_utxos_spent_externally(session, user_id, [utxo.id])
|
|
assert marked == 1
|
|
|
|
user = await session.get(User, user_id)
|
|
assert user.cached_balance_sats == 0
|
|
|
|
utxo = (await session.scalars(select(UtxoEvent).where(UtxoEvent.user_id == user_id))).one()
|
|
assert utxo.spent_txid == "external-spend"
|
|
|
|
audit_events = (await session.scalars(select(AuditLog))).all()
|
|
assert any(e.event_type == "utxo_spent_externally" for e in audit_events)
|
|
|
|
|
|
async def test_mark_utxos_spent_externally_skips_an_already_resolved_row(session_factory, user_id):
|
|
"""Something else (a legitimate platform spend, or a prior refresh) may have
|
|
resolved the row between the caller reading the candidate list and finishing
|
|
corroboration — mark_utxos_spent_externally must not clobber that."""
|
|
async with session_factory() as session:
|
|
await credit_confirmed_utxos(
|
|
session, user_id, [{"tx_hash": "dd" * 32, "tx_pos": 0, "height": 100, "value": 20_000_000}]
|
|
)
|
|
utxo = (await session.scalars(select(UtxoEvent).where(UtxoEvent.user_id == user_id))).one()
|
|
utxo_id = utxo.id
|
|
utxo.spent_txid = "some-real-platform-txid"
|
|
await session.commit()
|
|
|
|
async with session_factory() as session:
|
|
marked = await mark_utxos_spent_externally(session, user_id, [utxo_id])
|
|
assert marked == 0
|
|
utxo = await session.get(UtxoEvent, utxo_id)
|
|
assert utxo.spent_txid == "some-real-platform-txid" # untouched
|
|
|
|
|
|
async def test_reinstate_reappeared_utxos_clears_the_mark_and_restores_balance(session_factory, user_id):
|
|
async with session_factory() as session:
|
|
await credit_confirmed_utxos(
|
|
session, user_id, [{"tx_hash": "dd" * 32, "tx_pos": 0, "height": 100, "value": 20_000_000}]
|
|
)
|
|
utxo_id = (await session.scalars(select(UtxoEvent).where(UtxoEvent.user_id == user_id))).one().id
|
|
|
|
async with session_factory() as session:
|
|
await mark_utxos_spent_externally(session, user_id, [utxo_id])
|
|
|
|
# The outpoint reappears as unspent in a later refresh.
|
|
entries = [{"tx_hash": "dd" * 32, "tx_pos": 0, "height": 100, "value": 20_000_000}]
|
|
async with session_factory() as session:
|
|
reinstated = await reinstate_reappeared_utxos(session, user_id, entries)
|
|
assert reinstated == 1
|
|
|
|
user = await session.get(User, user_id)
|
|
assert user.cached_balance_sats == 20_000_000
|
|
|
|
utxo = await session.get(UtxoEvent, utxo_id)
|
|
assert utxo.spent_txid is None
|
|
|
|
audit_events = [e.event_type for e in (await session.scalars(select(AuditLog))).all()]
|
|
assert "utxo_external_spend_reinstated" in audit_events
|
|
|
|
|
|
async def test_reinstate_reappeared_utxos_ignores_unmarked_rows(session_factory, user_id):
|
|
entries = [{"tx_hash": "ee" * 32, "tx_pos": 0, "height": 100, "value": 3_000_000}]
|
|
async with session_factory() as session:
|
|
await credit_confirmed_utxos(session, user_id, entries)
|
|
|
|
async with session_factory() as session:
|
|
reinstated = await reinstate_reappeared_utxos(session, user_id, entries)
|
|
assert reinstated == 0
|