Disallow RPM upload with the same NVR [RHELDST-37824]

src/pubtools/_pulp/tasks/push/items/base.py

@@ -645,3 +645,7 @@ def upload_key(self):
         content types which can safely reuse uploads.
         """
         return None
+
+    def fail_if_duplicate(self, pulp_client):  # pylint:disable=unused-argument
+        """Subclasses may override this method to check for duplicates and fail if found."""
+        return None

src/pubtools/_pulp/tasks/push/items/rpm.py

@@ -144,3 +144,32 @@ def ensure_uploaded(self, ctx, repo_f=None):
 
     def upload_to_repo(self, repo):
         return repo.upload_rpm(self.pushsource_item.content(), cdn_path=self.cdn_path)
+
+    def fail_if_duplicate(self, pulp_client):
+        """
+        If there is already an RPM with identical CDN path of requested within push present in Pulp but has different checksum, raise an error.
+        Publishing such units would break integrity of repositories when published becuase of collision on origin path.
+        """
+        crit = Criteria.and_(
+            Criteria.with_unit_type(
+                RpmUnit, unit_fields=["filename", "cdn_path", "sha256sum"]
+            ),
+            Criteria.with_field("filename", self.pushsource_item.name),
+        )
+
+        results = pulp_client.search_content(crit)
+        for item in results:
+            if (
+                item.cdn_path
+                and item.cdn_path == self.cdn_path
+                and item.sha256sum != self.pushsource_item.sha256sum
+            ):
+                msg = (
+                    "Fatal error: Duplicate RPM present in Pulp: %s, sha256: %s, cdn_path: %s"
+                    % (
+                        item.filename,
+                        item.sha256sum,
+                        item.cdn_path,
+                    )
+                )
+                raise RuntimeError(msg)

src/pubtools/_pulp/tasks/push/phase/constants.py

@@ -112,3 +112,8 @@ def atom(name):
 
 OUT_MAX_FUTURES = int(os.getenv("PUBTOOLS_PULP_OUT_MAX_FUTURES") or "10")
 """Max number of pending futures in output buffer."""
+
+# Special workaround for allowing duplicate units to be uploaded to Pulp.
+ALLOW_DUPLICATE_UNITS = os.getenv(
+    "PUBTOOLS_PULP_ALLOW_DUPLICATE_UNITS", "True"
+).strip().lower() in ("true", "1", "yes", "y")

src/pubtools/_pulp/tasks/push/phase/upload.py

@@ -1,6 +1,8 @@
 import logging
 
+
 from .base import Phase
+from . import constants
 from ..items import State
 
 LOG = logging.getLogger("pubtools.pulp")
@@ -48,6 +50,14 @@ def run(self):
         upload_context = {}
 
         for item in self.iter_input():
+            # We don't want to allow any other operation on units with potentially duplicated origin path.
+            # Limited to units that would be associated to destination repos which would cause origin path collision.
+            if not constants.ALLOW_DUPLICATE_UNITS and item.pulp_state in [
+                State.MISSING,
+                State.PARTIAL,
+            ]:
+                item.fail_if_duplicate(self.pulp_client)
+
             if item.pulp_state in [State.IN_REPOS, State.PARTIAL, State.NEEDS_UPDATE]:
                 # This item is already in Pulp.
                 uploaded += 1
@@ -59,6 +69,7 @@ def run(self):
             else:
                 # This item is not in Pulp, or otherwise needs a reupload.
                 item_type = type(item)
+
                 if item.MULTI_UPLOAD_CONTEXT:
                     if item_type not in upload_context:
                         upload_context[item_type] = {}

tests/push/test_push.py

@@ -19,7 +19,7 @@
 from pubtools.pluggy import pm
 
 from pubtools._pulp.tasks.push import entry_point
-from pubtools._pulp.tasks.push.phase import context
+from pubtools._pulp.tasks.push.phase import context, constants
 
 from .util import hide_unit_ids
 
@@ -88,10 +88,15 @@ def test_typical_push(
     command_tester,
     hookspy,
     stub_collector,
+    monkeypatch,
 ):
     """Test a typical case of push using all sorts of content where the content
     is initially not present in Pulp.
     """
+    # patch this constant to disallow duplicate units to be uploaded to Pulp.
+    # should be a no-op for non-RPM units, should pass for RPM units.
+    monkeypatch.setattr(constants, "ALLOW_DUPLICATE_UNITS", False)
+
     # Sanity check that the Pulp server is, initially, empty.
     client = fake_controller.client
     assert list(client.search_content()) == []

tests/push/test_push_rpm_duplicate.py

@@ -0,0 +1,80 @@
+import functools
+
+import pytest
+
+from pushsource import Source, RpmPushItem
+from pubtools.pulplib import RpmUnit
+
+from pubtools._pulp.tasks.push import entry_point
+from pubtools._pulp.tasks.push.phase import constants
+
+
+def test_push_rpm_duplicate_fail(
+    fake_controller, fake_push, command_tester, caplog, monkeypatch
+):
+    """Test that push detects and fails in the case where a RPM with the same cdn_path but different checksum is pushed to Pulp in the destination repository."""
+    monkeypatch.setattr(constants, "ALLOW_DUPLICATE_UNITS", False)
+
+    client = fake_controller.client
+
+    rpm_dest = client.get_repository("dest1").result()
+
+    # Make this file exist.
+    existing_rpm = RpmUnit(
+        name="some-rpm",
+        version="1.0.0",
+        release="1",
+        arch="noarch",
+        sha256sum="db68c8a70f8383de71c107dca5fcfe53b1132186d1a6681d9ee3f4eea724fabb",
+        filename="some-rpm-1.0.0-1.noarch.rpm",
+        cdn_path="/content/origin/rpms/some-rpm/1.0.0/1/f21541eb/some-rpm-1.0.0-1.noarch.rpm",
+        signing_key="F21541EB",
+    )
+    fake_controller.insert_units(rpm_dest, [existing_rpm])
+
+    # Unit is now in dest1 repository.
+    # Set up a pushsource backend which requests push of the RPM with the same cdn_path/NVR/signing key but different checksum
+    # to different destination repository.
+    Source.register_backend(
+        "test",
+        lambda: [
+            RpmPushItem(
+                name="some-rpm-1.0.0-1.noarch.rpm",
+                dest=["dest2"],
+                # different checksum but same GPG key
+                sha256sum="e823456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef",
+                md5sum=32 * "a",
+                src="fake/path",
+                signing_key="F21541EB",
+            ),
+        ],
+    )
+
+    args = [
+        "",
+        "--source",
+        "test:",
+        "--pulp-url",
+        "https://pulp.example.com/",
+        "--allow-unsigned",
+    ]
+
+    run = functools.partial(entry_point, cls=lambda: fake_push)
+
+    # Ask it to push.
+    with pytest.raises(SystemExit) as excinfo:
+        command_tester.test(
+            run,
+            args,
+            # Can't guarantee a stable log order.
+            compare_plaintext=False,
+            compare_jsonl=False,
+        )
+
+    # It should have failed.
+    assert excinfo.value.code == 59
+
+    # It should tell us why it failed.
+    msg = "Duplicate RPM present in Pulp: some-rpm-1.0.0-1.noarch.rpm, sha256: db68c8a70f8383de71c107dca5fcfe53b1132186d1a6681d9ee3f4eea724fabb, cdn_path: /content/origin/rpms/some-rpm/1.0.0/1/f21541eb/some-rpm-1.0.0-1.noarch.rpm"
+
+    assert msg in caplog.text