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..54488434 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_key": "/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..d5dcdc18 --- /dev/null +++ b/CHANGES/1340.feature @@ -0,0 +1 @@ +Added support for verifying SLSA attestations with a configured public key. diff --git a/pulp_python/app/provenance.py b/pulp_python/app/provenance.py index 41e1c206..1c1eb78d 100644 --- a/pulp_python/app/provenance.py +++ b/pulp_python/app/provenance.py @@ -1,12 +1,50 @@ +import json +import logging from typing import Annotated, Literal, Union, get_args +from urllib.parse import urlparse -from pydantic import BaseModel, ConfigDict, Field +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 Base64Bytes, BaseModel, ConfigDict, Field, model_validator from pydantic.alias_generators import to_snake +from pypi_attestations import Attestation as _UpstreamAttestation from pypi_attestations import ( - Attestation, Distribution, + Envelope, # noqa - needed in module namespace for Pydantic model rebuild Publisher, + VerificationError, ) +from pypi_attestations import VerificationMaterial as _UpstreamVerificationMaterial +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 VerificationMaterial(_UpstreamVerificationMaterial): + """Extended verification material that supports optional certificate and public key. + + PEP 740 requires a certificate, but this extension allows attestations signed + with a custom key where the certificate is absent. + """ + + certificate: Base64Bytes | None = None + public_key: str | None = None + + @model_validator(mode="after") + def _require_certificate_or_public_key(self): + if self.certificate is None and self.public_key is None: + raise ValueError( + "verification_material must contain either a certificate or a public_key" + ) + return self class _PermissivePolicy: @@ -39,6 +77,21 @@ def _as_policy(self): ExtendedPublisher = Annotated[_ExtendedPublisherUnion, Field(union_mode="left_to_right")] +class Attestation(_UpstreamAttestation): + """ + 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 + """ + Cryptographic materials used to verify `message_signature`. + """ + + class AttestationBundle(BaseModel): """ AttestationBundle object as defined in PEP740. @@ -58,14 +111,148 @@ 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 + if cert_bytes is None: + return False + 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_embedded_key(attestation, server_key): + """Verify that the embedded public key matches the server-configured key.""" + vm = attestation.verification_material + if vm is None or vm.public_key is None: + return + try: + embedded_key = serialization.load_pem_public_key(vm.public_key.encode()) + except (ValueError, Exception) as e: + raise VerificationError(f"invalid embedded public key: {e}") + server_key_bytes = server_key.public_bytes( + serialization.Encoding.PEM, + serialization.PublicFormat.SubjectPublicKeyInfo, + ) + embedded_key_bytes = embedded_key.public_bytes( + serialization.Encoding.PEM, + serialization.PublicFormat.SubjectPublicKeyInfo, + ) + if server_key_bytes != embedded_key_bytes: + raise VerificationError( + "embedded public key does not match server-configured ATTESTATION_VERIFICATION_KEY" + ) + + 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. + + Attestations with valid Sigstore certificates are verified through the + standard Sigstore path. Attestations without certificates are verified + against a custom public key configured via ATTESTATION_VERIFICATION_KEY. + Currently, it supports RSA PKCS1v15 signatures and SLSA v0.2 provenance + publisher enrichment. + """ 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) + _verify_embedded_key(attestation, verification_key) + else: + raise VerificationError( + "Attestation has no Sigstore certificate or no custom " + "verification key configured via ATTESTATION_VERIFICATION_KEY " + "(embedded key in verification_material is optional)" + ) 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_attestations.py b/pulp_python/tests/functional/api/test_attestations.py index 147a68b3..4b916800 100644 --- a/pulp_python/tests/functional/api/test_attestations.py +++ b/pulp_python/tests/functional/api/test_attestations.py @@ -48,7 +48,10 @@ def test_crd_provenance(python_bindings, twine_package, python_content_factory, assert provenance.package == content.pulp_href r = requests.get(twine_package.provenance_url) assert r.status_code == 200 - assert r.json() == provenance.provenance + expected_att = r.json()["attestation_bundles"][0]["attestations"][0] + stored_att = provenance.provenance["attestation_bundles"][0]["attestations"][0] + assert stored_att["envelope"] == expected_att["envelope"] + assert stored_att["version"] == expected_att["version"] @pytest.mark.parallel @@ -130,7 +133,9 @@ def test_attestation_upload(python_bindings, twine_package, monitor_task): prov = python_bindings.ContentProvenanceApi.read(response.created_resources[1]) assert prov.package == response.created_resources[0] att_bundle = prov.provenance["attestation_bundles"][0] - assert att_bundle["attestations"] == attestations + stored_att = att_bundle["attestations"][0] + assert stored_att["envelope"] == attestations[0]["envelope"] + assert stored_att["version"] == attestations[0]["version"] assert att_bundle["publisher"]["kind"] == "Pulp User" @@ -150,7 +155,9 @@ def test_attestation_sync_upload(python_bindings, twine_package, download_python prov = provs.results[0] assert prov.package == content.pulp_href att_bundle = prov.provenance["attestation_bundles"][0] - assert att_bundle["attestations"] == attestations + stored_att = att_bundle["attestations"][0] + assert stored_att["envelope"] == attestations[0]["envelope"] + assert stored_att["version"] == attestations[0]["version"] assert att_bundle["publisher"]["kind"] == "Pulp User" diff --git a/pulp_python/tests/functional/api/test_slsa_attestations.py b/pulp_python/tests/functional/api/test_slsa_attestations.py new file mode 100644 index 00000000..d8256125 --- /dev/null +++ b/pulp_python/tests/functional/api/test_slsa_attestations.py @@ -0,0 +1,322 @@ +"""Functional tests for SLSA provenance attestation verification. + +These tests exercise the attestation / provenance upload paths with +attestations that carry an RSA signature instead of a Sigstore certificate. + +A static test keypair is shipped in .ci/assets/keys/ 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 an SLSA 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, verification_material=None): + """Return a single PEP-740 Attestation dict (SLSA flavour).""" + return { + "version": 1, + "verification_material": verification_material, + "envelope": { + "statement": _b64(statement_bytes), + "signature": _b64(signature_bytes), + }, + } + + +def _read_public_key_pem(): + """Read the test public key as a PEM string.""" + with open(TEST_PUBLIC_KEY_PATH) as f: + return f.read() + + +def _make_embedded_key_material(public_key_pem): + """Build a verification_material dict with an embedded public key.""" + return { + "certificate": None, + "public_key": public_key_pem, + "transparency_entries": [{"log_index": 0}], + } + + +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_slsa_provenance_stored( + python_bindings, python_content_factory, monitor_task, test_private_key, _provenance_file +): + """An SLSA 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_slsa_wrong_subject_name_rejected( + python_bindings, python_content_factory, monitor_task, test_private_key, _provenance_file +): + """Verification rejects an SLSA 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_slsa_wrong_digest_rejected( + python_bindings, python_content_factory, monitor_task, test_private_key, _provenance_file +): + """Verification rejects an SLSA 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_slsa_bad_signature_rejected( + python_bindings, python_content_factory, monitor_task, test_private_key, _provenance_file +): + """An SLSA 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_slsa_attestation_via_content_upload( + python_bindings, python_content_factory, monitor_task, test_private_key +): + """SLSA 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" + + +def test_slsa_embedded_key_accepted( + python_bindings, python_content_factory, monitor_task, test_private_key, _provenance_file +): + """An attestation with an embedded public key matching the server key is accepted.""" + content = python_content_factory() + + stmt = _build_statement(content.filename, content.sha256) + sig = _sign(stmt, test_private_key) + vm = _make_embedded_key_material(_read_public_key_pem()) + att = _make_attestation(stmt, sig, verification_material=vm) + 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_vm = prov_obj.provenance["attestation_bundles"][0]["attestations"][0][ + "verification_material" + ] + assert stored_vm["public_key"] == _read_public_key_pem() + + +def test_slsa_embedded_key_mismatch_rejected( + python_bindings, python_content_factory, monitor_task, test_private_key, _provenance_file +): + """An attestation with an embedded key that does not match the server key is rejected.""" + content = python_content_factory() + + stmt = _build_statement(content.filename, content.sha256) + sig = _sign(stmt, test_private_key) + wrong_key = ( + "-----BEGIN PUBLIC KEY-----\n" + "MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEArvxtuKOCCHGNd6agbC4b\n" + "zX9S7qMnMRdz9bxxzkPf9a4T75St3EcrNQYxLqzcDBfeiiLa1wXQoraqFSGkvqX+\n" + "YDFmtYRgnXEujUwYqHQBYrKtrQHxrQ2J3V1e+PqhJ4HCQq/uRZdx6KJ5DOlpYuJM\n" + "aL2m7IozNJJkgjQH1frmDfM5/GuPVbgwN0QDnC0oSzNJFKtdVMD3SPnQGMPQ9qxf\n" + "DiTZOk1tO3VD/xXoNJ6Zgf0D5RvGyLAfLflgojKImHezgpdE6iQyZAVU2ShKMq3M\n" + "K+XtX39/ajXHJ1FCLXlRFoNoHltWioyvkmXHbhb9OxGncB+x6idSjgp7AR5y15HN\n" + "DwIDAQAB\n" + "-----END PUBLIC KEY-----\n" + ) + vm = _make_embedded_key_material(wrong_key) + att = _make_attestation(stmt, sig, verification_material=vm) + 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 "embedded public key does not match" in exc_info.value.task.error["description"] diff --git a/template_config.yml b/template_config.yml index dbcfa481..9ee355c4 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.36" lint_ignore: [] @@ -43,6 +47,7 @@ pulp_settings: allowed_import_paths: "/tmp" api_root: "/pulp/" orphan_protection_time: 0 + attestation_verification_key: "/etc/pki/attestation/test-key.pem" pypi_api_hostname: "https://pulp:443" pulp_settings_azure: MEDIA_ROOT: ""