From 0df23e31bd2d1a87852dfe140a52172cd2a6a039 Mon Sep 17 00:00:00 2001 From: Jitka Halova Date: Wed, 12 Aug 2026 14:15:30 +0200 Subject: [PATCH 1/3] 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 + .../api/test_konflux_attestations.py | 248 ++++++++++++++++++ 4 files changed, 412 insertions(+), 8 deletions(-) create mode 100644 pulp_python/tests/functional/api/test_konflux_attestations.py 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 e75475ae..f3d6db94 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__) @@ -110,6 +117,7 @@ def validate(self, data): } ) + sha256 = data.get("sha256_digest") if attestations := data.get("attestations"): try: attestations = TypeAdapter(list[Attestation]).validate_python(attestations) @@ -117,8 +125,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"], diff --git a/pulp_python/tests/functional/api/test_konflux_attestations.py b/pulp_python/tests/functional/api/test_konflux_attestations.py new file mode 100644 index 00000000..2b63aec4 --- /dev/null +++ b/pulp_python/tests/functional/api/test_konflux_attestations.py @@ -0,0 +1,248 @@ +"""Functional tests for Konflux-style attestation verification. + +These tests exercise the attestation / provenance upload paths with +attestations that carry an RSA signature instead of a Sigstore certificate, +mirroring the format produced by Konflux / Calunga builds. + +The test signing key is generated at image build time and the matching +public key is configured as PULP_ATTESTATION_VERIFICATION_KEY so that +signature verification is fully exercised end-to-end. +""" + +import base64 +import json +import os + +import pytest +from cryptography.hazmat.primitives import hashes, serialization +from cryptography.hazmat.primitives.asymmetric import padding as crypto_padding + +from pulpcore.tests.functional.utils import PulpTaskError + +DSSE_PAYLOAD_TYPE = "application/vnd.in-toto+json" + +TEST_PRIVATE_KEY_PATH = "/etc/pki/attestation/test-key-private.pem" +TEST_PUBLIC_KEY_PATH = "/etc/pki/attestation/test-key.pem" + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _build_statement(filename, sha256): + """Build a minimal in-toto statement for a Konflux attestation.""" + return json.dumps( + { + "_type": "https://in-toto.io/Statement/v0.1", + "predicateType": "https://slsa.dev/provenance/v0.2", + "subject": [{"name": filename, "digest": {"sha256": sha256}}], + "predicate": { + "buildType": "https://konflux-ci.dev/PythonWheelBuild@v1", + "builder": {"id": "https://konflux-ci.dev/calunga"}, + }, + } + ).encode() + + +def _pae(payload_type, payload): + """Compute the DSSE Pre-Authentication Encoding.""" + return b"DSSEv1 %d %b %d %b" % ( + len(payload_type), + payload_type.encode(), + len(payload), + payload, + ) + + +def _sign(statement_bytes, private_key): + pae = _pae(DSSE_PAYLOAD_TYPE, statement_bytes) + return private_key.sign( + pae, + crypto_padding.PKCS1v15(), + hashes.SHA256(), + ) + + +def _b64(data: bytes) -> str: + return base64.b64encode(data).decode() + + +def _make_attestation(statement_bytes, signature_bytes): + """Return a single PEP-740 Attestation dict (Konflux flavour).""" + return { + "version": 1, + "verification_material": None, + "envelope": { + "statement": _b64(statement_bytes), + "signature": _b64(signature_bytes), + }, + } + + +def _make_provenance(attestation): + """Wrap an attestation into a full PEP-740 Provenance object.""" + return { + "version": 1, + "attestation_bundles": [ + { + "publisher": {"kind": "Konflux", "builder": "calunga"}, + "attestations": [attestation], + } + ], + } + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + + +@pytest.fixture(scope="session") +def test_private_key(pulp_settings): + """Load the CI-generated test attestation signing key.""" + if not os.path.exists(TEST_PRIVATE_KEY_PATH): + pytest.skip( + f"Test attestation private key not found at {TEST_PRIVATE_KEY_PATH} (not running in CI container?)" + ) + if pulp_settings.ATTESTATION_VERIFICATION_KEY != TEST_PUBLIC_KEY_PATH: + pytest.fail(f"ATTESTATION_VERIFICATION_KEY is not set to {TEST_PUBLIC_KEY_PATH}") + with open(TEST_PRIVATE_KEY_PATH, "rb") as f: + return serialization.load_pem_private_key(f.read(), password=None) + + +@pytest.fixture() +def _provenance_file(tmp_path): + """Return a helper that writes a provenance dict to a temp file.""" + + def _write(provenance_dict): + path = tmp_path / "provenance.json" + path.write_text(json.dumps(provenance_dict)) + return str(path) + + return _write + + +# --------------------------------------------------------------------------- +# Tests +# --------------------------------------------------------------------------- + + +def test_konflux_provenance_stored( + python_bindings, python_content_factory, monitor_task, test_private_key, _provenance_file +): + """A Konflux-style provenance is accepted and stored when verify=True.""" + content = python_content_factory() + + stmt = _build_statement(content.filename, content.sha256) + sig = _sign(stmt, test_private_key) + att = _make_attestation(stmt, sig) + prov = _make_provenance(att) + + task = python_bindings.ContentProvenanceApi.create( + package=content.pulp_href, + file=_provenance_file(prov), + verify=True, + ).task + result = monitor_task(task) + + prov_obj = python_bindings.ContentProvenanceApi.read(result.created_resources[-1]) + assert prov_obj.package == content.pulp_href + stored_att = prov_obj.provenance["attestation_bundles"][0]["attestations"][0] + assert stored_att["envelope"]["statement"] == _b64(stmt) + publisher = prov_obj.provenance["attestation_bundles"][0]["publisher"] + assert publisher["builder"] == "calunga" + assert publisher["kind"] == "Konflux" + + +def test_konflux_wrong_subject_name_rejected( + python_bindings, python_content_factory, monitor_task, test_private_key, _provenance_file +): + """Verification rejects a Konflux attestation whose subject name does not match.""" + content = python_content_factory() + + wrong_name = "wrong-package-0.1.tar.gz" + stmt = _build_statement(wrong_name, content.sha256) + sig = _sign(stmt, test_private_key) + att = _make_attestation(stmt, sig) + prov = _make_provenance(att) + + task = python_bindings.ContentProvenanceApi.create( + package=content.pulp_href, + file=_provenance_file(prov), + verify=True, + ).task + with pytest.raises(PulpTaskError) as exc_info: + monitor_task(task) + assert "subject does not match distribution name" in exc_info.value.task.error["description"] + + +def test_konflux_wrong_digest_rejected( + python_bindings, python_content_factory, monitor_task, test_private_key, _provenance_file +): + """Verification rejects a Konflux attestation whose digest does not match.""" + content = python_content_factory() + + bad_digest = "0" * 64 + stmt = _build_statement(content.filename, bad_digest) + sig = _sign(stmt, test_private_key) + att = _make_attestation(stmt, sig) + prov = _make_provenance(att) + + task = python_bindings.ContentProvenanceApi.create( + package=content.pulp_href, + file=_provenance_file(prov), + verify=True, + ).task + with pytest.raises(PulpTaskError) as exc_info: + monitor_task(task) + assert "subject does not match distribution digest" in exc_info.value.task.error["description"] + + +def test_konflux_bad_signature_rejected( + python_bindings, python_content_factory, monitor_task, test_private_key, _provenance_file +): + """An attestation with a valid subject but tampered signature is rejected.""" + content = python_content_factory() + + stmt = _build_statement(content.filename, content.sha256) + sig = _sign(stmt, test_private_key) + tampered_sig = bytes([b ^ 0xFF for b in sig[:32]]) + sig[32:] + att = _make_attestation(stmt, tampered_sig) + prov = _make_provenance(att) + + task = python_bindings.ContentProvenanceApi.create( + package=content.pulp_href, + file=_provenance_file(prov), + verify=True, + ).task + with pytest.raises(PulpTaskError) as exc_info: + monitor_task(task) + assert "signature verification failed" in exc_info.value.task.error["description"] + + +def test_konflux_attestation_via_content_upload( + python_bindings, python_content_factory, monitor_task, test_private_key +): + """Konflux-style attestations can be uploaded alongside a package via the content API.""" + content = python_content_factory() + + stmt = _build_statement(content.filename, content.sha256) + sig = _sign(stmt, test_private_key) + att = _make_attestation(stmt, sig) + + body = { + "artifact": content.artifact, + "relative_path": content.filename, + "sha256": content.sha256, + "attestations": json.dumps([att]), + } + task = python_bindings.ContentPackagesApi.create(**body).task + result = monitor_task(task) + + assert len(result.created_resources) == 2 + prov_obj = python_bindings.ContentProvenanceApi.read(result.created_resources[1]) + publisher = prov_obj.provenance["attestation_bundles"][0]["publisher"] + assert publisher["builder_id"] == "https://konflux-ci.dev/calunga" + assert publisher["build_type"] == "https://konflux-ci.dev/PythonWheelBuild@v1" + assert publisher["kind"] == "konflux-ci.dev" From 6bb66b91112e7e5079fda0ebfd32064f1c0c2caa Mon Sep 17 00:00:00 2001 From: Jitka Halova Date: Wed, 21 Oct 2026 13:32:15 +0200 Subject: [PATCH 2/3] Upstream patch (to be squashed) Assisted By: Claude Opus 5.5 --- .ci/assets/keys/test-key-private.pem | 52 +++++++ .ci/assets/keys/test-key.pem | 14 ++ .github/workflows/scripts/before_install.sh | 6 +- .gitleaks.toml | 1 + CHANGES/1340.feature | 1 + docs/user/guides/attestation.md | 37 +++++ pulp_python/app/provenance.py | 132 ++++++++---------- pulp_python/app/settings.py | 2 +- ...ons.py => test_key_signed_attestations.py} | 125 +++++++++++++---- template_config.yml | 8 +- 10 files changed, 277 insertions(+), 101 deletions(-) create mode 100644 .ci/assets/keys/test-key-private.pem create mode 100644 .ci/assets/keys/test-key.pem create mode 100644 CHANGES/1340.feature rename pulp_python/tests/functional/api/{test_konflux_attestations.py => test_key_signed_attestations.py} (59%) diff --git a/.ci/assets/keys/test-key-private.pem b/.ci/assets/keys/test-key-private.pem new file mode 100644 index 00000000..d445b5a0 --- /dev/null +++ b/.ci/assets/keys/test-key-private.pem @@ -0,0 +1,52 @@ +-----BEGIN PRIVATE KEY----- +MIIJQgIBADANBgkqhkiG9w0BAQEFAASCCSwwggkoAgEAAoICAQDLYCPSdloA94Te ++68CLqtHJ3qTKttyWyM1uHkcb+AEAExtQGLoKtysmE2QQz8xwpiBKsByAlPTAHZZ +pYIGzof45xEwiRzBqKxE4um1Uhtq+WCth96UeUkjz2G0xMxkFycjHDdpQQ92Hlg+ +mvU8VOrs4Xi7PvG5E+lTY7zez89QQ2ZY3OcbFSHVkC1b0COjYqJ7m5TmVWaRyNbW +Nn+00/uZ+hFzcsxAuKa9B9s4ngWELLA1TnWp5yi4q2Pkold4prI3PwF7IfGWtrfo +TgyzCKji12QLLjbd5KiKecGqx9zaV8xs7GnCif47JmOFAEDW5dtUstepH8ysy8wC +IXoZYvOSOsOv2mVz99JikyaWeV0rjNk7p9UPrhiaGKOfzfn+g6bzCJYBlo425pmJ +S/d1EoBoqgeCGniyLkFByQ9zDmzr1NlCkNKtNII1qNoRSMyqJY0qKPAd2bNMHUfT +FB3Eb/AvcOjb4fa5kkxl4gMQwasNaFmDgusqt1JjHK4XcL8UIcvzGPj8LsDrx5iT +HCFSF1gJvAXiLG0o6KFgrHfTBw2V3mxaKZNGr2NA5IMTs2UeGud590VmSvsKhr6R +tekHHGLCrZT9kPxbicBq6OZOhSb0usSpclSvwMzMSIOFNzACXq33nPYqlY0r1vGN +5Gi2P9W2CxAYbnOGgqpzHbt4YoLVrQIDAQABAoICAA2SR3xZPtL8XCGFKhM7NLLK +2k3NI60TPTDt3nxx+sD0RCVbkT4XniI7skauNh6xSFFWSQFSpnADgoz46R8LKTJd +jl1uyOcjb7DTyO9l9e5tixtetbuysZlyJ/2oJFDe5VMHzwAhwfu1NVj4KOVIcA+J +UZqCg3RA9TuwrCnc3uNm8VgnZZn+ZFjxW2raY4ZuTpQbuGlRHvHGNTqWPcS+C0wl +zoRQZNDs1t5GZ+/0m5RLvHZ82zKQpRGts5IjmIgJ7QVCxGvdwqwMBWRl0PMhi3jt +EVVYVXuj2/C3BKAg4NwGf93E6PR9FjoG6x00/HQFYrLZkbVM1JuzUynRPL9KRcvz +It1vxbmW71oWAXXkWcW7MHWR3n7LoQ8x0zcuf2A1PRuAG4zb8hTDyWud9d10NmSd +eVrNjRix54kW54n39o82dR2heKSYkaM59Z/9NMAHevaZwyQvXwoXLJO4V4iL1GXJ +JmDHd7yUtLTde/5H1jxUsNVvtkYqX4ePbqmw2cpcpu4dVMc4ivweValv6j3Ezokl +J0LdU9pRhpUdA72pbyoij7+5mvdrBm87XM3el9Nf+RhhGAdRVx7jR1FjdWQXfcyW +0LzuHafPL3ns9qQYaOt7LbgVyH08wRupcGKS9iPZM3Wmo7BYLy8fMT3YReum5+xq +bIVvWv7eTUFRUir4thABAoIBAQDvdDd9gscmAtEMfZMFyowZw4K2NvlQ3l5TT0bH +Gu9u4auratK5cGFJ/SQiBZqvyPT6E6q8HwxA6pVoagrGk+RYV0UiZDYdmBAvhQuG +msEiInVn8RYZs+0LnGhOVOdJccvc/CyMnBRb7i3EHEycpV5NaOUxGKfbb0/oDDxS +LcTLZ0L7ds7S4m9DY5AowJXCt/OQidpRidfgl5QbbTsKy+Gay2JjZ9vzee72guG0 +JBpSDlGKeJs/1u0MBM1clMLvEt9H1hR6Ru+KA28bSsJeRZfFvwTRB3Sz40izKAaJ +u86wQitquryAnzdvigT2vaJB7ZeWxliU/Y/KYff/WNG5sipdAoIBAQDZbbg1XSHG +rDsL6oCdHUWPIdqGS/DuuAn2g+YJ6+9HgN1hcFsRoJpDbem6F5W8tz9TCYXLJ/So +pGQKIZwiWnt8QSBS5z3jCZytkzD87CPtO11f9SkB/oFb5NkAcWWx6WCZdNhbaWTL +1Yq4Y0SLeFPt1mz1+dDerPO8EqYTH/N8t7nbJ1/HcoGXJZI7521mowQopqTuyZTG +bLBCKYAcmTjYl3fZ1iC0i1+Poz7BGx+84Rknx+4V2Nze46PfzPS/7t5xw/+iQ7+K +PRMdLydeUEbdEaCtwYCs2lkhw7Bu0SZLpt6mIluXdGUzjwt+TnBGcFqSDEFW8CLR +zD/ijVOaZMORAoIBABKmCmxL8xaSwZUncnvQ+nhHMbbfMSuLJe13DxwSjPMlwCjp +eN/YULtia536scFe9TVEstdT07B6lIg9OfmdKvt2UHwNMem8HgaVZgBlrQTrihk+ +PWpjCOMOm1D+a8Tch/P977pDrZI7SnUrfwv0FRQSR0c7lFcSpDZ+PXRo/BqbQCw1 +ZIYn/GJTLrb9yKwRh3aKReZzxcxIAdDhAOgmWLule1Qiko6zwFiSeOF+rk4Vr2QJ +YI3oPy3gcd3z9/qGjb0afx3GyIEHI3AMsnaFFPzhk45z9jLMUK0jQN8ZMU+o15jI +UkXyIG8fYKOWwTxBNL0ZVWzFEp3AY4APesXrikECggEAcZihZUADJYlWUCN0jqF8 +dgt54DBM8Gu2yNSgmw5pNTJed0n8SnleH1yNgGxSDwauTvPqsvltGa7JlYF50Xj0 +izZ7bNTjwHqFISqFrZ6yJn+diUTM5/3QF/K4bULRnuIPVh117ExkHRq0HyG97iAv +uVMOGnUqayxxKxGTMuq+i6pxr84ifFGW4yD+Bc4jmjwRMCvgf+FRmVmvvOFxzX8/ +8+ku9OCqtakbhDAF2V4SdtwkCGSsPz3OJ6VHOOYb+SsTjNyZ8mzy5YaDNfws1Vmb +rGRJTn6Ke8SYTMuwojUjjOLh8GqC794gAY/6sULJ5gNNToCYopNTncjYl8S+qSt8 +AQKCAQEAgWaRzjuVO4YJ/SAisaItnNov73PgdTM82fkB2uWip/zykdMPu4UKivOe +5/OVKsDgKOdfCI3kP2I+/h//4aiW6JyHtXJkPojD9QLwPC887cmcc2A0726nhmoj +VV5vZpJyTFPktxNc/snVID3uKT0uk4xfBAUNejRjlFFpC3fyxk3rB/0cG2UJWb0A +VapDSXTg+S3kySS4uZbwW19jofDMqFfsVwlIFlMRSkE4jj5+Qs6/P1xnpAceIJVJ +orG/DhLTqAwcRSjmNw/FpKNUbrFZ8NtUVU4XIEzvWmoqDMKrkiHMGOvfzU9jza1f +Y3acysRQ2KI9Ero+Ga3hDfaohueVFg== +-----END PRIVATE KEY----- diff --git a/.ci/assets/keys/test-key.pem b/.ci/assets/keys/test-key.pem new file mode 100644 index 00000000..101aa884 --- /dev/null +++ b/.ci/assets/keys/test-key.pem @@ -0,0 +1,14 @@ +-----BEGIN PUBLIC KEY----- +MIICIjANBgkqhkiG9w0BAQEFAAOCAg8AMIICCgKCAgEAy2Aj0nZaAPeE3vuvAi6r +Ryd6kyrbclsjNbh5HG/gBABMbUBi6CrcrJhNkEM/McKYgSrAcgJT0wB2WaWCBs6H ++OcRMIkcwaisROLptVIbavlgrYfelHlJI89htMTMZBcnIxw3aUEPdh5YPpr1PFTq +7OF4uz7xuRPpU2O83s/PUENmWNznGxUh1ZAtW9Ajo2Kie5uU5lVmkcjW1jZ/tNP7 +mfoRc3LMQLimvQfbOJ4FhCywNU51qecouKtj5KJXeKayNz8BeyHxlra36E4Mswio +4tdkCy423eSoinnBqsfc2lfMbOxpwon+OyZjhQBA1uXbVLLXqR/MrMvMAiF6GWLz +kjrDr9plc/fSYpMmlnldK4zZO6fVD64Ymhijn835/oOm8wiWAZaONuaZiUv3dRKA +aKoHghp4si5BQckPcw5s69TZQpDSrTSCNajaEUjMqiWNKijwHdmzTB1H0xQdxG/w +L3Do2+H2uZJMZeIDEMGrDWhZg4LrKrdSYxyuF3C/FCHL8xj4/C7A68eYkxwhUhdY +CbwF4ixtKOihYKx30wcNld5sWimTRq9jQOSDE7NlHhrnefdFZkr7Coa+kbXpBxxi +wq2U/ZD8W4nAaujmToUm9LrEqXJUr8DMzEiDhTcwAl6t95z2KpWNK9bxjeRotj/V +tgsQGG5zhoKqcx27eGKC1a0CAwEAAQ== +-----END PUBLIC KEY----- diff --git a/.github/workflows/scripts/before_install.sh b/.github/workflows/scripts/before_install.sh index edf0b5bb..a8c98c9c 100755 --- a/.github/workflows/scripts/before_install.sh +++ b/.github/workflows/scripts/before_install.sh @@ -50,7 +50,7 @@ legacy_component_name: "pulp_python" component_name: "python" component_version: "${COMPONENT_VERSION}" pulp_env: {} -pulp_settings: {"allowed_export_paths": "/tmp", "allowed_import_paths": "/tmp", "api_root": "/pulp/", "orphan_protection_time": 0, "pypi_api_hostname": "https://pulp:443"} +pulp_settings: {"allowed_export_paths": "/tmp", "allowed_import_paths": "/tmp", "api_root": "/pulp/", "attestation_verification_keys": {"default": "/etc/pki/attestation/test-key.pem"}, "orphan_protection_time": 0, "pypi_api_hostname": "https://pulp:443"} pulp_scheme: "https" image: name: "pulp" @@ -64,6 +64,10 @@ image: extra_files: - origin: "pulp_python" destination: "pulp_python" + - origin: "pulp_python/.ci/assets/keys/test-key.pem" + destination: "/etc/pki/attestation/test-key.pem" + - origin: "pulp_python/.ci/assets/keys/test-key-private.pem" + destination: "/etc/pki/attestation/test-key-private.pem" services: - name: "pulp" image: "pulp:ci_build" diff --git a/.gitleaks.toml b/.gitleaks.toml index de0b82e7..ae742957 100644 --- a/.gitleaks.toml +++ b/.gitleaks.toml @@ -1,6 +1,7 @@ [allowlist] description = "Allow specific test-keys." paths = [ + '''.ci/assets/keys/test-key-private.pem''', ] regexes = [ '''AKIAIT2Z5TDYPX3ARJBA''', diff --git a/CHANGES/1340.feature b/CHANGES/1340.feature new file mode 100644 index 00000000..da594900 --- /dev/null +++ b/CHANGES/1340.feature @@ -0,0 +1 @@ +Added custom (non-PEP 740) support for verifying attestations signed with a per-domain public key configured via `ATTESTATION_VERIFICATION_KEYS`. diff --git a/docs/user/guides/attestation.md b/docs/user/guides/attestation.md index c8a2fca3..4606590c 100644 --- a/docs/user/guides/attestation.md +++ b/docs/user/guides/attestation.md @@ -89,3 +89,40 @@ http $PULP_API/pypi/foo/simple/twine/ "Accept:application/vnd.pypi.simple.v1+jso http $PULP_API/pypi/foo/integrity/twine/6.2.0/twine-6.2.0.tar.gz/provenance/ ``` + +## Custom Key Verification (non-PEP 740) + +!!! note + This is a custom extension outside the PEP 740 spec. Standard PEP 740 users using Sigstore + keyless signing do not need this feature. + +Pulp Python supports verifying attestations signed with a custom public key instead of a Sigstore +certificate. This is useful for environments that sign packages with their own key (e.g. internal +build systems) rather than using Sigstore's keyless signing. + +To enable this, configure the `ATTESTATION_VERIFICATION_KEYS` setting, which maps a domain name to +the path of a PEM public key file. Each domain trusts only its own key. Without domains enabled, all +content belongs to the `default` domain. + +``` +PULP_ATTESTATION_VERIFICATION_KEYS=@json {"default": "/path/to/public-key.pem"} +``` + +Pulp uses two verification paths that coexist on the same server: + +- **Attestations with a Sigstore certificate** are verified through the standard PEP 740 path, + regardless of `ATTESTATION_VERIFICATION_KEYS`. +- **Attestations without a Sigstore certificate** are verified against the key configured for the + current domain using RSA PKCS1v15 signatures. In a domain without a configured key, they are + rejected. + +This means a single Pulp instance can host packages with Sigstore certificates and +internally-signed packages with key-based attestations. Each attestation is independently +routed to the appropriate verification path based on whether it contains a certificate. +Key-signed attestations must omit `verification_material` or set it to `null`. +Attestations with a certificate that can't be parsed are rejected. + +### Limitations + +- Each domain can have only one key. +- The key-based path only supports RSA PKCS1v15 with SHA-256 signatures. diff --git a/pulp_python/app/provenance.py b/pulp_python/app/provenance.py index 9cbd4128..f3c487f9 100644 --- a/pulp_python/app/provenance.py +++ b/pulp_python/app/provenance.py @@ -1,18 +1,18 @@ import json -import logging from typing import Annotated, Literal, Union, get_args -from urllib.parse import urlparse -from cryptography.exceptions import InvalidSignature +from cryptography.exceptions import InvalidSignature, UnsupportedAlgorithm from cryptography.hazmat.primitives import hashes, serialization from cryptography.hazmat.primitives.asymmetric import padding as crypto_padding +from cryptography.hazmat.primitives.asymmetric import rsa 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 as _UpstreamAttestation from pypi_attestations import ( Distribution, - Envelope, + Envelope, # noqa - needed in module namespace for Pydantic model rebuild Publisher, VerificationError, VerificationMaterial, @@ -20,12 +20,10 @@ from sigstore.dsse import Envelope as DSSEEnvelope from sigstore.dsse import _pae -log = logging.getLogger(__name__) +from pulpcore.plugin.util import get_domain _verification_key_cache = {} -SLSA_PROVENANCE_V02 = "https://slsa.dev/provenance/v0.2" - class _PermissivePolicy: """A permissive verification policy that always succeeds.""" @@ -57,12 +55,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 +69,6 @@ class Attestation(BaseModel): Cryptographic materials used to verify `message_signature`. """ - envelope: Envelope - """ - The enveloped attestation statement and signature. - """ - class AttestationBundle(BaseModel): """ @@ -95,35 +89,48 @@ 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) +def _load_verification_key(domain_name): + """Load the attestation verification public key configured for the domain, with caching.""" + keys = settings.ATTESTATION_VERIFICATION_KEYS + if not isinstance(keys, dict): + raise VerificationError( + "ATTESTATION_VERIFICATION_KEYS must map domain names to public key paths" + ) + key_path = keys.get(domain_name) 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()) + try: + with open(key_path, "rb") as f: + key = serialization.load_pem_public_key(f.read()) + except (OSError, ValueError, UnsupportedAlgorithm) as e: + raise VerificationError( + f"could not load the verification key for domain '{domain_name}' " + f"from {key_path}: {e}" + ) + if not isinstance(key, rsa.RSAPublicKey): + raise VerificationError( + f"the verification key for domain '{domain_name}' in {key_path} " + "must be an RSA public key" + ) + _verification_key_cache[key_path] = key 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): +def _has_certificate(attestation): + """Check whether the attestation carries a certificate, rejecting unparseable ones.""" + vm = attestation.verification_material + if vm is None: return False + try: + load_der_x509_certificate(vm.certificate) + except ValueError as e: + raise VerificationError(f"invalid X.509 certificate in verification_material: {e}") + return True def _verify_statement_subject(attestation, dist): - """Validate that the in-toto statement subject matches the distribution. - - Returns the parsed statement dict for downstream use. - """ + """Validate that the in-toto statement subject matches the distribution.""" try: stmt = json.loads(attestation.envelope.statement) except (json.JSONDecodeError, UnicodeDecodeError) as e: @@ -142,30 +149,6 @@ def _verify_statement_subject(attestation, dist): 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.""" @@ -179,30 +162,37 @@ def _verify_signature(attestation, public_key): crypto_padding.PKCS1v15(), hashes.SHA256(), ) - except InvalidSignature as e: - raise VerificationError(f"signature verification failed: {e}") + except InvalidSignature: + raise VerificationError("signature verification failed") def verify_provenance(filename, sha256, provenance, offline=True): - """Verify the provenance object is valid for the package.""" + """Verify the provenance object is valid for the package. + + Each attestation is routed to one of two coexisting verification paths: + 1. Attestations with a valid Sigstore certificate follow the standard PEP 740 path. + 2. Attestations without a certificate are verified against the key configured for the + current domain in ATTESTATION_VERIFICATION_KEYS using RSA PKCS1v15 signatures. + Attestations with a certificate that can't be parsed are rejected. + """ 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: - if _has_valid_certificate(attestation): - policy = publisher._as_policy() + if _has_certificate(attestation): 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) + domain_name = get_domain().name + verification_key = _load_verification_key(domain_name) if verification_key: + _verify_statement_subject(attestation, dist) _verify_signature(attestation, verification_key) else: - log.warning( - "Attestation without valid certificate accepted without " - "signature verification (ATTESTATION_VERIFICATION_KEY not set)" + raise VerificationError( + "attestation has no Sigstore certificate, and " + f"ATTESTATION_VERIFICATION_KEYS has no key for domain '{domain_name}'" ) diff --git a/pulp_python/app/settings.py b/pulp_python/app/settings.py index 65fedd28..80a393d7 100644 --- a/pulp_python/app/settings.py +++ b/pulp_python/app/settings.py @@ -4,7 +4,7 @@ PYPI_API_HOSTNAME = "https://" + socket.getfqdn() PYPI_PATH_PREFIX = "/pypi/" -ATTESTATION_VERIFICATION_KEY = None +ATTESTATION_VERIFICATION_KEYS = {} DRF_ACCESS_POLICY = { "dynaconf_merge_unique": True, diff --git a/pulp_python/tests/functional/api/test_konflux_attestations.py b/pulp_python/tests/functional/api/test_key_signed_attestations.py similarity index 59% rename from pulp_python/tests/functional/api/test_konflux_attestations.py rename to pulp_python/tests/functional/api/test_key_signed_attestations.py index 2b63aec4..a20c7d9d 100644 --- a/pulp_python/tests/functional/api/test_konflux_attestations.py +++ b/pulp_python/tests/functional/api/test_key_signed_attestations.py @@ -1,24 +1,27 @@ -"""Functional tests for Konflux-style attestation verification. +"""Functional tests for verifying attestations signed with a server-configured key. These tests exercise the attestation / provenance upload paths with -attestations that carry an RSA signature instead of a Sigstore certificate, -mirroring the format produced by Konflux / Calunga builds. +attestations that carry an RSA signature instead of a Sigstore certificate. -The test signing key is generated at image build time and the matching -public key is configured as PULP_ATTESTATION_VERIFICATION_KEY so that +A static test keypair is shipped in .ci/assets/keys/ and the matching +public key is configured for the default domain in PULP_ATTESTATION_VERIFICATION_KEYS so that signature verification is fully exercised end-to-end. """ import base64 +import hashlib import json import os import pytest from cryptography.hazmat.primitives import hashes, serialization from cryptography.hazmat.primitives.asymmetric import padding as crypto_padding +from cryptography.hazmat.primitives.asymmetric import rsa from pulpcore.tests.functional.utils import PulpTaskError +from pulp_python.tests.functional.constants import PYTHON_EGG_FILENAME, PYTHON_URL + DSSE_PAYLOAD_TYPE = "application/vnd.in-toto+json" TEST_PRIVATE_KEY_PATH = "/etc/pki/attestation/test-key-private.pem" @@ -31,15 +34,15 @@ def _build_statement(filename, sha256): - """Build a minimal in-toto statement for a Konflux attestation.""" + """Build a minimal in-toto statement with an SLSA v1 provenance predicate.""" return json.dumps( { - "_type": "https://in-toto.io/Statement/v0.1", - "predicateType": "https://slsa.dev/provenance/v0.2", + "_type": "https://in-toto.io/Statement/v1", + "predicateType": "https://slsa.dev/provenance/v1", "subject": [{"name": filename, "digest": {"sha256": sha256}}], "predicate": { - "buildType": "https://konflux-ci.dev/PythonWheelBuild@v1", - "builder": {"id": "https://konflux-ci.dev/calunga"}, + "buildDefinition": {"buildType": "https://konflux-ci.dev/PythonWheelBuild@v1"}, + "runDetails": {"builder": {"id": "https://konflux-ci.dev/calunga"}}, }, } ).encode() @@ -69,7 +72,7 @@ def _b64(data: bytes) -> str: def _make_attestation(statement_bytes, signature_bytes): - """Return a single PEP-740 Attestation dict (Konflux flavour).""" + """Return a single key-signed Attestation dict without verification material.""" return { "version": 1, "verification_material": None, @@ -100,13 +103,15 @@ def _make_provenance(attestation): @pytest.fixture(scope="session") def test_private_key(pulp_settings): - """Load the CI-generated test attestation signing key.""" + """Load the static test attestation signing key from .ci/assets/keys/.""" if not os.path.exists(TEST_PRIVATE_KEY_PATH): pytest.skip( f"Test attestation private key not found at {TEST_PRIVATE_KEY_PATH} (not running in CI container?)" ) - if pulp_settings.ATTESTATION_VERIFICATION_KEY != TEST_PUBLIC_KEY_PATH: - pytest.fail(f"ATTESTATION_VERIFICATION_KEY is not set to {TEST_PUBLIC_KEY_PATH}") + if pulp_settings.ATTESTATION_VERIFICATION_KEYS.get("default") != TEST_PUBLIC_KEY_PATH: + pytest.fail( + f"ATTESTATION_VERIFICATION_KEYS does not map the default domain to {TEST_PUBLIC_KEY_PATH}" + ) with open(TEST_PRIVATE_KEY_PATH, "rb") as f: return serialization.load_pem_private_key(f.read(), password=None) @@ -128,10 +133,10 @@ def _write(provenance_dict): # --------------------------------------------------------------------------- -def test_konflux_provenance_stored( +def test_key_signed_provenance_stored( python_bindings, python_content_factory, monitor_task, test_private_key, _provenance_file ): - """A Konflux-style provenance is accepted and stored when verify=True.""" + """A key-signed provenance is accepted and stored when verify=True.""" content = python_content_factory() stmt = _build_statement(content.filename, content.sha256) @@ -155,10 +160,10 @@ def test_konflux_provenance_stored( assert publisher["kind"] == "Konflux" -def test_konflux_wrong_subject_name_rejected( +def test_key_signed_wrong_subject_name_rejected( python_bindings, python_content_factory, monitor_task, test_private_key, _provenance_file ): - """Verification rejects a Konflux attestation whose subject name does not match.""" + """Verification rejects a key-signed attestation whose subject name does not match.""" content = python_content_factory() wrong_name = "wrong-package-0.1.tar.gz" @@ -177,10 +182,10 @@ def test_konflux_wrong_subject_name_rejected( assert "subject does not match distribution name" in exc_info.value.task.error["description"] -def test_konflux_wrong_digest_rejected( +def test_key_signed_wrong_digest_rejected( python_bindings, python_content_factory, monitor_task, test_private_key, _provenance_file ): - """Verification rejects a Konflux attestation whose digest does not match.""" + """Verification rejects a key-signed attestation whose digest does not match.""" content = python_content_factory() bad_digest = "0" * 64 @@ -199,10 +204,10 @@ def test_konflux_wrong_digest_rejected( assert "subject does not match distribution digest" in exc_info.value.task.error["description"] -def test_konflux_bad_signature_rejected( +def test_key_signed_bad_signature_rejected( python_bindings, python_content_factory, monitor_task, test_private_key, _provenance_file ): - """An attestation with a valid subject but tampered signature is rejected.""" + """A key-signed attestation with a valid subject but tampered signature is rejected.""" content = python_content_factory() stmt = _build_statement(content.filename, content.sha256) @@ -221,10 +226,10 @@ def test_konflux_bad_signature_rejected( assert "signature verification failed" in exc_info.value.task.error["description"] -def test_konflux_attestation_via_content_upload( +def test_key_signed_attestation_via_content_upload( python_bindings, python_content_factory, monitor_task, test_private_key ): - """Konflux-style attestations can be uploaded alongside a package via the content API.""" + """Key-signed attestations can be uploaded alongside a package via the content API.""" content = python_content_factory() stmt = _build_statement(content.filename, content.sha256) @@ -243,6 +248,72 @@ def test_konflux_attestation_via_content_upload( assert len(result.created_resources) == 2 prov_obj = python_bindings.ContentProvenanceApi.read(result.created_resources[1]) publisher = prov_obj.provenance["attestation_bundles"][0]["publisher"] - assert publisher["builder_id"] == "https://konflux-ci.dev/calunga" - assert publisher["build_type"] == "https://konflux-ci.dev/PythonWheelBuild@v1" - assert publisher["kind"] == "konflux-ci.dev" + assert publisher["kind"] == "Pulp User" + + +def test_invalid_certificate_not_verified_with_key( + python_bindings, python_content_factory, monitor_task, test_private_key, _provenance_file +): + """An attestation with an unparseable certificate is rejected, not verified with the key.""" + content = python_content_factory() + + stmt = _build_statement(content.filename, content.sha256) + att = _make_attestation(stmt, _sign(stmt, test_private_key)) + att["verification_material"] = { + "certificate": _b64(b"not-a-certificate"), + "transparency_entries": [{"log_index": 0}], + } + prov = _make_provenance(att) + + task = python_bindings.ContentProvenanceApi.create( + package=content.pulp_href, + file=_provenance_file(prov), + verify=True, + ).task + with pytest.raises(PulpTaskError) as exc_info: + monitor_task(task) + assert "invalid X.509 certificate" in exc_info.value.task.error["description"] + + +def test_key_signed_attestation_with_other_key_rejected( + python_bindings, python_content_factory, monitor_task, test_private_key, _provenance_file +): + """An attestation signed with a key other than the domain's configured key is rejected.""" + content = python_content_factory() + other_key = rsa.generate_private_key(public_exponent=65537, key_size=2048) + + stmt = _build_statement(content.filename, content.sha256) + att = _make_attestation(stmt, _sign(stmt, other_key)) + prov = _make_provenance(att) + + task = python_bindings.ContentProvenanceApi.create( + package=content.pulp_href, + file=_provenance_file(prov), + verify=True, + ).task + with pytest.raises(PulpTaskError) as exc_info: + monitor_task(task) + assert "signature verification failed" in exc_info.value.task.error["description"] + + +def test_key_signed_attestation_rejected_in_domain_without_key( + python_bindings, domain_factory, download_python_file, monitor_task, test_private_key +): + """A key configured for the default domain is not trusted in another domain.""" + domain = domain_factory() + python_file = download_python_file(PYTHON_EGG_FILENAME, PYTHON_URL) + with open(python_file, "rb") as f: + sha256 = hashlib.sha256(f.read()).hexdigest() + + stmt = _build_statement(PYTHON_EGG_FILENAME, sha256) + att = _make_attestation(stmt, _sign(stmt, test_private_key)) + body = { + "file": python_file, + "relative_path": PYTHON_EGG_FILENAME, + "attestations": json.dumps([att]), + } + task = python_bindings.ContentPackagesApi.create(**body, pulp_domain=domain.name).task + with pytest.raises(PulpTaskError) as exc_info: + monitor_task(task) + err_msg = f"ATTESTATION_VERIFICATION_KEYS has no key for domain '{domain.name}'" + assert err_msg in exc_info.value.task.error["description"] diff --git a/template_config.yml b/template_config.yml index de9ee816..39e49174 100644 --- a/template_config.yml +++ b/template_config.yml @@ -20,7 +20,11 @@ deploy_client_to_rubygems: true deploy_to_pypi: true disabled_redis_runners: [] docker_fixtures: false -extra_files: [] +extra_files: + - origin: "pulp_python/.ci/assets/keys/test-key.pem" + destination: "/etc/pki/attestation/test-key.pem" + - origin: "pulp_python/.ci/assets/keys/test-key-private.pem" + destination: "/etc/pki/attestation/test-key-private.pem" github_org: "pulp" latest_release_branch: "3.37" lint_ignore: [] @@ -43,6 +47,8 @@ pulp_settings: allowed_import_paths: "/tmp" api_root: "/pulp/" orphan_protection_time: 0 + attestation_verification_keys: + default: "/etc/pki/attestation/test-key.pem" pypi_api_hostname: "https://pulp:443" pulp_settings_azure: MEDIA_ROOT: "" From f0d7c8827b957c589c1afa512561cca12bb3e3a7 Mon Sep 17 00:00:00 2001 From: Jitka Halova Date: Thu, 22 Oct 2026 15:41:49 +0200 Subject: [PATCH 3/3] Adress rabbit (to be squashed) Assisted By: Claude Opus 5.5 --- docs/admin/reference/settings.md | 6 +++ docs/user/guides/attestation.md | 11 ++++- pulp_python/app/provenance.py | 31 +++++++++---- .../api/test_key_signed_attestations.py | 46 +++++++++++++++++++ 4 files changed, 83 insertions(+), 11 deletions(-) diff --git a/docs/admin/reference/settings.md b/docs/admin/reference/settings.md index 914e498c..ff46edd9 100644 --- a/docs/admin/reference/settings.md +++ b/docs/admin/reference/settings.md @@ -2,6 +2,12 @@ `pulp_python` adds configuration options to those offered by `pulpcore`. +## ATTESTATION_VERIFICATION_KEYS + +> A mapping of domain names to PEM public key files used to verify attestations signed with a +> custom key. See [Custom Key Verification](site:pulp_python/docs/user/guides/attestation/) for +> details. Defaults to `{}`. + ## PYTHON_GROUP_UPLOADS > This setting controls whether the uploading of packages through the PyPI APIs can be grouped diff --git a/docs/user/guides/attestation.md b/docs/user/guides/attestation.md index 4606590c..25792d8a 100644 --- a/docs/user/guides/attestation.md +++ b/docs/user/guides/attestation.md @@ -104,8 +104,14 @@ To enable this, configure the `ATTESTATION_VERIFICATION_KEYS` setting, which map the path of a PEM public key file. Each domain trusts only its own key. Without domains enabled, all content belongs to the `default` domain. +```python +# settings.py +ATTESTATION_VERIFICATION_KEYS = {"default": "/path/to/public-key.pem"} ``` -PULP_ATTESTATION_VERIFICATION_KEYS=@json {"default": "/path/to/public-key.pem"} + +```bash +# or as an environment variable +export PULP_ATTESTATION_VERIFICATION_KEYS='@json {"default": "/path/to/public-key.pem"}' ``` Pulp uses two verification paths that coexist on the same server: @@ -120,6 +126,9 @@ This means a single Pulp instance can host packages with Sigstore certificates a internally-signed packages with key-based attestations. Each attestation is independently routed to the appropriate verification path based on whether it contains a certificate. Key-signed attestations must omit `verification_material` or set it to `null`. +After downloading, key-signed attestations cannot be verified with `sigstore` or +`pypi-attestations`, since both require a certificate. Consumers must verify them against the same +public key, with a tool that supports DSSE signatures, such as `cosign`. Attestations with a certificate that can't be parsed are rejected. ### Limitations diff --git a/pulp_python/app/provenance.py b/pulp_python/app/provenance.py index f3c487f9..67433d56 100644 --- a/pulp_python/app/provenance.py +++ b/pulp_python/app/provenance.py @@ -48,6 +48,8 @@ def _as_policy(self): # Get the underlying Union type of the original Publisher # Publisher is Annotated[Union[...], Field(discriminator="kind")] _OriginalPublisherTypes = get_args(Publisher.__origin__) +# Get the kinds of the original publisher types, they can only be proven by a Sigstore certificate +_TrustedPublisherKinds = {t.model_fields["kind"].default for t in _OriginalPublisherTypes} # Add AnyPublisher to the list of original publisher types _ExtendedPublisherTypes = (*_OriginalPublisherTypes, AnyPublisher) _ExtendedPublisherUnion = Union[_ExtendedPublisherTypes] @@ -136,18 +138,23 @@ def _verify_statement_subject(attestation, dist): 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") + try: + 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}") + 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") + digest = subject.get("digest", {}).get("sha256") + if digest != dist.digest: + raise VerificationError("subject does not match distribution digest") + except (AttributeError, TypeError) as e: + raise VerificationError(f"malformed statement: {e}") def _verify_signature(attestation, public_key): @@ -186,6 +193,10 @@ def verify_provenance(filename, sha256, provenance, offline=True): staging = "sigstage.dev" in checkpoint.envelope attestation.verify(policy, dist, staging=staging, offline=offline) else: + if publisher.kind in _TrustedPublisherKinds: + raise VerificationError( + f"publisher kind '{publisher.kind}' requires a Sigstore certificate" + ) domain_name = get_domain().name verification_key = _load_verification_key(domain_name) if verification_key: diff --git a/pulp_python/tests/functional/api/test_key_signed_attestations.py b/pulp_python/tests/functional/api/test_key_signed_attestations.py index a20c7d9d..2787bb9b 100644 --- a/pulp_python/tests/functional/api/test_key_signed_attestations.py +++ b/pulp_python/tests/functional/api/test_key_signed_attestations.py @@ -317,3 +317,49 @@ def test_key_signed_attestation_rejected_in_domain_without_key( monitor_task(task) err_msg = f"ATTESTATION_VERIFICATION_KEYS has no key for domain '{domain.name}'" assert err_msg in exc_info.value.task.error["description"] + + +def test_key_signed_attestation_with_trusted_publisher_rejected( + python_bindings, python_content_factory, monitor_task, test_private_key, _provenance_file +): + """A key-signed attestation cannot claim a Trusted Publisher, which needs a certificate.""" + content = python_content_factory() + + stmt = _build_statement(content.filename, content.sha256) + att = _make_attestation(stmt, _sign(stmt, test_private_key)) + prov = _make_provenance(att) + prov["attestation_bundles"][0]["publisher"] = { + "kind": "GitHub", + "repository": "pypa/sampleproject", + "workflow": "release.yml", + } + + task = python_bindings.ContentProvenanceApi.create( + package=content.pulp_href, + file=_provenance_file(prov), + verify=True, + ).task + with pytest.raises(PulpTaskError) as exc_info: + monitor_task(task) + err_msg = "publisher kind 'GitHub' requires a Sigstore certificate" + assert err_msg in exc_info.value.task.error["description"] + + +def test_key_signed_malformed_subject_rejected( + python_bindings, python_content_factory, monitor_task, test_private_key, _provenance_file +): + """A key-signed attestation with a malformed subject is rejected, not crashed on.""" + content = python_content_factory() + + stmt = json.dumps({"_type": "https://in-toto.io/Statement/v1", "subject": None}).encode() + att = _make_attestation(stmt, _sign(stmt, test_private_key)) + prov = _make_provenance(att) + + task = python_bindings.ContentProvenanceApi.create( + package=content.pulp_href, + file=_provenance_file(prov), + verify=True, + ).task + with pytest.raises(PulpTaskError) as exc_info: + monitor_task(task) + assert "malformed statement" in exc_info.value.task.error["description"]