feat: Adds stripe tap to pay flow for TPoS running on Android PoS/phone devices (#3334)

Co-authored-by: Vlad Stan <stan.v.vlad@gmail.com>

lnbits/core/services/payments.py

@@ -139,7 +139,9 @@ async def create_fiat_invoice(
         payment_hash=internal_payment.payment_hash,
         currency=invoice_data.unit,
         memo=invoice_data.memo,
+        extra=invoice_data.extra or {},
     )
+
     if fiat_invoice.failed:
         logger.warning(fiat_invoice.error_message)
         internal_payment.status = PaymentState.FAILED

lnbits/core/views/fiat_api.py

@@ -1,10 +1,11 @@
 from http import HTTPStatus
 
-from fastapi import APIRouter, Depends
+from fastapi import APIRouter, Depends, HTTPException
 
 from lnbits.core.models.misc import SimpleStatus
 from lnbits.core.services.fiat_providers import test_connection
 from lnbits.decorators import check_admin
+from lnbits.fiat import StripeWallet, get_fiat_provider
 
 fiat_router = APIRouter(tags=["Fiat API"], prefix="/api/v1/fiat")
 
@@ -16,3 +17,29 @@ fiat_router = APIRouter(tags=["Fiat API"], prefix="/api/v1/fiat")
 )
 async def api_test_fiat_provider(provider: str) -> SimpleStatus:
     return await test_connection(provider)
+
+
+@fiat_router.post(
+    "/{provider}/connection_token",
+    status_code=HTTPStatus.OK,
+    dependencies=[Depends(check_admin)],
+)
+async def connection_token(provider: str):
+    provider_wallet = await get_fiat_provider(provider)
+    if provider == "stripe":
+        if not isinstance(provider_wallet, StripeWallet):
+            raise HTTPException(
+                status_code=500, detail="Stripe wallet/provider not configured"
+            )
+        try:
+            tok = await provider_wallet.create_terminal_connection_token()
+            secret = tok.get("secret")
+            if not secret:
+                raise HTTPException(
+                    status_code=502, detail="Stripe returned no connection token"
+                )
+            return {"secret": secret}
+        except Exception as e:
+            raise HTTPException(
+                status_code=500, detail="Failed to create connection token"
+            ) from e

lnbits/fiat/base.py

@@ -2,7 +2,7 @@ from __future__ import annotations
 
 from abc import ABC, abstractmethod
 from collections.abc import AsyncGenerator, Coroutine
-from typing import TYPE_CHECKING, NamedTuple
+from typing import TYPE_CHECKING, Any, NamedTuple
 
 if TYPE_CHECKING:
     pass
@@ -106,6 +106,7 @@ class FiatProvider(ABC):
         payment_hash: str,
         currency: str,
         memo: str | None = None,
+        extra: dict[str, Any] | None = None,
         **kwargs,
     ) -> Coroutine[None, None, FiatInvoiceResponse]:
         pass

lnbits/fiat/stripe.py

@@ -2,10 +2,12 @@ import asyncio
 import json
 from collections.abc import AsyncGenerator
 from datetime import datetime, timedelta, timezone
+from typing import Any, Literal
 from urllib.parse import urlencode
 
 import httpx
 from loguru import logger
+from pydantic import BaseModel, Field, ValidationError
 
 from lnbits.helpers import normalize_endpoint
 from lnbits.settings import settings
@@ -21,6 +23,34 @@ from .base import (
     FiatStatusResponse,
 )
 
+FiatMethod = Literal["checkout", "terminal"]
+
+
+class StripeTerminalOptions(BaseModel):
+    class Config:
+        extra = "ignore"
+
+    capture_method: Literal["automatic", "manual"] = "automatic"
+    metadata: dict[str, str] = Field(default_factory=dict)
+
+
+class StripeCheckoutOptions(BaseModel):
+    class Config:
+        extra = "ignore"
+
+    success_url: str | None = None
+    metadata: dict[str, str] = Field(default_factory=dict)
+    line_item_name: str | None = None
+
+
+class StripeCreateInvoiceOptions(BaseModel):
+    class Config:
+        extra = "ignore"
+
+    fiat_method: FiatMethod = "checkout"
+    terminal: StripeTerminalOptions | None = None
+    checkout: StripeCheckoutOptions | None = None
+
 
 class StripeWallet(FiatProvider):
     """https://docs.stripe.com/api"""
