Add XOAUTH2 authentication

mailmerge/__main__.py

@@ -248,6 +248,16 @@ def create_sample_input_files(template_path, database_path, config_path):
             # username = YOUR_USERNAME_HERE
             # ratelimit = 0
 
+            # Example: XOAUTH
+            # Enter your token at the password prompt.  For Microsoft OAuth
+            # authentication, a token can be obtained with the oauth2ms tool
+            # https://github.com/harishkrupo/oauth2ms
+            # [smtp_server]
+            # host = smtp.office365.com
+            # port = 587
+            # security = XOAUTH
+            # username = username@example.com
+
             # Example: No security
             # [smtp_server]
             # host = newman.eecs.umich.edu

mailmerge/sendmail_client.py

@@ -9,9 +9,9 @@
 import configparser
 import getpass
 import datetime
+import base64
 from . import exceptions
 
-
 # Type to store info read from config file
 MailmergeConfig = collections.namedtuple(
     "MailmergeConfig",
@@ -49,7 +49,7 @@ def read_config(self):
             security = None
 
         # Verify security type
-        if security not in [None, "SSL/TLS", "STARTTLS", "PLAIN"]:
+        if security not in [None, "SSL/TLS", "STARTTLS", "PLAIN", "XOAUTH"]:
             raise exceptions.MailmergeError(
                 f"{self.config_path}: unrecognized security type: '{security}'"
             )
@@ -86,27 +86,18 @@ def sendmail(self, sender, recipients, message):
             )
 
         # Send
+        host, port = self.config.host, self.config.port
         try:
-            message_flattened = str(message)
-            host, port = self.config.host, self.config.port
             if self.config.security == "SSL/TLS":
-                with smtplib.SMTP_SSL(host, port) as smtp:
-                    smtp.login(self.config.username, self.password)
-                    smtp.sendmail(sender, recipients, message_flattened)
+                self.sendmail_ssltls(sender, recipients, message)
             elif self.config.security == "STARTTLS":
-                with smtplib.SMTP(host, port) as smtp:
-                    smtp.ehlo()
-                    smtp.starttls()
-                    smtp.ehlo()
-                    smtp.login(self.config.username, self.password)
-                    smtp.sendmail(sender, recipients, message_flattened)
+                self.sendmail_starttls(sender, recipients, message)
             elif self.config.security == "PLAIN":
-                with smtplib.SMTP(host, port) as smtp:
-                    smtp.login(self.config.username, self.password)
-                    smtp.sendmail(sender, recipients, message_flattened)
+                self.sendmail_plain(sender, recipients, message)
+            elif self.config.security == "XOAUTH":
+                self.sendmail_xoauth(sender, recipients, message)
             elif self.config.security is None:
