1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380
|
import io
import textwrap
import unittest
from email import message_from_string, message_from_bytes
from email.message import EmailMessage
from email.generator import Generator, BytesGenerator
from email.headerregistry import Address
from email import policy
import email.errors
from test.test_email import TestEmailBase, parameterize
@parameterize
class TestGeneratorBase:
policy = policy.default
def msgmaker(self, msg, policy=None):
policy = self.policy if policy is None else policy
return self.msgfunc(msg, policy=policy)
refold_long_expected = {
0: textwrap.dedent("""\
To: whom_it_may_concern@example.com
From: nobody_you_want_to_know@example.com
Subject: We the willing led by the unknowing are doing the
impossible for the ungrateful. We have done so much for so long with so little
we are now qualified to do anything with nothing.
None
"""),
40: textwrap.dedent("""\
To: whom_it_may_concern@example.com
From:
nobody_you_want_to_know@example.com
Subject: We the willing led by the
unknowing are doing the impossible for
the ungrateful. We have done so much
for so long with so little we are now
qualified to do anything with nothing.
None
"""),
20: textwrap.dedent("""\
To:
whom_it_may_concern@example.com
From:
nobody_you_want_to_know@example.com
Subject: We the
willing led by the
unknowing are doing
the impossible for
the ungrateful. We
have done so much
for so long with so
little we are now
qualified to do
anything with
nothing.
None
"""),
}
refold_long_expected[100] = refold_long_expected[0]
refold_all_expected = refold_long_expected.copy()
refold_all_expected[0] = (
"To: whom_it_may_concern@example.com\n"
"From: nobody_you_want_to_know@example.com\n"
"Subject: We the willing led by the unknowing are doing the "
"impossible for the ungrateful. We have done so much for "
"so long with so little we are now qualified to do anything "
"with nothing.\n"
"\n"
"None\n")
refold_all_expected[100] = (
"To: whom_it_may_concern@example.com\n"
"From: nobody_you_want_to_know@example.com\n"
"Subject: We the willing led by the unknowing are doing the "
"impossible for the ungrateful. We have\n"
" done so much for so long with so little we are now qualified "
"to do anything with nothing.\n"
"\n"
"None\n")
length_params = [n for n in refold_long_expected]
def length_as_maxheaderlen_parameter(self, n):
msg = self.msgmaker(self.typ(self.refold_long_expected[0]))
s = self.ioclass()
g = self.genclass(s, maxheaderlen=n, policy=self.policy)
g.flatten(msg)
self.assertEqual(s.getvalue(), self.typ(self.refold_long_expected[n]))
def length_as_max_line_length_policy(self, n):
msg = self.msgmaker(self.typ(self.refold_long_expected[0]))
s = self.ioclass()
g = self.genclass(s, policy=self.policy.clone(max_line_length=n))
g.flatten(msg)
self.assertEqual(s.getvalue(), self.typ(self.refold_long_expected[n]))
def length_as_maxheaderlen_parm_overrides_policy(self, n):
msg = self.msgmaker(self.typ(self.refold_long_expected[0]))
s = self.ioclass()
g = self.genclass(s, maxheaderlen=n,
policy=self.policy.clone(max_line_length=10))
g.flatten(msg)
self.assertEqual(s.getvalue(), self.typ(self.refold_long_expected[n]))
def length_as_max_line_length_with_refold_none_does_not_fold(self, n):
msg = self.msgmaker(self.typ(self.refold_long_expected[0]))
s = self.ioclass()
g = self.genclass(s, policy=self.policy.clone(refold_source='none',
max_line_length=n))
g.flatten(msg)
self.assertEqual(s.getvalue(), self.typ(self.refold_long_expected[0]))
def length_as_max_line_length_with_refold_all_folds(self, n):
msg = self.msgmaker(self.typ(self.refold_long_expected[0]))
s = self.ioclass()
g = self.genclass(s, policy=self.policy.clone(refold_source='all',
max_line_length=n))
g.flatten(msg)
self.assertEqual(s.getvalue(), self.typ(self.refold_all_expected[n]))
def test_crlf_control_via_policy(self):
source = "Subject: test\r\n\r\ntest body\r\n"
expected = source
msg = self.msgmaker(self.typ(source))
s = self.ioclass()
g = self.genclass(s, policy=policy.SMTP)
g.flatten(msg)
self.assertEqual(s.getvalue(), self.typ(expected))
def test_flatten_linesep_overrides_policy(self):
source = "Subject: test\n\ntest body\n"
expected = source
msg = self.msgmaker(self.typ(source))
s = self.ioclass()
g = self.genclass(s, policy=policy.SMTP)
g.flatten(msg, linesep='\n')
self.assertEqual(s.getvalue(), self.typ(expected))
def test_set_mangle_from_via_policy(self):
source = textwrap.dedent("""\
Subject: test that
from is mangled in the body!
From time to time I write a rhyme.
""")
variants = (
(None, True),
(policy.compat32, True),
(policy.default, False),
(policy.default.clone(mangle_from_=True), True),
)
for p, mangle in variants:
expected = source.replace('From ', '>From ') if mangle else source
with self.subTest(policy=p, mangle_from_=mangle):
msg = self.msgmaker(self.typ(source))
s = self.ioclass()
g = self.genclass(s, policy=p)
g.flatten(msg)
self.assertEqual(s.getvalue(), self.typ(expected))
def test_compat32_max_line_length_does_not_fold_when_none(self):
msg = self.msgmaker(self.typ(self.refold_long_expected[0]))
s = self.ioclass()
g = self.genclass(s, policy=policy.compat32.clone(max_line_length=None))
g.flatten(msg)
self.assertEqual(s.getvalue(), self.typ(self.refold_long_expected[0]))
def test_rfc2231_wrapping(self):
# This is pretty much just to make sure we don't have an infinite
# loop; I don't expect anyone to hit this in the field.
msg = self.msgmaker(self.typ(textwrap.dedent("""\
To: nobody
Content-Disposition: attachment;
filename="afilenamelongenoghtowraphere"
None
""")))
expected = textwrap.dedent("""\
To: nobody
Content-Disposition: attachment;
filename*0*=us-ascii''afilename;
filename*1*=longenoghtowraphere
None
""")
s = self.ioclass()
g = self.genclass(s, policy=self.policy.clone(max_line_length=33))
g.flatten(msg)
self.assertEqual(s.getvalue(), self.typ(expected))
def test_rfc2231_wrapping_switches_to_default_len_if_too_narrow(self):
# This is just to make sure we don't have an infinite loop; I don't
# expect anyone to hit this in the field, so I'm not bothering to make
# the result optimal (the encoding isn't needed).
msg = self.msgmaker(self.typ(textwrap.dedent("""\
To: nobody
Content-Disposition: attachment;
filename="afilenamelongenoghtowraphere"
None
""")))
expected = textwrap.dedent("""\
To: nobody
Content-Disposition:
attachment;
filename*0*=us-ascii''afilenamelongenoghtowraphere
None
""")
s = self.ioclass()
g = self.genclass(s, policy=self.policy.clone(max_line_length=20))
g.flatten(msg)
self.assertEqual(s.getvalue(), self.typ(expected))
def test_keep_encoded_newlines(self):
msg = self.msgmaker(self.typ(textwrap.dedent("""\
To: nobody
Subject: Bad subject=?UTF-8?Q?=0A?=Bcc: injection@example.com
None
""")))
expected = textwrap.dedent("""\
To: nobody
Subject: Bad subject=?UTF-8?Q?=0A?=Bcc: injection@example.com
None
""")
s = self.ioclass()
g = self.genclass(s, policy=self.policy.clone(max_line_length=80))
g.flatten(msg)
self.assertEqual(s.getvalue(), self.typ(expected))
def test_keep_long_encoded_newlines(self):
msg = self.msgmaker(self.typ(textwrap.dedent("""\
To: nobody
Subject: Bad subject=?UTF-8?Q?=0A?=Bcc: injection@example.com
None
""")))
expected = textwrap.dedent("""\
To: nobody
Subject: Bad subject
=?utf-8?q?=0A?=Bcc:
injection@example.com
None
""")
s = self.ioclass()
g = self.genclass(s, policy=self.policy.clone(max_line_length=30))
g.flatten(msg)
self.assertEqual(s.getvalue(), self.typ(expected))
class TestGenerator(TestGeneratorBase, TestEmailBase):
msgfunc = staticmethod(message_from_string)
genclass = Generator
ioclass = io.StringIO
typ = str
def test_verify_generated_headers(self):
"""gh-121650: by default the generator prevents header injection"""
class LiteralHeader(str):
name = 'Header'
def fold(self, **kwargs):
return self
for text in (
'Value\r\nBad Injection\r\n',
'NoNewLine'
):
with self.subTest(text=text):
message = message_from_string(
"Header: Value\r\n\r\nBody",
policy=self.policy,
)
del message['Header']
message['Header'] = LiteralHeader(text)
with self.assertRaises(email.errors.HeaderWriteError):
message.as_string()
class TestBytesGenerator(TestGeneratorBase, TestEmailBase):
msgfunc = staticmethod(message_from_bytes)
genclass = BytesGenerator
ioclass = io.BytesIO
typ = lambda self, x: x.encode('ascii')
def test_cte_type_7bit_handles_unknown_8bit(self):
source = ("Subject: Maintenant je vous présente mon "
"collègue\n\n").encode('utf-8')
expected = ('Subject: Maintenant je vous =?unknown-8bit?q?'
'pr=C3=A9sente_mon_coll=C3=A8gue?=\n\n').encode('ascii')
msg = message_from_bytes(source)
s = io.BytesIO()
g = BytesGenerator(s, policy=self.policy.clone(cte_type='7bit'))
g.flatten(msg)
self.assertEqual(s.getvalue(), expected)
def test_cte_type_7bit_transforms_8bit_cte(self):
source = textwrap.dedent("""\
From: foo@bar.com
To: Dinsdale
Subject: Nudge nudge, wink, wink
Mime-Version: 1.0
Content-Type: text/plain; charset="latin-1"
Content-Transfer-Encoding: 8bit
oh là là, know what I mean, know what I mean?
""").encode('latin1')
msg = message_from_bytes(source)
expected = textwrap.dedent("""\
From: foo@bar.com
To: Dinsdale
Subject: Nudge nudge, wink, wink
Mime-Version: 1.0
Content-Type: text/plain; charset="iso-8859-1"
Content-Transfer-Encoding: quoted-printable
oh l=E0 l=E0, know what I mean, know what I mean?
""").encode('ascii')
s = io.BytesIO()
g = BytesGenerator(s, policy=self.policy.clone(cte_type='7bit',
linesep='\n'))
g.flatten(msg)
self.assertEqual(s.getvalue(), expected)
def test_smtputf8_policy(self):
msg = EmailMessage()
msg['From'] = "Páolo <főo@bar.com>"
msg['To'] = 'Dinsdale'
msg['Subject'] = 'Nudge nudge, wink, wink \u1F609'
msg.set_content("oh là là, know what I mean, know what I mean?")
expected = textwrap.dedent("""\
From: Páolo <főo@bar.com>
To: Dinsdale
Subject: Nudge nudge, wink, wink \u1F609
Content-Type: text/plain; charset="utf-8"
Content-Transfer-Encoding: 8bit
MIME-Version: 1.0
oh là là, know what I mean, know what I mean?
""").encode('utf-8').replace(b'\n', b'\r\n')
s = io.BytesIO()
g = BytesGenerator(s, policy=policy.SMTPUTF8)
g.flatten(msg)
self.assertEqual(s.getvalue(), expected)
def test_smtp_policy(self):
msg = EmailMessage()
msg["From"] = Address(addr_spec="foo@bar.com", display_name="Páolo")
msg["To"] = Address(addr_spec="bar@foo.com", display_name="Dinsdale")
msg["Subject"] = "Nudge nudge, wink, wink"
msg.set_content("oh boy, know what I mean, know what I mean?")
expected = textwrap.dedent("""\
From: =?utf-8?q?P=C3=A1olo?= <foo@bar.com>
To: Dinsdale <bar@foo.com>
Subject: Nudge nudge, wink, wink
Content-Type: text/plain; charset="utf-8"
Content-Transfer-Encoding: 7bit
MIME-Version: 1.0
oh boy, know what I mean, know what I mean?
""").encode().replace(b"\n", b"\r\n")
s = io.BytesIO()
g = BytesGenerator(s, policy=policy.SMTP)
g.flatten(msg)
self.assertEqual(s.getvalue(), expected)
if __name__ == '__main__':
unittest.main()
|