From e2c1becd757e8ca3a5cbc49ace8f8b26f99921a3 Mon Sep 17 00:00:00 2001 From: sijandh35 Date: Fri, 18 Sep 2026 11:13:09 +0000 Subject: [PATCH 1/2] Improvements in Asset --- geonode/assets/tests.py | 51 +++++++++++++++++++++++++++++++++++++++++ geonode/assets/views.py | 35 +++++++++++++++++++++++++++- 2 files changed, 85 insertions(+), 1 deletion(-) diff --git a/geonode/assets/tests.py b/geonode/assets/tests.py index 8ece6c35b34..3f123ed6a41 100644 --- a/geonode/assets/tests.py +++ b/geonode/assets/tests.py @@ -40,6 +40,7 @@ from geonode.assets.utils import create_asset, create_asset_and_link, unlink_asset from geonode.base.models import ResourceBase, Link from geonode.security.registry import permissions_registry +from rest_framework import status logger = logging.getLogger(__name__) @@ -631,3 +632,53 @@ def test_delete_asset_and_link(self): self.assertFalse(Asset.objects.filter(pk=asset_pk).exists()) self.assertFalse(Link.objects.filter(pk=self.link1.pk).exists()) self.assertFalse(os.path.exists(asset_file_path)) + + +class AssetViewSetPermissionsTests(GeoNodeBaseTestSupport): + def setUp(self): + super().setUp() + self.admin = get_user_model().objects.get(username="admin") + self.user = get_user_model().objects.create_user(username="asset_user", password="password") + self.resource = ResourceBase.objects.create(owner=self.admin, title="Private resource") + self.asset, self.link = create_asset_and_link( + self.resource, + self.admin, + [ONE_JSON], + title="Private asset", + ) + + def test_anonymous_cannot_retrieve_private_linked_asset(self): + response = self.client.get(reverse("assets-detail", kwargs={"pk": self.asset.pk})) + + self.assertIn(response.status_code, [status.HTTP_401_UNAUTHORIZED, status.HTTP_403_FORBIDDEN]) + + def test_user_with_view_resourcebase_can_retrieve_linked_asset(self): + self.resource.set_permissions({"users": {self.user.username: ["view_resourcebase"]}, "groups": {}}) + + self.client.force_login(self.user) + response = self.client.get(reverse("assets-detail", kwargs={"pk": self.asset.pk})) + + self.assertEqual(response.status_code, status.HTTP_200_OK) + + def test_user_without_change_resourcebase_cannot_patch_linked_asset(self): + self.resource.set_permissions({"users": {self.user.username: ["view_resourcebase"]}, "groups": {}}) + + self.client.force_login(self.user) + response = self.client.patch( + reverse("assets-detail", kwargs={"pk": self.asset.pk}), + data=json.dumps({"title": "SHOULD-NOT-WORK"}), + content_type="application/json", + ) + + self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) + self.asset.refresh_from_db() + self.assertEqual(self.asset.title, "Private asset") + + def test_user_without_change_resourcebase_cannot_delete_linked_asset(self): + self.resource.set_permissions({"users": {self.user.username: ["view_resourcebase"]}, "groups": {}}) + + self.client.force_login(self.user) + response = self.client.delete(reverse("assets-detail", kwargs={"pk": self.asset.pk})) + + self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) + self.assertTrue(Asset.objects.filter(pk=self.asset.pk).exists()) diff --git a/geonode/assets/views.py b/geonode/assets/views.py index 11bebea3013..06d0f466ce0 100644 --- a/geonode/assets/views.py +++ b/geonode/assets/views.py @@ -36,16 +36,49 @@ DynamicSearchFilter, ) from geonode.base.api.pagination import GeoNodeApiPagination +from geonode.base.models import ResourceBase +from geonode.security.registry import permissions_registry +from geonode.security.utils import get_visible_resources +from rest_framework import permissions logger = logging.getLogger(__name__) +class UserHasAssetPerms(permissions.BasePermission): + def has_object_permission(self, request, view, obj): + user = request.user + + if user and user.is_authenticated and user.is_superuser: + return True + + resources = ResourceBase.objects.filter(link__asset=obj).distinct() + has_linked_resources = resources.exists() + + if request.method in permissions.SAFE_METHODS: + if has_linked_resources: + return get_visible_resources(queryset=resources, user=user).exists() + + return user and user.is_authenticated and obj.owner_id == user.id + + if not user or not user.is_authenticated: + return False + + if not has_linked_resources: + return obj.owner_id == user.id + + return all( + "change_resourcebase" in permissions_registry.get_perms(instance=resource, user=user) + for resource in resources + ) + + class AssetViewSet(DynamicModelViewSet): """ API endpoint that allows Assets to be viewed or edited. """ - permission_classes = [IsAuthenticatedOrReadOnly] + permission_classes = [IsAuthenticatedOrReadOnly, UserHasAssetPerms] + http_method_names = ["get", "put", "patch", "delete"] filter_backends = [ DynamicFilterBackend, DynamicSortingFilter, From fed2eb04316416c0e524065314377abe5b5ab464 Mon Sep 17 00:00:00 2001 From: mattiagiupponi Date: Fri, 25 Sep 2026 12:07:37 +0200 Subject: [PATCH 2/2] Move logic into permissions registry --- geonode/assets/tests.py | 51 ++++++++++++++++++++++++++++++++++++ geonode/assets/views.py | 32 +++++----------------- geonode/security/registry.py | 42 +++++++++++++++++++++++++++++ 3 files changed, 99 insertions(+), 26 deletions(-) diff --git a/geonode/assets/tests.py b/geonode/assets/tests.py index 3f123ed6a41..ce707636fde 100644 --- a/geonode/assets/tests.py +++ b/geonode/assets/tests.py @@ -682,3 +682,54 @@ def test_user_without_change_resourcebase_cannot_delete_linked_asset(self): self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) self.assertTrue(Asset.objects.filter(pk=self.asset.pk).exists()) + + +class PermissionsRegistryAssetPermTests(GeoNodeBaseTestSupport): + """ + The Asset permission logic lives in the permissions registry now (not in the + DRF permission class), so it's tested directly here. + """ + + def setUp(self): + super().setUp() + self.admin = get_user_model().objects.get(username="admin") + self.owner = get_user_model().objects.create_user(username="asset_owner", password="password") + self.other_user = get_user_model().objects.create_user(username="other_user", password="password") + + def test_superuser_always_allowed(self): + asset, _ = create_asset_and_link( + ResourceBase.objects.create(owner=self.owner, title="r1"), self.owner, [ONE_JSON] + ) + self.assertTrue(permissions_registry.user_has_asset_perm(self.admin, asset, method="GET")) + self.assertTrue(permissions_registry.user_has_asset_perm(self.admin, asset, method="DELETE")) + + def test_asset_without_linked_resource_falls_back_to_owner(self): + asset = LocalAsset.objects.create(title="orphan", owner=self.owner, type="test") + + self.assertTrue(permissions_registry.user_has_asset_perm(self.owner, asset, method="GET")) + self.assertTrue(permissions_registry.user_has_asset_perm(self.owner, asset, method="PATCH")) + self.assertFalse(permissions_registry.user_has_asset_perm(self.other_user, asset, method="GET")) + + def test_write_requires_change_resourcebase_on_every_linked_resource(self): + resource = ResourceBase.objects.create(owner=self.owner, title="r2") + asset, _ = create_asset_and_link(resource, self.owner, [ONE_JSON]) + + # only view perm granted -> read ok, write denied + resource.set_permissions({"users": {self.other_user.username: ["view_resourcebase"]}, "groups": {}}) + self.assertTrue(permissions_registry.user_has_asset_perm(self.other_user, asset, method="GET")) + self.assertFalse(permissions_registry.user_has_asset_perm(self.other_user, asset, method="PATCH")) + + # bump to change perm -> write allowed too + resource.set_permissions( + {"users": {self.other_user.username: ["view_resourcebase", "change_resourcebase"]}, "groups": {}} + ) + self.assertTrue(permissions_registry.user_has_asset_perm(self.other_user, asset, method="PATCH")) + + def test_anonymous_denied_on_write(self): + resource = ResourceBase.objects.create(owner=self.owner, title="r3") + asset, _ = create_asset_and_link(resource, self.owner, [ONE_JSON]) + from guardian.shortcuts import get_anonymous_user + + anonymous = get_anonymous_user() + + self.assertFalse(permissions_registry.user_has_asset_perm(anonymous, asset, method="DELETE")) diff --git a/geonode/assets/views.py b/geonode/assets/views.py index 06d0f466ce0..ec5826a919a 100644 --- a/geonode/assets/views.py +++ b/geonode/assets/views.py @@ -36,40 +36,20 @@ DynamicSearchFilter, ) from geonode.base.api.pagination import GeoNodeApiPagination -from geonode.base.models import ResourceBase from geonode.security.registry import permissions_registry -from geonode.security.utils import get_visible_resources from rest_framework import permissions logger = logging.getLogger(__name__) class UserHasAssetPerms(permissions.BasePermission): - def has_object_permission(self, request, view, obj): - user = request.user - - if user and user.is_authenticated and user.is_superuser: - return True - - resources = ResourceBase.objects.filter(link__asset=obj).distinct() - has_linked_resources = resources.exists() - - if request.method in permissions.SAFE_METHODS: - if has_linked_resources: - return get_visible_resources(queryset=resources, user=user).exists() - - return user and user.is_authenticated and obj.owner_id == user.id - - if not user or not user.is_authenticated: - return False - - if not has_linked_resources: - return obj.owner_id == user.id + """ + Thin DRF adapter: the actual decision is delegated to the permissions registry, + which knows how to resolve an Asset's perms through its linked ResourceBase(s). + """ - return all( - "change_resourcebase" in permissions_registry.get_perms(instance=resource, user=user) - for resource in resources - ) + def has_object_permission(self, request, view, obj): + return permissions_registry.user_has_asset_perm(request.user, obj, method=request.method) class AssetViewSet(DynamicModelViewSet): diff --git a/geonode/security/registry.py b/geonode/security/registry.py index c8c8aa10290..35c5c6ce22b 100644 --- a/geonode/security/registry.py +++ b/geonode/security/registry.py @@ -86,6 +86,48 @@ def user_has_perm(self, user, instance=None, perm="", include_virtual=False): return perm in resolved_perms + def user_has_asset_perm(self, user, asset, method="GET"): + """ + Returns True if the user is allowed to access/edit an Asset. + + Assets are not a ResourceBase themselves, so the permission is resolved through the + ResourceBase(s) they are linked to (via `Link`): + - superusers are always allowed + - an asset with no linked resource yet (e.g. mid-upload) falls back to ownership + - safe/read methods require the asset to be visible through at least one linked resource + - unsafe/write methods require `change_resourcebase` on every linked resource + """ + from django.conf import settings + from geonode.base.models import ResourceBase + + if not asset: + return False + + if user and user.is_authenticated and user.is_superuser: + return True + + resources = ResourceBase.objects.filter(link__asset=asset).distinct() + has_linked_resources = resources.exists() + + if method in ("GET", "HEAD", "OPTIONS"): + if has_linked_resources: + return self.get_visible_resources( + queryset=resources, + user=user, + admin_approval_required=settings.ADMIN_MODERATE_UPLOADS, + unpublished_not_visible=settings.RESOURCE_PUBLISHING, + private_groups_not_visibile=settings.GROUP_PRIVATE_RESOURCES, + ).exists() + return bool(user and user.is_authenticated and asset.owner_id == user.id) + + if not user or not user.is_authenticated: + return False + + if not has_linked_resources: + return asset.owner_id == user.id + + return all("change_resourcebase" in self.get_perms(instance=resource, user=user) for resource in resources) + def user_can_feature(self, user, resource): """ Utility method to check if the user can set a resource as "featured" in the metadata