Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/sigstore/_utils.py: 30%

1# Copyright 2022 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""" 

16Shared utilities. 

17""" 

18 

19from __future__ import annotations 

20 

21import base64 

22import hashlib 

23from datetime import UTC, datetime 

24from importlib import resources 

25from typing import IO, NewType 

26from urllib import parse 

27 

28from cryptography.hazmat.primitives import serialization 

29from cryptography.hazmat.primitives.asymmetric import ec, ed25519, rsa 

30from cryptography.x509 import ( 

31 Certificate, 

32 ExtensionNotFound, 

33 Version, 

34 load_der_x509_certificate, 

35) 

36from cryptography.x509.oid import ExtendedKeyUsageOID, ExtensionOID 

37from sigstore_models.common.v1 import HashAlgorithm, TimeRange 

38 

39from sigstore import hashes as sigstore_hashes 

40from sigstore.errors import VerificationError 

41 

42PublicKey = rsa.RSAPublicKey | ec.EllipticCurvePublicKey | ed25519.Ed25519PublicKey 

43 

44PublicKeyTypes = ( 

45 type[rsa.RSAPublicKey] 

46 | type[ec.EllipticCurvePublicKey] 

47 | type[ed25519.Ed25519PublicKey] 

48) 

49 

50HexStr = NewType("HexStr", str) 

51""" 

52A newtype for `str` objects that contain hexadecimal strings (e.g. `ffabcd00ff`). 

53""" 

54B64Str = NewType("B64Str", str) 

55""" 

56A newtype for `str` objects that contain base64 encoded strings. 

57""" 

58KeyID = NewType("KeyID", bytes) 

59""" 

60A newtype for `bytes` objects that contain a key id. 

61""" 

62 

63 

64def load_pem_public_key( 

65 key_pem: bytes, 

66 *, 

67 types: tuple[PublicKeyTypes, ...] = ( 

68 rsa.RSAPublicKey, 

69 ec.EllipticCurvePublicKey, 

70 ed25519.Ed25519PublicKey, 

71 ), 

72) -> PublicKey: 

73 """ 

74 A specialization of `cryptography`'s `serialization.load_pem_public_key` 

75 with a uniform exception type (`VerificationError`) and filtering on valid key types 

76 for Sigstore purposes. 

77 """ 

78 

79 try: 

80 key = serialization.load_pem_public_key(key_pem) 

81 except Exception as exc: 

82 raise VerificationError("could not load PEM-formatted public key") from exc 

83 

84 if not isinstance(key, types): 

85 raise VerificationError(f"invalid key format: not one of {types}") 

86 

87 return key 

88 

89 

90def load_der_public_key( 

91 key_der: bytes, 

92 *, 

93 types: tuple[PublicKeyTypes, ...] = ( 

94 rsa.RSAPublicKey, 

95 ec.EllipticCurvePublicKey, 

96 ed25519.Ed25519PublicKey, 

97 ), 

98) -> PublicKey: 

99 """ 

100 The `load_pem_public_key` specialization, but DER. 

101 """ 

102 

103 try: 

104 key = serialization.load_der_public_key(key_der) 

105 except Exception as exc: 

106 raise VerificationError("could not load DER-formatted public key") from exc 

107 

108 if not isinstance(key, types): 

109 raise VerificationError(f"invalid key format: not one of {types}") 

110 

111 return key 

112 

113 

114def base64_encode_pem_cert(cert: Certificate) -> B64Str: 

115 """ 

116 Returns a string containing a base64-encoded PEM-encoded X.509 certificate. 

117 """ 

118 

119 return B64Str( 

120 base64.b64encode(cert.public_bytes(serialization.Encoding.PEM)).decode() 

121 ) 

122 

123 

124def cert_der_to_pem(der: bytes) -> str: 

125 """ 

126 Converts a DER-encoded X.509 certificate into its PEM encoding. 

127 

128 Returns a string containing a PEM-encoded X.509 certificate. 

129 """ 

130 

131 # NOTE: Technically we don't have to round-trip like this, since 

132 # the DER-to-PEM transformation is entirely mechanical. 

133 cert = load_der_x509_certificate(der) 

134 return cert.public_bytes(serialization.Encoding.PEM).decode() 

135 

136 

137def key_id(key: PublicKey) -> KeyID: 

138 """ 

139 Returns an RFC 6962-style "key ID" for the given public key. 

140 

141 See: <https://www.rfc-editor.org/rfc/rfc6962#section-3.2> 

