diff --git a/tests/test_session.py b/tests/test_session.py new file mode 100644 index 0000000..00e5092 --- /dev/null +++ b/tests/test_session.py @@ -0,0 +1,250 @@ +""" +/session, and the shared tap-authentication helper it introduced, exercised +with a genuine SUN: the test encrypts the PICC data with the card's k1 and +CMACs it with k2 exactly as an NTAG424 does, so the decrypt/verify path in +`_authenticate_tap` runs for real. Storage and LNbits core are stubbed at the +module boundary. +""" + +from datetime import datetime +from types import SimpleNamespace + +import pytest +from Cryptodome.Cipher import AES + +from .. import views_lnurl as v +from ..models import Card, Hit +from ..nxp424 import get_sun_mac + +K1 = bytes.fromhex("0F1E2D3C4B5A69788796A5B4C3D2E1F0") +K2 = bytes.fromhex("F0E1D2C3B4A5968778695A4B3C2D1E0F") +UID = bytes.fromhex("04A1B2C3D4E5F6") # 7-byte NTAG424 UID + + +def make_sun(counter: int) -> tuple[str, str]: + """(p, c) as the card would emit them for this read counter.""" + ctr = counter.to_bytes(3, "little") + plain = b"\xc7" + UID + ctr + b"\x00" * 5 # 16 bytes: tag, uid, ctr, pad + p = AES.new(K1, AES.MODE_CBC, b"\x00" * 16).encrypt(plain).hex().upper() + c = get_sun_mac(UID, ctr, K2).hex().upper() + return p, c + + +def make_card( + counter: int = 5, enable: bool = True, daily_limit: int = 100_000 +) -> Card: + return Card( + id="card1", + wallet="wallet1", + card_name="Alice", + uid=UID.hex().upper(), + external_id="ext123", + counter=counter, + tx_limit=50_000, + daily_limit=daily_limit, + enable=enable, + k0="00" * 16, + k1=K1.hex(), + k2=K2.hex(), + prev_k0="", + prev_k1="", + prev_k2="", + otp="", + time=datetime(2026, 1, 1), + ) + + +def make_hit(amount: int = 0, hit_id: str = "hit1") -> Hit: + return Hit( + id=hit_id, + card_id="card1", + ip="1.2.3.4", + spent=False, + useragent="test", + old_ctr=5, + new_ctr=6, + amount=amount, + time=datetime(2026, 1, 1), + ) + + +class FakeRequest: + def __init__(self) -> None: + self.client = SimpleNamespace(host="1.2.3.4") + self.headers = {"user-agent": "bitspire-test"} + + def url_for(self, name: str, **params: str) -> str: + return f"https://lnbits.test/boltcards/{name}/{params['hit_id']}" + + +@pytest.fixture +def stubs(monkeypatch): + """Stub storage + LNbits core; returns a dict the tests can tweak/inspect.""" + state = { + "card": make_card(), + "hits_today": [], + "updated_counter": None, + "created_hits": [], + "wallet": SimpleNamespace(balance_msat=123_456_000, currency="USD"), + "default_currency": None, + } + + async def get_card_by_external_id(external_id): + return state["card"] if external_id == state["card"].external_id else None + + async def update_card_counter(ctr, card_id): + state["updated_counter"] = ctr + + async def get_hits_today(card_id): + return state["hits_today"] + + async def create_hit(card_id, ip, agent, old_ctr, new_ctr): + hit = make_hit(hit_id=f"hit{len(state['created_hits']) + 1}") + state["created_hits"].append((card_id, ip, agent, old_ctr, new_ctr)) + return hit + + async def get_wallet(wallet_id): + return state["wallet"] + + async def satoshis_amount_as_fiat(amount, currency): + assert currency == "USD" + return amount * 0.0008 # 123456 sats → 98.7648 + + monkeypatch.setattr(v, "get_card_by_external_id", get_card_by_external_id) + monkeypatch.setattr(v, "update_card_counter", update_card_counter) + monkeypatch.setattr(v, "get_hits_today", get_hits_today) + monkeypatch.setattr(v, "create_hit", create_hit) + monkeypatch.setattr(v, "get_wallet", get_wallet) + monkeypatch.setattr(v, "satoshis_amount_as_fiat", satoshis_amount_as_fiat) + monkeypatch.setattr( + v.settings, "lnbits_default_accounting_currency", state["default_currency"] + ) + return state + + +@pytest.mark.asyncio +async def test_session_grants_balance_and_both_callbacks(stubs): + p, c = make_sun(counter=6) + out = await v.api_session(p, c, FakeRequest(), "ext123") + + assert out["authenticated"] is True + assert out["external_id"] == "ext123" + assert out["card_name"] == "Alice" + assert out["balance_msat"] == 123_456_000 + assert out["currency"] == "USD" + assert out["fiat"] == pytest.approx(98.7648) + # One hit, and both second steps are keyed by it. + assert len(stubs["created_hits"]) == 1 + assert out["withdraw"] == { + "callback": "https://lnbits.test/boltcards/boltcards.lnurl_callback/hit1", + "k1": "hit1", + "minWithdrawable": 1000, + "maxWithdrawable": 50_000_000, + } + assert out["withdraw_blocked_reason"] is None + assert ( + out["pay"]["callback"] + == "https://lnbits.test/boltcards/boltcards.pay_callback/hit1" + ) + assert out["pay"]["maxSendable"] == 50_000_000 + assert out["pay"]["metadata"] == v._TOPUP_METADATA + # The SUN counter advanced, so the same p/c can't be replayed. + assert stubs["updated_counter"] == 6 + + +@pytest.mark.asyncio +async def test_session_accepts_lowercase_p_and_c(stubs): + p, c = make_sun(counter=6) + out = await v.api_session(p.lower(), c.lower(), FakeRequest(), "ext123") + assert out["authenticated"] is True + + +@pytest.mark.asyncio +async def test_session_rejects_replayed_counter(stubs): + p, c = make_sun(counter=5) # card.counter is already 5 + out = await v.api_session(p, c, FakeRequest(), "ext123") + assert out == {"authenticated": False, "reason": "This link is already used."} + assert stubs["updated_counter"] is None + assert stubs["created_hits"] == [] + + +@pytest.mark.asyncio +async def test_session_rejects_bad_cmac_and_unknown_card(stubs): + p, c = make_sun(counter=6) + bad = await v.api_session(p, "00" * 8, FakeRequest(), "ext123") + assert bad == {"authenticated": False, "reason": "CMAC does not check."} + missing = await v.api_session(p, c, FakeRequest(), "nope") + assert missing == {"authenticated": False, "reason": "Card not found."} + stubs["card"] = make_card(enable=False) + disabled = await v.api_session(p, c, FakeRequest(), "ext123") + assert disabled == {"authenticated": False, "reason": "Card is disabled."} + + +@pytest.mark.asyncio +async def test_session_withholds_withdraw_over_daily_limit_but_keeps_pay(stubs): + stubs["card"] = make_card(daily_limit=1_000) + stubs["hits_today"] = [make_hit(amount=900), make_hit(amount=200)] + p, c = make_sun(counter=6) + out = await v.api_session(p, c, FakeRequest(), "ext123") + assert out["authenticated"] is True + assert out["withdraw"] is None + assert out["withdraw_blocked_reason"] == "Max daily limit spent." + assert out["pay"]["callback"].endswith("/hit1") + + +@pytest.mark.asyncio +async def test_session_without_currency_has_no_fiat(stubs): + stubs["wallet"] = SimpleNamespace(balance_msat=5_000, currency=None) + p, c = make_sun(counter=6) + out = await v.api_session(p, c, FakeRequest(), "ext123") + assert out["balance_msat"] == 5_000 + assert out["currency"] is None + assert out["fiat"] is None + + +@pytest.mark.asyncio +async def test_session_falls_back_to_instance_default_currency(stubs, monkeypatch): + stubs["wallet"] = SimpleNamespace(balance_msat=5_000, currency=None) + monkeypatch.setattr(v.settings, "lnbits_default_accounting_currency", "USD") + p, c = make_sun(counter=6) + out = await v.api_session(p, c, FakeRequest(), "ext123") + assert out["currency"] == "USD" + assert out["fiat"] == pytest.approx(5 * 0.0008) + + +@pytest.mark.asyncio +async def test_session_survives_rate_failure(stubs, monkeypatch): + async def boom(amount, currency): + raise ValueError("no rate") + + monkeypatch.setattr(v, "satoshis_amount_as_fiat", boom) + p, c = make_sun(counter=6) + out = await v.api_session(p, c, FakeRequest(), "ext123") + assert out["authenticated"] is True + assert out["fiat"] is None + + +# The refactor moved /verify and /pay onto the shared helper — pin their +# behaviour so the extraction can't have changed it. + + +@pytest.mark.asyncio +async def test_verify_still_authenticates_via_shared_helper(stubs): + p, c = make_sun(counter=6) + out = await v.api_verify(p, c, "ext123") + assert out == {"authenticated": True, "external_id": "ext123", "card_name": "Alice"} + assert stubs["updated_counter"] == 6 + # The counter stub doesn't persist, so replay against a card already at 6. + stubs["card"] = make_card(counter=6) + replay = await v.api_verify(p, c, "ext123") + assert replay == {"authenticated": False, "reason": "This link is already used."} + + +@pytest.mark.asyncio +async def test_pay_still_returns_pay_request_via_shared_helper(stubs): + p, c = make_sun(counter=6) + out = await v.api_pay(p, c, FakeRequest(), "ext123") + assert str(out.callback).endswith("/boltcards.pay_callback/hit1") + assert int(out.maxSendable) == 50_000_000 + bad = await v.api_pay(p, "00" * 8, FakeRequest(), "ext123") + assert bad.reason == "CMAC does not check." diff --git a/views_lnurl.py b/views_lnurl.py index 54d5915..4436737 100644 --- a/views_lnurl.py +++ b/views_lnurl.py @@ -5,7 +5,10 @@ from urllib.parse import urlparse import bolt11 from fastapi import APIRouter, HTTPException, Query, Request +from lnbits.core.crud import get_wallet from lnbits.core.services import create_invoice, pay_invoice +from lnbits.settings import settings +from lnbits.utils.exchange_rates import satoshis_amount_as_fiat from lnurl import ( CallbackUrl, LightningInvoice, @@ -33,7 +36,7 @@ from .crud import ( update_card_counter, update_card_otp, ) -from .models import UIDPost +from .models import Card, UIDPost from .nxp424 import decrypt_sun, get_sun_mac boltcards_lnurl_router = APIRouter() @@ -293,6 +296,59 @@ async def lnurlp_response( ) +###############SHARED TAP AUTHENTICATION (fork endpoints)################# +# /pay, /verify and /session all authenticate a tap exactly the way upstream's +# /scan does — same lookups, same checks, same order, same reasons — and then +# advance the stored SUN counter so a captured p/c can't be replayed. Keeping +# that in one place means the fork endpoints can't drift from /scan's +# acceptance rules. /scan itself is left untouched (upstream code). + + +async def _authenticate_tap( + external_id: str, p: str, c: str +) -> tuple[Card | None, int, str | None]: + """Look up the card, verify the SUN and advance the counter. + + Returns ``(card, new_counter, None)`` on success or ``(None, 0, reason)`` + with a /scan-compatible reason on failure. + """ + # some wallets send everything as lower case, no bueno + p = p.upper() + c = c.upper() + card = await get_card_by_external_id(external_id) + if not card: + return None, 0, "Card not found." + if not card.enable: + return None, 0, "Card is disabled." + try: + card_uid, counter = decrypt_sun(bytes.fromhex(p), bytes.fromhex(card.k1)) + if card.uid.upper() != card_uid.hex().upper(): + return None, 0, "Card UID mis-match." + if c != get_sun_mac(card_uid, counter, bytes.fromhex(card.k2)).hex().upper(): + return None, 0, "CMAC does not check." + except Exception: + return None, 0, "Error decrypting card." + + ctr_int = int.from_bytes(counter, "little") + if ctr_int <= card.counter: + return None, 0, "This link is already used." + await update_card_counter(ctr_int, card.id) + return card, ctr_int, None + + +def _client_info(request: Request) -> tuple[str, str] | None: + """(ip, user-agent) for the hit record, as /scan gathers them.""" + if not request.client: + return None + ip = request.client.host + if "x-real-ip" in request.headers: + ip = request.headers["x-real-ip"] + elif "x-forwarded-for" in request.headers: + ip = request.headers["x-forwarded-for"] + agent = request.headers["user-agent"] if "user-agent" in request.headers else "" + return ip, agent + + ###############LNURLPAY TAP-TO-RECEIVE (top-up)################# # Deposit sats to a card's wallet by tapping the card — the receive/cash-in # counterpart of the /scan withdraw. A Bolt Card only emits its lnurlw (a spend @@ -317,38 +373,16 @@ _TOPUP_METADATA = json.dumps([["text/plain", "Bolt Card top-up"]]) async def api_pay( p, c, request: Request, external_id: str ) -> LnurlPayResponse | LnurlErrorResponse: - # Mirror /scan's SUN verification exactly (some wallets lowercase p/c). - p = p.upper() - c = c.upper() - card = await get_card_by_external_id(external_id) + card, ctr_int, reason = await _authenticate_tap(external_id, p, c) if not card: - return LnurlErrorResponse(reason="Card not found.") - if not card.enable: - return LnurlErrorResponse(reason="Card is disabled.") - try: - card_uid, counter = decrypt_sun(bytes.fromhex(p), bytes.fromhex(card.k1)) - if card.uid.upper() != card_uid.hex().upper(): - return LnurlErrorResponse(reason="Card UID mis-match.") - if c != get_sun_mac(card_uid, counter, bytes.fromhex(card.k2)).hex().upper(): - return LnurlErrorResponse(reason="CMAC does not check.") - except Exception: - return LnurlErrorResponse(reason="Error decrypting card.") - - ctr_int = int.from_bytes(counter, "little") - if ctr_int <= card.counter: - return LnurlErrorResponse(reason="This link is already used.") - await update_card_counter(ctr_int, card.id) + return LnurlErrorResponse(reason=reason or "Card not found.") # Record the tap; the hit id is the single-use bearer for the callback. # (No daily-limit check here — that gates spending, and this only deposits.) - if not request.client: + client = _client_info(request) + if not client: return LnurlErrorResponse(reason="Cannot get client info.") - ip = request.client.host - if "x-real-ip" in request.headers: - ip = request.headers["x-real-ip"] - elif "x-forwarded-for" in request.headers: - ip = request.headers["x-forwarded-for"] - agent = request.headers["user-agent"] if "user-agent" in request.headers else "" + ip, agent = client hit = await create_hit(card.id, ip, agent, card.counter, ctr_int) callback_url = parse_obj_as( @@ -406,29 +440,95 @@ async def pay_callback( # the card-programming OTP endpoint. @boltcards_lnurl_router.get("/api/v1/verify/{external_id}") async def api_verify(p, c, external_id: str): - p = p.upper() - c = c.upper() - card = await get_card_by_external_id(external_id) + card, _ctr_int, reason = await _authenticate_tap(external_id, p, c) if not card: - return {"authenticated": False, "reason": "Card not found."} - if not card.enable: - return {"authenticated": False, "reason": "Card is disabled."} - try: - card_uid, counter = decrypt_sun(bytes.fromhex(p), bytes.fromhex(card.k1)) - if card.uid.upper() != card_uid.hex().upper(): - return {"authenticated": False, "reason": "Card UID mis-match."} - if c != get_sun_mac(card_uid, counter, bytes.fromhex(card.k2)).hex().upper(): - return {"authenticated": False, "reason": "CMAC does not check."} - except Exception: - return {"authenticated": False, "reason": "Error decrypting card."} - - ctr_int = int.from_bytes(counter, "little") - if ctr_int <= card.counter: - return {"authenticated": False, "reason": "This link is already used."} - await update_card_counter(ctr_int, card.id) + return {"authenticated": False, "reason": reason} return { "authenticated": True, "external_id": card.external_id, "card_name": card.card_name, } + + +###############ACCESS-CONTROL SESSION (terminal tap-to-enter)################# +# /boltcards/api/v1/session/{external_id}?p=<32-hex>&c=<16-hex> +# One tap → one session. For a terminal (bitSpire ATM) that unlocks on a card +# tap and lets the holder finish a buy or sell later in the same visit. A tap +# yields a single-use SUN, so anything that verifies it — /scan, /pay, /verify +# — spends it; a terminal that verified at entry could not reuse the p/c to +# move sats afterwards. This endpoint verifies ONCE (advancing the counter, +# exactly like /scan), records ONE hit, and returns everything the rest of the +# visit needs: +# - the card wallet's balance and its fiat equivalent (display only), +# - the LUD-03 second step (withdraw callback, k1 = hit) to pull a payment, +# - the LUD-06 second step (pay callback) to top the wallet up. +# Both callbacks are keyed by the hit — the same single-use bearer /scan and +# /pay already hand out — so the terminal holds no p/c, and the first withdraw +# spends the hit just as it would after a /scan. A top-up leaves it unspent, as +# /pay does. Reasons mirror /scan so a terminal can show them verbatim. + + +@boltcards_lnurl_router.get("/api/v1/session/{external_id}") +async def api_session(p, c, request: Request, external_id: str): + card, ctr_int, reason = await _authenticate_tap(external_id, p, c) + if not card: + return {"authenticated": False, "reason": reason} + + client = _client_info(request) + if not client: + return {"authenticated": False, "reason": "Cannot get client info."} + ip, agent = client + + # /scan refuses a withdraw voucher once today's hits exceed the daily limit. + # Mirror that by withholding the withdraw step; the top-up step only + # deposits and stays available, so the session itself is still granted. + todays_hits = await get_hits_today(card.id) + spent_today = sum(hit.amount for hit in todays_hits) + hit = await create_hit(card.id, ip, agent, card.counter, ctr_int) + + withdraw = None + withdraw_blocked_reason = None + if spent_today > int(card.daily_limit): + withdraw_blocked_reason = "Max daily limit spent." + else: + withdraw = { + "callback": str(request.url_for("boltcards.lnurl_callback", hit_id=hit.id)), + "k1": hit.id, + "minWithdrawable": 1000, + "maxWithdrawable": int(card.tx_limit) * 1000, + } + pay = { + "callback": str(request.url_for("boltcards.pay_callback", hit_id=hit.id)), + "minSendable": 1000, + "maxSendable": int(card.tx_limit) * 1000, + "metadata": _TOPUP_METADATA, + } + + # Balance + fiat, the way the LNbits wallet page shows them: the wallet's + # own currency first (per-wallet setting, LNbits ≥ 1.6), then the + # instance's default accounting currency; no currency → no fiat, and a + # rate failure never fails the session. + wallet = await get_wallet(card.wallet) + balance_msat = int(wallet.balance_msat) if wallet else 0 + currency = (getattr(wallet, "currency", None) if wallet else None) or getattr( + settings, "lnbits_default_accounting_currency", None + ) + fiat = None + if currency: + try: + fiat = await satoshis_amount_as_fiat(balance_msat / 1000, currency) + except Exception: + fiat = None + + return { + "authenticated": True, + "external_id": card.external_id, + "card_name": card.card_name, + "balance_msat": balance_msat, + "currency": currency, + "fiat": fiat, + "withdraw": withdraw, + "withdraw_blocked_reason": withdraw_blocked_reason, + "pay": pay, + }