diff --git a/lnbits/core/models/extensions.py b/lnbits/core/models/extensions.py index e8faaecec..262cafe13 100644 --- a/lnbits/core/models/extensions.py +++ b/lnbits/core/models/extensions.py @@ -472,6 +472,30 @@ class InstallableExtension(BaseModel): os.remove(ext_zip_file) raise AssertionError("File hash missmatch. Will not install.") + def load_archive_config(self) -> dict[str, Any]: + if not self.zip_path.is_file(): + return {} + + try: + with zipfile.ZipFile(self.zip_path, "r") as archive: + config_name = self._archive_config_name(archive.namelist()) + if not config_name: + return {} + with archive.open(config_name) as config_file: + config = json.load(config_file) + except Exception as exc: + raise ValueError(f"Cannot read extension config for '{self.id}'.") from exc + + return config if isinstance(config, dict) else {} + + @staticmethod + def _archive_config_name(names: list[str]) -> str | None: + for name in names: + path = PurePosixPath(name) + if len(path.parts) == 2 and path.name == "config.json": + return name + return None + def extract_archive(self): logger.info(f"Extracting extension {self.name} ({self.installed_version}).") Path(settings.lnbits_extensions_upgrade_path).mkdir(parents=True, exist_ok=True) diff --git a/lnbits/core/services/extensions.py b/lnbits/core/services/extensions.py index c82bc304f..7c2f60ae7 100644 --- a/lnbits/core/services/extensions.py +++ b/lnbits/core/services/extensions.py @@ -1,9 +1,5 @@ import asyncio import importlib -import json -import zipfile -from pathlib import PurePosixPath -from typing import Any from loguru import logger @@ -56,7 +52,7 @@ async def install_extension( if not skip_download: await ext_info.download_archive() - extension_config = _load_extension_archive_config(ext_info) + extension_config = ext_info.load_archive_config() ext_info.permissions = validate_wasm_extension_permissions( ext_info, granted_permissions, extension_config ) @@ -84,31 +80,6 @@ async def install_extension( return extension -def _load_extension_archive_config(ext_info: InstallableExtension) -> dict[str, Any]: - if not ext_info.zip_path.is_file(): - return {} - - try: - with zipfile.ZipFile(ext_info.zip_path, "r") as archive: - config_name = _archive_config_name(archive.namelist()) - if not config_name: - return {} - with archive.open(config_name) as config_file: - config = json.load(config_file) - except Exception as exc: - raise ValueError(f"Cannot read extension config for '{ext_info.id}'.") from exc - - return config if isinstance(config, dict) else {} - - -def _archive_config_name(names: list[str]) -> str | None: - for name in names: - path = PurePosixPath(name) - if len(path.parts) == 2 and path.name == "config.json": - return name - return None - - async def check_extensions_limit(installed_ext: InstallableExtension | None = None): if settings.lnbits_max_extensions == 0 or installed_ext: return