sigstore.dsse

  1# Copyright 2022 The Sigstore Authors
  2#
  3# Licensed under the Apache License, Version 2.0 (the "License");
  4# you may not use this file except in compliance with the License.
  5# You may obtain a copy of the License at
  6#
  7#      http://www.apache.org/licenses/LICENSE-2.0
  8#
  9# Unless required by applicable law or agreed to in writing, software
 10# distributed under the License is distributed on an "AS IS" BASIS,
 11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 12# See the License for the specific language governing permissions and
 13# limitations under the License.
 14
 15"""
 16Functionality for building and manipulating in-toto Statements and DSSE envelopes.
 17"""
 18
 19from __future__ import annotations
 20
 21import logging
 22from typing import Any, Dict, List, Literal, Optional, Union
 23
 24from cryptography.exceptions import InvalidSignature
 25from cryptography.hazmat.primitives import hashes
 26from cryptography.hazmat.primitives.asymmetric import ec
 27from pydantic import BaseModel, ConfigDict, Field, RootModel, StrictStr, ValidationError
 28from sigstore_protobuf_specs.dev.sigstore.common.v1 import HashAlgorithm
 29from sigstore_protobuf_specs.io.intoto import Envelope as _Envelope
 30from sigstore_protobuf_specs.io.intoto import Signature
 31
 32from sigstore.errors import Error, VerificationError
 33from sigstore.hashes import Hashed
 34
 35_logger = logging.getLogger(__name__)
 36
 37Digest = Union[
 38    Literal["sha256"],
 39    Literal["sha384"],
 40    Literal["sha512"],
 41    Literal["sha3_256"],
 42    Literal["sha3_384"],
 43    Literal["sha3_512"],
 44]
 45"""
 46NOTE: in-toto's DigestSet contains all kinds of hash algorithms that
 47we intentionally do not support. This model is limited to common members of the
 48SHA-2 and SHA-3 family that are at least as strong as SHA-256.
 49
 50See: <https://github.com/in-toto/attestation/blob/main/spec/v1/digest_set.md>
 51"""
 52
 53DigestSet = RootModel[Dict[Digest, str]]
 54"""
 55An internal validation model for in-toto subject digest sets.
 56"""
 57
 58
 59class Subject(BaseModel):
 60    """
 61    A single in-toto statement subject.
 62    """
 63
 64    name: Optional[StrictStr]
 65    digest: DigestSet = Field(...)
 66
 67
 68class _Statement(BaseModel):
 69    """
 70    An internal validation model for in-toto statements.
 71    """
 72
 73    model_config = ConfigDict(populate_by_name=True)
 74
 75    type_: Literal["https://in-toto.io/Statement/v1"] = Field(..., alias="_type")
 76    subjects: List[Subject] = Field(..., min_length=1, alias="subject")
 77    predicate_type: StrictStr = Field(..., alias="predicateType")
 78    predicate: Optional[Dict[str, Any]] = Field(None, alias="predicate")
 79
 80
 81class Statement:
 82    """
 83    Represents an in-toto statement.
 84
 85    This type deals with opaque bytes to ensure that the encoding does not
 86    change, but Statements are internally checked for conformance against
 87    the JSON object layout defined in the in-toto attestation spec.
 88
 89    See: <https://github.com/in-toto/attestation/blob/main/spec/v1/statement.md>
 90    """
 91
 92    def __init__(self, contents: bytes | _Statement) -> None:
 93        """
 94        Construct a new Statement.
 95
 96        This takes an opaque `bytes` containing the statement; use
 97        `StatementBuilder` to manually construct an in-toto statement
 98        from constituent pieces.
 99        """
