Add new field to repository to prevent duplicate filename uploads

fixes: #1162
Generated by: Claude-Opus-4.6

Author: gerrod3

CHANGES/1162.feature

@@ -0,0 +1,4 @@
+Added a new `upload_duplicate_filenames` boolean field to PythonRepository (default: True).
+When set to False, uploading a package whose filename already exists in the repository with a
+different sha256 checksum is rejected, preventing unintentional content substitution. Uploads
+with a matching checksum succeed idempotently.

pulp_python/app/migrations/0021_pythonrepository_upload_duplicate_filenames.py

@@ -0,0 +1,16 @@
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("python", "0020_pythonpackagecontent_name_normalized"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="pythonrepository",
+            name="upload_duplicate_filenames",
+            field=models.BooleanField(default=True),
+        ),
+    ]

pulp_python/app/models.py

@@ -362,6 +362,7 @@ class PythonRepository(Repository, AutoAddObjPermsMixin):
     PULL_THROUGH_SUPPORTED = True
 
     autopublish = models.BooleanField(default=False)
+    upload_duplicate_filenames = models.BooleanField(default=True)
 
     class Meta:
         default_related_name = "%(app_label)s_%(model_name)s"

pulp_python/app/serializers.py

@@ -53,9 +53,23 @@ class PythonRepositorySerializer(core_serializers.RepositorySerializer):
         default=False,
         required=False,
     )
+    upload_duplicate_filenames = serializers.BooleanField(
+        help_text=_(
+            "Whether to allow uploading Python packages whose filename already exists in the "
+            "repository. When False, uploading a file whose filename already exists in the "
+            "repository with a different sha256 checksum will be rejected, while uploads with a "
+            "matching checksum will succeed idempotently. When True (the default), duplicate "
+            "filenames are allowed and will replace the existing content."
+        ),
+        default=True,
+        required=False,
+    )
 
     class Meta:
-        fields = core_serializers.RepositorySerializer.Meta.fields + ("autopublish",)
+        fields = core_serializers.RepositorySerializer.Meta.fields + (
+            "autopublish",
+            "upload_duplicate_filenames",
+        )
         model = python_models.PythonRepository
 
 
@@ -448,7 +462,7 @@ def create(self, validated_data):
         content = super().create(validated_data)
         if provenance:
             prov_sha256 = python_models.PackageProvenance.calculate_sha256(provenance)
