From 4714f77f2848f33a111122be84b667b2a34f95a2 Mon Sep 17 00:00:00 2001 From: Jitka Halova Date: Wed, 12 Aug 2026 14:15:30 +0200 Subject: [PATCH 1/2] Apply pulp-service patch 0048 + ruff changes --- pulp_python/app/provenance.py | 149 ++++++++++++++++++++++++++-- pulp_python/app/pypi/serializers.py | 21 +++- pulp_python/app/settings.py | 2 + 3 files changed, 164 insertions(+), 8 deletions(-) diff --git a/pulp_python/app/provenance.py b/pulp_python/app/provenance.py index 41e1c206..9cbd4128 100644 --- a/pulp_python/app/provenance.py +++ b/pulp_python/app/provenance.py @@ -1,12 +1,30 @@ +import json +import logging from typing import Annotated, Literal, Union, get_args +from urllib.parse import urlparse +from cryptography.exceptions import InvalidSignature +from cryptography.hazmat.primitives import hashes, serialization +from cryptography.hazmat.primitives.asymmetric import padding as crypto_padding +from cryptography.x509 import load_der_x509_certificate +from django.conf import settings from pydantic import BaseModel, ConfigDict, Field from pydantic.alias_generators import to_snake from pypi_attestations import ( - Attestation, Distribution, + Envelope, Publisher, + VerificationError, + VerificationMaterial, ) +from sigstore.dsse import Envelope as DSSEEnvelope +from sigstore.dsse import _pae + +log = logging.getLogger(__name__) + +_verification_key_cache = {} + +SLSA_PROVENANCE_V02 = "https://slsa.dev/provenance/v0.2" class _PermissivePolicy: @@ -39,6 +57,25 @@ def _as_policy(self): ExtendedPublisher = Annotated[_ExtendedPublisherUnion, Field(union_mode="left_to_right")] +class Attestation(BaseModel): + """Attestation object as defined in PEP 740.""" + + version: Literal[1] + """ + The attestation format's version, which is always 1. + """ + + verification_material: VerificationMaterial | None = None + """ + Cryptographic materials used to verify `message_signature`. + """ + + envelope: Envelope + """ + The enveloped attestation statement and signature. + """ + + class AttestationBundle(BaseModel): """ AttestationBundle object as defined in PEP740. @@ -58,14 +95,114 @@ class Provenance(BaseModel): attestation_bundles: list[AttestationBundle] +def _load_verification_key(): + """Load the configured attestation verification public key, with caching.""" + key_path = getattr(settings, "ATTESTATION_VERIFICATION_KEY", None) + if not key_path: + return None + if key_path not in _verification_key_cache: + with open(key_path, "rb") as f: + _verification_key_cache[key_path] = serialization.load_pem_public_key(f.read()) + return _verification_key_cache[key_path] + + +def _has_valid_certificate(attestation): + """Check whether the attestation contains a valid X.509 certificate.""" + try: + vm = attestation.verification_material + if vm is None: + return False + cert_bytes = vm.certificate + load_der_x509_certificate(cert_bytes) + return True + except (ValueError, Exception): + return False + + +def _verify_statement_subject(attestation, dist): + """Validate that the in-toto statement subject matches the distribution. + + Returns the parsed statement dict for downstream use. + """ + try: + stmt = json.loads(attestation.envelope.statement) + except (json.JSONDecodeError, UnicodeDecodeError) as e: + raise VerificationError(f"invalid statement: {e}") + + subjects = stmt.get("subject", []) + if len(subjects) != 1: + raise VerificationError("expected exactly one subject in statement") + + subject = subjects[0] + name = subject.get("name", "") + if name != dist.name: + raise VerificationError(f"subject does not match distribution name: {name} != {dist.name}") + + digest = subject.get("digest", {}).get("sha256") + if digest != dist.digest: + raise VerificationError("subject does not match distribution digest") + + return stmt + + +def _enrich_publisher_from_statement(stmt, publisher): + """Populate publisher fields from an SLSA v0.2 provenance statement.""" + if stmt.get("predicateType") != SLSA_PROVENANCE_V02: + return + + predicate = stmt.get("predicate", {}) + builder_id = predicate.get("builder", {}).get("id") + build_type = predicate.get("buildType") + + if builder_id: + publisher.builder_id = builder_id + try: + hostname = urlparse(builder_id).hostname + if hostname: + publisher.kind = hostname + except Exception: + pass + + if build_type: + publisher.build_type = build_type + + +def _verify_signature(attestation, public_key): + """Verify the attestation's RSA signature over the DSSE PAE bytes.""" + statement_bytes = attestation.envelope.statement + signature_bytes = attestation.envelope.signature + pae = _pae(DSSEEnvelope._TYPE, statement_bytes) + try: + public_key.verify( + signature_bytes, + pae, + crypto_padding.PKCS1v15(), + hashes.SHA256(), + ) + except InvalidSignature as e: + raise VerificationError(f"signature verification failed: {e}") + + def verify_provenance(filename, sha256, provenance, offline=True): """Verify the provenance object is valid for the package.""" dist = Distribution(name=filename, digest=sha256) + verification_key = _load_verification_key() for bundle in provenance.attestation_bundles: publisher = bundle.publisher - policy = publisher._as_policy() for attestation in bundle.attestations: - sig_bundle = attestation.to_bundle() - checkpoint = sig_bundle.log_entry._inner.inclusion_proof.checkpoint - staging = "sigstage.dev" in checkpoint.envelope - attestation.verify(policy, dist, staging=staging, offline=offline) + if _has_valid_certificate(attestation): + policy = publisher._as_policy() + sig_bundle = attestation.to_bundle() + checkpoint = sig_bundle.log_entry._inner.inclusion_proof.checkpoint + staging = "sigstage.dev" in checkpoint.envelope + attestation.verify(policy, dist, staging=staging, offline=offline) + else: + stmt = _verify_statement_subject(attestation, dist) + _enrich_publisher_from_statement(stmt, publisher) + if verification_key: + _verify_signature(attestation, verification_key) + else: + log.warning( + "Attestation without valid certificate accepted without " + "signature verification (ATTESTATION_VERIFICATION_KEY not set)" + ) diff --git a/pulp_python/app/pypi/serializers.py b/pulp_python/app/pypi/serializers.py index bfa1a0ae..6261a6e2 100644 --- a/pulp_python/app/pypi/serializers.py +++ b/pulp_python/app/pypi/serializers.py @@ -3,12 +3,19 @@ from django.db.utils import IntegrityError from pydantic import TypeAdapter, ValidationError +from pypi_attestations import AttestationError from rest_framework import serializers from pulpcore.plugin.models import Artifact from pulpcore.plugin.util import get_domain -from pulp_python.app.provenance import Attestation +from pulp_python.app.provenance import ( + AnyPublisher, + Attestation, + AttestationBundle, + Provenance, + verify_provenance, +) from pulp_python.app.utils import DIST_EXTENSIONS, SUPPORTED_METADATA_VERSIONS log = logging.getLogger(__name__) @@ -107,6 +114,7 @@ def validate(self, data): } ) + sha256 = data.get("sha256_digest") if attestations := data.get("attestations"): try: attestations = TypeAdapter(list[Attestation]).validate_python(attestations) @@ -114,8 +122,17 @@ def validate(self, data): raise serializers.ValidationError( {"attestations": _("The uploaded attestations are not valid: {}").format(e)} ) + if attestations and sha256: + publisher = AnyPublisher(kind="Pulp User") + att_bundle = AttestationBundle(publisher=publisher, attestations=attestations) + provenance = Provenance(attestation_bundles=[att_bundle]) + try: + verify_provenance(file.name, sha256, provenance, offline=True) + except AttestationError as e: + raise serializers.ValidationError( + {"attestations": _("Attestations failed verification: {}").format(e)} + ) - sha256 = data.get("sha256_digest") digests = {"sha256": sha256} if sha256 else None artifact = Artifact.init_and_validate(file, expected_digests=digests) try: diff --git a/pulp_python/app/settings.py b/pulp_python/app/settings.py index c45438d5..65fedd28 100644 --- a/pulp_python/app/settings.py +++ b/pulp_python/app/settings.py @@ -4,6 +4,8 @@ PYPI_API_HOSTNAME = "https://" + socket.getfqdn() PYPI_PATH_PREFIX = "/pypi/" +ATTESTATION_VERIFICATION_KEY = None + DRF_ACCESS_POLICY = { "dynaconf_merge_unique": True, "reusable_conditions": ["pulp_python.app.global_access_conditions"], From 97ef8dd3288023e035cbf582b6b49301ca35cb0a Mon Sep 17 00:00:00 2001 From: Jitka Halova Date: Thu, 13 Aug 2026 14:20:11 +0200 Subject: [PATCH 2/2] Inherit upstream Attestation to keep Sigstore path --- pulp_python/app/provenance.py | 19 ++++++++----------- 1 file changed, 8 insertions(+), 11 deletions(-) diff --git a/pulp_python/app/provenance.py b/pulp_python/app/provenance.py index 9cbd4128..12f6d8ee 100644 --- a/pulp_python/app/provenance.py +++ b/pulp_python/app/provenance.py @@ -10,9 +10,10 @@ from django.conf import settings from pydantic import BaseModel, ConfigDict, Field from pydantic.alias_generators import to_snake +from pypi_attestations import Attestation as _UpstreamAttestation from pypi_attestations import ( Distribution, - Envelope, + Envelope, # noqa - needed in module namespace for Pydantic model rebuild Publisher, VerificationError, VerificationMaterial, @@ -57,12 +58,13 @@ def _as_policy(self): ExtendedPublisher = Annotated[_ExtendedPublisherUnion, Field(union_mode="left_to_right")] -class Attestation(BaseModel): - """Attestation object as defined in PEP 740.""" - - version: Literal[1] +class Attestation(_UpstreamAttestation): """ - The attestation format's version, which is always 1. + Attestation object as defined in PEP 740. + + Inherits from the upstream pypi_attestations.Attestation to keep Sigstore + verification methods (to_bundle, verify), but makes verification_material + optional to support attestations signed with a custom key instead of Sigstore. """ verification_material: VerificationMaterial | None = None @@ -70,11 +72,6 @@ class Attestation(BaseModel): Cryptographic materials used to verify `message_signature`. """ - envelope: Envelope - """ - The enveloped attestation statement and signature. - """ - class AttestationBundle(BaseModel): """