142 """ 

143 public_bytes = key.public_bytes( 

144 encoding=serialization.Encoding.DER, 

145 format=serialization.PublicFormat.SubjectPublicKeyInfo, 

146 ) 

147 

148 return KeyID(hashlib.sha256(public_bytes).digest()) 

149 

150 

151def sha256_digest( 

152 input_: bytes | IO[bytes] | sigstore_hashes.Hashed, 

153) -> sigstore_hashes.Hashed: 

154 """ 

155 Compute the SHA256 digest of an input stream or buffer or, 

156 if given a `Hashed`, return it directly. 

157 """ 

158 if isinstance(input_, sigstore_hashes.Hashed): 

159 return input_ 

160 

161 # If the input is already buffered into memory, there's no point in 

162 # going back through an I/O abstraction. 

163 if isinstance(input_, bytes): 

164 return sigstore_hashes.Hashed( 

165 digest=hashlib.sha256(input_).digest(), algorithm=HashAlgorithm.SHA2_256 

166 ) 

167 

168 return sigstore_hashes.Hashed( 

169 digest=_sha256_streaming(input_), algorithm=HashAlgorithm.SHA2_256 

170 ) 

171 

172 

173def _sha256_streaming(io: IO[bytes]) -> bytes: 

174 """ 

175 Compute the SHA256 of a stream. 

176 

177 This function does its own internal buffering, so an unbuffered stream 

178 should be supplied for optimal performance. 

179 """ 

180 

181 # NOTE: This function performs a SHA256 digest over a stream. 

182 # The stream's size is not checked, meaning that the stream's source 

183 # is implicitly trusted: if an attacker is able to truncate the stream's 

184 # source prematurely, then they could conceivably produce a digest 

185 # for a partial stream. This in turn could conceivably result 

186 # in a valid signature for an unintended (truncated) input. 

187 # 

188 # This is currently outside of sigstore-python's threat model: we 

189 # assume that the stream is trusted. 

190 # 

191 # See: https://github.com/sigstore/sigstore-python/pull/329#discussion_r1041215972 

192 

193 sha256 = hashlib.sha256() 

194 # Per coreutils' ioblksize.h: 128KB performs optimally across a range 

195 # of systems in terms of minimizing syscall overhead. 

196 view = memoryview(bytearray(128 * 1024)) 

197 

198 nbytes = io.readinto(view) # type: ignore[attr-defined] 

199 while nbytes: 

200 sha256.update(view[:nbytes]) 

201 nbytes = io.readinto(view) # type: ignore[attr-defined] 

202 

203 return sha256.digest() 

204 

205 

206def read_embedded(name: str, url: str) -> bytes: 

207 """ 

208 Read a resource for a given TUF repository embedded in this distribution 

209 of sigstore-python, returning its contents as bytes. 

210 """ 

211 embed_dir = parse.quote(url, safe="") 

212 b: bytes = resources.files("sigstore._store").joinpath(embed_dir, name).read_bytes() 

213 return b 

214 

215 

216def cert_is_ca(cert: Certificate) -> bool: 

217 """ 

218 Returns `True` if and only if the given `Certificate` 

219 is a CA certificate. 

220 

221 This function doesn't indicate the trustworthiness of the given 

222 `Certificate`, only whether it has the appropriate interior state. 

223 

224 This function is **not** naively invertible: users **must** use the 

225 dedicated `cert_is_leaf` utility function to determine whether a particular 

226 leaf upholds Sigstore's invariants. 

227 """ 

228 

229 # Only v3 certificates should appear in the context of Sigstore; 

230 # earlier versions of X.509 lack extensions and have ambiguous CA 

231 # behavior. 

232 if cert.version != Version.v3: 

233 raise VerificationError(f"invalid X.509 version: {cert.version}") 

234 

235 # Valid CA certificates must have the following set: 

236 # 

237 # * `BasicKeyUsage.keyCertSign` 

238 # * `BasicConstraints.ca` 

239 # 

240 # Any other combination of states is inconsistent and invalid, meaning 

241 # that we won't consider the certificate a valid non-CA leaf. 

242 

243 try: 

244 basic_constraints = cert.extensions.get_extension_for_oid( 

245 ExtensionOID.BASIC_CONSTRAINTS 

246 ) 

247 

248 # BasicConstraints must be marked as critical, per RFC 5280 4.2.1.9. 

249 if not basic_constraints.critical: 

250 raise VerificationError( 

251 "invalid X.509 certificate: non-critical BasicConstraints in CA" 

252 ) 

253 

254 ca = basic_constraints.value.ca # type: ignore[attr-defined] 

255 except ExtensionNotFound: 

256 # No BasicConstrains means that this can't possibly be a CA. 

257 return False 

258 

259 key_cert_sign = False 

260 try: 

261 key_usage = cert.extensions.get_extension_for_oid(ExtensionOID.KEY_USAGE) 

262 key_cert_sign = key_usage.value.key_cert_sign # type: ignore[attr-defined] 

263 except ExtensionNotFound: 

264 raise VerificationError("invalid X.509 certificate: missing KeyUsage") 

265 

266 # If both states are set, this is a CA. 

267 if ca and key_cert_sign: 

268 return True 

269 

270 if not (ca or key_cert_sign): 

271 return False 

272 

273 # Anything else is an invalid state that should never occur. 

274 raise VerificationError( 

275 f"invalid X.509 certificate states: KeyUsage.keyCertSign={key_cert_sign}" 

276 f", BasicConstraints.ca={ca}" 

277 ) 

278 

279 

280def cert_is_root_ca(cert: Certificate) -> bool: 

281 """ 

