Authenticate Outlook SMTP connection using OAuth

outlook-smtp-oauth.py

# Authenticate Outlook SMTP connection using OAuth
#
# 1. Create account at https://portal.azure.com/
#
# 2. Register New App to Microsoft Entra ID.
#    Select Native Application.
#    Select "Personal Microsoft accounts only".
#    Set redirect_uri.
#    It seems that No API permission required.
#
# https://learn.microsoft.com/ja-jp/exchange/client-developer/legacy-protocols/how-to-authenticate-an-imap-pop-smtp-application-by-using-oauth
# https://learn.microsoft.com/ja-jp/entra/identity-platform/v2-oauth2-auth-code-flow
# https://learn.microsoft.com/ja-jp/previous-versions/office/office-365-api/api/version-2.0/use-outlook-rest-api
#
# /// script
# dependencies = ["httpx"]
# ///

import http.server
import json
import logging
import smtplib
import threading
import urllib.parse
import webbrowser
from email.message import EmailMessage

import httpx

# FIXME: Set your app's client_id
client_id = "368dc748-3fde-45d4-8909-b3416f6f2bf9"

# FIXME: Set your app's redirect_uri
redirect_uri = "http://localhost:8000/auth"
redirect_host = "localhost"
redirect_port = 8000

# User.Read is used to get email address of authenticated user.
# offline_access is used to get refresh_token.
scope = "https://outlook.office.com/SMTP.Send https://outlook.office.com/User.Read offline_access"

smtp_host = "smtp.office365.com"
smtp_port = 587

logger = logging.getLogger()


class OauthAuthorizeRedirectHandler(http.server.SimpleHTTPRequestHandler):
    code = None

    def do_GET(self):
        pr = urllib.parse.urlparse(self.path)
        params = urllib.parse.parse_qs(pr.query)
        payload = b"\r\n".join(f"{k} = {v[0]}".encode("ascii") for k, v in params.items())
        self.send_response(200)
        self.send_header("Content-Length", len(payload))
        self.end_headers()
        self.wfile.write(payload)
        self.wfile.write(b"\r\n")

        OauthAuthorizeRedirectHandler.code = params["code"][0]


# Open authorize URL with web browser and receive result with local http server.
def oauth_request_authorization_code():
    url = "https://login.microsoftonline.com/consumers/oauth2/v2.0/authorize?" + urllib.parse.urlencode(
        {
            "client_id": client_id,
            "response_type": "code",
            "response_mode": "query",
            "redirect_uri": redirect_uri,
            "scope": scope,
            "state": "12345",
        }
    )

    t = threading.Timer(0.1, webbrowser.open, args=[url])
    t.start()

    with http.server.HTTPServer((redirect_host, redirect_port), OauthAuthorizeRedirectHandler) as httpd:
        httpd.handle_request()

    return OauthAuthorizeRedirectHandler.code


def oauth_exchange_code_with_token(code):
    r = httpx.post(
        "https://login.microsoftonline.com/consumers/oauth2/v2.0/token",
        data={
            "client_id": client_id,
            "scope": scope,
            "code": code,
            "redirect_uri": redirect_uri,
            "grant_type": "authorization_code",
        },
    )
    logger.debug("oauth_exchange_code_with_token: " + r.text)
    return r.json()


def oauth_refresh_token(token):
    r = httpx.post(
        "https://login.microsoftonline.com/consumers/oauth2/v2.0/token",
        data={
            "client_id": client_id,
            "scope": scope,
            "refresh_token": token["refresh_token"],
            "grant_type": "refresh_token",
        },
    )
    logger.debug("oauth_refresh_token: " + r.text)
    return r.json()


def outlook_get_me(access_token):
    r = httpx.get(
        "https://outlook.office.com/api/v2.0/me/",
        headers={
            "Authorization": f"Bearer {access_token}",
        },
    )
    logging.debug("outlook_get_me: " + r.text)
    return r.json()


def outlook_sendmail(message, authobject):
    with smtplib.SMTP(smtp_host, smtp_port) as server:
        if logger.isEnabledFor(logging.DEBUG):
            server.set_debuglevel(2)
        server.starttls()
        server.ehlo()
        server.auth("XOAUTH2", authobject)
        server.send_message(message)
        server.quit()


class Xoauth2Format:
    def __init__(self, user, access_token):
        self._user = user
        self._access_token = access_token

    def __call__(self):
        return f"user={self._user}\1auth=Bearer {self._access_token}\1\1"


def build_message(from_, to, subject, body):
    message = EmailMessage()
    message["From"] = from_
    message["To"] = to
    message["Subject"] = subject
    message.set_content(body)
    return message


# FIXME: Not implemented
def get_cached_token():
    return None


# FIXME: Not implemented
def set_cached_token(token):
    pass


# FIXME: Not implemented
def is_token_expired(token):
    # Is it token["expires_in"] seconds later?
    return False


def main():
    logging.basicConfig(level=logging.DEBUG)

    token = get_cached_token()

    if token is None:
        code = oauth_request_authorization_code()
        token = oauth_exchange_code_with_token(code)
        set_cached_token(token)

    # elif
    if is_token_expired(token):
        token = oauth_refresh_token(token)
        set_cached_token(token)

    me = outlook_get_me(token["access_token"])

    auth = Xoauth2Format(me["EmailAddress"], token["access_token"])
    message = build_message(me["EmailAddress"], me["EmailAddress"], "SMTP.Send access granted", json.dumps(token))
    outlook_sendmail(message, auth)


if __name__ == "__main__":
    main()