pypi_attestations

150class Attestation(BaseModel):
151    """Attestation object as defined in PEP 740."""
152
153    version: Literal[1]
154    """
155    The attestation format's version, which is always 1.
156    """
157
158    verification_material: VerificationMaterial
159    """
160    Cryptographic materials used to verify `message_signature`.
161    """
162
163    envelope: Envelope
164    """
165    The enveloped attestation statement and signature.
166    """
167
168    @property
169    def statement(self) -> dict[str, Any]:
170        """Return the statement within this attestation's envelope.
171
172        The value returned here is a dictionary, in the shape of an
173        in-toto statement.
174        """
175        return json.loads(self.envelope.statement)  # type: ignore[no-any-return]
176
177    @classmethod
178    def sign(cls, signer: Signer, dist: Distribution) -> Attestation:
179        """Create an envelope, with signature, from the given Python distribution.
180
181        On failure, raises `AttestationError`.
182        """
183        try:
184            stmt = (
185                StatementBuilder()
186                .subjects(
187                    [
188                        Subject(
189                            name=dist.name,
190                            digest=DigestSet(root={"sha256": dist.digest}),
191                        )
192                    ]
193                )
194                .predicate_type(AttestationType.PYPI_PUBLISH_V1)
195                .build()
196            )
197        except DsseError as e:
198            raise AttestationError(str(e))
199
200        try:
201            bundle = signer.sign_dsse(stmt)
202        except (ExpiredCertificate, ExpiredIdentity) as e:
203            raise AttestationError(str(e))
204
205        try:
206            return Attestation.from_bundle(bundle)
207        except ConversionError as e:
208            raise AttestationError(str(e))
209
210    @property
211    def certificate_claims(self) -> dict[str, str]:
212        """Return the claims present in the certificate.
213
214        We only return claims present in `_FULCIO_CLAIMS_OIDS`.
215        Values are decoded and returned as strings.
216        """
217        certificate = x509.load_der_x509_certificate(self.verification_material.certificate)
218        claims = {}
219        for extension in certificate.extensions:
220            if extension.oid in _FULCIO_CLAIMS_OIDS:
221                # 1.3.6.1.4.1.57264.1.8 through 1.3.6.1.4.1.57264.1.22 are formatted as DER-encoded
222                # strings; the ASN.1 tag is UTF8String (0x0C) and the tag class is universal.
223                value = extension.value.value
224                claims[extension.oid.dotted_string] = _der_decode_utf8string(value)
225
226        return claims
227
228    def verify(
229        self,
230        identity: VerificationPolicy | Publisher,
231        dist: Distribution,
232        *,
233        staging: bool = False,
234        offline: bool = False,
235    ) -> tuple[str, Optional[dict[str, Any]]]:
236        """Verify against an existing Python distribution.
237
238        The `identity` can be an object confirming to
239        `sigstore.policy.VerificationPolicy` or a `Publisher`, which will be
240        transformed into an appropriate verification policy.
241
242        By default, Sigstore's production verifier will be used. The
243        `staging` parameter can be toggled to enable the staging verifier
244        instead.
245
246        If `offline` is `True`, the verifier will not attempt to refresh the
247        TUF repository.
248
249        On failure, raises an appropriate subclass of `AttestationError`.
250        """
251        # NOTE: Can't do `isinstance` with `Publisher` since it's
252        # a `_GenericAlias`; instead we punch through to the inner
253        # `_Publisher` union.
254        # Use of typing.get_args is needed for Python < 3.10
255        if isinstance(identity, get_args(_Publisher)):
256            policy = identity._as_policy()  # noqa: SLF001
257        else:
258            policy = identity
259
260        if staging:
261            verifier = Verifier.staging(offline=offline)
262        else:
263            verifier = Verifier.production(offline=offline)
264
265        bundle = self.to_bundle()
266        try:
267            type_, payload = verifier.verify_dsse(bundle, policy)
268        except sigstore.errors.VerificationError as err:
269            raise VerificationError(str(err)) from err
270
271        if type_ != DsseEnvelope._TYPE:  # noqa: SLF001
272            raise VerificationError(f"expected JSON envelope, got {type_}")
273
274        try:
275            statement = _Statement.model_validate_json(payload)
276        except ValidationError as e:
277            raise VerificationError(f"invalid statement: {str(e)}")
278
279        if len(statement.subjects) != 1:
280            raise VerificationError("too many subjects in statement (must be exactly one)")
281        subject = statement.subjects[0]
282
283        if not subject.name:
284            raise VerificationError("invalid subject: missing name")
285
286        try:
287            # We always ultranormalize when signing, but other signers may not.
288            subject_name = _ultranormalize_dist_filename(subject.name)
289        except ValueError as e:
290            raise VerificationError(f"invalid subject: {str(e)}")
291
292        if subject_name != dist.name:
293            raise VerificationError(
294                f"subject does not match distribution name: {subject_name} != {dist.name}"
295            )
296
297        digest = subject.digest.root.get("sha256")
298        if digest is None or digest != dist.digest:
299            raise VerificationError("subject does not match distribution digest")
300
301        try:
302            AttestationType(statement.predicate_type)
303        except ValueError:
304            raise VerificationError(f"unknown attestation type: {statement.predicate_type}")
305
306        return statement.predicate_type, statement.predicate
307
308    def to_bundle(self) -> Bundle:
309        """Convert a PyPI attestation object as defined in PEP 740 into a Sigstore Bundle."""
310        cert_bytes = self.verification_material.certificate
311        statement = self.envelope.statement
312        signature = self.envelope.signature
313
314        evp = DsseEnvelope(
315            _Envelope(
316                payload=statement,
317                payload_type=DsseEnvelope._TYPE,  # noqa: SLF001
318                signatures=[_Signature(sig=signature)],
319            )
320        )
321
322        tlog_entry = self.verification_material.transparency_entries[0]
323        try:
324            certificate = x509.load_der_x509_certificate(cert_bytes)
325        except ValueError as err:
326            raise ConversionError("invalid X.509 certificate") from err
327
328        try:
329            log_entry = LogEntry._from_dict_rekor(tlog_entry)  # noqa: SLF001
330        except (ValidationError, sigstore.errors.Error) as err:
331            raise ConversionError("invalid transparency log entry") from err
332
333        return Bundle._from_parts(  # noqa: SLF001
334            cert=certificate,
335            content=evp,
336            log_entry=log_entry,
337        )
338
339    @classmethod
340    def from_bundle(cls, sigstore_bundle: Bundle) -> Attestation:
341        """Convert a Sigstore Bundle into a PyPI attestation as defined in PEP 740."""
342        certificate = sigstore_bundle.signing_certificate.public_bytes(
343            encoding=serialization.Encoding.DER
344        )
345
346        envelope = sigstore_bundle._inner.dsse_envelope  # noqa: SLF001
347
348        if len(envelope.signatures) != 1:
349            raise ConversionError(f"expected exactly one signature, got {len(envelope.signatures)}")
350
351        return cls(
352            version=1,
353            verification_material=VerificationMaterial(
354                certificate=base64.b64encode(certificate),
355                transparency_entries=[
356                    sigstore_bundle.log_entry._to_rekor().to_dict()  # noqa: SLF001
357                ],
358            ),
359            envelope=Envelope(
360                statement=base64.b64encode(envelope.payload),
361                signature=base64.b64encode(envelope.signatures[0].sig),
362            ),
363        )

