Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# This file is dual licensed under the terms of the Apache License, Version
2# 2.0, and the BSD License. See the LICENSE file in the root of this repository
3# for complete details.
5from __future__ import annotations
7import email.base64mime
8import email.generator
9import email.message
10import email.policy
11import io
12import typing
13from collections.abc import Iterable
15from cryptography import utils, x509
16from cryptography.exceptions import UnsupportedAlgorithm, _Reasons
17from cryptography.hazmat.bindings._rust import pkcs7 as rust_pkcs7
18from cryptography.hazmat.primitives import hashes, serialization
19from cryptography.hazmat.primitives.asymmetric import ec, padding, rsa
20from cryptography.hazmat.primitives.ciphers import (
21 algorithms,
22)
23from cryptography.utils import _check_byteslike
25load_pem_pkcs7_certificates = rust_pkcs7.load_pem_pkcs7_certificates
27load_der_pkcs7_certificates = rust_pkcs7.load_der_pkcs7_certificates
29serialize_certificates = rust_pkcs7.serialize_certificates
31PKCS7HashTypes = typing.Union[
32 hashes.SHA224,
33 hashes.SHA256,
34 hashes.SHA384,
35 hashes.SHA512,
36]
38PKCS7PrivateKeyTypes = typing.Union[
39 rsa.RSAPrivateKey, ec.EllipticCurvePrivateKey
40]
42ContentEncryptionAlgorithm = typing.Union[
43 type[algorithms.AES128], type[algorithms.AES256]
44]
47class PKCS7Options(utils.Enum):
48 Text = "Add text/plain MIME type"
49 Binary = "Don't translate input data into canonical MIME format"
50 DetachedSignature = "Don't embed data in the PKCS7 structure"
51 NoCapabilities = "Don't embed SMIME capabilities"
52 NoAttributes = "Don't embed authenticatedAttributes"
53 NoCerts = "Don't embed signer certificate"
56class PKCS7SignatureBuilder:
57 def __init__(
58 self,
59 data: utils.Buffer | None = None,
60 signers: list[
61 tuple[
62 x509.Certificate,
63 PKCS7PrivateKeyTypes,
64 PKCS7HashTypes,
65 padding.PSS | padding.PKCS1v15 | None,
66 ]
67 ] = [],
68 additional_certs: list[x509.Certificate] = [],
69 ):
70 self._data = data
71 self._signers = signers
72 self._additional_certs = additional_certs
74 def set_data(self, data: utils.Buffer) -> PKCS7SignatureBuilder:
75 _check_byteslike("data", data)
76 if self._data is not None:
77 raise ValueError("data may only be set once")
79 return PKCS7SignatureBuilder(data, self._signers)
81 def add_signer(
82 self,
83 certificate: x509.Certificate,
84 private_key: PKCS7PrivateKeyTypes,
85 hash_algorithm: PKCS7HashTypes,
86 *,
87 rsa_padding: padding.PSS | padding.PKCS1v15 | None = None,
88 ) -> PKCS7SignatureBuilder:
89 if not isinstance(
90 hash_algorithm,
91 (
92 hashes.SHA224,
93 hashes.SHA256,
94 hashes.SHA384,
95 hashes.SHA512,
96 ),
97 ):
98 raise TypeError(
99 "hash_algorithm must be one of hashes.SHA224, "
100 "SHA256, SHA384, or SHA512"
101 )
102 if not isinstance(certificate, x509.Certificate):
103 raise TypeError("certificate must be a x509.Certificate")
105 if not isinstance(
106 private_key, (rsa.RSAPrivateKey, ec.EllipticCurvePrivateKey)
107 ):
108 raise TypeError("Only RSA & EC keys are supported at this time.")
110 if rsa_padding is not None:
111 if not isinstance(rsa_padding, (padding.PSS, padding.PKCS1v15)):
112 raise TypeError("Padding must be PSS or PKCS1v15")
113 if not isinstance(private_key, rsa.RSAPrivateKey):
114 raise TypeError("Padding is only supported for RSA keys")
116 return PKCS7SignatureBuilder(
117 self._data,
118 [
119 *self._signers,
120 (certificate, private_key, hash_algorithm, rsa_padding),
121 ],
122 )
124 def add_certificate(
125 self, certificate: x509.Certificate
126 ) -> PKCS7SignatureBuilder:
127 if not isinstance(certificate, x509.Certificate):
128 raise TypeError("certificate must be a x509.Certificate")
130 return PKCS7SignatureBuilder(
131 self._data, self._signers, [*self._additional_certs, certificate]
132 )
134 def sign(
135 self,
136 encoding: serialization.Encoding,
137 options: Iterable[PKCS7Options],
138 backend: typing.Any = None,
139 ) -> bytes:
140 if len(self._signers) == 0:
141 raise ValueError("Must have at least one signer")
142 if self._data is None:
143 raise ValueError("You must add data to sign")
144 options = list(options)
145 if not all(isinstance(x, PKCS7Options) for x in options):
146 raise ValueError("options must be from the PKCS7Options enum")
147 if encoding not in (
148 serialization.Encoding.PEM,
149 serialization.Encoding.DER,
150 serialization.Encoding.SMIME,
151 ):
152 raise ValueError(
153 "Must be PEM, DER, or SMIME from the Encoding enum"
154 )
156 # Text is a meaningless option unless it is accompanied by
157 # DetachedSignature
158 if (
159 PKCS7Options.Text in options
160 and PKCS7Options.DetachedSignature not in options
161 ):
162 raise ValueError(
163 "When passing the Text option you must also pass "
164 "DetachedSignature"
165 )
167 if PKCS7Options.Text in options and encoding in (
168 serialization.Encoding.DER,
169 serialization.Encoding.PEM,
170 ):
171 raise ValueError(
172 "The Text option is only available for SMIME serialization"
173 )
175 # No attributes implies no capabilities so we'll error if you try to
176 # pass both.
177 if (
178 PKCS7Options.NoAttributes in options
179 and PKCS7Options.NoCapabilities in options
180 ):
181 raise ValueError(
182 "NoAttributes is a superset of NoCapabilities. Do not pass "
183 "both values."
184 )
186 return rust_pkcs7.sign_and_serialize(self, encoding, options)
189class PKCS7EnvelopeBuilder:
190 def __init__(
191 self,
192 *,
193 _data: bytes | None = None,
194 _recipients: list[x509.Certificate] | None = None,
195 _content_encryption_algorithm: ContentEncryptionAlgorithm
196 | None = None,
197 ):
198 from cryptography.hazmat.backends.openssl.backend import (
199 backend as ossl,
200 )
202 if not ossl.rsa_encryption_supported(padding=padding.PKCS1v15()):
203 raise UnsupportedAlgorithm(
204 "RSA with PKCS1 v1.5 padding is not supported by this version"
205 " of OpenSSL.",
206 _Reasons.UNSUPPORTED_PADDING,
207 )
208 self._data = _data
209 self._recipients = _recipients if _recipients is not None else []
210 self._content_encryption_algorithm = _content_encryption_algorithm
212 def set_data(self, data: bytes) -> PKCS7EnvelopeBuilder:
213 _check_byteslike("data", data)
214 if self._data is not None:
215 raise ValueError("data may only be set once")
217 return PKCS7EnvelopeBuilder(
218 _data=data,
219 _recipients=self._recipients,
220 _content_encryption_algorithm=self._content_encryption_algorithm,
221 )
223 def add_recipient(
224 self,
225 certificate: x509.Certificate,
226 ) -> PKCS7EnvelopeBuilder:
227 if not isinstance(certificate, x509.Certificate):
228 raise TypeError("certificate must be a x509.Certificate")
230 if not isinstance(certificate.public_key(), rsa.RSAPublicKey):
231 raise TypeError("Only RSA keys are supported at this time.")
233 return PKCS7EnvelopeBuilder(
234 _data=self._data,
235 _recipients=[
236 *self._recipients,
237 certificate,
238 ],
239 _content_encryption_algorithm=self._content_encryption_algorithm,
240 )
242 def set_content_encryption_algorithm(
243 self, content_encryption_algorithm: ContentEncryptionAlgorithm
244 ) -> PKCS7EnvelopeBuilder:
245 if self._content_encryption_algorithm is not None:
246 raise ValueError("Content encryption algo may only be set once")
247 if content_encryption_algorithm not in {
248 algorithms.AES128,
249 algorithms.AES256,
250 }:
251 raise TypeError("Only AES128 and AES256 are supported")
253 return PKCS7EnvelopeBuilder(
254 _data=self._data,
255 _recipients=self._recipients,
256 _content_encryption_algorithm=content_encryption_algorithm,
257 )
259 def encrypt(
260 self,
261 encoding: serialization.Encoding,
262 options: Iterable[PKCS7Options],
263 ) -> bytes:
264 if len(self._recipients) == 0:
265 raise ValueError("Must have at least one recipient")
266 if self._data is None:
267 raise ValueError("You must add data to encrypt")
269 # The default content encryption algorithm is AES-128-CBC, which the
270 # S/MIME v3.2 RFC specifies as MUST support (https://datatracker.ietf.org/doc/html/rfc5751#section-2.7)
271 # however rest of S/MIME v3.2 is not currently supported
272 content_encryption_algorithm = (
273 self._content_encryption_algorithm or algorithms.AES128
274 )
276 options = list(options)
277 if not all(isinstance(x, PKCS7Options) for x in options):
278 raise ValueError("options must be from the PKCS7Options enum")
279 if encoding not in (
280 serialization.Encoding.PEM,
281 serialization.Encoding.DER,
282 serialization.Encoding.SMIME,
283 ):
284 raise ValueError(
285 "Must be PEM, DER, or SMIME from the Encoding enum"
286 )
288 # Only allow options that make sense for encryption
289 if any(
290 opt not in [PKCS7Options.Text, PKCS7Options.Binary]
291 for opt in options
292 ):
293 raise ValueError(
294 "Only the following options are supported for encryption: "
295 "Text, Binary"
296 )
297 elif PKCS7Options.Text in options and PKCS7Options.Binary in options:
298 # OpenSSL accepts both options at the same time, but ignores Text.
299 # We fail defensively to avoid unexpected outputs.
300 raise ValueError(
301 "Cannot use Binary and Text options at the same time"
302 )
304 return rust_pkcs7.encrypt_and_serialize(
305 self, content_encryption_algorithm, encoding, options
306 )
309pkcs7_decrypt_der = rust_pkcs7.decrypt_der
310pkcs7_decrypt_pem = rust_pkcs7.decrypt_pem
311pkcs7_decrypt_smime = rust_pkcs7.decrypt_smime
314def _smime_signed_encode(
315 data: bytes, signature: bytes, micalg: str, text_mode: bool
316) -> bytes:
317 # This function works pretty hard to replicate what OpenSSL does
318 # precisely. For good and for ill.
320 m = email.message.Message()
321 m.add_header("MIME-Version", "1.0")
322 m.add_header(
323 "Content-Type",
324 "multipart/signed",
325 protocol="application/x-pkcs7-signature",
326 micalg=micalg,
327 )
329 m.preamble = "This is an S/MIME signed message\n"
331 msg_part = OpenSSLMimePart()
332 msg_part.set_payload(data)
333 if text_mode:
334 msg_part.add_header("Content-Type", "text/plain")
335 m.attach(msg_part)
337 sig_part = email.message.MIMEPart()
338 sig_part.add_header(
339 "Content-Type", "application/x-pkcs7-signature", name="smime.p7s"
340 )
341 sig_part.add_header("Content-Transfer-Encoding", "base64")
342 sig_part.add_header(
343 "Content-Disposition", "attachment", filename="smime.p7s"
344 )
345 sig_part.set_payload(
346 email.base64mime.body_encode(signature, maxlinelen=65)
347 )
348 del sig_part["MIME-Version"]
349 m.attach(sig_part)
351 fp = io.BytesIO()
352 g = email.generator.BytesGenerator(
353 fp,
354 maxheaderlen=0,
355 mangle_from_=False,
356 policy=m.policy.clone(linesep="\r\n"),
357 )
358 g.flatten(m)
359 return fp.getvalue()
362def _smime_enveloped_encode(data: bytes) -> bytes:
363 m = email.message.Message()
364 m.add_header("MIME-Version", "1.0")
365 m.add_header("Content-Disposition", "attachment", filename="smime.p7m")
366 m.add_header(
367 "Content-Type",
368 "application/pkcs7-mime",
369 smime_type="enveloped-data",
370 name="smime.p7m",
371 )
372 m.add_header("Content-Transfer-Encoding", "base64")
374 m.set_payload(email.base64mime.body_encode(data, maxlinelen=65))
376 return m.as_bytes(policy=m.policy.clone(linesep="\n", max_line_length=0))
379def _smime_enveloped_decode(data: bytes) -> bytes:
380 m = email.message_from_bytes(data)
381 if m.get_content_type() not in {
382 "application/x-pkcs7-mime",
383 "application/pkcs7-mime",
384 }:
385 raise ValueError("Not an S/MIME enveloped message")
386 return bytes(m.get_payload(decode=True))
389def _smime_remove_text_headers(data: bytes) -> bytes:
390 m = email.message_from_bytes(data)
391 # Using get() instead of get_content_type() since it has None as default,
392 # where the latter has "text/plain". Both methods are case-insensitive.
393 content_type = m.get("content-type")
394 if content_type is None:
395 raise ValueError(
396 "Decrypted MIME data has no 'Content-Type' header. "
397 "Please remove the 'Text' option to parse it manually."
398 )
399 if "text/plain" not in content_type:
400 raise ValueError(
401 f"Decrypted MIME data content type is '{content_type}', not "
402 "'text/plain'. Remove the 'Text' option to parse it manually."
403 )
404 return bytes(m.get_payload(decode=True))
407class OpenSSLMimePart(email.message.MIMEPart):
408 # A MIMEPart subclass that replicates OpenSSL's behavior of not including
409 # a newline if there are no headers.
410 def _write_headers(self, generator) -> None:
411 if list(self.raw_items()):
412 generator._write_headers(self)