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 # OMS v1.0 6.1.1: "The verifier MUST apply the same allow_symlinks
109 # policy recorded in serialization.allow_symlinks". Every verify
110 # subcommand passes a hashing config built from its own
111 # --allow-symlinks flag, so without this the caller's flag silently
112 # replaced the signed policy and _guess_hashing_config, which does read
113 # it, never ran (issue #666).
114 recorded_allow_symlinks = expected_manifest.serialization_type.get(
115 "allow_symlinks"
116 )
117 if recorded_allow_symlinks is not None:
118 hashing_config.set_allow_symlinks(recorded_allow_symlinks)
120 if "ignore_paths" in expected_manifest.serialization_type:
121 hashing_config.add_ignored_paths(
122 model_path=model_path,
123 paths=expected_manifest.serialization_type["ignore_paths"],
124 )
126 if self._ignore_unsigned_files:
127 files_to_hash = [
128 model_path / rd.identifier
129 for rd in expected_manifest.resource_descriptors()
130 ]
131 else:
132 files_to_hash = None
134 actual_manifest = hashing_config.hash(
135 model_path, files_to_hash=files_to_hash
136 )
138 if actual_manifest != expected_manifest:
139 diff_message = self._get_manifest_diff(
140 actual_manifest, expected_manifest
141 )
142 raise ValueError(f"Signature mismatch: {diff_message}")
144 def _get_manifest_diff(self, actual, expected) -> list[str]:
145 diffs = []
147 actual_hashes = {
148 rd.identifier: rd.digest for rd in actual.resource_descriptors()
149 }
150 expected_hashes = {
151 rd.identifier: rd.digest for rd in expected.resource_descriptors()
152 }
154 extra_actual_files = set(actual_hashes.keys()) - set(
155 expected_hashes.keys()
156 )
157 if extra_actual_files:
158 diffs.append(
159 f"Extra files found in model '{actual.model_name}': "
160 f"{', '.join(sorted(extra_actual_files))}"
161 )
163 missing_actual_files = set(expected_hashes.keys()) - set(
164 actual_hashes.keys()
165 )
166 if missing_actual_files:
167 diffs.append(
168 f"Missing files in model '{actual.model_name}': "
169 f"{', '.join(sorted(missing_actual_files))}"
170 )
172 common_files = set(actual_hashes.keys()) & set(expected_hashes.keys())
173 for identifier in sorted(common_files):
174 if actual_hashes[identifier] != expected_hashes[identifier]:
175 diffs.append(
176 f"Hash mismatch for '{identifier}': "
177 f"Expected '{expected_hashes[identifier]}', "
178 f"Actual '{actual_hashes[identifier]}'"
179 )
180 return diffs
182 def set_hashing_config(self, hashing_config: hashing.Config) -> Self:
183 """Sets the new configuration for hashing models.
185 After calling this method, the automatic guessing of the hashing
186 configuration used during signing is no longer possible from within one
187 instance of this class.
189 Args:
190 hashing_config: The new hashing configuration.
192 Returns:
193 The new signing configuration.
194 """
195 self._hashing_config = hashing_config
196 return self
198 def set_ignore_unsigned_files(self, ignore_unsigned_files: bool) -> Self:
199 """Sets whether files that were not signed are to be ignored.
201 This method allows to ignore those files that are not part of the
202 manifest and therefor were not originally signed.
204 Args:
205 ignore_unsigned_files: whether to ignore unsigned files
206 """
207 self._ignore_unsigned_files = ignore_unsigned_files
208 return self
210 def _guess_hashing_config(
211 self, source_manifest: manifest.Manifest
212 ) -> hashing.Config:
213 """Attempts to guess the hashing config from a manifest."""
214 args = source_manifest.serialization_type
215 method = args["method"]
216 match method:
217 case "files":
218 return hashing.Config().use_file_serialization(
219 hashing_algorithm=args["hash_type"],
220 allow_symlinks=args["allow_symlinks"],
221 ignore_paths=args.get("ignore_paths", frozenset()),
222 )
223 case "shards":
224 return hashing.Config().use_shard_serialization(
225 hashing_algorithm=args["hash_type"],
226 shard_size=args["shard_size"],
227 allow_symlinks=args["allow_symlinks"],
228 ignore_paths=args.get("ignore_paths", frozenset()),
229 )
230 case _:
231 raise ValueError("Cannot guess the hashing configuration")
233 def use_sigstore_verifier(
234 self,
235 *,
236 identity: str,
237 oidc_issuer: str,
238 use_staging: bool = False,
239 trust_config: pathlib.Path | None = None,
240 ) -> Self:
241 """Configures the verification of signatures produced by Sigstore.
243 The verifier in this configuration is changed to one that performs
244 verification of Sigstore signatures (sigstore bundles signed by
245 keyless signing via Sigstore).
247 Args:
248 identity: The expected identity that has signed the model.
249 oidc_issuer: The expected OpenID Connect issuer that provided the
250 certificate used for the signature.
251 use_staging: Use staging configurations, instead of production. This
252 is supposed to be set to True only when testing. Default is False.
253 trust_config: A path to a custom trust configuration. When provided,
254 the signature verification process will rely on the supplied
255 PKI and trust configurations, instead of the default Sigstore
256 setup. If not specified, the default Sigstore configuration
257 is used.
259 Return:
260 The new verification configuration.
261 """
262 self._uses_sigstore = True
263 self._verifier = sigstore.Verifier(
264 identity=identity,
265 oidc_issuer=oidc_issuer,
266 use_staging=use_staging,
267 trust_config=trust_config,
268 )
269 return self
271 def use_elliptic_key_verifier(
272 self, *, public_key: hashing.PathLike
273 ) -> Self:
274 """Configures the verification of signatures generated by a private key.
276 The verifier in this configuration is changed to one that performs
277 verification of sigstore bundles signed by an elliptic curve private
278 key. The public key used in the configuration must match the private key
279 used during signing.
281 Args:
282 public_key: The path to the public key to verify with.
284 Return:
285 The new verification configuration.
286 """
287 self._uses_sigstore = False
288 self._verifier = ec_key.Verifier(pathlib.Path(public_key))
289 return self
291 def use_certificate_verifier(
292 self,
293 *,
294 certificate_chain: Iterable[hashing.PathLike] = frozenset(),
295 log_fingerprints: bool = False,
296 expected_san_uris: Iterable[str] = frozenset(),
297 ) -> Self:
298 """Configures the verification of signatures generated by a certificate.
300 The verifier in this configuration is changed to one that performs
301 verification of sigstore bundles signed by a signing certificate.
303 Args:
304 certificate_chain: Certificate chain to establish root of trust. If
305 empty, the operating system's one is used.
306 log_fingerprints: Log certificates' SHA256 fingerprints
307 expected_san_uris: Optional URIs that must appear in the leaf
308 certificate's SubjectAltName. Binds the signature to a specific
309 signer identity (e.g. a SPIFFE ID) in addition to
310 chain-of-trust.
312 Return:
313 The new verification configuration.
314 """
315 self._uses_sigstore = False
316 self._verifier = certificate.Verifier(
317 [pathlib.Path(c) for c in certificate_chain],
318 log_fingerprints=log_fingerprints,
319 expected_san_uris=expected_san_uris,
320 )
321 return self