-            prov_model, _ = python_models.PackageProvenance.objects.get_or_create(
+            prov_model, _created = python_models.PackageProvenance.objects.get_or_create(
                 sha256=prov_sha256,
                 _pulp_domain=get_domain(),
                 defaults={"package": content, "provenance": provenance},
@@ -457,7 +471,22 @@ def create(self, validated_data):
                 core_models.CreatedResource.objects.create(content_object=prov_model)
             setattr(content, "provenance", prov_model)
         if repository:
-            repository.cast()
+            repository = repository.cast()
+            if not repository.upload_duplicate_filenames:
+                latest_version = repository.latest_version()
+                if latest_version:
+                    existing = python_models.PythonPackageContent.objects.filter(
+                        pk__in=latest_version.content,
+                        filename=content.filename,
+                    ).exclude(sha256=content.sha256)
+                    if existing.exists():
+                        raise serializers.ValidationError(
+                            _(
+                                "A file with filename '{}' already exists in repository '{}' "
+                                "with a different checksum. To allow this, set "
+                                "'upload_duplicate_filenames' to True on the repository."
+                            ).format(content.filename, repository.name)
+                        )
             content_to_add = [content.pk, content.provenance.pk] if provenance else [content.pk]
             content_to_add = core_models.Content.objects.filter(pk__in=content_to_add)
             with repository.new_version() as new_version:

pulp_python/tests/functional/api/test_crud_content_unit.py

@@ -144,44 +144,41 @@ def test_content_create_new_metadata(
         assert getattr(content, k) == v
 
 
+def get_package_url(package, filename):
+    with PyPISimple() as client:
+        page = client.get_project_page(package)
+        for package in page.packages:
+            if package.filename == filename:
+                return package.url
+        raise ValueError(f"Package {filename} not found")
+
+
 @pytest.mark.parallel
 def test_upload_metadata_23_spec(python_content_factory):
     """Test that packages using metadata spec 2.3 can be uploaded to pulp."""
     filename = "urllib3-2.2.2-py3-none-any.whl"
-    with PyPISimple() as client:
-        page = client.get_project_page("urllib3")
-        for package in page.packages:
-            if package.filename == filename:
-                content = python_content_factory(filename, url=package.url)
-                assert content.metadata_version == "2.3"
-                break
+    url = get_package_url("urllib3", filename)
+    content = python_content_factory(filename, url=url)
+    assert content.metadata_version == "2.3"
 
 
 @pytest.mark.parallel
 def test_upload_requires_python(python_content_factory):
     filename = "pip-24.3.1-py3-none-any.whl"
-    with PyPISimple() as client:
-        page = client.get_project_page("pip")
-        for package in page.packages:
-            if package.filename == filename:
-                content = python_content_factory(filename, url=package.url)
-                assert content.requires_python == ">=3.8"
-                break
+    url = get_package_url("pip", filename)
+    content = python_content_factory(filename, url=url)
+    assert content.requires_python == ">=3.8"
 
 
 @pytest.mark.parallel
 def test_upload_metadata_24_spec(python_content_factory):
     """Test that packages using metadata spec 2.4 can be uploaded to pulp."""
     filename = "setuptools-80.9.0.tar.gz"
-    with PyPISimple() as client:
-        page = client.get_project_page("setuptools")
-        for package in page.packages:
-            if package.filename == filename:
-                content = python_content_factory(filename, url=package.url)
-                assert content.metadata_version == "2.4"
-                assert content.license_expression == "MIT"
-                assert content.license_file == '["LICENSE"]'
-                break
+    url = get_package_url("setuptools", filename)
+    content = python_content_factory(filename, url=url)
+    assert content.metadata_version == "2.4"
+    assert content.license_expression == "MIT"
+    assert content.license_file == '["LICENSE"]'
 
 
 @pytest.mark.parallel
@@ -203,3 +200,85 @@ def test_package_creation_with_metadata(
     ensure_metadata(
         pulp_content_url, distro.base_path, PYTHON_WHEEL_FILENAME, "shelf-reader", "0.1"
     )
+
+
+@pytest.mark.parallel
+def test_upload_duplicate_filenames_disabled(
+    monitor_task,
+    python_bindings,
+    python_repo_factory,
+):
+    """
+    When upload_duplicate_filenames=False, uploading a file whose filename already exists
+    in the repository with a different sha256 is rejected, while re-uploading the same
+    file (matching sha256) succeeds idempotently.
+    """
+    repo = python_repo_factory(upload_duplicate_filenames=False)
+
+    # First upload succeeds
+    content_body = {"relative_path": PYTHON_EGG_FILENAME, "file_url": PYTHON_EGG_URL}
+    response = python_bindings.ContentPackagesApi.create(repository=repo.pulp_href, **content_body)
+    task = monitor_task(response.task)
+    content = python_bindings.ContentPackagesApi.read(task.created_resources[-1])
+    assert content.filename == PYTHON_EGG_FILENAME
+
+    # Re-upload same artifact with same filename — should succeed (idempotent)
+    response = python_bindings.ContentPackagesApi.create(repository=repo.pulp_href, **content_body)
+    task = monitor_task(response.task)
+    assert content.pulp_href in task.created_resources
+
+    # Upload a different artifact with the same filename — should be rejected
+    second_filename = "pip-26.0.1.tar.gz"
+    second_url = get_package_url("pip", second_filename)
+    content_body = {"relative_path": PYTHON_EGG_FILENAME, "file_url": second_url}
+    with pytest.raises(PulpTaskError) as exc:
+        response = python_bindings.ContentPackagesApi.create(
+            repository=repo.pulp_href, **content_body
+        )
+        monitor_task(response.task)
+    assert "already exists in repository" in exc.value.task.error["description"]
+    assert "upload_duplicate_filenames" in exc.value.task.error["description"]
+
+    # Verify the repository still has only the original content
+    repo = python_bindings.RepositoriesPythonApi.read(repo.pulp_href)
+    content_list = python_bindings.ContentPackagesApi.list(
+        repository_version=repo.latest_version_href
+    )
+    assert content_list.count == 1
+    assert content_list.results[0].sha256 == content.sha256
+
+
+@pytest.mark.parallel
+def test_upload_duplicate_filename_allowed_by_default(
+    monitor_task,
+    python_bindings,
+    python_repo_factory,
+):
+    """
+    By default (upload_duplicate_filenames=True), uploading a file with the same filename
+    but different sha256 replaces the existing content in the repository.
+    """
+    repo = python_repo_factory()
+    assert repo.upload_duplicate_filenames is True
+
+    content_body = {"relative_path": PYTHON_EGG_FILENAME, "file_url": PYTHON_EGG_URL}
+    response = python_bindings.ContentPackagesApi.create(repository=repo.pulp_href, **content_body)
+    task = monitor_task(response.task)
+    content1 = python_bindings.ContentPackagesApi.read(task.created_resources[-1])
+
+    # Upload a different artifact with the same filename — should succeed and replace
+    second_filename = "pip-26.0.tar.gz"
+    second_url = get_package_url("pip", second_filename)
+    content_body = {"relative_path": PYTHON_EGG_FILENAME, "file_url": second_url}
+    response = python_bindings.ContentPackagesApi.create(repository=repo.pulp_href, **content_body)
+    task = monitor_task(response.task)
+    content2 = python_bindings.ContentPackagesApi.read(task.created_resources[-1])
+    assert content2.pulp_href != content1.pulp_href
+
+    # Verify the repo has only the new content
+    repo = python_bindings.RepositoriesPythonApi.read(repo.pulp_href)
+    content_list = python_bindings.ContentPackagesApi.list(
+        repository_version=repo.latest_version_href
+    )
+    assert content_list.count == 1
+    assert content_list.results[0].sha256 == content2.sha256