[PULP-2548] Add key-signed attestation verification support

.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-----

.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-----

.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"

.gitleaks.toml

@@ -1,6 +1,7 @@
 [allowlist]
   description = "Allow specific test-keys."
   paths = [
+    '''.ci/assets/keys/test-key-private.pem''',
   ]
   regexes = [
     '''AKIAIT2Z5TDYPX3ARJBA''',

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`.

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

docs/user/guides/attestation.md

@@ -89,3 +89,49 @@ 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.
+
+```python
+# settings.py
+ATTESTATION_VERIFICATION_KEYS = {"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:
+
+- **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`.
+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
+
+- Each domain can have only one key.
+- The key-based path only supports RSA PKCS1v15 with SHA-256 signatures.

pulp_python/app/provenance.py

@@ -1,12 +1,28 @@
+import json
 from typing import Annotated, Literal, Union, get_args
 
+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 (
-    Attestation,
     Distribution,
+    Envelope,  # noqa - needed in module namespace for Pydantic model rebuild
     Publisher,
+    VerificationError,
+    VerificationMaterial,
 )
+from sigstore.dsse import Envelope as DSSEEnvelope
+from sigstore.dsse import _pae
+
+from pulpcore.plugin.util import get_domain
+
+_verification_key_cache = {}
 
 
 class _PermissivePolicy:
@@ -32,13 +48,30 @@ 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]
 # Create a new type that fallbacks to AnyPublisher
 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 +91,119 @@ class Provenance(BaseModel):
     attestation_bundles: list[AttestationBundle]
 
 
+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:
+        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_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."""
+    try:
+        stmt = json.loads(attestation.envelope.statement)
+    except (json.JSONDecodeError, UnicodeDecodeError) as e:
+        raise VerificationError(f"invalid statement: {e}")
+
+    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}"
+            )
+
+        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):
+    """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:
+        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)
     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_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:
+                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:
+                    _verify_statement_subject(attestation, dist)
+                    _verify_signature(attestation, verification_key)
+                else:
+                    raise VerificationError(
+                        "attestation has no Sigstore certificate, and "
+                        f"ATTESTATION_VERIFICATION_KEYS has no key for domain '{domain_name}'"
+                    )

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,15 +117,25 @@ def validate(self, data):
                 }
             )
 
+        sha256 = data.get("sha256_digest")
         if attestations := data.get("attestations"):
             try:
                 attestations = TypeAdapter(list[Attestation]).validate_python(attestations)
             except ValidationError as e:
                 raise serializers.ValidationError(
                     {"attestations": _("The uploaded attestations are not valid: {}").format(e)}
                 )
+            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: