model_signing.verifying

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