From 3e614428c09545c128b6bbac34087c3a7e443eb8 Mon Sep 17 00:00:00 2001 From: Padreug Date: Sun, 20 Sep 2026 17:01:14 +0200 Subject: [PATCH] refactor(lnurl): share tap authentication across the fork endpoints /pay and /verify each carried a copy of /scan's card lookup + SUN decrypt/CMAC/replay checks + counter advance. Extract _authenticate_tap() (same checks, same order, same reasons) and _client_info() so the fork endpoints can't drift from upstream's acceptance rules. /scan itself is untouched (upstream code). No behaviour change. Co-Authored-By: Claude Fable 5.1 --- views_lnurl.py | 112 ++++++++++++++++++++++++++++--------------------- 1 file changed, 65 insertions(+), 47 deletions(-) diff --git a/views_lnurl.py b/views_lnurl.py index 54d5915..aabd365 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,13 @@ 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, } +