Restrict non-admin users from uploading artifacts

fixes: pulp#5525

CHANGES/5525.bugfix

@@ -0,0 +1,2 @@
+Added a default access policy where only admin users will be able to upload and
+read/list artifacts.

CHANGES/5525.deprecation

@@ -0,0 +1,2 @@
+The `DELETE` request method from Artifact viewset has been deprecated and
+planned for removal.

pulp_file/tests/functional/api/test_domains.py

@@ -127,8 +127,10 @@ def test_artifact_upload(
     assert "default/api/v3/" in dup_artifact.pulp_href
     assert dup_artifact.sha256 == second_artifact.sha256
 
-    # Delete second artifact so domain can be deleted
-    pulpcore_bindings.ArtifactsApi.delete(second_artifact_href)
+    # Delete the artifacts so domain can be deleted
+    body = {"orphan_protection_time": 0, "content_hrefs": second_artifact_href}
+    task = pulpcore_bindings.OrphansCleanupApi.cleanup(body, pulp_domain=domain.name).task
+    monitor_task(task)
 
 
 @pytest.mark.parallel

pulpcore/app/viewsets/content.py

@@ -3,7 +3,7 @@
 from django.conf import settings
 from django.db import models
 from django_filters import NumberFilter
-from rest_framework import mixins, permissions, status
+from rest_framework import mixins, status
 from rest_framework.response import Response
 
 from pulpcore.filters import BaseFilterSet
@@ -15,6 +15,7 @@
 )
 from pulpcore.app.util import get_viewset_for_model
 from pulpcore.app.viewsets.base import NamedModelViewSet
