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

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""" 

16API for signing artifacts. 

17 

18Example: 

19 

20```python 

21from pathlib import Path 

22from sigstore.models import ClientTrustConfig 

23from sigstore.oidc import Issuer 

24from sigstore.sign import SigningContext 

25 

26artifact_path = Path("README.md") 

27 

28# Construct OIDC Issuer and SigningContext for the 

29# Sigstore Public Good instance 

30trust_config = ClientTrustConfig.production() 

31issuer = Issuer(trust_config.signing_config.get_oidc_url()) 

32context = SigningContext.from_trust_config(trust_config) 

33 

34# Get an identity token from OIDC provider using interactive auth 

35token = issuer.identity_token() 

36 

37# Sign artifact with the identity 

38with context.signer(token, cache = True) as signer: 

39 bundle = signer.sign_artifact(artifact_path.read_bytes()) 

40 

41with Path("README.md.sigstore.json").open("w") as f: 

42 f.write(bundle.to_json()) 

43``` 

44""" 

45 

46from __future__ import annotations 

47 

48import base64 

49import logging 

50from collections.abc import Iterator 

51from contextlib import contextmanager 

52from datetime import datetime, timezone 

53 

54import cryptography.x509 as x509 

55from cryptography.hazmat.primitives import hashes 

56from cryptography.hazmat.primitives.asymmetric import ec 

57from sigstore_models.common.v1 import HashOutput, MessageSignature 

58 

59from sigstore import dsse 

60from sigstore import hashes as sigstore_hashes 

61from sigstore._internal.fulcio import ( 

62 ExpiredCertificate, 

63 FulcioClient, 

64) 

65from sigstore._internal.rekor import EntryRequestBody, RekorLogSubmitter 

66from sigstore._internal.sct import verify_sct 

67from sigstore._internal.timestamp import TimestampAuthorityClient, TimestampError 

68from sigstore._internal.trust import KeyringPurpose 

69from sigstore._utils import sha256_digest 

70from sigstore.models import Bundle, ClientTrustConfig, TrustedRoot 

71from sigstore.oidc import ExpiredIdentity, IdentityToken 

72 

73_logger = logging.getLogger(__name__) 

74 

75 

76class Signer: 

77 """ 

78 The primary API for signing operations. 

79 """ 

80 

81 def __init__( 

82 self, 

83 identity_token: IdentityToken, 

84 signing_ctx: SigningContext, 

85 cache: bool = True, 

86 ) -> None: 

87 """ 

88 Create a new `Signer`. 

89 

90 `identity_token` is the identity token used to request a signing certificate 

91 from Fulcio. 

92 

93 `signing_ctx` is a `SigningContext` that keeps information about the signing 

94 configuration. 

95 

96 `cache` determines whether the signing certificate and ephemeral private key 

97 should be reused (until the certificate expires) to sign different artifacts. 

98 Default is `True`. 

99 """ 

100 self._identity_token = identity_token 

101 self._signing_ctx: SigningContext = signing_ctx 

102 self.__cached_private_key: ec.EllipticCurvePrivateKey | None = None 

103 self.__cached_signing_certificate: x509.Certificate | None = None 

104 if cache: 

105 _logger.debug("Generating ephemeral keys...") 

106 self.__cached_private_key = ec.generate_private_key(ec.SECP256R1()) 

107 _logger.debug("Requesting ephemeral certificate...") 

108 self.__cached_signing_certificate = self._signing_cert() 

109 

110 @property 

111 def _private_key(self) -> ec.EllipticCurvePrivateKey: 

112 """Get or generate a signing key.""" 

113 if self.__cached_private_key is None: 

114 _logger.debug("no cached key; generating ephemeral key") 

115 return ec.generate_private_key(ec.SECP256R1()) 

116 return self.__cached_private_key 

117 

118 def _build_csr(self) -> x509.CertificateSigningRequest: 

