feat(pairing): bunker pairing service — mint per-spire key + seed URL (#9)

Operator-side producer for seed-URL pairing (S0/#9, model A1). pair_spire()
orchestrates the nsecbunkerd admin chain via lnbits' NsecBunkerAdminClient:
  create_new_key(spire-<id>) -> _ensure_spire_policy -> create_new_token ->
  get_key_tokens -> package the <npub>#secret token into a bunker:// URL +
  base64url seed URL {spire_npub, spire_pubkey, bunker_url, relays}.

The spire later self-signs all its events as that bunker-held key; lnbits'
path-B roster maps the npub to the operator wallet — no nsec on the spire.
spirekeeper does steps 1-4 only; the NIP-46 connect/bind happens spire-side
(bitspire#52) with the spire's own client keypair.

Scoped policy 'spirekeeper-spire': sign_event 21000/21001-3/30078 + nip44
(kind-less via add_policy_rule). Local _ensure_spire_policy (no cache)
avoids lnbits' admin-pubkey-keyed _ensure_policy cache (policy-name-blind).

9 unit tests with a fake bunker (orchestration, policy reconcile, seed/
bunker:// wire shape, error paths); npub<->hex via lnbits' real helpers.
200 tests green.

Known gaps (lnbits NsecBunkerAdminClient): no token-expiry param, no revoke
RPC — re-pair works; 'revoke spire access' deferred to a bunker follow-up.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

tests/test_pairing.py

@@ -0,0 +1,240 @@
+"""Unit tests for the seed-URL pairing service (S0 / #9, model A1).
+
+The bunker admin client is faked — these exercise the orchestration
+(create_new_key -> ensure-policy -> create_new_token -> get_key_tokens),
+the policy reconciliation, and the seed-URL / bunker:// wire shape, with
+no live nsecbunkerd. npub<->hex round-trips through lnbits' real helpers
+so the parsing is exercised for real.
+
+Async is driven via asyncio.run (this venv has no pytest-asyncio), matching
+the rest of the suite.
+"""
+
+import asyncio
+import base64
+import json
+from datetime import datetime, timezone
+
+import pytest
+from lnbits.utils.nostr import hex_to_npub
+
+from ..models import Machine
+from ..pairing import (
+    SEED_URL_SCHEME,
+    SPIRE_POLICY_METHODS_NO_KIND,
+    SPIRE_POLICY_NAME,
+    SPIRE_POLICY_RULES,
+    PairingError,
+    build_seed_url,
+    pair_spire,
+    spire_key_name,
+)
+
+_NOW = datetime(2026, 6, 16, tzinfo=timezone.utc)
+_SPIRE_HEX = "522a4538f1df96508d9ee8b14072344dd4a566acfe03c25a92a39179c6fca891"
+_SPIRE_NPUB = hex_to_npub(_SPIRE_HEX)
+_RELAYS = ["wss://lnbits.demo.aiolabs.dev/nostrrelay/demo"]
+_BUNKER_RELAY = "wss://bunker.internal/relay"
+_PASSPHRASE = "keystore-pass"  # pragma: allowlist secret
+
+
+def _machine(mid: str = "m1") -> Machine:
+    return Machine(
+        id=mid,
+        operator_user_id="op1",
+        machine_npub="placeholder",
+        wallet_id="w1",
+        name="sintra",
+        location=None,
+        fiat_code="EUR",
+        is_active=True,
+        created_at=_NOW,
+        updated_at=_NOW,
+    )
+
+
+class FakeBunker:
+    """Records calls; returns canned bunker responses."""
+
+    # pragma: allowlist secret
+    def __init__(self, *, policies=None, token_secret="s3cr3t"):
+        self._policies = policies or []
+        self._token_secret = token_secret
+        self.calls: list[tuple] = []
+        self._next_policy_id = 7
+
+    async def create_new_key(self, name, passphrase):
+        self.calls.append(("create_new_key", name, passphrase))
+        return _SPIRE_NPUB
+
+    async def get_policies(self):
+        self.calls.append(("get_policies",))
+        return list(self._policies)
+
+    async def create_new_policy(self, name, rules):
+        self.calls.append(("create_new_policy", name, rules))
+        pid = self._next_policy_id
+        self._policies.append({"id": pid, "name": name, "rules": list(rules)})
+        return pid
+
+    async def add_policy_rule(self, policy_id, rule):
+        self.calls.append(("add_policy_rule", policy_id, rule))
+
+    async def create_new_token(self, key_name, client_name, policy_id):
+        self.calls.append(("create_new_token", key_name, client_name, policy_id))
+
+    async def get_key_tokens(self, key_name):
+        self.calls.append(("get_key_tokens", key_name))
+        return [
+            {
+                "clientName": f"spire-client-{key_name.split('-', 1)[1]}",
+                "token": f"{_SPIRE_NPUB}#{self._token_secret}",
+            }
+        ]
+
+    def named(self, name):
+        return [c for c in self.calls if c[0] == name]
+
+
+def _pair(bunker, machine=None):
+    return asyncio.run(
+        pair_spire(
+            machine or _machine(),
+            relays=_RELAYS,
+            admin_client=bunker,
+            bunker_relay=_BUNKER_RELAY,
+            keystore_passphrase=_PASSPHRASE,
+        )
+    )
+
+
+def test_pair_happy_path_mints_key_policy_token():
+    bunker = FakeBunker(token_secret="abc123")  # pragma: allowlist secret
+    result = _pair(bunker)
+
+    assert ("create_new_key", "spire-m1", _PASSPHRASE) in bunker.calls
+    assert result.bunker_key_name == spire_key_name("m1") == "spire-m1"
+
+    assert result.spire_npub == _SPIRE_NPUB
+    assert result.spire_pubkey_hex == _SPIRE_HEX
+
+    created = bunker.named("create_new_policy")
+    assert created and created[0][1] == SPIRE_POLICY_NAME
+    token_call = bunker.named("create_new_token")[0]
+    assert token_call[1] == "spire-m1"  # key_name
+    assert token_call[2] == "spire-client-m1"  # client_name
+    assert token_call[3] == 7  # policy_id from the fake's create_new_policy
+
+
+def test_bunker_url_carries_pubkey_relay_secret():
+    result = _pair(FakeBunker(token_secret="topsecret"))  # pragma: allowlist secret
+    assert result.bunker_url.startswith(f"bunker://{_SPIRE_HEX}?")
+    assert "relay=wss%3A%2F%2Fbunker.internal%2Frelay" in result.bunker_url
+    assert "secret=topsecret" in result.bunker_url
+
+
+def test_seed_url_decodes_to_contract():
+    result = _pair(FakeBunker(token_secret="zzz"))  # pragma: allowlist secret
+    assert result.seed_url.startswith(SEED_URL_SCHEME)
+    blob = result.seed_url[len(SEED_URL_SCHEME) :]
+    payload = json.loads(base64.urlsafe_b64decode(blob + "=" * (-len(blob) % 4)))
+    assert payload == {
+        "v": 1,
+        "spire_npub": _SPIRE_NPUB,
+        "spire_pubkey": _SPIRE_HEX,
+        "bunker_url": result.bunker_url,
+        "relays": _RELAYS,
+    }
+
+
+def test_fresh_policy_adds_kindless_nip44_rules():
+    bunker = FakeBunker()  # no existing policies
+    _pair(bunker)
+    added = [c[2]["method"] for c in bunker.named("add_policy_rule")]
+    # kind-scoped rules went in via create_new_policy; only the kind-less
+    # nip44 methods are reconciled in via add_policy_rule.
+    assert set(added) == set(SPIRE_POLICY_METHODS_NO_KIND)
+
+
+def test_existing_policy_reused_not_recreated():
+    existing = [
+        {
+            "id": 42,
+            "name": SPIRE_POLICY_NAME,
+            "rules": [dict(r) for r in SPIRE_POLICY_RULES],
+        }
+    ]
+    bunker = FakeBunker(policies=existing)
+    _pair(bunker)
+    assert not bunker.named("create_new_policy")  # reused, not recreated
+    assert bunker.named("create_new_token")[0][3] == 42  # used existing id
+    added = [c[2]["method"] for c in bunker.named("add_policy_rule")]
+    assert set(added) == set(SPIRE_POLICY_METHODS_NO_KIND)
+
+
+def test_fully_provisioned_policy_adds_nothing():
+    rules = [dict(r) for r in SPIRE_POLICY_RULES] + [
+        {"method": m, "kind": None} for m in SPIRE_POLICY_METHODS_NO_KIND
+    ]
+    bunker = FakeBunker(policies=[{"id": 9, "name": SPIRE_POLICY_NAME, "rules": rules}])
+    _pair(bunker)
+    assert not bunker.named("add_policy_rule")
+    assert not bunker.named("create_new_policy")
+
+
+def test_malformed_token_raises():
+    bunker = FakeBunker()
+
+    async def _bad_tokens(key_name):
+        _ = key_name
+        return [{"token": "no-hash-here"}]
+
+    bunker.get_key_tokens = _bad_tokens
+    with pytest.raises(PairingError, match="malformed token"):
+        _pair(bunker)
+
+
+def test_missing_relay_or_passphrase_raises():
+    with pytest.raises(PairingError, match="LNBITS_NSEC_BUNKER_URL"):
+        asyncio.run(
+            pair_spire(
+                _machine(),
+                relays=_RELAYS,
+                admin_client=FakeBunker(),
+                bunker_relay="",
+                keystore_passphrase=_PASSPHRASE,
+            )
+        )
+    with pytest.raises(PairingError, match="PASSPHRASE"):
+        asyncio.run(
+            pair_spire(
+                _machine(),
+                relays=_RELAYS,
+                admin_client=FakeBunker(),
+                bunker_relay=_BUNKER_RELAY,
+                keystore_passphrase="",
+            )
+        )
+    with pytest.raises(PairingError, match="relay is required"):
+        asyncio.run(
+            pair_spire(
+                _machine(),
+                relays=[],
+                admin_client=FakeBunker(),
+                bunker_relay=_BUNKER_RELAY,
+                keystore_passphrase=_PASSPHRASE,
+            )
+        )
+
+
+def test_build_seed_url_roundtrip():
+    url = build_seed_url(
+        spire_npub=_SPIRE_NPUB,
+        spire_pubkey_hex=_SPIRE_HEX,
+        bunker_url="bunker://x?relay=r&secret=s",
+        relays=_RELAYS,
+    )
+    blob = url[len(SEED_URL_SCHEME) :]
+    payload = json.loads(base64.urlsafe_b64decode(blob + "=" * (-len(blob) % 4)))
+    assert payload["spire_pubkey"] == _SPIRE_HEX
+    assert payload["relays"] == _RELAYS