boltcards/views_lnurl.py

import json
import secrets
from http import HTTPStatus
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.cache import cache
from lnurl import (
    CallbackUrl,
    LightningInvoice,
    LnurlErrorResponse,
    LnurlPayActionResponse,
    LnurlPayMetadata,
    LnurlPayResponse,
    LnurlSuccessResponse,
    LnurlWithdrawResponse,
    Max144Str,
    MessageAction,
    MilliSatoshi,
)
from pydantic import parse_obj_as

from .crud import (
    create_hit,
    get_card,
    get_card_by_external_id,
    get_card_by_otp,
    get_card_by_uid,
    get_hit,
    get_hits_today,
    spend_hit,
    update_card_counter,
    update_card_otp,
)
from .models import Card, UIDPost
from .nxp424 import decrypt_sun, get_sun_mac

boltcards_lnurl_router = APIRouter()


# /boltcards/api/v1/scan?p=00000000000000000000000000000000&c=0000000000000000
@boltcards_lnurl_router.get("/api/v1/scan/{external_id}")
async def api_scan(
    p, c, request: Request, external_id: str
) -> LnurlWithdrawResponse | LnurlErrorResponse:
    # some wallets send everything as lower case, no bueno
    p = p.upper()
    c = c.upper()
    card = None
    counter = b""
    card = await get_card_by_external_id(external_id)
    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)

    # gathering some info for hit record
    if not request.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 ""
    todays_hits = await get_hits_today(card.id)

    hits_amount = 0
    for hit in todays_hits:
        hits_amount += hit.amount
    if hits_amount > int(card.daily_limit):
        return LnurlErrorResponse(reason="Max daily limit spent.")
    hit = await create_hit(card.id, ip, agent, card.counter, ctr_int)

    # create a lud17 lnurlp to support lud19, add payLink field of the withdrawRequest
    lnurlpay_url = str(request.url_for("boltcards.lnurlp_response", hit_id=hit.id))
    pay_link = lnurlpay_url.replace("http://", "lnurlp://").replace(
        "https://", "lnurlp://"
    )
    callback_url = parse_obj_as(
        CallbackUrl, str(request.url_for("boltcards.lnurl_callback", hit_id=hit.id))
    )
    return LnurlWithdrawResponse(
        callback=callback_url,
        k1=hit.id,
        minWithdrawable=MilliSatoshi(1000),
        maxWithdrawable=MilliSatoshi(int(card.tx_limit) * 1000),
        defaultDescription=f"Boltcard (refund address {pay_link})",
        payLink=pay_link,  # type: ignore
    )


@boltcards_lnurl_router.get(
    "/api/v1/lnurl/cb/{hit_id}",
    status_code=HTTPStatus.OK,
    name="boltcards.lnurl_callback",
)
async def lnurl_callback(
    hit_id: str,
    k1: str = Query(None),
    pr: str = Query(None),
) -> LnurlErrorResponse | LnurlSuccessResponse:
    if not k1:
        return LnurlErrorResponse(reason="Missing K1 token")
    if k1 != hit_id:
        return LnurlErrorResponse(reason="K1 token does not match.")

    hit = await get_hit(hit_id)
    if not hit:
        return LnurlErrorResponse(reason="LNURL-withdraw record not found.")
    if hit.spent:
        return LnurlErrorResponse(reason="Payment already claimed.")
    if not pr:
        return LnurlErrorResponse(reason="Missing payment request.")

    try:
        invoice = bolt11.decode(pr)
    except bolt11.Bolt11Exception:
        return LnurlErrorResponse(reason="Failed to decode payment request.")
    if not invoice.amount_msat:
        return LnurlErrorResponse(reason="Invoice has no amount.")
    card = await get_card(hit.card_id)
    if not card:
        return LnurlErrorResponse(reason="Card not found.")
    hit = await spend_hit(card_id=hit.id, amount=int(invoice.amount_msat / 1000))
    if not hit:
        return LnurlErrorResponse(reason="Failed to update hit as spent.")
    try:
        await pay_invoice(
            wallet_id=card.wallet,
            payment_request=pr,
            max_sat=int(card.tx_limit),
            extra={"tag": "boltcards", "hit": hit.id},
        )
        return LnurlSuccessResponse()
    except Exception as exc:
        return LnurlErrorResponse(reason=f"Payment failed - {exc}")