282 Returns `True` if and only if the given `Certificate` indicates 

283 that it's a root CA. 

284 

285 This is **not** a verification function, and it does not establish 

286 the trustworthiness of the given certificate. 

287 """ 

288 

289 # NOTE(ww): This function is obnoxiously long to make the different 

290 # states explicit. 

291 

292 # Only v3 certificates should appear in the context of Sigstore; 

293 # earlier versions of X.509 lack extensions and have ambiguous CA 

294 # behavior. 

295 if cert.version != Version.v3: 

296 raise VerificationError(f"invalid X.509 version: {cert.version}") 

297 

298 # Non-CAs can't possibly be root CAs. 

299 if not cert_is_ca(cert): 

300 return False 

301 

302 # A certificate that is its own issuer and signer is considered a root CA. 

303 try: 

304 cert.verify_directly_issued_by(cert) 

305 return True 

306 except Exception: 

307 return False 

308 

309 

310def cert_is_leaf(cert: Certificate) -> bool: 

311 """ 

312 Returns `True` if and only if the given `Certificate` is a valid 

313 leaf certificate for Sigstore purposes. This means that: 

314 

315 * It is not a root or intermediate CA; 

316 * It has `KeyUsage.digitalSignature`; 

317 * It has `CODE_SIGNING` as an `ExtendedKeyUsage`. 

318 

319 This is **not** a verification function, and it does not establish 

320 the trustworthiness of the given certificate. 

321 """ 

322 

323 # Only v3 certificates should appear in the context of Sigstore; 

324 # earlier versions of X.509 lack extensions and have ambiguous CA 

325 # behavior. 

326 if cert.version != Version.v3: 

327 raise VerificationError(f"invalid X.509 version: {cert.version}") 

328 

329 # CAs are not leaves. 

330 if cert_is_ca(cert): 

331 return False 

332 

333 key_usage = cert.extensions.get_extension_for_oid(ExtensionOID.KEY_USAGE) 

334 digital_signature = key_usage.value.digital_signature # type: ignore[attr-defined] 

335 

336 if not digital_signature: 

337 raise VerificationError( 

338 "invalid certificate for Sigstore purposes: missing digital signature usage" 

339 ) 

340 

341 # Finally, we check to make sure the leaf has an `ExtendedKeyUsages` 

342 # extension that includes a codesigning entitlement. Sigstore should 

343 # never issue a leaf that doesn't have this extended usage. 

344 try: 

345 extended_key_usage = cert.extensions.get_extension_for_oid( 

346 ExtensionOID.EXTENDED_KEY_USAGE 

347 ) 

348 

349 return ExtendedKeyUsageOID.CODE_SIGNING in extended_key_usage.value # type: ignore[operator] 

350 except ExtensionNotFound: 

351 raise VerificationError("invalid X.509 certificate: missing ExtendedKeyUsage") 

352 

353 

354def is_timerange_valid(period: TimeRange | None, *, allow_expired: bool) -> bool: 

355 """ 

356 Given a `period`, checks that the the current time is not before `start`. If 

357 `allow_expired` is `False`, also checks that the current time is not after 

358 `end`. 

359 """ 

360 now = datetime.now(UTC) 

361 

362 # If there was no validity period specified, the key is always valid. 

363 if not period: 

364 return True 

365 

366 # Active: if the current time is before the starting period, we are not yet 

367 # valid. 

368 if now < period.start: 

369 return False 

370 

371 # If we want Expired keys, the key is valid at this point. Otherwise, check 

372 # that we are within range. 

373 return allow_expired or (period.end is None or now <= period.end)