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,5 @@
+Added a new `allow_package_substitution` boolean field to PythonRepository (default: True).
+When set to False, any new repository version that would implicitly replace existing content
+with content that has the same filename but a different sha256 checksum is rejected. This
+applies to all repository version creation paths including uploads, modify, and sync. Content
+with a matching checksum is allowed through 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="allow_package_substitution",
+            field=models.BooleanField(default=True),
+        ),
+    ]

pulp_python/app/models.py

@@ -31,7 +31,11 @@
     PYPI_LAST_SERIAL,
     PYPI_SERIAL_CONSTANT,
 )
-from pulpcore.plugin.repo_version_utils import remove_duplicates, validate_repo_version
+from pulpcore.plugin.repo_version_utils import (
+    collect_duplicates,
+    remove_duplicates,
+    validate_repo_version,
+)
 from pulpcore.plugin.util import get_domain_pk, get_domain
 
 log = getLogger(__name__)
@@ -362,6 +366,7 @@ class PythonRepository(Repository, AutoAddObjPermsMixin):
     PULL_THROUGH_SUPPORTED = True
 
     autopublish = models.BooleanField(default=False)
+    allow_package_substitution = models.BooleanField(default=True)
 
     class Meta:
         default_related_name = "%(app_label)s_%(model_name)s"
@@ -390,6 +395,24 @@ def on_new_version(self, version):
     def finalize_new_version(self, new_version):
         """
         Remove duplicate packages that have the same filename.
+
+        When allow_package_substitution is False, reject any new version that would implicitly
+        replace existing content with different checksums (content substitution).
         """
+        if not self.allow_package_substitution:
+            self._check_for_package_substitution(new_version)
         remove_duplicates(new_version)
         validate_repo_version(new_version)
+
+    def _check_for_package_substitution(self, new_version):
+        """
+        Raise a ValueError if newly added packages would replace existing packages that have the
+        same filename but a different sha256 checksum.
+        """
+        duplicates = collect_duplicates(new_version.content, ("filename",))
+        if duplicates:
+            raise ValueError(
+                "Found duplicate packages being added with the same filename but different checksums."
+                "To allow this, set 'allow_package_substitution' to True on the repository."
+                f"Duplicate packages: {duplicates}"
+            )

pulp_python/app/serializers.py

@@ -53,9 +53,23 @@ class PythonRepositorySerializer(core_serializers.RepositorySerializer):
         default=False,
         required=False,
     )
+    allow_package_substitution = serializers.BooleanField(
+        help_text=_(
+            "Whether to allow package substitution (replacing existing packages with packages "
+            "that have the same filename but a different checksum). When False, any new "
+            "repository version that would cause such a substitution will be rejected. This "
+            "applies to all repository version creation paths including uploads, modify, and "
+            "sync. When True (the default), package substitution is allowed."
+        ),
+        default=True,
+        required=False,
+    )
 
     class Meta:
-        fields = core_serializers.RepositorySerializer.Meta.fields + ("autopublish",)
+        fields = core_serializers.RepositorySerializer.Meta.fields + (
+            "autopublish",
+            "allow_package_substitution",
+        )
         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},

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,95 @@ 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_disallow_package_substitution(
+    monitor_task,
+    python_bindings,
+    python_repo_factory,
+):
+    """
+    When allow_package_substitution=False, any new repository version that would substitute
+    existing content (same filename, different sha256) is rejected. This applies to both
+    content uploads and repository modify operations. Re-adding content with a matching
+    sha256 succeeds idempotently.
+    """
+    repo = python_repo_factory(allow_package_substitution=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_body2 = {"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_body2
+        )
+        monitor_task(response.task)
+    assert "already exists in repository" in exc.value.task.error["description"]
+    assert "allow_package_substitution" in exc.value.task.error["description"]
+
+    # Also create the conflicting content without a repo, then try to add via modify
+    response = python_bindings.ContentPackagesApi.create(**content_body2)
+    task = monitor_task(response.task)
+    content2 = python_bindings.ContentPackagesApi.read(task.created_resources[0])
+    body = {"add_content_units": [content2.pulp_href]}
+    with pytest.raises(PulpTaskError) as exc:
+        monitor_task(python_bindings.RepositoriesPythonApi.modify(repo.pulp_href, body).task)
+    assert "already exists in repository" 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_package_substitution_allowed_by_default(
+    monitor_task,
+    python_bindings,
+    python_repo_factory,
+):
+    """
+    By default (allow_package_substitution=True), uploading a file with the same filename
+    but different sha256 replaces the existing content in the repository.
+    """
+    repo = python_repo_factory()
+    assert repo.allow_package_substitution 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