-                with smtplib.SMTP(host, port) as smtp:
-                    smtp.sendmail(sender, recipients, message_flattened)
+                self.sendmail_clear(sender, recipients, message)
         except smtplib.SMTPAuthenticationError as err:
             raise exceptions.MailmergeError(
                 f"{host}:{port} failed to authenticate "
@@ -123,3 +114,55 @@ def sendmail(self, sender, recipients, message):
 
         # Update timestamp of last sent message
         self.lastsent = now
+
+    def sendmail_ssltls(self, sender, recipients, message):
+        """Send email message with SSL/TLS security."""
+        message_flattened = str(message)
+        with smtplib.SMTP_SSL(self.config.host, self.config.port) as smtp:
+            smtp.login(self.config.username, self.password)
+            smtp.sendmail(sender, recipients, message_flattened)
+
+    def sendmail_starttls(self, sender, recipients, message):
+        """Send email message with STARTTLS security."""
+        message_flattened = str(message)
+        with smtplib.SMTP(self.config.host, self.config.port) as smtp:
+            smtp.ehlo()
+            smtp.starttls()
+            smtp.ehlo()
+            smtp.login(self.config.username, self.password)
+            smtp.sendmail(sender, recipients, message_flattened)
+
+    def sendmail_plain(self, sender, recipients, message):
+        """Send email message with plain security."""
+        message_flattened = str(message)
+        with smtplib.SMTP(self.config.host, self.config.port) as smtp:
+            smtp.login(self.config.username, self.password)
+            smtp.sendmail(sender, recipients, message_flattened)
+
+    def sendmail_clear(self, sender, recipients, message):
+        """Send email message with no security."""
+        message_flattened = str(message)
+        with smtplib.SMTP(self.config.host, self.config.port) as smtp:
+            smtp.sendmail(sender, recipients, message_flattened)
+
+    def sendmail_xoauth(self, sender, recipients, message):
+        """Send email message with XOAUTH security."""
+        xoauth2 = (
+            f"user={self.config.username}\x01"
+            f"auth=Bearer {self.password}\x01\x01"
+        )
+        try:
+            xoauth2 = xoauth2.encode("ascii")
+        except UnicodeEncodeError as err:
+            raise exceptions.MailmergeError(
+                f"Username and XOAUTH access token must be ASCII '{xoauth2}'. "
+                f"{err}, "
+            )
+        message_flattened = str(message)
+        with smtplib.SMTP(self.config.host, self.config.port) as smtp:
+            smtp.ehlo()
+            smtp.starttls()
+            smtp.ehlo()
+            smtp.docmd('AUTH XOAUTH2')
+            smtp.docmd(str(base64.b64encode(xoauth2).decode("utf-8")))
+            smtp.sendmail(sender, recipients, message_flattened)

tests/test_sendmail_client.py

@@ -8,6 +8,7 @@
 import smtplib
 import email
 import email.parser
+import base64
 import pytest
 from mailmerge import SendmailClient, MailmergeError
 
@@ -250,6 +251,90 @@ def test_security_starttls(mocker, tmp_path):
     assert smtp.sendmail.call_count == 1
 
 
+def test_security_xoauth(mocker, tmp_path):
+    """Verify XOAUTH security configuration."""
+    # Config for XOAUTH SMTP server
+    config_path = tmp_path/"server.conf"
+    config_path.write_text(textwrap.dedent("""\
+        [smtp_server]
+        host = smtp.office365.com
+        port = 587
+        security = XOAUTH
+        username = username@example.com
+    """))
+
+    # Simple template
+    sendmail_client = SendmailClient(config_path, dry_run=False)
+    message = email.message_from_string("Hello world")
+
+    # Mock SMTP
+    mock_smtp = mocker.patch('smtplib.SMTP')
+    mock_smtp_ssl = mocker.patch('smtplib.SMTP_SSL')
+
+    # Mock the password entry
+    mock_getpass = mocker.patch('getpass.getpass')
+    mock_getpass.return_value = "password"
+
+    # Send a message
+    sendmail_client.sendmail(
+        sender="test@test.com",
+        recipients=["test@test.com"],
+        message=message,
+    )
+
+    # Verify SMTP library calls
+    assert mock_getpass.call_count == 1
+    assert mock_smtp.call_count == 1
+    assert mock_smtp_ssl.call_count == 0
+    smtp = mock_smtp.return_value.__enter__.return_value
+    assert smtp.ehlo.call_count == 2
+    assert smtp.starttls.call_count == 1
+    assert smtp.login.call_count == 0
+    assert smtp.docmd.call_count == 2
+    assert smtp.sendmail.call_count == 1
+
+    # Verify authentication token format.  The first call to docmd() is always
+    # the same.  Second call to docmd() contains a base64 encoded username and
+    # password.
+    assert smtp.docmd.call_args_list[0].args[0] == "AUTH XOAUTH2"
+    user_pass = smtp.docmd.call_args_list[1].args[0]
+    user_pass = base64.b64decode(user_pass)
+    assert user_pass == \
+        b'user=username@example.com\x01auth=Bearer password\x01\x01'
+
+
+def test_security_xoauth_bad_username(mocker, tmp_path):
+    """Verify exception is thrown for UTF-8 username."""
+    # Config for XOAUTH SMTP server
+    config_path = tmp_path/"server.conf"
+    config_path.write_text(textwrap.dedent("""\
+        [smtp_server]
+        host = smtp.office365.com
+        port = 587
+        security = XOAUTH
+        username = Laȝamon@example.com
+    """))
+
+    # Simple template
+    sendmail_client = SendmailClient(config_path, dry_run=False)
+    message = email.message_from_string("Hello world")
+
+    # Mock the password entry
+    mock_getpass = mocker.patch('getpass.getpass')
+    mock_getpass.return_value = "password"
+
+    # Send a message
+    with pytest.raises(MailmergeError) as err:
+        sendmail_client.sendmail(
+            sender="test@test.com",
+            recipients=["test@test.com"],
+            message=message,
+        )
+
+    # Verify exception string
+    assert "Username and XOAUTH access token must be ASCII" in str(err.value)
+
+
 def test_security_plain(mocker, tmp_path):
     """Verify plain security configuration."""
     # Config for Plain SMTP server