643class AttestationBundle(BaseModel):
644    """AttestationBundle object as defined in PEP 740."""
645
646    publisher: Publisher
647    """
648    The publisher associated with this set of attestations.
649    """
650
651    attestations: list[Attestation]
652    """
653    The list of attestations included in this bundle.
654    """

116class AttestationError(ValueError):
117    """Base error for all APIs."""

109class AttestationType(str, Enum):
110    """Attestation types known to PyPI."""
111
112    SLSA_PROVENANCE_V1 = "https://slsa.dev/provenance/v1"
113    PYPI_PUBLISH_V1 = "https://docs.pypi.org/attestations/publish/v1"

120class ConversionError(AttestationError):
121    """The base error for all errors during conversion."""

 81class Distribution(BaseModel):
 82    """Represents a Python package distribution.
 83
 84    A distribution is identified by its (sdist or wheel) filename, which
 85    provides the package name and version (at a minimum) plus a SHA-256
 86    digest, which uniquely identifies its contents.
 87    """
 88
 89    name: str
 90    digest: str
 91
 92    @field_validator("name")
 93    @classmethod
 94    def _validate_name(cls, v: str) -> str:
 95        return _ultranormalize_dist_filename(v)
 96
 97    @classmethod
 98    def from_file(cls, dist: Path) -> Distribution:
 99        """Construct a `Distribution` from the given path."""
