views_lnurl.py

import json
import math
from http import HTTPStatus

from fastapi import APIRouter
from loguru import logger
from starlette.requests import Request

from .crud import (
    create_bleskomat_lnurl,
    get_bleskomat_by_api_key_id,
    get_bleskomat_lnurl,
)
from .exchange_rates import fetch_fiat_exchange_rate
from .helpers import (
    LnurlHttpError,
    LnurlValidationError,
    generate_bleskomat_lnurl_secret,
    generate_bleskomat_lnurl_signature,
    prepare_lnurl_params,
    query_to_signing_payload,
    unshorten_lnurl_query,
)

bleskomat_lnurl_router = APIRouter()


# Handles signed URL from Bleskomat ATMs and "action" callback of auto-generated LNURLs.
@bleskomat_lnurl_router.get("/u", name="bleskomat.api_bleskomat_lnurl")
async def api_bleskomat_lnurl(req: Request):
    try:
        query = dict(req.query_params)

        # Unshorten query if "s" is used instead of "signature".
        if "s" in query:
            query = unshorten_lnurl_query(query)

        if "signature" in query:

            # Signature provided. Use signature to verify that the URL was generated by
            # an authorized device. Later validate parameters, auto-generate LNURL,
            # reply with LNURL response object.
            signature = query["signature"]

            # The API key ID, nonce, and tag should be present in the query string.
            for field in ["id", "nonce", "tag"]:
                if field not in query:
                    raise LnurlHttpError(
                        f'Failed API key signature check: Missing "{field}"',
                        HTTPStatus.BAD_REQUEST,
                    )

            # URL signing scheme is described here:
            # https://github.com/chill117/lnurl-node#how-to-implement-url-signing-scheme
            payload = query_to_signing_payload(query)
            api_key_id = query["id"]
            bleskomat = await get_bleskomat_by_api_key_id(api_key_id)
            if not bleskomat:
                raise LnurlHttpError("Unknown API key", HTTPStatus.BAD_REQUEST)
            api_key_secret = bleskomat.api_key_secret
            api_key_encoding = bleskomat.api_key_encoding
            expected_signature = generate_bleskomat_lnurl_signature(
                payload, api_key_secret, api_key_encoding
            )
            if signature != expected_signature:
                raise LnurlHttpError("Invalid API key signature", HTTPStatus.FORBIDDEN)

            # Signature is valid.
            # In the case of signed URLs, the secret is deterministic based on
            # the API key ID and signature.
            secret = generate_bleskomat_lnurl_secret(api_key_id, signature)
            lnurl = await get_bleskomat_lnurl(secret)
            if not lnurl:
                try:
                    tag = query["tag"]
                    params = prepare_lnurl_params(tag, query)
                    if "f" in query:
                        rate = await fetch_fiat_exchange_rate(
                            currency=query["f"],
                            provider=bleskomat.exchange_rate_provider,
                        )
                        # Convert fee (%) to decimal:
                        fee = float(bleskomat.fee) / 100
                        if tag == "withdrawRequest":
                            for key in ["minWithdrawable", "maxWithdrawable"]:
                                amount_sats = int(
                                    math.floor((params[key] / rate) * 1e8)
                                )
                                fee_sats = int(math.floor(amount_sats * fee))
                                amount_sats_less_fee = amount_sats - fee_sats
                                # Convert to msats:
                                params[key] = int(amount_sats_less_fee * 1e3)
                except LnurlValidationError as exc:
                    raise LnurlHttpError(str(exc), HTTPStatus.BAD_REQUEST) from exc
                # Create a new LNURL using query parameters provided in the signed URL.
                json_params = json.JSONEncoder().encode(params)
                lnurl = await create_bleskomat_lnurl(
                    bleskomat=bleskomat,
                    secret=secret,
                    tag=tag,
                    params=json_params,
                    uses=1,
                )

            # Reply with LNURL response object.
            return lnurl.get_info_response_object(secret, req)

        # No signature provided.
        # Treat as "action" callback.

        if "k1" not in query:
            raise LnurlHttpError("Missing secret", HTTPStatus.BAD_REQUEST)

        secret = query["k1"]
        lnurl = await get_bleskomat_lnurl(secret)
        if not lnurl:
            raise LnurlHttpError("Invalid secret", HTTPStatus.BAD_REQUEST)

        if not lnurl.has_uses_remaining():
            raise LnurlHttpError(
                "Maximum number of uses already reached", HTTPStatus.BAD_REQUEST
            )

        try:
            await lnurl.execute_action(query)
        except LnurlValidationError as exc:
            raise LnurlHttpError(str(exc), HTTPStatus.BAD_REQUEST) from exc

    except LnurlHttpError as e:
        return {"status": "ERROR", "reason": str(e)}
    except Exception as e:
        logger.error(str(e))
        return {"status": "ERROR", "reason": "Unexpected error"}

    return {"status": "OK"}