Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/model_signing/verifying.py: 73%

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