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."

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).
+    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 LnurlErrorResponse(reason="Card not found.")
+        return LnurlErrorResponse(reason=reason or "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.)
-    if not request.client:
+    client = _client_info(request)
+    if not client:
         return LnurlErrorResponse(reason="Cannot get client info.")
-    ip = request.client.host
+    ip, agent = client
-    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(
@@ -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()
+    card, _ctr_int, reason = await _authenticate_tap(external_id, p, c)
-    c = c.upper()
-    card = await get_card_by_external_id(external_id)
     if not card:
-        return {"authenticated": False, "reason": "Card not found."}
+        return {"authenticated": False, "reason": reason}
-    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,
+    }