Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# Copyright 2025 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.
15"""Sigstore based signature, signers and verifiers, using protobuf.
17The difference between this module and `sign_sigstore` is that here we use
18Sigstore via the protobuf specs instead of `sigstore-python`. This is to enable
19support for traditional signing and verification. These require additional data
20to be stored in the sigstore bundle used for the signature but `sigstore-python`
21validation does not allow those.
22"""
24import abc
25import base64
26from collections.abc import Callable
27from datetime import datetime
28from functools import cached_property
29import json
30import pathlib
31from typing import Self, cast
32import urllib.error
33import urllib.request
35import rfc3161_client
36from sigstore import models as sigstore_models
37from sigstore_models.bundle import v1 as bundle_pb
38from typing_extensions import override
40from model_signing._signing import signing
43_TSA_CLIENT_TIMEOUT: int = 5
44_USER_AGENT: str = "model-signing"
47def _request_timestamp(
48 url: str, signature: bytes
49) -> rfc3161_client.TimeStampResponse:
50 """Request a timestamp from a TSA for the given signature."""
51 try:
52 timestamp_request = (
53 rfc3161_client.TimestampRequestBuilder()
54 .hash_algorithm(rfc3161_client.HashAlgorithm.SHA256)
55 .data(signature)
56 .nonce(nonce=True)
57 .build()
58 )
59 except ValueError as error:
60 raise ValueError(f"invalid TSA request: {error}") from error
62 req = urllib.request.Request(
63 url,
64 data=timestamp_request.as_bytes(),
65 headers={
66 "Content-Type": "application/timestamp-query",
67 "User-Agent": _USER_AGENT,
68 },
69 )
71 try:
72 with urllib.request.urlopen(req, timeout=_TSA_CLIENT_TIMEOUT) as resp:
73 response_bytes = resp.read()
74 except urllib.error.URLError as error:
75 raise ValueError(f"TSA request failed: {error}") from error
77 try:
78 return rfc3161_client.decode_timestamp_response(response_bytes)
79 except ValueError as error:
80 raise ValueError(f"invalid TSA response: {error}") from error
83# The media type to use when creating the protobuf based Sigstore bundle
84_BUNDLE_MEDIA_TYPE: str = "application/vnd.dev.sigstore.bundle.v0.3+json"
87def pae(raw_payload: bytes) -> bytes:
88 """Generates the PAE encoding of statement from the payload.
90 This is an internal of `sigstore_python`, but since in this module and
91 classes derived from the signer and verifier defined here we cannot use
92 `sigstore_python`, we have to reimplement this.
94 See https://github.com/secure-systems-lab/dsse/blob/v1.0.0/protocol.md
95 for details.
97 Args:
98 raw_payload: The raw payload to encode.
100 Returns:
101 The encoded statement from the payload.
102 """
103 payload_type = signing._IN_TOTO_JSON_PAYLOAD_TYPE
104 payload_type_length = len(payload_type)
105 payload_length = len(raw_payload)
106 pae_str = f"DSSEv1 {payload_type_length} {payload_type} {payload_length}"
107 return b" ".join([pae_str.encode("utf-8"), raw_payload])
110def pae_compat(raw_payload: bytes) -> bytes:
111 """Generates the PAE encoding of statement from the payload.
113 This is the same as `pae`, but using the version as defined in v0.2.0 of the
114 `model_signing` library. Due to a bug in that implementation, signatures
115 generated at that version have to be verified using this compat patch. The
116 issue is that the raw payload, which is bytes, is added to a string and then
117 encoded back as bytes, so we get additional escape characters included.
119 Args:
120 raw_payload: The raw payload to encode.
122 Returns:
123 The encoded statement from the payload.
124 """
125 payload_type = signing._IN_TOTO_JSON_PAYLOAD_TYPE
126 payload_type_length = len(payload_type)
127 payload_length = len(raw_payload)
128 # Notice bug here!
129 pae_str = (
130 f"DSSEV1 {payload_type_length} {payload_type} "
131 f"{payload_length} {raw_payload}"
132 )
133 return pae_str.encode("utf-8")
136class Signature(signing.Signature):
137 """Sigstore signature support, wrapping around `bundle_pb.Bundle`."""
139 def __init__(self, bundle: bundle_pb.Bundle):
140 """Builds an instance of this signature.
142 Args:
143 bundle: the sigstore bundle (in `bundle_pb.Bundle` format).
144 """
145 self.bundle = bundle
147 @override
148 def write(self, path: pathlib.Path) -> None:
149 path.write_bytes(self.to_bytes())
151 @override
152 def to_bytes(self) -> bytes:
153 return self.bundle.to_json().encode("utf-8")
155 @classmethod
156 @override
157 def read(cls, path: pathlib.Path) -> Self:
158 content = path.read_text(encoding="utf-8")
159 parsed_dict = json.loads(content)
161 # adjust parsed_dict due to previous usage of protobufs
162 if "tlogEntries" not in parsed_dict["verificationMaterial"]:
163 parsed_dict["verificationMaterial"]["tlogEntries"] = []
164 if "publicKey" in parsed_dict["verificationMaterial"]:
165 if "hint" not in parsed_dict["verificationMaterial"]["publicKey"]:
166 parsed_dict["verificationMaterial"]["publicKey"]["hint"] = None
167 for k in ["rawBytes", "keyDetails"]:
168 if k in parsed_dict["verificationMaterial"]["publicKey"]:
169 del parsed_dict["verificationMaterial"]["publicKey"][k]
171 return cls(bundle_pb.Bundle.from_dict(parsed_dict))
174class Signer(signing.Signer):
175 """Signer for traditional signing.
177 This is subclassed for each traditional signing method we support.
178 """
181class Verifier(signing.Verifier):
182 """Verifier for traditional signature verification.
184 This is subclassed for each traditional signing method we support.
185 """
187 def __init__(
188 self,
189 trust_config: pathlib.Path | None = None,
190 require_timestamp: bool = False,
191 ) -> None:
192 """Initializes optional timestamp verification trust material.
194 Args:
195 trust_config: A Sigstore client trust configuration or trusted root.
196 When omitted, the Sigstore production trusted root is used for
197 timestamp verification.
198 require_timestamp: Reject bundles without an RFC 3161 timestamp.
199 """
200 self._trust_config = trust_config
201 self._require_timestamp = require_timestamp
203 @cached_property
204 def _trusted_root(self) -> sigstore_models.TrustedRoot:
205 return _load_trusted_root(self._trust_config)
207 @override
208 def _verify_signed_content(
209 self, signature: signing.Signature
210 ) -> tuple[str, bytes]:
211 # We are guaranteed to only use the local signature type
212 signature = cast(Signature, signature)
213 bundle = signature.bundle
215 # Since the bundle is done via protobuf, check media type first
216 if bundle.media_type != _BUNDLE_MEDIA_TYPE:
217 raise ValueError(
218 f"Invalid sigstore bundle, got media type {bundle.media_type} "
219 f"but expected {_BUNDLE_MEDIA_TYPE}"
220 )
222 envelope = bundle.dsse_envelope
223 verification_time = verify_timestamp(
224 bundle.verification_material,
225 envelope.signatures[0].sig,
226 self._trust_config,
227 require_timestamp=self._require_timestamp,
228 trusted_root_loader=lambda: self._trusted_root,
229 )
230 return self._verify_bundle(bundle, verification_time)
232 @abc.abstractmethod
233 def _verify_bundle(
234 self, bundle: bundle_pb.Bundle, verification_time: datetime | None
235 ) -> tuple[str, bytes]:
236 """Verifies the bundle to extract the payload type and payload.
238 Since the bundle is generated via proto, we need to do more checks to
239 replace what `verify_dsse` from `sigstore_python` does.
240 """
243def request_timestamp(
244 signature_bytes: bytes, tsa_url: str
245) -> bundle_pb.TimestampVerificationData:
246 """Requests a timestamp from a TSA and returns verification data.
248 Args:
249 signature_bytes: The signature bytes to timestamp.
250 tsa_url: The URL of the RFC 3161 Timestamp Authority.
252 Returns:
253 TimestampVerificationData to include in the bundle.
254 """
255 response = _request_timestamp(tsa_url, signature_bytes)
257 return bundle_pb.TimestampVerificationData(
258 rfc3161_timestamps=[
259 bundle_pb.RFC3161SignedTimestamp(
260 # sigstore-models constructors expect bytes in base64 form.
261 signed_timestamp=base64.b64encode(response.as_bytes())
262 )
263 ]
264 )
267def verify_timestamp(
268 verification_material: bundle_pb.VerificationMaterial,
269 signature_bytes: bytes,
270 trust_config: pathlib.Path | None = None,
271 *,
272 require_timestamp: bool = False,
273 trusted_root_loader: (
274 Callable[[], sigstore_models.TrustedRoot] | None
275 ) = None,
276) -> datetime | None:
277 """Verifies and extracts an RFC 3161 timestamp from a bundle.
279 Args:
280 verification_material: The bundle's verification material.
281 signature_bytes: The signature bytes covered by the timestamp.
282 trust_config: A Sigstore client trust configuration or trusted root.
283 When omitted, the Sigstore production trusted root is used.
284 require_timestamp: Reject verification material without a timestamp.
285 trusted_root_loader: Optional lazy source of cached timestamp trust.
287 Returns:
288 The verified timestamp, or None if the bundle has no timestamp.
290 Raises:
291 ValueError: The timestamp is malformed, untrusted, or does not cover
292 the supplied signature.
293 """
294 ts_data = verification_material.timestamp_verification_data
295 if not ts_data or not ts_data.rfc3161_timestamps:
296 if require_timestamp:
297 raise ValueError("An RFC 3161 timestamp is required")
298 return None
300 if len(ts_data.rfc3161_timestamps) != 1:
301 raise ValueError("Expected exactly one RFC 3161 timestamp")
302 try:
303 response = rfc3161_client.decode_timestamp_response(
304 ts_data.rfc3161_timestamps[0].signed_timestamp
305 )
306 except ValueError as error:
307 raise ValueError(f"Invalid RFC 3161 timestamp: {error}") from error
309 trusted_root = (
310 trusted_root_loader()
311 if trusted_root_loader
312 else _load_trusted_root(trust_config)
313 )
314 authorities = trusted_root.get_timestamp_authorities()
315 if not authorities:
316 raise ValueError("No timestamp authorities found in the trusted root")
318 for authority in authorities:
319 certificates = authority.certificates(allow_expired=True)
320 if len(certificates) < 2:
321 continue
323 try:
324 verifier = rfc3161_client.VerifierBuilder(
325 tsa_certificate=certificates[0],
326 intermediates=certificates[1:-1],
327 roots=[certificates[-1]],
328 ).build()
329 verifier.verify_message(response, signature_bytes)
330 except (ValueError, rfc3161_client.VerificationError):
331 continue
333 timestamp = response.tst_info.gen_time
334 if authority.validity_period_start <= timestamp and (
335 authority.validity_period_end is None
336 or timestamp < authority.validity_period_end
337 ):
338 return timestamp
340 raise ValueError("Invalid RFC 3161 timestamp or untrusted authority")
343def _load_trusted_root(
344 trust_config: pathlib.Path | None,
345) -> sigstore_models.TrustedRoot:
346 """Loads timestamp trust from a client config or standalone trusted root."""
347 if trust_config is None:
348 return sigstore_models.ClientTrustConfig.production().trusted_root
350 contents = trust_config.read_text()
351 document = json.loads(contents)
352 if "trustedRoot" in document:
353 config = sigstore_models.ClientTrustConfig.from_json(contents)
354 return config.trusted_root
355 return sigstore_models.TrustedRoot.from_file(str(trust_config))