mirror of
https://github.com/lnbits/lnbits-legend.git
synced 2025-02-25 07:07:48 +01:00
221 lines
6.8 KiB
Python
221 lines
6.8 KiB
Python
import json
|
|
import secrets
|
|
from http import HTTPStatus
|
|
from urllib.parse import urlparse
|
|
|
|
from fastapi import HTTPException, Query, Request
|
|
from lnurl import encode as lnurl_encode
|
|
from lnurl.types import LnurlPayMetadata
|
|
from starlette.responses import HTMLResponse
|
|
|
|
from lnbits import bolt11
|
|
from lnbits.core.services import create_invoice
|
|
from lnbits.core.views.api import pay_invoice
|
|
|
|
from . import boltcards_ext
|
|
from .crud import (
|
|
create_hit,
|
|
get_card,
|
|
get_card_by_external_id,
|
|
get_card_by_otp,
|
|
get_hit,
|
|
get_hits_today,
|
|
spend_hit,
|
|
update_card_counter,
|
|
update_card_otp,
|
|
)
|
|
from .nxp424 import decryptSUN, getSunMAC
|
|
|
|
###############LNURLWITHDRAW#################
|
|
|
|
# /boltcards/api/v1/scan?p=00000000000000000000000000000000&c=0000000000000000
|
|
@boltcards_ext.get("/api/v1/scan/{external_id}")
|
|
async def api_scan(p, c, request: Request, external_id: str = Query(None)):
|
|
# 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 {"status": "ERROR", "reason": "No card."}
|
|
if not card.enable:
|
|
return {"status": "ERROR", "reason": "Card is disabled."}
|
|
try:
|
|
card_uid, counter = decryptSUN(bytes.fromhex(p), bytes.fromhex(card.k1))
|
|
if card.uid.upper() != card_uid.hex().upper():
|
|
return {"status": "ERROR", "reason": "Card UID mis-match."}
|
|
if c != getSunMAC(card_uid, counter, bytes.fromhex(card.k2)).hex().upper():
|
|
return {"status": "ERROR", "reason": "CMAC does not check."}
|
|
except:
|
|
return {"status": "ERROR", "reason": "Error decrypting card."}
|
|
|
|
ctr_int = int.from_bytes(counter, "little")
|
|
|
|
if ctr_int <= card.counter:
|
|
return {"status": "ERROR", "reason": "This link is already used."}
|
|
|
|
await update_card_counter(ctr_int, card.id)
|
|
|
|
# gathering some info for hit record
|
|
assert request.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 ""
|
|
todays_hits = await get_hits_today(card.id)
|
|
|
|
hits_amount = 0
|
|
for hit in todays_hits:
|
|
hits_amount = hits_amount + hit.amount
|
|
if hits_amount > card.daily_limit:
|
|
return {"status": "ERROR", "reason": "Max daily limit spent."}
|
|
hit = await create_hit(card.id, ip, agent, card.counter, ctr_int)
|
|
lnurlpay = lnurl_encode(request.url_for("boltcards.lnurlp_response", hit_id=hit.id))
|
|
return {
|
|
"tag": "withdrawRequest",
|
|
"callback": request.url_for("boltcards.lnurl_callback", hitid=hit.id),
|
|
"k1": hit.id,
|
|
"minWithdrawable": 1 * 1000,
|
|
"maxWithdrawable": card.tx_limit * 1000,
|
|
"defaultDescription": f"Boltcard (refund address lnurl://{lnurlpay})",
|
|
}
|
|
|
|
|
|
@boltcards_ext.get(
|
|
"/api/v1/lnurl/cb/{hitid}",
|
|
status_code=HTTPStatus.OK,
|
|
name="boltcards.lnurl_callback",
|
|
)
|
|
async def lnurl_callback(
|
|
pr: str = Query(None),
|
|
k1: str = Query(None),
|
|
):
|
|
if not k1:
|
|
return {"status": "ERROR", "reason": "Missing K1 token"}
|
|
|
|
hit = await get_hit(k1)
|
|
|
|
if not hit:
|
|
return {
|
|
"status": "ERROR",
|
|
"reason": "Record not found for this charge (bad k1)",
|
|
}
|
|
if hit.spent:
|
|
return {"status": "ERROR", "reason": "Payment already claimed"}
|
|
if not pr:
|
|
return {"status": "ERROR", "reason": "Missing payment request"}
|
|
|
|
try:
|
|
invoice = bolt11.decode(pr)
|
|
except:
|
|
return {"status": "ERROR", "reason": "Failed to decode payment request"}
|
|
|
|
card = await get_card(hit.card_id)
|
|
assert card
|
|
hit = await spend_hit(id=hit.id, amount=int(invoice.amount_msat / 1000))
|
|
assert hit
|
|
try:
|
|
await pay_invoice(
|
|
wallet_id=card.wallet,
|
|
payment_request=pr,
|
|
max_sat=card.tx_limit,
|
|
extra={"tag": "boltcard", "tag": hit.id},
|
|
)
|
|
return {"status": "OK"}
|
|
except Exception as exc:
|
|
return {"status": "ERROR", "reason": f"Payment failed - {exc}"}
|
|
|
|
|
|
# /boltcards/api/v1/auth?a=00000000000000000000000000000000
|
|
@boltcards_ext.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
|
|
|
|
|
|
###############LNURLPAY REFUNDS#################
|
|
|
|
|
|
@boltcards_ext.get(
|
|
"/api/v1/lnurlp/{hit_id}",
|
|
response_class=HTMLResponse,
|
|
name="boltcards.lnurlp_response",
|
|
)
|
|
async def lnurlp_response(req: Request, hit_id: str = Query(None)):
|
|
hit = await get_hit(hit_id)
|
|
assert hit
|
|
card = await get_card(hit.card_id)
|
|
assert card
|
|
if not hit:
|
|
return {"status": "ERROR", "reason": f"LNURL-pay record not found."}
|
|
if not card.enable:
|
|
return {"status": "ERROR", "reason": "Card is disabled."}
|
|
payResponse = {
|
|
"tag": "payRequest",
|
|
"callback": req.url_for("boltcards.lnurlp_callback", hit_id=hit_id),
|
|
"metadata": LnurlPayMetadata(json.dumps([["text/plain", "Refund"]])),
|
|
"minSendable": 1 * 1000,
|
|
"maxSendable": card.tx_limit * 1000,
|
|
}
|
|
return json.dumps(payResponse)
|
|
|
|
|
|
@boltcards_ext.get(
|
|
"/api/v1/lnurlp/cb/{hit_id}",
|
|
response_class=HTMLResponse,
|
|
name="boltcards.lnurlp_callback",
|
|
)
|
|
async def lnurlp_callback(hit_id: str = Query(None), amount: str = Query(None)):
|
|
hit = await get_hit(hit_id)
|
|
assert hit
|
|
card = await get_card(hit.card_id)
|
|
assert card
|
|
if not hit:
|
|
return {"status": "ERROR", "reason": f"LNURL-pay record not found."}
|
|
|
|
_, payment_request = 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},
|
|
)
|
|
|
|
payResponse = {"pr": payment_request, "routes": []}
|
|
|
|
return json.dumps(payResponse)
|