diff --git a/lnbits/core/extensions/permissions.py b/lnbits/core/extensions/permissions.py new file mode 100644 index 000000000..7ee190bd4 --- /dev/null +++ b/lnbits/core/extensions/permissions.py @@ -0,0 +1,59 @@ +from collections.abc import Iterable +from typing import Any + +from lnbits.core.extensions.api import extension_api_permission_ids +from lnbits.core.models.extensions import ExtensionPermission, InstallableExtension + + +def validate_extension_permissions( + ext_id: str, + permissions: Iterable[ExtensionPermission], + *, + strict: bool = True, +) -> list[ExtensionPermission]: + known_permission_ids = extension_api_permission_ids() + normalized_permissions: list[ExtensionPermission] = [] + unknown_ids: list[str] = [] + + for permission in permissions: + if permission.id not in known_permission_ids: + unknown_ids.append(permission.id) + if strict: + continue + normalized_permissions.append(permission.copy(update={"label": None})) + + if unknown_ids and strict: + raise ValueError( + f"Extension '{ext_id}' requests unknown permissions: " + + ", ".join(sorted(set(unknown_ids))) + ) + + return normalized_permissions + + +def validate_wasm_extension_permissions( + ext_info: InstallableExtension, + granted_permissions: list[ExtensionPermission] | None, + extension_config: dict[str, Any], +) -> list[ExtensionPermission]: + if extension_config.get("extension_type") != "wasm": + return [] + + requested_permissions = validate_extension_permissions( + ext_info.id, + ExtensionPermission.list_from_config(extension_config), + ) + if not requested_permissions: + return [] + + if granted_permissions is None: + raise ValueError(f"Extension '{ext_info.id}' requires permission approval.") + + requested_ids = {permission.id for permission in requested_permissions} + granted_ids = {permission.id for permission in granted_permissions} + if requested_ids != granted_ids: + raise ValueError( + f"Extension '{ext_info.id}' was not granted all requested permissions." + ) + + return requested_permissions diff --git a/lnbits/core/services/extensions.py b/lnbits/core/services/extensions.py index 86c904f4d..c82bc304f 100644 --- a/lnbits/core/services/extensions.py +++ b/lnbits/core/services/extensions.py @@ -2,7 +2,6 @@ import asyncio import importlib import json import zipfile -from collections.abc import Iterable from pathlib import PurePosixPath from typing import Any @@ -21,7 +20,7 @@ from lnbits.core.crud.extensions import ( get_installed_extensions, update_installed_extension, ) -from lnbits.core.extensions.api import extension_api_permission_ids +from lnbits.core.extensions.permissions import validate_wasm_extension_permissions from lnbits.core.helpers import migrate_extension_database from lnbits.db import Connection from lnbits.settings import settings @@ -58,7 +57,7 @@ async def install_extension( await ext_info.download_archive() extension_config = _load_extension_archive_config(ext_info) - ext_info.permissions = _validate_extension_permissions( + ext_info.permissions = validate_wasm_extension_permissions( ext_info, granted_permissions, extension_config ) @@ -85,60 +84,6 @@ async def install_extension( return extension -def validate_extension_permissions( - ext_id: str, - permissions: Iterable[ExtensionPermission], - *, - strict: bool = True, -) -> list[ExtensionPermission]: - known_permission_ids = extension_api_permission_ids() - normalized_permissions: list[ExtensionPermission] = [] - unknown_ids: list[str] = [] - - for permission in permissions: - if permission.id not in known_permission_ids: - unknown_ids.append(permission.id) - if strict: - continue - normalized_permissions.append(permission.copy(update={"label": None})) - - if unknown_ids and strict: - raise ValueError( - f"Extension '{ext_id}' requests unknown permissions: " - + ", ".join(sorted(set(unknown_ids))) - ) - - return normalized_permissions - - -def _validate_extension_permissions( - ext_info: InstallableExtension, - granted_permissions: list[ExtensionPermission] | None, - extension_config: dict[str, Any], -) -> list[ExtensionPermission]: - if extension_config.get("extension_type") != "wasm": - return [] - - requested_permissions = validate_extension_permissions( - ext_info.id, - ExtensionPermission.list_from_config(extension_config), - ) - if not requested_permissions: - return [] - - if granted_permissions is None: - raise ValueError(f"Extension '{ext_info.id}' requires permission approval.") - - requested_ids = {permission.id for permission in requested_permissions} - granted_ids = {permission.id for permission in granted_permissions} - if requested_ids != granted_ids: - raise ValueError( - f"Extension '{ext_info.id}' was not granted all requested permissions." - ) - - return requested_permissions - - def _load_extension_archive_config(ext_info: InstallableExtension) -> dict[str, Any]: if not ext_info.zip_path.is_file(): return {} diff --git a/lnbits/core/views/extension_api.py b/lnbits/core/views/extension_api.py index cd0817ca7..1a1afc127 100644 --- a/lnbits/core/views/extension_api.py +++ b/lnbits/core/views/extension_api.py @@ -11,6 +11,7 @@ from loguru import logger from lnbits.core.crud.extensions import get_user_extensions from lnbits.core.crud.wallets import get_wallets_ids from lnbits.core.db import db +from lnbits.core.extensions.permissions import validate_extension_permissions from lnbits.core.models import ( SimpleStatus, ) @@ -40,7 +41,6 @@ from lnbits.core.services.extensions import ( get_valid_extensions, install_extension, uninstall_extension, - validate_extension_permissions, ) from lnbits.db import Page from lnbits.decorators import (