Merge pull request 'feat(v2): collision guard — refuse machines whose npub matches an operator account (closes #32)' (#33) from feat/collision-detection into v2-bitspire

Reviewed-on: aiolabs/satmachineadmin#33

tests/test_collision_guard.py

@@ -0,0 +1,124 @@
+"""
+Tests for `views_api._assert_no_pubkey_collision` (aiolabs/satmachineadmin#32).
+
+Defends against the silent-drop failure mode reproduced on 2026-05-30T21:33Z:
+Greg's operator account pubkey had been seeded identical to the Sintra ATM's
+machine_npub, which masked the routing problem until Greg's pubkey rotated
+during the bunker migration — then `auto-account-from-npub` fired for the
+orphaned ATM npub and the cash-out invoice silently landed on a fresh
+auto-account wallet.
+
+The guard refuses to register a machine whose npub matches any LNbits
+operator account's `accounts.pubkey`, so this state cannot be entered
+through the satmachineadmin UI in the first place.
+
+Monkeypatches `views_api.get_account_by_pubkey` to avoid needing a live
+LNbits DB; this matches the assertion-style of tests/test_nostr_attribution
+(both isolate the assertion function for unit-testability).
+"""
+
+import asyncio
+from types import SimpleNamespace
+
+import pytest
+
+from .. import views_api
+from ..views_api import _assert_no_pubkey_collision
+
+# Canonical x-only pubkey for the integer 1 secret (matches NIP-44 reference vector).
+_PUBKEY_HEX = "79be667ef9dcbbac55a06295ce870b07029bfcdb2dce28d959f2815b16f81798"
+# Bech32 form of the same pubkey — operators may enter either form in the UI.
+_PUBKEY_NPUB = "npub10xlxvlhemja6c4dqv22uapctqupfhlxm9h8z3k2e72q4k9hcz7vqpkge6d"
+
+
+def _fake_account(pubkey: str = _PUBKEY_HEX):
+    """Account-shaped duck-typed object. _assert_no_pubkey_collision only
+    cares whether get_account_by_pubkey returns non-None; the returned
+    shape doesn't matter beyond that."""
+    return SimpleNamespace(id="op1", username="alice", pubkey=pubkey)
+
+
+def _patch_lookup(monkeypatch, return_value):
+    """Replace `views_api.get_account_by_pubkey` with an async stub that
+    captures the canonical-hex argument the guard normalised to and
+    returns the configured value."""
+    captured = {}
+
+    async def fake_lookup(pubkey: str):
+        captured["called_with"] = pubkey
+        return return_value
+
+    monkeypatch.setattr(views_api, "get_account_by_pubkey", fake_lookup)
+    return captured
+
+
+class TestCollisionDetected:
+    """Positive cases: machine_npub collides with an operator account's
+    pubkey. Each form (hex / bech32 / uppercase) must normalise to the
+    same canonical lookup + raise the same 400."""
+
+    def test_collision_with_hex_input_raises(self, monkeypatch):
+        _patch_lookup(monkeypatch, return_value=_fake_account())
+        with pytest.raises(Exception) as exc:
+            asyncio.run(_assert_no_pubkey_collision(_PUBKEY_HEX))
+        assert exc.value.status_code == 400
+        assert "collides with an existing LNbits operator account" in exc.value.detail
+        assert "aiolabs/satmachineadmin#32" in exc.value.detail
+
+    def test_collision_with_bech32_input_raises(self, monkeypatch):
+        """Operator may enter `npub1...` in the UI; the guard must
+        canonicalise to hex BEFORE the lookup, otherwise a colliding
+        npub-form input would silently miss the hex-stored
+        accounts.pubkey row."""
+        captured = _patch_lookup(monkeypatch, return_value=_fake_account())
+        with pytest.raises(Exception) as exc:
+            asyncio.run(_assert_no_pubkey_collision(_PUBKEY_NPUB))
+        assert exc.value.status_code == 400
+        # The bech32 input must be canonicalised to lowercase hex before the lookup.
+        assert captured["called_with"] == _PUBKEY_HEX
+
+    def test_collision_with_uppercase_hex_input_raises(self, monkeypatch):
+        """Hex inputs from manual entry / paste can land uppercase; the
+        guard's `normalize_public_key().lower()` should bring it to the
+        canonical lowercase hex that get_account_by_pubkey itself also
+        lowercases internally."""
+        captured = _patch_lookup(monkeypatch, return_value=_fake_account())
+        with pytest.raises(Exception) as exc:
+            asyncio.run(_assert_no_pubkey_collision(_PUBKEY_HEX.upper()))
+        assert exc.value.status_code == 400
+        assert captured["called_with"] == _PUBKEY_HEX
+
+
+class TestNoCollision:
+    """Negative cases: machine_npub does not match any account → guard
+    returns silently, machine creation can proceed."""
+
+    def test_no_collision_returns_silently(self, monkeypatch):
+        _patch_lookup(monkeypatch, return_value=None)
+        # Should NOT raise.
+        asyncio.run(_assert_no_pubkey_collision(_PUBKEY_HEX))
+
+    def test_no_collision_bech32_form_returns_silently(self, monkeypatch):
+        captured = _patch_lookup(monkeypatch, return_value=None)
+        asyncio.run(_assert_no_pubkey_collision(_PUBKEY_NPUB))
+        # The lookup still gets called with the canonicalised hex form.
+        assert captured["called_with"] == _PUBKEY_HEX
+
+
+class TestErrorMessage:
+    """The 400 detail must be operator-actionable: explains the failure,
+    points at the issue, and gives the remediation path."""
+
+    def test_error_includes_truncated_pubkey(self, monkeypatch):
+        _patch_lookup(monkeypatch, return_value=_fake_account())
+        with pytest.raises(Exception) as exc:
+            asyncio.run(_assert_no_pubkey_collision(_PUBKEY_HEX))
+        # First 12 chars of the canonical lowercase hex, followed by an ellipsis.
+        assert _PUBKEY_HEX[:12] in exc.value.detail
+
+    def test_error_includes_remediation_hint(self, monkeypatch):
+        _patch_lookup(monkeypatch, return_value=_fake_account())
+        with pytest.raises(Exception) as exc:
+            asyncio.run(_assert_no_pubkey_collision(_PUBKEY_HEX))
+        assert "lamassu-next" in exc.value.detail
+        assert "ATM_PRIVATE_KEY" in exc.value.detail