New attempt at getting secure SMTP working

quick_email/__init__.py

@@ -1,6 +1,6 @@
 from quick_email.builder import build_msg, Attachment
 from quick_email.sender import send_msg
 
-def send_email(host, port, send_from, subject, send_to=None, send_cc=None, send_bcc=None, plain_text=None, html_text=None, attachment_list=None, inline_attachment_dict=None, username=None, password=None, ssl=False):
+def send_email(host, port, send_from, subject, send_to=None, send_cc=None, send_bcc=None, plain_text=None, html_text=None, attachment_list=None, inline_attachment_dict=None, username=None, password=None, require_starttls=False):
     msg = build_msg(send_from, subject, send_to=send_to, send_cc=send_cc, send_bcc=send_bcc, plain_text=plain_text, html_text=html_text, attachment_list=attachment_list, inline_attachment_dict=inline_attachment_dict)
-    send_msg(msg, host, port, username=username, password=password, ssl=ssl)
+    send_msg(msg, host, port, username=username, password=password, require_starttls=require_starttls)

quick_email/sender.py

@@ -2,12 +2,11 @@
 
 from email.utils import COMMASPACE
 
-def send_msg(msg, host, port, username=None, password=None, ssl=False):
-    smtp = None
-    if not ssl:
-        smtp = smtplib.SMTP(host, port)
-    else:
-        smtp = smtplib.SMTP_SSL(host, port)
+def send_msg(msg, host, port, username=None, password=None, require_starttls=False):
+    smtp = smtplib.SMTP(host, port)
+
+    if require_starttls:
+        smtp.starttls()
 
     if username and password:
         smtp.login(username, password)

tests/smtp/aiosmtpd_fake.py

@@ -0,0 +1,61 @@
+import os
+import time
+import ssl
+import asyncio
+from six.moves import _thread
+
+from aiosmtpd.smtp import SMTP as Server
+from aiosmtpd.controller import Controller
+from aiosmtpd.handlers import Sink as Handler
+
+SMTP_PORT = int(os.environ.get(u'SMTP_PORT', u'8080'))
+
+
+class MyController(Controller):
+    def __init__(self, *args, **kwargs):
+        Controller.__init__(self, *args, **kwargs)
+        self._ssl_context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
+
+    def factory(self):
+        return MyServer(self.handler, tls_context=self._ssl_context)
+
+
+class MyServer(Server):
+    async def smtp_AUTH(self, arg):
+        if arg != 'PLAIN':
+            await self.push('501 Syntax: AUTH PLAIN')
+            return
+
+        await self.push('334')
+
+        second_line = await self._reader.readline()
+
+        try:
+            second_line = second_line.rstrip(b'\r\n').decode('ascii')
+        except UnicodeDecodeError:
+            await self.push('500 Error: Challenge must be ASCII')
+            return
+
+        if second_line != '':
+            self.authenticated = True
+            await self.push('235 Authentication successful')
+        else:
+            await self.push('535 Invalid credentials')
+
+
+async def _run():
+    controller = MyController(Handler(), hostname=u'localhost', port=SMTP_PORT)
+    controller.start()
+
+
+def _smtp_server_func():
+    loop = asyncio.new_event_loop()
+    loop.create_task(_run())
+    loop.run_forever()
+
+_smtp_server_thead = None
+def smtp_server_start():
+    global _smtp_server_thead
+    if _smtp_server_thead is None:
+        _smtp_server_thead = _thread.start_new_thread(_smtp_server_func, ())
+        time.sleep(1.0)

tests/smtpd_fake.py → tests/smtp/smtpd_fake.py

@@ -21,8 +21,12 @@ def _smtp_server_func():
 
 
 _smtp_server_thead = None
-def smtp_server_thread():
+def smtp_server_start():
     global _smtp_server_thead
     if _smtp_server_thead is None:
         _smtp_server_thead = _thread.start_new_thread(_smtp_server_func, ())
         time.sleep(1.0)
+
+
+def secure_smtp_server_start():
+    raise NotImplementedError(u'not available in this environment')

tests/test_quick_email.py

@@ -1,34 +1,31 @@
 import unittest
 
 from quick_email import send_email
