pypi_attestations

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

766class AttestationBundle(BaseModel):
767    """AttestationBundle object as defined in PEP 740."""
768
769    publisher: Publisher
770    """
771    The publisher associated with this set of attestations.
772    """
773
774    attestations: list[Attestation]
775    """
776    The list of attestations included in this bundle.
777    """

123class AttestationError(ValueError):
124    """Base error for all APIs."""

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

721class CircleCIPublisher(_PublisherBase):
722    """A CircleCI-based Trusted Publisher."""
723
724    kind: Literal["CircleCI"] = "CircleCI"
725
726    project_id: str
727    """
728    The CircleCI project ID (UUID) that performed the publishing action.
729    """
730
731    pipeline_definition_id: str
732    """
733    The CircleCI pipeline definition ID (UUID) that defines the pipeline configuration.
734    This uniquely identifies the specific pipeline definition allowed to publish.
735    """
736
737    vcs_origin: str | None = None
738    """
739    The optional source repository URI that triggered the pipeline.
740    This comes from the oidc.circleci.com/vcs-origin claim and looks like
741    "github.com/org/repo" (without the https:// prefix).
742    Not present for pipelines triggered by custom webhooks.
743    """
744
745    vcs_ref: str | None = None
746    """
747    The optional git ref that triggered the pipeline.
748    This comes from the oidc.circleci.com/vcs-ref claim and looks like
749    "refs/heads/main" or "refs/tags/v1.0.0".
750    Not present for pipelines triggered by custom webhooks.
751    """
752
753    def _as_policy(self) -> VerificationPolicy:
754        return _CircleCITrustedPublisherPolicy(
755            self.project_id,
756            self.pipeline_definition_id,
757            self.vcs_origin,
758            self.vcs_ref,
759        )

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

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

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

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

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

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

780class Provenance(BaseModel):
781    """Provenance object as defined in PEP 740."""
782
783    version: Literal[1] = 1
784    """
785    The provenance object's version, which is always 1.
786    """
787
788    attestation_bundles: list[AttestationBundle]
789    """
790    One or more attestation "bundles".
791    """

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

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