119 """ 

120 Build the X.509 Certificate Signing Request submitted to Fulcio. 

121 

122 The CSR carries an empty subject. Fulcio does not validate or use the 

123 CSR's subject (see sigstore/fulcio#863): the certificate's identity is 

124 taken from the OIDC token, not from this field. We previously embedded 

125 the token's identity as an EMAIL_ADDRESS attribute, but that attribute 

126 is encoded as an IA5String (ASCII-only). Some issuers (e.g. GitHub 

127 Actions, whose `sub` claim can contain a non-ASCII environment name) 

128 yield non-ASCII identities; pyca/cryptography does not reject those, so 

129 we emitted a CSR whose IA5String held non-ASCII bytes, which Fulcio's 

130 stricter ASN.1 parser then rejected with HTTP 400 (see 

131 sigstore/sigstore-python#1507). Since the subject is unused, we omit it 

132 entirely. 

133 """ 

134 builder = ( 

135 x509.CertificateSigningRequestBuilder() 

136 .subject_name(x509.Name([])) 

137 .add_extension( 

138 x509.BasicConstraints(ca=False, path_length=None), 

139 critical=True, 

140 ) 

141 ) 

142 return builder.sign(self._private_key, hashes.SHA256()) 

143 

144 def _signing_cert( 

145 self, 

146 ) -> x509.Certificate: 

147 """ 

148 Get or request a signing certificate from Fulcio. 

149 

150 Internally, this performs a CSR against Fulcio and verifies that 

151 the returned certificate is present in Fulcio's CT log. 

152 """ 

153 

154 # If a cached certificate exists, use it until it expires. 

155 if self.__cached_signing_certificate: 

156 not_valid_after = self.__cached_signing_certificate.not_valid_after_utc 

157 if datetime.now(timezone.utc) > not_valid_after: 

158 raise ExpiredCertificate 

159 return self.__cached_signing_certificate 

160 

161 # Our CSR cannot possibly succeed if our underlying identity token 

162 # is expired. 

163 if not self._identity_token.in_validity_period(): 

164 raise ExpiredIdentity 

165 

166 _logger.debug("Retrieving signed certificate...") 

167 

168 certificate_request = self._build_csr() 

169 

170 certificate_response = self._signing_ctx._fulcio.signing_cert.post( 

171 certificate_request, self._identity_token 

172 ) 

173 

174 verify_sct( 

175 certificate_response.cert, 

176 certificate_response.chain, 

177 self._signing_ctx._trusted_root.ct_keyring(KeyringPurpose.SIGN), 

178 ) 

179 

180 _logger.debug("Successfully verified SCT...") 

181 

182 return certificate_response.cert 

183 

184 def _finalize_sign( 

185 self, 

186 cert: x509.Certificate, 

187 content: MessageSignature | dsse.Envelope, 

188 proposed_entry: EntryRequestBody, 

189 ) -> Bundle: 

190 """ 

191 Perform the common "finalizing" steps in a Sigstore signing flow. 

192 """ 

193 # If the user provided TSA urls, timestamps the response 

194 signed_timestamp = [] 

195 for tsa_client in self._signing_ctx._tsa_clients: 

196 try: 

197 signed_timestamp.append(tsa_client.request_timestamp(content.signature)) 

198 except TimestampError as e: 

199 _logger.warning( 

200 f"Unable to use {tsa_client.url} to timestamp the bundle. Failed with {e}" 

201 ) 

202 

203 # Submit the proposed entry to the transparency log 

204 entry = self._signing_ctx._rekor.create_entry(proposed_entry) 

205 _logger.debug( 

206 f"Transparency log entry created with index: {entry._inner.log_index}" 

207 ) 

208 

209 return Bundle._from_parts(cert, content, entry, signed_timestamp) 

210 

211 def sign_dsse( 

212 self, 

213 input_: dsse.Statement, 

214 ) -> Bundle: 

215 """ 

216 Sign the given in-toto statement as a DSSE envelope, and return a 

217 `Bundle` containing the signed result. 

218 

219 This API is **only** for in-toto statements; to sign arbitrary artifacts, 

220 use `sign_artifact` instead. 

221 """ 