# /boltcards/api/v1/auth?a=00000000000000000000000000000000
@boltcards_lnurl_router.get("/api/v1/auth")
async def api_auth(a, request: Request):
    if a == "00000000000000000000000000000000":
        response = {"k0": "0" * 32, "k1": "1" * 32, "k2": "2" * 32}
        return response

    card = await get_card_by_otp(a)
    if not card:
        raise HTTPException(
            detail="Card does not exist.", status_code=HTTPStatus.NOT_FOUND
        )

    new_otp = secrets.token_hex(16)
    await update_card_otp(new_otp, card.id)

    lnurlw_base = (
        f"{urlparse(str(request.url)).netloc}/boltcards/api/v1/scan/{card.external_id}"
    )

    response = {
        "card_name": card.card_name,
        "id": str(1),
        "k0": card.k0,
        "k1": card.k1,
        "k2": card.k2,
        "k3": card.k1,
        "k4": card.k2,
        "lnurlw_base": "lnurlw://" + lnurlw_base,
        "protocol_name": "new_bolt_card_response",
        "protocol_version": str(1),
    }

    return response


# /boltcards/api/v1/auth?a=00000000000000000000000000000000
@boltcards_lnurl_router.post("/api/v1/auth")
async def api_auth_post(a: str, request: Request, data: UIDPost, wipe: bool = False):
    card = None
    if wipe:
        card = await get_card_by_otp(a)
    else:
        if not data.UID:
            raise HTTPException(
                detail="Missing UID.", status_code=HTTPStatus.BAD_REQUEST
            )

        card = await get_card_by_uid(data.UID)
    if not card:
        raise HTTPException(
            detail="Card does not exist.", status_code=HTTPStatus.NOT_FOUND
        )
    new_otp = secrets.token_hex(16)
    await update_card_otp(new_otp, card.id)
    lnurlw_base = (
        f"{urlparse(str(request.url)).netloc}/boltcards/api/v1/scan/{card.external_id}"
    )
    response = {
        "CARD_NAME": card.card_name,
        "ID": str(1),
        "K0": card.k0,
        "K1": card.k1,
        "K2": card.k2,
        "K3": card.k1,
        "K4": card.k2,
        "LNURLW_BASE": "LNURLW://" + lnurlw_base,
        "LNURLW": "LNURLW://" + lnurlw_base,
        "PROTOCOL_NAME": "NEW_BOLT_CARD_RESPONSE",
        "PROTOCOL_VERSION": str(1),
    }
    if wipe:
        response["action"] = "wipe"
    return response


###############LNURLPAY REFUNDS#################
@boltcards_lnurl_router.get(
    "/api/v1/lnurlp/cb/{hit_id}",
    name="boltcards.lnurlp_callback",
)
async def lnurlp_callback(
    hit_id: str, amount: str = Query(None)
) -> LnurlPayActionResponse | LnurlErrorResponse:
    hit = await get_hit(hit_id)
    if not hit:
        return LnurlErrorResponse(reason="LNURL-pay record not found.")
    card = await get_card(hit.card_id)
    if not card:
        return LnurlErrorResponse(reason="Card not found.")
    if not card.enable:
        return LnurlErrorResponse(reason="Card is disabled.")
    if not amount:
        return LnurlErrorResponse(reason="Missing amount.")
    if int(amount) < 1000:
        return LnurlErrorResponse(reason="Amount too low.")
    if int(amount) > int(card.tx_limit) * 1000:
        return LnurlErrorResponse(reason="Amount too high.")

    payment = await create_invoice(
        wallet_id=card.wallet,
        amount=int(int(amount) / 1000),
        memo=f"Refund {hit_id}",
        unhashed_description=LnurlPayMetadata(
            json.dumps([["text/plain", "Refund"]])
        ).encode(),
        extra={"refund": hit_id},
    )
    action = MessageAction(message=Max144Str("Refunded!"))
    invoice = parse_obj_as(LightningInvoice, payment.bolt11)
    return LnurlPayActionResponse(pr=invoice, successAction=action)


@boltcards_lnurl_router.get(
    "/api/v1/lnurlp/{hit_id}",
    name="boltcards.lnurlp_response",
)
async def lnurlp_response(
    req: Request, hit_id: str
) -> LnurlPayResponse | LnurlErrorResponse:
    hit = await get_hit(hit_id)
    if not hit:
        return LnurlErrorResponse(reason="LNURL-pay hit not found.")
    card = await get_card(hit.card_id)
    if not card:
        return LnurlErrorResponse(reason="Card not found.")
    if not card.enable:
        return LnurlErrorResponse(reason="Card is disabled.")
    callback_url = parse_obj_as(
        CallbackUrl, str(req.url_for("boltcards.lnurlp_callback", hit_id=hit_id))
    )
    return LnurlPayResponse(
        callback=callback_url,
        minSendable=MilliSatoshi(1000),
        maxSendable=MilliSatoshi(int(card.tx_limit) * 1000),
        metadata=LnurlPayMetadata(json.dumps([["text/plain", "Refund"]])),
    )


