We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
S/MIME signature: "Content-Type: text/plain" added to message (39.0.0)
Hey,

I'm observing a weird behavior when signing with S/MIME since version 39.0.0: it seems like an additional "Content-Type: text/plain" is added to the message after signing it.

My code looks like this:

```
from cryptography import x509
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.serialization import pkcs7

with open("test.cert", "rb") as cc:
    cert = x509.load_pem_x509_certificate(cc.read())

with open("test.key", "rb") as ck:
    key = serialization.load_pem_private_key(ck.read(), None)

msg = "test"

options = [pkcs7.PKCS7Options.DetachedSignature, pkcs7.PKCS7Options.Text]

signed_msg = (
    pkcs7.PKCS7SignatureBuilder()
    .set_data(bytes(msg, "utf-8"))
    .add_signer(cert, key, hashes.SHA256())
    .sign(serialization.Encoding.SMIME, options)
)

with open("/tmp/msg.txt", "wb") as msg_file:
    msg_file.write(signed_msg)
```

With `cryptography 38.0.4`, the relevant part is
```
This is an S/MIME signed message

------A78EE028A05FC73322259A20ED9EAAAE
Content-Type: text/plain

test
------A78EE028A05FC73322259A20ED9EAAAE
```
and `openssl smime -verify -in /tmp/msg.txt -noverify` is successful.

