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        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            )
113
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
121
122        actual_manifest = hashing_config.hash(
123            model_path, files_to_hash=files_to_hash
124        )
125
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}")
131
132    def _get_manifest_diff(self, actual, expected) -> list[str]:
133        diffs = []
134
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        }
141
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            )
150
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            )
159
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
169
170    def set_hashing_config(self, hashing_config: hashing.Config) -> Self:
171        """Sets the new configuration for hashing models.
172
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.
176
177        Args:
178            hashing_config: The new hashing configuration.
179
180        Returns:
181            The new signing configuration.
182        """
183        self._hashing_config = hashing_config
184        return self
185
186    def set_ignore_unsigned_files(self, ignore_unsigned_files: bool) -> Self:
187        """Sets whether files that were not signed are to be ignored.
188
189        This method allows to ignore those files that are not part of the
190        manifest and therefor were not originally signed.
191
192        Args:
193            ignore_unsigned_files: whether to ignore unsigned files
194        """
195        self._ignore_unsigned_files = ignore_unsigned_files
196        return self
197
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")
220
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.
230
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).
234
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.
246
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
258
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.
263
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.
268
269        Args:
270            public_key: The path to the public key to verify with.
271
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
278
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.
287
288        The verifier in this configuration is changed to one that performs
289        verification of sigstore bundles signed by a signing certificate.
290
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.
299
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