mirror of
https://github.com/lnbits/lnbits.git
synced 2026-08-13 12:42:47 +02:00
fix: check extension type when installing (#4101)
This commit is contained in:
Vlad Stan
GitHub
parent
7e2e116d46
commit
2b5a513693
6 changed files with 273 additions and 24 deletions
|
|
@ -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())
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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))
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue