feat: partial implementation

lnbits/core/wasm_ext/api/host.py

@@ -16,6 +16,7 @@ from ..storage.crud import (
     storage_count_rows,
     storage_delete_row,
     storage_get_paginated_rows,
+    storage_get_public_paginated_rows,
     storage_get_public_row,
     storage_get_row,
     storage_get_row_owner_id,
@@ -56,8 +57,11 @@ from .models import (
     UserWalletSummary,
     WalletBalanceRequest,
     WalletBalanceResponse,
+    WebsocketPublishRequest,
+    WebsocketPublishResponse,
 )
 from .registry import extension_api_method
+from .websockets import scoped_websocket_item_id
 
 logger = logging.getLogger("lnbits.extensions")
 PUBLIC_APPEND_DEFAULT_MAX_ROWS_PER_SOURCE = 10_000
@@ -224,6 +228,46 @@ class ExtensionHostAPI:
             total=page["total"],
         )
 
+    @extension_api_method(
+        method_id="storage.get_public_paginated",
+        namespace="storage",
+        name="Get paginated public storage rows",
+        host_name="storage_get_public_paginated",
+        sdk_name="getPublicPaginated",
+        description="Get filtered, searched, sorted, paginated public storage rows.",
+        required_permission="ext.storage.read_public",
+        require_auth=False,
+    )
+    async def storage_get_public_paginated(
+        self, request: StoragePaginatedRequest
+    ) -> StoragePaginatedResponse:
+        public_fields = self._public_storage_fields(request.table)
+        self._validate_public_storage_query_fields(request, public_fields)
+        page = await storage_get_public_paginated_rows(
+            self.extension_id,
+            request.table,
+            request.filters,
+            search=request.search,
+            search_fields=request.search_fields,
+            sort_by=request.sort_by,
+            descending=request.descending,
+            limit=request.limit,
+            offset=request.offset,
+        )
+        return StoragePaginatedResponse(
+            rows_json=json.dumps(
+                [
+                    {
+                        field_name: value
+                        for field_name, value in row.items()
+                        if field_name in public_fields
+                    }
+                    for row in page["data"]
+                ]
+            ),
+            total=page["total"],
+        )
+
     @extension_api_method(
         method_id="storage.delete",
         namespace="storage",
@@ -245,6 +289,25 @@ class ExtensionHostAPI:
         )
         return StorageDeleteResponse()
 