100        name = dist.name
101        with dist.open(mode="rb", buffering=0) as io:
102            # Replace this with `hashlib.file_digest()` once
103            # our minimum supported Python is >=3.11
104            digest = _sha256_streaming(io).hex()
105
106        return cls(name=name, digest=digest)

366class Envelope(BaseModel):
367    """The attestation envelope, containing the attested-for payload and its signature."""
368
369    statement: Base64Bytes
370    """
371    The attestation statement.
372
373    This is represented as opaque bytes on the wire (encoded as base64),
374    but it MUST be an JSON in-toto v1 Statement.
375    """
376
377    signature: Base64Bytes
378    """
379    A signature for the above statement, encoded as base64.
380    """

520class GitHubPublisher(_PublisherBase):
521    """A GitHub-based Trusted Publisher."""
522
523    kind: Literal["GitHub"] = "GitHub"
524
525    repository: str
526    """
527    The fully qualified publishing repository slug, e.g. `foo/bar` for
528    repository `bar` owned by `foo`.
529    """
530
531    workflow: str
532    """
533    The filename of the GitHub Actions workflow that performed the publishing
534    action.
535    """
536
537    environment: Optional[str] = None
538    """
539    The optional name GitHub Actions environment that the publishing
540    action was performed from.
541    """
542
543    def _as_policy(self) -> VerificationPolicy:
544        return _GitHubTrustedPublisherPolicy(self.repository, self.workflow)

612class GitLabPublisher(_PublisherBase):
613    """A GitLab-based Trusted Publisher."""
614
615    kind: Literal["GitLab"] = "GitLab"
616
617    repository: str
618    """
619    The fully qualified publishing repository slug, e.g. `foo/bar` for
620    repository `bar` owned by `foo` or `foo/baz/bar` for repository
621    `bar` owned by group `foo` and subgroup `baz`.
622    """
623
624    workflow_filepath: str
625    """
626    The path for the CI/CD configuration file. This is usually ".gitlab-ci.yml",
627    but can be customized.
628    """
629
630    environment: Optional[str] = None
631    """
632    The optional environment that the publishing action was performed from.
633    """
634
635    def _as_policy(self) -> VerificationPolicy:
636        return _GitLabTrustedPublisherPolicy(self.repository, self.workflow_filepath)

657class Provenance(BaseModel):
658    """Provenance object as defined in PEP 740."""
659
660    version: Literal[1] = 1
661    """
662    The provenance object's version, which is always 1.
663    """
664
665    attestation_bundles: list[AttestationBundle]
666    """
667    One or more attestation "bundles".
668    """

124class VerificationError(AttestationError):
125    """The PyPI Attestation failed verification."""
126
127    def __init__(self: VerificationError, msg: str) -> None:
128        """Initialize an `VerificationError`."""
129        super().__init__(f"Verification failed: {msg}")

135class VerificationMaterial(BaseModel):
136    """Cryptographic materials used to verify attestation objects."""
137
138    certificate: Base64Bytes
139    """
140    The signing certificate, as `base64(DER(cert))`.
141    """
142
143    transparency_entries: Annotated[list[TransparencyLogEntry], MinLen(1)]
144    """
145    One or more transparency log entries for this attestation's signature
146    and certificate.
147    """