diff --git a/tests/test_session.py b/tests/test_session.py deleted file mode 100644 index 00e5092..0000000 --- a/tests/test_session.py +++ /dev/null @@ -1,250 +0,0 @@ -""" -/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 4436737..54d5915 100644 --- a/views_lnurl.py +++ b/views_lnurl.py @@ -5,10 +5,7 @@ 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, @@ -36,7 +33,7 @@ from .crud import ( update_card_counter, update_card_otp, ) -from .models import Card, UIDPost +from .models import UIDPost from .nxp424 import decrypt_sun, get_sun_mac boltcards_lnurl_router = APIRouter() @@ -296,59 +293,6 @@ 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 @@ -373,16 +317,38 @@ _TOPUP_METADATA = json.dumps([["text/plain", "Bolt Card top-up"]]) async def api_pay( p, c, request: Request, external_id: str ) -> LnurlPayResponse | LnurlErrorResponse: - card, ctr_int, reason = await _authenticate_tap(external_id, p, c) + # 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) if not card: - return LnurlErrorResponse(reason=reason or "Card not found.") + 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) # 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.) - client = _client_info(request) - if not client: + if not request.client: return LnurlErrorResponse(reason="Cannot get client info.") - ip, agent = client + 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 "" hit = await create_hit(card.id, ip, agent, card.counter, ctr_int) callback_url = parse_obj_as( @@ -440,95 +406,29 @@ 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): - card, _ctr_int, reason = await _authenticate_tap(external_id, p, c) + p = p.upper() + c = c.upper() + card = await get_card_by_external_id(external_id) if not card: - return {"authenticated": False, "reason": reason} + 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": 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, - }