@@ -30,9 +60,9 @@ class StripeWallet(FiatProvider):
         self._settings_fields = self._settings_connection_fields()
         if not settings.stripe_api_endpoint:
             raise ValueError("Cannot initialize StripeWallet: missing endpoint.")
-
         if not settings.stripe_api_secret_key:
             raise ValueError("Cannot initialize StripeWallet: missing API secret key.")
+
         self.endpoint = normalize_endpoint(settings.stripe_api_endpoint)
         self.headers = {
             "Authorization": f"Bearer {settings.stripe_api_secret_key}",
@@ -60,8 +90,10 @@ class StripeWallet(FiatProvider):
             r.raise_for_status()
             data = r.json()
 
-            available_balance = data.get("available", [{}])[0].get("amount", 0)
+            available = data.get("available") or []
-            # pending_balance = data.get("pending", {}).get("amount", 0)
+            available_balance = 0
+            if available and isinstance(available, list):
+                available_balance = int(available[0].get("amount", 0))
 
             return FiatStatusResponse(balance=available_balance)
         except json.JSONDecodeError:
@@ -76,81 +108,47 @@ class StripeWallet(FiatProvider):
         payment_hash: str,
         currency: str,
         memo: str | None = None,
+        extra: dict[str, Any] | None = None,
         **kwargs,
     ) -> FiatInvoiceResponse:
         amount_cents = int(amount * 100)
-        form_data = [
+        opts = self._parse_create_opts(extra or {})
-            ("mode", "payment"),
+        if not opts:
-            (
+            return FiatInvoiceResponse(ok=False, error_message="Invalid Stripe options")
-                "success_url",
-                settings.stripe_payment_success_url or "https://lnbits.com",
-            ),
-            ("metadata[payment_hash]", payment_hash),
-            ("line_items[0][price_data][currency]", currency.lower()),
-            ("line_items[0][price_data][product_data][name]", memo or "LNbits Invoice"),
-            ("line_items[0][price_data][unit_amount]", amount_cents),
-            ("line_items[0][quantity]", "1"),
-        ]
-        encoded_data = urlencode(form_data)
 
