pypi_attestations

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

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

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

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

121class ConversionError(AttestationError):
122    """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        _check_dist_filename(v)
 96        return v
 97
 98    @classmethod
 99    def from_file(cls, dist: Path) -> Distribution:
100        """Construct a `Distribution` from the given path."""
101        name = dist.name
102        with dist.open(mode="rb", buffering=0) as io:
103            # Replace this with `hashlib.file_digest()` once
104            # our minimum supported Python is >=3.11
105            digest = _sha256_streaming(io).hex()
106
107        return cls(name=name, digest=digest)

374class Envelope(BaseModel):
375    """The attestation envelope, containing the attested-for payload and its signature."""
376
377    statement: Base64Bytes
378    """
379    The attestation statement.
380
381    This is represented as opaque bytes on the wire (encoded as base64),
382    but it MUST be an JSON in-toto v1 Statement.
383    """
384
385    signature: Base64Bytes
386    """
387    A signature for the above statement, encoded as base64.
388    """

508class GitHubPublisher(_PublisherBase):
509    """A GitHub-based Trusted Publisher."""
510
511    kind: Literal["GitHub"] = "GitHub"
512
513    repository: str
514    """
515    The fully qualified publishing repository slug, e.g. `foo/bar` for
516    repository `bar` owned by `foo`.
517    """
518
519    workflow: str
520    """
521    The filename of the GitHub Actions workflow that performed the publishing
522    action.
523    """
524
525    environment: Optional[str] = None
526    """
527    The optional name GitHub Actions environment that the publishing
528    action was performed from.
529    """
530
531    def _as_policy(self) -> VerificationPolicy:
532        return _GitHubTrustedPublisherPolicy(self.repository, self.workflow)

600class GitLabPublisher(_PublisherBase):
601    """A GitLab-based Trusted Publisher."""
602
603    kind: Literal["GitLab"] = "GitLab"
604
605    repository: str
606    """
607    The fully qualified publishing repository slug, e.g. `foo/bar` for
608    repository `bar` owned by `foo` or `foo/baz/bar` for repository
609    `bar` owned by group `foo` and subgroup `baz`.
610    """
611
612    workflow_filepath: str
613    """
614    The path for the CI/CD configuration file. This is usually ".gitlab-ci.yml",
615    but can be customized.
616    """
617
618    environment: Optional[str] = None
619    """
620    The optional environment that the publishing action was performed from.
621    """
622
623    def _as_policy(self) -> VerificationPolicy:
624        return _GitLabTrustedPublisherPolicy(self.repository, self.workflow_filepath)

627class GooglePublisher(_PublisherBase):
628    """A Google Cloud-based Trusted Publisher."""
629
630    kind: Literal["Google"] = "Google"
631
632    email: str
633    """
634    The email address of the Google Cloud service account that performed
635    the publishing action.
636    """
637
638    def _as_policy(self) -> VerificationPolicy:
639        return policy.Identity(identity=self.email, issuer="https://accounts.google.com")

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

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

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