From 2b5a513693480117998d47df7c2db0417481d0db Mon Sep 17 00:00:00 2001 From: Vlad Stan Date: Mon, 27 Jul 2026 14:53:12 +0300 Subject: [PATCH] fix: check extension type when installing (#4101) --- lnbits/core/models/extensions.py | 97 ++++++++++++++++++--- lnbits/core/services/extensions.py | 1 + lnbits/core/views/extension_api.py | 18 ++-- tests/api/test_extension_api.py | 58 ++++++++++++- tests/unit/test_models_extensions.py | 9 +- tests/unit/test_services_extensions.py | 114 ++++++++++++++++++++++++- 6 files changed, 273 insertions(+), 24 deletions(-) diff --git a/lnbits/core/models/extensions.py b/lnbits/core/models/extensions.py index 9f143f08f..4cc498b6d 100644 --- a/lnbits/core/models/extensions.py +++ b/lnbits/core/models/extensions.py @@ -28,6 +28,10 @@ from lnbits.task_manager import task_manager from lnbits.utils.cache import cache +class ExtensionArchiveValidationError(ValueError): + pass + + class ExplicitRelease(BaseModel): id: str name: str @@ -66,6 +70,11 @@ class Manifest(BaseModel): categories: dict[str, list[str]] = {} +class ExtensionManifestType(str, Enum): + PYTHON = "python" + WASM = "wasm" + + class GitHubRepoRelease(BaseModel): name: str tag_name: str @@ -356,6 +365,7 @@ class ExtensionRelease(BaseModel): icon: str | None = None details_link: str | None = None extension_type: str | None = None + manifest_type: ExtensionManifestType | None = Field(default=None, exclude=True) permissions: list[ExtensionPermission] = [] paid_features: str | None = None @@ -639,6 +649,41 @@ class InstallableExtension(BaseModel): return config if isinstance(config, dict) else {} + def validate_archive(self, config: Mapping[str, Any]) -> None: + release = self.meta.installed_release if self.meta else None + manifest_type = release.manifest_type if release else None + is_wasm = config.get("extension_type") == "wasm" + + if manifest_type == ExtensionManifestType.PYTHON and is_wasm: + raise ExtensionArchiveValidationError( + f"Python extension manifest cannot install WASM extension '{self.id}'." + ) + if manifest_type == ExtensionManifestType.WASM and not is_wasm: + raise ExtensionArchiveValidationError( + "WASM extension manifest requires extension_type 'wasm' " + f"for extension '{self.id}'." + ) + if not is_wasm: + return + + with zipfile.ZipFile(self.zip_path, "r") as archive: + python_file = next( + ( + item.filename + for item in archive.infolist() + if not item.is_dir() + and PurePosixPath(item.filename).suffix.lower() + in {".py", ".pyc", ".pyo", ".so", ".pyd"} + ), + None, + ) + + if python_file: + raise ExtensionArchiveValidationError( + f"WASM extension '{self.id}' contains forbidden Python file " + f"'{python_file}'." + ) + 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) @@ -947,14 +992,7 @@ class InstallableExtension(BaseModel): ) -> list[InstallableExtension]: extension_list: list[InstallableExtension] = [] - manifest_urls = dict.fromkeys( - [ - *settings.lnbits_extensions_manifests, - *settings.lnbits_wasm_extensions_manifests, - *settings.lnbits_wasm_extensions_manifests, - ] - ) - for url in manifest_urls: + for url, manifest_type in _extension_manifest_sources(): try: manifest = await cls.fetch_manifest(url) @@ -962,6 +1000,8 @@ class InstallableExtension(BaseModel): ext = await InstallableExtension.from_github_release(r) if not ext: continue + if ext.meta and ext.meta.latest_release: + ext.meta.latest_release.manifest_type = manifest_type existing_ext = next( (ee for ee in extension_list if ee.id == r.id), None ) @@ -981,6 +1021,7 @@ class InstallableExtension(BaseModel): for e in manifest.extensions: release = ExtensionRelease.from_explicit_release(url, e) + release.manifest_type = manifest_type existing_ext = next( (ee for ee in extension_list if ee.id == e.id), None ) @@ -1007,12 +1048,9 @@ class InstallableExtension(BaseModel): @classmethod async def get_extension_releases(cls, ext_id: str) -> list[ExtensionRelease]: extension_releases: list[ExtensionRelease] = [] - all_manifests = [ - *settings.lnbits_extensions_manifests, - *settings.lnbits_wasm_extensions_manifests, - settings.lnbits_extensions_builder_manifest_url, - ] - for url in all_manifests: + for url, manifest_type in _extension_manifest_sources( + include_builder=True, deduplicate=False + ): try: manifest = await cls.fetch_manifest(url) for r in manifest.repos: @@ -1021,6 +1059,8 @@ class InstallableExtension(BaseModel): repo_releases = await ExtensionRelease.get_github_releases( r.organisation, r.repository ) + for release in repo_releases: + release.manifest_type = manifest_type extension_releases += repo_releases for e in manifest.extensions: @@ -1035,6 +1075,7 @@ class InstallableExtension(BaseModel): explicit_release.details_link ) explicit_release.apply_config(config) + explicit_release.manifest_type = manifest_type await explicit_release.check_payment_requirements() extension_releases.append(explicit_release) @@ -1170,3 +1211,31 @@ def _archive_config_name(names: list[str]) -> str | None: if len(path.parts) == 2 and path.name == "config.json": return name return None + + +def _extension_manifest_sources( + *, include_builder: bool = False, deduplicate: bool = True +) -> list[tuple[str, ExtensionManifestType]]: + sources = [ + *( + (url, ExtensionManifestType.PYTHON) + for url in settings.lnbits_extensions_manifests + ), + *( + (url, ExtensionManifestType.WASM) + for url in settings.lnbits_wasm_extensions_manifests + ), + ] + if include_builder: + sources.append( + ( + settings.lnbits_extensions_builder_manifest_url, + ExtensionManifestType.PYTHON, + ) + ) + if not deduplicate: + return sources + unique_sources: dict[str, ExtensionManifestType] = {} + for url, manifest_type in sources: + unique_sources.setdefault(url, manifest_type) + return list(unique_sources.items()) diff --git a/lnbits/core/services/extensions.py b/lnbits/core/services/extensions.py index 3e788f1e7..29060e36d 100644 --- a/lnbits/core/services/extensions.py +++ b/lnbits/core/services/extensions.py @@ -210,6 +210,7 @@ async def install_extension( await ext_info.download_archive() extension_config = ext_info.load_archive_config() + ext_info.validate_archive(extension_config) ext_info.permissions = validate_wasm_extension_permissions( ext_info, granted_permissions, diff --git a/lnbits/core/views/extension_api.py b/lnbits/core/views/extension_api.py index 36a98cf6a..8b2c6b1c1 100644 --- a/lnbits/core/views/extension_api.py +++ b/lnbits/core/views/extension_api.py @@ -19,6 +19,7 @@ from lnbits.core.models.extensions import ( CreateExtension, CreateExtensionReview, Extension, + ExtensionArchiveValidationError, ExtensionBackgroundPaymentDestinationPolicy, ExtensionBackgroundPaymentGrant, ExtensionBackgroundPaymentGrantRequest, @@ -129,14 +130,17 @@ async def api_install_extension(data: CreateExtension): logger.warning(exc) etype, _, tb = sys.exc_info() traceback.print_exception(etype, exc, tb) - try: - archive_config = ext_info.load_archive_config() - except ValueError: - archive_config = {} - if archive_config.get("extension_type") == "wasm": - ext_info.clean_wasm_extension_files() + if isinstance(exc, ExtensionArchiveValidationError): + ext_info.zip_path.unlink(missing_ok=True) else: - ext_info.clean_extension_files() + try: + archive_config = ext_info.load_archive_config() + except ValueError: + archive_config = {} + if archive_config.get("extension_type") == "wasm": + ext_info.clean_wasm_extension_files() + else: + ext_info.clean_extension_files() detail = ( str(exc) if isinstance(exc, (AssertionError, ValueError)) diff --git a/tests/api/test_extension_api.py b/tests/api/test_extension_api.py index f27a2cef7..583a7bb9c 100644 --- a/tests/api/test_extension_api.py +++ b/tests/api/test_extension_api.py @@ -23,7 +23,9 @@ from lnbits.core.models.extensions import ( CreateExtensionReview, ExplicitRelease, Extension, + ExtensionArchiveValidationError, ExtensionConfig, + ExtensionManifestType, ExtensionPermission, ExtensionPermissionsUpdate, ExtensionRelease, @@ -175,6 +177,57 @@ async def test_extension_api_install_details_and_release_endpoints(mocker): assert release_info["is_version_compatible"] is True +@pytest.mark.anyio +async def test_extension_api_archive_validation_failure_only_removes_zip( + tmp_path, + settings, + mocker, +): + ext_id = f"ext_{uuid4().hex[:8]}" + release = make_extension_release(ext_id) + create_data = CreateExtension( + ext_id=ext_id, + archive=release.archive, + source_repo=release.source_repo, + version=release.version, + ) + original_data_folder = settings.lnbits_data_folder + clean_python_mock = mocker.patch.object( + InstallableExtension, "clean_extension_files" + ) + clean_wasm_mock = mocker.patch.object( + InstallableExtension, "clean_wasm_extension_files" + ) + mocker.patch.object( + InstallableExtension, + "get_extension_release", + mocker.AsyncMock(return_value=release), + ) + mocker.patch( + "lnbits.core.views.extension_api.install_extension", + mocker.AsyncMock( + side_effect=ExtensionArchiveValidationError("Invalid extension archive.") + ), + ) + + try: + settings.lnbits_data_folder = str(tmp_path / "data") + zip_path = Path(settings.lnbits_data_folder, "zips", f"{ext_id}.zip") + zip_path.parent.mkdir(parents=True) + zip_path.write_bytes(b"archive") + + with pytest.raises(HTTPException) as exc: + await api_install_extension(create_data) + finally: + settings.lnbits_data_folder = original_data_folder + + assert exc.value.status_code == 400 + assert exc.value.detail == "Invalid extension archive." + assert not zip_path.exists() + clean_python_mock.assert_not_called() + clean_wasm_mock.assert_not_called() + + @pytest.mark.anyio async def test_explicit_wasm_release_loads_install_permissions( settings, @@ -227,7 +280,8 @@ async def test_explicit_wasm_release_loads_install_permissions( return Manifest(extensions=[explicit_release, non_wasm_release]) return Manifest() - mocker.patch.object(settings, "lnbits_extensions_manifests", [manifest_url]) + mocker.patch.object(settings, "lnbits_extensions_manifests", []) + mocker.patch.object(settings, "lnbits_wasm_extensions_manifests", [manifest_url]) mocker.patch.object( settings, "lnbits_extensions_builder_manifest_url", @@ -248,6 +302,7 @@ async def test_explicit_wasm_release_loads_install_permissions( assert len(releases) == 1 assert releases[0].extension_type == "wasm" + assert releases[0].manifest_type == ExtensionManifestType.WASM assert releases[0].permissions == config_permissions fetch_config_mock.assert_awaited_once_with(details_link) @@ -257,6 +312,7 @@ async def test_explicit_wasm_release_loads_install_permissions( ) assert len(non_wasm_releases) == 1 assert non_wasm_releases[0].extension_type is None + assert non_wasm_releases[0].manifest_type == ExtensionManifestType.WASM assert non_wasm_releases[0].permissions == [] fetch_config_mock.assert_not_awaited() diff --git a/tests/unit/test_models_extensions.py b/tests/unit/test_models_extensions.py index 904b21f5a..e19c02dfb 100644 --- a/tests/unit/test_models_extensions.py +++ b/tests/unit/test_models_extensions.py @@ -1,7 +1,11 @@ import pytest from pytest_mock.plugin import MockerFixture -from lnbits.core.models.extensions import InstallableExtension, Manifest +from lnbits.core.models.extensions import ( + ExtensionManifestType, + InstallableExtension, + Manifest, +) from lnbits.settings import Settings @@ -42,6 +46,9 @@ async def test_get_installable_extensions_loads_wasm_manifests( extensions = await InstallableExtension._get_installable_extensions() assert [extension.id for extension in extensions] == ["tips"] + assert extensions[0].meta + assert extensions[0].meta.latest_release + assert extensions[0].meta.latest_release.manifest_type == ExtensionManifestType.WASM assert [call.args[0] for call in fetch_manifest.await_args_list] == [ regular_manifest_url, wasm_manifest_url, diff --git a/tests/unit/test_services_extensions.py b/tests/unit/test_services_extensions.py index 96211ee10..85f353785 100644 --- a/tests/unit/test_services_extensions.py +++ b/tests/unit/test_services_extensions.py @@ -14,6 +14,7 @@ from lnbits.core.crud import ( ) from lnbits.core.models.extensions import ( Extension, + ExtensionManifestType, ExtensionPermission, InstallableExtension, ReleasePaymentInfo, @@ -63,6 +64,9 @@ async def test_install_extension_creates_new_extension_and_starts_background_wor ): ext_id = f"ext_{uuid4().hex[:8]}" ext_info = make_installable_extension(ext_id) + assert ext_info.meta + assert ext_info.meta.installed_release + ext_info.meta.installed_release.manifest_type = ExtensionManifestType.PYTHON original_data_folder = settings.lnbits_data_folder original_extensions_path = settings.lnbits_extensions_path download_mock = mocker.patch.object( @@ -162,6 +166,9 @@ async def test_install_wasm_extension_requires_permissions_and_skips_background_ ): ext_id = f"wasm_{uuid4().hex[:8]}" ext_info = make_installable_extension(ext_id) + assert ext_info.meta + assert ext_info.meta.installed_release + ext_info.meta.installed_release.manifest_type = ExtensionManifestType.WASM original_data_folder = settings.lnbits_data_folder original_extensions_path = settings.lnbits_extensions_path original_wasm_extensions_path = settings.lnbits_wasm_extensions_path @@ -216,6 +223,97 @@ async def test_install_wasm_extension_requires_permissions_and_skips_background_ assert not upgrade_dir.exists() +@pytest.mark.parametrize( + ("manifest_type", "error"), + [ + ( + ExtensionManifestType.PYTHON, + "Python extension manifest cannot install WASM extension", + ), + ( + ExtensionManifestType.WASM, + "WASM extension manifest requires extension_type 'wasm'", + ), + ], +) +@pytest.mark.anyio +async def test_install_extension_rejects_archive_from_wrong_manifest_type( + tmp_path, + settings: Settings, + mocker: MockerFixture, + manifest_type: ExtensionManifestType, + error: str, +): + ext_id = f"ext_{uuid4().hex[:8]}" + ext_info = make_installable_extension(ext_id) + assert ext_info.meta + assert ext_info.meta.installed_release + ext_info.meta.installed_release.manifest_type = manifest_type + original_data_folder = settings.lnbits_data_folder + mocker.patch( + "lnbits.core.services.extensions.get_installed_extension", + mocker.AsyncMock(return_value=None), + ) + mocker.patch( + "lnbits.core.services.extensions.check_extensions_limit", + mocker.AsyncMock(), + ) + + try: + settings.lnbits_data_folder = str(tmp_path / "data") + config = ( + _wasm_install_config(ext_id) + if manifest_type == ExtensionManifestType.PYTHON + else {"id": ext_id, "name": ext_id} + ) + _write_extension_archive(ext_info, config) + + with pytest.raises(ValueError, match=error): + await install_extension(ext_info, skip_download=True) + finally: + settings.lnbits_data_folder = original_data_folder + + +@pytest.mark.parametrize( + "python_file", + ["main.py", "cache.pyc", "legacy.PYO", "native.so", "native.PYD"], +) +@pytest.mark.anyio +async def test_install_wasm_extension_rejects_python_files( + tmp_path, + settings: Settings, + mocker: MockerFixture, + python_file: str, +): + ext_id = f"wasm_{uuid4().hex[:8]}" + ext_info = make_installable_extension(ext_id) + assert ext_info.meta + assert ext_info.meta.installed_release + ext_info.meta.installed_release.manifest_type = ExtensionManifestType.WASM + original_data_folder = settings.lnbits_data_folder + mocker.patch( + "lnbits.core.services.extensions.get_installed_extension", + mocker.AsyncMock(return_value=None), + ) + mocker.patch( + "lnbits.core.services.extensions.check_extensions_limit", + mocker.AsyncMock(), + ) + + try: + settings.lnbits_data_folder = str(tmp_path / "data") + _write_extension_archive( + ext_info, + _wasm_install_config(ext_id), + extra_files=[f"nested/{python_file}"], + ) + + with pytest.raises(ValueError, match="contains forbidden Python file"): + await install_extension(ext_info, skip_download=True) + finally: + settings.lnbits_data_folder = original_data_folder + + @pytest.mark.anyio async def test_uninstall_activate_and_deactivate_extensions( tmp_path, settings: Settings, mocker: MockerFixture @@ -500,12 +598,26 @@ def _reset_wasm_invocation_state(): def _write_wasm_extension_archive( ext_info: InstallableExtension, config: dict, +) -> None: + _write_extension_archive( + ext_info, + config, + extra_files=[config["wasm"]["module"]], + ) + + +def _write_extension_archive( + ext_info: InstallableExtension, + config: dict, + *, + extra_files: list[str] | None = None, ) -> None: ext_info.zip_path.parent.mkdir(parents=True, exist_ok=True) root = f"{ext_info.id}-{ext_info.version}" with zipfile.ZipFile(ext_info.zip_path, "w") as archive: archive.writestr(f"{root}/config.json", json.dumps(config)) - archive.writestr(f"{root}/{config['wasm']['module']}", b"\0asm") + for filename in extra_files or []: + archive.writestr(f"{root}/{filename}", b"\0asm") def _wasm_install_config(ext_id: str) -> dict: