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
 43import sys
 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
 53if sys.version_info >= (3, 11):
 54    from typing import Self
 55else:
 56    from typing_extensions import Self
 57
 58
 59class Config:
 60    """Configuration to use when verifying models against signatures.
 61
 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.
 65
 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    """
 71
 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
 78
 79    def verify(
 80        self, model_path: hashing.PathLike, signature_path: hashing.PathLike
 81    ):
 82        """Verifies that a model conforms to a signature.
 83
 84        Args:
 85            model_path: The path to the model to verify.
 86            signature_path: The path to the signature file.
 87
 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")
 93
 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))
 98
 99        expected_manifest = self._verifier.verify(signature)
100
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)
119
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            )
125
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
133
134        actual_manifest = hashing_config.hash(
135            model_path, files_to_hash=files_to_hash
136        )
137
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}")
143
144    def _get_manifest_diff(self, actual, expected) -> list[str]:
145        diffs = []
146
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        }
153
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            )
162
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            )
171
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
181
182    def set_hashing_config(self, hashing_config: hashing.Config) -> Self:
183        """Sets the new configuration for hashing models.
184
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.
188
189        Args:
190            hashing_config: The new hashing configuration.
191
192        Returns:
193            The new signing configuration.
194        """
195        self._hashing_config = hashing_config
196        return self
197
198    def set_ignore_unsigned_files(self, ignore_unsigned_files: bool) -> Self:
199        """Sets whether files that were not signed are to be ignored.
200
201        This method allows to ignore those files that are not part of the
202        manifest and therefor were not originally signed.
203
204        Args:
205            ignore_unsigned_files: whether to ignore unsigned files
206        """
207        self._ignore_unsigned_files = ignore_unsigned_files
208        return self
209
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")
232
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.
242
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).
246
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.
258
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
270
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.
275
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.
280
281        Args:
282            public_key: The path to the public key to verify with.
283
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
290
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.
299
300        The verifier in this configuration is changed to one that performs
301        verification of sigstore bundles signed by a signing certificate.
302
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.
311
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