###############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
# voucher), so the tap is used purely as an authenticated identity: the same
# SUN p/c that /scan verifies proves card possession, and we return an
# lnurl-PAY response (LUD-06) for the card's own wallet instead of a withdraw
# voucher. The single-use `hit` acts as the bearer token for the callback,
# mirroring how `k1` bridges the two withdraw steps. Unlike the LUD-19 refund
# lnurlp (keyed by a prior scan's hit), this is reachable directly by a tap.

# The pay metadata MUST be byte-identical between the response below and the
# callback's unhashed_description, or the invoice's description_hash won't match
# (LUD-06). Keep it static.
_TOPUP_METADATA = json.dumps([["text/plain", "Bolt Card top-up"]])


# /boltcards/api/v1/pay/{external_id}?p=<32-hex>&c=<16-hex>  (mirrors /scan)
@boltcards_lnurl_router.get(
    "/api/v1/pay/{external_id}",
    name="boltcards.pay_response",
)
async def api_pay(
    p, c, request: Request, external_id: str
) -> LnurlPayResponse | LnurlErrorResponse:
    card, ctr_int, reason = await _authenticate_tap(external_id, p, c)
    if not card:
        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.)
    client = _client_info(request)
    if not client:
        return LnurlErrorResponse(reason="Cannot get client info.")
    ip, agent = client
    hit = await create_hit(card.id, ip, agent, card.counter, ctr_int)

    callback_url = parse_obj_as(
        CallbackUrl, str(request.url_for("boltcards.pay_callback", hit_id=hit.id))
    )
    return LnurlPayResponse(
        callback=callback_url,
        minSendable=MilliSatoshi(1000),
        maxSendable=MilliSatoshi(int(card.tx_limit) * 1000),
        metadata=LnurlPayMetadata(_TOPUP_METADATA),
    )


@boltcards_lnurl_router.get(
    "/api/v1/pay/cb/{hit_id}",
    name="boltcards.pay_callback",
)
async def pay_callback(
    hit_id: str, amount: str = Query(None)
) -> LnurlPayActionResponse | LnurlErrorResponse:
    hit = await get_hit(hit_id)
    if not hit:
        return LnurlErrorResponse(reason="LNURL-pay record not found.")
    card = await get_card(hit.card_id)
    if not card:
        return LnurlErrorResponse(reason="Card not found.")
    if not card.enable:
        return LnurlErrorResponse(reason="Card is disabled.")
    if not amount:
        return LnurlErrorResponse(reason="Missing amount.")
    if int(amount) < 1000:
        return LnurlErrorResponse(reason="Amount too low.")
    if int(amount) > int(card.tx_limit) * 1000:
        return LnurlErrorResponse(reason="Amount too high.")

    payment = await create_invoice(
        wallet_id=card.wallet,
        amount=int(int(amount) / 1000),
        memo=f"Top-up {card.card_name}",
        unhashed_description=LnurlPayMetadata(_TOPUP_METADATA).encode(),
        extra={"tag": "boltcards", "topup": hit_id},
    )
    action = MessageAction(message=Max144Str("Topped up!"))
    invoice = parse_obj_as(LightningInvoice, payment.bolt11)
    return LnurlPayActionResponse(pr=invoice, successAction=action)


###############ACCESS-CONTROL VERIFY (doors)#################
# /boltcards/api/v1/verify/{external_id}?p=<32-hex>&c=<16-hex>
# Side-effect-light SUN check for access control (e.g. doors via the `access`
# extension): confirm the tap is a genuine, non-replayed card and return its
# identity — WITHOUT /scan's spend semantics (no withdrawRequest, no daily-limit
# check, no "hit" record). It DOES advance the SUN counter, exactly like /scan,
# so a captured p/c can't be replayed. Named `verify` because `/auth` is already
# 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)
    if not card:
        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. The fiat figure comes ONLY from
    # LNbits' already-warm rate cache: this response gates the terminal
    # unlocking, and a cold lookup queries external exchanges (~1 s). Cache
    # miss → `fiat: null`; the terminal prices the sats itself.
    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:
        price = cache.get(f"btc-price-{currency}")  # fiat per BTC, as btc_price()
        if isinstance(price, (int, float)) and price > 0:
            fiat = balance_msat / 1000 / 100_000_000 * float(price)

    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,
    }