55
66from __future__ import annotations
77
8- import binascii
9- from base64 import b64decode , b64encode
8+ import base64
109from typing import TYPE_CHECKING , Annotated , Any , Literal , NewType
1110
12- import rfc8785
1311import sigstore .errors
1412from annotated_types import MinLen # noqa: TCH002
1513from cryptography import x509
1614from cryptography .hazmat .primitives import serialization
17- from pydantic import BaseModel
15+ from packaging .utils import parse_sdist_filename , parse_wheel_filename
16+ from pydantic import Base64Bytes , BaseModel
1817from pydantic_core import ValidationError
1918from sigstore ._utils import _sha256_streaming
19+ from sigstore .dsse import Envelope as DsseEnvelope
20+ from sigstore .dsse import _DigestSet , _Statement , _StatementBuilder , _Subject
2021from sigstore .models import Bundle , LogEntry
22+ from sigstore_protobuf_specs .io .intoto import Envelope as _Envelope
23+ from sigstore_protobuf_specs .io .intoto import Signature as _Signature
2124
2225if TYPE_CHECKING :
2326 from pathlib import Path # pragma: no cover
2730 from sigstore .verify .policy import VerificationPolicy # pragma: no cover
2831
2932
30- class ConversionError (ValueError ):
31- """The base error for all errors during conversion."""
32-
33+ class AttestationError (ValueError ):
34+ """Base error for all APIs."""
3335
34- class InvalidAttestationError (ConversionError ):
35- """The PyPI Attestation given as input is not valid."""
3636
37- def __init__ (self : InvalidAttestationError , msg : str ) -> None :
38- """Initialize an `InvalidAttestationError`."""
39- super ().__init__ (f"Could not convert input Attestation: { msg } )
37+ class ConversionError (AttestationError ):
38+ """The base error for all errors during conversion."""
4039
4140
42- class VerificationError (ValueError ):
41+ class VerificationError (AttestationError ):
4342 """The PyPI Attestation failed verification."""
4443
4544 def __init__ (self : VerificationError , msg : str ) -> None :
@@ -53,7 +52,7 @@ def __init__(self: VerificationError, msg: str) -> None:
5352class VerificationMaterial (BaseModel ):
5453 """Cryptographic materials used to verify attestation objects."""
5554
56- certificate : str
55+ certificate : Base64Bytes
5756 """
5857 The signing certificate, as `base64(DER(cert))`.
5958 """
@@ -78,62 +77,99 @@ class Attestation(BaseModel):
7877 Cryptographic materials used to verify `message_signature`.
7978 """
8079
81- message_signature : str
80+ envelope : Envelope
8281 """
83- The attestation's signature, as `base64(raw-sig)`, where `raw-sig`
84- is the raw bytes of the signing operation over the attestation payload.
82+ The enveloped attestation statement and signature.
8583 """
8684
85+ @classmethod
86+ def sign (cls , signer : Signer , dist : Path ) -> Attestation :
87+ """Create an envelope, with signature, from a distribution file."""
88+ with dist .open (mode = "rb" , buffering = 0 ) as io :
89+ # Replace this with `hashlib.file_digest()` once
90+ # our minimum supported Python is >=3.11
91+ digest = _sha256_streaming (io ).hex ()
92+
93+ stmt = (
94+ _StatementBuilder ()
95+ .subjects (
96+ [
97+ _Subject (
98+ name = _ultranormalize_dist_filename (dist .name ),
99+ digest = _DigestSet (root = {"sha256" : digest }),
100+ )
101+ ]
102+ )
103+ .predicate_type ("https://docs.pypi.org/attestations/publish/v1" )
104+ .build ()
105+ )
106+ bundle = signer .sign_dsse (stmt )
107+
108+ return sigstore_to_pypi (bundle )
109+
87110 def verify (self , verifier : Verifier , policy : VerificationPolicy , dist : Path ) -> None :
88111 """Verify against an existing Python artifact.
89112
90- On failure, raises:
91- - `InvalidAttestationError` if the attestation could not be converted to
92- a Sigstore Bundle.
93- - `VerificationError` if the attestation could not be verified.
113+ On failure, raises an appropriate subclass of `AttestationError`.
94114 """
95- payload_to_verify = AttestationPayload .from_dist (dist )
115+ with dist .open (mode = "rb" , buffering = 0 ) as io :
116+ # Replace this with `hashlib.file_digest()` once
117+ # our minimum supported Python is >=3.11
118+ expected_digest = _sha256_streaming (io ).hex ()
119+
96120 bundle = pypi_to_sigstore (self )
97121 try :
98- verifier .verify_artifact ( bytes ( payload_to_verify ), bundle , policy )
122+ type_ , payload = verifier .verify_dsse ( bundle , policy )
99123 except sigstore .errors .VerificationError as err :
100124 raise VerificationError (str (err )) from err
101125
126+ if type_ != DsseEnvelope ._TYPE : # noqa: SLF001
127+ raise VerificationError (f"expected JSON envelope, got { type_ } )
102128
103- class AttestationPayload (BaseModel ):
104- """Attestation Payload object as defined in PEP 740."""
129+ try :
130+ statement = _Statement .model_validate_json (payload )
131+ except ValidationError as e :
132+ raise VerificationError (f"invalid statement: { str (e )} )
105133
106- distribution : str
107- """
108- The file name of the Python package distribution.
109- """
134+ if len (statement .subjects ) != 1 :
135+ raise VerificationError ("too many subjects in statement (must be exactly one)" )
136+ subject = statement .subjects [0 ]
110137
111- digest : str
112- """
113- The SHA-256 digest of the distribution's contents, as a hexadecimal string.
114- """
138+ if not subject .name :
139+ raise VerificationError ("invalid subject: missing name" )
115140
116- @classmethod
117- def from_dist (cls , dist : Path ) -> AttestationPayload :
118- """Create an `AttestationPayload` from a distribution file."""
119- with dist .open (mode = "rb" , buffering = 0 ) as io :
120- # Replace this with `hashlib.file_digest()` once
121- # our minimum supported Python is >=3.11
122- digest = _sha256_streaming (io ).hex ()
141+ try :
142+ # We always ultranormalize when signing, but other signers may not.
143+ subject_name = _ultranormalize_dist_filename (subject .name )
144+ except ValueError as e :
145+ raise VerificationError (f"invalid subject: { str (e )} )
146+
147+ normalized = _ultranormalize_dist_filename (dist .name )
148+ if subject_name != normalized :
149+ raise VerificationError (
150+ f"subject does not match distribution name: { subject_name } { normalized }
151+ )
152+
153+ digest = subject .digest .root .get ("sha256" )
154+ if digest is None or digest != expected_digest :
155+ raise VerificationError ("subject does not match distribution digest" )
123156
124- return AttestationPayload (
125- distribution = dist .name ,
126- digest = digest ,
127- )
128157
129- def sign (self , signer : Signer ) -> Attestation :
130- """Create a PEP 740 attestation by signing this payload."""
131- sigstore_bundle = signer .sign_artifact (bytes (self ))
132- return sigstore_to_pypi (sigstore_bundle )
158+ class Envelope (BaseModel ):
159+ """The attestation envelope, containing the attested-for payload and its signature."""
133160
134- def __bytes__ (self : AttestationPayload ) -> bytes :
135- """Convert to bytes using a canonicalized JSON representation (from RFC8785)."""
136- return rfc8785 .dumps (self .model_dump ())
161+ statement : Base64Bytes
162+ """
163+ The attestation statement.
164+
165+ This is represented as opaque bytes on the wire (encoded as base64),
166+ but it MUST be an JSON in-toto v1 Statement.
167+ """
168+
169+ signature : Base64Bytes
170+ """
171+ A signature for the above statement, encoded as base64.
172+ """
137173
138174
139175def sigstore_to_pypi (sigstore_bundle : Bundle ) -> Attestation :
@@ -142,38 +178,103 @@ def sigstore_to_pypi(sigstore_bundle: Bundle) -> Attestation:
142178 encoding = serialization .Encoding .DER
143179 )
144180
145- signature = sigstore_bundle ._inner .message_signature .signature # noqa: SLF001
181+ envelope = sigstore_bundle ._inner .dsse_envelope # noqa: SLF001
182+
183+ if len (envelope .signatures ) != 1 :
184+ raise ConversionError (f"expected exactly one signature, got { len (envelope .signatures )} )
185+
146186 return Attestation (
147187 version = 1 ,
148188 verification_material = VerificationMaterial (
149- certificate = b64encode (certificate ). decode ( "ascii" ),
189+ certificate = base64 . b64encode (certificate ),
150190 transparency_entries = [TransparencyLogEntry (sigstore_bundle .log_entry ._to_dict_rekor ())], # noqa: SLF001
151191 ),
152- message_signature = b64encode (signature ).decode ("ascii" ),
192+ envelope = Envelope (
193+ statement = base64 .b64encode (envelope .payload ),
194+ signature = base64 .b64encode (envelope .signatures [0 ].sig ),
195+ ),
153196 )
154197
155198
156199def pypi_to_sigstore (pypi_attestation : Attestation ) -> Bundle :
157200 """Convert a PyPI attestation object as defined in PEP 740 into a Sigstore Bundle."""
158- try :
159- certificate_bytes = b64decode (pypi_attestation .verification_material .certificate )
160- signature_bytes = b64decode (pypi_attestation .message_signature )
161- except binascii .Error as err :
162- raise InvalidAttestationError (str (err )) from err
201+ cert_bytes = pypi_attestation .verification_material .certificate
202+ statement = pypi_attestation .envelope .statement
203+ signature = pypi_attestation .envelope .signature
204+
205+ evp = DsseEnvelope (
206+ _Envelope (
207+ payload = statement ,
208+ payload_type = DsseEnvelope ._TYPE , # noqa: SLF001
209+ signatures = [_Signature (sig = signature )],
210+ )
211+ )
163212
164213 tlog_entry = pypi_attestation .verification_material .transparency_entries [0 ]
165214 try :
166- certificate = x509 .load_der_x509_certificate (certificate_bytes )
215+ certificate = x509 .load_der_x509_certificate (cert_bytes )
167216 except ValueError as err :
168- raise InvalidAttestationError ( str ( err ) ) from err
217+ raise ConversionError ( "invalid X.509 certificate" ) from err
169218
170219 try :
171220 log_entry = LogEntry ._from_dict_rekor (tlog_entry ) # noqa: SLF001
172221 except (ValidationError , sigstore .errors .Error ) as err :
173- raise InvalidAttestationError ( str ( err ) ) from err
222+ raise ConversionError ( "invalid transparency log entry" ) from err
174223
175- return Bundle .from_parts (
224+ return Bundle ._from_parts ( # noqa: SLF001
176225 cert = certificate ,
177- sig = signature_bytes ,
226+ content = evp ,
178227 log_entry = log_entry ,
179228 )
229+
230+
231+ def _ultranormalize_dist_filename (dist : str ) -> str :
232+ """Return an "ultranormalized" form of the given distribution filename.
233+
234+ This form is equivalent to the normalized form for sdist and wheel
235+ filenames, with the additional stipulation that compressed tag sets,
236+ if present, are also sorted alphanumerically.
237+
238+ Raises `ValueError` on any invalid distribution filename.
239+ """
240+ # NOTE: .whl and .tar.gz are assumed lowercase, since `packaging`
241+ # already rejects non-lowercase variants.
242+ if dist .endswith (".whl" ):
243+ # `parse_wheel_filename` raises a supertype of ValueError on failure.
244+ name , ver , build , tags = parse_wheel_filename (dist )
245+
246+ # The name has been normalized to replace runs of `[.-_]+` with `-`,
247+ # which then needs to be replaced with `_` for the wheel.
248+ name = name .replace ("-" , "_" )
249+
250+ # `parse_wheel_filename` normalizes the name and version for us,
251+ # so all we need to do is re-compress the tag set in a canonical
252+ # order.
253+ # NOTE(ww): This is written in a not very efficient manner, since
254+ # I wasn't feeling smart.
255+ impls , abis , platforms = set (), set (), set ()
256+ for tag in tags :
257+ impls .add (tag .interpreter )
258+ abis .add (tag .abi )
259+ platforms .add (tag .platform )
260+
261+ impl_tag = "." .join (sorted (impls ))
262+ abi_tag = "." .join (sorted (abis ))
263+ platform_tag = "." .join (sorted (platforms ))
264+
265+ if build :
266+ parts = "-" .join (
267+ [name , str (ver ), f"{ build [0 ]} { build [1 ]} , impl_tag , abi_tag , platform_tag ]
268+ )
269+ else :
270+ parts = "-" .join ([name , str (ver ), impl_tag , abi_tag , platform_tag ])
271+
272+ return f"{ parts }
273+
274+ elif dist .endswith (".tar.gz" ):
275+ # `parse_sdist_filename` raises a supertype of ValueError on failure.
276+ name , ver = parse_sdist_filename (dist )
277+ name = name .replace ("-" , "_" )
278+ return f"{ name } { ver }
279+ else :
280+ raise ValueError (f"unknown distribution format: { dist } )
0 commit comments