Add attestation upload support

fixes: #984

Author: gerrod3

CHANGES/984.feature

@@ -0,0 +1 @@
+Added attestations field to package upload that will create a PEP 740 Provenance object for that content.

MANIFEST.in

@@ -9,4 +9,5 @@ include functest_requirements.txt
 include test_requirements.txt
 include unittest_requirements.txt
 include pulp_python/app/webserver_snippets/*
+include pulp_python/tests/functional/assets/*
 exclude releasing.md

pulp_python/app/models.py

@@ -21,8 +21,8 @@
 )
 from pulpcore.plugin.responses import ArtifactResponse
 
-from pypi_attestations import Provenance
 from pathlib import PurePath
+from .provenance import Provenance
 from .utils import (
     artifact_to_python_content_data,
     canonicalize_name,

pulp_python/app/provenance.py

@@ -0,0 +1,67 @@
+from typing import Annotated, Literal, Union, get_args
+
+from pydantic import BaseModel, ConfigDict, Field
+from pydantic.alias_generators import to_snake
+from pypi_attestations import Attestation, Distribution, Publisher
+
+
+class _PermissivePolicy:
+    """A permissive verification policy that always succeeds."""
+
+    def verify(self, cert):
+        """Always succeed verification."""
+        pass
+
+
+class AnyPublisher(BaseModel):
+    """A fallback publisher for any kind not matching other publisher types."""
+
+    model_config = ConfigDict(alias_generator=to_snake, extra="allow")
+
+    kind: str
+
+    def _as_policy(self):
+        """Return a permissive policy that always succeed."""
+        return _PermissivePolicy()
+
+
+# Get the underlying Union type of the original Publisher
+# Publisher is Annotated[Union[...], Field(discriminator="kind")]
+_OriginalPublisherTypes = get_args(Publisher.__origin__)
+# Add AnyPublisher to the list of original publisher types
+_ExtendedPublisherTypes = (*_OriginalPublisherTypes, AnyPublisher)
+_ExtendedPublisherUnion = Union[_ExtendedPublisherTypes]
+# Create a new type that fallbacks to AnyPublisher
+ExtendedPublisher = Annotated[_ExtendedPublisherUnion, Field(union_mode="left_to_right")]
+
+
+class AttestationBundle(BaseModel):
+    """
+    AttestationBundle object as defined in PEP740.
+
+    PyPI only accepts attestations from TrustedPublishers (GitHub, GitLab, Google), but we will
+    accept from any user.
+    """
+
+    publisher: ExtendedPublisher
+    attestations: list[Attestation]
+
+
+class Provenance(BaseModel):
+    """Provenance object as defined in PEP740."""
+
+    version: Literal[1] = 1
+    attestation_bundles: list[AttestationBundle]
+
+
+def verify_provenance(filename, sha256, provenance, offline=False):
+    """Verify the provenance object is valid for the package."""
+    dist = Distribution(name=filename, digest=sha256)
+    for bundle in provenance.attestation_bundles:
+        publisher = bundle.publisher
+        policy = publisher._as_policy()
+        for attestation in bundle.attestations:
+            bundle = attestation.to_bundle()
+            checkpoint = bundle.log_entry._inner.inclusion_proof.checkpoint
+            staging = "sigstage.dev" in checkpoint.envelope
+            attestation.verify(policy, dist, staging=staging, offline=offline)

pulp_python/app/pypi/serializers.py

@@ -2,6 +2,8 @@
 from gettext import gettext as _
 
 from rest_framework import serializers
+from pydantic import TypeAdapter, ValidationError
+from pypi_attestations import Attestation
 from pulp_python.app.utils import DIST_EXTENSIONS, SUPPORTED_METADATA_VERSIONS
 from pulpcore.plugin.models import Artifact
 from pulpcore.plugin.util import get_domain
@@ -70,6 +72,11 @@ class PackageUploadSerializer(serializers.Serializer):
         required=False,
         choices=SUPPORTED_METADATA_VERSIONS,
     )
+    attestations = serializers.JSONField(
+        required=False,
+        help_text=_("A JSON list containing attestations for the package."),
+        write_only=True,
+    )
 
     def validate(self, data):
         """Validates the request."""
@@ -98,6 +105,14 @@ def validate(self, data):
                 }
             )
 
+        if attestations := data.get("attestations"):
+            try:
+                attestations = TypeAdapter(list[Attestation]).validate_python(attestations)
+            except ValidationError as e:
+                raise serializers.ValidationError(
+                    {"attestations": _("The uploaded attestations are not valid: {}".format(e))}
+                )
+
         sha256 = data.get("sha256_digest")
         digests = {"sha256": sha256} if sha256 else None
         artifact = Artifact.init_and_validate(file, expected_digests=digests)

pulp_python/app/pypi/views.py

@@ -181,53 +181,63 @@ def upload(self, request, path):
         serializer = PackageUploadSerializer(data=request.data)
         serializer.is_valid(raise_exception=True)
         artifact, filename = serializer.validated_data["content"]
+        attestations = serializer.validated_data.get("attestations", None)
         repo_content = self.get_content(self.get_repository_version(self.distribution))
         if repo_content.filter(filename=filename).exists():
             return HttpResponseBadRequest(reason=f"Package {filename} already exists in index")
 
         if settings.PYTHON_GROUP_UPLOADS:
-            return self.upload_package_group(repo, artifact, filename, request.session)
+            return self.upload_package_group(
+                repo, artifact, filename, attestations, request.session
+            )
 
         result = dispatch(
             tasks.upload,
             exclusive_resources=[artifact, repo],
             kwargs={
                 "artifact_sha256": artifact.sha256,
                 "filename": filename,
+                "attestations": attestations,
                 "repository_pk": str(repo.pk),
             },
         )
         return OperationPostponedResponse(result, request)
 
-    def upload_package_group(self, repo, artifact, filename, session):
+    def upload_package_group(self, repo, artifact, filename, attestations, session):
         """Steps 4 & 5, spawns tasks to add packages to index."""
         start_time = datetime.now(tz=timezone.utc) + timedelta(seconds=5)
         task = "updated"
         if not session.get("start"):
-            task = self.create_group_upload_task(session, repo, artifact, filename, start_time)
+            task = self.create_group_upload_task(
+                session, repo, artifact, filename, attestations, start_time
+            )
         else:
             sq = Session.objects.select_for_update(nowait=True).filter(pk=session.session_key)
             try:
                 with transaction.atomic():
                     sq.first()
                     current_start = datetime.fromisoformat(session["start"])
                     if current_start >= datetime.now(tz=timezone.utc):
-                        session["artifacts"].append((str(artifact.sha256), filename))
+                        session["artifacts"].append((str(artifact.sha256), filename, attestations))
                         session["start"] = str(start_time)
                         session.modified = False
                         session.save()
                     else:
                         raise DatabaseError
             except DatabaseError:
                 session.cycle_key()
-                task = self.create_group_upload_task(session, repo, artifact, filename, start_time)
+                task = self.create_group_upload_task(
+                    session, repo, artifact, filename, attestations, start_time
+                )
         data = {"session": session.session_key, "task": task, "task_start_time": start_time}
         return Response(data=data)
 
-    def create_group_upload_task(self, cur_session, repository, artifact, filename, start_time):
+    def create_group_upload_task(
+        self, cur_session, repository, artifact, filename, attestations, start_time
+    ):
         """Creates the actual task that adds the packages to the index."""
         cur_session["start"] = str(start_time)
-        cur_session["artifacts"] = [(str(artifact.sha256), filename)]
+        cur_session["artifacts"] = [(str(artifact.sha256), filename, attestations)]
         cur_session.modified = False
         cur_session.save()
         task = dispatch(

pulp_python/app/serializers.py

@@ -5,14 +5,20 @@
 from django.db.utils import IntegrityError
 from packaging.requirements import Requirement
 from rest_framework import serializers
-from pydantic import ValidationError
-from pypi_attestations import Distribution, Provenance, VerificationError
+from pydantic import TypeAdapter, ValidationError
+from pypi_attestations import Attestation, VerificationError
 
 from pulpcore.plugin import models as core_models
 from pulpcore.plugin import serializers as core_serializers
-from pulpcore.plugin.util import get_domain
+from pulpcore.plugin.util import get_domain, get_prn, get_current_authenticated_user
 
 from pulp_python.app import models as python_models
+from pulp_python.app.provenance import (
+    Provenance,
+    verify_provenance,
+    AttestationBundle,
+    AnyPublisher,
+)
 from pulp_python.app.utils import (
     DIST_EXTENSIONS,
     artifact_to_python_content_data,
@@ -296,6 +302,46 @@ class PythonPackageContentSerializer(core_serializers.SingleArtifactContentUploa
         allow_null=True,
         help_text=_("The SHA256 digest of the package's METADATA file."),
     )
+    # PEP740 Attestations/Provenance
+    attestations = serializers.JSONField(
+        required=False,
+        help_text=_("A JSON list containing attestations for the package."),
+        write_only=True,
+    )
+    provenance = serializers.SerializerMethodField(
+        read_only=True, help_text=_("The created provenance object on upload.")
+    )
+
+    def get_provenance(self, obj):
+        """Get the provenance for the package."""
+        if provenance := getattr(obj, "provenance", None):
+            return get_prn(provenance)
+        return None
+
+    def validate_attestations(self, value):
+        """Validate the attestations, turn into Attestation objects."""
+        try:
+            if isinstance(value, str):
+                attestations = TypeAdapter(list[Attestation]).validate_json(value)
+            else:
+                attestations = TypeAdapter(list[Attestation]).validate_python(value)
+        except ValidationError as e:
+            raise serializers.ValidationError(_("Invalid attestations: {}".format(e)))
+        return attestations
+
+    def handle_attestations(self, filename, sha256, attestations, offline=False):
+        """Handle converting attestations to a Provenance object."""
+        user = get_current_authenticated_user()
+        publisher = AnyPublisher(kind="Pulp User", prn=get_prn(user))
+        att_bundle = AttestationBundle(publisher=publisher, attestations=attestations)
+        provenance = Provenance(attestation_bundles=[att_bundle])
+        try:
+            verify_provenance(filename, sha256, provenance, offline=offline)
+        except VerificationError as e:
+            raise serializers.ValidationError(
+                {"attestations": _("Attestations failed verification: {}".format(e))}
+            )
+        return provenance.model_dump(mode="json")
 
     def deferred_validate(self, data):
         """
@@ -336,6 +382,8 @@ def deferred_validate(self, data):
             )
 
         data.update(_data)
+        if attestations := data.pop("attestations", None):
+            data["provenance"] = self.handle_attestations(filename, data["sha256"], attestations)
 
         return data
 
@@ -345,6 +393,29 @@ def retrieve(self, validated_data):
         )
         return content.first()
 
+    def create(self, validated_data):
+        """Create new PythonPackageContent object."""
+        repository = validated_data.pop("repository", None)
+        provenance = validated_data.pop("provenance", None)
+        content = super().create(validated_data)
+        if provenance:
+            prov_sha256 = python_models.PackageProvenance.calculate_sha256(provenance)
+            prov_model, _ = python_models.PackageProvenance.objects.get_or_create(
+                sha256=prov_sha256,
+                _pulp_domain=get_domain(),
+                defaults={"package": content, "provenance": provenance},
+            )
+            if core_models.Task.current():
+                core_models.CreatedResource.objects.create(content_object=prov_model)
+            setattr(content, "provenance", prov_model)
+        if repository:
+            repository.cast()
+            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:
+                new_version.add_content(content_to_add)
+        return content
+
     class Meta:
         fields = core_serializers.SingleArtifactContentUploadSerializer.Meta.fields + (
             "author",
@@ -381,6 +452,8 @@ class Meta:
             "size",
             "sha256",
             "metadata_sha256",
+            "attestations",
+            "provenance",
         )
         model = python_models.PythonPackageContent
 
@@ -436,6 +509,10 @@ def validate(self, data):
         data.update(parse_project_metadata(vars(metadata)))
         # Overwrite filename from metadata
         data["filename"] = filename
+        if attestations := data.pop("attestations", None):
+            data["provenance"] = self.handle_attestations(
+                filename, data["sha256"], attestations, offline=True
+            )
         return data
 
     class Meta(PythonPackageContentSerializer.Meta):
@@ -497,13 +574,8 @@ def deferred_validate(self, data):
                 _("The uploaded provenance is not valid: {}".format(e))
             )
         if data.pop("verify"):
-            dist = Distribution(name=data["package"].filename, digest=data["package"].sha256)
             try:
-                for attestation_bundle in provenance.attestation_bundles:
-                    publisher = attestation_bundle.publisher
-                    policy = publisher._as_policy()
-                    for attestation in attestation_bundle.attestations:
-                        attestation.verify(policy, dist)
+                verify_provenance(data["package"].filename, data["package"].sha256, provenance)
             except VerificationError as e:
                 raise serializers.ValidationError(_("Provenance verification failed: {}".format(e)))
         return data