With `cryptography 39.0.0`, I get
```
This is an S/MIME signed message

--===============0873037622973214044==
Content-Type: text/plain

Content-Type: text/plain

test
--===============0873037622973214044==
```
and the verification with `openssl` fails.
After manually removing the additional "Content-Type: text/plain" from the file, it succeeds again.

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `src/cryptography/hazmat/primitives/serialization/pkcs7.py`
Content:
```
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.
4 
5 import email.base64mime
6 import email.generator
7 import email.message
8 import io
9 import typing
10 
11 from cryptography import utils, x509
12 from cryptography.hazmat.bindings._rust import pkcs7 as rust_pkcs7
13 from cryptography.hazmat.primitives import hashes, serialization
14 from cryptography.hazmat.primitives.asymmetric import ec, rsa
15 from cryptography.utils import _check_byteslike
16 
17 
18 def load_pem_pkcs7_certificates(data: bytes) -> typing.List[x509.Certificate]:
19     from cryptography.hazmat.backends.openssl.backend import backend
20 
21     return backend.load_pem_pkcs7_certificates(data)
22 
23 
24 def load_der_pkcs7_certificates(data: bytes) -> typing.List[x509.Certificate]:
25     from cryptography.hazmat.backends.openssl.backend import backend
26 
27     return backend.load_der_pkcs7_certificates(data)
28 
29 
30 def serialize_certificates(
31     certs: typing.List[x509.Certificate],
32     encoding: serialization.Encoding,
33 ) -> bytes:
34     return rust_pkcs7.serialize_certificates(certs, encoding)
35 
36 
37 _ALLOWED_PKCS7_HASH_TYPES = typing.Union[
38     hashes.SHA224,
39     hashes.SHA256,
40     hashes.SHA384,
41     hashes.SHA512,
42 ]
43 
44 _ALLOWED_PRIVATE_KEY_TYPES = typing.Union[
45     rsa.RSAPrivateKey, ec.EllipticCurvePrivateKey
46 ]
47 
48 
49 class PKCS7Options(utils.Enum):
50     Text = "Add text/plain MIME type"
51     Binary = "Don't translate input data into canonical MIME format"
52     DetachedSignature = "Don't embed data in the PKCS7 structure"
53     NoCapabilities = "Don't embed SMIME capabilities"
54     NoAttributes = "Don't embed authenticatedAttributes"
55     NoCerts = "Don't embed signer certificate"
56 
57 
58 class PKCS7SignatureBuilder:
59     def __init__(
60         self,
61         data: typing.Optional[bytes] = None,
62         signers: typing.List[
63             typing.Tuple[
64                 x509.Certificate,
65                 _ALLOWED_PRIVATE_KEY_TYPES,
66                 _ALLOWED_PKCS7_HASH_TYPES,
67             ]
68         ] = [],
69         additional_certs: typing.List[x509.Certificate] = [],
70     ):
71         self._data = data
72         self._signers = signers
73         self._additional_certs = additional_certs
74 
75     def set_data(self, data: bytes) -> "PKCS7SignatureBuilder":
76         _check_byteslike("data", data)
77         if self._data is not None:
78             raise ValueError("data may only be set once")
79 
80         return PKCS7SignatureBuilder(bytes(data), self._signers)
81 
82     def add_signer(
83         self,
84         certificate: x509.Certificate,
85         private_key: _ALLOWED_PRIVATE_KEY_TYPES,
86         hash_algorithm: _ALLOWED_PKCS7_HASH_TYPES,
87     ) -> "PKCS7SignatureBuilder":
88         if not isinstance(
89             hash_algorithm,
90             (
91                 hashes.SHA1,
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.SHA1, SHA224, "
100                 "SHA256, SHA384, or SHA512"
101             )
102         if not isinstance(certificate, x509.Certificate):
103             raise TypeError("certificate must be a x509.Certificate")
104 
105         if not isinstance(
106             private_key, (rsa.RSAPrivateKey, ec.EllipticCurvePrivateKey)
107         ):
108             raise TypeError("Only RSA & EC keys are supported at this time.")
109 
110         return PKCS7SignatureBuilder(
111             self._data,
112             self._signers + [(certificate, private_key, hash_algorithm)],
113         )
114 
115     def add_certificate(
116         self, certificate: x509.Certificate
117     ) -> "PKCS7SignatureBuilder":
118         if not isinstance(certificate, x509.Certificate):
119             raise TypeError("certificate must be a x509.Certificate")
120 
121         return PKCS7SignatureBuilder(
122             self._data, self._signers, self._additional_certs + [certificate]
123         )
124 
125     def sign(
126         self,
127         encoding: serialization.Encoding,
128         options: typing.Iterable[PKCS7Options],
129         backend: typing.Any = None,
130     ) -> bytes:
131         if len(self._signers) == 0:
132             raise ValueError("Must have at least one signer")
133         if self._data is None:
134             raise ValueError("You must add data to sign")
135         options = list(options)
136         if not all(isinstance(x, PKCS7Options) for x in options):
137             raise ValueError("options must be from the PKCS7Options enum")
138         if encoding not in (
139             serialization.Encoding.PEM,
140             serialization.Encoding.DER,
141             serialization.Encoding.SMIME,
142         ):
143             raise ValueError(
144                 "Must be PEM, DER, or SMIME from the Encoding enum"
145             )
146 
147         # Text is a meaningless option unless it is accompanied by
148         # DetachedSignature
149         if (
150             PKCS7Options.Text in options
151             and PKCS7Options.DetachedSignature not in options
152         ):
153             raise ValueError(
154                 "When passing the Text option you must also pass "
155                 "DetachedSignature"
156             )
157 
158         if PKCS7Options.Text in options and encoding in (
159             serialization.Encoding.DER,
160             serialization.Encoding.PEM,
161         ):
162             raise ValueError(
163                 "The Text option is only available for SMIME serialization"
164             )
165 
166         # No attributes implies no capabilities so we'll error if you try to
167         # pass both.
168         if (
169             PKCS7Options.NoAttributes in options
170             and PKCS7Options.NoCapabilities in options
171         ):
172             raise ValueError(
173                 "NoAttributes is a superset of NoCapabilities. Do not pass "
174                 "both values."
175             )
176 
177         return rust_pkcs7.sign_and_serialize(self, encoding, options)
178 
179 
180 def _smime_encode(data: bytes, signature: bytes, micalg: str) -> bytes:
181     # This function works pretty hard to replicate what OpenSSL does
182     # precisely. For good and for ill.
183 
184     m = email.message.Message()
185     m.add_header("MIME-Version", "1.0")
186     m.add_header(
187         "Content-Type",
188         "multipart/signed",
189         protocol="application/x-pkcs7-signature",
190         micalg=micalg,
191     )
192 
193     m.preamble = "This is an S/MIME signed message\n"
194 
195     msg_part = email.message.MIMEPart()
196     msg_part.set_payload(data)
197     msg_part.add_header("Content-Type", "text/plain")
198     m.attach(msg_part)
199 
200     sig_part = email.message.MIMEPart()
201     sig_part.add_header(
202         "Content-Type", "application/x-pkcs7-signature", name="smime.p7s"
203     )
204     sig_part.add_header("Content-Transfer-Encoding", "base64")
205     sig_part.add_header(
206         "Content-Disposition", "attachment", filename="smime.p7s"
207     )
208     sig_part.set_payload(
209         email.base64mime.body_encode(signature, maxlinelen=65)
210     )
211     del sig_part["MIME-Version"]
212     m.attach(sig_part)
213 
214     fp = io.BytesIO()
215     g = email.generator.BytesGenerator(
216         fp, maxheaderlen=0, mangle_from_=False, policy=m.policy
217     )
218     g.flatten(m)
219     return fp.getvalue()
220
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/src/cryptography/hazmat/primitives/serialization/pkcs7.py b/src/cryptography/hazmat/primitives/serialization/pkcs7.py
--- a/src/cryptography/hazmat/primitives/serialization/pkcs7.py
+++ b/src/cryptography/hazmat/primitives/serialization/pkcs7.py
@@ -5,6 +5,7 @@
 import email.base64mime
 import email.generator
 import email.message
+import email.policy
 import io
 import typing
 
@@ -177,7 +178,9 @@
         return rust_pkcs7.sign_and_serialize(self, encoding, options)
 
 
-def _smime_encode(data: bytes, signature: bytes, micalg: str) -> bytes:
+def _smime_encode(
+    data: bytes, signature: bytes, micalg: str, text_mode: bool
+) -> bytes:
     # This function works pretty hard to replicate what OpenSSL does
     # precisely. For good and for ill.
 
@@ -192,9 +195,10 @@
 
     m.preamble = "This is an S/MIME signed message\n"
 
-    msg_part = email.message.MIMEPart()
+    msg_part = OpenSSLMimePart()
     msg_part.set_payload(data)
-    msg_part.add_header("Content-Type", "text/plain")
+    if text_mode:
+        msg_part.add_header("Content-Type", "text/plain")
     m.attach(msg_part)
 
     sig_part = email.message.MIMEPart()
@@ -213,7 +217,18 @@
 
     fp = io.BytesIO()
     g = email.generator.BytesGenerator(
-        fp, maxheaderlen=0, mangle_from_=False, policy=m.policy
+        fp,
+        maxheaderlen=0,
+        mangle_from_=False,
+        policy=m.policy.clone(linesep="\r\n"),
     )
     g.flatten(m)
     return fp.getvalue()
+
+
+class OpenSSLMimePart(email.message.MIMEPart):
+    # A MIMEPart subclass that replicates OpenSSL's behavior of not including
+    # a newline if there are no headers.
+    def _write_headers(self, generator) -> None:
+        if list(self.raw_items()):
+            generator._write_headers(self)

{"golden_diff": "diff --git a/src/cryptography/hazmat/primitives/serialization/pkcs7.py b/src/cryptography/hazmat/primitives/serialization/pkcs7.py\n--- a/src/cryptography/hazmat/primitives/serialization/pkcs7.py\n+++ b/src/cryptography/hazmat/primitives/serialization/pkcs7.py\n@@ -5,6 +5,7 @@\n import email.base64mime\n import email.generator\n import email.message\n+import email.policy\n import io\n import typing\n \n@@ -177,7 +178,9 @@\n         return rust_pkcs7.sign_and_serialize(self, encoding, options)\n \n \n-def _smime_encode(data: bytes, signature: bytes, micalg: str) -> bytes:\n+def _smime_encode(\n+    data: bytes, signature: bytes, micalg: str, text_mode: bool\n+) -> bytes:\n     # This function works pretty hard to replicate what OpenSSL does\n     # precisely. For good and for ill.\n \n@@ -192,9 +195,10 @@\n \n     m.preamble = \"This is an S/MIME signed message\\n\"\n \n-    msg_part = email.message.MIMEPart()\n+    msg_part = OpenSSLMimePart()\n     msg_part.set_payload(data)\n-    msg_part.add_header(\"Content-Type\", \"text/plain\")\n+    if text_mode:\n+        msg_part.add_header(\"Content-Type\", \"text/plain\")\n     m.attach(msg_part)\n \n     sig_part = email.message.MIMEPart()\n@@ -213,7 +217,18 @@\n \n     fp = io.BytesIO()\n     g = email.generator.BytesGenerator(\n-        fp, maxheaderlen=0, mangle_from_=False, policy=m.policy\n+        fp,\n+        maxheaderlen=0,\n+        mangle_from_=False,\n+        policy=m.policy.clone(linesep=\"\\r\\n\"),\n     )\n     g.flatten(m)\n     return fp.getvalue()\n+\n+\n+class OpenSSLMimePart(email.message.MIMEPart):\n+    # A MIMEPart subclass that replicates OpenSSL's behavior of not including\n+    # a newline if there are no headers.\n+    def _write_headers(self, generator) -> None:\n+        if list(self.raw_items()):\n+            generator._write_headers(self)\n", "issue": "S/MIME signature: \"Content-Type: text/plain\" added to message (39.0.0)\nHey,\r\n\r\nI'm observing a weird behavior when signing with S/MIME since version 39.0.0: it seems like an additional \"Content-Type: text/plain\" is added to the message after signing it.\r\n\r\nMy code looks like this:\r\n\r\n```\r\nfrom cryptography import x509\r\nfrom cryptography.hazmat.primitives import hashes, serialization\r\nfrom cryptography.hazmat.primitives.serialization import pkcs7\r\n\r\nwith open(\"test.cert\", \"rb\") as cc:\r\n    cert = x509.load_pem_x509_certificate(cc.read())\r\n\r\nwith open(\"test.key\", \"rb\") as ck:\r\n    key = serialization.load_pem_private_key(ck.read(), None)\r\n\r\nmsg = \"test\"\r\n\r\noptions = [pkcs7.PKCS7Options.DetachedSignature, pkcs7.PKCS7Options.Text]\r\n\r\nsigned_msg = (\r\n    pkcs7.PKCS7SignatureBuilder()\r\n    .set_data(bytes(msg, \"utf-8\"))\r\n    .add_signer(cert, key, hashes.SHA256())\r\n    .sign(serialization.Encoding.SMIME, options)\r\n)\r\n\r\nwith open(\"/tmp/msg.txt\", \"wb\") as msg_file:\r\n    msg_file.write(signed_msg)\r\n```\r\n\r\nWith `cryptography 38.0.4`, the relevant part is\r\n```\r\nThis is an S/MIME signed message\r\n\r\n------A78EE028A05FC73322259A20ED9EAAAE\r\nContent-Type: text/plain\r\n\r\ntest\r\n------A78EE028A05FC73322259A20ED9EAAAE\r\n```\r\nand `openssl smime -verify -in /tmp/msg.txt -noverify` is successful.\r\n\r\nWith `cryptography 39.0.0`, I get\r\n```\r\nThis is an S/MIME signed message\r\n\r\n--===============0873037622973214044==\r\nContent-Type: text/plain\r\n\r\nContent-Type: text/plain\r\n\r\ntest\r\n--===============0873037622973214044==\r\n```\r\nand the verification with `openssl` fails.\r\nAfter manually removing the additional \"Content-Type: text/plain\" from the file, it succeeds again.\n", "before_files": [{"content": "# This file is dual licensed under the terms of the Apache License, Version\n# 2.0, and the BSD License. See the LICENSE file in the root of this repository\n# for complete details.\n\nimport email.base64mime\nimport email.generator\nimport email.message\nimport io\nimport typing\n\nfrom cryptography import utils, x509\nfrom cryptography.hazmat.bindings._rust import pkcs7 as rust_pkcs7\nfrom cryptography.hazmat.primitives import hashes, serialization\nfrom cryptography.hazmat.primitives.asymmetric import ec, rsa\nfrom cryptography.utils import _check_byteslike\n\n\ndef load_pem_pkcs7_certificates(data: bytes) -> typing.List[x509.Certificate]:\n    from cryptography.hazmat.backends.openssl.backend import backend\n\n    return backend.load_pem_pkcs7_certificates(data)\n\n\ndef load_der_pkcs7_certificates(data: bytes) -> typing.List[x509.Certificate]:\n    from cryptography.hazmat.backends.openssl.backend import backend\n\n    return backend.load_der_pkcs7_certificates(data)\n\n\ndef serialize_certificates(\n    certs: typing.List[x509.Certificate],\n    encoding: serialization.Encoding,\n) -> bytes:\n    return rust_pkcs7.serialize_certificates(certs, encoding)\n\n\n_ALLOWED_PKCS7_HASH_TYPES = typing.Union[\n    hashes.SHA224,\n    hashes.SHA256,\n    hashes.SHA384,\n    hashes.SHA512,\n]\n\n_ALLOWED_PRIVATE_KEY_TYPES = typing.Union[\n    rsa.RSAPrivateKey, ec.EllipticCurvePrivateKey\n]\n\n\nclass PKCS7Options(utils.Enum):\n    Text = \"Add text/plain MIME type\"\n    Binary = \"Don't translate input data into canonical MIME format\"\n    DetachedSignature = \"Don't embed data in the PKCS7 structure\"\n    NoCapabilities = \"Don't embed SMIME capabilities\"\n    NoAttributes = \"Don't embed authenticatedAttributes\"\n    NoCerts = \"Don't embed signer certificate\"\n\n\nclass PKCS7SignatureBuilder:\n    def __init__(\n        self,\n        data: typing.Optional[bytes] = None,\n        signers: typing.List[\n            typing.Tuple[\n                x509.Certificate,\n                _ALLOWED_PRIVATE_KEY_TYPES,\n                _ALLOWED_PKCS7_HASH_TYPES,\n            ]\n        ] = [],\n        additional_certs: typing.List[x509.Certificate] = [],\n    ):\n        self._data = data\n        self._signers = signers\n        self._additional_certs = additional_certs\n\n    def set_data(self, data: bytes) -> \"PKCS7SignatureBuilder\":\n        _check_byteslike(\"data\", data)\n        if self._data is not None:\n            raise ValueError(\"data may only be set once\")\n\n        return PKCS7SignatureBuilder(bytes(data), self._signers)\n\n    def add_signer(\n        self,\n        certificate: x509.Certificate,\n        private_key: _ALLOWED_PRIVATE_KEY_TYPES,\n        hash_algorithm: _ALLOWED_PKCS7_HASH_TYPES,\n    ) -> \"PKCS7SignatureBuilder\":\n        if not isinstance(\n            hash_algorithm,\n            (\n                hashes.SHA1,\n                hashes.SHA224,\n                hashes.SHA256,\n                hashes.SHA384,\n                hashes.SHA512,\n            ),\n        ):\n            raise TypeError(\n                \"hash_algorithm must be one of hashes.SHA1, SHA224, \"\n                \"SHA256, SHA384, or SHA512\"\n            )\n        if not isinstance(certificate, x509.Certificate):\n            raise TypeError(\"certificate must be a x509.Certificate\")\n\n        if not isinstance(\n            private_key, (rsa.RSAPrivateKey, ec.EllipticCurvePrivateKey)\n        ):\n            raise TypeError(\"Only RSA & EC keys are supported at this time.\")\n\n        return PKCS7SignatureBuilder(\n            self._data,\n            self._signers + [(certificate, private_key, hash_algorithm)],\n        )\n\n    def add_certificate(\n        self, certificate: x509.Certificate\n    ) -> \"PKCS7SignatureBuilder\":\n        if not isinstance(certificate, x509.Certificate):\n            raise TypeError(\"certificate must be a x509.Certificate\")\n\n        return PKCS7SignatureBuilder(\n            self._data, self._signers, self._additional_certs + [certificate]\n        )\n\n    def sign(\n        self,\n        encoding: serialization.Encoding,\n        options: typing.Iterable[PKCS7Options],\n        backend: typing.Any = None,\n    ) -> bytes:\n        if len(self._signers) == 0:\n            raise ValueError(\"Must have at least one signer\")\n        if self._data is None:\n            raise ValueError(\"You must add data to sign\")\n        options = list(options)\n        if not all(isinstance(x, PKCS7Options) for x in options):\n            raise ValueError(\"options must be from the PKCS7Options enum\")\n        if encoding not in (\n            serialization.Encoding.PEM,\n            serialization.Encoding.DER,\n            serialization.Encoding.SMIME,\n        ):\n            raise ValueError(\n                \"Must be PEM, DER, or SMIME from the Encoding enum\"\n            )\n\n        # Text is a meaningless option unless it is accompanied by\n        # DetachedSignature\n        if (\n            PKCS7Options.Text in options\n            and PKCS7Options.DetachedSignature not in options\n        ):\n            raise ValueError(\n                \"When passing the Text option you must also pass \"\n                \"DetachedSignature\"\n            )\n\n        if PKCS7Options.Text in options and encoding in (\n            serialization.Encoding.DER,\n            serialization.Encoding.PEM,\n        ):\n            raise ValueError(\n                \"The Text option is only available for SMIME serialization\"\n            )\n\n        # No attributes implies no capabilities so we'll error if you try to\n        # pass both.\n        if (\n            PKCS7Options.NoAttributes in options\n            and PKCS7Options.NoCapabilities in options\n        ):\n            raise ValueError(\n                \"NoAttributes is a superset of NoCapabilities. Do not pass \"\n                \"both values.\"\n            )\n\n        return rust_pkcs7.sign_and_serialize(self, encoding, options)\n\n\ndef _smime_encode(data: bytes, signature: bytes, micalg: str) -> bytes:\n    # This function works pretty hard to replicate what OpenSSL does\n    # precisely. For good and for ill.\n\n    m = email.message.Message()\n    m.add_header(\"MIME-Version\", \"1.0\")\n    m.add_header(\n        \"Content-Type\",\n        \"multipart/signed\",\n        protocol=\"application/x-pkcs7-signature\",\n        micalg=micalg,\n    )\n\n    m.preamble = \"This is an S/MIME signed message\\n\"\n\n    msg_part = email.message.MIMEPart()\n    msg_part.set_payload(data)\n    msg_part.add_header(\"Content-Type\", \"text/plain\")\n    m.attach(msg_part)\n\n    sig_part = email.message.MIMEPart()\n    sig_part.add_header(\n        \"Content-Type\", \"application/x-pkcs7-signature\", name=\"smime.p7s\"\n    )\n    sig_part.add_header(\"Content-Transfer-Encoding\", \"base64\")\n    sig_part.add_header(\n        \"Content-Disposition\", \"attachment\", filename=\"smime.p7s\"\n    )\n    sig_part.set_payload(\n        email.base64mime.body_encode(signature, maxlinelen=65)\n    )\n    del sig_part[\"MIME-Version\"]\n    m.attach(sig_part)\n\n    fp = io.BytesIO()\n    g = email.generator.BytesGenerator(\n        fp, maxheaderlen=0, mangle_from_=False, policy=m.policy\n    )\n    g.flatten(m)\n    return fp.getvalue()\n", "path": "src/cryptography/hazmat/primitives/serialization/pkcs7.py"}], "after_files": [{"content": "# This file is dual licensed under the terms of the Apache License, Version\n# 2.0, and the BSD License. See the LICENSE file in the root of this repository\n# for complete details.\n\nimport email.base64mime\nimport email.generator\nimport email.message\nimport email.policy\nimport io\nimport typing\n\nfrom cryptography import utils, x509\nfrom cryptography.hazmat.bindings._rust import pkcs7 as rust_pkcs7\nfrom cryptography.hazmat.primitives import hashes, serialization\nfrom cryptography.hazmat.primitives.asymmetric import ec, rsa\nfrom cryptography.utils import _check_byteslike\n\n\ndef load_pem_pkcs7_certificates(data: bytes) -> typing.List[x509.Certificate]:\n    from cryptography.hazmat.backends.openssl.backend import backend\n\n    return backend.load_pem_pkcs7_certificates(data)\n\n\ndef load_der_pkcs7_certificates(data: bytes) -> typing.List[x509.Certificate]:\n    from cryptography.hazmat.backends.openssl.backend import backend\n\n    return backend.load_der_pkcs7_certificates(data)\n\n\ndef serialize_certificates(\n    certs: typing.List[x509.Certificate],\n    encoding: serialization.Encoding,\n) -> bytes:\n    return rust_pkcs7.serialize_certificates(certs, encoding)\n\n\n_ALLOWED_PKCS7_HASH_TYPES = typing.Union[\n    hashes.SHA224,\n    hashes.SHA256,\n    hashes.SHA384,\n    hashes.SHA512,\n]\n\n_ALLOWED_PRIVATE_KEY_TYPES = typing.Union[\n    rsa.RSAPrivateKey, ec.EllipticCurvePrivateKey\n]\n\n\nclass PKCS7Options(utils.Enum):\n    Text = \"Add text/plain MIME type\"\n    Binary = \"Don't translate input data into canonical MIME format\"\n    DetachedSignature = \"Don't embed data in the PKCS7 structure\"\n    NoCapabilities = \"Don't embed SMIME capabilities\"\n    NoAttributes = \"Don't embed authenticatedAttributes\"\n    NoCerts = \"Don't embed signer certificate\"\n\n\nclass PKCS7SignatureBuilder:\n    def __init__(\n        self,\n        data: typing.Optional[bytes] = None,\n        signers: typing.List[\n            typing.Tuple[\n                x509.Certificate,\n                _ALLOWED_PRIVATE_KEY_TYPES,\n                _ALLOWED_PKCS7_HASH_TYPES,\n            ]\n        ] = [],\n        additional_certs: typing.List[x509.Certificate] = [],\n    ):\n        self._data = data\n        self._signers = signers\n        self._additional_certs = additional_certs\n\n    def set_data(self, data: bytes) -> \"PKCS7SignatureBuilder\":\n        _check_byteslike(\"data\", data)\n        if self._data is not None:\n            raise ValueError(\"data may only be set once\")\n\n        return PKCS7SignatureBuilder(bytes(data), self._signers)\n\n    def add_signer(\n        self,\n        certificate: x509.Certificate,\n        private_key: _ALLOWED_PRIVATE_KEY_TYPES,\n        hash_algorithm: _ALLOWED_PKCS7_HASH_TYPES,\n    ) -> \"PKCS7SignatureBuilder\":\n        if not isinstance(\n            hash_algorithm,\n            (\n                hashes.SHA1,\n                hashes.SHA224,\n                hashes.SHA256,\n                hashes.SHA384,\n                hashes.SHA512,\n            ),\n        ):\n            raise TypeError(\n                \"hash_algorithm must be one of hashes.SHA1, SHA224, \"\n                \"SHA256, SHA384, or SHA512\"\n            )\n        if not isinstance(certificate, x509.Certificate):\n            raise TypeError(\"certificate must be a x509.Certificate\")\n\n        if not isinstance(\n            private_key, (rsa.RSAPrivateKey, ec.EllipticCurvePrivateKey)\n        ):\n            raise TypeError(\"Only RSA & EC keys are supported at this time.\")\n\n        return PKCS7SignatureBuilder(\n            self._data,\n            self._signers + [(certificate, private_key, hash_algorithm)],\n        )\n\n    def add_certificate(\n        self, certificate: x509.Certificate\n    ) -> \"PKCS7SignatureBuilder\":\n        if not isinstance(certificate, x509.Certificate):\n            raise TypeError(\"certificate must be a x509.Certificate\")\n\n        return PKCS7SignatureBuilder(\n            self._data, self._signers, self._additional_certs + [certificate]\n        )\n\n    def sign(\n        self,\n        encoding: serialization.Encoding,\n        options: typing.Iterable[PKCS7Options],\n        backend: typing.Any = None,\n    ) -> bytes:\n        if len(self._signers) == 0:\n            raise ValueError(\"Must have at least one signer\")\n        if self._data is None:\n            raise ValueError(\"You must add data to sign\")\n        options = list(options)\n        if not all(isinstance(x, PKCS7Options) for x in options):\n            raise ValueError(\"options must be from the PKCS7Options enum\")\n        if encoding not in (\n            serialization.Encoding.PEM,\n            serialization.Encoding.DER,\n            serialization.Encoding.SMIME,\n        ):\n            raise ValueError(\n                \"Must be PEM, DER, or SMIME from the Encoding enum\"\n            )\n\n        # Text is a meaningless option unless it is accompanied by\n        # DetachedSignature\n        if (\n            PKCS7Options.Text in options\n            and PKCS7Options.DetachedSignature not in options\n        ):\n            raise ValueError(\n                \"When passing the Text option you must also pass \"\n                \"DetachedSignature\"\n            )\n\n        if PKCS7Options.Text in options and encoding in (\n            serialization.Encoding.DER,\n            serialization.Encoding.PEM,\n        ):\n            raise ValueError(\n                \"The Text option is only available for SMIME serialization\"\n            )\n\n        # No attributes implies no capabilities so we'll error if you try to\n        # pass both.\n        if (\n            PKCS7Options.NoAttributes in options\n            and PKCS7Options.NoCapabilities in options\n        ):\n            raise ValueError(\n                \"NoAttributes is a superset of NoCapabilities. Do not pass \"\n                \"both values.\"\n            )\n\n        return rust_pkcs7.sign_and_serialize(self, encoding, options)\n\n\ndef _smime_encode(\n    data: bytes, signature: bytes, micalg: str, text_mode: bool\n) -> bytes:\n    # This function works pretty hard to replicate what OpenSSL does\n    # precisely. For good and for ill.\n\n    m = email.message.Message()\n    m.add_header(\"MIME-Version\", \"1.0\")\n    m.add_header(\n        \"Content-Type\",\n        \"multipart/signed\",\n        protocol=\"application/x-pkcs7-signature\",\n        micalg=micalg,\n    )\n\n    m.preamble = \"This is an S/MIME signed message\\n\"\n\n    msg_part = OpenSSLMimePart()\n    msg_part.set_payload(data)\n    if text_mode:\n        msg_part.add_header(\"Content-Type\", \"text/plain\")\n    m.attach(msg_part)\n\n    sig_part = email.message.MIMEPart()\n    sig_part.add_header(\n        \"Content-Type\", \"application/x-pkcs7-signature\", name=\"smime.p7s\"\n    )\n    sig_part.add_header(\"Content-Transfer-Encoding\", \"base64\")\n    sig_part.add_header(\n        \"Content-Disposition\", \"attachment\", filename=\"smime.p7s\"\n    )\n    sig_part.set_payload(\n        email.base64mime.body_encode(signature, maxlinelen=65)\n    )\n    del sig_part[\"MIME-Version\"]\n    m.attach(sig_part)\n\n    fp = io.BytesIO()\n    g = email.generator.BytesGenerator(\n        fp,\n        maxheaderlen=0,\n        mangle_from_=False,\n        policy=m.policy.clone(linesep=\"\\r\\n\"),\n    )\n    g.flatten(m)\n    return fp.getvalue()\n\n\nclass OpenSSLMimePart(email.message.MIMEPart):\n    # A MIMEPart subclass that replicates OpenSSL's behavior of not including\n    # a newline if there are no headers.\n    def _write_headers(self, generator) -> None:\n        if list(self.raw_items()):\n            generator._write_headers(self)\n", "path": "src/cryptography/hazmat/primitives/serialization/pkcs7.py"}]}