Adding the STARTTLS and AUTH extensions

quick_email/sender.py

@@ -8,8 +8,8 @@ def send_msg(msg, host, port, username=None, password=None, require_starttls=Fal
     if require_starttls:
         smtp.starttls()
 
-    if username and password:
-        smtp.login(username, password)
+        if username and password:
+            smtp.login(username, password)
 
     _all_recipients = all_recipients(msg)
 

tests/smtp/smtpd.py

@@ -3,7 +3,8 @@
 import socket
 import asyncore
 import asynchat
-
+import ssl
+import base64
 
 __version__ = u'Python SMTP proxy version 0.2 - edit'
 
@@ -176,11 +177,63 @@ def smtp_DATA(self, arg):
         self.set_terminator(b'\r\n.\r\n')
         self.push(b'354 End data with <CR><LF>.<CR><LF>')
 
+    # Extensions Start
+
+    def smtp_AUTH(self, arg):
+        auth = self.__server._auth
+        if isinstance(auth, dict) and u'user' in auth and u'password' in auth:
+            user = auth[u'user']
+            password = auth[u'password']
+
+            s = base64.stanard_b64encode(u'\0%s\0%s'.format(user, password).encode()).decode()
+
+            if arg == u'PLAIN {}'.format(s).encode():
+                self.push(b'235 Authentication successful')
+            else:
+                self.push(b'535 SMTP Authentication unsuccessful/Bad username or password')
+        else:
+            self.push(b'454 Temporary authentication failure')
+
+    def smtp_EHLO(self, arg):
+        if not arg:
+            self.push(b'501 Syntax: HELO hostname')
+        elif self.__greeting:
+            self.push(b'503 Duplicate HELO/EHLO')
+        else:
+            self.__greeting = arg
+            if isinstance(self.__conn, ssl.SSLSocket):
+                self.push(u'250 {}'.format(self.__fqdn).encode())
+            else:
+                self.push(u'250-{}'.format(self.__fqdn).encode())
+                self.push(b'250 STARTTLS')
+
+    def smtp_STARTTLS(self, arg):
+        if arg:
+            self.push(b'501 Syntax error (no parameters allowed)')
+        elif self.__server._starttls and not isinstance(self.__conn, ssl.SSLSocket):
+            self.push(b'220 Ready to start TLS')
+            self.__conn.settimeout(30)
+            self.__conn = self.__server._ssl_ctx.wrap_socket(self.__conn, server_side=True)
+            self.__conn.settimeout(None)
+            # re-init channel
+            asynchat.async_chat.__init__(self, self.__conn)
+            self.__line = []
+            self.__state = self.COMMAND
+            self.__greeting = 0
+            self.__mailfrom = None
+            self.__rcpttos = []
+            self.__data = ''
+        else:
+            self.push(b'454 TLS not available due to temporary reason')
+
 
 class SMTPServer(asyncore.dispatcher):
-    def __init__(self, localaddr, remoteaddr):
+    def __init__(self, localaddr, remoteaddr, ssl_ctx=None, starttls=True, auth=None):
         self._localaddr = localaddr
         self._remoteaddr = remoteaddr
+        self._ssl_ctx = ssl_ctx
+        self._starttls = starttls
+        self._auth = auth
         asyncore.dispatcher.__init__(self)
         try:
             self.create_socket(socket.AF_INET, socket.SOCK_STREAM)
@@ -197,6 +250,8 @@ def handle_accept(self):
         pair = self.accept()
         if pair is not None:
             conn, addr = pair
+            if self._ssl_ctx and not self._starttls:
+                conn = self._ssl_ctx.wrap_socket(conn, server_side=True)
             channel = SMTPChannel(self, conn, addr)
 
     def process_message(self, peer, mailfrom, rcpttos, data):

tests/smtp/smtpd_fake.py

@@ -10,15 +10,19 @@
 
 
 class FakeSMTPServer(SMTPServer):
-    def __init__(self, localaddr, remoteaddr):
-        SMTPServer.__init__(self, localaddr, remoteaddr)
+    def __init__(self, localaddr, remoteaddr, ssl_ctx=None, starttls=True, auth=None):
+        SMTPServer.__init__(self, localaddr, remoteaddr, ssl_ctx=ssl_ctx, starttls=starttls, auth=auth)
 
     def process_message(self, peer, mailfrom, rcpttos, data):
         pass
 
 
 def _smtp_server_func():
-    smtp_server = FakeSMTPServer((u'localhost', SMTP_PORT), None)
+    ssl_ctx = ssl.create_default_context()
+    smtp_server = FakeSMTPServer((u'localhost', SMTP_PORT), None, ssl_ctx=ssl_ctx, starttls=True, auth={
+            'user': 'testuser',
+            'password': 'password',
+        })
     asyncore.loop()
 
 

tests/test_quick_email.py

@@ -11,15 +11,10 @@ def setUpClass(cls):
     def test_send_email(self):
         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.skip('not working yet')
-    def test_send_email_auth(self):
-        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')
 
-    @unittest.skip('not working yet')
     def test_send_email_starttls(self):
         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.skip('not working yet')
     def test_send_email_starttls_auth(self):
         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)