Fix provenance sync to use Pulp's extended Provenance model

Sync was validating provenance with pypi_attestations.Provenance, which
rejects Pulp User publishers created during upload. Use the extended model
from pulp_python.app.provenance instead.

Add a cross-domain functional test that uploads attestations in one domain
and syncs provenance into a repository in another.

Assisted-by: Cursor
Co-authored-by: Cursor <cursoragent@cursor.com>

Author: gerrod3

CHANGES/+pulp-attestation-sync.bugfix

@@ -0,0 +1 @@
+Fixed provenance sync failing when syncing Pulp-created attestations that use a `Pulp User` publisher.

pulp_python/app/tasks/sync.py

@@ -9,7 +9,6 @@
 from bandersnatch.mirror import Mirror
 from lxml.etree import LxmlError
 from packaging.requirements import Requirement
-from pypi_attestations import Provenance
 from pypi_simple import IndexPage
 
 from pulpcore.plugin.download import HttpDownloader
@@ -28,6 +27,7 @@
     PythonPackageContent,
     PythonRemote,
 )
+from pulp_python.app.provenance import Provenance
 from pulp_python.app.utils import PYPI_LAST_SERIAL, aget_remote_simple_page, parse_metadata
 
 logger = logging.getLogger(__name__)

pulp_python/tests/functional/api/test_attestations.py

@@ -8,6 +8,7 @@
 import requests
 from pypi_simple import PyPISimple
 
+from pulpcore.app import settings  # noqa: TID251
 from pulpcore.tests.functional.utils import PulpTaskError
 
 
@@ -154,6 +155,55 @@ def test_attestation_sync_upload(python_bindings, twine_package, download_python
     assert att_bundle["publisher"]["kind"] == "Pulp User"
 
 
+@pytest.mark.skipif(not settings.DOMAIN_ENABLED, reason="Domain not enabled")
+@pytest.mark.parallel
+def test_sync_pulp_created_provenance(
+    domain_factory,
+    python_bindings,
+    twine_package,
+    python_repo_factory,
+    python_distribution_factory,
+    python_remote_factory,
+    python_repo_with_sync,
+    python_content_summary,
+    monitor_task,
+):
+    """Test syncing provenance across domains from a Pulp upload."""
+    source_domain = domain_factory()
+    attestations = get_attestations(twine_package.provenance_url)
+    source_repo = python_repo_factory(pulp_domain=source_domain.name)
+    body = {
+        "relative_path": twine_package.filename,
+        "file_url": twine_package.url,
+        "attestations": json.dumps(attestations),
+        "repository": source_repo.pulp_href,
+    }
+    task = python_bindings.ContentPackagesApi.create(pulp_domain=source_domain.name, **body).task
+    monitor_task(task)
+
+    source_distro = python_distribution_factory(
+        repository=source_repo, pulp_domain=source_domain.name
+    )
+    dest_domain = domain_factory()
+    remote = python_remote_factory(
+        url=source_distro.base_url,
+        provenance=True,
+        includes=[f"twine=={twine_package.version}"],
+        pulp_domain=dest_domain.name,
+    )
+    dest_repo = python_repo_with_sync(remote=remote, pulp_domain=dest_domain.name)
+
+    summary = python_content_summary(repository_version=dest_repo.latest_version_href)
+    assert summary.present["python.provenance"]["count"] == 1
+
+    provs = python_bindings.ContentProvenanceApi.list(
+        repository_version=dest_repo.latest_version_href,
+        pulp_domain=dest_domain.name,
+    )
+    assert provs.count == 1
+    assert provs.results[0].provenance["attestation_bundles"][0]["publisher"]["kind"] == "Pulp User"
+
+
 def test_attestation_twine_upload(
     pulpcore_bindings,
     python_content_summary,