+from pulpcore.app.loggers import deprecation_logger
 
 from .custom_filters import (
     ArtifactRepositoryVersionFilter,
@@ -73,12 +74,28 @@ class ArtifactViewSet(
     queryset = Artifact.objects.all()
     serializer_class = ArtifactSerializer
     filterset_class = ArtifactFilter
-    permission_classes = (permissions.IsAuthenticated,)
 
+    DEFAULT_ACCESS_POLICY = {
+        "statements": [
+            {
+                "action": ["create", "list", "retrieve"],
+                "principal": "admin",
+                "effect": "allow",
+            },
+        ],
+    }
+
+    # Deleting artifacts is a risky operation and will be removed in a future release.
+    # However, for compatibility reasons, it is still possible to execute the DELETE
+    # request by overriding the DEFAULT_ACCESS_POLICY.
     def destroy(self, request, pk):
         """
         Remove Artifact only if it is not associated with any Content.
         """
+        deprecation_logger.warning(
+            "destroy is deprecated. Deleting artifacts is a dangerous operation, "
+            "use orphan cleanup instead."
+        )
         try:
             return super().destroy(request, pk)
         except models.ProtectedError:

pulpcore/pytest_plugin.py

@@ -20,6 +20,7 @@
 from time import sleep
 from yarl import URL
 
+from pulpcore.client.pulpcore.api.artifacts_api import ArtifactsApi
 from pulpcore.tests.functional.utils import (
     SLEEP_TIME,
     TASK_TIMEOUT,
@@ -879,7 +880,7 @@ def redis_status(pulp_status):
 
 
 @pytest.fixture(scope="class")
-def add_to_cleanup(monitor_task):
+def add_to_cleanup(pulpcore_bindings, monitor_task):
     """Fixture to allow pulp objects to be deleted in reverse order after the test module."""
     obj_refs = []
 
@@ -894,7 +895,12 @@ def _add_to_cleanup(api_client, pulp_href):
         with suppress(Exception):
             # There was no delete task for this unit or the unit may already have been deleted.
             # Also we can never be sure which one is the right ApiException to catch.
-            task_url = api_client.delete(pulp_href).task
+            if isinstance(api_client, ArtifactsApi):
+                task_url = pulpcore_bindings.OrphansCleanupApi.cleanup(
+                    {"orphan_protection_time": 0, "content_hrefs": pulp_href}
+                ).task
+            else:
+                task_url = api_client.delete(pulp_href).task
             delete_task_hrefs.append(task_url)
 
     for deleted_task_href in delete_task_hrefs:

pulpcore/tests/functional/api/test_crd_artifacts.py

@@ -21,26 +21,30 @@ def pulpcore_random_file(tmp_path):
     return {"name": name, "size": 1024, "digest": digest}
 
 
-def _do_upload_valid_attrs(artifact_api, file, data):
-    """Upload a file with the given attributes."""
-    artifact = artifact_api.create(file, **data)
-    # assumes ALLOWED_CONTENT_CHECKSUMS does NOT contain "md5"
-    assert artifact.md5 is None, "MD5 {}".format(artifact.md5)
-    read_artifact = artifact_api.read(artifact.pulp_href)
-    # assumes ALLOWED_CONTENT_CHECKSUMS does NOT contain "md5"
-    assert read_artifact.md5 is None
-    for key, val in artifact.to_dict().items():
-        assert getattr(read_artifact, key) == val
-    # Delete the Artifact
-    artifact_api.delete(read_artifact.pulp_href)
-    with pytest.raises(ApiException) as e:
-        artifact_api.read(read_artifact.pulp_href)
-
-    assert e.value.status == 404
+@pytest.fixture
+def upload_valid_attrs(monitor_task, pulpcore_bindings):
+    def _upload_valid_attrs(artifact_api, file, data):
+        """Upload a file with the given attributes."""
+        artifact = artifact_api.create(file, **data)
+        # assumes ALLOWED_CONTENT_CHECKSUMS does NOT contain "md5"
+        assert artifact.md5 is None, "MD5 {}".format(artifact.md5)
+        read_artifact = artifact_api.read(artifact.pulp_href)
+        # assumes ALLOWED_CONTENT_CHECKSUMS does NOT contain "md5"
+        assert read_artifact.md5 is None
+        for key, val in artifact.to_dict().items():
+            assert getattr(read_artifact, key) == val
+        # Delete the Artifact
+        body = {"orphan_protection_time": 0, "content_hrefs": read_artifact.pulp_href}
+        monitor_task(pulpcore_bindings.OrphansCleanupApi.cleanup(body).task)
+        with pytest.raises(ApiException) as e:
+            artifact_api.read(read_artifact.pulp_href)
+        assert e.value.status == 404
+
+    return _upload_valid_attrs
 
 
 @pytest.mark.parallel
-def test_upload_valid_attrs(pulpcore_bindings, pulpcore_random_file):
+def test_upload_valid_attrs(pulpcore_bindings, pulpcore_random_file, upload_valid_attrs):
     """Upload a file, and provide valid attributes.
 
     For each possible combination of ``sha256`` and ``size`` (including
@@ -56,12 +60,10 @@ def test_upload_valid_attrs(pulpcore_bindings, pulpcore_random_file):
     for i in range(len(file_attrs) + 1):
         for keys in itertools.combinations(file_attrs, i):
             data = {key: file_attrs[key] for key in keys}
-            _do_upload_valid_attrs(
-                pulpcore_bindings.ArtifactsApi, pulpcore_random_file["name"], data
-            )
+            upload_valid_attrs(pulpcore_bindings.ArtifactsApi, pulpcore_random_file["name"], data)
 
 
-def test_upload_empty_file(delete_orphans_pre, pulpcore_bindings, tmp_path):
+def test_upload_empty_file(delete_orphans_pre, pulpcore_bindings, tmp_path, upload_valid_attrs):
     """Upload an empty file.
 
     For each possible combination of ``sha256`` and ``size`` (including
@@ -80,7 +82,7 @@ def test_upload_empty_file(delete_orphans_pre, pulpcore_bindings, tmp_path):
     for i in range(len(file_attrs) + 1):
         for keys in itertools.combinations(file_attrs, i):
             data = {key: file_attrs[key] for key in keys}
-            _do_upload_valid_attrs(pulpcore_bindings.ArtifactsApi, file, data)
+            upload_valid_attrs(pulpcore_bindings.ArtifactsApi, file, data)
 
 
 @pytest.mark.parallel
@@ -146,7 +148,7 @@ def test_upload_mixed_attrs(pulpcore_bindings, pulpcore_random_file):
 
 
 @pytest.mark.parallel
-def test_delete_artifact(pulpcore_bindings, pulpcore_random_file):
+def test_delete_artifact(pulpcore_bindings, pulpcore_random_file, gen_user, monitor_task):
     """Delete an artifact, it is removed from the filesystem."""
     if settings.DEFAULT_FILE_STORAGE != "pulpcore.app.models.storage.FileSystem":
         pytest.skip("this test only works for filesystem storage")
@@ -157,7 +159,45 @@ def test_delete_artifact(pulpcore_bindings, pulpcore_random_file):
     file_exists = os.path.exists(path_to_file)
     assert file_exists
 
-    # delete
-    pulpcore_bindings.ArtifactsApi.delete(artifact.pulp_href)
+    # try to delete as a regular (non-admin) user
+    regular_user = gen_user()
+    with regular_user, pytest.raises(ApiException) as e:
+        pulpcore_bindings.ArtifactsApi.delete(artifact.pulp_href)
+    assert e.value.status == 403
+
+    # destroy artifact api is not allowed, even for admins
+    with pytest.raises(ApiException) as e:
+        pulpcore_bindings.ArtifactsApi.delete(artifact.pulp_href)
+    assert e.value.status == 403
+
+    # remove artifact through orphan cleanup
+    body = {"orphan_protection_time": 0, "content_hrefs": artifact.pulp_href}
+    monitor_task(pulpcore_bindings.OrphansCleanupApi.cleanup(body).task)
     file_exists = os.path.exists(path_to_file)
     assert not file_exists
+
+
+@pytest.mark.parallel
+def test_upload_artifact_as_a_regular_user(pulpcore_bindings, gen_user, pulpcore_random_file):
+    regular_user = gen_user()
+    with regular_user, pytest.raises(ApiException) as e:
+        pulpcore_bindings.ArtifactsApi.create(pulpcore_random_file["name"])
+    assert e.value.status == 403
+
+
+@pytest.mark.parallel
+def test_list_and_retrieve_artifact_as_a_regular_user(
+    pulpcore_bindings, gen_user, pulpcore_random_file
+):
+    regular_user = gen_user()
+    artifact = pulpcore_bindings.ArtifactsApi.create(pulpcore_random_file["name"])
+
+    # check if list is not allowed
+    with regular_user, pytest.raises(ApiException) as e:
+        pulpcore_bindings.ArtifactsApi.list()
+    assert e.value.status == 403
+
+    # check if retrieve is also not allowed
+    with regular_user, pytest.raises(ApiException) as e:
+        pulpcore_bindings.ArtifactsApi.read(artifact.pulp_href)
+    assert e.value.status == 403

pulpcore/tests/functional/api/test_upload.py

@@ -72,11 +72,6 @@ def _upload_chunks(size, chunks, sha256, include_chunk_sha256=False):
         return upload, artifact
 
     yield _upload_chunks
-    for href in artifacts:
-        try:
-            pulpcore_bindings.ArtifactsApi.delete(href)
-        except ApiException:
-            pass
 
 
 @pytest.mark.parallel