Improvements in Asset

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,104 @@ 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())
+
+
+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"))

geonode/assets/views.py

@@ -36,16 +36,29 @@
     DynamicSearchFilter,
 )
 from geonode.base.api.pagination import GeoNodeApiPagination
+from geonode.security.registry import permissions_registry
+from rest_framework import permissions
 
 logger = logging.getLogger(__name__)
 
 
+class UserHasAssetPerms(permissions.BasePermission):
+    """
+    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).
+    """
+
+    def has_object_permission(self, request, view, obj):
+        return permissions_registry.user_has_asset_perm(request.user, obj, method=request.method)
+
+
 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,

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