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 InvalidEnvelope(Error):
194    """
195    Raised when the associated `Envelope` is invalid in some way.
196    """
197
198
199class Envelope:
200    """
201    Represents a DSSE envelope.
202
203    This class cannot be constructed directly; you must use `sign` or `from_json`.
204
205    See: <https://github.com/secure-systems-lab/dsse/blob/v1.0.0/envelope.md>
206    """
207
208    _TYPE = "application/vnd.in-toto+json"
209
210    def __init__(self, inner: _Envelope) -> None:
211        """
212        @private
213        """
214
215        self._inner = inner
216        self._verify()
217
218    def _verify(self) -> None:
219        """
220        Verify and load the Envelope.
221        """
222        if len(self._inner.signatures) != 1:
223            raise InvalidEnvelope("envelope must contain exactly one signature")
224
225        if not self._inner.signatures[0].sig:
226            raise InvalidEnvelope("envelope signature must be non-empty")
227
228        self._signature_bytes = self._inner.signatures[0].sig
229
230    @classmethod
231    def _from_json(cls, contents: bytes | str) -> Envelope:
232        """Return a DSSE envelope from the given JSON representation."""
233        inner = _Envelope().from_json(contents)
234        return cls(inner)
235
236    def to_json(self) -> str:
237        """
238        Return a JSON string with this DSSE envelope's contents.
239        """
240        return self._inner.to_json()
241
242    def __eq__(self, other: object) -> bool:
243        """Equality for DSSE envelopes."""
244
245        if not isinstance(other, Envelope):
246            return NotImplemented
247
248        return self._inner == other._inner
249
250    @property
251    def signature(self) -> bytes:
252        """Return the decoded bytes of the Envelope signature."""
253        return self._signature_bytes
254
255
256def _pae(type_: str, body: bytes) -> bytes:
257    """
258    Compute the PAE encoding for the given `type_` and `body`.
259    """
260
261    # See:
262    # https://github.com/secure-systems-lab/dsse/blob/v1.0.0/envelope.md
263    # https://github.com/in-toto/attestation/blob/v1.0/spec/v1.0/envelope.md
264    pae = f"DSSEv1 {len(type_)} {type_} ".encode()
265    pae += b" ".join([str(len(body)).encode(), body])
266    return pae
267
268
269def _sign(key: ec.EllipticCurvePrivateKey, stmt: Statement) -> Envelope:
270    """
271    Sign for the given in-toto `Statement`, and encapsulate the resulting
272    signature in a DSSE `Envelope`.
273    """
274    pae = stmt._pae()
275    _logger.debug(f"DSSE PAE: {pae!r}")
276
277    signature = key.sign(pae, ec.ECDSA(hashes.SHA256()))
278    return Envelope(
279        _Envelope(
280            payload=stmt._contents,
281            payload_type=Envelope._TYPE,
282            signatures=[Signature(sig=signature)],
283        )
284    )
285
286
287def _verify(key: ec.EllipticCurvePublicKey, evp: Envelope) -> bytes:
288    """
289    Verify the given in-toto `Envelope`, returning the verified inner payload.
290
291    This function does **not** check the envelope's payload type. The caller
292    is responsible for performing this check.
293    """
294
295    pae = _pae(evp._inner.payload_type, evp._inner.payload)
296
297    nsigs = len(evp._inner.signatures)
298    if nsigs != 1:
299        raise VerificationError(f"DSSE: exactly 1 signature allowed, got {nsigs}")
300
301    signature = evp._inner.signatures[0].sig
302
303    try:
304        key.verify(signature, pae, ec.ECDSA(hashes.SHA256()))
305    except InvalidSignature:
306        raise VerificationError("DSSE: invalid signature")
307
308    return evp._inner.payload