+    @extension_api_method(
+        method_id="websocket.publish",
+        namespace="websocket",
+        name="Publish websocket message",
+        host_name="websocket_publish",
+        sdk_name="publish",
+        description="Publish a JSON message on an extension-local websocket channel.",
+        required_permission="websocket.publish",
+        require_auth=False,
+    )
+    async def websocket_publish(
+        self, request: WebsocketPublishRequest
+    ) -> WebsocketPublishResponse:
+        from lnbits.core.services import websocket_manager
+
+        item_id = scoped_websocket_item_id(self.extension_id, request.item_id)
+        await websocket_manager.send(item_id, request.data_json)
+        return WebsocketPublishResponse()
+
     @extension_api_method(
         method_id="wallet.create_invoice",
         namespace="wallet",
@@ -698,6 +761,20 @@ class ExtensionHostAPI:
 
         raise PermissionError(f"Storage table '{table}' is not publicly readable.")
 
+    def _validate_public_storage_query_fields(
+        self, request: StoragePaginatedRequest, public_fields: set[str]
+    ) -> None:
+        query_fields = set(request.filters)
+        query_fields.update(request.search_fields)
+        if request.sort_by:
+            query_fields.add(request.sort_by)
+        private_fields = sorted(query_fields - public_fields)
+        if private_fields:
+            raise PermissionError(
+                "Public storage query uses non-public fields: "
+                + ", ".join(private_fields)
+            )
+
     async def _public_storage_append_policy(
         self, table: str, source_id: str
     ) -> tuple[dict[str, Any], str]:

lnbits/core/wasm_ext/api/models.py

@@ -114,6 +114,38 @@ class StoragePaginatedResponse(BaseModel):
     total: int = 0
 
 
+class WebsocketPublishRequest(BaseModel):
+    item_id: str = Field(..., min_length=1, max_length=128)
+    data: Any = Field(default_factory=dict)
+
+    @root_validator(pre=True)
+    def parse_data_json(cls, values: dict[str, Any]) -> dict[str, Any]:
+        data_json = values.get("data_json")
+        if data_json is not None and "data" not in values:
+            values["data"] = json.loads(data_json)
+        return values
+
+    @root_validator
+    def validate_data_size(cls, values: dict[str, Any]) -> dict[str, Any]:
+        data = values.get("data")
+        try:
+            encoded = json.dumps(data, separators=(",", ":"))
+        except TypeError as exc:
+            raise ValueError("websocket data must be JSON serializable.") from exc
+        if len(encoded.encode()) > 65536:
+            raise ValueError("websocket data must not exceed 65536 bytes.")
+        values["data"] = data
+        return values
+
+    @property
+    def data_json(self) -> str:
+        return json.dumps(self.data, separators=(",", ":"))
+
+
+class WebsocketPublishResponse(BaseModel):
+    sent: bool = True
+
+
 class StorageDeleteRequest(BaseModel):
     table: str = Field(..., min_length=1, max_length=128)
     id: str = Field(..., min_length=1, max_length=512)

lnbits/core/wasm_ext/api/registry.py

@@ -15,6 +15,7 @@ _EXTENSION_RUNTIME_PERMISSION_IDS = {
     "wallet.pay_invoice",
     "wallet.pay_invoice_background",
     "wallet.payments.watch",
+    "websocket.subscribe",
 }
 _RequestModel = TypeVar("_RequestModel", bound=BaseModel)
 _ResponseModel = TypeVar("_ResponseModel", bound=BaseModel)

lnbits/core/wasm_ext/api/websockets.py

@@ -0,0 +1,17 @@
+from __future__ import annotations
+
+import re
+
+_EXTENSION_ID_RE = re.compile(r"^[A-Za-z0-9_-]{1,128}$")
+_LOCAL_ITEM_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9:_-]{0,127}$")
+
+
+def scoped_websocket_item_id(extension_id: str, item_id: str) -> str:
+    if not _EXTENSION_ID_RE.fullmatch(extension_id):
+        raise ValueError("Extension websocket namespace is invalid.")
+    if not _LOCAL_ITEM_ID_RE.fullmatch(item_id):
+        raise ValueError(
+            "Extension websocket item ID must be 1-128 characters and contain "
+            "only letters, numbers, colon, underscore, or dash."
+        )
+    return f"ext:{extension_id}:{item_id}"

lnbits/core/wasm_ext/storage/__init__.py

