diff --git a/lnbits/core/models/extensions.py b/lnbits/core/models/extensions.py index c0cfa3260..ab6b29cf1 100644 --- a/lnbits/core/models/extensions.py +++ b/lnbits/core/models/extensions.py @@ -43,6 +43,8 @@ class ExplicitRelease(BaseModel): details_link: str | None paid_features: str | None pay_link: str | None + admin_only: bool = False + super_user_only: bool = False def is_version_compatible(self): return is_lnbits_version_ok(self.min_lnbits_version, self.max_lnbits_version) @@ -83,6 +85,8 @@ class ExtensionConfig(BaseModel): warning: str | None = "" min_lnbits_version: str | None max_lnbits_version: str | None + admin_only: bool = False + super_user_only: bool = False def is_version_compatible(self) -> bool: return is_lnbits_version_ok(self.min_lnbits_version, self.max_lnbits_version) @@ -195,6 +199,8 @@ class ExtensionRelease(BaseModel): cost_sats: int | None = None paid_sats: int | None = 0 payment_hash: str | None = None + admin_only: bool = False + super_user_only: bool = False @property def archive_url(self) -> str: @@ -263,6 +269,8 @@ class ExtensionRelease(BaseModel): paid_features=e.paid_features, repo=e.repo, icon=e.icon, + admin_only=e.admin_only, + super_user_only=e.super_user_only, ) @classmethod @@ -289,6 +297,8 @@ class ExtensionRelease(BaseModel): release.min_lnbits_version = config.min_lnbits_version release.max_lnbits_version = config.max_lnbits_version release.is_version_compatible = config.is_version_compatible() + release.admin_only = config.admin_only + release.super_user_only = config.super_user_only release.icon = icon_to_github_url(f"{org}/{repo}", config.tile) @@ -336,6 +346,8 @@ class ExtensionMeta(BaseModel): paid_features: str | None = None has_paid_release: bool = False has_free_release: bool = False + admin_only: bool = False + super_user_only: bool = False class InstallableExtension(BaseModel): @@ -399,6 +411,16 @@ class InstallableExtension(BaseModel): return False return self.meta.pay_to_enable.required is True + @property + def is_admin_only(self) -> bool: + return settings.is_admin_extension(self.id) or bool( + self.meta and self.meta.admin_only + ) + + @property + def is_super_user_only(self) -> bool: + return bool(self.meta and self.meta.super_user_only) + async def download_archive(self): logger.info(f"Downloading extension {self.name} ({self.installed_version}).") ext_zip_file = self.zip_path @@ -454,6 +476,16 @@ class InstallableExtension(BaseModel): self.name = config_json.get("name") self.short_description = config_json.get("short_description") + if self.meta: + self.meta.admin_only = config_json.get("admin_only", False) + self.meta.super_user_only = config_json.get("super_user_only", False) + if self.meta.installed_release: + self.meta.installed_release.admin_only = config_json.get( + "admin_only", False + ) + self.meta.installed_release.super_user_only = config_json.get( + "super_user_only", False + ) if ( self.meta @@ -496,6 +528,10 @@ class InstallableExtension(BaseModel): return if not release.is_version_compatible: return + if not self.meta: + self.meta = ExtensionMeta() + self.meta.admin_only = release.admin_only + self.meta.super_user_only = release.super_user_only if not self.meta or not self.meta.latest_release: meta = self.meta or ExtensionMeta() meta.latest_release = release @@ -569,6 +605,8 @@ class InstallableExtension(BaseModel): config.tile, ), meta=ExtensionMeta( + admin_only=config.admin_only, + super_user_only=config.super_user_only, latest_release=ExtensionRelease.from_github_release( source_repo, latest_release ), @@ -580,7 +618,12 @@ class InstallableExtension(BaseModel): @classmethod def from_explicit_release(cls, e: ExplicitRelease) -> InstallableExtension: - meta = ExtensionMeta(archive=e.archive, dependencies=e.dependencies) + meta = ExtensionMeta( + archive=e.archive, + dependencies=e.dependencies, + admin_only=e.admin_only, + super_user_only=e.super_user_only, + ) return InstallableExtension( id=e.id, name=e.name, @@ -610,6 +653,8 @@ class InstallableExtension(BaseModel): short_description=config_json.get("short_description"), icon=config_json.get("tile"), meta=ExtensionMeta( + admin_only=config_json.get("admin_only", False), + super_user_only=config_json.get("super_user_only", False), installed_release=ExtensionRelease( name=ext_id, version=version, @@ -617,6 +662,10 @@ class InstallableExtension(BaseModel): source_repo=f"{conf_path}", min_lnbits_version=config_json.get("min_lnbits_version"), max_lnbits_version=config_json.get("max_lnbits_version"), + admin_only=config_json.get("admin_only", False), + super_user_only=config_json.get( + "super_user_only", False + ), ) ), ) diff --git a/lnbits/core/views/extension_api.py b/lnbits/core/views/extension_api.py index 46bf3b080..603104b3c 100644 --- a/lnbits/core/views/extension_api.py +++ b/lnbits/core/views/extension_api.py @@ -79,7 +79,11 @@ async def api_install_extension(data: CreateExtension): raise HTTPException(HTTPStatus.BAD_REQUEST, "Incompatible extension version.") release.payment_hash = data.payment_hash - ext_meta = ExtensionMeta(installed_release=release) + ext_meta = ExtensionMeta( + installed_release=release, + admin_only=release.admin_only, + super_user_only=release.super_user_only, + ) ext_info = InstallableExtension( id=data.ext_id, name=data.ext_id, @@ -191,6 +195,14 @@ async def api_enable_extension( raise ValueError(f"Extension '{ext_id}' is not installed.") if not ext.active: raise ValueError(f"Extension '{ext_id}' is not activated.") + if ext.is_super_user_only and not settings.is_super_user(account_id.id): + raise HTTPException( + HTTPStatus.FORBIDDEN, f"User not authorized for extension '{ext_id}'." + ) + if ext.is_admin_only and not settings.is_admin_user(account_id.id): + raise HTTPException( + HTTPStatus.FORBIDDEN, f"User not authorized for extension '{ext_id}'." + ) user_ext = await get_user_extension(account_id.id, ext_id) if not user_ext: @@ -464,6 +476,8 @@ async def get_extension_release(org: str, repo: str, tag_name: str): "min_lnbits_version": config.min_lnbits_version, "is_version_compatible": config.is_version_compatible(), "warning": config.warning, + "admin_only": config.admin_only, + "super_user_only": config.super_user_only, } except Exception as exc: raise HTTPException( @@ -573,7 +587,8 @@ async def extensions(account_id: AccountId = Depends(check_account_id_exists)): (True for version in db_versions if version.db == ext.id), False ), "isAvailable": ext.id in all_ext_ids, - "isAdminOnly": ext.id in settings.lnbits_admin_extensions, + "isAdminOnly": ext.is_admin_only, + "isSuperUserOnly": ext.is_super_user_only, "isActive": ext.id not in inactive_extensions, "latestRelease": ( dict(ext.meta.latest_release) diff --git a/lnbits/decorators.py b/lnbits/decorators.py index 1673b860f..91a933423 100644 --- a/lnbits/decorators.py +++ b/lnbits/decorators.py @@ -14,6 +14,7 @@ from lnbits.core.crud import ( get_account, get_account_by_email, get_account_by_username, + get_installed_extension, get_user_active_extensions_ids, get_user_from_account, get_wallet_for_key, @@ -424,7 +425,19 @@ async def check_user_extension_access( Check if the user has access to a particular extension. Raises HTTP Forbidden if the user is not allowed. """ - if settings.is_admin_extension(ext_id) and not settings.is_admin_user(user_id): + ext = await get_installed_extension(ext_id, conn=conn) + is_admin_only = settings.is_admin_extension(ext_id) or bool( + ext and ext.meta and ext.meta.admin_only + ) + is_super_user_only = bool(ext and ext.meta and ext.meta.super_user_only) + + if is_super_user_only and not settings.is_super_user(user_id): + return SimpleStatus( + success=False, + message=f"User not authorized for extension '{ext_id}'.", + ) + + if is_admin_only and not settings.is_admin_user(user_id): return SimpleStatus( success=False, message=f"User not authorized for extension '{ext_id}'." ) diff --git a/lnbits/templates/pages/extensions.vue b/lnbits/templates/pages/extensions.vue index 0bfcbef83..2eee2feec 100644 --- a/lnbits/templates/pages/extensions.vue +++ b/lnbits/templates/pages/extensions.vue @@ -307,7 +307,13 @@ :label="$t('disable')" > + Super User Only + + + @@ -316,7 +322,9 @@ v-else-if=" extension.isInstalled && extension.isActive && - !g.user.extensions.includes(extension.id) + !g.user.extensions.includes(extension.id) && + (!extension.isAdminOnly || g.user.admin) && + (!extension.isSuperUserOnly || g.user.super_user) " flat color="primary" diff --git a/tests/api/test_extension_api.py b/tests/api/test_extension_api.py index c3d64ffaa..e9aa3521b 100644 --- a/tests/api/test_extension_api.py +++ b/tests/api/test_extension_api.py @@ -152,6 +152,8 @@ async def test_extension_api_install_details_and_release_endpoints(mocker): short_description="Config", min_lnbits_version="0.1.0", max_lnbits_version=None, + admin_only=True, + super_user_only=True, ) mocker.patch.object( ExtensionConfig, @@ -160,6 +162,8 @@ async def test_extension_api_install_details_and_release_endpoints(mocker): ) release_info = await get_extension_release("org", ext_id, "v1.0.0") assert release_info["is_version_compatible"] is True + assert release_info["admin_only"] is True + assert release_info["super_user_only"] is True @pytest.mark.anyio @@ -258,6 +262,8 @@ async def test_extension_api_pay_to_enable_and_catalog_views(mocker, admin_user) catalog_entry = make_installable_extension( ext_id, pay_to_enable=PayToEnableInfo(required=True, amount=21, wallet=admin_wallet.id), + admin_only=True, + super_user_only=True, ) mocker.patch.object( InstallableExtension, @@ -267,6 +273,8 @@ async def test_extension_api_pay_to_enable_and_catalog_views(mocker, admin_user) catalog = await extensions(AccountId(id=regular_user.id)) catalog_item = next(item for item in catalog if item["id"] == ext_id) assert catalog_item["payToEnable"]["wallet"] is None + assert catalog_item["isAdminOnly"] is True + assert catalog_item["isSuperUserOnly"] is True @pytest.mark.anyio @@ -428,3 +436,31 @@ async def test_extension_api_review_endpoints(mocker): CreateExtensionReview(tag=ext_id, name="Alice", rating=900, comment="Great") ) assert payment_request.payment_hash.startswith("hash_") + + +@pytest.mark.anyio +async def test_extension_api_enable_rejects_admin_and_super_only_extensions( + admin_user, +): + regular_user = await create_user_account( + Account( + id=uuid4().hex, + username=f"user_{uuid4().hex[:8]}", + email=f"user_{uuid4().hex[:8]}@lnbits.com", + ) + ) + admin_only_ext = f"admin_{uuid4().hex[:8]}" + super_only_ext = f"super_{uuid4().hex[:8]}" + + await create_installed_extension( + make_installable_extension(admin_only_ext, admin_only=True) + ) + await create_installed_extension( + make_installable_extension(super_only_ext, super_user_only=True) + ) + + with pytest.raises(HTTPException, match="User not authorized"): + await api_enable_extension(admin_only_ext, AccountId(id=regular_user.id)) + + with pytest.raises(HTTPException, match="User not authorized"): + await api_enable_extension(super_only_ext, AccountId(id=admin_user.id)) diff --git a/tests/helpers.py b/tests/helpers.py index 393fcddb5..524ab2b6c 100644 --- a/tests/helpers.py +++ b/tests/helpers.py @@ -141,9 +141,13 @@ def make_installable_extension( pay_to_enable: PayToEnableInfo | None = None, dependencies: list[str] | None = None, payments: list[ReleasePaymentInfo] | None = None, + admin_only: bool = False, + super_user_only: bool = False, ) -> InstallableExtension: release = make_extension_release(ext_id, version) release.is_version_compatible = compatible + release.admin_only = admin_only + release.super_user_only = super_user_only return InstallableExtension( id=ext_id, name=f"Extension {ext_id}", @@ -156,6 +160,8 @@ def make_installable_extension( pay_to_enable=pay_to_enable, dependencies=dependencies or [], payments=payments or [], + admin_only=admin_only, + super_user_only=super_user_only, ), ) diff --git a/tests/unit/test_decorators.py b/tests/unit/test_decorators.py index 304aebabf..08f3c9c4d 100644 --- a/tests/unit/test_decorators.py +++ b/tests/unit/test_decorators.py @@ -8,6 +8,7 @@ from fastapi.exceptions import HTTPException from httpx import AsyncClient from pydantic.types import UUID4 +from lnbits.core.crud.extensions import create_installed_extension from lnbits.core.crud.users import delete_account from lnbits.core.models import User from lnbits.core.models.users import AccessTokenPayload @@ -18,10 +19,12 @@ from lnbits.decorators import ( check_extension_builder, check_first_install, check_user_exists, + check_user_extension_access, optional_user_id, ) from lnbits.helpers import create_access_token from lnbits.settings import AuthMethods, Settings, settings +from tests.helpers import make_installable_extension @pytest.mark.anyio @@ -225,3 +228,35 @@ async def test_check_extension_builder_requires_admin_when_disabled_for_users( admin_user = user_alan.copy(deep=True) admin_user.admin = True await check_extension_builder(admin_user) + + +@pytest.mark.anyio +async def test_check_user_extension_access_honors_extension_metadata( + settings: Settings, user_alan: User, admin_user: User +): + admin_only_ext = f"admin_{uuid4().hex[:8]}" + super_only_ext = f"super_{uuid4().hex[:8]}" + + await create_installed_extension( + make_installable_extension(admin_only_ext, admin_only=True) + ) + await create_installed_extension( + make_installable_extension(super_only_ext, super_user_only=True) + ) + + regular_status = await check_user_extension_access(user_alan.id, admin_only_ext) + assert regular_status.success is False + + admin_status = await check_user_extension_access(admin_user.id, admin_only_ext) + assert admin_status.success is True + + admin_super_status = await check_user_extension_access(admin_user.id, super_only_ext) + assert admin_super_status.success is False + + previous_super_user = settings.super_user + settings.super_user = admin_user.id + try: + super_status = await check_user_extension_access(admin_user.id, super_only_ext) + assert super_status.success is True + finally: + settings.super_user = previous_super_user