Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
56 changes: 56 additions & 0 deletions node/test_utxo_db.py
Original file line number Diff line number Diff line change
Expand Up @@ -343,6 +343,62 @@ def test_mining_reward_empty_inputs_allowed(self):
self.assertTrue(ok)
self.assertEqual(self.db.get_balance('alice'), 100 * UNIT)

# -- bounty #2819 LOW: validation gaps & edge cases ----------------------

def test_duplicate_input_rejected(self):
"""Same box_id listed twice in inputs must be rejected.
Without explicit dedup, input_total is inflated 2x (LOW-2)."""
self._apply_coinbase('alice', 100 * UNIT)
boxes = self.db.get_unspent_for_address('alice')
box_id = boxes[0]['box_id']

ok = self.db.apply_transaction({
'tx_type': 'transfer',
'inputs': [
{'box_id': box_id, 'spending_proof': 'sig'},
{'box_id': box_id, 'spending_proof': 'sig'}, # duplicate
],
'outputs': [{'address': 'attacker', 'value_nrtc': 200 * UNIT}],
'fee_nrtc': 0,
}, block_height=10)
self.assertFalse(ok)
self.assertEqual(self.db.get_balance('alice'), 100 * UNIT)
self.assertEqual(self.db.get_balance('attacker'), 0)

def test_self_transfer(self):
"""Self-transfer (from == to) must work correctly."""
self._apply_coinbase('alice', 100 * UNIT)
boxes = self.db.get_unspent_for_address('alice')

ok = self.db.apply_transaction({
'tx_type': 'transfer',
'inputs': [{'box_id': boxes[0]['box_id'],
'spending_proof': 'sig'}],
'outputs': [{'address': 'alice', 'value_nrtc': 100 * UNIT}],
'fee_nrtc': 0,
}, block_height=10)
self.assertTrue(ok)
self.assertEqual(self.db.get_balance('alice'), 100 * UNIT)

def test_spending_proof_accepted_without_verification(self):
"""The UTXO layer accepts any spending_proof without verification.
Signature verification is the endpoint layer's responsibility.
This test documents the behavior so future changes don't
accidentally rely on it (LOW-3)."""
self._apply_coinbase('alice', 100 * UNIT)
boxes = self.db.get_unspent_for_address('alice')

# Bogus spending_proof is accepted at the UTXO layer
ok = self.db.apply_transaction({
'tx_type': 'transfer',
'inputs': [{'box_id': boxes[0]['box_id'],
'spending_proof': 'TOTALLY_BOGUS'}],
'outputs': [{'address': 'bob', 'value_nrtc': 100 * UNIT}],
'fee_nrtc': 0,
}, block_height=10)
self.assertTrue(ok, "UTXO layer should accept any spending_proof "
"(verification is endpoint's job)")

def test_mempool_empty_inputs_rejected_for_transfer(self):
"""Mempool must also reject non-minting txs with empty inputs."""
tx = {
Expand Down
18 changes: 18 additions & 0 deletions node/utxo_db.py
Original file line number Diff line number Diff line change
Expand Up @@ -292,6 +292,12 @@ def apply_transaction(self, tx: dict, block_height: int,
"""
Atomically apply a transaction: spend inputs, create outputs.

.. warning::
This method does **not** verify ``spending_proof``. Callers
MUST authenticate the spender (e.g. Ed25519 signature check)
before calling this method. See ``utxo_endpoints.py`` for
the endpoint-level verification.

``tx`` keys:
tx_type: str
inputs: list of {box_id: str, spending_proof: str}
Expand All @@ -316,6 +322,18 @@ def apply_transaction(self, tx: dict, block_height: int,
try:
conn.execute("BEGIN IMMEDIATE")

# -- reject duplicate input box_ids --------------------------------
# Keyed on box_id alone (the PK of the UTXO being consumed).
# Different spending_proof values for the same box_id are still
# a duplicate — the proof content is irrelevant to dedup.
# Without this, the same box_id counted twice inflates
# input_total. The spend-phase rowcount check catches it
# today, but only accidentally. Defense in depth.
input_box_ids = [i['box_id'] for i in inputs]
if len(input_box_ids) != len(set(input_box_ids)):
conn.execute("ROLLBACK")
return False

# -- validate inputs exist and are unspent -----------------------
input_total = 0
for inp in inputs:
Expand Down
Loading