222 cert = self._signing_cert() 

223 

224 # Sign the statement, producing a DSSE envelope 

225 content = dsse._sign(self._private_key, input_) 

226 

227 # Create the proposed DSSE log entry 

228 proposed_entry = self._signing_ctx._rekor._build_dsse_request( 

229 envelope=content, certificate=cert 

230 ) 

231 

232 return self._finalize_sign(cert, content, proposed_entry) 

233 

234 def sign_artifact( 

235 self, 

236 input_: bytes | sigstore_hashes.Hashed, 

237 ) -> Bundle: 

238 """ 

239 Sign an artifact, and return a `Bundle` corresponding to the signed result. 

240 

241 The input can be one of two forms: 

242 

243 1. A `bytes` buffer; 

244 2. A `Hashed` object, containing a pre-hashed input (e.g., for inputs 

245 that are too large to buffer into memory). 

246 

247 Regardless of the input format, the signing operation will produce a 

248 `hashedrekord` entry within the bundle. No other entry types 

249 are supported by this API. 

250 """ 

251 

252 cert = self._signing_cert() 

253 

254 # Sign artifact 

255 hashed_input = sha256_digest(input_) 

256 

257 artifact_signature = self._private_key.sign( 

258 hashed_input.digest, ec.ECDSA(hashed_input._as_prehashed()) 

259 ) 

260 

261 content = MessageSignature( 

262 message_digest=HashOutput( 

263 algorithm=hashed_input.algorithm, 

264 digest=base64.b64encode(hashed_input.digest), 

265 ), 

266 signature=base64.b64encode(artifact_signature), 

267 ) 

268 

269 # Create the proposed hashedrekord entry 

270 proposed_entry = self._signing_ctx._rekor._build_hashed_rekord_request( 

271 hashed_input=hashed_input, signature=artifact_signature, certificate=cert 

272 ) 

273 

274 return self._finalize_sign(cert, content, proposed_entry) 

275 

276 

277class SigningContext: 

278 """ 

279 Keep a context between signing operations. 

280 """ 

281 

282 def __init__( 

283 self, 

284 *, 

285 fulcio: FulcioClient, 

286 rekor: RekorLogSubmitter, 

287 trusted_root: TrustedRoot, 

288 tsa_clients: list[TimestampAuthorityClient] | None = None, 

289 ): 

290 """ 

291 Create a new `SigningContext`. 

292 

293 `fulcio` is a `FulcioClient` capable of connecting to a Fulcio instance 

294 and returning signing certificates. 

295 

296 `rekor` is a `RekorClient` capable of connecting to a Rekor instance 

297 and creating transparency log entries. 

298 """ 

299 self._fulcio = fulcio 

300 self._rekor = rekor 

301 self._trusted_root = trusted_root 

302 self._tsa_clients = tsa_clients or [] 

303 

304 @classmethod 

305 def from_trust_config(cls, trust_config: ClientTrustConfig) -> SigningContext: 

306 """ 

307 Create a `SigningContext` from the given `ClientTrustConfig`. 

308 

309 @api private 

310 """ 

311 signing_config = trust_config.signing_config 

312 return cls( 

313 fulcio=signing_config.get_fulcio(), 

314 rekor=signing_config.get_tlogs()[0], 

315 trusted_root=trust_config.trusted_root, 

316 tsa_clients=signing_config.get_tsas(), 

317 ) 

318 

319 @contextmanager 

320 def signer( 

321 self, identity_token: IdentityToken, *, cache: bool = True 

322 ) -> Iterator[Signer]: 

323 """ 

324 A context manager for signing operations. 

325 

326 `identity_token` is the identity token passed to the `Signer` instance 

327 and used to request a signing certificate from Fulcio. 

328 

329 `cache` determines whether the signing certificate and ephemeral private key 

330 generated by the `Signer` instance should be reused (until the certificate expires) 

331 to sign different artifacts. 

332 Default is `True`. 

333 """ 

334 yield Signer(identity_token, self, cache)