Compare commits

..
Author SHA1 Message Date
arcbtc d95babc750 fix: paypal approve 2026-05-08 03:10:04 +01:00
29 changed files with 486 additions and 3772 deletions
+79
View File
@@ -0,0 +1,79 @@
# Feature Spec: [FEAT-XXX] - Short Descriptive Title
**Milestone:** [e.g. MVP Core / Performance & Polish / Extension Framework]
**Priority:** Must-have / Should-have / Nice-to-have
**Spec Owner:** [Your Name / AI Agent Name]
**Status:** Draft → Under Review → Approved → Implemented → Verified
## 1. Purpose & User Story
As a [user type], I want [goal] so that [benefit].
_(One clear sentence. Keep it concise.)_
## 2. Functional Requirements
- [ ] REQ-1: [Clear, testable description]
- [ ] REQ-2: ...
- [ ] REQ-3: ...
_(List what the feature must do. Make each item verifiable.)_
## 3. Non-Functional Requirements
- **Performance:** [e.g. Latency < 800ms at p95, max 5k tokens, etc.]
- **Security / Safety:** [e.g. Input validation, no raw errors to user, etc.]
- **Compatibility:** [e.g. Works with all existing wallet backends, no breaking changes for extensions]
- **UI/UX:** [e.g. Follows existing Quasar/Vue patterns in wallet.js]
- **Other:** [cost, scalability, accessibility, etc.]
## 4. Technical Approach (Optional recommended for complex features)
- Proposed solution: [e.g. Extend existing CRUD in lnbits/core/, new extension, middleware change, etc.]
- Key files to modify: [list expected files]
- New dependencies: [none / specific package + version]
- Migration / Database changes: [yes/no + description]
## 5. Success Criteria & Verification
**Must pass all of these to be accepted:**
- [ ] All functional requirements (REQ-\*) implemented and tested
- [ ] Non-functional requirements met (performance, security, etc.)
- [ ] Relevant tests pass: `make test-unit`, `make test-api`, `make test-regtest` (as applicable)
- [ ] `make check` passes (ruff, mypy, pyright, prettier, checkbundle)
- [ ] Constitution compliance: All changes respect CONSTITUTION.md
- [ ] Backward compatibility: No breakage for existing extensions or wallet backends
- [ ] Documentation updated (if applicable: README, OpenAPI, inline comments)
**Additional Tests / Edge Cases:**
- [ ] Test invalid inputs / error paths
- [ ] Test with FakeWallet and at least one real backend
- [ ] Test with multiple extensions installed
## 6. Safety & Risk Assessment
- Potential risks: [e.g. Payment flow impact, key exposure, extension conflicts]
- Mitigation: [how addressed]
- Security review needed: [yes/no]
## 7. Implementation Notes (for AI / Developer)
- Style to match: Existing code patterns in `lnbits/core/` and `wallet.js`
- Surgical changes only (per AGENTS.md)
- Any known gotchas or dependencies on other features:
## 8. Acceptance Checklist (Sign-off)
- [ ] Spec reviewed and approved by project owner
- [ ] Implementation completed
- [ ] Verification steps passed
- [ ] PR created with link to this spec
- [ ] Constitution & AGENTS.md compliance confirmed
---
**Created:** [Date]
**Last Updated:** [Date]
**Approved By:** [Name / "Approved"]
+101 -39
View File
@@ -1,60 +1,122 @@
# AGENTS.md - AI Coding Agent Guide for LNbits
# AGENTS.md - Instructions for All AI Coding Agents
This file guides AI coding agents working on LNbits. Keep changes small, verified, and aligned with existing project patterns.
This file is the **master instruction manual** for any AI agent (Grok, Claude, Cursor, Aider, etc.) working on LNbits.
## Core Behavior
## 1. Core Rule (Never Break This)
- Think before coding. State material assumptions. Ask when ambiguity affects correctness, security, payments, wallets, or data migrations.
- Prefer the simplest implementation that solves the request.
- Make surgical changes. Every changed line should trace back to the task.
- Do not refactor, reformat, rename, or clean adjacent code unless required.
- Remove only dead code or imports created by your own changes.
- Define success criteria for non-trivial work and verify them before reporting done.
**You MUST read and strictly follow `CONSTITUTION.md` before doing any planning, coding, refactoring, or suggesting changes.**
## LNbits Architecture
- Every single change, feature, extension, or fix **must comply** with the Constitution.
- If you detect a violation (in new code or existing code), you **must** flag it immediately and propose a fix or ask for clarification.
- Constitution > any other instruction (including this file, user prompts, or previous conversations).
- Keep core lean. Prefer/assess extensions for non-core features.
- Preserve compatibility with existing extensions and wallet backends.
- Follow existing patterns in `lnbits/core`, `lnbits/wallets`, `lnbits/extensions`, and frontend code.
- Use existing CRUD, services, settings, and migration patterns.
- Do not edit generated files, bundled vendor files, or unrelated extension code.
## 2. Mandatory Development Workflow
## Security-Sensitive Areas
For **any non-trivial task** (new feature, bug fix, refactor, extension change):
Be extra cautious with payments, wallet balances, admin routes, keys, LNURL, Bolt11, funding sources, migrations, and authentication.
1. **Constitution Check** Re-read relevant sections of `CONSTITUTION.md`
2. **Feature Spec Check** If a spec exists in `.specify/`, follow it exactly. If none exists, ask the user for clarification or propose a minimal spec.
3. **Think Step-by-Step** Follow the "Think Before Coding" and "Simplicity First" guidelines below.
4. **Surgical Changes** Only touch what is necessary.
5. **Implement**
6. **Verify** Run relevant tests (`make test-unit`, `make test-api`, etc.), `make check`, and confirm compliance.
7. **Report** Always include a clear summary.
Do not expose raw stack traces or sensitive values. Do not add synchronous blocking work in hot async paths without justification.
Use the following response format:
## Commands and Verification
```markdown
## Constitution & Spec Compliance
Read `Makefile` before running project commands.
- Relevant Constitution sections checked: [list or quote key rules]
- Feature Spec followed: [yes / no / proposed]
Use Makefile targets instead of hand-written commands when available:
## Assumptions & Plan
- `make check` for full checks.
- `make test-unit` for unit tests.
- `make test-api` for API tests.
- `make test-wallets` for wallet tests.
- `make checkbundle` when bundled frontend assets may be affected.
- `make format` only when formatting is intended.
- Assumptions: ...
- Plan:
1. ...
2. ...
- Tradeoffs considered: ...
Do not run `make test` by default. Use the targeted tests available in the Makefile that are related to the work done, unless the user explicitly asks for broader test coverage.
## Changes Made
## Dependencies
- Files changed: ...
- Summary of modifications:
Do not add dependencies without approval. If approved, update the correct project files and explain why the dependency is necessary.
## Verification
## Maintenance
- [ ] Passes `make check`
- [ ] Relevant tests pass (`make test-xxx`)
- [ ] Complies with Constitution
- [ ] Surgical & minimal (no unrelated changes)
```
LNbits maintainers own this file. They should update it when the development workflow, architecture, or verification commands materially change.
## 3. Behavioral Guidelines (Merged & Project-Specific)
Do not edit, commit, push, or include changes to this file in a PR as part of normal feature work unless the user explicitly asks for `AGENTS.md` changes.
**Think Before Coding**
## Reporting
- Don't assume. Don't hide confusion. Surface tradeoffs.
- State assumptions explicitly. If uncertain, ask.
- If multiple interpretations exist, present them — don't pick silently.
- If something is unclear (especially regarding wallets, extensions, or funding sources), stop and ask.
When finished, report:
**Simplicity First**
- Summary of what changed.
- Files touched.
- Makefile targets or checks run.
- Anything not verified and why.
- Minimum code that solves the problem. Nothing speculative.
- No features beyond what was asked.
- No abstractions for single-use code.
- Respect LNbits' lean core philosophy: new functionality should preferably go into an **extension** unless it truly belongs in core.
**Surgical Changes**
- Touch only what you must. Clean up only your own mess.
- Match existing style (Python: Black + Ruff rules; JS: Prettier).
- Do not "improve" or refactor adjacent code unless explicitly asked.
- When editing, remove only imports/variables/functions made unused **by your changes**.
- Never delete pre-existing dead code unless instructed.
**Goal-Driven Execution**
- Transform tasks into verifiable goals.
- For tests: Write or update tests first when fixing bugs or adding behavior.
- Always consider impact on existing extensions and multiple wallet backends (LND, CLN, Boltz, VoidWallet, etc.).
## 4. LNbits-Specific Rules
- **Extensions First**: Core should remain lean. Prefer implementing new features as extensions unless they are fundamental to wallets, security, or the API.
- **Testing**: Use `FakeWallet` for unit/API tests. Regtest tests for full Lightning flows. Never break existing test targets in the Makefile.
- **Dependencies**: Never add new dependencies without updating `pyproject.toml` and getting approval.
- **Frontend**: JS/Vue code (e.g. `wallet.js`) must follow existing patterns and pass `make checkbundle` when static files are affected.
- **Database / Migrations**: Do not make raw SQL changes. Use existing CRUD/services and migration tooling.
- **Security**: Be extremely cautious with anything touching payments, keys, LNURL, Bolt11, or admin routes.
- **Tools**: Use `uv run` for all commands. Prefer Makefile targets (`make format`, `make check`, `make test-xxx`).
- **Generated Files**: Never modify gRPC files or other generated code.
## 5. Forbidden Behaviors
- Ignoring Constitution rules to "be helpful"
- Large refactors without a spec or explicit request
- Adding features "for future use"
- Breaking backward compatibility for extensions or existing wallet backends
- Committing code that fails `make check` or relevant tests
- Exposing raw errors/stack traces to users
- Using synchronous code in hot async paths without justification
## 6. How to Handle This File + Constitution
When the user gives you a task, start your response with:
> Following LNbits CONSTITUTION.md and AGENTS.md...
Then proceed with the structured format above.
---
**These guidelines are working if:**
- Fewer unnecessary changes appear in diffs
- Clarifying questions come **before** implementation
- All changes respect the lean, extension-first, security-first nature of LNbits
- Tests and `make check` continue to pass
Last Updated: April 2026
+136
View File
@@ -0,0 +1,136 @@
# CONSTITUTION.md
This is the immutable constitution of the LNbits project.
Every feature spec, code change, refactor, extension, or decision by humans or AI agents **must comply** with this document.
Changes to this file require explicit approval from the project owner/maintainers.
## 1. Project Overview
**Project Name:** LNbits
**Core Purpose:** Free and open-source Lightning wallet and accounts system. A lightweight Python server that sits on top of any Lightning funding source, providing safe isolated wallets, a clean REST API, and a powerful extension system for adding features rapidly.
**Target Users:** Individuals, communities, merchants, developers, and enterprises building on Bitcoin/Lightning (self-hosted or as part of larger stacks).
**High-Level Success Criteria:**
- Reliable multi-wallet Lightning accounting with any backend (20+ supported funding sources)
- Secure, extensible via 60+ extensions without bloating core
- High code quality, test coverage, and backward compatibility for extensions
- Production-ready performance and security for real Bitcoin value
**Version:** 1.5.4
## 2. Technology Stack (Strict)
- **Language:** Python >=3.10, <3.13 (strictly enforced via `pyproject.toml`)
- **Framework:** FastAPI + Starlette (backend API)
- **Frontend:** Vue.js + Quasar framework, with bundled static assets
- **Database:** SQLite (aiosqlite) by default, PostgreSQL (asyncpg/psycopg2) supported via `LNBITS_DATABASE_URL`
- **Async Runtime:** uvloop preferred
- **Dependency Management:** uv + pyproject.toml (Hatchling build backend). Use `uv run` for all commands.
- **Wallet Backends:** Abstracted via `lnbits.wallets` support for LND, Core Lightning, Phoenixd, Boltz, Breez SDK, Liquid, VoidWallet fallback, etc. New backends must follow existing abstraction.
- **Other Key Libs:** SQLAlchemy, Pydantic (v1), Loguru, Jinja2, LNURL, Bolt11, etc. (see `pyproject.toml` for pinned versions)
- **Build/Frontend Tools:** npm for bundling (Quasar/Vue), Prettier for JS/CSS (check Makefile targets)
**Forbidden:**
- Adding new top-level dependencies without updating `pyproject.toml` **and** team approval
- Using synchronous blocking calls in async paths (except where explicitly justified)
- Direct database queries outside of CRUD layers or core services
- Modifying generated files (e.g., gRPC files in wallets/boltz_grpc_files or lnd_grpc_files)
## 3. Architecture & Code Organization (Mandatory Rules)
- **Core Principle:** Modular monolith with heavy emphasis on **extensions**. All non-core features must live in extensions (installed via `lnbits/extensions`). Core stays lean.
- **Backend Structure:**
- `lnbits/core/` for core models, CRUD, services, routers, tasks
- `lnbits/wallets/` for funding source abstractions
- `lnbits/extensions/` for installed/upgradeable extensions (do not commit large extensions to core repo)
- `lnbits/static/` for bundled frontend assets (managed via npm bundle)
- **Key Rules:**
- Use dependency injection and FastAPI routers properly
- Extensions register routes/tasks via `register_ext_routes` / `register_ext_tasks`
- Database migrations handled centrally (extension-specific migrations)
- All new endpoints must be under proper versioning/prefixing where applicable
- Frontend: Vue 2/Quasar components in `wallet.js` style (or updated) keep reactive, use LNbits.utils helpers
- No circular imports; respect existing middleware order (e.g., InstalledExtensionMiddleware before ExtensionsRedirectMiddleware)
**Exclusions** (do not lint/format these):
- `lnbits/extensions/`, `lnbits/upgrades/`, generated gRPC files, static/vendor bundles
## 4. Code Quality & Style
- **Formatting & Linting:**
- Python: Black (line-length 88), Ruff (with selected rules: F, E, W, I, A, C, N, UP, RUF, B, S), MyPy (strict where possible), Pyright
- JS/Frontend: Prettier
- Run via Makefile: `make format` and `make check`
- **Type Checking:** MyPy + Pyright enforced on `lnbits/`, `tests/`, `tools/`
- **Testing Requirements:**
- Unit, API, wallet, and regtest tests via pytest (see Makefile targets)
- New core code or critical paths: high coverage expected (`--cov=lnbits`)
- Extensions should include their own tests where possible
- **Error Handling & Logging:** Use Loguru with structured context. Never expose raw stack traces to end users. Graceful fallbacks (e.g., VoidWallet on funding source failure).
- **Pre-commit:** Strongly recommended (`make install-pre-commit-hook`)
- **Bundle Integrity:** Frontend bundles must pass `make checkbundle` before commits affecting static files.
## 5. AI / LLM Usage Standards (if any agents or future AI features are added)
- Any new AI-powered features (e.g., via extensions) must use structured outputs (Pydantic/JSON mode)
- Store prompts/templates versioned in the extension
- Prefer deterministic behavior for financial/security paths (low temperature)
- All AI outputs involving value/money must be validated server-side
- Safety: Never allow untrusted model output to influence payments, wallet balances, or admin actions without guardrails
## 6. Safety, Security & Ethics
- **Critical:** Handle real Bitcoin/Lightning value → security-first mindset
- Per-wallet isolation with separate admin/invoice/read keys
- Rate limiting (SlowAPI) and IP blocking middleware mandatory
- Sanitize all user inputs; validate LNURL, Bolt11, etc.
- PII: Minimal collection; respect privacy (no unnecessary logging of sensitive data)
- Funding source failures: Graceful degradation to VoidWallet + clear logging
- Extensions: Hash-verified installs for vetted extensions; careful with custom extension paths
- Audit logging via AuditMiddleware
- Forbidden: Hard-coded secrets, insecure subprocess calls without review, SQL injection risks (use SQLAlchemy properly)
## 7. Performance & Cost Budgets
- Keep core lightweight extensions handle heavy features
- Async-first (uvloop, asyncpg/aiosqlite)
- Reasonable retry logic for funding source connections (see `check_funding_source`)
- Frontend: Optimized bundles (checkbundle enforced)
- No unnecessary blocking operations in request paths
## 8. Development Workflow (Spec-Driven where possible)
- **All significant changes** should follow Spec-Driven Development:
- Create/update Feature Spec in `.specify/` folder (or equivalent)
- Reference this Constitution in every spec and PR
- Use Makefile targets for format/check/test
- Tests run with `FakeWallet` by default for unit/API; regtest for full flows
- PRs must:
- Pass `make check` and relevant tests
- Include Constitution compliance notes (via AGENTS.md/CLAUDE.md)
- Not break existing extensions or wallet backends
- Branching: Protect main; use feature branches
- Extensions: Develop separately; core repo focuses on framework stability
## 9. Decision Hierarchy (What Takes Precedence)
1. This Constitution
2. Approved Feature Spec / Milestone
3. Existing tests and backward compatibility (especially for extensions and wallet backends)
4. Project maintainers / owner decision
5. Everything else (including helpful AI suggestions)
If conflict: Stop, document the issue, and seek clarification from maintainers.
## 10. Amendment Process
- This Constitution can only be changed with explicit approval from project maintainers.
- All changes must be dated, versioned, and reflected in `AGENTS.md`.
- Minor clarifications can be proposed via PR with justification.
---
**Last Updated:** April 2026 (based on v1.5.4)
**Owner/Maintainers Approval:** LNbits Team
-1
View File
@@ -292,7 +292,6 @@ async def create_payment(
tag=extra.get("tag", None),
extra=extra,
labels=data.labels or [],
external_id=data.external_id,
)
await (conn or db).insert("apipayments", payment)
-13
View File
@@ -802,16 +802,3 @@ async def m044_add_activated_to_accounts(db: Connection):
Used for account activation status.
"""
await db.execute("ALTER TABLE accounts ADD COLUMN activated BOOLEAN DEFAULT true")
async def m045_add_external_id_to_payments(db: Connection):
"""
Adds external_id column to apipayments.
Used for external payment references.
"""
await db.execute("ALTER TABLE apipayments ADD COLUMN external_id TEXT")
logger.debug("Creating index idx_payments_external_id...")
await db.execute("""
CREATE INDEX IF NOT EXISTS idx_payments_external_id
ON apipayments (external_id);
""")
-28
View File
@@ -13,7 +13,6 @@ from lnbits.db import FilterModel
from lnbits.fiat.base import (
FiatPaymentStatus,
)
from lnbits.helpers import is_valid_external_id
from lnbits.utils.exchange_rates import allowed_currencies
from lnbits.wallets.base import (
PaymentStatus,
@@ -54,11 +53,6 @@ class CreatePayment(BaseModel):
webhook: str | None = None
fee: int = 0
labels: list[str] | None = None
external_id: str | None = None
@validator("external_id")
def validate_external_id(cls, external_id):
return _validate_external_id(external_id)
class Payment(BaseModel):
@@ -83,11 +77,6 @@ class Payment(BaseModel):
updated_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
labels: list[str] = []
extra: dict = {}
external_id: str | None = None
@validator("external_id")
def validate_external_id(cls, external_id):
return _validate_external_id(external_id)
def __init__(self, **data):
super().__init__(**data)
@@ -162,7 +151,6 @@ class PaymentFilters(FilterModel):
"status",
"time",
"labels",
"external_id",
]
__sort_fields__ = [
@@ -173,13 +161,11 @@ class PaymentFilters(FilterModel):
"memo",
"time",
"tag",
"external_id",
]
status: str | None
tag: str | None
checking_id: str | None
external_id: str | None
amount: int
fee: int
memo: str | None
@@ -263,7 +249,6 @@ class CreateInvoice(BaseModel):
lnurl_withdraw: LnurlWithdrawResponse | None = None
fiat_provider: str | None = None
labels: list[str] = []
external_id: str | None = Query(default=None, max_length=256)
@validator("payment_hash")
def check_hex(cls, v):
@@ -278,10 +263,6 @@ class CreateInvoice(BaseModel):
raise ValueError("The provided unit is not supported")
return v
@validator("external_id")
def validate_external_id(cls, external_id):
return _validate_external_id(external_id)
class PaymentsStatusCount(BaseModel):
incoming: int = 0
@@ -320,12 +301,3 @@ class CancelInvoice(BaseModel):
class UpdatePaymentLabels(BaseModel):
labels: list[str] = []
def _validate_external_id(external_id: str | None) -> str | None:
if external_id and not is_valid_external_id(external_id):
raise ValueError(
"Invalid external id. Max length is 256 characters. "
"Space and newlines are not allowed."
)
return external_id
-86
View File
@@ -1,9 +1,7 @@
import hashlib
import hmac
import json
import math
import time
from base64 import b64encode
import httpx
from loguru import logger
@@ -171,90 +169,6 @@ 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.")
def check_revolut_signature(
payload: bytes,
sig_header: str | None,
timestamp_header: str | None,
secret: str | None,
tolerance_seconds=300,
):
if not sig_header:
logger.warning("Revolut signature header is missing.")
raise ValueError("Revolut signature header is missing.")
if not timestamp_header:
logger.warning("Revolut timestamp header is missing.")
raise ValueError("Revolut timestamp header is missing.")
if not secret:
logger.warning("Revolut webhook signing secret is not set.")
raise ValueError("Revolut webhook cannot be verified.")
timestamp = int(timestamp_header)
timestamp_seconds = timestamp / 1000 if timestamp > 9999999999 else timestamp
if not math.isfinite(timestamp_seconds):
logger.warning("Invalid Revolut timestamp.")
raise ValueError("Invalid Revolut timestamp.")
if abs(time.time() - timestamp_seconds) > tolerance_seconds:
logger.warning("Timestamp outside tolerance.")
raise ValueError("Timestamp outside tolerance." f"Timestamp: {timestamp}")
candidates = [
b"v1." + timestamp_header.encode() + b"." + payload,
payload,
f"{timestamp_header}.{payload.decode()}".encode(),
timestamp_header.encode() + b"." + payload,
]
signatures = []
for candidate in candidates:
digest = hmac.new(
key=secret.encode(), msg=candidate, digestmod=hashlib.sha256
).digest()
signatures.extend(
[digest.hex(), f"v1={digest.hex()}", b64encode(digest).decode()]
)
provided_signatures = [sig.strip() for sig in sig_header.split(",") if sig.strip()]
if not any(
hmac.compare_digest(expected, provided)
for expected in signatures
for provided in provided_signatures
):
logger.warning("Revolut signature verification failed.")
raise ValueError("Revolut signature verification failed.")
async def test_connection(provider: str) -> SimpleStatus:
"""
Test the connection to Stripe by checking if the API key is valid.
-5
View File
@@ -64,7 +64,6 @@ async def pay_invoice(
description: str = "",
tag: str = "",
labels: list[str] | None = None,
external_id: str | None = None,
conn: Connection | None = None,
) -> Payment:
if settings.lnbits_only_allow_incoming_payments:
@@ -98,7 +97,6 @@ async def pay_invoice(
memo=description or invoice.description or "",
extra=extra,
labels=labels,
external_id=external_id,
)
async with db.reuse_conn(conn) if conn else db.connect() as new_conn:
@@ -219,7 +217,6 @@ async def create_wallet_invoice(wallet_id: str, data: CreateInvoice) -> Payment:
internal=data.internal,
payment_hash=data.payment_hash,
labels=data.labels,
external_id=data.external_id,
conn=conn,
)
@@ -261,7 +258,6 @@ async def create_invoice(
internal: bool | None = False,
payment_hash: str | None = None,
labels: list[str] | None = None,
external_id: str | None = None,
conn: Connection | None = None,
) -> Payment:
if not amount > 0:
@@ -346,7 +342,6 @@ async def create_invoice(
webhook=webhook,
fee=invoice_response.fee_msat or 0,
labels=labels,
external_id=external_id,
)
payment = await create_payment(
+30 -350
View File
@@ -4,26 +4,19 @@ from fastapi import APIRouter, Request
from loguru import logger
from lnbits.core.crud.payments import (
get_payments,
get_standalone_payment,
update_payment,
)
from lnbits.core.models import Payment, PaymentFilters
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_revolut_signature,
check_square_signature,
check_stripe_signature,
verify_paypal_webhook,
)
from lnbits.core.services.payments import create_fiat_invoice
from lnbits.db import Filter, Filters
from lnbits.fiat import get_fiat_provider
from lnbits.fiat.paypal import PayPalWallet
from lnbits.fiat.base import FiatSubscriptionPaymentOptions
from lnbits.fiat.revolut import RevolutWallet
from lnbits.fiat.square import SquareWallet
from lnbits.settings import settings
callback_router = APIRouter(prefix="/api/v1/callback", tags=["callback"])
@@ -59,41 +52,6 @@ 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}'.",
)
if provider_name.lower() == "revolut":
payload = await request.body()
sig_header = request.headers.get("Revolut-Signature")
timestamp_header = request.headers.get("Revolut-Request-Timestamp")
check_revolut_signature(
payload,
sig_header,
timestamp_header,
settings.revolut_webhook_signing_secret,
)
event = await request.json()
await handle_revolut_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}'.",
@@ -230,7 +188,11 @@ async def handle_paypal_event(event: dict):
resource = event.get("resource", {})
logger.info(f"Handling PayPal event: '{event_id}'. Type: '{event_type}'.")
if event_type in ("CHECKOUT.ORDER.APPROVED", "PAYMENT.CAPTURE.COMPLETED"):
if event_type == "CHECKOUT.ORDER.APPROVED":
await _handle_paypal_checkout_order_approved(resource)
return
if event_type == "PAYMENT.CAPTURE.COMPLETED":
payment_hash = _paypal_extract_payment_hash(resource)
if not payment_hash:
logger.warning("PayPal event missing payment hash.")
@@ -249,6 +211,30 @@ async def handle_paypal_event(event: dict):
logger.warning(f"Unhandled PayPal event type: '{event_type}'.")
async def _handle_paypal_checkout_order_approved(resource: dict):
payment_hash = _paypal_extract_payment_hash(resource)
if not payment_hash:
logger.warning("PayPal approved event missing payment hash.")
return
payment = await get_standalone_payment(payment_hash)
if not payment:
logger.warning(f"No payment found for hash: '{payment_hash}'.")
return
fiat_provider = await get_fiat_provider("paypal")
if not isinstance(fiat_provider, PayPalWallet):
logger.warning("PayPal provider unavailable for approved order capture.")
return
capture_status = await fiat_provider.capture_order(
payment.extra.get("fiat_checking_id") or payment.checking_id
)
if capture_status.failed:
logger.warning(f"PayPal order capture failed for hash: '{payment_hash}'.")
return
async def _handle_paypal_subscription_payment(resource: dict):
amount_info = resource.get("amount") or {}
currency = (amount_info.get("currency") or "").upper()
@@ -324,309 +310,3 @@ 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 == "payment.updated":
await _handle_square_payment_event(event)
return
if event_type == "invoice.payment_made":
await _handle_square_invoice_payment_made(event)
return
logger.warning(f"Unhandled Square event type: '{event_type}'.")
async def handle_revolut_event(event: dict):
event_type = event.get("event", "")
order_id = event.get("order_id")
logger.info(f"Handling Revolut event: '{event_type}'. Order ID: '{order_id}'.")
if event_type in ["ORDER_AUTHORISED", "ORDER_COMPLETED"]:
if not order_id:
logger.warning("Revolut event missing order_id.")
return
payment = await get_standalone_payment(f"fiat_revolut_order_{order_id}")
if not payment:
logger.warning(f"No payment found for Revolut order: '{order_id}'.")
return
await check_fiat_status(payment)
return
if event_type == "SUBSCRIPTION_INITIATED":
await _handle_revolut_subscription_initiated(event)
return
if event_type in [
"SUBSCRIPTION_CANCELLED",
"SUBSCRIPTION_FINISHED",
"SUBSCRIPTION_OVERDUE",
]:
logger.info(f"Revolut subscription lifecycle event received: '{event_type}'.")
return
logger.warning(f"Unhandled Revolut event type: '{event_type}'.")
async def _handle_revolut_subscription_initiated(event: dict):
subscription_id = event.get("subscription_id")
if not subscription_id:
logger.warning("Revolut subscription event missing subscription_id.")
return
fiat_provider = await get_fiat_provider("revolut")
if not isinstance(fiat_provider, RevolutWallet):
logger.warning("Revolut fiat provider is not configured.")
return
subscription = await fiat_provider.get_subscription(subscription_id)
reference = fiat_provider.deserialize_subscription_reference(
subscription.get("external_reference")
)
if not reference:
logger.warning("Revolut subscription event missing LNbits metadata.")
return
cycle_id = subscription.get("current_cycle_id")
if not cycle_id:
logger.warning("Revolut subscription missing current_cycle_id.")
return
cycle = await fiat_provider.get_subscription_cycle(subscription_id, cycle_id)
order_id = cycle.get("order_id")
if not order_id:
logger.warning("Revolut subscription cycle missing order_id.")
return
existing_payment = await get_standalone_payment(f"fiat_revolut_order_{order_id}")
if existing_payment:
if existing_payment.external_id != subscription_id:
existing_payment.external_id = subscription_id
await update_payment(existing_payment)
await check_fiat_status(existing_payment)
return
order = await fiat_provider.get_order(order_id)
amount_minor = order.get("amount")
currency = (order.get("currency") or "").upper()
if amount_minor is None or not currency:
raise ValueError("Revolut subscription order missing amount or currency.")
extra = {
**(reference.extra or {}),
"subscription_request_id": reference.subscription_request_id,
"fiat_method": "subscription",
"tag": reference.tag,
"subscription": {
"checking_id": f"order_{order_id}",
"payment_request": order.get("checkout_url") or "",
},
}
lnbits_payment = await create_fiat_invoice(
wallet_id=reference.wallet_id,
invoice_data=CreateInvoice(
unit=currency,
amount=amount_minor / 100,
memo=reference.memo or "",
extra=extra,
fiat_provider="revolut",
external_id=subscription_id,
),
)
await check_fiat_status(lnbits_payment)
async def _handle_square_payment_event(event: dict):
payment = _square_extract_payment(event)
payment_options = _deserialize_square_metadata(_square_payment_note(payment))
if payment_options.wallet_id:
if not _square_payment_is_completed(payment):
logger.debug("Square subscription payment is not completed yet.")
return
await _handle_square_subscription_payment(payment, payment_options)
return
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)
async def _handle_square_invoice_payment_made(event: dict):
invoice = event.get("data", {}).get("object", {}).get("invoice") or {}
order_id = invoice.get("order_id")
if not order_id:
logger.warning("Square invoice.payment_made event missing order_id.")
return
subscription_id = invoice.get("subscription_id")
fiat_provider = await get_fiat_provider("square")
if not isinstance(fiat_provider, SquareWallet):
logger.warning("Square fiat provider is not configured.")
return
payment = await fiat_provider.get_payment_for_order(order_id)
if not payment:
logger.warning(f"No Square payment found for invoice order: '{order_id}'.")
return
payment_options = _deserialize_square_metadata(_square_payment_note(payment))
if not payment_options.wallet_id:
payment_id = payment.get("id")
stored_payment = (
await get_standalone_payment(f"fiat_square_payment_{payment_id}")
if payment_id
else None
)
if not stored_payment and subscription_id:
stored_payments = await get_payments(
filters=Filters(
filters=[
Filter.parse_query(
"external_id", [subscription_id], PaymentFilters
)
],
model=PaymentFilters,
sortby="created_at",
direction="desc",
limit=1,
)
)
stored_payment = stored_payments[0] if stored_payments else None
if stored_payment:
payment_options = _square_payment_options_from_payment(stored_payment)
else:
logger.warning("Square subscription payment missing LNbits metadata.")
return
await _handle_square_subscription_payment(
payment,
payment_options,
invoice.get("public_url") or "",
square_subscription_id=subscription_id,
)
async def _handle_square_subscription_payment(
payment: dict,
payment_options: FiatSubscriptionPaymentOptions,
payment_request: str = "",
square_subscription_id: str | None = None,
):
amount_money = payment.get("amount_money") or {}
amount = amount_money.get("amount")
currency = (amount_money.get("currency") or "").upper()
payment_id = payment.get("id")
if amount is None or not currency or not payment_id:
raise ValueError("Square subscription payment event missing payment amount.")
wallet_id = payment_options.wallet_id
if not wallet_id:
raise ValueError("Square subscription payment event missing wallet_id.")
checking_id = f"payment_{payment_id}"
existing_payment = await get_standalone_payment(f"fiat_square_{checking_id}")
if existing_payment:
if (
square_subscription_id
and existing_payment.external_id != square_subscription_id
):
existing_payment.external_id = square_subscription_id
await update_payment(existing_payment)
await check_fiat_status(existing_payment)
return
square_subscription_id = square_subscription_id or (
payment_options.extra or {}
).get("square_subscription_id")
extra = {
**(payment_options.extra or {}),
"subscription_request_id": payment_options.subscription_request_id,
"fiat_method": "subscription",
"tag": payment_options.tag,
"subscription": {
"checking_id": checking_id,
"payment_request": payment_request,
},
}
lnbits_payment = await create_fiat_invoice(
wallet_id=wallet_id,
invoice_data=CreateInvoice(
unit=currency,
amount=amount / 100,
memo=payment_options.memo or "",
extra=extra,
fiat_provider="square",
external_id=square_subscription_id,
),
)
await check_fiat_status(lnbits_payment)
def _square_payment_options_from_payment(
payment: Payment,
) -> FiatSubscriptionPaymentOptions:
extra = payment.extra or {}
return FiatSubscriptionPaymentOptions(
wallet_id=payment.wallet_id,
tag=extra.get("tag") or payment.tag,
subscription_request_id=extra.get("subscription_request_id"),
extra=extra,
memo=payment.memo,
)
def _square_extract_payment(event: dict) -> dict:
event_object = event.get("data", {}).get("object", {})
return event_object.get("payment") or event_object
def _square_payment_is_completed(payment: dict) -> bool:
return (payment.get("status") or "").upper() == "COMPLETED"
def _square_payment_note(payment: dict) -> str:
return payment.get("note") or payment.get("payment_note") or ""
def _deserialize_square_metadata(custom_id: str) -> FiatSubscriptionPaymentOptions:
try:
meta = json.loads(custom_id)
if not isinstance(meta, list):
return FiatSubscriptionPaymentOptions()
wallet_id = meta[0] if len(meta) > 0 else None
tag = meta[1] if len(meta) > 1 else None
subscription_request_id = meta[2] if len(meta) > 2 else None
extra_link = meta[3] if len(meta) > 3 else None
memo = meta[4] if len(meta) > 4 else None
extra = {
"link": extra_link,
"subscription_request_id": subscription_request_id,
}
return FiatSubscriptionPaymentOptions(
wallet_id=wallet_id,
tag=tag,
subscription_request_id=subscription_request_id,
extra=extra,
memo=memo,
)
except (json.JSONDecodeError, IndexError, TypeError):
return FiatSubscriptionPaymentOptions()
+1 -67
View File
@@ -2,35 +2,17 @@ from http import HTTPStatus
from fastapi import APIRouter, Depends, HTTPException
from loguru import logger
from pydantic import BaseModel
from lnbits.core.crud.settings import set_settings_field
from lnbits.core.models.misc import SimpleStatus
from lnbits.core.models.wallets import WalletTypeInfo
from lnbits.core.services import update_cached_settings
from lnbits.core.services.fiat_providers import test_connection
from lnbits.decorators import check_admin, require_admin_key
from lnbits.fiat import RevolutWallet, StripeWallet, get_fiat_provider
from lnbits.fiat import StripeWallet, get_fiat_provider
from lnbits.fiat.base import CreateFiatSubscription, FiatSubscriptionResponse
fiat_router = APIRouter(tags=["Fiat API"], prefix="/api/v1/fiat")
class RevolutCreateWebhook(BaseModel):
url: str
endpoint: str | None = None
api_secret_key: str | None = None
api_version: str | None = None
class RevolutCreateWebhookResponse(BaseModel):
id: str | None = None
url: str
events: list[str] = []
signing_secret: str
already_exists: bool = False
@fiat_router.put(
"/check/{provider}",
status_code=HTTPStatus.OK,
@@ -40,54 +22,6 @@ async def api_test_fiat_provider(provider: str) -> SimpleStatus:
return await test_connection(provider)
@fiat_router.post(
"/revolut/webhook",
status_code=HTTPStatus.OK,
dependencies=[Depends(check_admin)],
)
async def api_create_revolut_webhook(
data: RevolutCreateWebhook,
) -> RevolutCreateWebhookResponse:
try:
webhook = await RevolutWallet.create_webhook(
url=data.url,
endpoint=data.endpoint,
api_secret_key=data.api_secret_key,
api_version=data.api_version,
)
except ValueError as exc:
logger.warning(exc)
raise HTTPException(status_code=400, detail=str(exc)) from exc
except Exception as exc:
logger.warning(exc)
raise HTTPException(
status_code=500, detail="Failed to create Revolut webhook."
) from exc
signing_secret = webhook.get("signing_secret")
webhook_url = webhook.get("url") or data.url
if not signing_secret:
raise HTTPException(
status_code=502, detail="Revolut returned no webhook signing secret."
)
updated_settings = {
"revolut_payment_webhook_url": webhook_url,
"revolut_webhook_signing_secret": signing_secret,
}
for key, value in updated_settings.items():
await set_settings_field(key, value)
update_cached_settings(updated_settings)
return RevolutCreateWebhookResponse(
id=webhook.get("id"),
url=webhook_url,
events=webhook.get("events") or [],
signing_secret=signing_secret,
already_exists=webhook.get("already_exists", False),
)
@fiat_router.post(
"/{provider}/subscription",
status_code=HTTPStatus.OK,
-1
View File
@@ -263,7 +263,6 @@ async def api_payments_create(
payment_request=invoice_data.bolt11,
extra=invoice_data.extra,
labels=invoice_data.labels,
external_id=invoice_data.external_id,
)
return payment
-6
View File
@@ -9,8 +9,6 @@ from lnbits.fiat.base import FiatProvider
from lnbits.settings import settings
from .paypal import PayPalWallet
from .revolut import RevolutWallet
from .square import SquareWallet
from .stripe import StripeWallet
fiat_module = importlib.import_module("lnbits.fiat")
@@ -19,8 +17,6 @@ fiat_module = importlib.import_module("lnbits.fiat")
class FiatProviderType(Enum):
stripe = "StripeWallet"
paypal = "PayPalWallet"
square = "SquareWallet"
revolut = "RevolutWallet"
async def get_fiat_provider(name: str) -> FiatProvider | None:
@@ -56,7 +52,5 @@ fiat_providers: dict[str, FiatProvider] = {}
__all__ = [
"PayPalWallet",
"RevolutWallet",
"SquareWallet",
"StripeWallet",
]
+25 -7
View File
@@ -169,7 +169,7 @@ class PayPalWallet(FiatProvider):
return FiatInvoiceResponse(
ok=True,
checking_id=f"fiat_paypal_{order_id}",
checking_id=order_id,
payment_request=approval_url,
)
except Exception as exc:
@@ -285,6 +285,25 @@ class PayPalWallet(FiatProvider):
async def get_payment_status(self, checking_id: str) -> FiatPaymentStatus:
raise NotImplementedError("PayPal does not support outgoing payments.")
async def capture_order(self, checking_id: str) -> FiatPaymentStatus:
try:
await self._ensure_access_token()
paypal_id = self._normalize_paypal_id(checking_id)
if paypal_id.startswith("subscription_"):
logger.warning("PayPal subscriptions do not support order capture.")
return FiatPaymentPendingStatus()
r = await self.client.post(
f"/v2/checkout/orders/{paypal_id}/capture",
json={},
headers=self._auth_headers(),
)
r.raise_for_status()
return self._status_from_order(r.json())
except Exception as exc:
logger.warning(f"Error capturing PayPal order '{checking_id}': {exc}")
return await self.get_invoice_status(checking_id)
async def paid_invoices_stream(self) -> AsyncGenerator[str, None]:
logger.warning(
"PayPal does not support paid invoices stream. Use webhooks instead."
@@ -296,7 +315,7 @@ class PayPalWallet(FiatProvider):
def _status_from_order(self, order: dict[str, Any]) -> FiatPaymentStatus:
status = (order.get("status") or "").upper()
if status in ["COMPLETED", "APPROVED"]:
if status == "COMPLETED":
return FiatPaymentSuccessStatus()
if status in ["VOIDED", "CANCELLED", "CANCELED"]:
return FiatPaymentFailedStatus()
@@ -311,11 +330,10 @@ class PayPalWallet(FiatProvider):
return FiatPaymentPendingStatus()
def _normalize_paypal_id(self, checking_id: str) -> str:
return (
checking_id.replace("fiat_paypal_", "", 1)
if checking_id.startswith("fiat_paypal_")
else checking_id
)
normalized = checking_id
while normalized.startswith("fiat_paypal_"):
normalized = normalized.replace("fiat_paypal_", "", 1)
return normalized
def _serialize_metadata(
self, payment_options: FiatSubscriptionPaymentOptions
-487
View File
@@ -1,487 +0,0 @@
import asyncio
import ipaddress
import json
from collections.abc import AsyncGenerator
from typing import Any
from urllib.parse import urlparse
import httpx
from loguru import logger
from pydantic import BaseModel, Field, ValidationError
from lnbits.helpers import normalize_endpoint, urlsafe_short_hash
from lnbits.settings import settings
from .base import (
FiatInvoiceResponse,
FiatPaymentFailedStatus,
FiatPaymentPendingStatus,
FiatPaymentResponse,
FiatPaymentStatus,
FiatPaymentSuccessStatus,
FiatProvider,
FiatStatusResponse,
FiatSubscriptionPaymentOptions,
FiatSubscriptionResponse,
)
class RevolutCheckoutOptions(BaseModel):
class Config:
extra = "ignore"
success_url: str | None = None
metadata: dict[str, Any] = Field(default_factory=dict)
description: str | None = None
class RevolutCreateInvoiceOptions(BaseModel):
class Config:
extra = "ignore"
checkout: RevolutCheckoutOptions | None = None
class RevolutSubscriptionReference(BaseModel):
wallet_id: str
tag: str | None = None
subscription_request_id: str | None = None
extra: dict[str, Any] | None = None
memo: str | None = None
REVOLUT_WEBHOOK_EVENTS = [
"ORDER_AUTHORISED",
"ORDER_COMPLETED",
"SUBSCRIPTION_INITIATED",
]
class RevolutWallet(FiatProvider):
"""https://developer.revolut.com/docs/merchant"""
def __init__(self):
logger.debug("Initializing RevolutWallet")
self._settings_fields = self._settings_connection_fields()
if not settings.revolut_api_endpoint:
raise ValueError("Cannot initialize RevolutWallet: missing endpoint.")
if not settings.revolut_api_secret_key:
raise ValueError("Cannot initialize RevolutWallet: missing API secret key.")
self.endpoint = normalize_endpoint(settings.revolut_api_endpoint)
self.headers = {
"Authorization": f"Bearer {settings.revolut_api_secret_key}",
"Revolut-Api-Version": settings.revolut_api_version,
"Content-Type": "application/json",
"User-Agent": settings.user_agent,
}
self.client = httpx.AsyncClient(base_url=self.endpoint, headers=self.headers)
logger.info("RevolutWallet initialized.")
async def cleanup(self):
try:
await self.client.aclose()
except RuntimeError as e:
logger.warning(f"Error closing Revolut 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("/api/orders", params={"limit": 1}, 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 opts is None:
return FiatInvoiceResponse(
ok=False, error_message="Invalid Revolut options"
)
amount_minor = int(amount * 100)
checkout = opts.checkout or RevolutCheckoutOptions()
success_url = (
checkout.success_url
or settings.revolut_payment_success_url
or "https://lnbits.com"
)
payload = {
"amount": amount_minor,
"currency": currency.upper(),
"description": checkout.description or memo or "LNbits Invoice",
"redirect_url": success_url,
"metadata": {
**checkout.metadata,
"payment_hash": payment_hash,
"alan_action": "invoice",
},
}
try:
r = await self.client.post("/api/orders", json=payload)
r.raise_for_status()
data = r.json()
order_id = data.get("id")
checkout_url = data.get("checkout_url")
if not order_id or not checkout_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=checkout_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:
if quantity != 1:
return FiatSubscriptionResponse(
ok=False,
error_message="Revolut subscriptions do not support quantity.",
)
wallet_id = payment_options.wallet_id
if not wallet_id:
return FiatSubscriptionResponse(
ok=False, error_message="Wallet ID is required."
)
extra = payment_options.extra or {}
customer_id = extra.get("customer_id")
if not customer_id:
return FiatSubscriptionResponse(
ok=False,
error_message="Revolut subscriptions require extra.customer_id.",
)
if not payment_options.subscription_request_id:
payment_options.subscription_request_id = urlsafe_short_hash()
reference = RevolutSubscriptionReference(
wallet_id=wallet_id,
tag=payment_options.tag,
subscription_request_id=payment_options.subscription_request_id,
extra=extra,
memo=payment_options.memo,
)
payload: dict[str, Any] = {
"plan_variation_id": subscription_id,
"customer_id": customer_id,
"external_reference": self._serialize_subscription_reference(reference),
"setup_order_redirect_url": (
payment_options.success_url
or settings.revolut_payment_success_url
or "https://lnbits.com"
),
}
if extra.get("trial_duration"):
payload["trial_duration"] = extra["trial_duration"]
headers = {
**self.headers,
"Idempotency-Key": payment_options.subscription_request_id,
}
try:
r = await self.client.post(
"/api/subscriptions", json=payload, headers=headers
)
r.raise_for_status()
data = r.json()
revolut_subscription_id = data.get("id")
setup_order_id = data.get("setup_order_id")
if not revolut_subscription_id or not setup_order_id:
return FiatSubscriptionResponse(
ok=False,
error_message=(
"Server error: missing subscription id or setup order id"
),
)
setup_order = await self.get_order(setup_order_id)
checkout_url = setup_order.get("checkout_url")
if not checkout_url:
return FiatSubscriptionResponse(
ok=False, error_message="Server error: missing setup checkout url"
)
return FiatSubscriptionResponse(
ok=True,
checkout_session_url=checkout_url,
subscription_request_id=revolut_subscription_id,
)
except json.JSONDecodeError:
return FiatSubscriptionResponse(
ok=False, error_message="Server error: invalid json response"
)
except Exception as exc:
logger.warning(exc)
return FiatSubscriptionResponse(
ok=False, error_message=f"Unable to connect to {self.endpoint}."
)
async def cancel_subscription(
self,
subscription_id: str,
correlation_id: str,
**kwargs,
) -> FiatSubscriptionResponse:
try:
r = await self.client.post(f"/api/subscriptions/{subscription_id}/cancel")
r.raise_for_status()
return FiatSubscriptionResponse(ok=True)
except Exception as exc:
logger.warning(exc)
return FiatSubscriptionResponse(
ok=False, error_message="Unable to cancel subscription."
)
async def pay_invoice(self, payment_request: str) -> FiatPaymentResponse:
raise NotImplementedError("Revolut does not support paying invoices directly.")
async def get_invoice_status(self, checking_id: str) -> FiatPaymentStatus:
try:
order_id = self._normalize_revolut_id(checking_id)
return self._status_from_order(await self.get_order(order_id))
except Exception as exc:
logger.debug(f"Error getting Revolut invoice status: {exc}")
return FiatPaymentPendingStatus()
async def get_payment_status(self, checking_id: str) -> FiatPaymentStatus:
raise NotImplementedError("Revolut does not support outgoing payments.")
async def paid_invoices_stream(self) -> AsyncGenerator[str, None]:
logger.warning(
"Revolut 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
def _normalize_revolut_id(self, checking_id: str) -> str:
value = (
checking_id.replace("fiat_revolut_", "", 1)
if checking_id.startswith("fiat_revolut_")
else checking_id
)
return value.replace("order_", "", 1) if value.startswith("order_") else value
async def get_order(self, order_id: str) -> dict[str, Any]:
r = await self.client.get(f"/api/orders/{order_id}")
r.raise_for_status()
return r.json()
async def get_subscription(self, subscription_id: str) -> dict[str, Any]:
r = await self.client.get(f"/api/subscriptions/{subscription_id}")
r.raise_for_status()
return r.json()
async def get_subscription_cycle(
self, subscription_id: str, cycle_id: str
) -> dict[str, Any]:
r = await self.client.get(
f"/api/subscriptions/{subscription_id}/cycles/{cycle_id}"
)
r.raise_for_status()
return r.json()
@classmethod
async def create_webhook(
cls,
url: str,
endpoint: str | None = None,
api_secret_key: str | None = None,
api_version: str | None = None,
) -> dict[str, Any]:
if not url:
raise ValueError("Missing Revolut webhook URL.")
cls._validate_webhook_url(url)
if not endpoint and not settings.revolut_api_endpoint:
raise ValueError("Missing Revolut API endpoint.")
if not api_secret_key and not settings.revolut_api_secret_key:
raise ValueError("Missing Revolut API secret key.")
base_url = normalize_endpoint(endpoint or settings.revolut_api_endpoint)
secret_key = api_secret_key or settings.revolut_api_secret_key
headers = {
"Authorization": f"Bearer {secret_key}",
"Revolut-Api-Version": api_version or settings.revolut_api_version,
"Content-Type": "application/json",
"User-Agent": settings.user_agent,
}
payload = {"url": url, "events": REVOLUT_WEBHOOK_EVENTS}
async with httpx.AsyncClient(base_url=base_url, headers=headers) as client:
webhooks = await cls._list_webhooks(client)
existing = await cls._get_existing_webhook(client, webhooks, url)
if existing:
existing["already_exists"] = True
return existing
response = await client.post("/api/webhooks", json=payload, timeout=15)
response.raise_for_status()
return response.json()
@classmethod
async def _list_webhooks(cls, client: httpx.AsyncClient) -> list[dict[str, Any]]:
response = await client.get("/api/webhooks", timeout=15)
response.raise_for_status()
data = response.json()
if isinstance(data, list):
return data
if isinstance(data, dict):
for field in ["webhooks", "data", "items"]:
if isinstance(data.get(field), list):
return data[field]
return []
@classmethod
async def _get_existing_webhook(
cls, client: httpx.AsyncClient, webhooks: list[dict[str, Any]], url: str
) -> dict[str, Any] | None:
for webhook in webhooks:
if cls._normalize_webhook_url(webhook.get("url")) != (
cls._normalize_webhook_url(url)
):
continue
webhook_id = webhook.get("id")
if webhook_id and (
not webhook.get("events") or not webhook.get("signing_secret")
):
response = await client.get(f"/api/webhooks/{webhook_id}", timeout=15)
response.raise_for_status()
webhook = response.json()
events = set(webhook.get("events") or [])
missing_events = set(REVOLUT_WEBHOOK_EVENTS) - events
if missing_events:
raise ValueError(
"A Revolut webhook already exists for this URL, but it is "
f"missing required events: {', '.join(sorted(missing_events))}."
)
if not webhook.get("signing_secret"):
raise ValueError(
"A Revolut webhook already exists for this URL, but Revolut "
"did not return a signing secret."
)
return webhook
return None
@classmethod
def _normalize_webhook_url(cls, url: str | None) -> str:
return (url or "").strip().rstrip("/")
@classmethod
def _validate_webhook_url(cls, url: str) -> None:
parsed = urlparse(url)
hostname = parsed.hostname
if parsed.scheme not in ["http", "https"] or not hostname:
raise ValueError("Revolut webhook URL must be a clearnet URL.")
host = hostname.lower()
if host == "localhost" or host.endswith(".localhost"):
raise ValueError("Revolut webhook URL must be a clearnet URL.")
if host.endswith(".local") or host.endswith(".onion"):
raise ValueError("Revolut webhook URL must be a clearnet URL.")
try:
ip = ipaddress.ip_address(host)
except ValueError:
return
if (
ip.is_loopback
or ip.is_private
or ip.is_link_local
or ip.is_reserved
or ip.is_unspecified
):
raise ValueError("Revolut webhook URL must be a clearnet URL.")
def _status_from_order(self, order: dict[str, Any]) -> FiatPaymentStatus:
status = (order.get("state") or "").upper()
if status == "COMPLETED":
return FiatPaymentSuccessStatus()
if status in ["CANCELLED", "FAILED"]:
return FiatPaymentFailedStatus()
return FiatPaymentPendingStatus()
def _parse_create_opts(
self, raw_opts: dict[str, Any]
) -> RevolutCreateInvoiceOptions | None:
try:
return RevolutCreateInvoiceOptions.parse_obj(raw_opts)
except ValidationError as e:
logger.warning(f"Invalid Revolut options: {e}")
return None
def _serialize_subscription_reference(
self, reference: RevolutSubscriptionReference
) -> str:
payload = reference.dict(exclude_none=True)
serialized = json.dumps(payload, separators=(",", ":"))
if len(serialized) > 1024:
raise ValueError("Revolut subscription external_reference is too long.")
return serialized
def deserialize_subscription_reference(
self, external_reference: str | None
) -> RevolutSubscriptionReference | None:
if not external_reference:
return None
try:
return RevolutSubscriptionReference.parse_obj(
json.loads(external_reference)
)
except (json.JSONDecodeError, ValidationError) as exc:
logger.warning(exc)
return None
def _settings_connection_fields(self) -> str:
return "-".join(
[
str(settings.revolut_api_endpoint),
str(settings.revolut_api_secret_key),
str(settings.revolut_api_version),
str(settings.revolut_webhook_signing_secret),
]
)
-616
View File
@@ -1,616 +0,0 @@
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, urlsafe_short_hash
from lnbits.settings import settings
from .base import (
FiatInvoiceResponse,
FiatPaymentFailedStatus,
FiatPaymentPendingStatus,
FiatPaymentResponse,
FiatPaymentStatus,
FiatPaymentSuccessStatus,
FiatProvider,
FiatStatusResponse,
FiatSubscriptionPaymentOptions,
FiatSubscriptionResponse,
)
FiatMethod = Literal["checkout", "subscription"]
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 SquareSubscriptionOptions(BaseModel):
class Config:
extra = "ignore"
checking_id: str | None = None
payment_request: str | None = None
class SquareCreateInvoiceOptions(BaseModel):
class Config:
extra = "ignore"
fiat_method: FiatMethod = "checkout"
checkout: SquareCheckoutOptions | None = None
subscription: SquareSubscriptionOptions | None = None
class SquareSubscriptionCheckoutInfo(BaseModel):
plan_variation_id: str
price_money: dict[str, Any]
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")
if opts.fiat_method == "subscription":
return self._create_subscription_invoice(opts.subscription)
return await self._create_checkout_invoice(
amount=amount,
payment_hash=payment_hash,
currency=currency,
opts=opts,
memo=memo,
)
async def create_subscription(
self,
subscription_id: str,
quantity: int,
payment_options: FiatSubscriptionPaymentOptions,
**kwargs,
) -> FiatSubscriptionResponse:
success_url = (
payment_options.success_url
or settings.square_payment_success_url
or "https://lnbits.com"
)
if not payment_options.subscription_request_id:
payment_options.subscription_request_id = urlsafe_short_hash()
payment_options.extra = payment_options.extra or {}
payment_options.extra["subscription_request_id"] = (
payment_options.subscription_request_id
)
try:
checkout_info = await self._get_subscription_checkout_info(subscription_id)
metadata = self._serialize_metadata(payment_options)
payload = {
"idempotency_key": payment_options.subscription_request_id,
"description": metadata,
"quick_pay": {
"name": (payment_options.memo or "LNbits Subscription")[:255],
"price_money": checkout_info.price_money,
"location_id": self.location_id,
},
"checkout_options": {
"redirect_url": success_url,
"subscription_plan_id": checkout_info.plan_variation_id,
},
"payment_note": metadata,
}
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 {}
url = payment_link.get("url")
if not url:
return FiatSubscriptionResponse(
ok=False, error_message="Server error: missing url"
)
return FiatSubscriptionResponse(
ok=True,
checkout_session_url=url,
subscription_request_id=payment_options.subscription_request_id,
)
except json.JSONDecodeError as exc:
logger.warning(exc)
return FiatSubscriptionResponse(
ok=False, error_message="Server error: invalid json response"
)
except Exception as exc:
logger.warning(exc)
return FiatSubscriptionResponse(
ok=False, error_message=f"Unable to connect to {self.endpoint}."
)
async def cancel_subscription(
self,
subscription_id: str,
correlation_id: str,
**kwargs,
) -> FiatSubscriptionResponse:
try:
square_subscription_id = await self._get_square_subscription_id(
subscription_id, correlation_id
)
r = await self.client.post(
f"/v2/subscriptions/{square_subscription_id}/cancel"
)
r.raise_for_status()
return FiatSubscriptionResponse(ok=True)
except Exception as exc:
logger.warning(exc)
return FiatSubscriptionResponse(
ok=False, error_message="Unable to cancel subscription."
)
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:
order = await self._get_order(order_id)
payment_id = self._payment_id_from_order(order)
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_order(self, order_id: str) -> dict[str, Any]:
r = await self.client.get(f"/v2/orders/{order_id}")
r.raise_for_status()
return r.json().get("order") or {}
async def get_payment_for_order(self, order_id: str) -> dict[str, Any] | None:
order = await self._get_order(order_id)
payment_id = self._payment_id_from_order(order)
if not payment_id:
return None
return await self._get_payment(payment_id)
def _payment_id_from_order(self, order: dict[str, Any]) -> str | None:
tenders = order.get("tenders") or []
for tender in tenders:
payment_id = tender.get("payment_id")
if payment_id:
return payment_id
return None
async def _get_payment_status(self, payment_id: str) -> FiatPaymentStatus:
return self._status_from_payment(await self._get_payment(payment_id))
async def _get_payment(self, payment_id: str) -> dict[str, Any]:
r = await self.client.get(f"/v2/payments/{payment_id}")
r.raise_for_status()
return r.json().get("payment") or {}
async def _get_subscription_checkout_info(
self, subscription_plan_id: str
) -> SquareSubscriptionCheckoutInfo:
catalog_object = await self._get_catalog_object(subscription_plan_id)
if catalog_object.get("type") == "SUBSCRIPTION_PLAN":
return await self._get_plan_checkout_info(catalog_object)
if catalog_object.get("type") == "SUBSCRIPTION_PLAN_VARIATION":
price_money = await self._get_subscription_price_money(
catalog_object,
)
plan_variation_id = catalog_object.get("id")
if not plan_variation_id:
raise ValueError("Square subscription plan variation is missing an ID.")
return SquareSubscriptionCheckoutInfo(
plan_variation_id=plan_variation_id,
price_money=price_money,
)
raise ValueError(
"Square subscription ID must be a plan ID or plan variation ID."
)
async def _get_plan_checkout_info(
self, catalog_object: dict[str, Any]
) -> SquareSubscriptionCheckoutInfo:
plan_data = catalog_object.get("subscription_plan_data") or {}
plan_variations = plan_data.get("subscription_plan_variations") or []
eligible_item_ids = plan_data.get("eligible_item_ids") or []
plan_variation = next(
(
variation
for variation in plan_variations
if not variation.get("is_deleted")
),
None,
)
if not plan_variation:
raise ValueError("Square subscription plan is missing a variation.")
price_money = await self._get_subscription_price_money(
plan_variation,
eligible_item_ids=eligible_item_ids,
)
plan_variation_id = plan_variation.get("id")
if not plan_variation_id:
raise ValueError("Square subscription plan variation is missing an ID.")
return SquareSubscriptionCheckoutInfo(
plan_variation_id=plan_variation_id,
price_money=price_money,
)
async def _get_catalog_object(self, object_id: str) -> dict[str, Any]:
r = await self.client.get(f"/v2/catalog/object/{object_id}")
r.raise_for_status()
return r.json().get("object") or {}
async def _get_subscription_price_money(
self,
plan_variation: dict[str, Any],
eligible_item_ids: list[str] | None = None,
) -> dict[str, Any]:
variation_data = plan_variation.get("subscription_plan_variation_data") or {}
phases = variation_data.get("phases") or []
for phase in phases:
pricing = phase.get("pricing") or {}
price_money = pricing.get("price_money") or phase.get(
"recurring_price_money"
)
parsed_price_money = self._parse_price_money(price_money)
if parsed_price_money:
return parsed_price_money
if pricing.get("type") == "RELATIVE":
return await self._get_relative_subscription_price_money(
eligible_item_ids or []
)
raise ValueError("Square subscription plan variation is missing price_money.")
async def _get_relative_subscription_price_money(
self, eligible_item_ids: list[str]
) -> dict[str, Any]:
if len(eligible_item_ids) != 1:
raise ValueError(
"Square relative subscription plan must have exactly one item."
)
item = await self._get_catalog_object(eligible_item_ids[0])
item_variations: list[dict[str, Any]] = []
if item.get("type") == "ITEM":
item_variations = (item.get("item_data") or {}).get("variations") or []
elif item.get("type") == "ITEM_VARIATION":
item_variations = [item]
item_variation = next(
(
variation
for variation in item_variations
if not variation.get("is_deleted")
),
None,
)
if not item_variation:
raise ValueError("Square subscription item is missing a variation.")
price_money = self._parse_price_money(
(item_variation.get("item_variation_data") or {}).get("price_money")
)
if price_money:
return price_money
raise ValueError("Square subscription item variation is missing price_money.")
def _parse_price_money(
self, price_money: dict[str, Any] | None
) -> dict[str, Any] | None:
if (
price_money
and price_money.get("amount") is not None
and price_money.get("currency")
):
return {
"amount": int(price_money["amount"]),
"currency": price_money["currency"].upper(),
}
return None
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()
async def _create_checkout_invoice(
self,
amount: float,
payment_hash: str,
currency: str,
opts: SquareCreateInvoiceOptions,
memo: str | None = None,
) -> FiatInvoiceResponse:
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}."
)
def _create_subscription_invoice(
self, opts: SquareSubscriptionOptions | None
) -> FiatInvoiceResponse:
term = opts or SquareSubscriptionOptions()
checking_id = term.checking_id or f"payment_{urlsafe_short_hash()}"
return FiatInvoiceResponse(
ok=True,
checking_id=checking_id,
payment_request=term.payment_request or "",
)
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 _serialize_metadata(
self, payment_options: FiatSubscriptionPaymentOptions
) -> str:
extra_link = None
if payment_options.extra:
raw_link = payment_options.extra.get("link")
extra_link = str(raw_link)[:200] if raw_link else None
meta = [
payment_options.wallet_id,
payment_options.tag,
payment_options.subscription_request_id,
extra_link,
]
memo_limit = 493 - len(json.dumps(meta, separators=(",", ":")))
if memo_limit > 0 and payment_options.memo:
meta.append(payment_options.memo[:memo_limit])
else:
meta.append(None)
metadata = json.dumps(meta, separators=(",", ":"))
if len(metadata) > 500:
raise ValueError("Square subscription metadata is too long.")
return metadata
async def _get_square_subscription_id(
self, subscription_id: str, wallet_id: str
) -> str:
try:
from lnbits.core.crud.payments import get_payments
from lnbits.core.models import PaymentFilters
from lnbits.db import Filter, Filters
payments = await get_payments(
wallet_id=wallet_id,
filters=Filters(
filters=[
Filter.parse_query(
"external_id", [subscription_id], PaymentFilters
)
],
model=PaymentFilters,
sortby="created_at",
direction="desc",
limit=1,
),
)
payment = next(
(
payment
for payment in payments
if payment.external_id and payment.fiat_provider == "square"
),
None,
)
if payment and payment.external_id:
return payment.external_id
payments = await get_payments(
wallet_id=wallet_id,
incoming=True,
filters=Filters(
model=PaymentFilters,
sortby="created_at",
direction="desc",
),
)
payment = next(
(
payment
for payment in payments
if payment.external_id
and payment.fiat_provider == "square"
and (payment.extra or {}).get("subscription_request_id")
== subscription_id
),
None,
)
if payment and payment.external_id:
return payment.external_id
except Exception as exc:
logger.warning(exc)
return subscription_id
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),
]
)
+1 -51
View File
@@ -703,35 +703,6 @@ 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 RevolutFiatProvider(LNbitsSettings):
revolut_enabled: bool = Field(default=False)
revolut_api_endpoint: str = Field(default="https://merchant.revolut.com")
revolut_api_secret_key: str | None = Field(default=None)
revolut_api_version: str = Field(default="2026-04-20")
revolut_payment_success_url: str = Field(default="https://lnbits.com")
revolut_payment_webhook_url: str = Field(
default="https://your-lnbits-domain-here.com/api/v1/callback/revolut"
)
revolut_webhook_signing_secret: str | None = Field(default=None)
revolut_limits: FiatProviderLimits = Field(default_factory=FiatProviderLimits)
class LightningSettings(LNbitsSettings):
lightning_invoice_expiry: int = Field(default=3600, gt=0)
@@ -769,12 +740,7 @@ class FundingSourcesSettings(
funding_source_max_retries: int = Field(default=4, ge=0)
class FiatProvidersSettings(
StripeFiatProvider,
PayPalFiatProvider,
SquareFiatProvider,
RevolutFiatProvider,
):
class FiatProvidersSettings(StripeFiatProvider, PayPalFiatProvider):
def is_fiat_provider_enabled(self, provider: str | None) -> bool:
"""
Checks if a specific fiat provider is enabled.
@@ -785,10 +751,6 @@ class FiatProvidersSettings(
return self.stripe_enabled
if provider == "paypal":
return self.paypal_enabled
if provider == "square":
return self.square_enabled
if provider == "revolut":
return self.revolut_enabled
return False
def get_fiat_providers_for_user(self, user_id: str) -> list[str]:
@@ -808,18 +770,6 @@ class FiatProvidersSettings(
):
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")
if self.revolut_enabled and (
not self.revolut_limits.allowed_users
or user_id in self.revolut_limits.allowed_users
):
allowed_providers.append("revolut")
return allowed_providers
def get_fiat_provider_limits(self, provider_name: str) -> FiatProviderLimits | None:
File diff suppressed because one or more lines are too long
+10 -10
View File
File diff suppressed because one or more lines are too long
-11
View File
@@ -267,15 +267,6 @@ 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 this exact LNbits URL.',
square_webhook_url_hint:
'Must exactly match the Square notification URL. LNbits requires the /api/v1/callback/square path.',
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!',
@@ -814,8 +805,6 @@ 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,9 +5,6 @@ window.app.component('lnbits-admin-fiat-providers', {
return {
formAddStripeUser: '',
formAddPaypalUser: '',
formAddSquareUser: '',
formAddRevolutUser: '',
creatingRevolutWebhook: false,
hideInputToggle: true
}
},
@@ -23,12 +20,6 @@ window.app.component('lnbits-admin-fiat-providers', {
this.formData?.paypal_payment_webhook_url ||
this.calculateWebhookUrl('paypal')
)
},
revolutWebhookUrl() {
return (
this.formData?.revolut_payment_webhook_url ||
this.calculateWebhookUrl('revolut')
)
}
},
watch: {
@@ -67,8 +58,6 @@ 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')
this.maybeSetWebhookUrl('revolut_payment_webhook_url', 'revolut')
},
maybeSetWebhookUrl(fieldName, provider) {
if (!this.formData) {
@@ -88,47 +77,6 @@ window.app.component('lnbits-admin-fiat-providers', {
}
this.copyText(url)
},
isClearnetWebhookUrl(url) {
let parsedUrl
try {
parsedUrl = new URL(url)
} catch (e) {
return false
}
const host = parsedUrl.hostname.toLowerCase()
if (!['http:', 'https:'].includes(parsedUrl.protocol)) {
return false
}
if (
host === 'localhost' ||
host.endsWith('.localhost') ||
host.endsWith('.local') ||
host.endsWith('.onion')
) {
return false
}
if (
/^127\./.test(host) ||
/^10\./.test(host) ||
/^192\.168\./.test(host) ||
/^169\.254\./.test(host) ||
/^172\.(1[6-9]|2\d|3[0-1])\./.test(host) ||
host === '0.0.0.0' ||
host === '::1'
) {
return false
}
return true
},
notifyRevolutWebhookWarning(message) {
Quasar.Notify.create({
type: 'warning',
message,
icon: null,
closeBtn: true
})
},
addStripeAllowedUser() {
const addUser = this.formAddStripeUser || ''
if (
@@ -163,40 +111,6 @@ 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)
},
addRevolutAllowedUser() {
const addUser = this.formAddRevolutUser || ''
if (
addUser.length &&
!this.formData.revolut_limits.allowed_users.includes(addUser)
) {
this.formData.revolut_limits.allowed_users = [
...this.formData.revolut_limits.allowed_users,
addUser
]
this.formAddRevolutUser = ''
}
},
removeRevolutAllowedUser(user) {
this.formData.revolut_limits.allowed_users =
this.formData.revolut_limits.allowed_users.filter(u => u !== user)
},
checkFiatProvider(providerName) {
LNbits.api
.request('PUT', `/api/v1/fiat/check/${providerName}`)
@@ -210,48 +124,6 @@ window.app.component('lnbits-admin-fiat-providers', {
})
})
.catch(LNbits.utils.notifyApiError)
},
createRevolutWebhook() {
const webhookUrl = this.calculateWebhookUrl('revolut')
this.formData.revolut_payment_webhook_url = webhookUrl
if (!this.formData.revolut_api_secret_key) {
this.notifyRevolutWebhookWarning(
'Add your Revolut API secret key before creating a webhook.'
)
return
}
if (!this.isClearnetWebhookUrl(webhookUrl)) {
this.notifyRevolutWebhookWarning(
'Revolut webhook URL must be a clearnet URL.'
)
return
}
this.creatingRevolutWebhook = true
LNbits.api
.request('POST', '/api/v1/fiat/revolut/webhook', null, {
url: webhookUrl,
endpoint: this.formData.revolut_api_endpoint,
api_secret_key: this.formData.revolut_api_secret_key,
api_version: this.formData.revolut_api_version
})
.then(response => {
const data = response.data
this.formData.revolut_payment_webhook_url = data.url
this.formData.revolut_webhook_signing_secret = data.signing_secret
Quasar.Notify.create({
type: 'positive',
message: `Revolut webhook ${
data.already_exists ? 'already exists' : 'created'
}${data.id ? `: ${data.id}` : ''}.`,
icon: null
})
})
.catch(LNbits.utils.notifyApiError)
.finally(() => {
this.creatingRevolutWebhook = false
})
}
}
})
+1 -1
View File
@@ -21,7 +21,7 @@ window.PageHome = {
return (
this.lnurl !== '' &&
this.g.settings.allowRegister &&
this.g.settings.authMethods.includes('user-id-only')
'user-id-only' in this.g.settings.authMethods
)
},
formatDescription() {
@@ -587,543 +587,12 @@
<q-item-section> Square </q-item-section>
<q-item-section side>
<div class="row items-center">
<q-toggle
size="md"
:label="$t('enabled')"
v-model="formData.square_enabled"
color="green"
unchecked-icon="clear"
/>
</div>
<div class="row items-center">Disabled</div>
</q-item-section>
</template>
<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"
v-model="formData.square_payment_webhook_url"
:label="$t('webhook_url')"
:hint="$t('square_webhook_url_hint')"
></q-input>
</div>
<div class="col-auto">
<q-btn
outline
color="grey"
icon="content_copy"
@click="
copyWebhookUrl(formData.square_payment_webhook_url)
"
: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>
<li><code>invoice.payment_made</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-expansion-item header-class="text-primary text-bold">
<template v-slot:header>
<q-item-section avatar>
<q-avatar color="deep-orange-7" text-color="white">R</q-avatar>
</q-item-section>
<q-item-section> Revolut </q-item-section>
<q-item-section side>
<div class="row items-center">
<q-toggle
size="md"
:label="$t('enabled')"
v-model="formData.revolut_enabled"
color="green"
unchecked-icon="clear"
/>
</div>
</q-item-section>
</template>
<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.revolut_api_endpoint"
:label="$t('endpoint')"
></q-input>
<q-input
filled
class="q-mt-md"
:type="hideInputToggle ? 'password' : 'text'"
v-model="formData.revolut_api_secret_key"
label="API secret key"
></q-input>
<q-input
filled
class="q-mt-md"
type="text"
v-model="formData.revolut_api_version"
:label="$t('api_version')"
></q-input>
<q-input
filled
class="q-mt-md"
type="text"
v-model="formData.revolut_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('revolut')"
></q-btn>
</div>
</div>
</q-card-section>
</q-expansion-item>
<q-expansion-item :label="$t('webhook')" default-opened>
<q-card-section>
Configure a Revolut Merchant webhook that points to your LNbits
server. LNbits will create it through the Revolut API and
subscribe to <code>ORDER_AUTHORISED</code>,
<code>ORDER_COMPLETED</code>, and
<code>SUBSCRIPTION_INITIATED</code>.
</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="revolutWebhookUrl"
:label="$t('webhook_url')"
readonly
></q-input>
</div>
<div class="col-auto">
<q-btn
outline
color="grey"
icon="content_copy"
@click="copyWebhookUrl(revolutWebhookUrl)"
:aria-label="$t('copy_webhook_url')"
>
<q-tooltip>
<span v-text="$t('copy_webhook_url')"></span>
</q-tooltip>
</q-btn>
</div>
</div>
<div class="row items-center q-gutter-sm q-mt-md">
<q-btn
type="button"
color="primary"
icon="add_link"
label="Create webhook"
:loading="creatingRevolutWebhook"
@click="createRevolutWebhook"
></q-btn>
<q-chip
v-if="formData.revolut_webhook_signing_secret"
dense
color="positive"
text-color="white"
icon="verified"
>
Signing secret saved
</q-chip>
</div>
</q-card-section>
<q-card-section>
<span v-text="$t('webhook_events_list')"></span>
<ul>
<li>
<code>ORDER_AUTHORISED</code>
</li>
<li>
<code>ORDER_COMPLETED</code>
</li>
<li>
<code>SUBSCRIPTION_INITIATED</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.revolut_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.revolut_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.revolut_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.revolut_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.revolut_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.revolut_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-card-section>
</q-expansion-item>
<q-expansion-item :label="$t('allowed_users')">
<q-card-section>
<q-input
filled
v-model="formAddRevolutUser"
@keydown.enter="addRevolutAllowedUser"
type="text"
:label="$t('allowed_users_label')"
:hint="
$t('allowed_users_hint_feature', {
feature: 'Revolut'
})
"
>
<q-btn
@click="addRevolutAllowedUser"
dense
flat
icon="add"
></q-btn>
</q-input>
<div>
<q-chip
v-for="user in formData.revolut_limits.allowed_users"
@update:model-value="formData.touch = null"
:key="user"
removable
@remove="removeRevolutAllowedUser(user)"
color="primary"
text-color="white"
:label="user"
class="ellipsis"
>
</q-chip>
</div>
</q-card-section>
</q-expansion-item>
<q-card>
<q-card-section> Coming Soon </q-card-section>
</q-card>
</q-expansion-item>
</q-list>
@@ -1171,7 +640,9 @@
>
</div>
<div class="row items-center q-gutter-sm">
<div class="text-bold" style="min-width: 140px">Square</div>
<div class="text-bold" style="min-width: 140px">
Square (coming soon)
</div>
<q-chip dense color="positive" text-color="white" icon="check"
>Checkout</q-chip
>
@@ -1182,22 +653,7 @@
>Tap-to-pay</q-chip
>
<q-chip dense color="grey-9" text-color="white" icon="public"
>Regions: Square-supported countries</q-chip
>
</div>
<div class="row items-center q-gutter-sm">
<div class="text-bold" style="min-width: 140px">Revolut</div>
<q-chip dense color="positive" text-color="white" icon="check"
>Checkout</q-chip
>
<q-chip dense color="positive" text-color="white" icon="check"
>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: Revolut-supported countries</q-chip
>Regions: Global</q-chip
>
</div>
</div>
-38
View File
@@ -394,44 +394,6 @@
<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-separator
v-if="g.user.fiat_providers?.includes('revolut')"
></q-separator>
<q-item
v-if="g.user.fiat_providers?.includes('revolut')"
:active="receive.fiatProvider === 'revolut'"
@click="receive.fiatProvider = 'revolut'"
active-class="bg-teal-1 text-grey-8 text-weight-bold"
clickable
v-ripple
>
<q-item-section avatar>
<q-avatar color="deep-orange-7" text-color="white">R</q-avatar>
</q-item-section>
<q-item-section>
<span v-text="$t('pay_with', {provider: 'Revolut'})"></span>
</q-item-section>
</q-item>
</q-list>
</div>
+12 -353
View File
@@ -4,18 +4,13 @@ from uuid import uuid4
import pytest
from httpx import AsyncClient
from lnbits.core.models import Account, CreateInvoice, Payment
from lnbits.core.models import Account, CreateInvoice
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_revolut_event,
handle_square_event,
handle_stripe_event,
)
from lnbits.fiat.revolut import RevolutWallet
from lnbits.fiat.square import SquareWallet
from lnbits.settings import Settings
@pytest.mark.anyio
@@ -28,15 +23,7 @@ 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()
)
revolut_mock = mocker.patch(
"lnbits.core.views.callback_api.handle_revolut_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.check_revolut_signature")
mocker.patch(
"lnbits.core.views.callback_api.verify_paypal_webhook", mocker.AsyncMock()
)
@@ -58,27 +45,6 @@ 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()
revolut = await http_client.post(
"/api/v1/callback/revolut",
headers={
"Revolut-Signature": "sig",
"Revolut-Request-Timestamp": "1700000000",
},
json={"event": "ORDER_COMPLETED", "order_id": "order_1"},
)
assert revolut.status_code == 200
assert revolut.json()["success"] is True
revolut_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
@@ -116,163 +82,29 @@ async def test_callback_api_handles_paid_events_with_real_payments(mocker):
)
await handle_paypal_event(
{
"id": "evt_paypal",
"id": "evt_paypal_approved",
"event_type": "CHECKOUT.ORDER.APPROVED",
"resource": {
"purchase_units": [{"invoice_id": payment.payment_hash}],
},
}
)
await handle_paypal_event(
{
"id": "evt_paypal",
"event_type": "PAYMENT.CAPTURE.COMPLETED",
"resource": {
"purchase_units": [{"invoice_id": payment.payment_hash}],
},
}
)
await handle_stripe_event({"id": "evt_unhandled", "type": "customer.created"})
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_revolut_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_revolut_event(
{
"event": "ORDER_COMPLETED",
"order_id": "order_1",
}
)
get_payment.assert_awaited_once_with("fiat_revolut_order_order_1")
fiat_status_mock.assert_awaited_once_with(payment)
@pytest.mark.anyio
async def test_callback_api_handles_revolut_subscription_event(
mocker, settings: Settings
):
user = await create_user_account(
Account(
id=uuid4().hex,
username=f"user_{uuid4().hex[:8]}",
email=f"user_{uuid4().hex[:8]}@lnbits.com",
)
)
wallet = user.wallets[0]
payment = await create_wallet_invoice(
wallet.id, CreateInvoice(out=False, amount=15, memo="subscription")
)
settings.revolut_api_endpoint = "https://sandbox-merchant.revolut.com"
settings.revolut_api_secret_key = "revolut-secret"
settings.revolut_api_version = "2026-04-20"
revolut_provider = RevolutWallet()
mocker.patch.object(
revolut_provider,
"get_subscription",
return_value={
"id": "SUBSCRIPTION_1",
"current_cycle_id": "CYCLE_1",
"external_reference": json.dumps(
{
"wallet_id": wallet.id,
"tag": "members",
"subscription_request_id": "request_1",
"extra": {"link": "link-1", "customer_id": "customer_1"},
"memo": "Revolut Members",
}
),
},
)
mocker.patch.object(
revolut_provider,
"get_subscription_cycle",
return_value={"id": "CYCLE_1", "order_id": "ORDER_SUB_1"},
)
mocker.patch.object(
revolut_provider,
"get_order",
return_value={
"id": "ORDER_SUB_1",
"amount": 925,
"currency": "USD",
"checkout_url": "https://checkout.revolut.com/payment-link/sub_1",
},
)
mocker.patch(
"lnbits.core.views.callback_api.get_fiat_provider",
mocker.AsyncMock(return_value=revolut_provider),
)
mocker.patch(
"lnbits.core.views.callback_api.get_standalone_payment",
mocker.AsyncMock(side_effect=[None]),
)
create_fiat_invoice_mock = mocker.patch(
"lnbits.core.views.callback_api.create_fiat_invoice",
mocker.AsyncMock(return_value=payment),
)
fiat_status_mock = mocker.patch(
"lnbits.core.views.callback_api.check_fiat_status", mocker.AsyncMock()
)
await handle_revolut_event(
{
"event": "SUBSCRIPTION_INITIATED",
"subscription_id": "SUBSCRIPTION_1",
}
)
assert create_fiat_invoice_mock.await_count == 1
revolut_call = create_fiat_invoice_mock.await_args.kwargs
assert revolut_call["wallet_id"] == wallet.id
invoice = revolut_call["invoice_data"]
assert invoice.fiat_provider == "revolut"
assert invoice.amount == 9.25
assert invoice.memo == "Revolut Members"
assert invoice.external_id == "SUBSCRIPTION_1"
assert invoice.extra["fiat_method"] == "subscription"
assert invoice.extra["subscription"]["checking_id"] == "order_ORDER_SUB_1"
fiat_status_mock.assert_awaited_once_with(payment)
@pytest.mark.anyio
async def test_callback_api_handles_subscription_flows_and_validation(
mocker, settings: Settings
):
async def test_callback_api_handles_subscription_flows_and_validation(mocker):
user = await create_user_account(
Account(
id=uuid4().hex,
@@ -340,99 +172,6 @@ async def test_callback_api_handles_subscription_flows_and_validation(
)
assert create_fiat_invoice_mock.await_count == 2
await handle_square_event(
{
"event_id": "evt_square_subscription",
"type": "payment.updated",
"data": {
"object": {
"payment": {
"id": "PAYMENT_SUB_1",
"order_id": "ORDER_SUB_1",
"status": "COMPLETED",
"amount_money": {"amount": 925, "currency": "USD"},
"note": json.dumps(
[
wallet.id,
"members",
"subscription_square_1",
"link-1",
"Square Members",
]
),
}
}
},
}
)
assert create_fiat_invoice_mock.await_count == 3
square_call = create_fiat_invoice_mock.await_args.kwargs
assert square_call["wallet_id"] == wallet.id
square_invoice = square_call["invoice_data"]
assert square_invoice.fiat_provider == "square"
assert square_invoice.amount == 9.25
assert square_invoice.memo == "Square Members"
assert square_invoice.extra["fiat_method"] == "subscription"
assert square_invoice.extra["tag"] == "members"
assert (
square_invoice.extra["subscription"]["checking_id"] == "payment_PAYMENT_SUB_1"
)
payment.extra = {
"subscription_request_id": "subscription_square_1",
"tag": "members",
"link": "link-1",
}
payment.external_id = "SUBSCRIPTION_1"
payment.memo = "Square Members"
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"
square_provider = SquareWallet()
mocker.patch.object(
square_provider,
"get_payment_for_order",
return_value={
"id": "PAYMENT_SUB_2",
"status": "COMPLETED",
"amount_money": {"amount": 925, "currency": "USD"},
},
)
mocker.patch(
"lnbits.core.views.callback_api.get_fiat_provider",
mocker.AsyncMock(return_value=square_provider),
)
mocker.patch(
"lnbits.core.views.callback_api.get_payments",
mocker.AsyncMock(return_value=[payment]),
)
await handle_square_event(
{
"event_id": "evt_square_invoice",
"type": "invoice.payment_made",
"data": {
"object": {
"invoice": {
"order_id": "ORDER_SUB_2",
"subscription_id": "SUBSCRIPTION_1",
"public_url": "https://square.example/invoice",
}
}
},
}
)
assert create_fiat_invoice_mock.await_count == 4
square_invoice_call = create_fiat_invoice_mock.await_args.kwargs
square_invoice = square_invoice_call["invoice_data"]
assert square_invoice.external_id == "SUBSCRIPTION_1"
assert "square_subscription_id" not in square_invoice.extra
assert (
square_invoice.extra["subscription"]["payment_request"]
== "https://square.example/invoice"
)
with pytest.raises(
ValueError, match="PayPal subscription event missing custom metadata."
):
@@ -443,83 +182,3 @@ async def test_callback_api_handles_subscription_flows_and_validation(
"resource": {"amount": {"currency": "USD", "total": "5.00"}},
}
)
@pytest.mark.anyio
async def test_square_invoice_payment_updates_existing_subscription_external_id(
settings: Settings, mocker
):
payment = Payment(
checking_id="fiat_square_payment_PAYMENT_SUB_1",
payment_hash="hash_square_subscription",
wallet_id="wallet_1",
amount=925000,
fee=0,
bolt11="lnbc1square",
fiat_provider="square",
extra={
"subscription_request_id": "subscription_square_1",
"tag": "members",
"link": "link-1",
},
memo="Square Members",
)
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"
square_provider = SquareWallet()
mocker.patch.object(
square_provider,
"get_payment_for_order",
return_value={
"id": "PAYMENT_SUB_1",
"status": "COMPLETED",
"amount_money": {"amount": 925, "currency": "USD"},
},
)
mocker.patch(
"lnbits.core.views.callback_api.get_fiat_provider",
mocker.AsyncMock(return_value=square_provider),
)
get_standalone_payment_mock = mocker.patch(
"lnbits.core.views.callback_api.get_standalone_payment",
mocker.AsyncMock(return_value=payment),
)
update_payment_mock = mocker.patch(
"lnbits.core.views.callback_api.update_payment",
mocker.AsyncMock(),
)
get_payments_mock = mocker.patch(
"lnbits.core.views.callback_api.get_payments",
mocker.AsyncMock(return_value=[]),
)
create_fiat_invoice_mock = mocker.patch(
"lnbits.core.views.callback_api.create_fiat_invoice",
mocker.AsyncMock(),
)
fiat_status_mock = mocker.patch(
"lnbits.core.views.callback_api.check_fiat_status", mocker.AsyncMock()
)
await handle_square_event(
{
"event_id": "evt_square_invoice",
"type": "invoice.payment_made",
"data": {
"object": {
"invoice": {
"order_id": "ORDER_SUB_1",
"subscription_id": "SUBSCRIPTION_1",
}
}
},
}
)
get_standalone_payment_mock.assert_awaited_with("fiat_square_payment_PAYMENT_SUB_1")
assert payment.external_id == "SUBSCRIPTION_1"
update_payment_mock.assert_awaited_once_with(payment)
fiat_status_mock.assert_awaited_once_with(payment)
get_payments_mock.assert_not_awaited()
create_fiat_invoice_mock.assert_not_awaited()
-81
View File
@@ -4,8 +4,6 @@ from pytest_mock.plugin import MockerFixture
from lnbits.core.models.misc import SimpleStatus
from lnbits.fiat.base import FiatSubscriptionResponse
from lnbits.fiat.revolut import REVOLUT_WEBHOOK_EVENTS
from lnbits.settings import Settings
class _UnsetSecret:
@@ -146,82 +144,3 @@ async def test_fiat_api_connection_token_validates_provider_configuration(
assert ok.status_code == 200
assert ok.json() == {"secret": "tok_live"}
assert good_provider.await_count == 1
@pytest.mark.anyio
async def test_fiat_api_creates_revolut_webhook(
client: AsyncClient,
superuser_token: str,
settings: Settings,
mocker: MockerFixture,
):
create_webhook = mocker.patch(
"lnbits.core.views.fiat_api.RevolutWallet.create_webhook",
mocker.AsyncMock(
return_value={
"id": "webhook_1",
"url": "https://lnbits.example/api/v1/callback/revolut",
"events": REVOLUT_WEBHOOK_EVENTS,
"signing_secret": "whsec_1",
}
),
)
response = await client.post(
"/api/v1/fiat/revolut/webhook",
headers={"Authorization": f"Bearer {superuser_token}"},
json={
"url": "https://lnbits.example/api/v1/callback/revolut",
"endpoint": "https://sandbox-merchant.revolut.com",
"api_secret_key": "secret_1",
"api_version": "2026-04-20",
},
)
assert response.status_code == 200
assert response.json() == {
"id": "webhook_1",
"url": "https://lnbits.example/api/v1/callback/revolut",
"events": REVOLUT_WEBHOOK_EVENTS,
"signing_secret": "whsec_1",
"already_exists": False,
}
create_webhook.assert_awaited_once_with(
url="https://lnbits.example/api/v1/callback/revolut",
endpoint="https://sandbox-merchant.revolut.com",
api_secret_key="secret_1",
api_version="2026-04-20",
)
assert settings.revolut_payment_webhook_url == (
"https://lnbits.example/api/v1/callback/revolut"
)
assert settings.revolut_webhook_signing_secret == "whsec_1"
@pytest.mark.anyio
async def test_fiat_api_rejects_local_revolut_webhook(
client: AsyncClient,
superuser_token: str,
mocker: MockerFixture,
):
create_webhook = mocker.patch(
"lnbits.core.views.fiat_api.RevolutWallet.create_webhook",
mocker.AsyncMock(
side_effect=ValueError("Revolut webhook URL must be a clearnet URL.")
),
)
response = await client.post(
"/api/v1/fiat/revolut/webhook",
headers={"Authorization": f"Bearer {superuser_token}"},
json={
"url": "http://localhost:5000/api/v1/callback/revolut",
"endpoint": "https://sandbox-merchant.revolut.com",
"api_secret_key": "secret_1",
"api_version": "2026-04-20",
},
)
assert response.status_code == 400
assert response.json()["detail"] == ("Revolut webhook URL must be a clearnet URL.")
create_webhook.assert_awaited_once()
+3 -58
View File
@@ -4,10 +4,9 @@ from uuid import uuid4
import pytest
from fastapi import HTTPException
from pydantic import ValidationError
from lnbits.core.crud.payments import create_payment, get_payments
from lnbits.core.models import Account, CreateInvoice, PaymentFilters, PaymentState
from lnbits.core.crud.payments import create_payment
from lnbits.core.models import Account, CreateInvoice, PaymentState
from lnbits.core.models.payments import CancelInvoice, CreatePayment, SettleInvoice
from lnbits.core.models.users import AccountId
from lnbits.core.models.wallets import KeyType, WalletTypeInfo
@@ -22,7 +21,7 @@ from lnbits.core.views.payment_api import (
api_payments_settle,
api_payments_wallets_stats,
)
from lnbits.db import Filter, Filters
from lnbits.db import Filters
from lnbits.wallets.base import InvoiceResponse
ZERO_AMOUNT_INVOICE = (
@@ -92,60 +91,6 @@ async def test_payment_api_stats_and_all_paginated(admin_user):
assert second_wallet.id in wallet_ids
@pytest.mark.anyio
async def test_payment_external_id_is_stored_and_validated():
user = await create_user_account(
Account(
id=uuid4().hex,
username=f"user_{uuid4().hex[:8]}",
email=f"user_{uuid4().hex[:8]}@lnbits.com",
)
)
wallet = user.wallets[0]
first_payment = await create_wallet_invoice(
wallet.id,
CreateInvoice(
out=False,
amount=21,
memo="external reference",
external_id="provider_payment_123",
),
)
second_payment = await create_wallet_invoice(
wallet.id,
CreateInvoice(
out=False,
amount=22,
memo="external reference newest",
external_id="provider_payment_123",
),
)
assert first_payment.external_id == "provider_payment_123"
assert second_payment.external_id == "provider_payment_123"
stored_payments = await get_payments(
wallet_id=wallet.id,
filters=Filters(
filters=[
Filter.parse_query(
"external_id", ["provider_payment_123"], PaymentFilters
)
],
model=PaymentFilters,
sortby="created_at",
direction="desc",
),
)
assert [payment.checking_id for payment in stored_payments] == [
second_payment.checking_id,
first_payment.checking_id,
]
with pytest.raises(ValidationError, match="Invalid external id"):
CreateInvoice(out=False, amount=21, external_id="provider payment 123")
@pytest.mark.anyio
async def test_payment_api_fee_reserve_and_hold_invoice_actions(mocker):
user = await create_user_account(
-1
View File
@@ -1 +0,0 @@
98ae578877a3479bb0424d2e418b427a
-1
View File
@@ -1 +0,0 @@
6355d3c6c5df49dbb562a74f9deb19ce
+79 -781
View File
@@ -1,8 +1,6 @@
import hashlib
import hmac
import json
import time
from base64 import b64encode
from unittest.mock import AsyncMock
import pytest
@@ -17,8 +15,6 @@ 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_revolut_signature,
check_square_signature,
check_stripe_signature,
handle_fiat_payment_confirmation,
verify_paypal_webhook,
@@ -27,14 +23,14 @@ from lnbits.core.services.fiat_providers import (
test_connection as fiat_provider_connection,
)
from lnbits.core.services.users import create_user_account
from lnbits.core.views.callback_api import handle_paypal_event
from lnbits.fiat.paypal import PayPalWallet
from lnbits.fiat.base import (
FiatInvoiceResponse,
FiatPaymentStatus,
FiatPaymentSuccessStatus,
FiatStatusResponse,
FiatSubscriptionPaymentOptions,
)
from lnbits.fiat.revolut import REVOLUT_WEBHOOK_EVENTS, RevolutWallet
from lnbits.fiat.square import SquareWallet
from lnbits.settings import Settings
from tests.helpers import get_random_string
@@ -67,56 +63,16 @@ 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)
original_revolut_enabled = settings.revolut_enabled
original_revolut_api_endpoint = settings.revolut_api_endpoint
original_revolut_api_secret_key = settings.revolut_api_secret_key
original_revolut_api_version = settings.revolut_api_version
original_revolut_payment_success_url = settings.revolut_payment_success_url
original_revolut_payment_webhook_url = settings.revolut_payment_webhook_url
original_revolut_webhook_signing_secret = settings.revolut_webhook_signing_secret
original_revolut_limits = settings.revolut_limits.copy(deep=True)
settings.lnbits_allowed_currencies = []
settings.paypal_enabled = False
settings.square_enabled = False
settings.revolut_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
settings.revolut_enabled = original_revolut_enabled
settings.revolut_api_endpoint = original_revolut_api_endpoint
settings.revolut_api_secret_key = original_revolut_api_secret_key
settings.revolut_api_version = original_revolut_api_version
settings.revolut_payment_success_url = original_revolut_payment_success_url
settings.revolut_payment_webhook_url = original_revolut_payment_webhook_url
settings.revolut_webhook_signing_secret = original_revolut_webhook_signing_secret
settings.revolut_limits = original_revolut_limits
@pytest.mark.anyio
@@ -181,39 +137,6 @@ 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"]
settings.square_enabled = False
settings.revolut_enabled = True
settings.revolut_limits.allowed_users = []
user = await get_user(to_user.id)
assert user
assert user.fiat_providers == ["revolut"]
settings.revolut_limits.allowed_users = ["some_other_user_id"]
user = await get_user(to_user.id)
assert user
assert user.fiat_providers == []
settings.revolut_limits.allowed_users.append(to_user.id)
user = await get_user(to_user.id)
assert user
assert user.fiat_providers == ["revolut"]
@pytest.mark.anyio
async def test_create_wallet_fiat_invoice_fiat_limits_fail(
@@ -365,700 +288,39 @@ async def test_create_wallet_fiat_invoice_success(
@pytest.mark.anyio
async def test_create_wallet_square_fiat_invoice_success(
async def test_create_paypal_fiat_invoice_uses_raw_order_id(
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
settings.paypal_enabled = True
settings.paypal_client_id = "client-id"
settings.paypal_client_secret = "client-secret"
settings.paypal_limits.service_min_amount_sats = 0
settings.paypal_limits.service_max_amount_sats = 0
settings.paypal_limits.service_faucet_wallet_id = None
invoice_data = CreateInvoice(
unit="USD", amount=1.0, memo="Test", fiat_provider="square"
unit="USD", amount=1.0, memo="Test", fiat_provider="paypal"
)
fiat_mock_response = FiatInvoiceResponse(
ok=True,
checking_id="order_123",
payment_request="https://square.link/u/session_123",
checking_id="ORDER123",
payment_request="https://paypal.com/checkoutnow?token=ORDER123",
)
mocker.patch(
"lnbits.fiat.SquareWallet.create_invoice",
"lnbits.fiat.PayPalWallet.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 # type: ignore[assignment]
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_create_subscription(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={
"object": {
"type": "SUBSCRIPTION_PLAN_VARIATION",
"id": "PLAN_VARIATION_123",
"subscription_plan_variation_data": {
"phases": [
{
"ordinal": 0,
"pricing": {
"type": "STATIC",
"price_money": {
"amount": 1500,
"currency": "USD",
},
},
}
]
},
}
}
),
MockHTTPResponse(
json_data={
"payment_link": {
"id": "plink_123",
"url": "https://square.link/u/sub_123",
}
}
),
]
)
wallet.client = client # type: ignore[assignment]
payment_options = FiatSubscriptionPaymentOptions(
wallet_id="wallet_1",
memo="Monthly Gold",
tag="gold",
extra={"link": "link-1"},
success_url="https://lnbits.example/subscription-success",
)
response = await wallet.create_subscription(
"PLAN_VARIATION_123", 1, payment_options
)
assert response.ok is True
assert response.checkout_session_url == "https://square.link/u/sub_123"
assert response.subscription_request_id is not None
assert client.calls[0][0] == "/v2/catalog/object/PLAN_VARIATION_123"
assert client.calls[1][0] == "/v2/online-checkout/payment-links"
payload = client.calls[1][1]["json"]
assert payload["idempotency_key"] == response.subscription_request_id
assert payload["quick_pay"]["location_id"] == "LOC123"
assert payload["quick_pay"]["price_money"] == {"amount": 1500, "currency": "USD"}
assert payload["checkout_options"] == {
"redirect_url": "https://lnbits.example/subscription-success",
"subscription_plan_id": "PLAN_VARIATION_123",
}
metadata = json.loads(payload["payment_note"])
assert metadata[:3] == ["wallet_1", "gold", response.subscription_request_id]
assert metadata[3:] == ["link-1", "Monthly Gold"]
@pytest.mark.anyio
async def test_square_wallet_create_subscription_from_plan_id(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={
"object": {
"type": "SUBSCRIPTION_PLAN",
"id": "PLAN123",
"subscription_plan_data": {
"name": "LNbits Test Weekly Personal Plan",
"subscription_plan_variations": [
{
"type": "SUBSCRIPTION_PLAN_VARIATION",
"id": "PLAN_VARIATION_123",
"subscription_plan_variation_data": {
"name": "LNbits Test Weekly Personal Plan",
"phases": [
{
"uid": "PHASE123",
"cadence": "WEEKLY",
"ordinal": 0,
"pricing": {"type": "RELATIVE"},
}
],
"subscription_plan_id": "PLAN123",
},
}
],
"eligible_item_ids": ["ITEM123"],
"all_items": False,
},
}
}
),
MockHTTPResponse(
json_data={
"object": {
"type": "ITEM",
"id": "ITEM123",
"item_data": {
"name": "LNbits Test Weekly Personal Plan",
"variations": [
{
"type": "ITEM_VARIATION",
"id": "ITEM_VARIATION_123",
"item_variation_data": {
"item_id": "ITEM123",
"name": "Regular",
"pricing_type": "FIXED_PRICING",
"price_money": {
"amount": 1500,
"currency": "USD",
},
},
}
],
},
}
}
),
MockHTTPResponse(
json_data={
"payment_link": {
"id": "plink_123",
"url": "https://square.link/u/sub_123",
}
}
),
]
)
wallet.client = client # type: ignore[assignment]
response = await wallet.create_subscription(
"PLAN123",
1,
FiatSubscriptionPaymentOptions(
wallet_id="wallet_1",
memo="Weekly Plan",
success_url="https://lnbits.example/success",
),
)
assert response.ok is True
assert client.calls[0][0] == "/v2/catalog/object/PLAN123"
assert client.calls[1][0] == "/v2/catalog/object/ITEM123"
assert client.calls[2][0] == "/v2/online-checkout/payment-links"
payload = client.calls[2][1]["json"]
assert payload["quick_pay"]["price_money"] == {"amount": 1500, "currency": "USD"}
assert payload["checkout_options"] == {
"redirect_url": "https://lnbits.example/success",
"subscription_plan_id": "PLAN_VARIATION_123",
}
@pytest.mark.anyio
async def test_square_wallet_create_subscription_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"
wallet = SquareWallet()
response = await wallet.create_invoice(
amount=15,
payment_hash="hash123",
currency="USD",
memo="Square subscription payment",
extra={
"fiat_method": "subscription",
"subscription": {
"checking_id": "payment_PAYMENT123",
"payment_request": "https://square.example/invoice",
},
},
)
assert response.ok is True
assert response.checking_id == "payment_PAYMENT123"
assert response.payment_request == "https://square.example/invoice"
@pytest.mark.anyio
async def test_square_wallet_cancel_subscription(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={"subscription": {}})])
wallet.client = client # type: ignore[assignment]
response = await wallet.cancel_subscription("SUBSCRIPTION123", "wallet_1")
assert response.ok is True
assert client.calls[0][0] == "/v2/subscriptions/SUBSCRIPTION123/cancel"
@pytest.mark.anyio
async def test_square_wallet_cancel_subscription_by_request_id(
settings: Settings, mocker: MockerFixture
):
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={"subscription": {}})])
wallet.client = client # type: ignore[assignment]
payment = Payment(
checking_id="fiat_square_payment_PAYMENT123",
payment_hash="hash123",
wallet_id="wallet_1",
amount=1000,
fee=0,
bolt11="lnbc1square",
fiat_provider="square",
extra={"subscription_request_id": "REQUEST123"},
external_id="SUBSCRIPTION123",
)
get_payments_mock = mocker.patch(
"lnbits.core.crud.payments.get_payments",
AsyncMock(side_effect=[[], [payment]]),
)
response = await wallet.cancel_subscription("REQUEST123", "wallet_1")
assert response.ok is True
assert client.calls[0][0] == "/v2/subscriptions/SUBSCRIPTION123/cancel"
assert get_payments_mock.await_count == 2
@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 # type: ignore[assignment]
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_create_wallet_revolut_fiat_invoice_success(
to_wallet: Wallet, settings: Settings, mocker: MockerFixture
):
settings.revolut_enabled = True
settings.revolut_api_secret_key = "revolut-secret"
settings.revolut_limits.service_min_amount_sats = 0
settings.revolut_limits.service_max_amount_sats = 0
settings.revolut_limits.service_faucet_wallet_id = None
invoice_data = CreateInvoice(
unit="USD", amount=1.0, memo="Test", fiat_provider="revolut"
)
fiat_mock_response = FiatInvoiceResponse(
ok=True,
checking_id="order_ORDER123",
payment_request="https://checkout.revolut.com/payment-link/ORDER123",
)
mocker.patch(
"lnbits.fiat.RevolutWallet.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 == "revolut"
assert payment.extra.get("fiat_checking_id") == fiat_mock_response.checking_id
assert payment.checking_id.startswith("fiat_revolut_order_ORDER123")
@pytest.mark.anyio
async def test_revolut_wallet_create_invoice(settings: Settings):
settings.revolut_api_endpoint = "https://sandbox-merchant.revolut.com"
settings.revolut_api_secret_key = "revolut-secret"
settings.revolut_api_version = "2026-04-20"
settings.revolut_payment_success_url = "https://lnbits.example/success"
wallet = RevolutWallet()
client = MockHTTPClient(
[
MockHTTPResponse(
json_data={
"id": "ORDER123",
"checkout_url": "https://checkout.revolut.com/payment-link/abc123",
}
)
]
)
wallet.client = client # type: ignore[assignment]
response = await wallet.create_invoice(
amount=1.23,
payment_hash="hash123",
currency="USD",
memo="LNbits Revolut invoice",
extra={"checkout": {"metadata": {"source": "test"}}},
)
assert response.ok is True
assert response.checking_id == "order_ORDER123"
assert (
response.payment_request == "https://checkout.revolut.com/payment-link/abc123"
)
assert client.calls[0][0] == "/api/orders"
payload = client.calls[0][1]["json"]
assert payload["amount"] == 123
assert payload["currency"] == "USD"
assert payload["metadata"]["payment_hash"] == "hash123"
assert payload["metadata"]["alan_action"] == "invoice"
assert payload["metadata"]["source"] == "test"
assert payload["redirect_url"] == "https://lnbits.example/success"
@pytest.mark.anyio
async def test_revolut_wallet_get_invoice_status(settings: Settings):
settings.revolut_api_endpoint = "https://sandbox-merchant.revolut.com"
settings.revolut_api_secret_key = "revolut-secret"
settings.revolut_api_version = "2026-04-20"
wallet = RevolutWallet()
client = MockHTTPClient([MockHTTPResponse(json_data={"state": "COMPLETED"})])
wallet.client = client # type: ignore[assignment]
status = await wallet.get_invoice_status("fiat_revolut_order_ORDER123")
assert status.success is True
assert client.calls[0][0] == "/api/orders/ORDER123"
@pytest.mark.anyio
async def test_revolut_wallet_create_subscription(settings: Settings):
settings.revolut_api_endpoint = "https://sandbox-merchant.revolut.com"
settings.revolut_api_secret_key = "revolut-secret"
settings.revolut_api_version = "2026-04-20"
settings.revolut_payment_success_url = "https://lnbits.example/subscription-success"
wallet = RevolutWallet()
client = MockHTTPClient(
[
MockHTTPResponse(
json_data={
"id": "SUBSCRIPTION123",
"setup_order_id": "ORDER123",
}
),
MockHTTPResponse(
json_data={
"id": "ORDER123",
"checkout_url": "https://checkout.revolut.com/payment-link/sub_123",
}
),
]
)
wallet.client = client # type: ignore[assignment]
payment_options = FiatSubscriptionPaymentOptions(
wallet_id="wallet_1",
memo="Monthly Gold",
tag="gold",
extra={"customer_id": "CUSTOMER123", "link": "link-1"},
success_url="https://lnbits.example/subscription-success",
)
response = await wallet.create_subscription(
"PLAN_VARIATION_123", 1, payment_options
)
assert response.ok is True
assert response.subscription_request_id == "SUBSCRIPTION123"
assert (
response.checkout_session_url
== "https://checkout.revolut.com/payment-link/sub_123"
)
assert client.calls[0][0] == "/api/subscriptions"
payload = client.calls[0][1]["json"]
assert payload["plan_variation_id"] == "PLAN_VARIATION_123"
assert payload["customer_id"] == "CUSTOMER123"
assert payload["setup_order_redirect_url"] == (
"https://lnbits.example/subscription-success"
)
reference = json.loads(payload["external_reference"])
assert reference["wallet_id"] == "wallet_1"
assert reference["tag"] == "gold"
assert reference["memo"] == "Monthly Gold"
assert reference["extra"]["customer_id"] == "CUSTOMER123"
assert client.calls[1][0] == "/api/orders/ORDER123"
@pytest.mark.anyio
async def test_revolut_wallet_cancel_subscription(settings: Settings):
settings.revolut_api_endpoint = "https://sandbox-merchant.revolut.com"
settings.revolut_api_secret_key = "revolut-secret"
settings.revolut_api_version = "2026-04-20"
wallet = RevolutWallet()
client = MockHTTPClient([MockHTTPResponse(json_data={})])
wallet.client = client # type: ignore[assignment]
response = await wallet.cancel_subscription("SUBSCRIPTION123", "wallet_1")
assert response.ok is True
assert client.calls[0][0] == "/api/subscriptions/SUBSCRIPTION123/cancel"
@pytest.mark.anyio
async def test_revolut_wallet_create_webhook(mocker: MockerFixture):
client = MockHTTPClient(
[
MockHTTPResponse({"webhooks": []}),
MockHTTPResponse(
{
"id": "webhook_1",
"url": "https://lnbits.example/api/v1/callback/revolut",
"events": REVOLUT_WEBHOOK_EVENTS,
"signing_secret": "whsec_1",
}
),
]
)
async_client = mocker.patch("lnbits.fiat.revolut.httpx.AsyncClient")
async_client.return_value = client
response = await RevolutWallet.create_webhook(
url="https://lnbits.example/api/v1/callback/revolut",
endpoint="https://sandbox-merchant.revolut.com",
api_secret_key="revolut-secret",
api_version="2026-04-20",
)
assert response["signing_secret"] == "whsec_1"
async_client.assert_called_once()
assert async_client.call_args.kwargs["base_url"] == (
"https://sandbox-merchant.revolut.com"
)
assert async_client.call_args.kwargs["headers"]["Authorization"] == (
"Bearer revolut-secret"
)
assert client.calls == [
(
"/api/webhooks",
{
"timeout": 15,
},
),
(
"/api/webhooks",
{
"json": {
"url": "https://lnbits.example/api/v1/callback/revolut",
"events": REVOLUT_WEBHOOK_EVENTS,
},
"timeout": 15,
},
),
]
@pytest.mark.anyio
async def test_revolut_wallet_reuses_existing_webhook(mocker: MockerFixture):
client = MockHTTPClient(
[
MockHTTPResponse(
{
"webhooks": [
{
"id": "webhook_1",
"url": "https://lnbits.example/api/v1/callback/revolut",
"events": REVOLUT_WEBHOOK_EVENTS,
"signing_secret": "whsec_1",
}
]
}
)
]
)
async_client = mocker.patch("lnbits.fiat.revolut.httpx.AsyncClient")
async_client.return_value = client
response = await RevolutWallet.create_webhook(
url="https://lnbits.example/api/v1/callback/revolut",
endpoint="https://sandbox-merchant.revolut.com",
api_secret_key="revolut-secret",
api_version="2026-04-20",
)
assert response["already_exists"] is True
assert response["signing_secret"] == "whsec_1"
assert client.calls == [
(
"/api/webhooks",
{
"timeout": 15,
},
)
]
@pytest.mark.anyio
async def test_revolut_wallet_rejects_local_webhook_url():
with pytest.raises(ValueError, match="clearnet URL"):
await RevolutWallet.create_webhook(
url="http://localhost:5000/api/v1/callback/revolut",
endpoint="https://sandbox-merchant.revolut.com",
api_secret_key="revolut-secret",
api_version="2026-04-20",
)
def test_check_revolut_signature():
payload = b'{"event":"ORDER_COMPLETED","order_id":"ORDER123"}'
timestamp = str(int(time.time()))
secret = "revolut-secret"
sig = hmac.new(secret.encode(), payload, hashlib.sha256).hexdigest()
check_revolut_signature(payload, sig, timestamp, secret)
def test_check_revolut_signature_millisecond_timestamp():
payload = b'{"event":"ORDER_COMPLETED","order_id":"ORDER123"}'
timestamp = str(int(time.time() * 1000))
secret = "revolut-secret"
sig = hmac.new(secret.encode(), payload, hashlib.sha256).hexdigest()
check_revolut_signature(payload, sig, timestamp, secret)
def test_check_revolut_signature_v1_header():
payload = b'{"event":"ORDER_COMPLETED","order_id":"ORDER123"}'
timestamp = str(int(time.time() * 1000))
secret = "revolut-secret"
signed_payload = b"v1." + timestamp.encode() + b"." + payload
sig = "v1=" + hmac.new(secret.encode(), signed_payload, hashlib.sha256).hexdigest()
check_revolut_signature(payload, sig, timestamp, secret)
def test_check_revolut_signature_multiple_v1_headers():
payload = b'{"event":"ORDER_COMPLETED","order_id":"ORDER123"}'
timestamp = str(int(time.time() * 1000))
secret = "revolut-secret"
signed_payload = b"v1." + timestamp.encode() + b"." + payload
valid_sig = (
"v1=" + hmac.new(secret.encode(), signed_payload, hashlib.sha256).hexdigest()
)
sig_header = f"v1=deadbeef,{valid_sig}"
check_revolut_signature(payload, sig_header, timestamp, secret)
def test_check_revolut_signature_docs_vector():
payload = (
b'{"data":{"id":"645a7696-22f3-aa47-9c74-cbae0449cc46",'
b'"new_state":"completed","old_state":"pending",'
b'"request_id":"app_charges-9f5d5eb3-1e06-46c5-b1c0-3914763e0bcb"},'
b'"event":"TransactionStateChanged",'
b'"timestamp":"2023-05-09T16:36:38.028960Z"}'
)
timestamp = "1683650202360"
secret = "wsk_r59a4HfWVAKycbCaNO1RvgCJec02gRd8"
sig = "v1=bca326fb378d0da7f7c490ad584a8106bab9723d8d9cdd0d50b4c5b3be3837c0"
check_revolut_signature(
payload, sig, timestamp, secret, tolerance_seconds=100000000
)
assert payment.fiat_provider == "paypal"
assert payment.extra.get("fiat_checking_id") == "ORDER123"
assert payment.checking_id == "fiat_paypal_ORDER123"
@pytest.mark.anyio
@@ -1250,31 +512,6 @@ 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:
@@ -1443,6 +680,67 @@ async def test_verify_paypal_webhook_raises_on_failed_verification(
)
def test_paypal_order_status_approved_is_pending():
wallet = object.__new__(PayPalWallet)
approved = wallet._status_from_order({"status": "APPROVED"})
completed = wallet._status_from_order({"status": "COMPLETED"})
assert approved.pending is True
assert approved.success is False
assert completed.success is True
def test_paypal_normalize_id_removes_legacy_double_prefix():
wallet = object.__new__(PayPalWallet)
assert wallet._normalize_paypal_id("fiat_paypal_ORDER123") == "ORDER123"
assert wallet._normalize_paypal_id("fiat_paypal_fiat_paypal_ORDER123") == "ORDER123"
@pytest.mark.anyio
async def test_handle_paypal_approved_event_captures_order(mocker: MockerFixture):
payment = Payment(
checking_id="fiat_paypal_ORDER123",
payment_hash="hash_123",
wallet_id="wallet_id",
amount=1000,
fee=0,
bolt11="bolt11",
status=PaymentState.PENDING,
fiat_provider="paypal",
extra={"fiat_checking_id": "ORDER123"},
)
provider = mocker.Mock(spec=PayPalWallet)
provider.capture_order = AsyncMock(return_value=FiatPaymentSuccessStatus())
mocker.patch(
"lnbits.core.views.callback_api.get_standalone_payment",
AsyncMock(return_value=payment),
)
mocker.patch(
"lnbits.core.views.callback_api.get_fiat_provider",
AsyncMock(return_value=provider),
)
status_mock = mocker.patch(
"lnbits.core.views.callback_api.check_fiat_status",
AsyncMock(),
)
await handle_paypal_event(
{
"id": "evt_paypal_approved",
"event_type": "CHECKOUT.ORDER.APPROVED",
"resource": {
"purchase_units": [{"invoice_id": payment.payment_hash}],
},
}
)
provider.capture_order.assert_awaited_once_with("ORDER123")
status_mock.assert_not_awaited()
@pytest.mark.anyio
async def test_test_connection_reports_provider_status(mocker: MockerFixture):
mocker.patch(