1+ import json
12from typing import Annotated, Literal, Union, get_args
23
3- from pydantic import BaseModel, ConfigDict, Field
4+ from cryptography.exceptions import InvalidSignature, UnsupportedAlgorithm
5+ from cryptography.hazmat.primitives import hashes, serialization
6+ from cryptography.hazmat.primitives.asymmetric import padding as crypto_padding
7+ from cryptography.hazmat.primitives.asymmetric import rsa
8+ from cryptography.x509 import load_der_x509_certificate
9+ from django.conf import settings
10+ from pydantic import BaseModel, ConfigDict, Field, ValidationError
411from pydantic.alias_generators import to_snake
12+ from pypi_attestations import Attestation as _UpstreamAttestation
513from pypi_attestations import (
6- Attestation,
714 Distribution,
15+ Envelope, # noqa - needed in module namespace for Pydantic model rebuild
816 Publisher,
17+ VerificationError,
18+ VerificationMaterial,
919)
20+ from sigstore.dsse import Envelope as DSSEEnvelope
21+ from sigstore.dsse import _pae
22+
23+ from pulpcore.plugin.util import get_domain
24+
25+ _verification_key_cache = {}
1026
1127
1228class _PermissivePolicy:
@@ -32,13 +48,30 @@ def _as_policy(self):
3248# Get the underlying Union type of the original Publisher
3349# Publisher is Annotated[Union[...], Field(discriminator="kind")]
3450_OriginalPublisherTypes = get_args(Publisher.__origin__)
51+ # Get the kinds of the original publisher types, they can only be proven by a Sigstore certificate
52+ _TrustedPublisherKinds = {t.model_fields["kind"].default for t in _OriginalPublisherTypes}
3553# Add AnyPublisher to the list of original publisher types
3654_ExtendedPublisherTypes = (*_OriginalPublisherTypes, AnyPublisher)
3755_ExtendedPublisherUnion = Union[_ExtendedPublisherTypes]
3856# Create a new type that fallbacks to AnyPublisher
3957ExtendedPublisher = Annotated[_ExtendedPublisherUnion, Field(union_mode="left_to_right")]
4058
4159
60+ class Attestation(_UpstreamAttestation):
61+ """
62+ Attestation object as defined in PEP 740.
63+
64+ Inherits from the upstream pypi_attestations.Attestation to keep Sigstore
65+ verification methods (to_bundle, verify), but makes verification_material
66+ optional to support attestations signed with a custom key instead of Sigstore.
67+ """
68+
69+ verification_material: VerificationMaterial | None = None
70+ """
71+ Cryptographic materials used to verify `message_signature`.
72+ """
73+
74+
4275class AttestationBundle(BaseModel):
4376 """
4477 AttestationBundle object as defined in PEP740.
@@ -58,14 +91,122 @@ class Provenance(BaseModel):
5891 attestation_bundles: list[AttestationBundle]
5992
6093
94+ def _load_verification_key(domain_name):
95+ """Load the attestation verification public key configured for the domain, with caching."""
96+ keys = settings.ATTESTATION_VERIFICATION_KEYS
97+ if not isinstance(keys, dict):
98+ raise VerificationError(
99+ "ATTESTATION_VERIFICATION_KEYS must map domain names to public key paths"
100+ )
101+ key_path = keys.get(domain_name)
102+ if not key_path:
103+ return None
104+ if key_path not in _verification_key_cache:
105+ try:
106+ with open(key_path, "rb") as f:
107+ key = serialization.load_pem_public_key(f.read())
108+ except (OSError, ValueError, UnsupportedAlgorithm) as e:
109+ raise VerificationError(
110+ f"could not load the verification key for domain '{domain_name}' "
111+ f"from {key_path}: {e}"
112+ )
113+ if not isinstance(key, rsa.RSAPublicKey):
114+ raise VerificationError(
115+ f"the verification key for domain '{domain_name}' in {key_path} "
116+ "must be an RSA public key"
117+ )
118+ _verification_key_cache[key_path] = key
119+ return _verification_key_cache[key_path]
120+
121+
122+ def _has_certificate(attestation):
123+ """Check whether the attestation carries a certificate, rejecting unparseable ones."""
124+ vm = attestation.verification_material
125+ if vm is None:
126+ return False
127+ try:
128+ load_der_x509_certificate(vm.certificate)
129+ except ValueError as e:
130+ raise VerificationError(f"invalid X.509 certificate in verification_material: {e}")
131+ return True
132+
133+
134+ def _verify_statement_subject(attestation, dist):
135+ """Validate that the in-toto statement subject matches the distribution."""
136+ try:
137+ stmt = json.loads(attestation.envelope.statement)
138+ except (json.JSONDecodeError, UnicodeDecodeError) as e:
139+ raise VerificationError(f"invalid statement: {e}")
140+
141+ try:
142+ subjects = stmt.get("subject", [])
143+ if len(subjects) != 1:
144+ raise VerificationError("expected exactly one subject in statement")
145+
146+ subject = subjects[0]
147+ name = subject.get("name", "")
148+ if name != dist.name:
149+ raise VerificationError(
150+ f"subject does not match distribution name: {name} != {dist.name}"
151+ )
152+
153+ digest = subject.get("digest", {}).get("sha256")
154+ if digest != dist.digest:
155+ raise VerificationError("subject does not match distribution digest")
156+ except (AttributeError, TypeError) as e:
157+ raise VerificationError(f"malformed statement: {e}")
158+
159+
160+ def _verify_signature(attestation, public_key):
161+ """Verify the attestation's RSA signature over the DSSE PAE bytes."""
162+ statement_bytes = attestation.envelope.statement
163+ signature_bytes = attestation.envelope.signature
164+ pae = _pae(DSSEEnvelope._TYPE, statement_bytes)
165+ try:
166+ public_key.verify(
167+ signature_bytes,
168+ pae,
169+ crypto_padding.PKCS1v15(),
170+ hashes.SHA256(),
171+ )
172+ except InvalidSignature:
173+ raise VerificationError("signature verification failed")
174+
175+
61176def verify_provenance(filename, sha256, provenance, offline=True):
62- """Verify the provenance object is valid for the package."""
63- dist = Distribution(name=filename, digest=sha256)
177+ """Verify the provenance object is valid for the package.
178+
179+ Each attestation is routed to one of two coexisting verification paths:
180+ 1. Attestations with a valid Sigstore certificate follow the standard PEP 740 path.
181+ 2. Attestations without a certificate are verified against the key configured for the
182+ current domain in ATTESTATION_VERIFICATION_KEYS using RSA PKCS1v15 signatures.
183+ Attestations with a certificate that can't be parsed are rejected.
184+ """
185+ try:
186+ dist = Distribution(name=filename, digest=sha256)
187+ except ValidationError as e:
188+ raise VerificationError(f"invalid distribution filename {filename}: {e}")
64189 for bundle in provenance.attestation_bundles:
65190 publisher = bundle.publisher
66191 policy = publisher._as_policy()
67192 for attestation in bundle.attestations:
68- sig_bundle = attestation.to_bundle()
69- checkpoint = sig_bundle.log_entry._inner.inclusion_proof.checkpoint
70- staging = "sigstage.dev" in checkpoint.envelope
71- attestation.verify(policy, dist, staging=staging, offline=offline)
193+ if _has_certificate(attestation):
194+ sig_bundle = attestation.to_bundle()
195+ checkpoint = sig_bundle.log_entry._inner.inclusion_proof.checkpoint
196+ staging = "sigstage.dev" in checkpoint.envelope
197+ attestation.verify(policy, dist, staging=staging, offline=offline)
198+ else:
199+ if publisher.kind in _TrustedPublisherKinds:
200+ raise VerificationError(
201+ f"publisher kind '{publisher.kind}' requires a Sigstore certificate"
202+ )
203+ domain_name = get_domain().name
204+ verification_key = _load_verification_key(domain_name)
205+ if verification_key:
206+ _verify_statement_subject(attestation, dist)
207+ _verify_signature(attestation, verification_key)
208+ else:
209+ raise VerificationError(
210+ "attestation has no Sigstore certificate, and "
211+ f"ATTESTATION_VERIFICATION_KEYS has no key for domain '{domain_name}'"
212+ )
0 commit comments