From 9366b47346c35cb95c39edc7eb1599d4c46f0684 Mon Sep 17 00:00:00 2001
From: Erwan Rouchet <rouchet@teklia.com>
Date: Fri, 31 Mar 2023 13:36:46 +0200
Subject: [PATCH] Rewrite CreateArtifact permissions
---
arkindex/ponos/api.py | 15 +++++-------
arkindex/ponos/permissions.py | 32 ++++++++++++++++++++++++
arkindex/ponos/tests/test_api.py | 42 ++++++++++++++++++++++++++++++++
3 files changed, 80 insertions(+), 9 deletions(-)
diff --git a/arkindex/ponos/api.py b/arkindex/ponos/api.py
index 79593e4ac5..1f6010881b 100644
--- a/arkindex/ponos/api.py
+++ b/arkindex/ponos/api.py
@@ -28,6 +28,7 @@ from arkindex.ponos.permissions import (
IsAgentOrArtifactAdmin,
IsAgentOrTask,
IsAgentOrTaskAdmin,
+ IsAgentOrTaskAdminOrReadOnly,
IsAssignedAgentOrReadOnly,
)
from arkindex.ponos.recipe import parse_recipe
@@ -49,7 +50,7 @@ from arkindex.ponos.serializers import (
)
from arkindex.process.models import Process
from arkindex.project.mixins import ProcessACLMixin
-from arkindex.project.permissions import IsAuthenticated, IsVerified
+from arkindex.project.permissions import IsVerified
from arkindex.users.models import Role
from rest_framework_simplejwt.views import TokenRefreshView
@@ -314,26 +315,22 @@ class TaskArtifacts(ListCreateAPIView):
# Used for OpenAPI schema serialization: the ID in the path is the task ID
queryset = Task.objects.none()
- permission_classes = (IsAuthenticated, )
+ permission_classes = (IsAgentOrTaskAdminOrReadOnly, )
serializer_class = ArtifactSerializer
# Force no pagination, even when global settings add them
pagination_class = None
def get_task(self):
- if "pk" not in self.kwargs:
- return Artifact.objects.none()
- return get_object_or_404(Task, pk=self.kwargs["pk"])
+ task = get_object_or_404(Task, pk=self.kwargs["pk"])
+ self.check_object_permissions(self.request, task)
+ return task
def get_queryset(self):
task = self.get_task()
return task.artifacts.all()
def perform_create(self, serializer):
- user = self.request.user
- if not user.is_staff and not getattr(user, "is_agent", False):
- raise PermissionDenied()
-
# Assign task when creating through the API
serializer.save(task=self.get_task())
diff --git a/arkindex/ponos/permissions.py b/arkindex/ponos/permissions.py
index 0dcdb04f6f..d059cd2d37 100644
--- a/arkindex/ponos/permissions.py
+++ b/arkindex/ponos/permissions.py
@@ -3,6 +3,7 @@ from rest_framework.permissions import SAFE_METHODS
from arkindex.ponos.models import Task
from arkindex.project.mixins import CorpusACLMixin
from arkindex.project.permissions import IsAuthenticated
+from arkindex.users.models import Role
def require_agent_or_admin(request, view):
@@ -74,6 +75,37 @@ class IsAgentOrTaskAdmin(CorpusACLMixin, IsAuthenticated):
)
+class IsAgentOrTaskAdminOrReadOnly(CorpusACLMixin, IsAuthenticated):
+ """
+ Instance admins, agents, process admins, and the task itself are always allowed.
+ For GET/HEAD, only a Guest level on the process is required for regular users.
+ """
+
+ def has_object_permission(self, request, view, task):
+ # Allow agents and instance admins
+ if require_agent_or_admin(request, view):
+ return True
+
+ # Allow a task to access itself
+ if isinstance(request.auth, Task) and request.auth.id == task.id:
+ return True
+
+ # Add request to attributes for the ACL mixin to work with self.user
+ self.request = request
+ level = self.process_access_level(task.workflow.process)
+
+ # Require *some* access to the process
+ if level is None:
+ return False
+
+ # Require only a guest access for GET/HEAD
+ if request.method in SAFE_METHODS:
+ return level >= Role.Guest.value
+
+ # Require admin access for other methods
+ return level >= Role.Admin.value
+
+
class IsAgentOrArtifactAdmin(IsAgentOrTaskAdmin):
"""
Permission to access an artifact with high privilege, based on
diff --git a/arkindex/ponos/tests/test_api.py b/arkindex/ponos/tests/test_api.py
index 596ed50343..cc69589586 100644
--- a/arkindex/ponos/tests/test_api.py
+++ b/arkindex/ponos/tests/test_api.py
@@ -1479,6 +1479,48 @@ class TestAPI(APITestCase):
self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST)
self.assertEqual(resp.json(), {"size": [expected_error]})
+ @override_settings(PONOS_ARTIFACT_MAX_SIZE=99999)
+ def test_create_artifact_task(self):
+ Process.objects.create(
+ creator=self.user,
+ mode=ProcessMode.Repository,
+ workflow=self.workflow,
+ )
+
+ with self.assertNumQueries(5):
+ resp = self.client.post(
+ reverse("api:task-artifacts", args=[self.task1.id]),
+ data={
+ "path": "some/path.txt",
+ "content_type": "text/plain",
+ "size": 1000,
+ },
+ HTTP_AUTHORIZATION=f"Ponos {self.task1.token}",
+ )
+ self.assertEqual(resp.status_code, status.HTTP_201_CREATED, resp.json())
+
+ data = resp.json()
+ s3_put_url = data.pop("s3_put_url")
+ self.assertTrue(
+ s3_put_url.startswith(
+ f"http://somewhere/ponos-artifacts/{self.workflow.id}/{self.task1.id}/some/path.txt"
+ )
+ )
+
+ artifact = self.task1.artifacts.get()
+ self.assertDictEqual(
+ resp.json(),
+ {
+ "content_type": "text/plain",
+ "created": str_date(artifact.created),
+ "id": str(artifact.id),
+ "path": "some/path.txt",
+ "size": 1000,
+ "updated": str_date(artifact.updated),
+ "url": f"http://testserver/api/v1/task/{self.task1.id}/artifact/some/path.txt",
+ },
+ )
+
def test_artifact_download(self):
# Add a new artifact
--
GitLab