Compare commits

..
Author SHA1 Message Date
dni 842fb53742 mock import 2026-07-09 08:19:25 +02:00
dni 2f9b3723ae black 2026-07-09 08:05:59 +02:00
dni 04397b85f0 also activate extension for start task in test 2026-07-09 08:05:59 +02:00
dni 8982852516 dont start tasks twice 2026-07-09 08:05:59 +02:00
dni e2bee23daf fix pydantic v1 errors 2026-07-09 08:05:59 +02:00
dni a822523460 clear cached modules 2026-07-09 08:05:59 +02:00
dni 4971f4c5c0 fix import 2026-07-09 08:05:59 +02:00
dni d157c43ebb remove old files 2026-07-09 08:05:59 +02:00
dni bb53fe41c8 remove uppgrade_hash 2026-07-09 08:05:59 +02:00
dni 5865a3f96a refactor: get rid of url rewrites for extensions /upgrades/
should open the way for custom root-path
2026-07-09 08:05:59 +02:00
Vlad StanandGitHub cabb58f8fe chore: make bundle (#4045) 2026-07-08 16:59:28 +03:00
aac04efd0e Merge commit from fork
* fix: don't show unsafe html in lnurlp notifications

* fix: lnurlp success aes decryption

* regenerate bundle

---------

Co-authored-by: Vlad Stan <stan.v.vlad@gmail.com>
2026-07-08 16:49:43 +03:00
fd9009f760 feat: create TaskManager singleton class (#3775)
Co-authored-by: dadofsambonzuki <dadofsambonzuki@users.noreply.github.com>
2026-07-08 14:07:38 +02:00
dni ⚡andGitHub 6ab413775a chore: update to v1.5.6 (#4044) 2026-07-08 13:54:35 +02:00
7eb9965205 feat: keep funding source on default reset, force seed backup (#4040)
Co-authored-by: alan <alan@lnbits.com>
2026-07-08 13:54:11 +02:00
4634ad5a5a feat: use lnbits price aggregator to save loads of request (#4026)
Co-authored-by: alan <alan@lnbits.com>
2026-07-06 09:21:55 +02:00
dni ⚡andGitHub 648aaa17c5 CI: appimage upload-urls with gh upload command (#4036) 2026-07-03 12:35:14 +01:00
blackcoffeexbtandGitHub 7db5c986b3 NWC funding source: Reduce polling (#4008) 2026-07-02 13:00:49 +01:00
Vlad StanandGitHub 26c31b626d refactor: set extension field for payments (#4031) 2026-07-01 16:55:15 +03:00
53 changed files with 2085 additions and 670 deletions
+1 -14
View File
@@ -7,10 +7,6 @@ on:
description: 'The tag name for the release'
required: true
type: string
upload_url:
description: 'The upload URL for the release'
required: true
type: string
workflow_dispatch:
inputs:
@@ -18,10 +14,6 @@ on:
description: 'The tag name for the release'
required: true
type: string
upload_url:
description: 'The upload URL for the release'
required: true
type: string
jobs:
build-linux-package:
@@ -113,11 +105,6 @@ jobs:
shell: bash
- name: Upload Linux Release Asset
uses: actions/upload-release-asset@v1
with:
upload_url: ${{ inputs.upload_url }}
asset_path: ${{ env.APPIMAGE_NAME }}
asset_name: ${{ env.APPIMAGE_NAME }}
asset_content_type: application/octet-stream
env:
GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
run: gh release upload "${{ inputs.tag_name }}" "${{ env.APPIMAGE_NAME }}" --clobber
-11
View File
@@ -12,8 +12,6 @@ jobs:
release:
runs-on: ubuntu-24.04
outputs:
upload_url: ${{ steps.get_upload_url.outputs.upload_url }}
steps:
- uses: actions/checkout@v4
- name: Create github pre-release
@@ -22,14 +20,6 @@ jobs:
tag: ${{ github.ref_name }}
run: |
gh release create "$tag" --prerelease --generate-notes --draft
- id: get_upload_url
name: Get upload url of Github release
env:
GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
tag: ${{ github.ref_name }}
run: |
upload_url=$(gh release view "$tag" --json uploadUrl -q ".uploadUrl")
echo "upload_url=$upload_url" >> "$GITHUB_OUTPUT"
docker:
if: github.repository == 'lnbits/lnbits'
@@ -74,4 +64,3 @@ jobs:
uses: ./.github/workflows/appimage.yml
with:
tag_name: ${{ github.ref_name }}
upload_url: ${{ needs.release.outputs.upload_url }}
-11
View File
@@ -13,8 +13,6 @@ jobs:
release:
runs-on: ubuntu-24.04
outputs:
upload_url: ${{ steps.get_upload_url.outputs.upload_url }}
steps:
- uses: actions/checkout@v4
- name: Create github release
@@ -23,14 +21,6 @@ jobs:
tag: ${{ github.ref_name }}
run: |
gh release create "$tag" --generate-notes --draft
- id: get_upload_url
name: Get upload url of Github release
env:
GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
tag: ${{ github.ref_name }}
run: |
upload_url=$(gh release view "$tag" --json uploadUrl -q ".uploadUrl")
echo "upload_url=$upload_url" >> "$GITHUB_OUTPUT"
docker:
if: github.repository == 'lnbits/lnbits'
@@ -85,4 +75,3 @@ jobs:
uses: ./.github/workflows/appimage.yml
with:
tag_name: ${{ github.ref_name }}
upload_url: ${{ needs.release.outputs.upload_url }}
+97 -45
View File
@@ -23,30 +23,35 @@ from lnbits.core.crud import (
get_installed_extensions,
update_installed_extension_state,
)
from lnbits.core.crud.audit import delete_expired_audit_entries
from lnbits.core.crud.extensions import create_installed_extension
from lnbits.core.helpers import migrate_extension_database
from lnbits.core.models.notifications import NotificationType
from lnbits.core.services.extensions import deactivate_extension, get_valid_extensions
from lnbits.core.services.notifications import enqueue_admin_notification
from lnbits.core.services.payments import check_pending_payments
from lnbits.core.services.funding_source import (
check_balance_delta_changed,
check_server_balance_against_node,
)
from lnbits.core.services.notifications import (
dispatch_payment_notification,
enqueue_admin_notification,
process_next_notification,
)
from lnbits.core.services.payments import (
check_pending_payments,
fundingsource_invoice_producer,
)
from lnbits.core.tasks import (
audit_queue,
collect_exchange_rates_data,
purge_audit_data,
run_by_the_minute_tasks,
wait_for_audit_data,
wait_for_paid_invoices,
wait_notification_messages,
notify_server_status,
process_next_audit_entry,
refresh_extension_cache,
)
from lnbits.exceptions import register_exception_handlers
from lnbits.helpers import version_parse
from lnbits.llms_txt import create_llms_txt_route
from lnbits.settings import settings
from lnbits.tasks import (
cancel_all_tasks,
create_permanent_task,
register_invoice_listener,
)
from lnbits.utils.cache import cache
from lnbits.utils.logger import (
configure_logger,
@@ -69,7 +74,7 @@ from .middleware import (
add_profiler_middleware,
add_ratelimit_middleware,
)
from .tasks import internal_invoice_listener, invoice_listener, run_interval
from .task_manager import task_manager
async def startup(app: FastAPI):
@@ -133,7 +138,7 @@ async def shutdown():
settings.lnbits_running = False
# shutdown event
cancel_all_tasks()
task_manager.cancel_all_tasks()
# wait a bit to allow them to finish, so that cleanup can run without problems
await asyncio.sleep(0.1)
@@ -381,7 +386,6 @@ def register_custom_extensions_path():
upgrades_dir = settings.lnbits_extensions_upgrade_path
shutil.rmtree(upgrades_dir, True)
Path(upgrades_dir).mkdir(parents=True, exist_ok=True)
sys.path.append(str(upgrades_dir))
if settings.has_default_extension_path:
return
@@ -436,10 +440,54 @@ def register_ext_tasks(ext: Extension) -> None:
def register_ext_routes(app: FastAPI, ext: Extension) -> None:
"""Register FastAPI routes for extension."""
ext_module = importlib.import_module(ext.module_name)
module_name = ext.module_name
# Clear all cached sub-modules so a fresh import picks up new files from ext_dir.
# A simple reload() would reuse cached sub-modules (e.g. views_api) and serve
# stale code even after the extension files have been replaced on disk.
stale = [
k for k in sys.modules if k == module_name or k.startswith(f"{module_name}.")
]
for k in stale:
del sys.modules[k]
if stale:
# Pydantic v1 keeps a global _FUNCS set of validator qualnames to detect
# duplicates. Clear the extension's entries so reimport doesn't raise
# "duplicate validator" errors for validators with the same qualname.
try:
import pydantic.class_validators as _pydantic_cv
_pydantic_cv._FUNCS = {
f for f in _pydantic_cv._FUNCS if not f.startswith(f"{module_name}.")
}
except (ImportError, AttributeError):
pass
ext_module = importlib.import_module(module_name)
ext_route = getattr(ext_module, f"{ext.code}_ext")
ext_redirects = (
getattr(ext_module, f"{ext.code}_redirect_paths")
if hasattr(ext_module, f"{ext.code}_redirect_paths")
else []
)
settings.activate_extension_paths(ext.code, ext_redirects)
# Remove existing routes for this extension before re-registering so that
# an upgraded extension replaces the old one at the same paths (no prefix).
ext_prefix = f"/{ext.code}"
app.router.routes = [
r
for r in app.router.routes
if not (
getattr(r, "path", "") == ext_prefix
or getattr(r, "path", "").startswith(f"{ext_prefix}/")
)
]
# Invalidate FastAPI's cached OpenAPI schema so the next /openapi.json
# request reflects the updated routes.
app.openapi_schema = None
if hasattr(ext_module, f"{ext.code}_static_files"):
ext_statics = getattr(ext_module, f"{ext.code}_static_files")
for s in ext_statics:
@@ -448,17 +496,8 @@ def register_ext_routes(app: FastAPI, ext: Extension) -> None:
)
app.mount(s["path"], StaticFiles(directory=static_dir), s["name"])
ext_redirects = (
getattr(ext_module, f"{ext.code}_redirect_paths")
if hasattr(ext_module, f"{ext.code}_redirect_paths")
else []
)
settings.activate_extension_paths(ext.code, ext.upgrade_hash, ext_redirects)
logger.trace(f"Adding route for extension {ext_module}.")
prefix = f"/upgrades/{ext.upgrade_hash}" if ext.upgrade_hash != "" else ""
app.include_router(router=ext_route, prefix=prefix)
app.include_router(router=ext_route)
async def check_and_register_extensions(app: FastAPI) -> None:
@@ -473,29 +512,42 @@ async def check_and_register_extensions(app: FastAPI) -> None:
def register_async_tasks() -> None:
create_permanent_task(wait_for_audit_data)
create_permanent_task(wait_notification_messages)
task_manager.init()
create_permanent_task(
run_interval(
settings.lnbits_funding_source_pending_interval_seconds,
check_pending_payments,
)
# listen to all incoming payments and dispatch payment notifications
# note: should be the first in task list for a bit quicker notifications
task_manager.register_invoice_listener(dispatch_payment_notification, "core")
# periodic tasks
task_manager.create_permanent_task(cache.invalidate_cache, interval=10)
task_manager.create_permanent_task(delete_expired_audit_entries, interval=60 * 60)
task_manager.create_permanent_task(
check_pending_payments,
interval=settings.lnbits_funding_source_pending_interval_seconds,
)
create_permanent_task(invoice_listener)
create_permanent_task(internal_invoice_listener)
create_permanent_task(cache.invalidate_forever)
task_manager.create_permanent_task(
collect_exchange_rates_data,
interval=max(60, settings.lnbits_exchange_history_refresh_interval_seconds),
)
task_manager.create_permanent_task(check_balance_delta_changed, interval=60)
task_manager.create_permanent_task(
check_server_balance_against_node,
interval=60 * settings.lnbits_watchdog_interval_minutes,
)
task_manager.create_permanent_task(
notify_server_status,
interval=60 * 60 * settings.lnbits_notification_server_status_hours,
)
task_manager.create_permanent_task(refresh_extension_cache, interval=60)
# core invoice listener
invoice_queue: asyncio.Queue = asyncio.Queue()
register_invoice_listener(invoice_queue, "core")
create_permanent_task(lambda: wait_for_paid_invoices(invoice_queue))
create_permanent_task(run_by_the_minute_tasks)
create_permanent_task(purge_audit_data)
create_permanent_task(collect_exchange_rates_data)
# permanent tasks run in a loop, will be restarted if they fail
task_manager.create_permanent_task(fundingsource_invoice_producer)
task_manager.create_permanent_task(process_next_notification)
task_manager.create_permanent_task(process_next_audit_entry)
# server logs for websocket
if settings.lnbits_admin_ui:
server_log_task = initialize_server_websocket_logger()
create_permanent_task(server_log_task)
task_manager.create_permanent_task(
server_log_task, name="server_websocket_logger"
)
-5
View File
@@ -1,6 +1,5 @@
import asyncio
import importlib
import sys
import time
from functools import wraps
from getpass import getpass
@@ -377,10 +376,6 @@ async def extensions_update( # noqa: C901
if not await _can_run_operation(url):
return
upgrades_dir = settings.lnbits_extensions_upgrade_path
Path(upgrades_dir).mkdir(parents=True, exist_ok=True)
sys.path.append(str(upgrades_dir))
if extension:
await update_extension(extension, repo_index, source_repo, url, admin_user)
return
+1
View File
@@ -290,6 +290,7 @@ async def create_payment(
webhook=data.webhook,
fee=-abs(data.fee),
tag=extra.get("tag", None),
extension=data.extension,
extra=extra,
labels=data.labels or [],
external_id=data.external_id,
+15 -10
View File
@@ -8,11 +8,18 @@ from lnbits.db import dict_to_model
from lnbits.settings import (
AdminSettings,
EditableSettings,
FundingSourcesSettings,
SettingsField,
SuperSettings,
settings,
)
RESET_PRESERVED_SETTINGS = (
"lnbits_webpush_pubkey",
"lnbits_webpush_privkey",
*FundingSourcesSettings.__fields__,
)
async def get_super_settings() -> SuperSettings | None:
data = await get_settings_by_tag("core")
@@ -69,16 +76,14 @@ async def delete_admin_settings(tag: str | None = "core") -> None:
async def reset_core_settings() -> None:
await db.execute(
"""
DELETE FROM system_settings WHERE tag = 'core'
AND id NOT IN (
'super_user',
'lnbits_webpush_pubkey',
'lnbits_webpush_privkey'
)
""",
)
core_settings = await get_settings_by_tag("core") or {}
super_user = await get_settings_field("super_user")
await delete_admin_settings()
if super_user:
await set_settings_field("super_user", super_user.value)
for field in RESET_PRESERVED_SETTINGS:
if field in core_settings:
await set_settings_field(field, core_settings[field])
async def create_admin_settings(super_user: str, new_settings: dict) -> SuperSettings:
+6 -14
View File
@@ -6,7 +6,6 @@ import json
import os
import shutil
import zipfile
from asyncio.tasks import create_task
from pathlib import Path
from typing import Any
@@ -21,6 +20,7 @@ from lnbits.helpers import (
version_parse,
)
from lnbits.settings import settings
from lnbits.task_manager import task_manager
from lnbits.utils.cache import cache
@@ -147,21 +147,13 @@ class Extension(BaseModel):
name: str | None = None
short_description: str | None = None
tile: str | None = None
upgrade_hash: str | None = ""
@property
def module_name(self) -> str:
if self.is_upgrade_extension:
return f"{self.code}-{self.upgrade_hash}"
if settings.has_default_extension_path:
return f"lnbits.extensions.{self.code}"
return self.code
@property
def is_upgrade_extension(self) -> bool:
return self.upgrade_hash != ""
@classmethod
def from_installable_ext(cls, ext_info: InstallableExtension) -> Extension:
return Extension(
@@ -170,7 +162,6 @@ class Extension(BaseModel):
name=ext_info.name,
short_description=ext_info.short_description,
tile=ext_info.icon,
upgrade_hash=ext_info.hash if ext_info.ext_upgrade_dir.is_dir() else "",
)
@@ -375,9 +366,6 @@ class InstallableExtension(BaseModel):
@property
def module_name(self) -> str:
if self.ext_upgrade_dir.is_dir():
return f"{self.id}-{self.hash}"
if settings.has_default_extension_path:
return f"lnbits.extensions.{self.id}"
return self.id
@@ -468,6 +456,7 @@ class InstallableExtension(BaseModel):
shutil.rmtree(self.ext_dir, True)
shutil.copytree(Path(self.ext_upgrade_dir), Path(self.ext_dir))
shutil.rmtree(self.ext_upgrade_dir, True)
logger.info(f"Extension {self.name} ({self.installed_version}) extracted.")
def clean_extension_files(self):
@@ -642,7 +631,10 @@ class InstallableExtension(BaseModel):
if cache_value.older_than(10 * 60) or post_refresh_cache:
# refresh cache in background if older than 10 minutes or requested
create_task(cls._refresh_installable_extensions_cache())
task_manager.create_task(
cls._refresh_installable_extensions_cache(),
"refresh_installable_extensions_cache",
)
extension_list = cache_value.value # type: ignore
return extension_list
+2
View File
@@ -54,6 +54,7 @@ class CreatePayment(BaseModel):
amount_msat: int
memo: str
extra: dict | None = {}
extension: str | None = None
preimage: str | None = None
expiry: datetime | None = None
webhook: str | None = None
@@ -259,6 +260,7 @@ class CreateInvoice(BaseModel):
)
expiry: int | None = None
extra: dict | None = None
extension: str | None = None
webhook: str | None = None
bolt11: str | None = None
lnurl_withdraw: LnurlWithdrawResponse | None = None
+15 -19
View File
@@ -56,14 +56,10 @@ async def install_extension(
else:
await update_installed_extension(ext_info)
extension = Extension.from_installable_ext(ext_info)
if extension.is_upgrade_extension:
# call stop while the old routes are still active
if installed_ext:
await stop_extension_background_work(ext_info.id)
await start_extension_background_work(ext_info.id)
return extension
return Extension.from_installable_ext(ext_info)
async def check_extensions_limit(installed_ext: InstallableExtension | None = None):
@@ -103,16 +99,16 @@ async def stop_extension_background_work(ext_id: str) -> bool:
Stop background work for extension (like asyncio.Tasks, WebSockets, etc).
Extension must expose a `myextension_stop()` function if it is starting tasks.
"""
upgrade_hash = settings.extension_upgrade_hash(ext_id)
ext = Extension(code=ext_id, is_valid=True, upgrade_hash=upgrade_hash)
ext = Extension(code=ext_id, is_valid=True)
module_name = ext.module_name
try:
logger.info(f"Stopping background work for extension '{ext.module_name}'.")
old_module = importlib.import_module(ext.module_name)
logger.info(f"Stopping background work for extension '{module_name}'.")
old_module = importlib.import_module(module_name)
stop_fn_name = f"{ext_id}_stop"
if not hasattr(old_module, stop_fn_name):
raise ValueError(f"No stop function found for '{ext.module_name}'.")
raise ValueError(f"No stop function found for '{module_name}'.")
stop_fn = getattr(old_module, stop_fn_name)
if stop_fn:
@@ -120,9 +116,9 @@ async def stop_extension_background_work(ext_id: str) -> bool:
await stop_fn()
else:
stop_fn()
logger.info(f"Stopped background work for extension '{ext.module_name}'.")
logger.info(f"Stopped background work for extension '{module_name}'.")
except Exception as ex:
logger.warning(f"Failed to stop background work for '{ext.module_name}'.")
logger.warning(f"Failed to stop background work for '{module_name}'.")
logger.warning(ex)
return False
@@ -135,12 +131,12 @@ async def start_extension_background_work(ext_id: str) -> bool:
Extension CAN expose a `myextension_start()` function if it is starting tasks.
Extension MUST expose a `myextension_stop()` in that case.
"""
upgrade_hash = settings.extension_upgrade_hash(ext_id)
ext = Extension(code=ext_id, is_valid=True, upgrade_hash=upgrade_hash)
ext = Extension(code=ext_id, is_valid=True)
module_name = ext.module_name
try:
logger.info(f"Starting background work for extension '{ext.module_name}'.")
new_module = importlib.import_module(ext.module_name)
logger.info(f"Starting background work for extension '{module_name}'.")
new_module = importlib.import_module(module_name)
start_fn_name = f"{ext_id}_start"
# start function is optional, return False if not found
@@ -153,10 +149,10 @@ async def start_extension_background_work(ext_id: str) -> bool:
await start_fn()
else:
start_fn()
logger.info(f"Started background work for extension '{ext.module_name}'.")
logger.info(f"Started background work for extension '{module_name}'.")
return True
except Exception as ex:
logger.warning(f"Failed to start background work for '{ext.module_name}'.")
logger.warning(f"Failed to start background work for '{module_name}'.")
logger.warning(ex)
return False
+2 -5
View File
@@ -20,6 +20,7 @@ from lnbits.fiat.base import (
FiatPaymentSuccessStatus,
)
from lnbits.settings import settings
from lnbits.task_manager import task_manager
async def handle_fiat_payment_confirmation(
@@ -60,11 +61,7 @@ async def check_fiat_status(payment: Payment) -> FiatPaymentStatus:
payment.status = PaymentState.SUCCESS.value
await update_payment(payment)
await handle_fiat_payment_confirmation(payment)
# notify receivers asynchronously
from lnbits.tasks import internal_invoice_queue
await internal_invoice_queue.put(payment.checking_id)
task_manager.internal_invoice_queue.put_nowait(payment)
return fiat_status
+2
View File
@@ -66,6 +66,8 @@ async def check_server_balance_against_node():
async def check_balance_delta_changed():
if settings.notification_balance_delta_threshold_sats <= 0:
return
status = await get_balance_delta()
if settings.latest_balance_delta_sats is None:
settings.latest_balance_delta_sats = status.delta_sats
+9
View File
@@ -237,6 +237,15 @@ async def send_email(
return False
async def dispatch_payment_notification(payment: Payment) -> None:
"""
This worker dispatches the payment notifications.
"""
wallet = await get_wallet(payment.wallet_id)
if wallet:
await send_payment_notification(wallet, payment)
async def dispatch_webhook(payment: Payment):
"""
Dispatches the webhook to the webhook url.
+25 -11
View File
@@ -19,6 +19,7 @@ from lnbits.exceptions import InvoiceError, PaymentError, UnsupportedError
from lnbits.fiat import get_fiat_provider
from lnbits.helpers import check_callback_url
from lnbits.settings import settings
from lnbits.task_manager import task_manager
from lnbits.utils.crypto import fake_privkey, random_secret_and_hash, verify_preimage
from lnbits.utils.exchange_rates import fiat_amount_as_satoshis, satoshis_amount_as_fiat
from lnbits.wallets import fake_wallet, get_funding_source
@@ -214,6 +215,7 @@ async def create_wallet_invoice(wallet_id: str, data: CreateInvoice) -> Payment:
unhashed_description=unhashed_description,
expiry=data.expiry,
extra=data.extra,
extension=data.extension,
webhook=data.webhook,
internal=data.internal,
payment_hash=data.payment_hash,
@@ -259,6 +261,7 @@ async def create_invoice(
webhook: str | None = None,
internal: bool | None = False,
payment_hash: str | None = None,
extension: str | None = None,
labels: list[str] | None = None,
external_id: str | None = None,
conn: Connection | None = None,
@@ -342,6 +345,7 @@ async def create_invoice(
expiry=invoice.expiry_date,
memo=memo,
extra=extra,
extension=extension,
webhook=webhook,
fee=invoice_response.fee_msat or 0,
labels=labels,
@@ -513,9 +517,7 @@ async def update_wallet_balance(
)
payment.status = PaymentState.SUCCESS
await update_payment(payment, conn=conn)
from lnbits.tasks import internal_invoice_queue_put
await internal_invoice_queue_put(payment.checking_id)
task_manager.internal_invoice_queue.put_nowait(payment)
async def check_wallet_limits(
@@ -786,10 +788,8 @@ async def _pay_internal_invoice(
) # notify the receiver
# notify receiver asynchronously (extension listeners)
from lnbits.tasks import internal_invoice_queue
logger.debug(f"enqueuing internal invoice {internal_payment.checking_id}")
await internal_invoice_queue.put(internal_payment.checking_id)
task_manager.internal_invoice_queue.put_nowait(internal_payment)
return payment
@@ -826,16 +826,15 @@ async def _pay_external_invoice(
fee_reserve_msat = fee_reserve(amount_msat, internal=False)
from lnbits.tasks import create_task
task = create_task(
_fundingsource_pay_invoice(checking_id, payment.bolt11, fee_reserve_msat)
task = task_manager.create_task(
_fundingsource_pay_invoice(checking_id, payment.bolt11, fee_reserve_msat),
f"fundingsource_pay_invoice_{checking_id}",
)
# make sure a hold invoice or deferred payment is not blocking the server
wait_time = max(1, settings.lnbits_funding_source_pay_invoice_wait_seconds)
try:
payment_response = await asyncio.wait_for(task, timeout=wait_time)
payment_response = await asyncio.wait_for(task.task, timeout=wait_time)
except asyncio.TimeoutError:
# return pending payment on timeout
logger.debug(
@@ -1105,3 +1104,18 @@ async def update_invoice_from_paid_invoices_stream(checking_id: str) -> Payment
payment = await update_payment(payment)
return payment
async def fundingsource_invoice_producer() -> None:
"""
will collect all invoices that come directly from the backend wallet.
Called registered in the app startup sequence and run by taskmanager.
"""
funding_source = get_funding_source()
async for checking_id in funding_source.paid_invoices_stream():
logger.info(f"got a payment notification {checking_id}")
payment = await update_invoice_from_paid_invoices_stream(checking_id)
if payment:
logger.success(f"fundingsource invoice {checking_id} settled")
task_manager.invoice_queue.put_nowait(payment)
+46 -123
View File
@@ -2,77 +2,43 @@ import asyncio
from loguru import logger
from lnbits.core.crud import (
create_audit_entry,
get_wallet,
)
from lnbits.core.crud.audit import delete_expired_audit_entries
from lnbits.core.crud import create_audit_entry
from lnbits.core.crud.payments import get_payments_status_count
from lnbits.core.crud.users import get_accounts
from lnbits.core.crud.wallets import get_wallets_count
from lnbits.core.models.audit import AuditEntry
from lnbits.core.models.extensions import InstallableExtension
from lnbits.core.models.notifications import NotificationType
from lnbits.core.services.funding_source import (
check_balance_delta_changed,
check_server_balance_against_node,
get_balance_delta,
)
from lnbits.core.services.funding_source import get_balance_delta
from lnbits.core.services.notifications import (
enqueue_admin_notification,
process_next_notification,
send_payment_notification,
)
from lnbits.db import Filters
from lnbits.settings import settings
from lnbits.utils.cache import cache
from lnbits.utils.exchange_rates import btc_rates
from lnbits.utils.exchange_rates import btc_price_from_aggregator, btc_rates
audit_queue: asyncio.Queue[AuditEntry] = asyncio.Queue()
async def run_by_the_minute_tasks() -> None:
minute_counter = 0
while settings.lnbits_running:
status_minutes = settings.lnbits_notification_server_status_hours * 60
if settings.notification_balance_delta_threshold_sats > 0:
try:
# runs by default every minute, the delta should not change that often
await check_balance_delta_changed()
except Exception as ex:
logger.error(ex)
if minute_counter % settings.lnbits_watchdog_interval_minutes == 0:
try:
await check_server_balance_against_node()
except Exception as ex:
logger.error(ex)
if minute_counter % status_minutes == 0:
try:
await _notify_server_status()
except Exception as ex:
logger.error(ex)
if minute_counter % 60 == 0:
try:
# initialize the list of all extensions
await InstallableExtension.get_installable_extensions(
post_refresh_cache=True
)
except Exception as ex:
logger.error(ex)
minute_counter += 1
await asyncio.sleep(60)
async def process_next_audit_entry() -> None:
"""
Waits for audit entries to be pushed to the queue.
Then it inserts the entries into the DB.
"""
data = await audit_queue.get()
await create_audit_entry(data)
async def _notify_server_status() -> None:
async def refresh_extension_cache() -> None:
# only refreshes every 10 minutes
await InstallableExtension.get_installable_extensions()
async def notify_server_status() -> None:
accounts = await get_accounts(filters=Filters(limit=0))
wallets_count = await get_wallets_count()
payments = await get_payments_status_count()
status = await get_balance_delta()
values = {
"up_time": settings.lnbits_server_up_time,
@@ -89,81 +55,38 @@ async def _notify_server_status() -> None:
enqueue_admin_notification(NotificationType.server_status, values)
async def wait_for_paid_invoices(invoice_paid_queue: asyncio.Queue) -> None:
"""
This worker dispatches events to all extensions and dispatches webhooks.
"""
while settings.lnbits_running:
payment = await invoice_paid_queue.get()
logger.trace("received invoice paid event")
# payment notification
wallet = await get_wallet(payment.wallet_id)
if wallet:
await send_payment_notification(wallet, payment)
async def wait_for_audit_data() -> None:
"""
Waits for audit entries to be pushed to the queue.
Then it inserts the entries into the DB.
"""
while settings.lnbits_running:
data = await audit_queue.get()
try:
await create_audit_entry(data)
except Exception as ex:
logger.warning(ex)
await asyncio.sleep(3)
async def wait_notification_messages() -> None:
while settings.lnbits_running:
try:
await process_next_notification()
except Exception as ex:
logger.warning("Payment notification error", ex)
await asyncio.sleep(3)
async def purge_audit_data() -> None:
"""
Remove audit entries which have passed their retention period.
"""
while settings.lnbits_running:
try:
await delete_expired_audit_entries()
except Exception as ex:
logger.warning(ex)
# clean every hour
await asyncio.sleep(60 * 60)
async def collect_exchange_rates_data() -> None:
"""
Collect exchange rates data. Used for monitoring only.
"""
while settings.lnbits_running:
currency = settings.lnbits_default_accounting_currency or "USD"
max_history_size = settings.lnbits_exchange_history_size
sleep_time = settings.lnbits_exchange_history_refresh_interval_seconds
if sleep_time > 0:
try:
rates = await btc_rates(currency)
if rates:
rates_values = [r[1] for r in rates]
lnbits_rate = sum(rates_values) / len(rates_values)
rates.append(("LNbits", lnbits_rate))
cache.set(
f"btc-price-{currency}",
lnbits_rate,
expiry=settings.lnbits_exchange_rate_cache_seconds,
)
settings.append_exchange_rate_datapoint(dict(rates), max_history_size)
except Exception as ex:
logger.warning(ex)
currency = settings.lnbits_default_accounting_currency or "USD"
max_history_size = settings.lnbits_exchange_history_size
try:
if (
settings.lnbits_price_aggregator_enabled
and settings.lnbits_price_aggregator_url
):
price = await btc_price_from_aggregator(currency)
if price:
cache.set(
f"btc-price-{currency}",
price,
expiry=settings.lnbits_exchange_rate_cache_seconds,
)
settings.append_exchange_rate_datapoint(
{"Aggregator": price}, max_history_size
)
else:
sleep_time = 60
await asyncio.sleep(sleep_time)
rates = await btc_rates(currency)
if rates:
rates_values = [r[1] for r in rates]
lnbits_rate = sum(rates_values) / len(rates_values)
rates.append(("LNbits", lnbits_rate))
cache.set(
f"btc-price-{currency}",
lnbits_rate,
expiry=settings.lnbits_exchange_rate_cache_seconds,
)
settings.append_exchange_rate_datapoint(dict(rates), max_history_size)
except Exception as ex:
logger.warning(ex)
+4 -5
View File
@@ -20,7 +20,7 @@ from lnbits.core.services.settings import dict_to_settings
from lnbits.decorators import check_admin, check_super_user
from lnbits.server import server_restart
from lnbits.settings import AdminSettings, Settings, UpdateSettings, settings
from lnbits.tasks import invoice_listeners
from lnbits.task_manager import PublicTask, task_manager
from .. import core_app_extra
from ..crud import get_admin_settings, reset_core_settings, update_admin_settings
@@ -44,11 +44,10 @@ async def api_auditor():
name="Monitor",
description="show the current listeners and other monitoring data",
dependencies=[Depends(check_admin)],
response_model=list[PublicTask],
)
async def api_monitor():
return {
"invoice_listeners": list(invoice_listeners.keys()),
}
async def api_monitor() -> list[PublicTask]:
return task_manager.get_public_tasks()
@admin_router.get(
+1 -6
View File
@@ -310,12 +310,7 @@ def get_api_routes(routes: list) -> dict[str, str]:
def path_segments(path: str) -> list[str]:
path = path.strip("/")
segments = path.split("/")
if len(segments) < 2:
return segments
if segments[0] == "upgrades":
return segments[2:]
return segments[0:]
return path.split("/")
def normalize_path(path: str | None) -> str:
-8
View File
@@ -51,14 +51,6 @@ class InstalledExtensionMiddleware:
await self.app(scope, receive, send)
return
# re-route all trafic if the extension has been upgraded
if top_path in settings.lnbits_upgraded_extensions:
upgrade_path = (
f"""{settings.lnbits_upgraded_extensions[top_path]}/{top_path}"""
)
tail = "/".join(rest)
scope["path"] = f"/upgrades/{upgrade_path}/{tail}"
await self.app(scope, receive, send)
def _response_by_accepted_type(
+7 -14
View File
@@ -166,8 +166,6 @@ class ExchangeRateProvider(BaseModel):
class InstalledExtensionsSettings(LNbitsSettings):
# installed extensions that have been deactivated
lnbits_deactivated_extensions: set[str] = Field(default=set())
# upgraded extensions that require API redirects
lnbits_upgraded_extensions: dict[str, str] = Field(default={})
# list of redirects that extensions want to perform
lnbits_extensions_redirects: list[RedirectPath] = Field(default=[])
@@ -190,18 +188,10 @@ class InstalledExtensionsSettings(LNbitsSettings):
def activate_extension_paths(
self,
ext_id: str,
upgrade_hash: str | None = None,
ext_redirects: list[dict] | None = None,
):
self.lnbits_deactivated_extensions.discard(ext_id)
"""
Update the list of upgraded extensions. The middleware will perform
redirects based on this
"""
if upgrade_hash:
self.lnbits_upgraded_extensions[ext_id] = upgrade_hash
if ext_redirects:
self._activate_extension_redirects(ext_id, ext_redirects)
@@ -211,9 +201,6 @@ class InstalledExtensionsSettings(LNbitsSettings):
self.lnbits_deactivated_extensions.add(ext_id)
self._remove_extension_redirects(ext_id)
def extension_upgrade_hash(self, ext_id: str) -> str:
return settings.lnbits_upgraded_extensions.get(ext_id, "")
def _activate_extension_redirects(self, ext_id: str, ext_redirects: list[dict]):
ext_redirect_paths = [
RedirectPath(**{"ext_id": ext_id, **er}) for er in ext_redirects
@@ -362,6 +349,8 @@ class ExchangeProvidersSettings(LNbitsSettings):
lnbits_exchange_rate_cache_seconds: int = Field(default=60, ge=0)
lnbits_exchange_history_size: int = Field(default=60, ge=0)
lnbits_exchange_history_refresh_interval_seconds: int = Field(default=300, ge=0)
lnbits_price_aggregator_enabled: bool = Field(default=True)
lnbits_price_aggregator_url: str = Field(default="https://price.lnbits.com")
lnbits_exchange_rate_providers: list[ExchangeRateProvider] = Field(
default=[
@@ -587,6 +576,7 @@ class PhoenixdFundingSource(LNbitsSettings):
phoenixd_api_password: str | None = Field(default=None)
phoenixd_data_dir: str | None = Field(default=None)
phoenixd_mnemonic: str | None = Field(default=None)
phoenixd_mnemonic_backup_confirmed: bool = Field(default=False)
class AlbyFundingSource(LNbitsSettings):
@@ -611,6 +601,7 @@ class SparkL2FundingSource(LNbitsSettings):
spark_l2_external_endpoint: str | None = Field(default="http://localhost:8765")
spark_l2_external_api_key: str | None = Field(default=None)
spark_l2_mnemonic: str | None = Field(default=None)
spark_l2_mnemonic_backup_confirmed: bool = Field(default=False)
spark_l2_pay_wait_ms: int = Field(default=4000, ge=0)
spark_l2_pay_poll_ms: int = Field(default=500, ge=0)
spark_l2_stream_keepalive_ms: int = Field(default=15000, ge=0)
@@ -648,6 +639,7 @@ class BoltzFundingSource(LNbitsSettings):
boltz_client_password: str = Field(default="")
boltz_client_cert: str | None = Field(default=None)
boltz_mnemonic: str | None = Field(default=None)
boltz_mnemonic_backup_confirmed: bool = Field(default=False)
class StrikeFundingSource(LNbitsSettings):
@@ -1072,11 +1064,12 @@ class EnvSettings(LNbitsSettings):
log_rotation: str = Field(default="100 MB")
log_retention: str = Field(default="3 months")
first_install_token: str | None = Field(default=None)
cleanup_wallets_days: int = Field(default=90, ge=0)
funding_source_max_retries: int = Field(default=4, ge=0)
lnbits_max_users: int = Field(default=0, ge=0)
lnbits_max_extensions: int = Field(default=0, ge=0)
task_heart_beat_verbose: bool = Field(default=False)
task_heart_beat_interval: int = Field(default=30)
@property
def has_default_extension_path(self) -> bool:
File diff suppressed because one or more lines are too long
+1 -1
View File
File diff suppressed because one or more lines are too long
@@ -62,12 +62,6 @@ window.app.component('lnbits-admin-exchange-providers', {
mounted() {
this.getExchangeRateHistory()
},
created() {
const hash = window.location.hash.replace('#', '')
if (hash === 'exchange_providers') {
this.showExchangeProvidersTab(hash)
}
},
methods: {
getDefaultSetting(fieldName) {
LNbits.api.getDefaultSetting(fieldName).then(response => {
@@ -127,18 +121,21 @@ window.app.component('lnbits-admin-exchange-providers', {
this.exchangeData.showTickerConversion = true
},
initExchangeChart(data) {
if (this.exchangeRatesChart) {
this.exchangeRatesChart.destroy()
this.exchangeRatesChart = null
}
const xValues = data.map(d =>
this.utils.formatTimestamp(d.timestamp, 'HH:mm')
)
const exchanges = [
...this.formData.lnbits_exchange_rate_providers,
{name: 'LNbits'}
]
const exchanges = this.formData.lnbits_price_aggregator_enabled
? [{name: 'Aggregator'}]
: [...this.formData.lnbits_exchange_rate_providers, {name: 'LNbits'}]
const datasets = exchanges.map(exchange => ({
label: exchange.name,
data: data.map(d => d.rates[exchange.name]),
pointStyle: true,
borderWidth: exchange.name === 'LNbits' ? 4 : 1,
borderWidth: exchange.name === 'LNbits' ? 4 : 2,
tension: 0.4
}))
this.exchangeRatesChart = new Chart(
@@ -148,7 +145,11 @@ window.app.component('lnbits-admin-exchange-providers', {
options: {
plugins: {
legend: {
display: false
display: true
},
title: {
display: true,
text: 'Bitcoin Price History'
}
}
},
@@ -0,0 +1,151 @@
window.app.component('lnbits-admin-funding-seed-backup', {
props: ['active', 'is-super-user', 'form-data', 'settings'],
template: '#lnbits-admin-funding-seed-backup',
data() {
return {
dialog: {
show: false,
step: 1,
seed: '',
visible: false,
challenge: [],
answers: {},
error: '',
confirmField: ''
}
}
},
watch: {
active(isActive) {
if (isActive) {
this.openIfRequired()
}
},
'formData.lnbits_backend_wallet_class'(walletClass, previousWalletClass) {
const source = this.seedBackupSource(walletClass)
if (previousWalletClass && source && this.formData[source.seedField]) {
this.formData[source.confirmField] = false
}
this.openIfRequired()
},
'formData.boltz_mnemonic'() {
this.formData.boltz_mnemonic_backup_confirmed =
this.formData.boltz_mnemonic === this.settings.boltz_mnemonic
? this.settings.boltz_mnemonic_backup_confirmed
: false
this.openIfRequired()
},
'formData.phoenixd_mnemonic'() {
this.formData.phoenixd_mnemonic_backup_confirmed =
this.formData.phoenixd_mnemonic === this.settings.phoenixd_mnemonic
? this.settings.phoenixd_mnemonic_backup_confirmed
: false
this.openIfRequired()
},
'formData.spark_l2_mnemonic'() {
this.formData.spark_l2_mnemonic_backup_confirmed =
this.formData.spark_l2_mnemonic === this.settings.spark_l2_mnemonic
? this.settings.spark_l2_mnemonic_backup_confirmed
: false
this.openIfRequired()
}
},
computed: {
seedWords() {
return this.dialog.seed
.split(/\s+/)
.filter(Boolean)
.map((word, index) => ({index, word}))
}
},
created() {
this.openIfRequired()
},
methods: {
seedBackupSource(walletClass = this.formData.lnbits_backend_wallet_class) {
if (walletClass === 'BoltzWallet') {
return {
seedField: 'boltz_mnemonic',
confirmField: 'boltz_mnemonic_backup_confirmed'
}
}
if (walletClass === 'PhoenixdWallet') {
return {
seedField: 'phoenixd_mnemonic',
confirmField: 'phoenixd_mnemonic_backup_confirmed'
}
}
if (walletClass === 'SparkL2Wallet') {
return {
seedField: 'spark_l2_mnemonic',
confirmField: 'spark_l2_mnemonic_backup_confirmed'
}
}
},
openIfRequired() {
if (!this.active || !this.isSuperUser) return
const source = this.seedBackupSource()
if (!source) return
const seed = (this.formData[source.seedField] || '').trim()
const confirmed = this.formData[source.confirmField]
if (!seed || confirmed || this.dialog.show) return
this.dialog = {
show: true,
step: 1,
seed,
visible: false,
challenge: [],
answers: {},
error: '',
confirmField: source.confirmField
}
},
prepareChallenge() {
const words = this.dialog.seed.split(/\s+/).filter(Boolean)
const count = Math.min(4, words.length)
const indexes = _.shuffle([...Array(words.length).keys()]).slice(0, count)
this.dialog.challenge = indexes
.sort((a, b) => a - b)
.map(index => ({index, word: words[index]}))
this.dialog.answers = {}
this.dialog.error = ''
this.dialog.step = 2
},
submitChallenge() {
const isValid = this.dialog.challenge.every(({index, word}) => {
const answer = this.dialog.answers[index] || ''
return answer.trim().toLowerCase() === word.toLowerCase()
})
if (!isValid) {
this.dialog.error =
'One or more words are incorrect. Check your backup and try again.'
return
}
const field = this.dialog.confirmField
LNbits.api
.request(
'PATCH',
'/admin/api/v1/settings',
this.g.user.wallets[0].adminkey,
{
[field]: true
}
)
.then(() => {
this.formData[field] = true
this.settings[field] = true
this.dialog.show = false
Quasar.Notify.create({
type: 'positive',
message: 'Seed backup confirmed',
icon: 'check'
})
})
.catch(LNbits.utils.notifyApiError)
}
}
})
@@ -1,5 +1,5 @@
window.app.component('lnbits-admin-funding', {
props: ['is-super-user', 'form-data', 'settings'],
props: ['active', 'is-super-user', 'form-data', 'settings'],
template: '#lnbits-admin-funding',
data() {
return {
+33 -12
View File
@@ -414,12 +414,19 @@ window.PageWallet = {
switch (action.tag) {
case 'url':
Quasar.Notify.create({
message: `<a target="_blank" style="color: inherit" href="${action.url}">${action.url}</a>`,
message: action.url,
caption: action.description,
html: true,
html: false,
type: 'positive',
timeout: 0,
closeBtn: true
closeBtn: true,
actions: [
{
label: 'Open link',
color: 'white',
handler: () => this.utils.openUrlInNewTab(action.url)
}
]
})
break
case 'message':
@@ -431,15 +438,29 @@ window.PageWallet = {
})
break
case 'aes':
this.utils.decryptLnurlPayAES(action, response.data.preimage)
Quasar.Notify.create({
message: value,
caption: extra.success_action.description,
html: true,
type: 'positive',
timeout: 0,
closeBtn: true
})
this.utils
.decryptLnurlPayAES(action, response.data.preimage)
.then(value => {
Quasar.Notify.create({
message: value,
caption: action.description,
html: false,
type: 'positive',
timeout: 0,
closeBtn: true
})
})
.catch(error => {
Quasar.Notify.create({
message: action.description || 'Payment successful.',
caption: 'Could not decrypt success action.',
html: false,
type: 'warning',
timeout: 0,
closeBtn: true
})
})
break
}
}
})
+22
View File
@@ -365,5 +365,27 @@ window._lnbitsUtils = {
let decoder = new TextDecoder('utf-8')
return decoder.decode(valueb)
})
},
validateBrowsableUrl(urlString, allowLoopback = false) {
const url = new URL(urlString)
if (url.protocol !== 'http:' && url.protocol !== 'https:') {
throw new Error('Invalid protocol')
}
if (!allowLoopback) {
const host = url.hostname
if (
host === 'localhost' ||
host === '[::1]' ||
host === '::1' ||
host.startsWith('127.') ||
host.startsWith('::ffff:127.')
) {
throw new Error('Loopback addresses are not allowed')
}
}
},
openUrlInNewTab(urlString, allowLoopback = false) {
this.validateBrowsableUrl(urlString, allowLoopback)
window.open(urlString, '_blank', 'noopener,noreferrer')
}
}
+1
View File
@@ -58,6 +58,7 @@
"js/pages/users.js",
"js/pages/account.js",
"js/pages/admin.js",
"js/components/admin/lnbits-admin-funding-seed-backup.js",
"js/components/admin/lnbits-admin-funding.js",
"js/components/admin/lnbits-admin-funding-sources.js",
"js/components/admin/lnbits-admin-fiat-providers.js",
+200
View File
@@ -0,0 +1,200 @@
import asyncio
import traceback
import uuid
from collections.abc import Callable, Coroutine
from datetime import datetime, timezone
from loguru import logger
from pydantic import BaseModel
from lnbits.core.models import Payment
from lnbits.settings import settings
class PublicTask(BaseModel):
"""Public model used to expose task information via the API."""
name: str
created_at: datetime
class Task:
"""Model used on the backend to keep track of background tasks."""
coro: Coroutine
name: str
created_at: datetime
task: asyncio.Task
invoice_queue: asyncio.Queue[Payment] | None = None
def __init__(
self,
coro: Coroutine,
name: str | None = None,
invoice_queue: asyncio.Queue | None = None,
) -> None:
self.coro = coro
self.name = name or f"task_{uuid.uuid4()}"
self.created_at = datetime.now(timezone.utc)
self.task = asyncio.create_task(self.coro, name=self.name)
self.invoice_queue = invoice_queue
class TaskManager:
"""Singleton class to manage background tasks."""
tasks: list[Task] = []
invoice_queue: asyncio.Queue[Payment] = asyncio.Queue()
internal_invoice_queue: asyncio.Queue[Payment] = asyncio.Queue()
def init(self) -> None:
self.create_permanent_task(
func=self._heart_beat,
interval=settings.task_heart_beat_interval,
)
self.create_permanent_task(self._invoice_listener_consumer)
self.create_permanent_task(self._internal_invoice_listener_consumer)
def get_task(self, name: str) -> Task | None:
"""Get a running task by name."""
for task in self.tasks:
if task.name == name:
return task
return None
def get_public_tasks(self) -> list[PublicTask]:
"""Get a list of public tasks."""
return [PublicTask(name=t.name, created_at=t.created_at) for t in self.tasks]
def cancel_task(self, task: Task) -> None:
"""Cancel a running task."""
self.tasks.remove(task)
try:
task.task.cancel()
except Exception as exc:
logger.warning(f"error while cancelling task `{task.name}`: {exc!s}")
def cancel_all_tasks(self) -> None:
"""Cancel all running tasks."""
for task in list(self.tasks):
self.cancel_task(task)
def create_task(
self,
coro: Coroutine,
name: str | None = None,
invoice_queue: asyncio.Queue | None = None,
) -> Task:
"""Create a task. If a task with the same name exists, it will be cancelled."""
if name:
task = self.get_task(name)
if task:
self.cancel_task(task)
task = Task(coro=coro, name=name, invoice_queue=invoice_queue)
self.tasks.append(task)
return task
def create_permanent_task(
self,
func: Callable[[], Coroutine],
invoice_queue: asyncio.Queue | None = None,
name: str | None = None,
interval: int = 0,
) -> Task:
"""Create a task that runs forever and restarts on failure."""
async def wrapper():
while settings.lnbits_running:
await self._catch_everything_and_restart(func)
if interval > 0:
await asyncio.sleep(interval)
return self.create_task(
coro=wrapper(), name=name or func.__name__, invoice_queue=invoice_queue
)
def register_invoice_listener(
self,
func: Callable[[Payment], Coroutine],
name: str | None = None,
) -> Task:
"""
A method intended for extensions to call when they want to be notified about
incoming payments. Will call provided Coroutine with the updated payment.
"""
name = f"{name or uuid.uuid4()}_invoice_listener"
queue: asyncio.Queue[Payment] = asyncio.Queue()
return self.create_permanent_task(
self._invoice_listener_worker(func, queue),
name=name,
invoice_queue=queue,
)
async def _heart_beat(self) -> None:
"""A heartbeat that removes done tasks logs the number of tasks."""
for task in self.tasks:
state = task.task._state if task.task else "NOT RUNNING"
if settings.task_heart_beat_verbose:
logger.debug(
f"Task Manager: `{task.name}` state: `{state}` "
f"created: {task.created_at.strftime('%Y-%m-%d %H:%M:%S')}`"
)
if task.task and task.task.done():
logger.debug(f"Task Manager: task `{task.name}` is done.")
self.cancel_task(task)
listeners_count = sum(1 for task in self.tasks if task.invoice_queue)
logger.debug(
f"Task Manager: {len(self.tasks) - listeners_count} tasks "
f"and {listeners_count} invoice listeners."
)
async def _catch_everything_and_restart(
self,
func: Callable[[], Coroutine],
restart_interval: int = 5,
) -> None:
"""Catches all exceptions from a function and restarts it after 5 seconds."""
while settings.lnbits_running:
try:
return await func()
except asyncio.CancelledError:
raise # because we must pass this up
except Exception as exc:
if not settings.lnbits_running:
return
logger.error(f"exception in background task `{func.__name__}`:", exc)
logger.error(traceback.format_exc())
logger.info(
f"`{func.__name__}` restarts in {restart_interval} seconds."
)
await asyncio.sleep(restart_interval)
def _invoice_listener_worker(
self, func: Callable[[Payment], Coroutine], queue: asyncio.Queue[Payment]
) -> Callable:
async def wrapper() -> None:
payment: Payment = await queue.get()
await func(payment)
return wrapper
def _invoice_dispatcher(self, payment: Payment) -> None:
"""Dispatches a payment to all registered invoice listeners."""
for task in self.tasks:
if not task.invoice_queue:
continue
logger.debug(f"Enqueing payment to task {task.name}")
task.invoice_queue.put_nowait(payment)
async def _invoice_listener_consumer(self) -> None:
payment = await self.invoice_queue.get()
logger.info(f"got a payment notification {payment.checking_id}")
self._invoice_dispatcher(payment)
async def _internal_invoice_listener_consumer(self) -> None:
payment = await self.internal_invoice_queue.get()
logger.info(f"got an internal payment notification {payment.checking_id}")
self._invoice_dispatcher(payment)
task_manager = TaskManager()
+33 -120
View File
@@ -1,146 +1,83 @@
import asyncio
import traceback
import uuid
from collections.abc import Callable, Coroutine
from loguru import logger
from lnbits.core.models import Payment
from lnbits.core.services.payments import (
get_standalone_payment,
update_invoice_from_paid_invoices_stream,
)
from lnbits.core.services.payments import get_standalone_payment
from lnbits.settings import settings
from lnbits.wallets import get_funding_source
tasks: list[asyncio.Task] = []
unique_tasks: dict[str, asyncio.Task] = {}
from lnbits.task_manager import task_manager
# DEPRECATED: use task_manager.create_task instead.
def create_task(coro: Coroutine) -> asyncio.Task:
task = asyncio.create_task(coro)
tasks.append(task)
return task
logger.debug("DEPRECATED: use task_manager.create_task instead.")
return task_manager.create_task(coro).task
# DEPRECATED: use task_manager.create_task with `name` kwarg.
def create_unique_task(name: str, coro: Coroutine) -> asyncio.Task:
if unique_tasks.get(name):
logger.warning(f"task `{name}` already exists, cancelling it")
try:
unique_tasks[name].cancel()
except Exception as exc:
logger.warning(f"error while cancelling task `{name}`: {exc!s}")
task = asyncio.create_task(coro)
unique_tasks[name] = task
return task
logger.debug("DEPRECATED: use task_manager.create_task instead.")
return task_manager.create_task(coro, name=name).task
# DEPRECATED: use task_manager.create_permanent_task instead.
def create_permanent_task(func: Callable[[], Coroutine]) -> asyncio.Task:
return create_task(catch_everything_and_restart(func))
logger.debug("DEPRECATED: use task_manager.create_permanent_task instead.")
return task_manager.create_permanent_task(func).task
# DEPRECATED: use task_manager.create_permanent_task with `name` argument instead.
def create_permanent_unique_task(
name: str, coro: Callable[[], Coroutine]
) -> asyncio.Task:
return create_unique_task(name, catch_everything_and_restart(coro, name))
def cancel_all_tasks() -> None:
for task in tasks:
try:
task.cancel()
except Exception as exc:
logger.warning(f"error while cancelling task: {exc!s}")
for name, task in unique_tasks.items():
try:
task.cancel()
except Exception as exc:
logger.warning(f"error while cancelling task `{name}`: {exc!s}")
# DEPRECATED don't use this, use task_manager.create_permanent_task instead.
async def catch_everything_and_restart(
func: Callable[[], Coroutine],
name: str = "unnamed",
) -> Coroutine:
try:
return await func()
except asyncio.CancelledError:
raise # because we must pass this up
except Exception as exc:
logger.error(f"exception in background task `{name}`:", exc)
logger.error(traceback.format_exc())
logger.error("will restart the task in 5 seconds.")
await asyncio.sleep(5)
return await catch_everything_and_restart(func, name)
) -> None:
_ = name
return await task_manager._catch_everything_and_restart(func)
invoice_listeners: dict[str, asyncio.Queue] = {}
# TODO: name should not be optional
# some extensions still dont use a name, but they should
def register_invoice_listener(send_chan: asyncio.Queue, name: str | None = None):
"""
A method intended for extensions (and core/tasks.py) to call when they want to be
notified about new invoice payments incoming. Will emit all incoming payments.
DEPRECATED: use task_manager.register_invoice_listener instead,
which also allows to pass a callback instead of a queue.
This method will still work but it is not recommended for new code.
"""
if not name:
# fallback to a random name if extension didn't provide one
name = f"no_name_{str(uuid.uuid4())[:8]}"
logger.debug("DEPRECATED: use task_manager.register_invoice_listener instead.")
name = f"forward_{name or str(uuid.uuid4())[:8]}"
if invoice_listeners.get(name):
logger.warning(f"invoice listener `{name}` already exists, replacing it")
# here we just forwarding the payments to the provided queue
async def forward_queue(payment: Payment):
send_chan.put_nowait(payment)
logger.trace(f"registering invoice listener `{name}`")
invoice_listeners[name] = send_chan
internal_invoice_queue: asyncio.Queue = asyncio.Queue(0)
task_manager.register_invoice_listener(forward_queue, name=name)
async def internal_invoice_queue_put(checking_id: str) -> None:
"""
DEPRECATED: use task_manager.internal_invoice_queue instead,
A method to call when it wants to notify about an internal invoice payment.
"""
await internal_invoice_queue.put(checking_id)
async def internal_invoice_listener() -> None:
"""
internal_invoice_queue will be filled directly in core/services.py
after the payment was deemed to be settled internally.
Called by the app startup sequence.
"""
while settings.lnbits_running:
checking_id = await internal_invoice_queue.get()
logger.info(f"got an internal payment notification {checking_id}")
payment = await get_standalone_payment(checking_id, incoming=True)
if payment:
logger.success(f"internal invoice {checking_id} settled")
await invoice_callback_dispatcher(payment)
async def invoice_listener() -> None:
"""
invoice_listener will collect all invoices that come directly
from the backend wallet.
Called by the app startup sequence.
"""
funding_source = get_funding_source()
async for checking_id in funding_source.paid_invoices_stream():
logger.info(f"got a payment notification {checking_id}")
payment = await update_invoice_from_paid_invoices_stream(checking_id)
if payment:
logger.success(f"fundingsource invoice {checking_id} settled")
await invoice_callback_dispatcher(payment)
payment = await get_standalone_payment(checking_id, incoming=True)
if not payment:
logger.warning(f"internal_invoice_queue_put: payment {checking_id} not found")
return
await task_manager.internal_invoice_queue.put(payment)
# DEPRECATED use task_manager.register_invoice_listener(coro, name="myext")
def wait_for_paid_invoices(
invoice_listener_name: str,
func: Callable[[Payment], Coroutine],
) -> Callable[[], Coroutine]:
logger.debug("DEPRECATED: use task_manager.register_invoice_listener instead.")
async def wrapper() -> None:
invoice_queue: asyncio.Queue = asyncio.Queue()
@@ -150,27 +87,3 @@ def wait_for_paid_invoices(
await func(payment)
return wrapper
def run_interval(
interval_seconds: int,
func: Callable[[], Coroutine],
) -> Callable[[], Coroutine]:
"""Run a function at a specified interval in seconds, while the server is running"""
async def wrapper() -> None:
while settings.lnbits_running:
try:
await func()
except Exception as e:
logger.error(f"Error occurred in interval task: {e}")
logger.warning(traceback.format_exc())
await asyncio.sleep(interval_seconds)
return wrapper
async def invoice_callback_dispatcher(payment: Payment):
for name, send_chan in invoice_listeners.items():
logger.trace(f"invoice listeners: sending to `{name}`")
await send_chan.put(payment)
+2 -1
View File
@@ -1,4 +1,5 @@
{% include('components/admin/funding.vue') %} {%
{% include('components/admin/funding_seed_backup.vue') %} {%
include('components/admin/funding.vue') %} {%
include('components/admin/funding_sources.vue') %} {%
include('components/admin/fiat_providers.vue') %} {%
include('components/admin/exchange_providers.vue') %} {%
@@ -1,7 +1,46 @@
<template id="lnbits-admin-exchange-providers">
<h6 class="q-my-none q-mb-sm">
<span v-text="$t('exchange_providers')"></span>
</h6>
<h6 class="q-my-none q-mb-xs">LNbits Price Aggregator</h6>
<p class="q-mb-md text-caption text-grey">
A privacy-friendly, open-source Bitcoin price aggregator maintained by the
LNbits team. Aggregates prices from multiple exchanges and returns a median,
no API keys required.
<a href="https://price.lnbits.com" target="_blank" rel="noopener"
>price.lnbits.com</a
>
&mdash;
<a
href="https://github.com/lnbits/lnbits-price-aggregator"
target="_blank"
rel="noopener"
>GitHub</a
>
</p>
<div class="row q-mb-md items-start">
<div class="col-auto q-mr-md q-mt-sm">
<q-toggle
v-model="formData.lnbits_price_aggregator_enabled"
@update:model-value="formData.touch = null"
label="Use Price Aggregator"
>
</q-toggle>
</div>
<div class="col-12 col-md-7">
<q-input
filled
v-model="formData.lnbits_price_aggregator_url"
type="text"
label="Price Aggregator URL"
hint="Fetch BTC price from this aggregator instead of individual providers below."
:disable="!formData.lnbits_price_aggregator_enabled"
@update:model-value="formData.touch = null"
>
</q-input>
</div>
</div>
<q-separator class="q-my-md"></q-separator>
<h6 class="q-my-none q-mb-sm">Bitcoin Price History</h6>
<div class="row">
<div class="col-12 col-md-8">
@@ -53,6 +92,11 @@
</div>
</div>
<q-separator class="q-my-md"></q-separator>
<h6 class="q-my-none q-mb-sm">
<span v-text="$t('exchange_providers')"></span>
</h6>
<div class="row q-mt-md">
<div class="col-6">
<q-btn
@@ -60,6 +104,7 @@
label="Add Exchange Provider"
color="primary"
class="q-mb-md"
:disable="formData.lnbits_price_aggregator_enabled"
>
</q-btn>
</div>
@@ -70,12 +115,20 @@
:label="$t('reset_defaults')"
color="primary"
class="float-right"
:disable="formData.lnbits_price_aggregator_enabled"
>
</q-btn>
</div>
</div>
<div class="overflow-auto">
<div
class="overflow-auto"
:style="
formData.lnbits_price_aggregator_enabled
? 'opacity:0.4;pointer-events:none'
: ''
"
>
<q-table
row-key="name"
:rows="formData.lnbits_exchange_rate_providers"
@@ -301,5 +301,11 @@
</div>
</div>
</div>
<lnbits-admin-funding-seed-backup
:active="active"
:is-super-user="isSuperUser"
:form-data="formData"
:settings="settings"
></lnbits-admin-funding-seed-backup>
</q-card-section>
</template>
@@ -0,0 +1,136 @@
<template id="lnbits-admin-funding-seed-backup">
<q-dialog v-model="dialog.show">
<q-card style="width: 760px; max-width: 95vw; border-radius: 8px">
<q-card-section class="q-pb-md">
<div class="row q-col-gutter-sm">
<div class="col-6">
<q-chip
square
class="full-width"
icon="looks_one"
:color="dialog.step === 1 ? 'primary' : 'grey-9'"
text-color="white"
label="Backup"
></q-chip>
</div>
<div class="col-6">
<q-chip
square
class="full-width"
icon="looks_two"
:color="dialog.step === 2 ? 'primary' : 'grey-9'"
text-color="white"
label="Verify"
></q-chip>
</div>
</div>
</q-card-section>
<q-separator></q-separator>
<q-card-section v-if="dialog.step === 1">
<div class="row items-center justify-between q-mb-md">
<div>
<div
class="text-subtitle1"
v-text="`${seedWords.length}-word recovery phrase`"
></div>
<div
class="text-caption text-grey-5"
v-text="'Write these words down in order.'"
></div>
</div>
<q-btn
outline
no-caps
color="primary"
:icon="dialog.visible ? 'visibility_off' : 'visibility'"
:label="dialog.visible ? 'Hide words' : 'Show words'"
@click="dialog.visible = !dialog.visible"
></q-btn>
</div>
<div class="row q-col-gutter-sm">
<div
class="col-4 col-md-3"
v-for="word in seedWords"
:key="word.index"
>
<div
class="row items-center no-wrap rounded-borders"
style="
min-height: 42px;
border: 1px solid rgba(255, 255, 255, 0.14);
background: rgba(255, 255, 255, 0.035);
"
>
<div
class="text-caption text-grey-5 text-center"
style="
width: 42px;
border-right: 1px solid rgba(255, 255, 255, 0.1);
"
v-text="word.index + 1"
></div>
<div
class="text-body2 text-weight-medium q-px-sm"
style="min-width: 0; overflow-wrap: anywhere"
v-text="dialog.visible ? word.word : '••••••'"
></div>
</div>
</div>
</div>
<div class="row justify-end q-mt-lg">
<q-btn
color="primary"
no-caps
label="I have written it down"
@click="prepareChallenge"
></q-btn>
</div>
</q-card-section>
<q-card-section v-if="dialog.step === 2">
<div class="q-mb-md">
<div class="text-subtitle1" v-text="'Confirm your backup'"></div>
<div
class="text-caption text-grey-5"
v-text="
'Enter the requested words from your written recovery phrase.'
"
></div>
</div>
<div class="row q-col-gutter-md">
<div
class="col-12 col-sm-6"
v-for="word in dialog.challenge"
:key="word.index"
>
<q-input
v-model.trim="dialog.answers[word.index]"
filled
:label="`Word ${word.index + 1}`"
></q-input>
</div>
</div>
<div
class="text-negative q-mt-sm"
v-if="dialog.error"
v-text="dialog.error"
></div>
<div class="row justify-between q-mt-lg">
<q-btn flat no-caps label="Back" @click="dialog.step = 1"></q-btn>
<q-btn
color="primary"
icon="check"
no-caps
label="Confirm backup"
@click="submitChallenge"
></q-btn>
</div>
</q-card-section>
</q-card>
</q-dialog>
</template>
@@ -6,6 +6,9 @@
:content-inset-level="0.5"
>
<q-card-section>
<q-banner dense rounded class="bg-warning text-black q-mb-md">
These keys should be kept safe, sharing them could risk losing funds.
</q-banner>
<q-list>
<q-item dense class="q-pa-none">
<q-item-section>
+1
View File
@@ -199,6 +199,7 @@
>
<q-tab-panel name="funding">
<lnbits-admin-funding
:active="tab === 'funding'"
:is-super-user="isSuperUser"
:settings="settings"
:form-data="formData"
+6 -17
View File
@@ -1,13 +1,8 @@
from __future__ import annotations
import asyncio
from time import time
from typing import Any, NamedTuple
from loguru import logger
from lnbits.settings import settings
class Cached(NamedTuple):
value: Any
@@ -22,8 +17,7 @@ class Cache:
Small caching utility providing simple get/set interface (very much like redis)
"""
def __init__(self, interval: float = 10) -> None:
self.interval = interval
def __init__(self) -> None:
self._values: dict[Any, Cached] = {}
def value(self, key: str) -> Cached | None:
@@ -59,16 +53,11 @@ class Cache:
self.set(key, value, expiry=expiry)
return value
async def invalidate_forever(self):
while settings.lnbits_running:
try:
await asyncio.sleep(self.interval)
ts = time()
expired = [k for k, v in self._values.items() if v.expiry < ts]
for k in expired:
self._values.pop(k)
except Exception:
logger.error("Error invalidating cache")
async def invalidate_cache(self):
ts = time()
expired = [k for k, v in self._values.items() if v.expiry < ts]
for k in expired:
self._values.pop(k)
cache = Cache()
+25
View File
@@ -289,7 +289,32 @@ async def btc_rates(currency: str) -> list[tuple[str, float]]:
return apply_trimmed_mean_filter(all_rates)
async def btc_price_from_aggregator(currency: str) -> float | None:
url = settings.lnbits_price_aggregator_url.rstrip("/")
try:
headers = {"User-Agent": settings.user_agent}
async with httpx.AsyncClient(headers=headers) as client:
r = await client.get(f"{url}/rate/{currency.upper()}", timeout=3)
r.raise_for_status()
data = r.json()
median = data.get("rates", {}).get("median")
if median:
return float(median)
except Exception as e:
logger.warning(f"Failed to fetch price from aggregator {url}: {e}")
return None
async def btc_price(currency: str) -> float:
if (
settings.lnbits_price_aggregator_enabled
and settings.lnbits_price_aggregator_url
):
price = await btc_price_from_aggregator(currency)
if price:
return price
logger.warning("Price aggregator failed, falling back to exchange providers.")
rates = await btc_rates(currency)
if not rates:
logger.warning("Could not fetch any Bitcoin price.")
+4 -7
View File
@@ -41,19 +41,16 @@ def log_server_info():
def initialize_server_websocket_logger() -> Callable:
super_user_hash = sha256(settings.super_user.encode("utf-8")).hexdigest()
serverlog_queue: asyncio.Queue = asyncio.Queue()
async def update_websocket_serverlog():
while settings.lnbits_running:
msg = await serverlog_queue.get()
await websocket_updater(super_user_hash, msg)
logger.add(
lambda msg: serverlog_queue.put_nowait(msg),
format=Formatter().format,
)
async def update_websocket_serverlog():
msg = await serverlog_queue.get()
await websocket_updater(super_user_hash, msg)
return update_websocket_serverlog
+921 -138
View File
File diff suppressed because it is too large Load Diff
+1
View File
@@ -111,6 +111,7 @@
"js/pages/users.js",
"js/pages/account.js",
"js/pages/admin.js",
"js/components/admin/lnbits-admin-funding-seed-backup.js",
"js/components/admin/lnbits-admin-funding.js",
"js/components/admin/lnbits-admin-funding-sources.js",
"js/components/admin/lnbits-admin-fiat-providers.js",
+1 -1
View File
@@ -1,6 +1,6 @@
[project]
name = "lnbits"
version = "1.5.5"
version = "1.5.6"
requires-python = ">=3.10,<3.13"
description = "LNbits, free and open-source Lightning wallet and accounts system."
authors = [{ name = "Alan Bits", email = "alan@lnbits.com" }]
+29 -1
View File
@@ -3,6 +3,7 @@ from pathlib import Path
import pytest
from httpx import AsyncClient
from lnbits.core.crud.settings import get_settings_field, set_settings_field
from lnbits.server import server_restart
from lnbits.settings import Settings
@@ -81,7 +82,8 @@ async def test_admin_audit_monitor_and_test_email(
headers={"Authorization": f"Bearer {superuser_token}"},
)
assert monitor.status_code == 200
assert "invoice_listeners" in monitor.json()
task_names = [t["name"] for t in monitor.json()]
assert any("invoice_listener" in name for name in task_names)
test_email = await client.get(
"/admin/api/v1/testemail",
@@ -150,6 +152,15 @@ async def test_admin_partial_reset_restart_and_backup(
async def test_admin_delete_settings_requires_superuser(
client: AsyncClient, superuser_token: str
):
await set_settings_field("lnbits_site_title", "Reset me")
await set_settings_field("lnbits_backend_wallet_class", "BoltzWallet")
await set_settings_field("boltz_mnemonic", "keep boltz seed")
await set_settings_field("boltz_mnemonic_backup_confirmed", True)
await set_settings_field("phoenixd_mnemonic", "keep phoenixd seed")
await set_settings_field("phoenixd_mnemonic_backup_confirmed", True)
await set_settings_field("spark_l2_mnemonic", "keep spark seed")
await set_settings_field("spark_l2_mnemonic_backup_confirmed", True)
server_restart.clear()
response = await client.delete(
"/admin/api/v1/settings",
@@ -157,4 +168,21 @@ async def test_admin_delete_settings_requires_superuser(
)
assert response.status_code == 200
assert server_restart.is_set() is True
assert await get_settings_field("lnbits_site_title") is None
backend_wallet = await get_settings_field("lnbits_backend_wallet_class")
boltz_seed = await get_settings_field("boltz_mnemonic")
boltz_confirmed = await get_settings_field("boltz_mnemonic_backup_confirmed")
phoenixd_seed = await get_settings_field("phoenixd_mnemonic")
phoenixd_confirmed = await get_settings_field("phoenixd_mnemonic_backup_confirmed")
spark_l2_seed = await get_settings_field("spark_l2_mnemonic")
spark_l2_confirmed = await get_settings_field("spark_l2_mnemonic_backup_confirmed")
assert backend_wallet and backend_wallet.value == "BoltzWallet"
assert boltz_seed and boltz_seed.value == "keep boltz seed"
assert boltz_confirmed and boltz_confirmed.value is True
assert phoenixd_seed and phoenixd_seed.value == "keep phoenixd seed"
assert phoenixd_confirmed and phoenixd_confirmed.value is True
assert spark_l2_seed and spark_l2_seed.value == "keep spark seed"
assert spark_l2_confirmed and spark_l2_confirmed.value is True
server_restart.clear()
+9 -6
View File
@@ -13,10 +13,13 @@ from lnbits.core.services import (
fee_reserve_total,
get_balance_delta,
)
from lnbits.core.services.payments import pay_invoice, update_wallet_balance
from lnbits.core.services.payments import (
pay_invoice,
update_wallet_balance,
)
from lnbits.core.services.users import create_user_account
from lnbits.exceptions import PaymentError
from lnbits.tasks import create_task, wait_for_paid_invoices
from lnbits.task_manager import task_manager
from lnbits.wallets import get_funding_source
from ..helpers import is_fake, is_regtest
@@ -160,12 +163,11 @@ async def test_create_real_invoice(
assert not payment_status["paid"]
on_paid_mock = mocker.AsyncMock()
create_task(wait_for_paid_invoices("test_create_invoice", on_paid_mock)())
task_manager.register_invoice_listener(on_paid_mock, "test_create_invoice")
pay_real_invoice(invoice["bolt11"])
await asyncio.sleep(1)
assert on_paid_mock.call_count == 1
payment = on_paid_mock.call_args_list[0][0][0]
@@ -393,12 +395,11 @@ async def test_receive_real_invoice_set_pending_and_check_state(
assert not payment_status["paid"]
on_paid_mock = mocker.AsyncMock()
create_task(wait_for_paid_invoices("test_create_invoice", on_paid_mock)())
task_manager.register_invoice_listener(on_paid_mock, "test_create_invoice")
pay_real_invoice(invoice["bolt11"])
await asyncio.sleep(1)
assert on_paid_mock.call_count == 1
payment = on_paid_mock.call_args_list[0][0][0]
@@ -412,6 +413,8 @@ async def test_receive_real_invoice_set_pending_and_check_state(
payment_status = response.json()
assert payment_status["paid"]
assert payment
# set the incoming invoice to pending
payment.status = PaymentState.PENDING
await update_payment(payment)
+24 -14
View File
@@ -5,6 +5,7 @@ import pytest
from pytest_mock.plugin import MockerFixture
from lnbits.settings import Settings
from lnbits.task_manager import task_manager
from lnbits.utils.cache import Cache, Cached
key = "foo"
@@ -13,11 +14,10 @@ value = "bar"
@pytest.fixture
async def cache():
cache = Cache(interval=0.1)
task = asyncio.create_task(cache.invalidate_forever())
cache = Cache()
task = task_manager.create_permanent_task(cache.invalidate_cache, interval=1)
yield cache
task.cancel()
task_manager.cancel_task(task)
@pytest.mark.anyio
@@ -31,13 +31,13 @@ async def test_cache_get_set(cache):
@pytest.mark.anyio
async def test_cache_expiry(cache):
# gets expired by `get` call
cache.set(key, value, expiry=0.01)
await asyncio.sleep(0.02)
cache.set(key, value, expiry=1)
await asyncio.sleep(2)
assert not cache.get(key)
# gets expired by invalidation task
cache.set(key, value, expiry=0.1)
await asyncio.sleep(0.2)
cache.set(key, value, expiry=1)
await asyncio.sleep(2)
assert key not in cache._values
assert not cache.get(key)
@@ -94,23 +94,33 @@ async def test_cache_pop_expired_returns_default(cache):
async def test_invalidate_forever_logs_and_recovers_from_errors(
settings: Settings, mocker: MockerFixture
):
test_cache = Cache(interval=0)
logger_error = mocker.patch("lnbits.utils.cache.logger.error")
test_cache = Cache()
original_running = settings.lnbits_running
calls = 0
async def fake_sleep(_interval):
original_invalidate = test_cache.invalidate_cache
async def fake_invalidate():
nonlocal calls
calls += 1
if calls == 1:
raise RuntimeError("boom")
settings.lnbits_running = False
await original_invalidate()
mocker.patch.object(test_cache, "invalidate_cache", side_effect=fake_invalidate)
mocker.patch("lnbits.task_manager.asyncio.sleep")
logger_error = mocker.patch("lnbits.task_manager.logger.error")
bg_task = None
try:
settings.lnbits_running = True
mocker.patch("lnbits.utils.cache.asyncio.sleep", side_effect=fake_sleep)
await test_cache.invalidate_forever()
bg_task = task_manager.create_permanent_task(test_cache.invalidate_cache)
await bg_task.task
finally:
settings.lnbits_running = original_running
if bg_task:
task_manager.cancel_task(bg_task)
logger_error.assert_called_once_with("Error invalidating cache")
assert logger_error.called
assert calls == 2
+4
View File
@@ -275,6 +275,10 @@ async def test_btc_rates_skips_unsupported_and_failing_providers(
@pytest.mark.anyio
async def test_btc_price_handles_empty_single_and_multiple_rates(mocker: MockerFixture):
mocker.patch(
"lnbits.utils.exchange_rates.btc_price_from_aggregator",
AsyncMock(return_value=None),
)
mocker.patch("lnbits.utils.exchange_rates.btc_rates", AsyncMock(return_value=[]))
assert await btc_price("usd") == 0.0
+10 -5
View File
@@ -1717,7 +1717,9 @@ async def test_check_fiat_status_handles_internal_states(mocker: MockerFixture):
"lnbits.core.services.fiat_providers.get_fiat_provider",
AsyncMock(return_value=provider),
)
queue_put = mocker.patch("lnbits.tasks.internal_invoice_queue.put", AsyncMock())
queue_put = mocker.patch(
"lnbits.task_manager.task_manager.internal_invoice_queue.put_nowait"
)
success_status = await check_fiat_status(
Payment(
@@ -1734,7 +1736,8 @@ async def test_check_fiat_status_handles_internal_states(mocker: MockerFixture):
)
assert success_status.success is True
queue_put.assert_awaited_once_with("fiat_pending")
queue_put.assert_called_once()
assert queue_put.call_args[0][0].checking_id == "fiat_pending"
await check_fiat_status(
Payment(
@@ -1749,7 +1752,7 @@ async def test_check_fiat_status_handles_internal_states(mocker: MockerFixture):
extra={"fiat_checking_id": "stripe_checking_id"},
)
)
assert queue_put.await_count == 1
assert queue_put.call_count == 1
@pytest.mark.anyio
@@ -1786,7 +1789,9 @@ async def test_check_fiat_status_persists_successful_payment(
"lnbits.fiat.StripeWallet.get_invoice_status",
AsyncMock(return_value=FiatPaymentStatus(paid=True)),
)
queue_put = mocker.patch("lnbits.tasks.internal_invoice_queue.put", AsyncMock())
queue_put = mocker.patch(
"lnbits.task_manager.task_manager.internal_invoice_queue.put_nowait"
)
status = await check_fiat_status(payment)
@@ -1794,7 +1799,7 @@ async def test_check_fiat_status_persists_successful_payment(
assert payment.status == PaymentState.SUCCESS
updated_payment = await get_payment(payment.checking_id)
assert updated_payment.status == PaymentState.SUCCESS
queue_put.assert_awaited_once_with(payment.checking_id)
queue_put.assert_called_once_with(payment)
@pytest.mark.anyio
-2
View File
@@ -259,9 +259,7 @@ def test_get_api_routes_extracts_v1_paths():
def test_path_and_case_helpers():
assert path_segments("/wallet/path") == ["wallet", "path"]
assert path_segments("/upgrades/ext/assets/app.js") == ["assets", "app.js"]
assert normalize_path(None) == "/"
assert normalize_path("/upgrades/ext/assets/app.js") == "/assets/app.js"
assert normalize_endpoint("example.com/") == "https://example.com"
assert normalize_endpoint("ws://socket.example.com") == "ws://socket.example.com"
assert (
+17 -12
View File
@@ -12,15 +12,12 @@ from lnbits.core.crud import create_wallet, get_standalone_payment, get_wallet
from lnbits.core.crud.payments import get_payment, get_payments_paginated
from lnbits.core.models import PaymentState, Wallet
from lnbits.core.services import create_invoice, create_user_account, pay_invoice
from lnbits.core.services.payments import update_wallet_balance
from lnbits.core.services.payments import (
update_wallet_balance,
)
from lnbits.exceptions import InvoiceError, PaymentError
from lnbits.settings import Settings
from lnbits.tasks import (
create_task,
internal_invoice_listener,
internal_invoice_queue,
wait_for_paid_invoices,
)
from lnbits.task_manager import task_manager
from lnbits.wallets.base import PaymentResponse
from lnbits.wallets.fake import FakeWallet
@@ -237,24 +234,30 @@ async def test_notification_for_internal_payment(
test_name = "test_notification_for_internal_payment"
# Drain stale items left by session-scoped fixtures (e.g. update_wallet_balance)
while not internal_invoice_queue.empty():
while not task_manager.internal_invoice_queue.empty():
try:
internal_invoice_queue.get_nowait()
task_manager.internal_invoice_queue.get_nowait()
except asyncio.QueueEmpty:
break
on_paid_mock = mocker.AsyncMock()
create_task(internal_invoice_listener())
create_task(wait_for_paid_invoices(test_name, on_paid_mock)())
# create_task(internal_invoice_listener())
task_manager.register_invoice_listener(on_paid_mock, test_name)
payment = await create_invoice(
wallet_id=to_wallet.id,
amount=123,
memo=test_name,
webhook="http://test.404.lnbits.com",
)
await pay_invoice(
paid_payment = await pay_invoice(
wallet_id=to_wallet.id, payment_request=payment.bolt11, extra={"tag": "lnurlp"}
)
assert paid_payment.status == PaymentState.SUCCESS.value
assert paid_payment.bolt11 == payment.bolt11
assert paid_payment.amount == -123_000
await asyncio.sleep(1)
assert on_paid_mock.call_count == 1
@@ -264,6 +267,8 @@ async def test_notification_for_internal_payment(
assert _payment.status == PaymentState.SUCCESS.value
assert _payment.bolt11 == payment.bolt11
assert _payment.amount == 123_000
assert _payment.checking_id == payment.checking_id
updated_payment = await get_payment(_payment.checking_id)
assert (
updated_payment.webhook_status is not None
+8 -3
View File
@@ -62,6 +62,9 @@ async def test_install_extension_creates_new_extension_and_starts_background_wor
"lnbits.core.services.extensions.start_extension_background_work",
mocker.AsyncMock(return_value=True),
)
mocker.patch(
"lnbits.core.services.extensions.core_app_extra.register_new_ext_routes"
)
mocker.patch(
"lnbits.core.services.extensions.get_db_version",
mocker.AsyncMock(return_value=0),
@@ -76,6 +79,7 @@ async def test_install_extension_creates_new_extension_and_starts_background_wor
settings.lnbits_extensions_path = str(tmp_path / "code")
extension = await install_extension(ext_info)
await activate_extension(extension) # starts background task
stored = await get_installed_extension(ext_id)
finally:
await delete_installed_extension(ext_id=ext_id)
@@ -111,6 +115,9 @@ async def test_install_extension_updates_existing_upgrade_and_preserves_payments
"lnbits.core.services.extensions.stop_extension_background_work",
mocker.AsyncMock(return_value=True),
)
mocker.patch(
"lnbits.core.services.extensions.core_app_extra.register_new_ext_routes"
)
mocker.patch(
"lnbits.core.services.extensions.get_db_version",
mocker.AsyncMock(return_value=1),
@@ -124,9 +131,8 @@ async def test_install_extension_updates_existing_upgrade_and_preserves_payments
settings.lnbits_data_folder = str(tmp_path / "data")
settings.lnbits_extensions_path = str(tmp_path / "code")
await create_installed_extension(existing_ext)
updated_ext.ext_upgrade_dir.mkdir(parents=True, exist_ok=True)
extension = await install_extension(updated_ext, skip_download=True)
await activate_extension(extension) # starts background task
stored = await get_installed_extension(ext_id)
finally:
await delete_installed_extension(ext_id=ext_id)
@@ -134,7 +140,6 @@ async def test_install_extension_updates_existing_upgrade_and_preserves_payments
settings.lnbits_extensions_path = original_extensions_path
assert extension.code == ext_id
assert extension.is_upgrade_extension is True
assert stored is not None
assert stored.meta is not None
assert stored.meta.payments == [existing_payment]
+3 -3
View File
@@ -197,8 +197,7 @@ async def test_update_wallet_balance_validates_credit_and_debit(
settings.lnbits_wallet_limit_max_balance = 0
queue_mock = mocker.patch(
"lnbits.tasks.internal_invoice_queue_put",
mocker.AsyncMock(),
"lnbits.task_manager.task_manager.internal_invoice_queue.put_nowait",
)
await update_wallet_balance(wallet, 5)
@@ -212,7 +211,8 @@ async def test_update_wallet_balance_validates_credit_and_debit(
]
assert credit_payments
assert credit_payments[0].status == PaymentState.SUCCESS
queue_mock.assert_awaited_once_with(credit_payments[0].checking_id)
queue_mock.assert_called_once()
assert queue_mock.call_args[0][0].checking_id == credit_payments[0].checking_id
@pytest.mark.anyio
+1 -6
View File
@@ -216,16 +216,11 @@ def test_installed_extensions_settings_activate_and_deactivate_paths():
}
]
installed.activate_extension_paths(
"lnurlp",
upgrade_hash="hash123",
ext_redirects=redirects,
)
installed.activate_extension_paths("lnurlp", ext_redirects=redirects)
redirect = installed.find_extension_redirect("/.well-known/lnurlp", [])
assert redirect is not None
assert redirect.ext_id == "lnurlp"
assert installed.lnbits_upgraded_extensions["lnurlp"] == "hash123"
assert "lnurlp" in installed.lnbits_installed_extensions_ids
installed.deactivate_extension_paths("lnurlp")
+127 -1
View File
@@ -1,3 +1,4 @@
import asyncio
import base64
import hashlib
import json
@@ -12,7 +13,7 @@ from Cryptodome.Util.Padding import pad, unpad
from websockets import ServerConnection
from websockets import serve as ws_serve
from lnbits.wallets.nwc import NWCWallet
from lnbits.wallets.nwc import NWCConnection, NWCWallet
from tests.wallets.helpers import (
WalletTest,
build_test_id,
@@ -99,6 +100,8 @@ async def handle( # noqa: C901
event,
)
await websocket.send(json.dumps(["EVENT", sub_id, event]))
elif 23195 in kinds:
assert sub_filter["authors"] == [mock_settings["service_public_key"]]
elif msg[0] == "EVENT":
event = msg[1]
decrypted_content = decrypt_content(
@@ -177,6 +180,129 @@ async def run(data: WalletTest):
await nwcwallet.cleanup()
@pytest.mark.anyio
async def test_nwc_rejects_event_from_unexpected_pubkey(mocker):
async def _noop(*args, **kwargs):
return None
mocker.patch("lnbits.wallets.nwc.NWCConnection._connect_to_relay", new=_noop)
mocker.patch("lnbits.wallets.nwc.NWCConnection._handle_timeouts", new=_noop)
service_private_key = PrivateKey()
service_public_key = service_private_key.public_key.format().hex()[2:]
attacker_private_key = PrivateKey()
attacker_public_key = attacker_private_key.public_key.format().hex()[2:]
account_private_key = PrivateKey()
conn = NWCConnection(
service_public_key,
account_private_key.secret.hex(),
"ws://127.0.0.1:8555",
)
try:
event = {
"kind": 23195,
"content": "{}",
"created_at": int(time.time()),
"tags": [["e", "request-event-id"]],
}
sign_event(attacker_public_key, attacker_private_key.secret.hex(), event)
with pytest.raises(Exception, match="Invalid event signature"):
await conn._on_event_message(["EVENT", "subid", event])
finally:
await conn.close()
@pytest.mark.anyio
async def test_nwc_marks_pending_invoice_settled_only_once():
wallet = NWCWallet.__new__(NWCWallet)
wallet.pending_invoice_details = {"checking-id": {"checking_id": "checking-id"}}
wallet.pending_invoices = ["checking-id"]
wallet.paid_invoices_queue = asyncio.Queue(0)
wallet._mark_invoice_settled("checking-id", source="notification")
wallet._mark_invoice_settled("checking-id", source="notification")
assert wallet.paid_invoices_queue.qsize() == 1
assert await wallet.paid_invoices_queue.get() == "checking-id"
@pytest.mark.anyio
async def test_nwc_registers_notification_subscriptions(mocker):
async def _noop(*args, **kwargs):
return None
mocker.patch("lnbits.wallets.nwc.NWCConnection._connect_to_relay", new=_noop)
mocker.patch("lnbits.wallets.nwc.NWCConnection._handle_timeouts", new=_noop)
service_private_key = PrivateKey()
service_public_key = service_private_key.public_key.format().hex()[2:]
account_private_key = PrivateKey()
conn = NWCConnection(
service_public_key,
account_private_key.secret.hex(),
"ws://127.0.0.1:8555",
)
send_mock = mocker.patch.object(conn, "_send", mocker.AsyncMock())
try:
await conn._subscribe_to_notifications()
assert len(conn.notification_subscription_ids) == 2
assert len(conn.subscriptions) == 2
assert set(conn.subscriptions.keys()) == conn.notification_subscription_ids
assert all(
subscription["method"] == "notification_sub"
and subscription["event_id"] == subscription["sub_id"]
for subscription in conn.subscriptions.values()
)
assert send_mock.await_count == 2
finally:
await conn.close()
@pytest.mark.anyio
async def test_nwc_spreads_fallback_lookups_with_cooldown(mocker):
def _schedule_next_lookup(
invoice: dict[str, object], now: float | None = None
) -> None:
assert now is not None
invoice["next_lookup_at"] = now + 1
wallet = NWCWallet.__new__(NWCWallet)
wallet.shutdown = False
wallet.pending_invoices = ["checking-1", "checking-2"]
wallet.pending_invoice_details = {
"checking-1": {
"checking_id": "checking-1",
"next_lookup_at": 0.0,
"lookup_attempts": 0,
},
"checking-2": {
"checking_id": "checking-2",
"next_lookup_at": 0.0,
"lookup_attempts": 0,
},
}
wallet.pending_invoices_lookup_cooldown = 1.0
wallet._is_shutting_down = lambda: False
wallet._payment_data_is_settled = lambda payment_data: False
wallet._cache_payment_data = lambda *args, **kwargs: None
wallet._schedule_next_lookup = _schedule_next_lookup
wallet.conn = mocker.Mock()
wallet.conn.get_info = mocker.AsyncMock()
wallet.conn.supports_method = mocker.Mock(return_value=True)
wallet.conn.call = mocker.AsyncMock(return_value={"settled_at": None})
sleep_mock = mocker.patch("lnbits.wallets.nwc.asyncio.sleep", mocker.AsyncMock())
await wallet._run_fallback_lookups(100.0)
assert wallet.conn.call.await_count == 2
sleep_mock.assert_awaited_once_with(1.0)
@pytest.mark.anyio
@pytest.mark.parametrize(
"test_data",
Generated
+1 -1
View File
@@ -1288,7 +1288,7 @@ wheels = [
[[package]]
name = "lnbits"
version = "1.5.5"
version = "1.5.6"
source = { editable = "." }
dependencies = [
{ name = "aiosqlite" },