-        try:
+        if opts.fiat_method == "checkout":
-            headers = self.headers.copy()
+            return await self._create_checkout_invoice(
-            headers["Content-Type"] = "application/x-www-form-urlencoded"
+                amount_cents, currency, payment_hash, memo, opts
-            r = await self.client.post(
+            )
-                url="/v1/checkout/sessions", headers=headers, content=encoded_data
+        if opts.fiat_method == "terminal":
+            return await self._create_terminal_invoice(
+                amount_cents, currency, payment_hash, opts
             )
-            r.raise_for_status()
-            data = r.json()
 
-            session_id = data.get("id")
+        return FiatInvoiceResponse(
-            if not session_id:
+            ok=False, error_message=f"Unsupported fiat_method: {opts.fiat_method}"
-                return FiatInvoiceResponse(
+        )
-                    ok=False, error_message="Server error: 'missing session id'"
-                )
-            payment_request = data.get("url")
-            if not payment_request:
-                return FiatInvoiceResponse(
-                    ok=False, error_message="Server error: 'missing payment URL'"
-                )
-
-            return FiatInvoiceResponse(
-                ok=True, checking_id=session_id, payment_request=payment_request
-            )
-        except json.JSONDecodeError:
-            return FiatInvoiceResponse(
-                ok=False, error_message="Server error: 'invalid json response'"
-            )
-        except Exception as exc:
-            logger.warning(exc)
-            return FiatInvoiceResponse(
-                ok=False, error_message=f"Unable to connect to {self.endpoint}."
-            )
 
     async def pay_invoice(self, payment_request: str) -> FiatPaymentResponse:
         raise NotImplementedError("Stripe does not support paying invoices directly.")
 
     async def get_invoice_status(self, checking_id: str) -> FiatPaymentStatus:
         try:
-            r = await self.client.get(
+            stripe_id = self._normalize_stripe_id(checking_id)
-                url=f"/v1/checkout/sessions/{checking_id}",
-            )
-            r.raise_for_status()
 
-            data = r.json()
+            if stripe_id.startswith("cs_"):
-            payment_status = data.get("payment_status")
+                r = await self.client.get(f"/v1/checkout/sessions/{stripe_id}")
-            if not payment_status:
+                r.raise_for_status()
-                return FiatPaymentPendingStatus()
+                return self._status_from_checkout_session(r.json())
-            if payment_status == "paid":
-                # todo: handle fee
-                return FiatPaymentSuccessStatus()
 
-            expires_at = data.get("expires_at")
+            if stripe_id.startswith("pi_"):
-            _24_hours_ago = datetime.now(timezone.utc) - timedelta(hours=24)
+                r = await self.client.get(f"/v1/payment_intents/{stripe_id}")
-            if expires_at and expires_at < _24_hours_ago.timestamp():
+                r.raise_for_status()
-                # be defensive: add a 24 hour buffer
+                return self._status_from_payment_intent(r.json())
-                return FiatPaymentFailedStatus()
 
+            logger.debug(f"Unknown Stripe id prefix: {checking_id}")
             return FiatPaymentPendingStatus()
+
         except Exception as exc:
             logger.debug(f"Error getting invoice status: {exc}")
             return FiatPaymentPendingStatus()
@@ -167,6 +165,169 @@ class StripeWallet(FiatProvider):
             value = await mock_queue.get()
             yield value
 
+    async def create_terminal_connection_token(self) -> dict:
+        r = await self.client.post("/v1/terminal/connection_tokens")
+        r.raise_for_status()
+        return r.json()
+
+    async def _create_checkout_invoice(
+        self,
+        amount_cents: int,
+        currency: str,
+        payment_hash: str,
+        memo: str | None,
+        opts: StripeCreateInvoiceOptions,
+    ) -> FiatInvoiceResponse:
+        co = opts.checkout or StripeCheckoutOptions()
+        success_url = (
+            co.success_url
+            or settings.stripe_payment_success_url
+            or "https://lnbits.com"
+        )
+        line_item_name = co.line_item_name or memo or "LNbits Invoice"
+
+        form_data: list[tuple[str, str]] = [
+            ("mode", "payment"),
+            ("success_url", success_url),
+            ("metadata[payment_hash]", payment_hash),
+            ("line_items[0][price_data][currency]", currency.lower()),
+            ("line_items[0][price_data][product_data][name]", line_item_name),
+            ("line_items[0][price_data][unit_amount]", str(amount_cents)),
+            ("line_items[0][quantity]", "1"),
+        ]
+        form_data += self._encode_metadata("metadata", co.metadata)
+
+        try:
+            r = await self.client.post(
+                "/v1/checkout/sessions",
+                headers=self._build_headers_form(),
+                content=urlencode(form_data),
+            )
+            r.raise_for_status()
+            data = r.json()
+            session_id, url = data.get("id"), data.get("url")
+            if not session_id or not url:
+                return FiatInvoiceResponse(
+                    ok=False, error_message="Server error: missing id or url"
+                )
+            return FiatInvoiceResponse(
+                ok=True, checking_id=session_id, payment_request=url
+            )
+        except json.JSONDecodeError:
+            return FiatInvoiceResponse(
+                ok=False, error_message="Server error: invalid json response"
+            )
+        except Exception as exc:
+            logger.warning(exc)
+            return FiatInvoiceResponse(
+                ok=False, error_message=f"Unable to connect to {self.endpoint}."
+            )
+
+    async def _create_terminal_invoice(
+        self,
+        amount_cents: int,
+        currency: str,
+        payment_hash: str,
+        opts: StripeCreateInvoiceOptions,
+    ) -> FiatInvoiceResponse:
+        term = opts.terminal or StripeTerminalOptions()
+        data: dict[str, str] = {
+            "amount": str(amount_cents),
+            "currency": currency.lower(),
+            "payment_method_types[]": "card_present",
+            "capture_method": term.capture_method,
+            "metadata[payment_hash]": payment_hash,
+            "metadata[source]": "lnbits",
+        }
+        for k, v in (term.metadata or {}).items():
+            data[f"metadata[{k}]"] = str(v)
+
+        try:
+            r = await self.client.post("/v1/payment_intents", data=data)
+            r.raise_for_status()
+            pi = r.json()
+            pi_id, client_secret = pi.get("id"), pi.get("client_secret")
+            if not pi_id or not client_secret:
+                return FiatInvoiceResponse(
+                    ok=False,
+                    error_message="Error: missing PaymentIntent or client_secret",
+                )
+            return FiatInvoiceResponse(
+                ok=True, checking_id=pi_id, payment_request=client_secret
+            )
+        except json.JSONDecodeError:
+            return FiatInvoiceResponse(
+                ok=False, error_message="Error: invalid json response"
+            )
+        except Exception as exc:
+            logger.warning(exc)
+            return FiatInvoiceResponse(
+                ok=False, error_message=f"Unable to connect to {self.endpoint}."
+            )
+
+    def _normalize_stripe_id(self, checking_id: str) -> str:
+        """Remove our internal prefix so Stripe sees a real id."""
+        return (
+            checking_id.replace("fiat_stripe_", "", 1)
+            if checking_id.startswith("fiat_stripe_")
+            else checking_id
+        )
+
+    def _status_from_checkout_session(self, data: dict) -> FiatPaymentStatus:
+        """Map a Checkout Session to LNbits fiat status."""
+        if data.get("payment_status") == "paid":
+            return FiatPaymentSuccessStatus()
+
+        # Consider an expired session a fail (existing 24h rule).
+        expires_at = data.get("expires_at")
+        _24h_ago = datetime.now(timezone.utc) - timedelta(hours=24)
+        if expires_at and float(expires_at) < _24h_ago.timestamp():
+            return FiatPaymentFailedStatus()
+
+        return FiatPaymentPendingStatus()
+
+    def _status_from_payment_intent(self, pi: dict) -> FiatPaymentStatus:
+        """Map a PaymentIntent to LNbits fiat status (card_present friendly)."""
+        status = pi.get("status")
+
+        if status == "succeeded":
+            return FiatPaymentSuccessStatus()
+
+        if status in ("canceled", "payment_failed"):
+            return FiatPaymentFailedStatus()
+
+        if status == "requires_payment_method":
+            if pi.get("last_payment_error"):
+                return FiatPaymentFailedStatus()
+
+            now_ts = datetime.now(timezone.utc).timestamp()
+            created_ts = float(pi.get("created") or now_ts)
+            is_stale = (now_ts - created_ts) > 300
+            if is_stale:
+                return FiatPaymentFailedStatus()
+
+        return FiatPaymentPendingStatus()
+
+    def _build_headers_form(self) -> dict[str, str]:
+        return {**self.headers, "Content-Type": "application/x-www-form-urlencoded"}
+
+    def _encode_metadata(
+        self, prefix: str, md: dict[str, Any]
+    ) -> list[tuple[str, str]]:
+        out: list[tuple[str, str]] = []
+        for k, v in (md or {}).items():
+            out.append((f"{prefix}[{k}]", str(v)))
+        return out
+
+    def _parse_create_opts(
+        self, raw_opts: dict[str, Any]
+    ) -> StripeCreateInvoiceOptions | None:
+        try:
+            return StripeCreateInvoiceOptions.parse_obj(raw_opts)
+        except ValidationError as e:
+            logger.warning(f"Invalid Stripe options: {e}")
+            return None
+
     def _settings_connection_fields(self) -> str:
         return "-".join(
             [str(settings.stripe_api_endpoint), str(settings.stripe_api_secret_key)]