Michilis/lnbits
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add wallet list

Browse Source
This commit is contained in:
Vlad Stan
2026-07-09 16:30:18 +03:00
parent f2920e8eb1
commit 20e3b4c0b4
7 changed files with 164 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lnbits/core/extensions/api.py
+15
View File
@@ -21,6 +21,7 @@ from .models import (
KvListResponse,
KvSetRequest,
KvSetResponse,
ListUserWalletsResponse,
LogRequest,
LogResponse,
NowResponse,
@@ -173,6 +174,20 @@ class ExtensionAPI:
) -> CreateInvoiceResponse:
self._raise_unwired_runtime("wallet_create_invoice")
@extension_api_method(
method_id="wallet.list_user_wallets",
namespace="wallet",
name="List user wallets",
host_name="list_user_wallets",
sdk_name="listUserWallets",
description="List wallets available to the authenticated extension user.",
required_permission="wallet.list",
)
async def wallet_list_user_wallets(
self, request: EmptyRequest
) -> ListUserWalletsResponse:
self._raise_unwired_runtime("wallet_list_user_wallets")
@extension_api_method(
method_id="payments.watch",
namespace="payments",
lnbits/core/extensions/codegen/typescript.py
+29 -4
View File
@@ -69,12 +69,31 @@ def _select_methods(
def _collect_models(methods: Sequence[ExtensionAPIMethod]) -> list[type[BaseModel]]:
models: dict[str, type[BaseModel]] = {}
for method in methods:
models[method.request_model.__name__] = method.request_model
models[method.response_model.__name__] = method.response_model
pending = [
model
for method in methods
for model in (method.request_model, method.response_model)
]
while pending:
model = pending.pop()
if model.__name__ in models:
continue
models[model.__name__] = model
for field in model.__fields__.values():
pending.extend(_nested_model_types(field.outer_type_))
return [models[name] for name in sorted(models)]
def _nested_model_types(type_: Any) -> list[type[BaseModel]]:
models: list[type[BaseModel]] = []
if _is_model_type(type_):
models.append(type_)
for arg in get_args(type_):
models.extend(_nested_model_types(arg))
return models
def _render_model_type(model: type[BaseModel]) -> list[str]:
name = _model_name(model)
fields = model.__fields__
@@ -105,7 +124,9 @@ def _python_type_to_ts(type_: Any, allow_none: bool = False) -> str:
if origin is Literal:
return " | ".join(_literal_to_ts(arg) for arg in args)
if origin in (list, Sequence):
if _is_model_type(type_):
ts = _model_name(type_)
elif origin in (list, Sequence):
item_type = _python_type_to_ts(args[0]) if args else "unknown"
ts = f"{item_type}[]"
elif origin is dict:
@@ -145,6 +166,10 @@ def _is_subclass(type_: Any, class_: type) -> bool:
return False
def _is_model_type(type_: Any) -> bool:
return _is_subclass(type_, BaseModel)
def _render_method_metadata(
methods: Sequence[ExtensionAPIMethod],
) -> list[str]:
lnbits/core/extensions/loader.py
+49 -22
View File
@@ -192,16 +192,21 @@ def _add_wasm_extension_api_route(
if _has_route(app, route_path, method):
return
async def invoke_wasm_extension_export(request: Request) -> dict[str, Any]:
from .wasm import invoke_wasm_extension_export as invoke_export
require_user = _require_wasm_user_extension(extension.id)
async def invoke_wasm_api_request(
request: Request, user: Any | None = None
) -> dict[str, Any]:
from .wasm import invoke_wasm_extension_export as invoke_wasm_export
try:
payload = await _read_api_payload(request, path_params)
return await invoke_export(
return await invoke_wasm_export(
app,
extension.id,
export_name,
payload,
user=user,
)
except KeyError as exc:
raise HTTPException(status_code=404, detail=str(exc)) from exc
@@ -210,12 +215,24 @@ def _add_wasm_extension_api_route(
except (TypeError, ValueError) as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
async def invoke_private_wasm_extension_export(
request: Request,
user: Any = Depends(require_user),
) -> dict[str, Any]:
return await invoke_wasm_api_request(request, user)
async def invoke_public_wasm_extension_export(request: Request) -> dict[str, Any]:
return await invoke_wasm_api_request(request)
app.add_api_route(
route_path,
invoke_wasm_extension_export,
(
invoke_public_wasm_extension_export
if auth == "public"
else invoke_private_wasm_extension_export
),
methods=[method],
name=f"{extension.id}:{method}:{route_path}",
dependencies=_wasm_extension_dependencies(extension, auth),
include_in_schema=False,
)
@@ -362,14 +379,8 @@ def _add_wasm_extension_frame_route(
async def serve_wasm_extension_frame(request: Request) -> FileResponse:
_consume_wasm_extension_frame_token(request, extension, frame_path)
response = FileResponse(entrypoint)
response.headers["Content-Security-Policy"] = (
"sandbox allow-scripts allow-forms; "
"default-src 'self' data: blob:; "
"connect-src 'none'; "
"form-action 'none'; "
"object-src 'none'; "
"base-uri 'none'; "
"frame-ancestors 'self'"
response.headers["Content-Security-Policy"] = _wasm_extension_frame_csp(
request, extension
)
response.headers["Cache-Control"] = "no-store"
response.headers["Cross-Origin-Opener-Policy"] = "same-origin"
@@ -423,6 +434,31 @@ def _wasm_extension_wrapper_response(
return response
def _wasm_extension_frame_csp(request: Request, extension: WasmExtension) -> str:
origin = str(request.base_url).rstrip("/")
extension_assets = f"{origin}/ext-assets/{extension.id}/"
static_assets = f"{origin}/static/"
return (
"sandbox allow-scripts; "
"default-src 'none'; "
f"script-src {extension_assets}; "
"script-src-attr 'none'; "
f"style-src {extension_assets} {static_assets}; "
"style-src-attr 'none'; "
f"img-src {extension_assets} {static_assets}; "
f"font-src {static_assets}; "
"connect-src 'none'; "
"form-action 'none'; "
"object-src 'none'; "
"base-uri 'none'; "
"frame-src 'none'; "
"worker-src 'none'; "
"media-src 'none'; "
"manifest-src 'none'; "
"frame-ancestors 'self'"
)
def _wasm_extension_frame_url(extension: WasmExtension, frame_path: str) -> str:
token = _create_wasm_extension_frame_token(extension, frame_path)
return f"{frame_path}?frame_token={token}"
@@ -506,15 +542,6 @@ def _path_template_pattern(path: str) -> str:
return f"^{pattern}$"
def _wasm_extension_dependencies(
extension: WasmExtension,
auth: str,
) -> list[Any]:
if auth == "public":
return []
return [Depends(_require_wasm_user_extension(extension.id))]
def _require_wasm_user_extension(ext_id: str) -> Any:
async def require_wasm_user_extension(
user: Any = Depends(check_user_exists),
lnbits/core/extensions/models.py
+10
View File
@@ -46,6 +46,16 @@ class CreateInvoiceResponse(BaseModel):
checking_id: str
class UserWalletSummary(BaseModel):
id: str
name: str
currency: str | None = None
class ListUserWalletsResponse(BaseModel):
wallets: list[UserWalletSummary] = Field(default_factory=list)
class WatchPaymentRequest(BaseModel):
payment_hash: str = Field(..., min_length=1, max_length=128)
callback_export: str = Field(..., min_length=1, max_length=128)
lnbits/core/extensions/prototype.py
+15
View File
@@ -9,12 +9,15 @@ from .api import ExtensionAPI
from .models import (
CreateInvoiceRequest,
CreateInvoiceResponse,
EmptyRequest,
KvGetRequest,
KvGetResponse,
KvListRequest,
KvListResponse,
KvSetRequest,
KvSetResponse,
ListUserWalletsResponse,
UserWalletSummary,
WatchPaymentRequest,
WatchPaymentResponse,
)
@@ -35,6 +38,7 @@ class InMemoryExtensionAPI(ExtensionAPI):
state: InMemoryExtensionState | None = None,
user_id: str | None = None,
wallet_id: str | None = None,
user_wallets: list[UserWalletSummary] | None = None,
) -> None:
super().__init__(
extension_id,
@@ -43,6 +47,7 @@ class InMemoryExtensionAPI(ExtensionAPI):
wallet_id=wallet_id,
)
self.state = state or InMemoryExtensionState()
self.user_wallets = list(user_wallets) if user_wallets is not None else None
async def storage_get(self, request: KvGetRequest) -> KvGetResponse:
self.require_permission("ext.storage.read_write")
@@ -77,6 +82,16 @@ class InMemoryExtensionAPI(ExtensionAPI):
checking_id=f"{self.extension_id}:{payment_hash}",
)
async def wallet_list_user_wallets(
self, _request: EmptyRequest
) -> ListUserWalletsResponse:
self.require_permission("wallet.list")
if self.user_wallets is None:
raise PermissionError(
"Listing user wallets requires an authenticated user context."
)
return ListUserWalletsResponse(wallets=self.user_wallets)
async def payments_watch(
self, request: WatchPaymentRequest
) -> WatchPaymentResponse:
lnbits/core/extensions/wasm.py
+45 -2
View File
@@ -11,6 +11,7 @@ from fastapi import FastAPI
from .api import list_extension_api_methods
from .loader import WasmExtension, register_wasm_extension
from .models import UserWalletSummary
from .prototype import InMemoryExtensionAPI, InMemoryExtensionState
from .runtime import ExtensionAPIHost
@@ -20,11 +21,19 @@ async def invoke_wasm_extension_export(
ext_id: str,
export_name: str,
payload: Mapping[str, Any] | None = None,
*,
user: Any | None = None,
) -> dict[str, Any]:
extension = _get_registered_extension(app, ext_id)
state = _get_extension_state(app)
permissions = _extension_permissions(extension)
api = InMemoryExtensionAPI(extension.id, permissions, state=state)
api = InMemoryExtensionAPI(
extension.id,
permissions,
state=state,
user_id=_user_id(user),
user_wallets=_user_wallet_summaries(user),
)
return await asyncio.to_thread(
_invoke_wasm_extension_export_sync,
@@ -143,10 +152,18 @@ def _dict_to_component_record(value: Mapping[str, Any]) -> Any:
record = component.Record()
for key, item in value.items():
setattr(record, _camel_to_kebab(key), item)
setattr(record, _camel_to_kebab(key), _to_component_value(item))
return record
def _to_component_value(value: Any) -> Any:
if isinstance(value, Mapping):
return _dict_to_component_record(value)
if isinstance(value, list):
return [_to_component_value(item) for item in value]
return value
def _parse_wasm_export_result(extension: WasmExtension, value: Any) -> dict[str, Any]:
if isinstance(value, bytes):
value = value.decode()
@@ -195,5 +212,31 @@ def _extension_permissions(extension: WasmExtension) -> set[str]:
return permissions
def _user_id(user: Any | None) -> str | None:
return getattr(user, "id", None) if user else None
def _user_wallet_summaries(user: Any | None) -> list[UserWalletSummary] | None:
if user is None:
return None
summaries: list[UserWalletSummary] = []
for wallet in getattr(user, "wallets", []) or []:
if not getattr(wallet, "can_receive_payments", False):
continue
wallet_id = getattr(wallet, "id", None)
wallet_name = getattr(wallet, "name", None)
if not wallet_id or not wallet_name:
continue
summaries.append(
UserWalletSummary(
id=wallet_id,
name=wallet_name,
currency=getattr(wallet, "currency", None),
)
)
return summaries
def _camel_to_kebab(value: str) -> str:
return re.sub(r"([a-z0-9])([A-Z])", r"\1-\2", value).replace("_", "-").lower()
lnbits/templates/wasm_extension.html
+1 -1
View File
@@ -16,7 +16,7 @@
class="wasm-extension-frame"
data-frame-url="{{ frame_url }}"
title="{{ extension.name }}"
sandbox="allow-scripts allow-forms"
sandbox="allow-scripts"
allow="clipboard-write"
referrerpolicy="no-referrer"
></iframe>