100        if isinstance(contents, bytes):
101            self._contents = contents
102            try:
103                self._inner = _Statement.model_validate_json(contents)
104            except ValidationError:
105                raise Error("malformed in-toto statement")
106        else:
107            self._contents = contents.model_dump_json(by_alias=True).encode()
108            self._inner = contents
109
110    def _matches_digest(self, digest: Hashed) -> bool:
111        """
112        Returns a boolean indicating whether this in-toto Statement contains a subject
113        matching the given digest. The subject's name is **not** checked.
114
115        No digests other than SHA256 are currently supported.
116        """
117        if digest.algorithm != HashAlgorithm.SHA2_256:
118            raise VerificationError(f"unexpected digest algorithm: {digest.algorithm}")
119
120        for sub in self._inner.subjects:
121            sub_digest = sub.digest.root.get("sha256")
122            if sub_digest is None:
123                continue
124            if sub_digest == digest.digest.hex():
125                return True
126
127        return False
128
129    def _pae(self) -> bytes:
130        """
131        Construct the PAE encoding for this statement.
132        """
133
134        return _pae(Envelope._TYPE, self._contents)
135
136
137class StatementBuilder:
138    """
139    A builder-style API for constructing in-toto Statements.
140    """
141
142    def __init__(
143        self,
144        subjects: Optional[List[Subject]] = None,
145        predicate_type: Optional[str] = None,
146        predicate: Optional[Dict[str, Any]] = None,
147    ):
148        """
149        Create a new `StatementBuilder`.
150        """
151        self._subjects = subjects or []
152        self._predicate_type = predicate_type
153        self._predicate = predicate
154
155    def subjects(self, subjects: list[Subject]) -> StatementBuilder:
156        """
157        Configure the subjects for this builder.
158        """
159        self._subjects = subjects
160        return self
161
162    def predicate_type(self, predicate_type: str) -> StatementBuilder:
163        """
164        Configure the predicate type for this builder.
165        """
166        self._predicate_type = predicate_type
167        return self
168
169    def predicate(self, predicate: dict[str, Any]) -> StatementBuilder:
170        """
171        Configure the predicate for this builder.
172        """
173        self._predicate = predicate
174        return self
175
176    def build(self) -> Statement:
177        """
178        Build a `Statement` from the builder's state.
179        """
180        try:
181            stmt = _Statement(
182                type_="https://in-toto.io/Statement/v1",
183                subjects=self._subjects,
184                predicate_type=self._predicate_type,
185                predicate=self._predicate,
186            )
187        except ValidationError as e:
188            raise Error(f"invalid statement: {e}")
189
190        return Statement(stmt)
191
192
193class Envelope:
194    """
195    Represents a DSSE envelope.
196
197    This class cannot be constructed directly; you must use `sign` or `from_json`.
198
199    See: <https://github.com/secure-systems-lab/dsse/blob/v1.0.0/envelope.md>
200    """
201
202    _TYPE = "application/vnd.in-toto+json"
203
204    def __init__(self, inner: _Envelope) -> None:
205        """
206        @private
207        """
208
209        self._inner = inner
210
211    @classmethod
212    def _from_json(cls, contents: bytes | str) -> Envelope:
213        """Return a DSSE envelope from the given JSON representation."""
214        inner = _Envelope().from_json(contents)
215        return cls(inner)
216
217    def to_json(self) -> str:
218        """
219        Return a JSON string with this DSSE envelope's contents.
220        """
221        return self._inner.to_json()
222
223    def __eq__(self, other: object) -> bool:
224        """Equality for DSSE envelopes."""
225
226        if not isinstance(other, Envelope):
227            return NotImplemented
228
229        return self._inner == other._inner
230
231
232def _pae(type_: str, body: bytes) -> bytes:
233    """
234    Compute the PAE encoding for the given `type_` and `body`.
235    """
236
237    # See:
238    # https://github.com/secure-systems-lab/dsse/blob/v1.0.0/envelope.md
239    # https://github.com/in-toto/attestation/blob/v1.0/spec/v1.0/envelope.md
240    pae = f"DSSEv1 {len(type_)} {type_} ".encode()
241    pae += b" ".join([str(len(body)).encode(), body])
242    return pae
243
244
245def _sign(key: ec.EllipticCurvePrivateKey, stmt: Statement) -> Envelope:
246    """
247    Sign for the given in-toto `Statement`, and encapsulate the resulting
248    signature in a DSSE `Envelope`.
249    """
250    pae = stmt._pae()
251    _logger.debug(f"DSSE PAE: {pae!r}")
252
253    signature = key.sign(pae, ec.ECDSA(hashes.SHA256()))
254    return Envelope(
255        _Envelope(
256            payload=stmt._contents,
257            payload_type=Envelope._TYPE,
258            signatures=[Signature(sig=signature)],
259        )
260    )
261
262
263def _verify(key: ec.EllipticCurvePublicKey, evp: Envelope) -> bytes:
264    """
265    Verify the given in-toto `Envelope`, returning the verified inner payload.
266
267    This function does **not** check the envelope's payload type. The caller
268    is responsible for performing this check.
269    """
270
271    pae = _pae(evp._inner.payload_type, evp._inner.payload)
272
273    nsigs = len(evp._inner.signatures)
274    if nsigs != 1:
275        raise VerificationError(f"DSSE: exactly 1 signature allowed, got {nsigs}")
276
277    signature = evp._inner.signatures[0].sig
278
279    try:
280        key.verify(signature, pae, ec.ECDSA(hashes.SHA256()))
281    except InvalidSignature:
282        raise VerificationError("DSSE: invalid signature")
283
284    return evp._inner.payload