-import smtpd_fake
 
 try:
-    import aiosmtpd
-    del aiosmtpd
+    import tests.smtp.aiosmtpd_fake as smtp
 
     _can_run_ssl_tests = True
     _can_run_auth_tests = True
-except ImportError:
+except (ImportError, SyntaxError):
+    import tests.smtp.smtpd_fake as smtp
     _can_run_ssl_tests = False
     _can_run_auth_tests = False
 
 class TestQuickEmail(unittest.TestCase):
     def test_send_email(self):
-        smtpd_fake.smtp_server_thread()
-        send_email(u'localhost', smtpd_fake.SMTP_PORT, u'Example <example@example.com>', u'The Subject', send_to=u'Test <test@test.com>', send_cc=None, send_bcc=None, plain_text=u'Some Text', html_text=u'<b>Some Bold Text</b>', attachment_list=None, inline_attachment_dict=None)
+        smtp.smtp_server_start()
+        send_email(u'localhost', smtp.SMTP_PORT, u'Example <example@example.com>', u'The Subject', send_to=u'Test <test@test.com>', plain_text=u'Some Text', html_text=u'<b>Some Bold Text</b>')
 
     @unittest.skipUnless(_can_run_ssl_tests, 'SSL tests unsupported')
-    def test_send_email_ssl(self):
-        pass
-
-    @unittest.skipUnless(_can_run_auth_tests, 'Auth tests unsupported')
-    def test_send_email_auth(self):
-        pass
+    def test_send_email_starttls(self):
+        smtp.smtp_server_start()
+        send_email(u'localhost', smtp.SMTP_PORT, u'Example <example@example.com>', u'The Subject', send_to=u'Test <test@test.com>', plain_text=u'Some Text', html_text=u'<b>Some Bold Text</b>', require_starttls=True)
 
     @unittest.skipUnless(_can_run_ssl_tests and _can_run_auth_tests, 'SSL or Auth tests unsupported')
-    def test_send_email_ssl_auth(self):
-        pass
+    def test_send_email_starttls_auth(self):
+        smtp.smtp_server_start()
+        send_email(u'localhost', smtp.SMTP_PORT, u'Example <example@example.com>', u'The Subject', send_to=u'Test <test@test.com>', plain_text=u'Some Text', html_text=u'<b>Some Bold Text</b>', username=u'testuser', password=u'password', require_starttls=True)
 
 if __name__ == u'__main__':
     unittest.main()

tests/test_sender.py

@@ -1,19 +1,28 @@
 import unittest
 
 from quick_email import builder, sender
-import smtpd_fake
+
+try:
+    import tests.smtp.aiosmtpd_fake as smtp
+
+    _can_run_ssl_tests = True
+    _can_run_auth_tests = True
+except (ImportError, SyntaxError):
+    import tests.smtp.smtpd_fake as smtp
+    _can_run_ssl_tests = False
+    _can_run_auth_tests = False
 
 
 class TestSender(unittest.TestCase):
     def test_minimal(self):
-        smtpd_fake.smtp_server_thread()
+        smtp.smtp_server_start()
         msg = builder.build_msg(u'Example <example@example.com>', u'The Subject', send_to=u'Test <test@test.com>', plain_text=u'Some Text', html_text=u'<b>Some Bold Text</b>')
-        sender.send_msg(msg, u'localhost', smtpd_fake.SMTP_PORT)
+        sender.send_msg(msg, u'localhost', smtp.SMTP_PORT)
 
     def test_multiple_recipients(self):
-        smtpd_fake.smtp_server_thread()
+        smtp.smtp_server_start()
         msg = builder.build_msg(u'Example <example@example.com>', u'The Subject', send_to=u'Test <test@test.com>', send_cc=[u'Example <example@example.com>', u'Example2 <example2@example.com>'], send_bcc=u'Example3 <example3@example.com>', plain_text=u'Some Text', html_text=u'<b>Some Bold Text</b>')
-        sender.send_msg(msg, u'localhost', smtpd_fake.SMTP_PORT)
+        sender.send_msg(msg, u'localhost', smtp.SMTP_PORT)
 
 
 if __name__ == u'__main__':