diff --git a/dje/api.py b/dje/api.py index 7a4bc0bf..c6a4b948 100644 --- a/dje/api.py +++ b/dje/api.py @@ -411,18 +411,19 @@ def get_queryset(self): # Support for `many=True` serializer_field = self.parent if isinstance(self.parent, ManyRelatedField) else self - - model_class = serializer_field.parent.Meta.model - field_name = serializer_field.source - field = model_class._meta.get_field(field_name) user = self.context["request"].user - if not queryset: - manager = field.related_model.objects - if is_secured(manager): - queryset = manager.get_queryset(user=user) - else: - queryset = manager.all() + if not queryset or self.scope_content_type: + model_class = serializer_field.parent.Meta.model + field_name = serializer_field.source + field = model_class._meta.get_field(field_name) + + if not queryset: + manager = field.related_model.objects + if is_secured(manager): + queryset = manager.get_queryset(user=user) + else: + queryset = manager.all() queryset = queryset.scope(user.dataspace) diff --git a/dje/api_permissions.py b/dje/api_permissions.py new file mode 100644 index 00000000..5254b6a2 --- /dev/null +++ b/dje/api_permissions.py @@ -0,0 +1,171 @@ +# +# Copyright (c) nexB Inc. and others. All rights reserved. +# DejaCode is a trademark of nexB Inc. +# SPDX-License-Identifier: AGPL-3.0-only +# See https://github.com/aboutcode-org/dejacode for support or download. +# See https://aboutcode.org for more information about AboutCode FOSS projects. +# + + +from django.contrib.auth import get_user_model +from django.contrib.auth.models import Group +from django.core.exceptions import ObjectDoesNotExist + +from guardian.shortcuts import assign_perm +from guardian.shortcuts import get_groups_with_perms +from guardian.shortcuts import get_users_with_perms +from guardian.shortcuts import remove_perm +from rest_framework import permissions +from rest_framework import serializers +from rest_framework import status +from rest_framework.decorators import action +from rest_framework.response import Response + +from dje.api import DataspacedSlugRelatedField + +User = get_user_model() + + +class CanManageObjectPermissions(permissions.BasePermission): + """ + Allows managing object-level permissions if the user is: + - a superuser, or + - the object's owner (configurable via ``owner_field`` on the View). + """ + + owner_field = "created_by" + + def has_object_permission(self, request, view, obj): + user = request.user + if not user.is_authenticated: + return False + + if user.is_superuser: + return True + + owner_field = getattr(view, "owner_field", self.owner_field) + owner = getattr(obj, owner_field, None) + return owner == user + + +class ObjectPermissionSerializer(serializers.Serializer): + """ + Validates POST/DELETE input for the manage_permissions action. + Exactly one of ``user`` or ``group`` must be provided alongside ``permissions``. + """ + + user = DataspacedSlugRelatedField( + queryset=User.objects.all(), + slug_field="username", + required=False, + allow_null=True, + default=None, + ) + group = serializers.SlugRelatedField( + queryset=Group.objects.all(), + slug_field="name", + required=False, + allow_null=True, + default=None, + ) + permissions = serializers.ListField(child=serializers.CharField(), allow_empty=False) + + class Meta: + fields = ("user", "group", "permissions") + + def validate(self, data): + has_user = data.get("user") is not None + has_group = data.get("group") is not None + if not has_user and not has_group: + raise serializers.ValidationError("Either 'user' or 'group' must be provided.") + if has_user and has_group: + raise serializers.ValidationError( + "Only one of 'user' or 'group' can be provided, not both." + ) + return data + + +class ObjectPermissionsMixin: + """ + Mixin that adds a ``/permissions/`` endpoint for any object-level ViewSet. + Supports GET (list), POST (assign), and DELETE (remove) operations for + both individual users and groups. + + GET /api/{model}/{uuid}/permissions/ + POST /api/{model}/{uuid}/permissions/ + DELETE /api/{model}/{uuid}/permissions/ + """ + + @action( + detail=True, + methods=["get", "post", "delete"], + url_path="permissions", + serializer_class=ObjectPermissionSerializer, + permission_classes=[permissions.IsAuthenticated, CanManageObjectPermissions], + ) + def manage_permissions(self, request, *args, **kwargs): + """ + Manage object-level permissions for this object. + + - GET: List users and groups with their permissions. + - POST: Assign permissions. Provide ``user`` or ``group`` and ``permissions`` list. + - DELETE: Remove permissions. Provide ``user`` or ``group`` and ``permissions`` list. + """ + obj = self.get_object() + serializer_context = {**self.get_serializer_context(), "object": obj} + + if request.method == "GET": + users_with_perms = get_users_with_perms(obj, attach_perms=True) + groups_with_perms = get_groups_with_perms(obj, attach_perms=True) + data = { + "users": [ + { + "dataspace": user.dataspace.name, + "username": user.get_username(), + "object_permissions": list(perms), + } + for user, perms in users_with_perms.items() + ], + "groups": [ + { + "name": group.name, + "object_permissions": list(perms), + } + for group, perms in groups_with_perms.items() + ], + } + return Response(data, status=status.HTTP_200_OK) + + # POST or DELETE + serializer = self.get_serializer(data=request.data, context=serializer_context) + if not serializer.is_valid(): + return Response({"errors": serializer.errors}, status=status.HTTP_400_BAD_REQUEST) + + target = serializer.validated_data["user"] or serializer.validated_data["group"] + perms = serializer.validated_data["permissions"] + + if request.method == "POST": + errors = [] + for perm in perms: + try: + assign_perm(perm, target, obj) + except ObjectDoesNotExist: + errors.append(f"Cannot assign permission '{perm}' due to an internal error.") + + if errors: + return Response({"errors": errors}, status=status.HTTP_400_BAD_REQUEST) + + return Response({"status": "permissions assigned"}, status=status.HTTP_200_OK) + + if request.method == "DELETE": + errors = [] + for perm in perms: + try: + remove_perm(perm, target, obj) + except ObjectDoesNotExist: + errors.append(f"Cannot remove permission '{perm}' due to an internal error.") + + if errors: + return Response({"errors": errors}, status=status.HTTP_400_BAD_REQUEST) + + return Response({"status": "permissions removed"}, status=status.HTTP_200_OK) diff --git a/docs/howto-5-product-object-permissions.rst b/docs/howto-5-product-object-permissions.rst index 3bf017ec..236315b6 100644 --- a/docs/howto-5-product-object-permissions.rst +++ b/docs/howto-5-product-object-permissions.rst @@ -92,3 +92,156 @@ examples and not recommendations. You have now made the Product visible, and optionally editable, by DejaCode Users that are not superusers. + +4. Manage Product Object Permissions via the REST API +----------------------------------------------------- + +Product object permissions can also be managed programmatically through the REST API. +This is especially useful for CI/CD pipelines that create Product versions automatically +and need to assign permissions without manual intervention. + +The endpoint is available at:: + + /api/v2/products/{uuid}/permissions/ + +**Authentication** + +All requests require authentication. The examples below use an API key passed via +the ``Authorization`` header:: + + Authorization: Token + +**Available permissions** + +The following permission codenames can be assigned to users or groups: + +- ``view_product`` -- allows viewing the product +- ``change_product`` -- allows editing the product +- ``delete_product`` -- allows deleting the product + +**Finding the Product UUID** + +Retrieve the UUID from the product list endpoint:: + + GET /api/v2/products/?name=MyApp&version=2.0 + +The ``uuid`` field is included in each product entry of the response. + +4.1 List current permissions +^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ + +Retrieve all users and groups that currently have permissions on a product:: + + GET /api/v2/products/{uuid}/permissions/ + +Response:: + + { + "users": [ + { + "dataspace": "nexB", + "username": "alice", + "object_permissions": ["view_product", "change_product"] + } + ], + "groups": [ + { + "name": "backend-team", + "object_permissions": ["view_product"] + } + ] + } + +4.2 Assign permissions to a user +^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ + +Provide ``user`` (username) and a ``permissions`` list:: + + POST /api/v2/products/{uuid}/permissions/ + Content-Type: application/json + + { + "user": "alice", + "permissions": ["view_product", "change_product"] + } + +Successful response:: + + {"status": "permissions assigned"} + +4.3 Assign permissions to a group +^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ + +Use ``group`` (group name) instead of ``user``. All members of the group will +inherit the assigned permissions:: + + POST /api/v2/products/{uuid}/permissions/ + Content-Type: application/json + + { + "group": "backend-team", + "permissions": ["view_product"] + } + +This is the recommended approach when multiple users need access to the same set +of products. Manage group membership via the DejaCode admin, then assign the group +to each product once. + +4.4 Remove permissions from a user or group +^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ + +Use the ``DELETE`` method with the same body format:: + + DELETE /api/v2/products/{uuid}/permissions/ + Content-Type: application/json + + { + "user": "alice", + "permissions": ["change_product"] + } + +Or for a group:: + + DELETE /api/v2/products/{uuid}/permissions/ + Content-Type: application/json + + { + "group": "backend-team", + "permissions": ["view_product"] + } + +Successful response:: + + {"status": "permissions removed"} + +4.5 Automate permissions in a CI/CD pipeline +^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ + +The following shell script illustrates how to create a Product version and immediately +assign permissions to a group, so that team members can view it without any manual +step:: + + BASE_URL="https://dejacode.example.com/api/v2" + TOKEN="your-api-token" + GROUP="backend-team" + + # Create the product version + RESPONSE=$(curl -s -X POST "$BASE_URL/products/" \ + -H "Authorization: Token $TOKEN" \ + -H "Content-Type: application/json" \ + -d '{"name": "MyApp", "version": "3.0"}') + + UUID=$(echo "$RESPONSE" | python3 -c "import sys,json; print(json.load(sys.stdin)['uuid'])") + + # Assign view permission to the team + curl -s -X POST "$BASE_URL/products/$UUID/permissions/" \ + -H "Authorization: Token $TOKEN" \ + -H "Content-Type: application/json" \ + -d "{\"group\": \"$GROUP\", \"permissions\": [\"view_product\"]}" + +**Access control for the permissions endpoint** + +Only the following users can call the ``/permissions/`` endpoint on a given product: + +- A **superuser** +- The user who **created** the product (``created_by`` field) diff --git a/product_portfolio/api.py b/product_portfolio/api.py index c1ea0ea6..c8d9998f 100644 --- a/product_portfolio/api.py +++ b/product_portfolio/api.py @@ -31,6 +31,7 @@ from dje.api import NameVersionHyperlinkedRelatedField from dje.api import ProductRelatedViewSet from dje.api import SPDXDocumentActionMixin +from dje.api_permissions import ObjectPermissionsMixin from dje.filters import LastModifiedDateFilter from dje.filters import MultipleCharFilter from dje.filters import MultipleUUIDFilter @@ -343,6 +344,7 @@ class Meta: class ProductViewSet( + ObjectPermissionsMixin, SendAboutFilesMixin, AboutCodeFilesActionMixin, SPDXDocumentActionMixin, diff --git a/product_portfolio/tests/test_api.py b/product_portfolio/tests/test_api.py index 0afaee9e..ce43a34e 100644 --- a/product_portfolio/tests/test_api.py +++ b/product_portfolio/tests/test_api.py @@ -586,6 +586,136 @@ def test_api_product_endpoint_cyclonedx_sbom_action(self): self.assertEqual(status.HTTP_400_BAD_REQUEST, response.status_code) self.assertEqual("Spec version 10.10 not supported", response.data) + def test_api_product_endpoint_manage_permissions_action(self): + from django.contrib.auth.models import Group + + url = reverse("api_v2:product-manage-permissions", args=[self.product1.uuid]) + + # Unauthenticated access is rejected with 403 + self.client.logout() + response = self.client.get(url) + self.assertEqual(status.HTTP_403_FORBIDDEN, response.status_code) + + # User without view_product gets 404 (object not in secured queryset) + self.client.login(username=self.base_user.username, password="secret") + response = self.client.get(url) + self.assertEqual(status.HTTP_404_NOT_FOUND, response.status_code) + + # User with view_product but not owner gets 403 + assign_perm("view_product", self.base_user, self.product1) + response = self.client.get(url) + self.assertEqual(status.HTTP_403_FORBIDDEN, response.status_code) + + # Superuser GET: response has "users" and "groups" keys + self.client.login(username=self.super_user.username, password="secret") + response = self.client.get(url) + self.assertEqual(status.HTTP_200_OK, response.status_code) + self.assertIn("users", response.data) + self.assertIn("groups", response.data) + # base_user has view_product at this point + self.assertEqual(1, len(response.data["users"])) + self.assertEqual(self.base_user.username, response.data["users"][0]["username"]) + self.assertEqual(self.dataspace.name, response.data["users"][0]["dataspace"]) + self.assertIn("view_product", response.data["users"][0]["object_permissions"]) + self.assertEqual([], response.data["groups"]) + + # Superuser can POST to assign multiple permissions to a user at once + data = { + "user": self.admin_user.username, + "permissions": ["view_product", "change_product"], + } + response = self.client.post(url, data, format="json") + self.assertEqual(status.HTTP_200_OK, response.status_code) + self.assertEqual({"status": "permissions assigned"}, response.data) + self.assertIn("view_product", get_perms(self.admin_user, self.product1)) + self.assertIn("change_product", get_perms(self.admin_user, self.product1)) + + # Superuser can DELETE to remove user permissions + data = {"user": self.admin_user.username, "permissions": ["view_product", "change_product"]} + response = self.client.delete(url, data, content_type="application/json") + self.assertEqual(status.HTTP_200_OK, response.status_code) + self.assertEqual({"status": "permissions removed"}, response.data) + self.assertNotIn("view_product", get_perms(self.admin_user, self.product1)) + self.assertNotIn("change_product", get_perms(self.admin_user, self.product1)) + + # Superuser can POST to assign permissions to a group + team = Group.objects.create(name="backend-team") + data = {"group": team.name, "permissions": ["view_product", "change_product"]} + response = self.client.post(url, data, format="json") + self.assertEqual(status.HTTP_200_OK, response.status_code) + self.assertEqual({"status": "permissions assigned"}, response.data) + self.assertIn("view_product", get_perms(team, self.product1)) + self.assertIn("change_product", get_perms(team, self.product1)) + + # GET lists the group with its permissions + response = self.client.get(url) + self.assertEqual(status.HTTP_200_OK, response.status_code) + self.assertEqual(1, len(response.data["groups"])) + self.assertEqual(team.name, response.data["groups"][0]["name"]) + self.assertIn("view_product", response.data["groups"][0]["object_permissions"]) + + # Superuser can DELETE to remove group permissions + data = {"group": team.name, "permissions": ["view_product", "change_product"]} + response = self.client.delete(url, data, content_type="application/json") + self.assertEqual(status.HTTP_200_OK, response.status_code) + self.assertNotIn("view_product", get_perms(team, self.product1)) + + # Product creator (created_by) can GET, POST, and DELETE + self.product1.created_by = self.admin_user + self.product1.save() + assign_perm("view_product", self.admin_user, self.product1) + self.client.login(username=self.admin_user.username, password="secret") + + response = self.client.get(url) + self.assertEqual(status.HTTP_200_OK, response.status_code) + + data = {"user": self.base_user.username, "permissions": ["change_product"]} + response = self.client.post(url, data, format="json") + self.assertEqual(status.HTTP_200_OK, response.status_code) + self.assertIn("change_product", get_perms(self.base_user, self.product1)) + + response = self.client.delete(url, data, content_type="application/json") + self.assertEqual(status.HTTP_200_OK, response.status_code) + self.assertNotIn("change_product", get_perms(self.base_user, self.product1)) + + # Invalid permission codename on POST returns 400 + self.client.login(username=self.super_user.username, password="secret") + data = {"user": self.base_user.username, "permissions": ["nonexistent_perm"]} + response = self.client.post(url, data, format="json") + self.assertEqual(status.HTTP_400_BAD_REQUEST, response.status_code) + self.assertIn("errors", response.data) + # DELETE with unknown codename is a no-op (guardian remove_perm is idempotent) + response = self.client.delete(url, data, content_type="application/json") + self.assertEqual(status.HTTP_200_OK, response.status_code) + + # Neither user nor group returns 400 + response = self.client.post(url, {"permissions": ["view_product"]}, format="json") + self.assertEqual(status.HTTP_400_BAD_REQUEST, response.status_code) + self.assertIn("errors", response.data) + + # Both user and group returns 400 + data = { + "user": self.base_user.username, + "group": team.name, + "permissions": ["view_product"], + } + response = self.client.post(url, data, format="json") + self.assertEqual(status.HTTP_400_BAD_REQUEST, response.status_code) + self.assertIn("errors", response.data) + + # Empty permissions list returns 400 + data = {"user": self.base_user.username, "permissions": []} + response = self.client.post(url, data, format="json") + self.assertEqual(status.HTTP_400_BAD_REQUEST, response.status_code) + self.assertIn("errors", response.data) + + # User from another dataspace is rejected (dataspace scoping) + other_dataspace_user = create_user("other_ds_user", self.alternate_dataspace) + data = {"user": other_dataspace_user.username, "permissions": ["view_product"]} + response = self.client.post(url, data, format="json") + self.assertEqual(status.HTTP_400_BAD_REQUEST, response.status_code) + self.assertIn("errors", response.data) + class ProductRelatedAPITestCase(TestCase): def setUp(self):