rfc compliantish header check

flask_mail.py

@@ -110,6 +110,12 @@ def sanitize_addresses(addresses, encoding='utf-8'):
     return map(lambda e: sanitize_address(e, encoding), addresses)
 
 
+def _has_newline(line):
+    """Used by has_bad_header to check for \\r or \\n"""
+    if line and ('\r' in line or '\n' in line):
+        return True
+    return False
+
 class Connection(object):
     """Handles connection to host."""
 
@@ -354,13 +360,25 @@ def __str__(self):
 
     def has_bad_headers(self):
         """Checks for bad headers i.e. newlines in subject, sender or recipients.
+        RFC5322: Allows multiline CRLF with trailing whitespace (FWS) in headers
         """
 
-        reply_to = self.reply_to or ''
-        for val in [self.subject, self.sender, reply_to] + self.recipients:
-            for c in '\r\n':
-                if c in val:
-                    return True
+        headers = [self.sender, self.reply_to] + self.recipients
+        for header in headers:
+            if _has_newline(header):
+                return True
+
+        if self.subject:
+            if _has_newline(self.subject):
+                for linenum, line in enumerate(self.subject.split('\r\n')):
+                    if not line:
+                        return True
+                    if linenum > 0 and line[0] not in '\t ':
+                        return True
+                    if _has_newline(line):
+                        return True
+                    if len(line.strip()) == 0:
+                        return True
         return False
 
     def is_bad_headers(self):

tests.py

@@ -153,15 +153,44 @@ def test_attach(self):
         self.assertEqual(a.data, b"this is a test")
 
     def test_bad_header_subject(self):
-        msg = Message(subject="testing\n\r",
+        msg = Message(subject="testing\r\n",
+                      sender="from@example.com",
+                      body="testing",
+                      recipients=["to@example.com"])
+        self.assertRaises(BadHeaderError, self.mail.send, msg)
+
+    def test_multiline_subject(self):
+        msg = Message(subject="testing\r\n testing\r\n testing \r\n \ttesting",
+                      sender="from@example.com",
+                      body="testing",
+                      recipients=["to@example.com"])
+        self.mail.send(msg)
+        response = msg.as_string()
+        self.assertIn("From: from@example.com", str(response))
+        self.assertIn("testing\r\n testing\r\n testing \r\n \ttesting", str(response))
+
+    def test_bad_multiline_subject(self):
+        msg = Message(subject="testing\r\n testing\r\n ",
+                      sender="from@example.com",
+                      body="testing",
+                      recipients=["to@example.com"])
+        self.assertRaises(BadHeaderError, self.mail.send, msg)
+
+        msg = Message(subject="testing\r\n testing\r\n\t",
+                      sender="from@example.com",
+                      body="testing",
+                      recipients=["to@example.com"])
+        self.assertRaises(BadHeaderError, self.mail.send, msg)
+
+        msg = Message(subject="testing\r\n testing\r\n\n",
                       sender="from@example.com",
                       body="testing",
                       recipients=["to@example.com"])
         self.assertRaises(BadHeaderError, self.mail.send, msg)
 
     def test_bad_header_sender(self):
         msg = Message(subject="testing",
-                      sender="from@example.com\n\r",
+                      sender="from@example.com\r\n",
                       recipients=["to@example.com"],
                       body="testing")
 
@@ -170,7 +199,7 @@ def test_bad_header_sender(self):
     def test_bad_header_reply_to(self):
         msg = Message(subject="testing",
                       sender="from@example.com",
-                      reply_to="evil@example.com\n\r",
+                      reply_to="evil@example.com\r",
                       recipients=["to@example.com"],
                       body="testing")