498 lines
16 KiB
Python
498 lines
16 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from fastapi import Depends, FastAPI, HTTPException, Request
|
|
from fastapi.responses import FileResponse
|
|
from fastapi.staticfiles import StaticFiles
|
|
from loguru import logger
|
|
|
|
from lnbits.decorators import check_user_exists, check_user_extension_access
|
|
from lnbits.helpers import template_renderer
|
|
from lnbits.settings import settings
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class WasmExtension:
|
|
id: str
|
|
name: str
|
|
version: str
|
|
root_path: Path
|
|
module_path: Path
|
|
wit_path: Path | None
|
|
world: str
|
|
host_api: str
|
|
exports: list[dict[str, Any]]
|
|
config: dict[str, Any]
|
|
|
|
|
|
def is_wasm_extension_id(ext_id: str) -> bool:
|
|
config = load_wasm_extension_config(ext_id)
|
|
return bool(config and config.get("extension_type") == "wasm")
|
|
|
|
|
|
def is_wasm_extension_dir(ext_dir: Path) -> bool:
|
|
config = _load_json(ext_dir / "config.json")
|
|
return bool(config and config.get("extension_type") == "wasm")
|
|
|
|
|
|
def load_wasm_extension_config(ext_id: str) -> dict[str, Any] | None:
|
|
ext_dir = Path(settings.lnbits_extensions_path, "extensions", ext_id)
|
|
return _load_json(ext_dir / "config.json")
|
|
|
|
|
|
def register_wasm_extension(app: FastAPI, ext_id: str) -> WasmExtension:
|
|
loaded = load_wasm_extension(ext_id)
|
|
from .wasm import warm_wasm_extension
|
|
|
|
warm_wasm_extension(loaded)
|
|
_mount_wasm_extension_static(app, loaded)
|
|
_register_wasm_extension_ui_routes(app, loaded)
|
|
_register_wasm_extension_api_routes(app, loaded)
|
|
|
|
extensions = getattr(app.state, "lnbits_wasm_extensions", {})
|
|
extensions[ext_id] = loaded
|
|
app.state.lnbits_wasm_extensions = extensions
|
|
|
|
settings.activate_extension_paths(ext_id, "", [])
|
|
logger.info(
|
|
f"Loaded WASM extension '{loaded.id}' "
|
|
f"({loaded.module_path.stat().st_size} bytes)."
|
|
)
|
|
return loaded
|
|
|
|
|
|
def load_wasm_extension(ext_id: str) -> WasmExtension:
|
|
ext_dir = Path(settings.lnbits_extensions_path, "extensions", ext_id)
|
|
config = load_wasm_extension_config(ext_id)
|
|
if not config:
|
|
raise FileNotFoundError(f"Missing WASM extension config for '{ext_id}'.")
|
|
if config.get("extension_type") != "wasm":
|
|
raise ValueError(f"Extension '{ext_id}' is not a WASM extension.")
|
|
|
|
wasm_config = config.get("wasm") or {}
|
|
module_path = _extension_path(ext_dir, wasm_config.get("module"))
|
|
wit_path = _optional_extension_path(ext_dir, wasm_config.get("wit"))
|
|
_check_wasm_module(module_path)
|
|
if wit_path and not wit_path.is_file():
|
|
raise FileNotFoundError(f"WIT file not found: {wit_path}")
|
|
|
|
return WasmExtension(
|
|
id=config.get("id") or ext_id,
|
|
name=config.get("name") or ext_id,
|
|
version=config.get("version") or "0.0",
|
|
root_path=ext_dir,
|
|
module_path=module_path,
|
|
wit_path=wit_path,
|
|
world=wasm_config.get("world") or "",
|
|
host_api=wasm_config.get("host_api") or "lnbits.core.extensions.ExtensionAPI",
|
|
exports=wasm_config.get("exports") or [],
|
|
config=config,
|
|
)
|
|
|
|
|
|
def _mount_wasm_extension_static(app: FastAPI, extension: WasmExtension) -> None:
|
|
static_path = extension.root_path / "static"
|
|
if not static_path.is_dir():
|
|
return
|
|
|
|
mount_path = f"/ext-assets/{extension.id}"
|
|
if any(getattr(route, "path", None) == mount_path for route in app.routes):
|
|
return
|
|
|
|
app.mount(
|
|
mount_path,
|
|
StaticFiles(directory=static_path),
|
|
name=f"{extension.id}-static",
|
|
)
|
|
|
|
|
|
def _register_wasm_extension_ui_routes(app: FastAPI, extension: WasmExtension) -> None:
|
|
for route_index, route_config in enumerate(extension.config.get("ui_routes") or []):
|
|
route_path = _wasm_extension_ui_route_path(extension, route_config.get("path"))
|
|
entrypoint = _wasm_extension_entrypoint(
|
|
extension, route_config.get("entrypoint")
|
|
)
|
|
frame_path = f"/ext-frame/{extension.id}/{route_index}"
|
|
auth = _wasm_extension_route_auth(extension, route_config.get("auth"))
|
|
path_params = route_config.get("path_params") or {}
|
|
_add_wasm_extension_frame_route(app, extension, frame_path, entrypoint, auth)
|
|
_add_wasm_extension_wrapper_route(
|
|
app,
|
|
extension,
|
|
route_path,
|
|
frame_path,
|
|
auth,
|
|
path_params,
|
|
)
|
|
|
|
|
|
def _register_wasm_extension_api_routes(app: FastAPI, extension: WasmExtension) -> None:
|
|
for route_config in extension.config.get("api_routes") or []:
|
|
_add_wasm_extension_api_route(app, extension, route_config)
|
|
|
|
|
|
def _add_wasm_extension_api_route(
|
|
app: FastAPI,
|
|
extension: WasmExtension,
|
|
route_config: dict[str, Any],
|
|
) -> None:
|
|
method = _wasm_extension_api_method(extension, route_config.get("method"))
|
|
route_path = _wasm_extension_api_path(extension, route_config.get("path"))
|
|
export_name = _wasm_extension_api_export(extension, route_config.get("export"))
|
|
path_params = route_config.get("path_params") or {}
|
|
auth = _wasm_extension_route_auth(extension, route_config.get("auth"))
|
|
|
|
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
|
|
|
|
try:
|
|
payload = await _read_api_payload(request, path_params)
|
|
return await invoke_export(
|
|
app,
|
|
extension.id,
|
|
export_name,
|
|
payload,
|
|
)
|
|
except KeyError as exc:
|
|
raise HTTPException(status_code=404, detail=str(exc)) from exc
|
|
except PermissionError as exc:
|
|
raise HTTPException(status_code=403, detail=str(exc)) from exc
|
|
except (TypeError, ValueError) as exc:
|
|
raise HTTPException(status_code=400, detail=str(exc)) from exc
|
|
|
|
app.add_api_route(
|
|
route_path,
|
|
invoke_wasm_extension_export,
|
|
methods=[method],
|
|
name=f"{extension.id}:{method}:{route_path}",
|
|
dependencies=_wasm_extension_dependencies(extension, auth),
|
|
include_in_schema=False,
|
|
)
|
|
|
|
|
|
async def _read_api_payload(
|
|
request: Request,
|
|
path_params: dict[str, str],
|
|
) -> dict[str, Any]:
|
|
payload = _read_api_path_params(request, path_params)
|
|
payload.update(_read_api_query_params(request))
|
|
if request.method in {"POST", "PUT", "PATCH"}:
|
|
payload.update(await _read_json_object(request))
|
|
return payload
|
|
|
|
|
|
async def _read_json_object(request: Request) -> dict[str, Any]:
|
|
body = await request.body()
|
|
if not body:
|
|
return {}
|
|
value = json.loads(body)
|
|
if not isinstance(value, dict):
|
|
raise TypeError("WASM extension API payload must be a JSON object.")
|
|
return value
|
|
|
|
|
|
def _read_api_path_params(
|
|
request: Request,
|
|
path_params: dict[str, str],
|
|
) -> dict[str, Any]:
|
|
payload: dict[str, Any] = {}
|
|
for key, value in request.path_params.items():
|
|
target = path_params.get(key) or _snake_to_camel(key)
|
|
payload[target] = value
|
|
return payload
|
|
|
|
|
|
def _read_api_query_params(request: Request) -> dict[str, Any]:
|
|
return {_snake_to_camel(key): value for key, value in request.query_params.items()}
|
|
|
|
|
|
def _wasm_extension_api_export(extension: WasmExtension, export_name: Any) -> str:
|
|
if not isinstance(export_name, str) or not export_name:
|
|
raise ValueError(f"Invalid API export for WASM extension '{extension.id}'.")
|
|
|
|
for export in extension.exports:
|
|
if export.get("name") != export_name:
|
|
continue
|
|
if export.get("visibility") in {"public", "authenticated"}:
|
|
return export_name
|
|
raise PermissionError(f"WASM export '{export_name}' is not callable over HTTP.")
|
|
raise KeyError(f"WASM extension '{extension.id}' has no export '{export_name}'.")
|
|
|
|
|
|
def _wasm_extension_api_method(extension: WasmExtension, method: Any) -> str:
|
|
if not isinstance(method, str):
|
|
raise ValueError(f"Invalid API method for WASM extension '{extension.id}'.")
|
|
method = method.upper()
|
|
if method not in {"GET", "POST", "PUT", "PATCH", "DELETE"}:
|
|
raise ValueError(f"Unsupported API method for WASM extension '{extension.id}'.")
|
|
return method
|
|
|
|
|
|
def _wasm_extension_api_path(extension: WasmExtension, path: Any) -> str:
|
|
if not isinstance(path, str) or not path.startswith("/"):
|
|
raise ValueError(f"Invalid API path for WASM extension '{extension.id}'.")
|
|
if path == "/":
|
|
return f"/api/v1/ext/{extension.id}"
|
|
return f"/api/v1/ext/{extension.id}{path}"
|
|
|
|
|
|
def _has_route(app: FastAPI, route_path: str, method: str) -> bool:
|
|
for route in app.routes:
|
|
if getattr(route, "path", None) != route_path:
|
|
continue
|
|
methods = getattr(route, "methods", set()) or set()
|
|
if method in methods:
|
|
return True
|
|
return False
|
|
|
|
|
|
def _snake_to_camel(value: str) -> str:
|
|
head, *tail = value.split("_")
|
|
return head + "".join(part.capitalize() for part in tail)
|
|
|
|
|
|
def _add_wasm_extension_wrapper_route(
|
|
app: FastAPI,
|
|
extension: WasmExtension,
|
|
route_path: str,
|
|
frame_path: str,
|
|
auth: str,
|
|
path_params: dict[str, str],
|
|
) -> None:
|
|
if _has_route(app, route_path, "GET"):
|
|
return
|
|
|
|
require_user = _require_wasm_user_extension(extension.id)
|
|
|
|
async def serve_private_wasm_extension_page(
|
|
request: Request,
|
|
user: Any = Depends(require_user),
|
|
) -> Any:
|
|
return _wasm_extension_wrapper_response(
|
|
request,
|
|
extension,
|
|
frame_path,
|
|
auth,
|
|
path_params,
|
|
user.json(),
|
|
)
|
|
|
|
async def serve_public_wasm_extension_page(request: Request) -> Any:
|
|
return _wasm_extension_wrapper_response(
|
|
request,
|
|
extension,
|
|
frame_path,
|
|
auth,
|
|
path_params,
|
|
None,
|
|
)
|
|
|
|
app.add_api_route(
|
|
route_path,
|
|
(
|
|
serve_public_wasm_extension_page
|
|
if auth == "public"
|
|
else serve_private_wasm_extension_page
|
|
),
|
|
methods=["GET"],
|
|
name=f"{extension.id}:{route_path}",
|
|
include_in_schema=False,
|
|
)
|
|
|
|
|
|
def _add_wasm_extension_frame_route(
|
|
app: FastAPI,
|
|
extension: WasmExtension,
|
|
frame_path: str,
|
|
entrypoint: Path,
|
|
auth: str,
|
|
) -> None:
|
|
if _has_route(app, frame_path, "GET"):
|
|
return
|
|
|
|
async def serve_wasm_extension_frame() -> FileResponse:
|
|
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["Cache-Control"] = "no-store"
|
|
response.headers["Cross-Origin-Opener-Policy"] = "same-origin"
|
|
response.headers["Cross-Origin-Resource-Policy"] = "same-origin"
|
|
# for access will use the bridge API, but we don't want to allow any other access
|
|
response.headers["Permissions-Policy"] = (
|
|
"camera=(), microphone=(), geolocation=(), payment=(), "
|
|
"clipboard-read=(), usb=()"
|
|
)
|
|
response.headers["Referrer-Policy"] = "no-referrer"
|
|
response.headers["X-Content-Type-Options"] = "nosniff"
|
|
return response
|
|
|
|
app.add_api_route(
|
|
frame_path,
|
|
serve_wasm_extension_frame,
|
|
methods=["GET"],
|
|
name=f"{extension.id}:frame:{frame_path}",
|
|
dependencies=_wasm_extension_dependencies(extension, auth),
|
|
include_in_schema=False,
|
|
)
|
|
|
|
|
|
def _wasm_extension_wrapper_response(
|
|
request: Request,
|
|
extension: WasmExtension,
|
|
frame_path: str,
|
|
auth: str,
|
|
path_params: dict[str, str],
|
|
user_json: str | None,
|
|
) -> Any:
|
|
public = auth == "public"
|
|
return template_renderer().TemplateResponse(
|
|
request,
|
|
"wasm_extension.html",
|
|
{
|
|
"extension": extension,
|
|
"frame_url": frame_path,
|
|
"bridge": {
|
|
"extensionId": extension.id,
|
|
"public": public,
|
|
"routeParams": _read_api_path_params(request, path_params),
|
|
"query": _read_api_query_params(request),
|
|
"apiRoutes": _wasm_extension_bridge_api_routes(extension, public),
|
|
},
|
|
"public": public,
|
|
"user": user_json,
|
|
},
|
|
)
|
|
|
|
|
|
def _wasm_extension_bridge_api_routes(
|
|
extension: WasmExtension,
|
|
public: bool,
|
|
) -> list[dict[str, str]]:
|
|
routes: list[dict[str, str]] = []
|
|
for route_config in extension.config.get("api_routes") or []:
|
|
auth = _wasm_extension_route_auth(extension, route_config.get("auth"))
|
|
if public and auth != "public":
|
|
continue
|
|
method = _wasm_extension_api_method(extension, route_config.get("method"))
|
|
path = _wasm_extension_api_path(extension, route_config.get("path"))
|
|
_wasm_extension_api_export(extension, route_config.get("export"))
|
|
routes.append(
|
|
{
|
|
"method": method,
|
|
"path": path,
|
|
"pattern": _path_template_pattern(path),
|
|
}
|
|
)
|
|
return routes
|
|
|
|
|
|
def _path_template_pattern(path: str) -> str:
|
|
pattern = re.sub(r"\\{[^/{}]+\\}", r"[^/]+", re.escape(path))
|
|
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),
|
|
) -> Any:
|
|
status = await check_user_extension_access(user.id, ext_id)
|
|
if not status.success:
|
|
raise HTTPException(status_code=403, detail=status.message)
|
|
return user
|
|
|
|
return require_wasm_user_extension
|
|
|
|
|
|
def _load_json(path: Path) -> dict[str, Any] | None:
|
|
if not path.is_file():
|
|
return None
|
|
with path.open("r", encoding="utf-8") as config_file:
|
|
value = json.load(config_file)
|
|
if not isinstance(value, dict):
|
|
raise ValueError(f"Expected JSON object in '{path}'.")
|
|
return value
|
|
|
|
|
|
def _extension_path(ext_dir: Path, value: Any) -> Path:
|
|
if not isinstance(value, str) or not value:
|
|
raise ValueError(f"Missing relative path for extension '{ext_dir.name}'.")
|
|
path = (ext_dir / value).resolve()
|
|
if ext_dir.resolve() not in path.parents:
|
|
raise ValueError(f"Extension path escapes extension root: {value}")
|
|
return path
|
|
|
|
|
|
def _optional_extension_path(ext_dir: Path, value: Any) -> Path | None:
|
|
if value is None:
|
|
return None
|
|
return _extension_path(ext_dir, value)
|
|
|
|
|
|
def _wasm_extension_route_auth(extension: WasmExtension, auth: Any) -> str:
|
|
if auth in {"public", "user"}:
|
|
return auth
|
|
raise ValueError(f"Invalid route auth for WASM extension '{extension.id}'.")
|
|
|
|
|
|
def _wasm_extension_ui_route_path(extension: WasmExtension, path: Any) -> str:
|
|
if not isinstance(path, str) or not path.startswith("/"):
|
|
raise ValueError(f"Invalid route path for WASM extension '{extension.id}'.")
|
|
if path == "/":
|
|
return "/ext"
|
|
return f"/ext{path}"
|
|
|
|
|
|
def _wasm_extension_entrypoint(extension: WasmExtension, entrypoint: Any) -> Path:
|
|
if not isinstance(entrypoint, str) or not entrypoint.startswith("/"):
|
|
raise ValueError(
|
|
f"Invalid route entrypoint for WASM extension '{extension.id}'."
|
|
)
|
|
|
|
static_prefix = f"/ext-assets/{extension.id}/"
|
|
if not entrypoint.startswith(static_prefix):
|
|
raise ValueError(
|
|
f"Route entrypoint for WASM extension '{extension.id}' must be under "
|
|
f"'{static_prefix}'."
|
|
)
|
|
|
|
path = extension.root_path / "static" / entrypoint.removeprefix(static_prefix)
|
|
path = path.resolve()
|
|
if extension.root_path.resolve() not in path.parents:
|
|
raise ValueError(f"Route entrypoint escapes extension root: {entrypoint}")
|
|
if not path.is_file():
|
|
raise FileNotFoundError(f"Route entrypoint not found: {path}")
|
|
return path
|
|
|
|
|
|
def _check_wasm_module(path: Path) -> None:
|
|
if not path.is_file():
|
|
raise FileNotFoundError(f"WASM module not found: {path}")
|
|
with path.open("rb") as wasm_file:
|
|
magic = wasm_file.read(4)
|
|
if magic != b"\0asm":
|
|
raise ValueError(f"Invalid WASM module: {path}")
|