diff --git a/geonode/assets/tests.py b/geonode/assets/tests.py index 8ece6c35b34..ce707636fde 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,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")) diff --git a/geonode/assets/views.py b/geonode/assets/views.py index 11bebea3013..ec5826a919a 100644 --- a/geonode/assets/views.py +++ b/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, 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