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 2024 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"""High level API for the verification interface of `model_signing` library.
17This module supports configuring the verification method used to verify a model,
18before performing the verification.
20```python
21model_signing.verifying.Config().use_sigstore_verifier(
22 identity=identity, oidc_issuer=oidc_provider
23).verify("finbert", "finbert.sig")
24```
26The same verification configuration can be used to verify multiple models:
28```python
29verifying_config = model_signing.signing.Config().use_elliptic_key_verifier(
30 public_key="key.pub"
31)
33for model in all_models:
34 verifying_config.verify(model, f"{model}_sharded.sig")
35```
37The API defined here is stable and backwards compatible.
38"""
40from collections.abc import Iterable
41import copy
42import pathlib
43import sys
45from model_signing import hashing
46from model_signing import manifest
47from model_signing._signing import sign_certificate as certificate
48from model_signing._signing import sign_ec_key as ec_key
49from model_signing._signing import sign_sigstore as sigstore
50from model_signing._signing import sign_sigstore_pb as sigstore_pb
53if sys.version_info >= (3, 11):
54 from typing import Self
55else:
56 from typing_extensions import Self
59class Config:
60 """Configuration to use when verifying models against signatures.
62 The verification configuration is needed to determine how to read and verify
63 the signature. Given we support multiple signing format, the verification
64 settings must match the signing ones.
66 The configuration also supports configuring the hashing configuration from
67 `model_signing.hashing`. This should also match the configuration used
68 during signing. However, by default, we can attempt to guess it from the
69 signature.
70 """
72 def __init__(self):
73 """Initializes the default configuration for verification."""
74 self._hashing_config = None
75 self._verifier = None
76 self._uses_sigstore = False
77 self._ignore_unsigned_files = False
79 def verify(
80 self, model_path: hashing.PathLike, signature_path: hashing.PathLike
81 ):
82 """Verifies that a model conforms to a signature.
84 Args:
85 model_path: The path to the model to verify.
86 signature_path: The path to the signature file.
88 Raises:
89 ValueError: No verifier has been configured.
90 """
91 if self._verifier is None:
92 raise ValueError("Attempting to verify with no configured verifier")
94 if self._uses_sigstore:
95 signature = sigstore.Signature.read(pathlib.Path(signature_path))
96 else:
97 signature = sigstore_pb.Signature.read(pathlib.Path(signature_path))
99 expected_manifest = self._verifier.verify(signature)
101 if self._hashing_config is not None:
102 # The signed manifest's ignore paths are applied below. Copy the
103 # config so they do not mutate the caller's config or accumulate
104 # into later verify() calls on a reused instance.
105 hashing_config = copy.deepcopy(self._hashing_config)
106 else:
107 hashing_config = self._guess_hashing_config(expected_manifest)
108 if "ignore_paths" in expected_manifest.serialization_type:
109 hashing_config.add_ignored_paths(
110 model_path=model_path,
111 paths=expected_manifest.serialization_type["ignore_paths"],
112 )
114 if self._ignore_unsigned_files:
115 files_to_hash = [
116 model_path / rd.identifier
117 for rd in expected_manifest.resource_descriptors()
118 ]
119 else:
120 files_to_hash = None
122 actual_manifest = hashing_config.hash(
123 model_path, files_to_hash=files_to_hash
124 )
126 if actual_manifest != expected_manifest:
127 diff_message = self._get_manifest_diff(
128 actual_manifest, expected_manifest
129 )
130 raise ValueError(f"Signature mismatch: {diff_message}")
132 def _get_manifest_diff(self, actual, expected) -> list[str]:
133 diffs = []
135 actual_hashes = {
136 rd.identifier: rd.digest for rd in actual.resource_descriptors()
137 }
138 expected_hashes = {
139 rd.identifier: rd.digest for rd in expected.resource_descriptors()
140 }
142 extra_actual_files = set(actual_hashes.keys()) - set(
143 expected_hashes.keys()
144 )
145 if extra_actual_files:
146 diffs.append(
147 f"Extra files found in model '{actual.model_name}': "
148 f"{', '.join(sorted(extra_actual_files))}"
149 )
151 missing_actual_files = set(expected_hashes.keys()) - set(
152 actual_hashes.keys()
153 )
154 if missing_actual_files:
155 diffs.append(
156 f"Missing files in model '{actual.model_name}': "
157 f"{', '.join(sorted(missing_actual_files))}"
158 )
160 common_files = set(actual_hashes.keys()) & set(expected_hashes.keys())
161 for identifier in sorted(common_files):
162 if actual_hashes[identifier] != expected_hashes[identifier]:
163 diffs.append(
164 f"Hash mismatch for '{identifier}': "
165 f"Expected '{expected_hashes[identifier]}', "
166 f"Actual '{actual_hashes[identifier]}'"
167 )
168 return diffs
170 def set_hashing_config(self, hashing_config: hashing.Config) -> Self:
171 """Sets the new configuration for hashing models.
173 After calling this method, the automatic guessing of the hashing
174 configuration used during signing is no longer possible from within one
175 instance of this class.
177 Args:
178 hashing_config: The new hashing configuration.
180 Returns:
181 The new signing configuration.
182 """
183 self._hashing_config = hashing_config
184 return self
186 def set_ignore_unsigned_files(self, ignore_unsigned_files: bool) -> Self:
187 """Sets whether files that were not signed are to be ignored.
189 This method allows to ignore those files that are not part of the
190 manifest and therefor were not originally signed.
192 Args:
193 ignore_unsigned_files: whether to ignore unsigned files
194 """
195 self._ignore_unsigned_files = ignore_unsigned_files
196 return self
198 def _guess_hashing_config(
199 self, source_manifest: manifest.Manifest
200 ) -> hashing.Config:
201 """Attempts to guess the hashing config from a manifest."""
202 args = source_manifest.serialization_type
203 method = args["method"]
204 match method:
205 case "files":
206 return hashing.Config().use_file_serialization(
207 hashing_algorithm=args["hash_type"],
208 allow_symlinks=args["allow_symlinks"],
209 ignore_paths=args.get("ignore_paths", frozenset()),
210 )
211 case "shards":
212 return hashing.Config().use_shard_serialization(
213 hashing_algorithm=args["hash_type"],
214 shard_size=args["shard_size"],
215 allow_symlinks=args["allow_symlinks"],
216 ignore_paths=args.get("ignore_paths", frozenset()),
217 )
218 case _:
219 raise ValueError("Cannot guess the hashing configuration")
221 def use_sigstore_verifier(
222 self,
223 *,
224 identity: str,
225 oidc_issuer: str,
226 use_staging: bool = False,
227 trust_config: pathlib.Path | None = None,
228 ) -> Self:
229 """Configures the verification of signatures produced by Sigstore.
231 The verifier in this configuration is changed to one that performs
232 verification of Sigstore signatures (sigstore bundles signed by
233 keyless signing via Sigstore).
235 Args:
236 identity: The expected identity that has signed the model.
237 oidc_issuer: The expected OpenID Connect issuer that provided the
238 certificate used for the signature.
239 use_staging: Use staging configurations, instead of production. This
240 is supposed to be set to True only when testing. Default is False.
241 trust_config: A path to a custom trust configuration. When provided,
242 the signature verification process will rely on the supplied
243 PKI and trust configurations, instead of the default Sigstore
244 setup. If not specified, the default Sigstore configuration
245 is used.
247 Return:
248 The new verification configuration.
249 """
250 self._uses_sigstore = True
251 self._verifier = sigstore.Verifier(
252 identity=identity,
253 oidc_issuer=oidc_issuer,
254 use_staging=use_staging,
255 trust_config=trust_config,
256 )
257 return self
259 def use_elliptic_key_verifier(
260 self, *, public_key: hashing.PathLike
261 ) -> Self:
262 """Configures the verification of signatures generated by a private key.
264 The verifier in this configuration is changed to one that performs
265 verification of sigstore bundles signed by an elliptic curve private
266 key. The public key used in the configuration must match the private key
267 used during signing.
269 Args:
270 public_key: The path to the public key to verify with.
272 Return:
273 The new verification configuration.
274 """
275 self._uses_sigstore = False
276 self._verifier = ec_key.Verifier(pathlib.Path(public_key))
277 return self
279 def use_certificate_verifier(
280 self,
281 *,
282 certificate_chain: Iterable[hashing.PathLike] = frozenset(),
283 log_fingerprints: bool = False,
284 expected_san_uris: Iterable[str] = frozenset(),
285 ) -> Self:
286 """Configures the verification of signatures generated by a certificate.
288 The verifier in this configuration is changed to one that performs
289 verification of sigstore bundles signed by a signing certificate.
291 Args:
292 certificate_chain: Certificate chain to establish root of trust. If
293 empty, the operating system's one is used.
294 log_fingerprints: Log certificates' SHA256 fingerprints
295 expected_san_uris: Optional URIs that must appear in the leaf
296 certificate's SubjectAltName. Binds the signature to a specific
297 signer identity (e.g. a SPIFFE ID) in addition to
298 chain-of-trust.
300 Return:
301 The new verification configuration.
302 """
303 self._uses_sigstore = False
304 self._verifier = certificate.Verifier(
305 [pathlib.Path(c) for c in certificate_chain],
306 log_fingerprints=log_fingerprints,
307 expected_san_uris=expected_san_uris,
308 )
309 return self