feat(lnurl): /session — one verified tap for a terminal visit #1

Merged
padreug merged 2 commits from feat/card-session-endpoint into main 2026-09-20 15:16:37 +00:00
Showing only changes of commit 3e614428c0 - Show all commits

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 <noreply@anthropic.com>
Padreug 2026-09-20 17:01:14 +02:00

View file

@ -5,7 +5,10 @@ from urllib.parse import urlparse
import bolt11 import bolt11
from fastapi import APIRouter, HTTPException, Query, Request 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.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 ( from lnurl import (
CallbackUrl, CallbackUrl,
LightningInvoice, LightningInvoice,
@ -33,7 +36,7 @@ from .crud import (
update_card_counter, update_card_counter,
update_card_otp, update_card_otp,
) )
from .models import UIDPost from .models import Card, UIDPost
from .nxp424 import decrypt_sun, get_sun_mac from .nxp424 import decrypt_sun, get_sun_mac
boltcards_lnurl_router = APIRouter() 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)################# ###############LNURLPAY TAP-TO-RECEIVE (top-up)#################
# Deposit sats to a card's wallet by tapping the card — the receive/cash-in # 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 # 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( async def api_pay(
p, c, request: Request, external_id: str p, c, request: Request, external_id: str
) -> LnurlPayResponse | LnurlErrorResponse: ) -> 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: 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. # 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.) # (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.") 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) hit = await create_hit(card.id, ip, agent, card.counter, ctr_int)
callback_url = parse_obj_as( callback_url = parse_obj_as(
@ -406,29 +440,13 @@ async def pay_callback(
# the card-programming OTP endpoint. # the card-programming OTP endpoint.
@boltcards_lnurl_router.get("/api/v1/verify/{external_id}") @boltcards_lnurl_router.get("/api/v1/verify/{external_id}")
async def api_verify(p, c, external_id: str): 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: 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 { return {
"authenticated": True, "authenticated": True,
"external_id": card.external_id, "external_id": card.external_id,
"card_name": card.card_name, "card_name": card.card_name,
} }