pypi_attestations

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

663class AttestationBundle(BaseModel):
664    """AttestationBundle object as defined in PEP 740."""
665
666    publisher: Publisher
667    """
668    The publisher associated with this set of attestations.
669    """
670
671    attestations: list[Attestation]
672    """
673    The list of attestations included in this bundle.
674    """

121class AttestationError(ValueError):
122    """Base error for all APIs."""

114class AttestationType(str, Enum):
115    """Attestation types known to PyPI."""
116
117    SLSA_PROVENANCE_V1 = "https://slsa.dev/provenance/v1"
118    PYPI_PUBLISH_V1 = "https://docs.pypi.org/attestations/publish/v1"

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

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

381class Envelope(BaseModel):
382    """The attestation envelope, containing the attested-for payload and its signature."""
383
384    statement: Base64Bytes
385    """
386    The attestation statement.
387
388    This is represented as opaque bytes on the wire (encoded as base64),
389    but it MUST be an JSON in-toto v1 Statement.
390    """
391
392    signature: Base64Bytes
393    """
394    A signature for the above statement, encoded as base64.
395    """

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

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

644class GooglePublisher(_PublisherBase):
645    """A Google Cloud-based Trusted Publisher."""
646
647    kind: Literal["Google"] = "Google"
648
649    email: str
650    """
651    The email address of the Google Cloud service account that performed
652    the publishing action.
653    """
654
655    def _as_policy(self) -> VerificationPolicy:
656        return policy.Identity(identity=self.email, issuer="https://accounts.google.com")

677class Provenance(BaseModel):
678    """Provenance object as defined in PEP 740."""
679
680    version: Literal[1] = 1
681    """
682    The provenance object's version, which is always 1.
683    """
684
685    attestation_bundles: list[AttestationBundle]
686    """
687    One or more attestation "bundles".
688    """

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

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