Skip to content

HTTPS clone URL

Subversion checkout URL

You can clone with
or
.
Download ZIP

Loading…

Message encoding support #3

Merged
merged 8 commits into from

2 participants

@rpatterson
Owner

Needed for the latest push to my fork of pyramid_mailer as well.

@mcdonc mcdonc merged commit 60ae523 into repoze:master
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Commits on Mar 14, 2012
  1. @rpatterson
  2. @rpatterson
  3. @rpatterson

    Add a helper function for predictable and sane encoding of message he…

    rpatterson authored
    …aders.
    
    This is the result of long conversations about the RFC's, the `email`
    package and various bugs and quirks in both.
Commits on Mar 15, 2012
  1. @rpatterson
  2. @rpatterson
  3. @rpatterson
  4. @rpatterson

    Tolerate some Python 2.7 and PyPy output differences.

    rpatterson authored
    For some reason, under Python 2.6 and 3.2, the encoding is included in
    the RFC 2047 prefix as 'utf_8' but under 2.7 and 3.2 it is 'utf-8'
    with a dash instead of an underscore.
  5. @rpatterson
This page is out of date. Refresh to see the latest.
View
6 CHANGES.txt
@@ -4,6 +4,12 @@ Change history
2.4 (Unreleased)
----------------
+- Provide improved support for encoding messages to bytes. It should
+ now be possible to represent your messages in
+ `email.message.Message` objects just with unicode (excepting bytes
+ for binary attachments) and the mailer will handler it as
+ appropriate.
+
- cPython 2.6, 2.7, 3.2 and pypy 1.8 compatibility.
2.3 (2011-05-17)
View
95 repoze/sendmail/encoding.py
@@ -0,0 +1,95 @@
+BBB_PY_2 = True
+try:
+ str = unicode
+except NameError:
+ BBB_PY_2 = False
+
+from email import utils
+from email import charset
+
+# From http://tools.ietf.org/html/rfc5322#section-3.6
+ADDR_HEADERS = ('resent-from',
+ 'resent-sender',
+ 'resent-to',
+ 'resent-cc',
+ 'resent-bcc',
+ 'from',
+ 'sender',
+ 'reply-to',
+ 'to',
+ 'cc',
+ 'bcc')
+
+PARAM_HEADERS = ('content-type',
+ 'content-disposition')
+
+
+def encode_message(message,
+ addr_headers=ADDR_HEADERS, param_headers=PARAM_HEADERS):
+ """
+ Encode a `Message` handling headers and payloads.
+
+ Headers are handled in the most sane way possible. Address names
+ are left in `ascii` if possible or encoded to `latin_1` or `utf-8`
+ and finally encoded according to RFC 2047 without encoding the
+ address, something the `email` stdlib package doesn't do.
+ Parameterized headers such as `filename` in the
+ `Content-Disposition` header, have their values encoded properly
+ while leaving the rest of the header to be handled without
+ encoding. Finally, all other header are left in `ascii` if
+ possible or encoded to `latin_1` or `utf-8` as a whole.
+
+ The return is a bytest string of the whole message.
+ """
+ for key, value in message.items():
+ if key.lower() in addr_headers:
+ addrs = []
+ for name, addr in utils.getaddresses([value]):
+ best, encoded = best_charset(name)
+ if BBB_PY_2:
+ name = encoded
+ name = charset.Charset(best).header_encode(name)
+ addrs.append(utils.formataddr((name, addr)))
+ value = ', '.join(addrs)
+ message.replace_header(key, value)
+ if key.lower() in param_headers:
+ for param_key, param_value in message.get_params(header=key):
+ if param_value:
+ best, encoded = best_charset(param_value)
+ if BBB_PY_2:
+ param_value = encoded
+ if best == 'ascii':
+ best = None
+ message.set_param(param_key, param_value,
+ header=key, charset=best)
+ else:
+ best, encoded = best_charset(value)
+ if BBB_PY_2:
+ value = encoded
+ value = charset.Charset(best).header_encode(value)
+ message.replace_header(key, value)
+
+ payload = message.get_payload()
+ if payload and isinstance(payload, str):
+ best, encoded = best_charset(payload)
+ if BBB_PY_2:
+ payload = encoded
+ message.set_payload(payload, charset=best)
+
+ return message.as_string().encode('ascii')
+
+
+def best_charset(text):
+ """
+ Find the most human-readable and/or conventional encoding for unicode text.
+
+ Prefers `ascii` or `latin_1` and falls back to `utf_8`.
+ """
+ encoded = text
+ for charset in 'ascii', 'latin_1', 'utf_8':
+ try:
+ encoded = text.encode(charset)
+ except UnicodeError:
+ pass
+ else:
+ return charset, encoded
View
10 repoze/sendmail/interfaces.py
@@ -64,8 +64,8 @@ def send(fromaddr, toaddrs, message):
`toaddrs` is a sequence of recipient addresses (byte strings).
- `message` is a byte string that contains both headers and body
- formatted according to RFC 2822. If it does not contain a Message-Id
+ `message` is a `Message` object from the stdlib
+ `email.message` module. If it does not contain a Message-Id
header, it will be generated and added automatically.
Returns the message ID.
@@ -87,9 +87,9 @@ def send(fromaddr, toaddrs, message):
`toaddrs` is a sequence of recipient addresses (unicode strings).
- `message` contains both headers and body formatted according to RFC
- 2822. It should contain at least Date, From, To, and Message-Id
- headers.
+ `message` is a `Message` object from the stdlib
+ `email.message` module. If it does not contain a Message-Id
+ header, it will be generated and added automatically.
Messages are sent immediatelly.
View
6 repoze/sendmail/mailer.py
@@ -20,6 +20,7 @@
from zope.interface import implementer
from repoze.sendmail.interfaces import IMailer
+from repoze.sendmail import encoding
have_ssl = hasattr(socket, 'ssl')
@@ -44,8 +45,9 @@ def smtp_factory(self):
return connection
def send(self, fromaddr, toaddrs, message):
- if isinstance(message, Message):
- message = message.as_string()
+ assert isinstance(message, Message), \
+ 'Message must be instance of email.message.Message'
+ message = encoding.encode_message(message)
connection = self.smtp_factory()
View
4 repoze/sendmail/tests/test_delivery.py
@@ -83,7 +83,7 @@ def testSend(self):
mailer.sent_messages = []
msgid = delivery.send(fromaddr, toaddrs, message)
- self.assert_('@' in msgid)
+ self.assertTrue('@' in msgid)
self.assertEquals(mailer.sent_messages, [])
transaction.commit()
self.assertEquals(len(mailer.sent_messages), 1)
@@ -184,7 +184,7 @@ def testSend(self):
MaildirMessageStub.commited_messages = []
msgid = delivery.send(fromaddr, toaddrs, message)
- self.assert_('@' in msgid)
+ self.assertTrue('@' in msgid)
self.assertEquals(MaildirMessageStub.commited_messages, [])
self.assertEquals(MaildirMessageStub.aborted_messages, [])
transaction.commit()
View
182 repoze/sendmail/tests/test_encoding.py
@@ -0,0 +1,182 @@
+##############################################################################
+#
+# Copyright (c) 2003 Zope Corporation and Contributors.
+# All Rights Reserved.
+#
+# This software is subject to the provisions of the Zope Public License,
+# Version 2.1 (ZPL). A copy of the ZPL should accompany this distribution.
+# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
+# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
+# FOR A PARTICULAR PURPOSE.
+#
+##############################################################################
+
+import unittest
+import base64
+import quopri
+from email import message
+from email.mime import multipart
+from email.mime import application
+
+try:
+ from urllib.parse import quote
+except ImportError:
+ # BBB Python 2 and 3 compat
+ from urllib import quote
+
+
+class TestEncoding(unittest.TestCase):
+
+ def setUp(self):
+ self.message = message.Message()
+ self.latin_1_encoded = b'LaPe\xf1a'
+ self.latin_1 = self.latin_1_encoded.decode('latin_1')
+ self.utf_8_encoded = b'mo \xe2\x82\xac'
+ self.utf_8 = self.utf_8_encoded.decode('utf_8')
+
+ def encode(self, message=None):
+ if message is None:
+ message = self.message
+ from repoze.sendmail import encoding
+ return encoding.encode_message(message)
+
+ def test_best_charset_ascii(self):
+ from repoze.sendmail import encoding
+ value = 'foo'
+ best, encoded = encoding.best_charset(value)
+ self.assertEqual(encoded, b'foo')
+ self.assertEqual(best, 'ascii')
+
+ def test_best_charset_latin_1(self):
+ from repoze.sendmail import encoding
+ value = self.latin_1
+ best, encoded = encoding.best_charset(value)
+ self.assertEqual(encoded, self.latin_1_encoded)
+ self.assertEqual(best, 'latin_1')
+
+ def test_best_charset_utf_8(self):
+ from repoze.sendmail import encoding
+ value = self.utf_8
+ best, encoded = encoding.best_charset(value)
+ self.assertEqual(encoded, self.utf_8_encoded)
+ self.assertEqual(best, 'utf_8')
+
+ def test_encoding_ascii_headers(self):
+ to = ', '.join(['Chris McDonough <chrism@example.com>',
+ '"Chris Rossi, M.D." <chrisr@example.com>'])
+ self.message['To'] = to
+ from_ = 'Ross Patterson <rpatterson@example.com>'
+ self.message['From'] = from_
+ subject = 'I know what you did last PyCon'
+ self.message['Subject'] = subject
+
+ encoded = self.encode()
+
+ self.assertTrue(
+ b'To: Chris McDonough <chrism@example.com>, "Chris Rossi,'
+ in encoded)
+ self.assertTrue(b'From: '+from_.encode('ascii') in encoded)
+ self.assertTrue(b'Subject: '+subject.encode('ascii') in encoded)
+
+ def test_encoding_latin_1_headers(self):
+ to = ', '.join([
+ '"'+self.latin_1+' McDonough, M.D." <chrism@example.com>',
+ 'Chris Rossi <chrisr@example.com>'])
+ self.message['To'] = to
+ from_ = self.latin_1+' Patterson <rpatterson@example.com>'
+ self.message['From'] = from_
+ subject = 'I know what you did last '+self.latin_1
+ self.message['Subject'] = subject
+
+ encoded = self.encode()
+
+ self.assertTrue(b'To: =?iso-8859-1?' in encoded)
+ self.assertTrue(b'From: =?iso-8859-1?' in encoded)
+ self.assertTrue(b'Subject: =?iso-8859-1?' in encoded)
+ self.assertTrue(b'<chrism@example.com>' in encoded)
+ self.assertTrue(b'<chrisr@example.com>' in encoded)
+ self.assertTrue(b'<rpatterson@example.com>' in encoded)
+
+ def test_encoding_utf_8_headers(self):
+ to = ', '.join([
+ '"'+self.utf_8+' McDonough, M.D." <chrism@example.com>',
+ 'Chris Rossi <chrisr@example.com>'])
+ self.message['To'] = to
+ from_ = self.utf_8+' Patterson <rpatterson@example.com>'
+ self.message['From'] = from_
+ subject = 'I know what you did last '+self.utf_8
+ self.message['Subject'] = subject
+
+ encoded = self.encode()
+
+ self.assertTrue(b'To: =?utf' in encoded)
+ self.assertTrue(b'From: =?utf' in encoded)
+ self.assertTrue(b'Subject: =?utf' in encoded)
+ self.assertTrue(b'<chrism@example.com>' in encoded)
+ self.assertTrue(b'<chrisr@example.com>' in encoded)
+ self.assertTrue(b'<rpatterson@example.com>' in encoded)
+
+ def test_encoding_ascii_header_parameters(self):
+ self.message['Content-Disposition'] = (
+ 'attachment; filename=foo.ppt')
+
+ encoded = self.encode()
+
+ self.assertTrue(
+ b'Content-Disposition: attachment; filename="foo.ppt"' in encoded)
+
+ def test_encoding_latin_1_header_parameters(self):
+ self.message['Content-Disposition'] = (
+ 'attachment; filename='+self.latin_1+'.ppt')
+
+ encoded = self.encode()
+
+ self.assertTrue(
+ b"Content-Disposition: attachment; filename*=" in encoded)
+ self.assertTrue(b"latin_1''"+quote(
+ self.latin_1_encoded).encode('ascii') in encoded)
+
+ def test_encoding_utf_8_header_parameters(self):
+ self.message['Content-Disposition'] = (
+ 'attachment; filename='+self.utf_8+'.ppt')
+
+ encoded = self.encode()
+
+ self.assertTrue(
+ b"Content-Disposition: attachment; filename*=" in encoded)
+ self.assertTrue(b"utf_8''"+quote(self.utf_8_encoded).encode('ascii')
+ in encoded)
+
+ def test_encoding_ascii_body(self):
+ body = 'I know what you did last PyCon'
+ self.message.set_payload(body)
+
+ encoded = self.encode()
+
+ self.assertTrue(body.encode('ascii') in encoded)
+
+ def test_encoding_latin_1_body(self):
+ body = 'I know what you did last '+self.latin_1
+ self.message.set_payload(body)
+
+ encoded = self.encode()
+
+ self.assertTrue(quopri.encodestring(body.encode('latin_1')) in encoded)
+
+ def test_encoding_utf_8_body(self):
+ body = 'I know what you did last '+self.utf_8
+ self.message.set_payload(body)
+
+ encoded = self.encode()
+
+ self.assertTrue(base64.encodestring(body.encode('utf_8')) in encoded)
+
+ def test_binary_body(self):
+ body = b'I know what you did last PyCon'
+ self.message = multipart.MIMEMultipart()
+ self.message.attach(application.MIMEApplication(body))
+
+ encoded = self.encode()
+
+ self.assertTrue(base64.encodestring(body) in encoded)
View
6 repoze/sendmail/tests/test_maildir.py
@@ -227,7 +227,7 @@ def test_add(self):
from repoze.sendmail.maildir import Maildir
m = Maildir('/path/to/maildir')
tx_message = m.add(Message())
- self.assert_(tx_message._pending_path,
+ self.assertTrue(tx_message._pending_path,
'/path/to/maildir/tmp/1234500002.4242.myhostname.')
def test_add_no_good_filenames(self):
@@ -254,7 +254,7 @@ def test_tx_msg_abort(self):
tx_msg.abort()
self.assertEquals(tx_msg._aborted, True)
self.assertEquals(tx_msg._committed, False)
- self.assert_(filename1 in self.fake_os_module._removed_files)
+ self.assertTrue(filename1 in self.fake_os_module._removed_files)
tx_msg.abort()
self.assertRaises(RuntimeError, tx_msg.commit)
@@ -269,7 +269,7 @@ def test_tx_msg_commit(self):
tx_msg.commit()
self.assertEquals(tx_msg._aborted, False)
self.assertEquals(tx_msg._committed, True)
- self.assert_((filename1, filename2)
+ self.assertTrue((filename1, filename2)
in self.fake_os_module._renamed_files)
self.assertRaises(RuntimeError, tx_msg.abort)
View
56 repoze/sendmail/tests/test_mailer.py
@@ -14,6 +14,7 @@
from zope.interface.verify import verifyObject
from repoze.sendmail.mailer import SMTPMailer
+import email
import ssl
import unittest
@@ -86,11 +87,12 @@ def test_send(self):
msg['Headers'] = 'headers'
msg.set_payload('bodybodybody\n-- \nsig\n')
self.mailer.send(fromaddr, toaddrs, msg)
- self.assertEquals(self.smtp.fromaddr, fromaddr)
- self.assertEquals(self.smtp.toaddrs, toaddrs)
- self.assertEquals(self.smtp.msgtext, msg.as_string())
- self.assert_(self.smtp.quitted)
- self.assert_(self.smtp.closed)
+ self.assertEqual(self.smtp.fromaddr, fromaddr)
+ self.assertEqual(self.smtp.toaddrs, toaddrs)
+ self.assertEqual(
+ self.smtp.msgtext, msg.as_string().encode('ascii'))
+ self.assertTrue(self.smtp.quitted)
+ self.assertTrue(self.smtp.closed)
def test_fail_ehlo(self):
from email.message import Message
@@ -114,34 +116,42 @@ def test_tls_required_not_available(self):
def test_send_auth(self):
fromaddr = 'me@example.com'
toaddrs = ('you@example.com', 'him@example.com')
- msgtext = 'Headers: headers\n\nbodybodybody\n-- \nsig\n'
+ headers = 'Headers: headers'
+ body='bodybodybody\n-- \nsig\n'
+ msgtext = headers+'\n\n'+body
+ msg = email.message_from_string(msgtext)
self.mailer.username = 'foo'
self.mailer.password = 'evil'
self.mailer.hostname = 'spamrelay'
self.mailer.port = 31337
- self.mailer.send(fromaddr, toaddrs, msgtext)
- self.assertEquals(self.smtp.username, 'foo')
- self.assertEquals(self.smtp.password, 'evil')
- self.assertEquals(self.smtp.hostname, 'spamrelay')
- self.assertEquals(self.smtp.port, '31337')
- self.assertEquals(self.smtp.fromaddr, fromaddr)
- self.assertEquals(self.smtp.toaddrs, toaddrs)
- self.assertEquals(self.smtp.msgtext, msgtext)
- self.assert_(self.smtp.quitted)
- self.assert_(self.smtp.closed)
+ self.mailer.send(fromaddr, toaddrs, msg)
+ self.assertEqual(self.smtp.username, 'foo')
+ self.assertEqual(self.smtp.password, 'evil')
+ self.assertEqual(self.smtp.hostname, 'spamrelay')
+ self.assertEqual(self.smtp.port, '31337')
+ self.assertEqual(self.smtp.fromaddr, fromaddr)
+ self.assertEqual(self.smtp.toaddrs, toaddrs)
+ self.assertTrue(body.encode('ascii') in self.smtp.msgtext)
+ self.assertTrue(headers.encode('ascii') in self.smtp.msgtext)
+ self.assertTrue(self.smtp.quitted)
+ self.assertTrue(self.smtp.closed)
def test_send_failQuit(self):
self.mailer.smtp.fail_on_quit = True
try:
fromaddr = 'me@example.com'
toaddrs = ('you@example.com', 'him@example.com')
- msgtext = 'Headers: headers\n\nbodybodybody\n-- \nsig\n'
- self.mailer.send(fromaddr, toaddrs, msgtext)
- self.assertEquals(self.smtp.fromaddr, fromaddr)
- self.assertEquals(self.smtp.toaddrs, toaddrs)
- self.assertEquals(self.smtp.msgtext, msgtext)
- self.assert_(not self.smtp.quitted)
- self.assert_(self.smtp.closed)
+ headers = 'Headers: headers'
+ body='bodybodybody\n-- \nsig\n'
+ msgtext = headers+'\n\n'+body
+ msg = email.message_from_string(msgtext)
+ self.mailer.send(fromaddr, toaddrs, msg)
+ self.assertEqual(self.smtp.fromaddr, fromaddr)
+ self.assertEqual(self.smtp.toaddrs, toaddrs)
+ self.assertTrue(body.encode('ascii') in self.smtp.msgtext)
+ self.assertTrue(headers.encode('ascii') in self.smtp.msgtext)
+ self.assertTrue(not self.smtp.quitted)
+ self.assertTrue(self.smtp.closed)
finally:
self.mailer.smtp.fail_on_quit = False
Something went wrong with that request. Please try again.