@@ -4,6 +4,7 @@ from .crud import (
     storage_count_rows,
     storage_delete_row,
     storage_get_paginated_rows,
+    storage_get_public_paginated_rows,
     storage_get_public_row,
     storage_get_row,
     storage_get_row_owner_id,
@@ -16,6 +17,7 @@ __all__ = [
     "storage_count_rows",
     "storage_delete_row",
     "storage_get_paginated_rows",
+    "storage_get_public_paginated_rows",
     "storage_get_public_row",
     "storage_get_row",
     "storage_get_row_owner_id",

lnbits/core/wasm_ext/storage/crud.py

@@ -191,6 +191,50 @@ async def storage_get_paginated_rows(
     }
 
 
+async def storage_get_public_paginated_rows(
+    ext_id: str,
+    table: str,
+    filters: dict[str, Any],
+    *,
+    search: str | None,
+    search_fields: list[str],
+    sort_by: str | None,
+    descending: bool,
+    limit: int,
+    offset: int,
+) -> dict[str, Any]:
+    table_schema = _load_table_schema(ext_id, table)
+    database = Database(f"ext_{ext_id}")
+    where_sql, values = _where_sql(
+        database, table_schema, filters, search, search_fields
+    )
+    order_sql = _order_sql(table_schema, sort_by, descending)
+    count_values = dict(values)
+    values.update({"limit": min(limit, 1000), "offset": offset})
+
+    table_ref = _table_ref_for_schema(ext_id, table)
+    rows_query = f"""
+        SELECT * FROM {table_ref}
+        {where_sql}
+        {order_sql}
+        LIMIT :limit
+        OFFSET :offset
+    """  # noqa: S608
+    count_query = f"""
+        SELECT COUNT(*) AS count FROM {table_ref}
+        {where_sql}
+    """  # noqa: S608
+
+    async with database.connect() as conn:
+        rows = await conn.fetchall(rows_query, values)
+        count_row = await conn.fetchone(count_query, count_values)
+
+    return {
+        "data": [_row_from_db(table_schema, row) for row in rows],
+        "total": int(count_row["count"]) if count_row else 0,
+    }
+
+
 async def storage_delete_row(
     ext_id: str,
     table: str,

lnbits/static/i18n/en.js

@@ -559,6 +559,9 @@ window.localisation.en = {
   extension_permission_http_request_hosts: 'Allowed hosts',
   extension_permission_utils_basic: 'Use basic LNbits utilities',
   extension_permission_ui_camera_scan_qr: 'Scan QR codes',
+  extension_permission_websocket: 'Use extension websockets',
+  extension_permission_websocket_publish: 'Publish websocket messages',
+  extension_permission_websocket_subscribe: 'Subscribe to websocket messages',
   extension_permission_wallet_payments_watch: 'Watch wallet payments',
   extension_permission_wallet_create_invoice: 'Create invoices',
   extension_permission_wallet_create_invoice_public:

lnbits/static/js/components/lnbits-extension-permissions.js

@@ -110,6 +110,9 @@
         'extension_permission_warning_wallet_payments_watch'
       )
     }
+    if (['websocket.publish', 'websocket.subscribe'].includes(permission.id)) {
+      return mediumRisk(translateFn)
+    }
     if (
       [
         'wallet.list',
@@ -143,6 +146,9 @@
       'extension.api.request',
       'http.request',
       'ui.camera.scan_qr',
+      'websocket',
+      'websocket.publish',
+      'websocket.subscribe',
       'ext.storage.read',
       'ext.storage.write',
       'ext.storage.read_public',
@@ -233,14 +239,26 @@
       permissions.length === 2 &&
       permissions.some(permission => permission.id === 'ext.storage.read') &&
       permissions.some(permission => permission.id === 'ext.storage.write')
+    const isWebsocket =
+      permissions.every(permission =>
+        ['websocket.publish', 'websocket.subscribe'].includes(permission.id)
+      ) &&
+      permissions.some(permission => permission.id === 'websocket.publish') &&
+      permissions.some(permission => permission.id === 'websocket.subscribe')
     const descriptions = permissions
       .map(permission => permissionManifestDescription(permission))
       .filter(Boolean)
     const item = {
-      id: isReadWriteStorage ? 'ext.storage.read_write' : permission.id,
+      id: isReadWriteStorage
+        ? 'ext.storage.read_write'
+        : isWebsocket
+          ? 'websocket'
+          : permission.id,
       label: isReadWriteStorage
         ? translate(translateFn, 'extension_permission_ext_storage_read_write')
-        : permissionLabel(permission, translateFn),
+        : isWebsocket
+          ? translate(translateFn, 'extension_permission_websocket')
+          : permissionLabel(permission, translateFn),
       risk: permissionRisk(permissions, extensions, translateFn),
       badges: [],
       descriptions,
@@ -298,7 +316,11 @@
     const hasReadWriteStorage =
       permissionsById.has('ext.storage.read') &&
       permissionsById.has('ext.storage.write')
+    const hasWebsocket =
+      permissionsById.has('websocket.publish') &&
+      permissionsById.has('websocket.subscribe')
     let addedReadWriteStorage = false
+    let addedWebsocket = false
 
     return permissionList
       .map((permission, index) => {
@@ -317,6 +339,21 @@
             ]
           }
         }
+        if (
+          hasWebsocket &&
+          ['websocket.publish', 'websocket.subscribe'].includes(permission.id)
+        ) {
+          if (addedWebsocket) return null
+          addedWebsocket = true
+          return {
+            index,
+            orderId: 'websocket',
+            permissions: [
+              permissionsById.get('websocket.publish'),
+              permissionsById.get('websocket.subscribe')
+            ]
+          }
+        }
         return {
           index,
           orderId: permission.id,

lnbits/static/js/wasm-extension-component.js

@@ -183,7 +183,8 @@ window.WasmExtensionComponent = {
       handleWindowMessage: null,
       loading: false,
       loadId: 0,
-      paymentSubscriptions: new Map()
+      paymentSubscriptions: new Map(),
+      websocketSubscriptions: new Map()
     }
   },
   created() {
@@ -357,6 +358,29 @@ window.WasmExtensionComponent = {
         )
       })
     },
+    extensionRoute(path) {
+      let url
+      try {
+        url = new URL(String(path || ''), window.location.origin)
+      } catch (_error) {
+        throw new Error('Invalid extension route.')
+      }
+      if (url.origin !== window.location.origin) {
+        throw new Error('Extension route must stay on this server.')
+      }
+
+      const basePath = `/ext/${encodeURIComponent(this.bridge.extensionId)}`
+      if (
+        url.pathname !== basePath &&
+        !url.pathname.startsWith(`${basePath}/`)
+      ) {
+        throw new Error('Extension route must stay inside this extension.')
+      }
+      return `${url.pathname}${url.search}${url.hash}`
+    },
+    replaceExtensionRoute(message) {
+      return this.$router.replace(this.extensionRoute(message.path))
+    },
     async callApi(message) {
       const method = String(message.method || 'GET').toUpperCase()
       const path = String(message.path || '')
@@ -844,6 +868,18 @@ window.WasmExtensionComponent = {
     isPaymentHash(value) {
       return typeof value === 'string' && /^[a-f0-9]{64}$/i.test(value)
     },
+    isWebsocketItemId(value) {
+      return (
+        typeof value === 'string' &&
+        /^[A-Za-z0-9][A-Za-z0-9:_-]{0,127}$/.test(value)
+      )
+    },
+    scopedWebsocketItemId(itemId) {
+      if (!this.isWebsocketItemId(itemId)) {
+        throw new Error('Invalid websocket item ID.')
+      }
+      return `ext:${this.bridge.extensionId}:${itemId}`
+    },
     websocketUrl(path) {
       const url = new URL(window.location.href)
       url.protocol = url.protocol === 'https:' ? 'wss:' : 'ws:'
@@ -874,8 +910,24 @@ window.WasmExtensionComponent = {
         this.closePaymentSubscription(subscriptionId)
       }
     },
+    closeWebsocketSubscription(subscriptionId) {
+      const subscription = this.websocketSubscriptions.get(subscriptionId)
+      if (!subscription) return
+      this.websocketSubscriptions.delete(subscriptionId)
+      try {
+        subscription.socket.close()
+      } catch (_error) {}
+    },
+    closeWebsocketSubscriptions() {
+      for (const subscriptionId of Array.from(
+        this.websocketSubscriptions.keys()
+      )) {
+        this.closeWebsocketSubscription(subscriptionId)
+      }
+    },
     closeBridgePort() {
       this.closePaymentSubscriptions()
+      this.closeWebsocketSubscriptions()
       this.bridgePort?.close()
       this.bridgePort = null
     },
@@ -937,6 +989,55 @@ window.WasmExtensionComponent = {
         this.paymentSubscriptions.delete(subscriptionId)
       })
     },
+    subscribeWebsocket(message) {
+      if (!this.hasBridgePermission('websocket.subscribe')) {
+        throw new Error('Extension is missing websocket subscribe permission.')
+      }
+
+      const subscriptionId = String(message.subscriptionId || '')
+      const itemId = String(message.itemId || '')
+
+      if (
+        !subscriptionId ||
+        subscriptionId.length > 128 ||
+        !this.isWebsocketItemId(itemId)
+      ) {
+        throw new Error('Invalid websocket subscription.')
+      }
+
+      this.closeWebsocketSubscription(subscriptionId)
+
+      const scopedItemId = this.scopedWebsocketItemId(itemId)
+      const socket = new WebSocket(
+        this.websocketUrl(`/api/v1/ws/${encodeURIComponent(scopedItemId)}`)
+      )
+      this.websocketSubscriptions.set(subscriptionId, {itemId, socket})
+
+      socket.addEventListener('message', event => {
+        let data = event.data
+        try {
+          data = JSON.parse(event.data)
+        } catch (_error) {}
+
+        this.sendBridgeEvent({
+          event: 'websocket.message',
+          subscriptionId,
+          itemId,
+          data
+        })
+      })
+      socket.addEventListener('error', () => {
+        this.sendBridgeEvent({
+          event: 'websocket.error',
+          subscriptionId,
+          itemId
+        })
+        this.closeWebsocketSubscription(subscriptionId)
+      })
+      socket.addEventListener('close', () => {
+        this.websocketSubscriptions.delete(subscriptionId)
+      })
+    },
     async handleBridgeRequest(message, reply) {
       if (!message || message.type !== 'lnbits-extension:request') return
 
@@ -966,6 +1067,15 @@ window.WasmExtensionComponent = {
           return
         }
 
+        if (message.action === 'navigation.replace') {
+          await this.replaceExtensionRoute(message)
+          this.sendResponse(reply, message.id, {
+            ok: true,
+            data: {ok: true}
+          })
+          return
+        }
+
         if (message.action === 'ui.scan_qr') {
           this.sendResponse(reply, message.id, {
             ok: true,
@@ -1016,6 +1126,24 @@ window.WasmExtensionComponent = {
           return
         }
 
+        if (message.action === 'websocket.subscribe') {
+          this.subscribeWebsocket(message)
+          this.sendResponse(reply, message.id, {
+            ok: true,
+            data: {ok: true}
+          })
+          return
+        }
+
+        if (message.action === 'websocket.unsubscribe') {
+          this.closeWebsocketSubscription(String(message.subscriptionId || ''))
+          this.sendResponse(reply, message.id, {
+            ok: true,
+            data: {ok: true}
+          })
+          return
+        }
+
         throw new Error('Unknown extension bridge action.')
       } catch (error) {
         this.sendResponse(reply, message.id, {

tests/unit/test_wasm_extension_frontend.py

@@ -14,7 +14,7 @@ def test_wasm_frontend_assets_are_registered_in_component_bundle():
     assert "js/components/admin/lnbits-admin-wasm-limit-config.js" in components
 
 
-def test_wasm_frontend_bridge_restricts_api_routes_and_payment_actions():
+def test_wasm_frontend_bridge_restricts_api_routes_and_realtime_actions():
     bridge = (ROOT / "lnbits/static/js/wasm-extension-component.js").read_text(
         encoding="utf-8"
     )
@@ -22,8 +22,15 @@ def test_wasm_frontend_bridge_restricts_api_routes_and_payment_actions():
     assert "allowedApiRoute(method, path)" in bridge
     assert "url.origin !== window.location.origin" in bridge
     assert "Extension API route is not allowed." in bridge
+    assert "extensionRoute(path)" in bridge
+    assert "Extension route must stay inside this extension." in bridge
     assert "message.action === 'payment.subscribe'" in bridge
     assert "message.action === 'payment.unsubscribe'" in bridge
+    assert "message.action === 'websocket.subscribe'" in bridge
+    assert "message.action === 'websocket.unsubscribe'" in bridge
+    assert "message.action === 'navigation.replace'" in bridge
+    assert "hasBridgePermission('websocket.subscribe')" in bridge
+    assert "ext:${this.bridge.extensionId}:${itemId}" in bridge
     assert "message.action === 'ui.scan_qr'" in bridge
 
 

tests/unit/test_wasm_extension_host_api.py

@@ -12,7 +12,9 @@ from lnbits.core.wasm_ext.api.models import (
     PayInvoiceRequest,
     StorageAppendPublicRequest,
     StorageGetRequest,
+    StoragePaginatedRequest,
     WalletBalanceRequest,
+    WebsocketPublishRequest,
 )
 from lnbits.exceptions import PaymentError
 from lnbits.helpers import sha256s
@@ -49,6 +51,128 @@ async def test_host_api_filters_public_storage_fields(mocker: MockerFixture):
     storage_mock.assert_awaited_once_with("demoext", "tips", "tip-1")
 
 
+@pytest.mark.anyio
+async def test_host_api_filters_public_paginated_storage_rows(
+    mocker: MockerFixture,
+):
+    storage_mock = mocker.patch(
+        "lnbits.core.wasm_ext.api.host.storage_get_public_paginated_rows",
+        mocker.AsyncMock(
+            return_value={
+                "data": [
+                    {
+                        "id": "message-1",
+                        "thread_id": "thread-1",
+                        "message": "Hello",
+                        "admin_note": "secret",
+                    }
+                ],
+                "total": 1,
+            }
+        ),
+    )
+    api = ExtensionHostAPI(
+        "demoext",
+        [
+            ExtensionPermission(
+                id="ext.storage.read_public",
+                policies=[
+                    {
+                        "table_name": "messages",
+                        "public_fields": ["id", "thread_id", "message"],
+                    }
+                ],
+            )
+        ],
+    )
+
+    response = await api.storage_get_public_paginated(
+        StoragePaginatedRequest(
+            table="messages",
+            filters={"thread_id": "thread-1"},
+            search="hello",
+            search_fields=["message"],
+            sort_by="id",
+            descending=False,
+            limit=25,
+            offset=0,
+        )
+    )
+
+    assert json.loads(response.rows_json) == [
+        {"id": "message-1", "thread_id": "thread-1", "message": "Hello"}
+    ]
+    assert response.total == 1
+    storage_mock.assert_awaited_once()
+
+
+@pytest.mark.anyio
+async def test_host_api_public_paginated_storage_rejects_private_query_fields():
+    api = ExtensionHostAPI(
+        "demoext",
+        [
+            ExtensionPermission(
+                id="ext.storage.read_public",
+                policies=[
+                    {
+                        "table_name": "messages",
+                        "public_fields": ["id", "message"],
+                    }
+                ],
+            )
+        ],
+    )
+
+    with pytest.raises(PermissionError, match="non-public fields"):
+        await api.storage_get_public_paginated(
+            StoragePaginatedRequest(
+                table="messages",
+                filters={"admin_note": "secret"},
+                search=None,
+                search_fields=[],
+                sort_by=None,
+                descending=False,
+                limit=25,
+                offset=0,
+            )
+        )
+
+
+@pytest.mark.anyio
+async def test_host_api_websocket_publish_scopes_item_id(mocker: MockerFixture):
+    send_mock = mocker.patch(
+        "lnbits.core.services.websocket_manager.send",
+        mocker.AsyncMock(),
+    )
+    api = ExtensionHostAPI("demoext", ["websocket.publish"])
+
+    response = await api.websocket_publish(
+        WebsocketPublishRequest(
+            item_id="conversation:abc_123",
+            data={"message": "Hello"},
+        )
+    )
+
+    assert response.sent is True
+    send_mock.assert_awaited_once_with(
+        "ext:demoext:conversation:abc_123",
+        '{"message":"Hello"}',
+    )
+
+
+@pytest.mark.anyio
+async def test_host_api_websocket_publish_rejects_invalid_item_id():
+    api = ExtensionHostAPI("demoext", ["websocket.publish"])
+
+    with pytest.raises(ValueError, match="item ID"):
+        await api.websocket_publish(
+            WebsocketPublishRequest(
+                item_id="../other",
+                data={"message": "Hello"},
+            )
+        )
+
+
 @pytest.mark.anyio
 async def test_host_api_storage_requires_owner_context_and_uses_user_hash(
     mocker: MockerFixture,

tests/unit/test_wasm_extension_permissions.py

@@ -269,6 +269,29 @@ def test_validate_wasm_permissions_allows_wallet_payments_watch_permission():
     ) == [ExtensionPermission(id="wallet.payments.watch")]
 
 
+def test_validate_wasm_permissions_allows_websocket_permissions():
+    ext_info = make_installable_extension("demoext")
+    extension_config = _wasm_config(
+        "demoext",
+        [
+            {"id": "websocket.publish"},
+            {"id": "websocket.subscribe"},
+        ],
+    )
+
+    assert validate_wasm_extension_permissions(
+        ext_info,
+        [
+            ExtensionPermission(id="websocket.publish"),
+            ExtensionPermission(id="websocket.subscribe"),
+        ],
+        extension_config,
+    ) == [
+        ExtensionPermission(id="websocket.publish"),
+        ExtensionPermission(id="websocket.subscribe"),
+    ]
+
+
 def test_background_payment_grant_lookup_and_policy_coverage():
     permissions = {
         WALLET_PAY_INVOICE_BACKGROUND_PERMISSION: [