feat: basic square integration

This commit is contained in:
Vlad Stan
2026-05-13 15:23:29 +03:00
parent 810a13722c
commit 3850f561d6
11 changed files with 939 additions and 9 deletions
+31
View File
@@ -2,6 +2,7 @@ import hashlib
import hmac
import json
import time
from base64 import b64encode
import httpx
from loguru import logger
@@ -169,6 +170,36 @@ async def verify_paypal_webhook(headers, payload: bytes):
raise ValueError("PayPal webhook cannot be verified.") from exc
def check_square_signature(
payload: bytes,
sig_header: str | None,
secret: str | None,
notification_url: str | None,
):
if not sig_header:
logger.warning("Square signature header is missing.")
raise ValueError("Square signature header is missing.")
if not secret:
logger.warning("Square webhook signature key is not set.")
raise ValueError("Square webhook cannot be verified.")
if not notification_url:
logger.warning("Square webhook notification URL is not set.")
raise ValueError("Square webhook cannot be verified.")
signed_payload = notification_url.encode() + payload
computed_signature = b64encode(
hmac.new(
key=secret.encode(), msg=signed_payload, digestmod=hashlib.sha256
).digest()
).decode()
if hmac.compare_digest(computed_signature, sig_header) is not True:
logger.warning("Square signature verification failed.")
raise ValueError("Square signature verification failed.")
async def test_connection(provider: str) -> SimpleStatus:
"""
Test the connection to Stripe by checking if the API key is valid.
+50
View File
@@ -10,6 +10,7 @@ from lnbits.core.models.misc import SimpleStatus
from lnbits.core.models.payments import CreateInvoice
from lnbits.core.services.fiat_providers import (
check_fiat_status,
check_square_signature,
check_stripe_signature,
verify_paypal_webhook,
)
@@ -50,6 +51,23 @@ async def api_generic_webhook_handler(
message=f"Callback received successfully from '{provider_name}'.",
)
if provider_name.lower() == "square":
payload = await request.body()
sig_header = request.headers.get("x-square-hmacsha256-signature")
check_square_signature(
payload,
sig_header,
settings.square_webhook_signature_key,
settings.square_payment_webhook_url,
)
event = await request.json()
await handle_square_event(event)
return SimpleStatus(
success=True,
message=f"Callback received successfully from '{provider_name}'.",
)
return SimpleStatus(
success=False,
message=f"Unknown fiat provider '{provider_name}'.",
@@ -280,3 +298,35 @@ def _deserialize_paypal_metadata(custom_id: str) -> FiatSubscriptionPaymentOptio
except (json.JSONDecodeError, IndexError) as e:
logger.warning(f"Failed to deserialize PayPal metadata: {e}")
return FiatSubscriptionPaymentOptions()
async def handle_square_event(event: dict):
event_id = event.get("event_id") or event.get("id", "")
event_type = event.get("type", "")
logger.info(f"Handling Square event: '{event_id}'. Type: '{event_type}'.")
if event_type in ("payment.created", "payment.updated"):
await _handle_square_payment_event(event)
return
logger.warning(f"Unhandled Square event type: '{event_type}'.")
async def _handle_square_payment_event(event: dict):
payment = _square_extract_payment(event)
order_id = payment.get("order_id")
if not order_id:
logger.warning("Square payment event missing order_id.")
return
lnbits_payment = await get_standalone_payment(f"fiat_square_order_{order_id}")
if not lnbits_payment:
logger.warning(f"No payment found for Square order: '{order_id}'.")
return
await check_fiat_status(lnbits_payment)
def _square_extract_payment(event: dict) -> dict:
event_object = event.get("data", {}).get("object", {})
return event_object.get("payment") or event_object
+3
View File
@@ -9,6 +9,7 @@ from lnbits.fiat.base import FiatProvider
from lnbits.settings import settings
from .paypal import PayPalWallet
from .square import SquareWallet
from .stripe import StripeWallet
fiat_module = importlib.import_module("lnbits.fiat")
@@ -17,6 +18,7 @@ fiat_module = importlib.import_module("lnbits.fiat")
class FiatProviderType(Enum):
stripe = "StripeWallet"
paypal = "PayPalWallet"
square = "SquareWallet"
async def get_fiat_provider(name: str) -> FiatProvider | None:
@@ -52,5 +54,6 @@ fiat_providers: dict[str, FiatProvider] = {}
__all__ = [
"PayPalWallet",
"SquareWallet",
"StripeWallet",
]
+279
View File
@@ -0,0 +1,279 @@
import asyncio
import json
from collections.abc import AsyncGenerator
from typing import Any, Literal
import httpx
from loguru import logger
from pydantic import BaseModel, Field, ValidationError
from lnbits.helpers import normalize_endpoint
from lnbits.settings import settings
from .base import (
FiatInvoiceResponse,
FiatPaymentFailedStatus,
FiatPaymentPendingStatus,
FiatPaymentResponse,
FiatPaymentStatus,
FiatPaymentSuccessStatus,
FiatProvider,
FiatStatusResponse,
FiatSubscriptionPaymentOptions,
FiatSubscriptionResponse,
)
FiatMethod = Literal["checkout"]
class SquareCheckoutOptions(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 SquareCreateInvoiceOptions(BaseModel):
class Config:
extra = "ignore"
fiat_method: FiatMethod = "checkout"
checkout: SquareCheckoutOptions | None = None
class SquareWallet(FiatProvider):
"""https://developer.squareup.com/reference/square"""
def __init__(self):
logger.debug("Initializing SquareWallet")
self._settings_fields = self._settings_connection_fields()
if not settings.square_api_endpoint:
raise ValueError("Cannot initialize SquareWallet: missing endpoint.")
if not settings.square_access_token:
raise ValueError("Cannot initialize SquareWallet: missing access token.")
if not settings.square_location_id:
raise ValueError("Cannot initialize SquareWallet: missing location ID.")
self.endpoint = normalize_endpoint(settings.square_api_endpoint)
self.location_id = settings.square_location_id
self.headers = {
"Authorization": f"Bearer {settings.square_access_token}",
"Square-Version": settings.square_api_version,
"Content-Type": "application/json",
"User-Agent": settings.user_agent,
}
self.client = httpx.AsyncClient(base_url=self.endpoint, headers=self.headers)
logger.info("SquareWallet initialized.")
async def cleanup(self):
try:
await self.client.aclose()
except RuntimeError as e:
logger.warning(f"Error closing Square wallet connection: {e}")
async def status(
self, only_check_settings: bool | None = False
) -> FiatStatusResponse:
if only_check_settings:
if self._settings_fields != self._settings_connection_fields():
return FiatStatusResponse("Connection settings have changed.", 0)
return FiatStatusResponse(balance=0)
try:
r = await self.client.get(f"/v2/locations/{self.location_id}", timeout=15)
r.raise_for_status()
_ = r.json()
return FiatStatusResponse(balance=0)
except json.JSONDecodeError:
return FiatStatusResponse("Server error: 'invalid json response'", 0)
except Exception as exc:
logger.warning(exc)
return FiatStatusResponse(f"Unable to connect to {self.endpoint}.", 0)
async def create_invoice(
self,
amount: float,
payment_hash: str,
currency: str,
memo: str | None = None,
extra: dict[str, Any] | None = None,
**kwargs,
) -> FiatInvoiceResponse:
opts = self._parse_create_opts(extra or {})
if not opts:
return FiatInvoiceResponse(ok=False, error_message="Invalid Square options")
amount_cents = int(amount * 100)
co = opts.checkout or SquareCheckoutOptions()
success_url = (
co.success_url
or settings.square_payment_success_url
or "https://lnbits.com"
)
line_item_name = (co.line_item_name or memo or "LNbits Invoice")[:255]
metadata = {
**co.metadata,
"payment_hash": payment_hash,
"alan_action": "invoice",
}
payload = {
"idempotency_key": payment_hash,
"order": {
"location_id": self.location_id,
"metadata": metadata,
"line_items": [
{
"name": line_item_name,
"quantity": "1",
"base_price_money": {
"amount": amount_cents,
"currency": currency.upper(),
},
}
],
},
"checkout_options": {"redirect_url": success_url},
}
if memo:
payload["payment_note"] = memo[:500]
try:
r = await self.client.post(
"/v2/online-checkout/payment-links", json=payload
)
r.raise_for_status()
data = r.json()
payment_link = data.get("payment_link") or {}
order_id = payment_link.get("order_id")
url = payment_link.get("url")
if not order_id or not url:
return FiatInvoiceResponse(
ok=False, error_message="Server error: missing order id or url"
)
return FiatInvoiceResponse(
ok=True,
checking_id=f"order_{order_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_subscription(
self,
subscription_id: str,
quantity: int,
payment_options: FiatSubscriptionPaymentOptions,
**kwargs,
) -> FiatSubscriptionResponse:
return FiatSubscriptionResponse(
ok=False, error_message="Square subscriptions are not supported."
)
async def cancel_subscription(
self,
subscription_id: str,
correlation_id: str,
**kwargs,
) -> FiatSubscriptionResponse:
return FiatSubscriptionResponse(
ok=False, error_message="Square subscriptions are not supported."
)
async def pay_invoice(self, payment_request: str) -> FiatPaymentResponse:
raise NotImplementedError("Square does not support paying invoices directly.")
async def get_invoice_status(self, checking_id: str) -> FiatPaymentStatus:
try:
square_id = self._normalize_square_id(checking_id)
if square_id.startswith("payment_"):
payment_id = square_id.replace("payment_", "", 1)
return await self._get_payment_status(payment_id)
order_id = (
square_id.replace("order_", "", 1)
if square_id.startswith("order_")
else square_id
)
return await self._get_order_status(order_id)
except Exception as exc:
logger.debug(f"Error getting Square invoice status: {exc}")
return FiatPaymentPendingStatus()
async def get_payment_status(self, checking_id: str) -> FiatPaymentStatus:
raise NotImplementedError("Square does not support outgoing payments.")
async def paid_invoices_stream(self) -> AsyncGenerator[str, None]:
logger.warning(
"Square does not support paid invoices stream. Use webhooks instead."
)
mock_queue: asyncio.Queue[str] = asyncio.Queue(0)
while settings.lnbits_running:
value = await mock_queue.get()
yield value
async def _get_order_status(self, order_id: str) -> FiatPaymentStatus:
r = await self.client.get(f"/v2/orders/{order_id}")
r.raise_for_status()
order = r.json().get("order") or {}
tenders = order.get("tenders") or []
payment_id = None
for tender in tenders:
payment_id = tender.get("payment_id")
if payment_id:
break
if payment_id:
return await self._get_payment_status(payment_id)
if (order.get("state") or "").upper() == "CANCELED":
return FiatPaymentFailedStatus()
return FiatPaymentPendingStatus()
async def _get_payment_status(self, payment_id: str) -> FiatPaymentStatus:
r = await self.client.get(f"/v2/payments/{payment_id}")
r.raise_for_status()
return self._status_from_payment(r.json().get("payment") or {})
def _status_from_payment(self, payment: dict[str, Any]) -> FiatPaymentStatus:
status = (payment.get("status") or "").upper()
if status == "COMPLETED":
return FiatPaymentSuccessStatus()
if status in ["CANCELED", "FAILED"]:
return FiatPaymentFailedStatus()
return FiatPaymentPendingStatus()
def _normalize_square_id(self, checking_id: str) -> str:
return (
checking_id.replace("fiat_square_", "", 1)
if checking_id.startswith("fiat_square_")
else checking_id
)
def _parse_create_opts(
self, raw_opts: dict[str, Any]
) -> SquareCreateInvoiceOptions | None:
try:
return SquareCreateInvoiceOptions.parse_obj(raw_opts)
except ValidationError as e:
logger.warning(f"Invalid Square options: {e}")
return None
def _settings_connection_fields(self) -> str:
return "-".join(
[
str(settings.square_api_endpoint),
str(settings.square_access_token),
str(settings.square_location_id),
str(settings.square_api_version),
]
)
+24 -1
View File
@@ -703,6 +703,21 @@ class PayPalFiatProvider(LNbitsSettings):
paypal_limits: FiatProviderLimits = Field(default_factory=FiatProviderLimits)
class SquareFiatProvider(LNbitsSettings):
square_enabled: bool = Field(default=False)
square_api_endpoint: str = Field(default="https://connect.squareup.com")
square_access_token: str | None = Field(default=None)
square_location_id: str | None = Field(default=None)
square_api_version: str = Field(default="2026-01-22")
square_payment_success_url: str = Field(default="https://lnbits.com")
square_payment_webhook_url: str = Field(
default="https://your-lnbits-domain-here.com/api/v1/callback/square"
)
square_webhook_signature_key: str | None = Field(default=None)
square_limits: FiatProviderLimits = Field(default_factory=FiatProviderLimits)
class LightningSettings(LNbitsSettings):
lightning_invoice_expiry: int = Field(default=3600, gt=0)
@@ -740,7 +755,7 @@ class FundingSourcesSettings(
funding_source_max_retries: int = Field(default=4, ge=0)
class FiatProvidersSettings(StripeFiatProvider, PayPalFiatProvider):
class FiatProvidersSettings(StripeFiatProvider, PayPalFiatProvider, SquareFiatProvider):
def is_fiat_provider_enabled(self, provider: str | None) -> bool:
"""
Checks if a specific fiat provider is enabled.
@@ -751,6 +766,8 @@ class FiatProvidersSettings(StripeFiatProvider, PayPalFiatProvider):
return self.stripe_enabled
if provider == "paypal":
return self.paypal_enabled
if provider == "square":
return self.square_enabled
return False
def get_fiat_providers_for_user(self, user_id: str) -> list[str]:
@@ -770,6 +787,12 @@ class FiatProvidersSettings(StripeFiatProvider, PayPalFiatProvider):
):
allowed_providers.append("paypal")
if self.square_enabled and (
not self.square_limits.allowed_users
or user_id in self.square_limits.allowed_users
):
allowed_providers.append("square")
return allowed_providers
def get_fiat_provider_limits(self, provider_name: str) -> FiatProviderLimits | None:
+9
View File
@@ -267,6 +267,13 @@ window.localisation.en = {
webhook_events_list: 'The following events must be supported by the webhook:',
webhook_stripe_description:
'One the stripe side you must configure a webhook with a URL that points to your LNbits server.',
webhook_square_description:
'On the Square side configure a webhook pointing to your LNbits server.',
access_token: 'Access Token',
location_id: 'Location ID',
square_location_id_hint:
'Square location ID to create payment links for. Use the endpoint to select sandbox or production.',
api_version: 'API Version',
payment_proof: 'Payment Proof',
update: 'Update',
update_available: 'Update {version} available!',
@@ -805,6 +812,8 @@ window.localisation.en = {
webhook_id_hint: 'PayPal webhook ID used to verify incoming events.',
webhook_paypal_description:
'On the PayPal side configure a webhook pointing to your LNbits server.',
square_webhook_signature_key_hint:
'Square webhook signature key used to verify incoming events.',
callback_success_url: 'Callback Success URL',
callback_success_url_hint:
'The user will be redirected to this URL after the payment is successful',
@@ -5,6 +5,7 @@ window.app.component('lnbits-admin-fiat-providers', {
return {
formAddStripeUser: '',
formAddPaypalUser: '',
formAddSquareUser: '',
hideInputToggle: true
}
},
@@ -20,6 +21,12 @@ window.app.component('lnbits-admin-fiat-providers', {
this.formData?.paypal_payment_webhook_url ||
this.calculateWebhookUrl('paypal')
)
},
squareWebhookUrl() {
return (
this.formData?.square_payment_webhook_url ||
this.calculateWebhookUrl('square')
)
}
},
watch: {
@@ -58,6 +65,7 @@ window.app.component('lnbits-admin-fiat-providers', {
syncWebhookUrls() {
this.maybeSetWebhookUrl('stripe_payment_webhook_url', 'stripe')
this.maybeSetWebhookUrl('paypal_payment_webhook_url', 'paypal')
this.maybeSetWebhookUrl('square_payment_webhook_url', 'square')
},
maybeSetWebhookUrl(fieldName, provider) {
if (!this.formData) {
@@ -111,6 +119,23 @@ window.app.component('lnbits-admin-fiat-providers', {
this.formData.paypal_limits.allowed_users =
this.formData.paypal_limits.allowed_users.filter(u => u !== user)
},
addSquareAllowedUser() {
const addUser = this.formAddSquareUser || ''
if (
addUser.length &&
!this.formData.square_limits.allowed_users.includes(addUser)
) {
this.formData.square_limits.allowed_users = [
...this.formData.square_limits.allowed_users,
addUser
]
this.formAddSquareUser = ''
}
},
removeSquareAllowedUser(user) {
this.formData.square_limits.allowed_users =
this.formData.square_limits.allowed_users.filter(u => u !== user)
},
checkFiatProvider(providerName) {
LNbits.api
.request('PUT', `/api/v1/fiat/check/${providerName}`)
@@ -587,12 +587,282 @@
<q-item-section> Square </q-item-section>
<q-item-section side>
<div class="row items-center">Disabled</div>
<div class="row items-center">
<q-toggle
size="md"
:label="$t('enabled')"
v-model="formData.square_enabled"
color="green"
unchecked-icon="clear"
/>
</div>
</q-item-section>
</template>
<q-card>
<q-card-section> Coming Soon </q-card-section>
<q-card class="q-pb-xl">
<q-expansion-item :label="$t('api')" default-opened>
<q-card-section class="q-pa-md">
<q-input
filled
type="text"
v-model="formData.square_api_endpoint"
:label="$t('endpoint')"
></q-input>
<q-input
filled
class="q-mt-md"
:type="hideInputToggle ? 'password' : 'text'"
v-model="formData.square_access_token"
:label="$t('access_token')"
></q-input>
<q-input
filled
class="q-mt-md"
type="text"
v-model="formData.square_location_id"
:label="$t('location_id')"
:hint="$t('square_location_id_hint')"
></q-input>
<q-input
filled
class="q-mt-md"
type="text"
v-model="formData.square_api_version"
:label="$t('api_version')"
></q-input>
<q-input
filled
class="q-mt-md"
type="text"
v-model="formData.square_payment_success_url"
:label="$t('callback_success_url')"
:hint="$t('callback_success_url_hint')"
></q-input>
</q-card-section>
<q-card-section class="q-pa-md">
<div class="row">
<div class="col">
<q-btn
outline
color="grey"
class="float-right"
:label="$t('check_connection')"
@click="checkFiatProvider('square')"
></q-btn>
</div>
</div>
</q-card-section>
</q-expansion-item>
<q-expansion-item :label="$t('webhook')" default-opened>
<q-card-section>
<span v-text="$t('webhook_square_description')"></span>
</q-card-section>
<q-card-section>
<div class="row items-center q-gutter-sm q-mt-md">
<div class="col">
<q-input
filled
type="text"
disable
:model-value="squareWebhookUrl"
:label="$t('webhook_url')"
:hint="$t('webhook_url_hint')"
readonly
></q-input>
</div>
<div class="col-auto">
<q-btn
outline
color="grey"
icon="content_copy"
@click="copyWebhookUrl(squareWebhookUrl)"
:aria-label="$t('copy_webhook_url')"
>
<q-tooltip>
<span v-text="$t('copy_webhook_url')"></span>
</q-tooltip>
</q-btn>
</div>
</div>
<q-input
filled
class="q-mt-md"
:type="hideInputToggle ? 'password' : 'text'"
v-model="formData.square_webhook_signature_key"
:label="$t('signing_secret')"
:hint="$t('square_webhook_signature_key_hint')"
></q-input>
</q-card-section>
<q-card-section>
<span v-text="$t('webhook_events_list')"></span>
<ul>
<li><code>payment.updated</code></li>
</ul>
</q-card-section>
</q-expansion-item>
<q-expansion-item :label="$t('service_fee')">
<q-card-section>
<div class="row">
<div class="col-md-4 col-sm-12">
<q-input
filled
class="q-ma-sm"
type="number"
min="0"
v-model="formData.square_limits.service_fee_percent"
@update:model-value="formData.touch = null"
:label="$t('service_fee_label')"
:hint="$t('service_fee_hint')"
></q-input>
</div>
<div class="col-md-4 col-sm-12">
<q-input
filled
class="q-ma-sm"
type="number"
min="0"
v-model="formData.square_limits.service_max_fee_sats"
@update:model-value="formData.touch = null"
:label="$t('service_fee_max')"
:hint="$t('service_fee_max_hint')"
></q-input>
</div>
<div class="col-md-4 col-sm-12">
<q-input
filled
class="q-ma-sm"
type="text"
v-model="formData.square_limits.service_fee_wallet_id"
@update:model-value="formData.touch = null"
:label="$t('fee_wallet_label')"
:hint="$t('fee_wallet_hint')"
></q-input>
</div>
</div>
</q-card-section>
</q-expansion-item>
<q-expansion-item :label="$t('amount_limits')">
<q-card-section>
<div class="row">
<div class="col-md-4 col-sm-12">
<q-input
filled
class="q-ma-sm"
type="number"
min="0"
v-model="formData.square_limits.service_min_amount_sats"
@update:model-value="formData.touch = null"
:label="$t('min_incoming_payment_amount')"
:hint="$t('min_incoming_payment_amount_desc')"
></q-input>
</div>
<div class="col-md-4 col-sm-12">
<q-input
filled
class="q-ma-sm"
type="number"
min="0"
v-model="formData.square_limits.service_max_amount_sats"
@update:model-value="formData.touch = null"
:label="$t('max_incoming_payment_amount')"
:hint="$t('max_incoming_payment_amount_desc')"
></q-input>
</div>
<div class="col-md-4 col-sm-12">
<q-input
filled
class="q-ma-sm"
v-model="formData.square_limits.service_faucet_wallet_id"
@update:model-value="formData.touch = null"
:label="$t('faucest_wallet_id')"
:hint="$t('faucest_wallet_id_hint')"
></q-input>
</div>
</div>
<q-item>
<q-item-section>
<q-item-label v-text="$t('faucest_wallet')"></q-item-label>
<q-item-label caption>
<ul>
<li>
<span
v-text="
$t('faucest_wallet_desc_1', {
provider: 'square'
})
"
></span>
</li>
<li>
<span
v-text="
$t('faucest_wallet_desc_2', {
provider: 'square'
})
"
></span>
</li>
<li>
<span v-text="$t('faucest_wallet_desc_3')"></span>
</li>
<li>
<span
v-text="
$t('faucest_wallet_desc_4', {
provider: 'square'
})
"
></span>
</li>
<li>
<span v-text="$t('faucest_wallet_desc_5')"></span>
</li>
</ul>
<br />
</q-item-label>
</q-item-section>
</q-item>
</q-card-section>
</q-expansion-item>
<q-expansion-item :label="$t('allowed_users')">
<q-card-section>
<q-input
filled
v-model="formAddSquareUser"
@keydown.enter="addSquareAllowedUser"
type="text"
:label="$t('allowed_users_label')"
:hint="
$t('allowed_users_hint_feature', {
feature: 'Square'
})
"
>
<q-btn
@click="addSquareAllowedUser"
dense
flat
icon="add"
></q-btn>
</q-input>
<div>
<q-chip
v-for="user in formData.square_limits.allowed_users"
@update:model-value="formData.touch = null"
:key="user"
removable
@remove="removeSquareAllowedUser(user)"
color="primary"
text-color="white"
:label="user"
class="ellipsis"
>
</q-chip>
</div>
</q-card-section>
</q-expansion-item>
</q-card>
</q-expansion-item>
</q-list>
@@ -640,20 +910,18 @@
>
</div>
<div class="row items-center q-gutter-sm">
<div class="text-bold" style="min-width: 140px">
Square (coming soon)
</div>
<div class="text-bold" style="min-width: 140px">Square</div>
<q-chip dense color="positive" text-color="white" icon="check"
>Checkout</q-chip
>
<q-chip dense color="positive" text-color="white" icon="check"
<q-chip dense color="negative" text-color="white" icon="close"
>Subscriptions</q-chip
>
<q-chip dense color="negative" text-color="white" icon="close"
>Tap-to-pay</q-chip
>
<q-chip dense color="grey-9" text-color="white" icon="public"
>Regions: Global</q-chip
>Regions: Square-supported countries</q-chip
>
</div>
</div>
+20
View File
@@ -394,6 +394,26 @@
<span v-text="$t('pay_with', {provider: 'PayPal'})"></span>
</q-item-section>
</q-item>
<q-separator
v-if="g.user.fiat_providers?.includes('square')"
></q-separator>
<q-item
v-if="g.user.fiat_providers?.includes('square')"
:active="receive.fiatProvider === 'square'"
@click="receive.fiatProvider = 'square'"
active-class="bg-teal-1 text-grey-8 text-weight-bold"
clickable
v-ripple
>
<q-item-section avatar>
<q-avatar>
<q-img src="/static/images/square_logo.png"></q-img>
</q-avatar>
</q-item-section>
<q-item-section>
<span v-text="$t('pay_with', {provider: 'Square'})"></span>
</q-item-section>
</q-item>
</q-list>
</div>
+45
View File
@@ -9,6 +9,7 @@ from lnbits.core.services.payments import create_wallet_invoice
from lnbits.core.services.users import create_user_account
from lnbits.core.views.callback_api import (
handle_paypal_event,
handle_square_event,
handle_stripe_event,
)
@@ -23,7 +24,11 @@ async def test_callback_api_generic_webhook_handler_routes_providers(
paypal_mock = mocker.patch(
"lnbits.core.views.callback_api.handle_paypal_event", mocker.AsyncMock()
)
square_mock = mocker.patch(
"lnbits.core.views.callback_api.handle_square_event", mocker.AsyncMock()
)
mocker.patch("lnbits.core.views.callback_api.check_stripe_signature")
mocker.patch("lnbits.core.views.callback_api.check_square_signature")
mocker.patch(
"lnbits.core.views.callback_api.verify_paypal_webhook", mocker.AsyncMock()
)
@@ -45,6 +50,15 @@ async def test_callback_api_generic_webhook_handler_routes_providers(
assert paypal.json()["success"] is True
paypal_mock.assert_awaited_once()
square = await http_client.post(
"/api/v1/callback/square",
headers={"x-square-hmacsha256-signature": "sig"},
json={"event_id": "evt_3", "type": "payment.updated"},
)
assert square.status_code == 200
assert square.json()["success"] is True
square_mock.assert_awaited_once()
unknown = await http_client.post("/api/v1/callback/unknown", json={"id": "evt_3"})
assert unknown.status_code == 200
assert unknown.json()["success"] is False
@@ -94,6 +108,37 @@ async def test_callback_api_handles_paid_events_with_real_payments(mocker):
assert fiat_status_mock.await_count == 2
@pytest.mark.anyio
async def test_callback_api_handles_square_paid_events(mocker):
payment = mocker.Mock()
get_payment = mocker.patch(
"lnbits.core.views.callback_api.get_standalone_payment",
mocker.AsyncMock(return_value=payment),
)
fiat_status_mock = mocker.patch(
"lnbits.core.views.callback_api.check_fiat_status", mocker.AsyncMock()
)
await handle_square_event(
{
"event_id": "evt_square",
"type": "payment.updated",
"data": {
"object": {
"payment": {
"id": "payment_1",
"order_id": "order_1",
"status": "COMPLETED",
}
}
},
}
)
get_payment.assert_awaited_once_with("fiat_square_order_order_1")
fiat_status_mock.assert_awaited_once_with(payment)
@pytest.mark.anyio
async def test_callback_api_handles_subscription_flows_and_validation(mocker):
user = await create_user_account(
+177
View File
@@ -1,6 +1,7 @@
import hashlib
import hmac
import time
from base64 import b64encode
from unittest.mock import AsyncMock
import pytest
@@ -15,6 +16,7 @@ from lnbits.core.models.wallets import Wallet
from lnbits.core.services import check_payment_status, payments
from lnbits.core.services.fiat_providers import (
check_fiat_status,
check_square_signature,
check_stripe_signature,
handle_fiat_payment_confirmation,
verify_paypal_webhook,
@@ -24,6 +26,7 @@ from lnbits.core.services.fiat_providers import (
)
from lnbits.core.services.users import create_user_account
from lnbits.fiat.base import FiatInvoiceResponse, FiatPaymentStatus, FiatStatusResponse
from lnbits.fiat.square import SquareWallet
from lnbits.settings import Settings
from tests.helpers import get_random_string
@@ -56,16 +59,39 @@ class MockHTTPClient:
self.calls.append((path, kwargs))
return self._responses.pop(0)
async def get(self, path: str, **kwargs):
self.calls.append((path, kwargs))
return self._responses.pop(0)
@pytest.fixture(autouse=True)
def fiat_provider_test_settings(settings: Settings):
original_allowed_currencies = settings.lnbits_allowed_currencies
original_paypal_enabled = settings.paypal_enabled
original_square_enabled = settings.square_enabled
original_square_api_endpoint = settings.square_api_endpoint
original_square_access_token = settings.square_access_token
original_square_location_id = settings.square_location_id
original_square_api_version = settings.square_api_version
original_square_payment_success_url = settings.square_payment_success_url
original_square_payment_webhook_url = settings.square_payment_webhook_url
original_square_webhook_signature_key = settings.square_webhook_signature_key
original_square_limits = settings.square_limits.copy(deep=True)
settings.lnbits_allowed_currencies = []
settings.paypal_enabled = False
settings.square_enabled = False
yield
settings.lnbits_allowed_currencies = original_allowed_currencies
settings.paypal_enabled = original_paypal_enabled
settings.square_enabled = original_square_enabled
settings.square_api_endpoint = original_square_api_endpoint
settings.square_access_token = original_square_access_token
settings.square_location_id = original_square_location_id
settings.square_api_version = original_square_api_version
settings.square_payment_success_url = original_square_payment_success_url
settings.square_payment_webhook_url = original_square_payment_webhook_url
settings.square_webhook_signature_key = original_square_webhook_signature_key
settings.square_limits = original_square_limits
@pytest.mark.anyio
@@ -130,6 +156,22 @@ async def test_create_wallet_fiat_invoice_allowed_users(
assert user
assert user.fiat_providers == []
settings.square_enabled = True
settings.square_limits.allowed_users = []
user = await get_user(to_user.id)
assert user
assert user.fiat_providers == ["square"]
settings.square_limits.allowed_users = ["some_other_user_id"]
user = await get_user(to_user.id)
assert user
assert user.fiat_providers == []
settings.square_limits.allowed_users.append(to_user.id)
user = await get_user(to_user.id)
assert user
assert user.fiat_providers == ["square"]
@pytest.mark.anyio
async def test_create_wallet_fiat_invoice_fiat_limits_fail(
@@ -280,6 +322,116 @@ async def test_create_wallet_fiat_invoice_success(
assert status.success is True
@pytest.mark.anyio
async def test_create_wallet_square_fiat_invoice_success(
to_wallet: Wallet, settings: Settings, mocker: MockerFixture
):
settings.square_enabled = True
settings.square_access_token = "square-token"
settings.square_location_id = "LOC123"
settings.square_limits.service_min_amount_sats = 0
settings.square_limits.service_max_amount_sats = 0
settings.square_limits.service_faucet_wallet_id = None
invoice_data = CreateInvoice(
unit="USD", amount=1.0, memo="Test", fiat_provider="square"
)
fiat_mock_response = FiatInvoiceResponse(
ok=True,
checking_id="order_123",
payment_request="https://square.link/u/session_123",
)
mocker.patch(
"lnbits.fiat.SquareWallet.create_invoice",
AsyncMock(return_value=fiat_mock_response),
)
mocker.patch(
"lnbits.utils.exchange_rates.get_fiat_rate_satoshis",
AsyncMock(return_value=1000),
)
payment = await payments.create_fiat_invoice(to_wallet.id, invoice_data)
assert payment.status == PaymentState.PENDING
assert payment.fiat_provider == "square"
assert payment.extra.get("fiat_checking_id") == fiat_mock_response.checking_id
assert payment.checking_id.startswith("fiat_square_order_123")
@pytest.mark.anyio
async def test_square_wallet_create_invoice(settings: Settings):
settings.square_api_endpoint = "https://connect.squareupsandbox.com"
settings.square_access_token = "square-token"
settings.square_location_id = "LOC123"
settings.square_api_version = "2026-01-22"
settings.square_payment_success_url = "https://lnbits.example/success"
wallet = SquareWallet()
client = MockHTTPClient(
[
MockHTTPResponse(
json_data={
"payment_link": {
"order_id": "ORDER123",
"url": "https://square.link/u/abc123",
}
}
)
]
)
wallet.client = client
response = await wallet.create_invoice(
amount=1.23,
payment_hash="hash123",
currency="USD",
memo="LNbits Square invoice",
extra={"checkout": {"metadata": {"source": "test"}}},
)
assert response.ok is True
assert response.checking_id == "order_ORDER123"
assert response.payment_request == "https://square.link/u/abc123"
assert client.calls[0][0] == "/v2/online-checkout/payment-links"
payload = client.calls[0][1]["json"]
assert payload["idempotency_key"] == "hash123"
assert payload["order"]["location_id"] == "LOC123"
assert payload["order"]["metadata"]["payment_hash"] == "hash123"
assert payload["order"]["metadata"]["alan_action"] == "invoice"
assert payload["order"]["metadata"]["source"] == "test"
assert payload["order"]["line_items"][0]["base_price_money"]["amount"] == 123
@pytest.mark.anyio
async def test_square_wallet_get_invoice_status(settings: Settings):
settings.square_api_endpoint = "https://connect.squareupsandbox.com"
settings.square_access_token = "square-token"
settings.square_location_id = "LOC123"
settings.square_api_version = "2026-01-22"
wallet = SquareWallet()
client = MockHTTPClient(
[
MockHTTPResponse(
json_data={
"order": {
"id": "ORDER123",
"state": "COMPLETED",
"tenders": [{"payment_id": "PAYMENT123"}],
}
}
),
MockHTTPResponse(json_data={"payment": {"status": "COMPLETED"}}),
]
)
wallet.client = client
status = await wallet.get_invoice_status("fiat_square_order_ORDER123")
assert status.success is True
assert client.calls[0][0] == "/v2/orders/ORDER123"
assert client.calls[1][0] == "/v2/payments/PAYMENT123"
@pytest.mark.anyio
async def test_fiat_service_fee(settings: Settings):
# settings.stripe_limits.service_min_amount_sats = 0
@@ -469,6 +621,31 @@ def test_check_stripe_signature_non_utf8_payload():
check_stripe_signature(payload, sig_header, secret)
def test_check_square_signature_success():
payload = b'{"type":"payment.updated"}'
secret = "signature-key"
notification_url = "https://lnbits.example/api/v1/callback/square"
signature = b64encode(
hmac.new(
key=secret.encode(),
msg=notification_url.encode() + payload,
digestmod=hashlib.sha256,
).digest()
).decode()
check_square_signature(payload, signature, secret, notification_url)
def test_check_square_signature_rejects_invalid_signature():
with pytest.raises(ValueError, match="Square signature verification failed."):
check_square_signature(
b'{"type":"payment.updated"}',
"invalid-signature",
"signature-key",
"https://lnbits.example/api/v1/callback/square",
)
# Helper to generate a valid Stripe signature header
def _make_stripe_sig_header(payload, secret, timestamp=None):
if timestamp is None: