Repository: aio-libs/aiosmtpd
Pull request #192 (2020-10-05 10:03:57)

Implement AUTH LOGIN and AUTH PLAIN -- with tests
<!-- Thank you for your contribution! -->

## What do these changes do?

**This PR obsoletes #10 and #66**

### MAJOR changes:

* Implement SMTP AUTH with the LOGIN and PLAIN mechanisms.
* Implement extensible AUTH mechanism via the "handler hook" system
* Returns Enhanced Status Code (RFC3463 and RFC5248) for many situations (not yet exhaustive -- focusing mostly on SMTP AUTH and impacted commands)
* Some commands now emits error if AUTH is required but has not happened

### Minor changes:

* No longer targets Python 3.5 because a hard dependency (`atpublic`) requires Python >= 3.6
* Smattering of type hints here there and everywhere

## Are there changes in behavior for the user?

Existing usage should not be impacted.

Users that implemented #66 must implement some changes, though. Namely:

* Change in `SMTP.__init__` kwarg names
* Change in authenticator 'callback' signature

## Related issue number

## Major:

#102 

### Minor / related / accidental:

#133 #167 #178 #188 #194 

## Checklist

- [x] I think the code is well written
- [x] Unit tests for the changes exist
- [x] Documentation reflects the changes
- [x] Add a new news fragment into the `CHANGES` folder


Patch:
diff --git a/.coveragerc b/.coveragerc
index 37152d5..0ed6f61 100644
--- a/.coveragerc
+++ b/.coveragerc
@@ -18,3 +18,4 @@ source =
 exclude_lines =
     pragma: nocover
     pragma: no${PLATFORM}
+    pragma: skipif_${PY_39}_py39
diff --git a/aiosmtpd/controller.py b/aiosmtpd/controller.py
index be7292a..90c080f 100644
--- a/aiosmtpd/controller.py
+++ b/aiosmtpd/controller.py
@@ -4,12 +4,14 @@ import threading
 
 from aiosmtpd.smtp import SMTP
 from public import public
+from typing import Any, Dict
 
 
 @public
 class Controller:
     def __init__(self, handler, loop=None, hostname=None, port=8025, *,
-                 ready_timeout=1.0, enable_SMTPUTF8=True, ssl_context=None):
+                 ready_timeout=1.0, enable_SMTPUTF8=True, ssl_context=None,
+                 server_kwargs: Dict[str, Any] = None):
         """
         `Documentation can be found here
         <http://aiosmtpd.readthedocs.io/en/latest/aiosmtpd\
@@ -26,10 +28,12 @@ class Controller:
         self._thread_exception = None
         self.ready_timeout = os.getenv(
             'AIOSMTPD_CONTROLLER_TIMEOUT', ready_timeout)
+        self.server_kwargs: Dict[str, Any] = server_kwargs or {}
 
     def factory(self):
         """Allow subclasses to customize the handler/server creation."""
-        return SMTP(self.handler, enable_SMTPUTF8=self.enable_SMTPUTF8)
+        return SMTP(self.handler, enable_SMTPUTF8=self.enable_SMTPUTF8,
+                    **self.server_kwargs)
 
     def _run(self, ready_event):
         asyncio.set_event_loop(self.loop)
@@ -68,7 +72,7 @@ class Controller:
         self.loop.stop()
         try:
             _all_tasks = asyncio.Task.all_tasks
-        except AttributeError:   # pragma: nocover
+        except AttributeError:   # pragma: skipif_lt_py39
             _all_tasks = asyncio.all_tasks
         for task in _all_tasks(self.loop):
             task.cancel()
diff --git a/aiosmtpd/docs/NEWS.rst b/aiosmtpd/docs/NEWS.rst
index 67c75f8..6ff6dcc 100644
--- a/aiosmtpd/docs/NEWS.rst
+++ b/aiosmtpd/docs/NEWS.rst
@@ -2,12 +2,30 @@
  NEWS for aiosmtpd
 ===================
 
-unreleased
-==========
+aiosmtpd-next
+=============
 
-Fixed
+Added
 -----
+* Support for SMTP ``AUTH``, with AUTH hooks feature
+* Built-in implementation for ``AUTH PLAIN`` and ``AUTH LOGIN`` logic (fixes #102)
+* Feature to inject keyword args during server class instantiation in ``Controller.factory``
+  (potentially fixes #194, #179)
+* Support for Python 3.8 and 3.9.0 (also fixes #188)
+
+Fixed/Improved
+--------------
 * Don't strip last ``\r\n`` prior to terminating dot.
+* Slight improvement to make Test Suite more maintainable
+* No more failures/DeprecationWarnings for Python 3.8 (fixes #167)
+* Faster ``_handle_client()`` processing
+* Faster method access for ``smtp_*``, ``handle_*``, and ``auth_*`` hooks
+
+Removed
+-------
+* Unit Tests that mocked too deep, possibly masking observable internal behaviors
+* Drop support for Python 3.5
+
 
 1.2 (2018-09-01)
 ================
diff --git a/aiosmtpd/docs/concepts.rst b/aiosmtpd/docs/concepts.rst
index 20b8bbd..15f4e84 100644
--- a/aiosmtpd/docs/concepts.rst
+++ b/aiosmtpd/docs/concepts.rst
@@ -71,6 +71,12 @@ object is created.
 
       This is the asyncio event loop instance.
 
+   .. attribute:: login_data
+
+      Contains the login information gathered during the ``AUTH`` procedure.
+      If it contains ``None``, that means authentication has not taken place
+      or has failed.
+
 
 Envelope
 --------
diff --git a/aiosmtpd/docs/controller.rst b/aiosmtpd/docs/controller.rst
index 71c6ec0..351ac79 100644
--- a/aiosmtpd/docs/controller.rst
+++ b/aiosmtpd/docs/controller.rst
@@ -161,7 +161,7 @@ The EHLO response does not include the ``SMTPUTF8`` ESMTP option.
 Controller API
 ==============
 
-.. class:: Controller(handler, loop=None, hostname=None, port=8025, *, ready_timeout=1.0, enable_SMTPUTF8=True, ssl_context=None)
+.. class:: Controller(handler, loop=None, hostname=None, port=8025, *, ready_timeout=1.0, enable_SMTPUTF8=True, ssl_context=None, server_kwargs=None)
 
    *handler* is an instance of a :ref:`handler <handlers>` class.
 
@@ -195,6 +195,10 @@ Controller API
    method. Note that this implies unconditional encryption of the connection,
    and prevents use of the ``STARTTLS`` mechanism.
 
+   *server_kwargs* is a dict that will be passed through as keyword arguments
+   to the server's class during server creation in the :meth:`Controller.factory`
+   method.
+
    .. attribute:: handler
 
       The instance of the event *handler* passed to the constructor.
diff --git a/aiosmtpd/docs/handlers.rst b/aiosmtpd/docs/handlers.rst
index 6a1657e..1c4a569 100644
--- a/aiosmtpd/docs/handlers.rst
+++ b/aiosmtpd/docs/handlers.rst
@@ -89,12 +89,22 @@ The following hooks are currently defined:
     instantiated with ``decode_data=False`` (the default) or
     ``decode_data=True``.  In the former case, both ``envelope.content`` and
     ``envelope.original_content`` will be the content bytes (normalized
-    according to the transparency rules in `RFC 5321, §4.5.2
-    <https://tools.ietf.org/html/rfc5321#section-4.5.2>`_).  In the latter
+    according to the transparency rules in |RFC 5321, §4.5.2|_).  In the latter
     case, ``envelope.original_content`` will be the normalized bytes, but
     ``envelope.content`` will be the UTF-8 decoded string of the original
     content.
 
+``handle_AUTH(server, session, envelope, args)``
+    Called to handle ``AUTH`` command, if you need custom AUTH behavior.
+    You *must* comply with |RFC 4954|_.
+    Most of the time, you don't *need* to implement this hook;
+    :ref:`AUTH hooks <auth_hooks>` are provided to override/implement selctive
+    SMTP AUTH mechanisms (see below).
+
+    ``args`` will contain the list of words following the ``AUTH`` command.
+    You will need to call some ``server`` methods and modify some ``session``
+    properties. ``envelope`` is usually ignored.
+
 In addition to the SMTP command hooks, the following hooks can also be
 implemented by handlers.  These have different APIs, and are called
 synchronously (i.e. they are **not** coroutines).
@@ -109,8 +119,59 @@ synchronously (i.e. they are **not** coroutines).
     handling of a connection (e.g. if an ``smtp_<command>()`` method raises an
     exception).  The exception object is passed in.  This method *must* return
     a status string, such as ``'542 Internal server error'``.  If the method
-    returns None or raises an exception, an exception will be logged, and a 500
-    code will be returned to the client.
+    returns ``None`` or raises an exception, an exception will be logged, and a
+    ``500`` code will be returned to the client.
+
+
+.. _auth_hooks:
+
+AUTH hooks
+=============
+
+In addition to the above SMTP hooks, you can also implement AUTH hooks.
+**These hooks are asynchronous**.
+Every AUTH hook is named ``auth_MECHANISM`` where ``MECHANISM`` is the all-uppercase
+mechanism that the hook will implement. AUTH hooks will be called with the SMTP
+server instance and a list of str following the ``AUTH`` command.
+
+The SMTP class provides built-in AUTH hooks for the ``LOGIN`` and ``PLAIN``
+mechanisms, named ``auth_LOGIN`` and ``auth_PLAIN``, respectively.
+If the handler class implements ``auth_LOGIN`` and/or ``auth_PLAIN``, then
+those methods of the handler instance will override the built-in methods.
+
+``auth_MECHANISM(server: SMTP, args: List[str])``
+
+  *server* is the instance of the ``SMTP`` class invoking the AUTH hook.
+  This allows the AUTH hook implementation to invoke facilities such as the
+  ``push()`` and ``_auth_interact()`` methods.
+
+  *args* is a list of string split from the string after the ``AUTH`` command.
+  ``args[0]`` is always equal to ``MECHANISM``.
+
+  The AUTH hook **must** perform the actual validation of AUTH credentials.
+  In the built-in AUTH hooks, this is done by invoking the function specified
+  by the ``auth_callback`` initialization argument. AUTH hooks in handlers
+  are NOT required to do the same.
+
+  The AUTH hook **must** return one of the following values:
+
+    * ``None`` -- an error happened during AUTH exchange/procedure, and has
+      been handled inside the hook. ``smtp_AUTH`` will not do anything more.
+
+    * ``MISSING`` -- no error during exchange, but the credentials received
+      are invalid/rejected. (``MISSING`` is a pre-instantiated object you
+      can import from ``aiosmtpd.smtp``)
+
+    * *Anything else* -- an 'identity' of the STMP user. Usually is the username
+      given during AUTH exchange/procedure, but not necessarily so; can also
+      be, for instance, a Session ID. This will be stored in the Session
+      object's ``login_data`` property (see
+      :ref:`Session and Envelopes <sessions_and_envelopes>`)
+
+**NOTE:** Defining *additional* AUTH hooks in your handler will NOT disable
+the built-in LOGIN and PLAIN hooks; if you do not want to offer the LOGIN and
+PLAIN mechanisms, specify them in the ``auth_exclude_mechanism`` parameter
+of the :ref:`SMTP class<smtp_api>`.
 
 
 Built-in handlers
@@ -228,3 +289,7 @@ We open up the mailbox again, and all three messages are waiting for us.
 .. _ArgumentParser: https://docs.python.org/3/library/argparse.html#argumentparser-objects
 .. _`email.message.Message`: https://docs.python.org/3/library/email.compat32-message.html#email.message.Message
 .. _Maildir: https://docs.python.org/3/library/mailbox.html#maildir
+.. _RFC 4954: https://tools.ietf.org/html/rfc4954
+.. |RFC 4954| replace:: **RFC 4954**
+.. _RFC 5321, §4.5.2: https://tools.ietf.org/html/rfc5321#section-4.5.2
+.. |RFC 5321, §4.5.2| replace:: **RFC 5321, §4.5.2**
diff --git a/aiosmtpd/docs/intro.rst b/aiosmtpd/docs/intro.rst
index b79eacb..f8bccf6 100644
--- a/aiosmtpd/docs/intro.rst
+++ b/aiosmtpd/docs/intro.rst
@@ -32,6 +32,8 @@ Relevant RFCs
   Extension for Returning Enhanced Error Codes
 * `RFC 6531 <http://www.faqs.org/rfcs/rfc6531.html>`__ - SMTP Extension for
   Internationalized Email
+* `RFC 4954 <http://www.faqs.org/rfcs/rfc4954.html>`__ - SMTP Service Extension
+  for Authentication
 
 
 Other references
diff --git a/aiosmtpd/docs/smtp.rst b/aiosmtpd/docs/smtp.rst
index 1dbeca3..38629eb 100644
--- a/aiosmtpd/docs/smtp.rst
+++ b/aiosmtpd/docs/smtp.rst
@@ -66,7 +66,7 @@ Because we prefixed the ``smtp_PING()`` method with the ``@syntax()``
 decorator, the command shows up in the ``HELP`` output.
 
     >>> print(client.help().decode('utf-8'))
-    Supported commands: DATA EHLO HELO HELP MAIL NOOP PING QUIT RCPT RSET VRFY
+    Supported commands: AUTH DATA EHLO HELO HELP MAIL NOOP PING QUIT RCPT RSET VRFY
 
 And we can get more detailed help on the new command.
 
@@ -98,10 +98,12 @@ override to provide additional responses.
     server.
 
 
+.. _smtp_api:
+
 SMTP API
 ========
 
-.. class:: SMTP(handler, *, data_size_limit=33554432, enable_SMTPUTF8=False, decode_data=False, hostname=None, ident=None, tls_context=None, require_starttls=False, timeout=300, loop=None)
+.. class:: SMTP(handler, *, data_size_limit=33554432, enable_SMTPUTF8=False, decode_data=False, hostname=None, ident=None, tls_context=None, require_starttls=False, timeout=300, auth_required=False, auth_require_tls=True, auth_exclude_mechanism=None, auth_callback=None, loop=None)
 
    *handler* is an instance of a :ref:`handler <handlers>` class.
 
@@ -135,6 +137,24 @@ SMTP API
    After this time the connection will be closed by the server.  The default
    is 300 seconds, as per `RFC 2821`_.
 
+   *auth_required* specifies whether SMTP Authentication is mandatory or
+   not for the session. This impacts some SMTP commands such as HELP, MAIL
+   FROM, RCPT TO, and others.
+
+   *auth_require_tls* specifies whether ``STARTTLS`` must be used before
+   AUTH exchange or not. If you set this to ``False`` then AUTH exchange can
+   be done outside a TLS context, but the class will warn you of security
+   considerations. Please note that *require_starttls* takes precedence
+   over this setting.
+
+   *auth_exclude_mechanism* is an ``Iterable[str]`` that specifies SMTP AUTH
+   mechanisms to NOT use.
+
+   *auth_callback* is a function that accepts three arguments: ``mechanism: str``,
+   ``login: bytes``, and ``password: bytes``. Based on these args, the function
+   must return a ``bool`` that indicates whether the client's authentication
+   attempt is accepted/successful or not.
+
    *loop* is the asyncio event loop to use.  If not given,
    :meth:`asyncio.new_event_loop()` is called to create the event loop.
 
@@ -184,6 +204,10 @@ SMTP API
       The event loop being used.  This will either be the given *loop*
       argument, or the new event loop that was created.
 
+   .. attribute:: authenticated
+
+      A flag that indicates whether authentication had succeeded.
+
    .. method:: _create_session()
 
       A method subclasses can override to return custom ``Session`` instances.
diff --git a/aiosmtpd/handlers.py b/aiosmtpd/handlers.py
index a864f23..2d201f6 100644
--- a/aiosmtpd/handlers.py
+++ b/aiosmtpd/handlers.py
@@ -100,11 +100,10 @@ class Proxy:
         # Look for the last header
         i = 0
         ending = CRLF
-        for line in lines:                          # pragma: nobranch
+        for i, line in enumerate(lines):  # pragma: nobranch
             if NLCRE.match(line):
                 ending = line
                 break
-            i += 1
         peer = session.peer[0].encode('ascii')
         lines.insert(i, b'X-Peer: %s%s' % (peer, ending))
         data = EMPTYBYTES.join(lines)
diff --git a/aiosmtpd/main.py b/aiosmtpd/main.py
index 8d84a1f..36d3dfd 100644
--- a/aiosmtpd/main.py
+++ b/aiosmtpd/main.py
@@ -137,23 +137,31 @@ def main(args=None):
     if args.debug > 2:
         loop.set_debug(enabled=True)
 
-    log.info('Server listening on %s:%s', args.host, args.port)
+    log.debug('Attempting to start server on %s:%s', args.host, args.port)
+    server = server_loop = None
+    try:
+        server = loop.create_server(factory, host=args.host, port=args.port)
+        server_loop = loop.run_until_complete(server)
+    except RuntimeError as e:  # pragma: nocover
+        log.critical("\u001b[33;1m" + "#" * 600 + "\u001b[0m",
+                     exc_info=e)
+        raise
+    log.debug(f"server_loop = {server_loop}")
+    log.info('Server is listening on %s:%s', args.host, args.port)
 
-    server = loop.run_until_complete(
-        loop.create_server(factory, host=args.host, port=args.port))
     # Signal handlers are only supported on *nix, so just ignore the failure
     # to set this on Windows.
     with suppress(NotImplementedError):
         loop.add_signal_handler(signal.SIGINT, loop.stop)
 
-    log.info('Starting asyncio loop')
+    log.debug('Starting asyncio loop')
     try:
         loop.run_forever()
     except KeyboardInterrupt:
         pass
-    server.close()
-    log.info('Completed asyncio loop')
-    loop.run_until_complete(server.wait_closed())
+    server_loop.close()
+    log.debug('Completed asyncio loop')
+    loop.run_until_complete(server_loop.wait_closed())
     loop.close()
 
 
diff --git a/aiosmtpd/smtp.py b/aiosmtpd/smtp.py
index f5253d7..b810b4c 100644
--- a/aiosmtpd/smtp.py
+++ b/aiosmtpd/smtp.py
@@ -2,12 +2,25 @@ import ssl
 import socket
 import asyncio
 import logging
+import binascii
 import collections
 import asyncio.sslproto as sslproto
 
+from base64 import b64decode
 from email._header_value_parser import get_addr_spec, get_angle_addr
 from email.errors import HeaderParseError
 from public import public
+from typing import (
+    Any,
+    Awaitable,
+    Callable,
+    Dict,
+    Iterable,
+    List,
+    NamedTuple,
+    Optional,
+    Union,
+)
 from warnings import warn
 
 
@@ -19,7 +32,22 @@ log = logging.getLogger('mail.log')
 DATA_SIZE_DEFAULT = 33554432
 EMPTYBYTES = b''
 NEWLINE = '\n'
-MISSING = object()
+
+
+class _Missing:
+    pass
+
+
+MISSING = _Missing()
+
+
+AuthMechanismType = Callable[["SMTP", List[str]], Awaitable[Any]]
+_TriStateType = Union[None, _Missing, bytes]
+
+
+class _AuthMechAttr(NamedTuple):
+    method: AuthMechanismType
+    is_builtin: bool
 
 
 @public
@@ -30,6 +58,7 @@ class Session:
         self.host_name = None
         self.extended_smtp = False
         self.loop = loop
+        self.login_data = None
 
 
 @public
@@ -38,7 +67,7 @@ class Envelope:
         self.mail_from = None
         self.mail_options = []
         self.smtp_utf8 = False
-        self.content = None
+        self.content: Union[None, bytes, str] = None
         self.original_content = None
         self.rcpt_tos = []
         self.rcpt_options = []
@@ -60,6 +89,10 @@ def syntax(text, extended=None, when=None):
     return decorator
 
 
+def login_always_fail(mechanism, login, password):
+    return False
+
+
 @public
 class SMTP(asyncio.StreamReaderProtocol):
     command_size_limit = 512
@@ -76,6 +109,10 @@ class SMTP(asyncio.StreamReaderProtocol):
                  tls_context=None,
                  require_starttls=False,
                  timeout=300,
+                 auth_required=False,
+                 auth_require_tls=True,
+                 auth_exclude_mechanism: Optional[Iterable[str]] = None,
+                 auth_callback: Callable[[str, bytes, bytes], bool] = None,
                  loop=None):
         self.__ident__ = ident or __ident__
         self.loop = loop if loop else make_loop()
@@ -108,6 +145,37 @@ class SMTP(asyncio.StreamReaderProtocol):
         self.envelope = None
         self.transport = None
         self._handler_coroutine = None
+        if not auth_require_tls and auth_required:
+            warn("Requiring AUTH while not requiring TLS "
+                 "can lead to security vulnerabilities!")
+        self._auth_require_tls = auth_require_tls
+        self._auth_callback = auth_callback or login_always_fail
+        self._auth_required = auth_required
+        self.authenticated = False
+        # Get hooks & methods to significantly speedup getattr's
+        self._auth_methods: Dict[str, _AuthMechAttr] = {
+            m.replace("auth_", ""): _AuthMechAttr(getattr(h, m), h is self)
+            for h in (self, handler)
+            for m in dir(h)
+            if m.startswith("auth_")
+        }
+        for m in (auth_exclude_mechanism or []):
+            self._auth_methods.pop(m, None)
+        msg = "Available AUTH mechanisms:"
+        for m, impl in sorted(
+                self._auth_methods.items()):  # type: str, _AuthMechAttr
+            msg += f" {m}{'(builtin)' if impl.is_builtin else ''}"
+        log.info(msg)
+        self._handle_hooks: Dict[str, Callable] = {
+            m.replace("handle_", ""): getattr(handler, m)
+            for m in dir(handler)
+            if m.startswith("handle_")
+        }
+        self._smtp_methods: Dict[str, Any] = {
+            m.replace("smtp_", ""): getattr(self, m)
+            for m in dir(self)
+            if m.startswith("smtp_")
+        }
 
     def _create_session(self):
         return Session(self.loop)
@@ -116,7 +184,7 @@ class SMTP(asyncio.StreamReaderProtocol):
         return Envelope()
 
     async def _call_handler_hook(self, command, *args):
-        hook = getattr(self.event_handler, 'handle_' + command, None)
+        hook = self._handle_hooks.get(command)
         if hook is None:
             return MISSING
         status = await hook(self, self.session, self.envelope, *args)
@@ -144,11 +212,11 @@ class SMTP(asyncio.StreamReaderProtocol):
             # Do SSL certificate checking as rfc3207 part 4.1 says.  Why is
             # _extra a protected attribute?
             self.session.ssl = self._tls_protocol._extra
-            handler = getattr(self.event_handler, 'handle_STARTTLS', None)
-            if handler is None:
+            hook = self._handle_hooks.get("STARTTLS")
+            if hook is None:
                 self._tls_handshake_okay = True
             else:
-                self._tls_handshake_okay = handler(
+                self._tls_handshake_okay = hook(
                     self, self.session, self.envelope)
         else:
             super().connection_made(transport)
@@ -239,7 +307,7 @@ class SMTP(asyncio.StreamReaderProtocol):
         while self.transport is not None:   # pragma: nobranch
             # XXX Put the line limit stuff into the StreamReader?
             try:
-                line = await self._reader.readline()
+                line: bytes = await self._reader.readline()
                 log.debug('_handle_client readline: %s', line)
                 # XXX this rstrip may not completely preserve old behavior.
                 line = line.rstrip(b'\r\n')
@@ -247,28 +315,21 @@ class SMTP(asyncio.StreamReaderProtocol):
                 if not line:
                     await self.push('500 Error: bad syntax')
                     continue
-                i = line.find(b' ')
+                command, _, arg = line.partition(b" ")
                 # Decode to string only the command name part, which must be
                 # ASCII as per RFC.  If there is an argument, it is decoded to
                 # UTF-8/surrogateescape so that non-UTF-8 data can be
                 # re-encoded back to the original bytes when the SMTP command
                 # is handled.
-                if i < 0:
-                    try:
-                        command = line.upper().decode(encoding='ascii')
-                    except UnicodeDecodeError:
-                        await self.push('500 Error: bad syntax')
-                        continue
-
+                try:
+                    command = command.upper().decode(encoding='ascii')
+                except UnicodeDecodeError:
+                    await self.push('500 Error: bad syntax')
+                    continue
+                if not arg:
                     arg = None
                 else:
-                    try:
-                        command = line[:i].upper().decode(encoding='ascii')
-                    except UnicodeDecodeError:
-                        await self.push('500 Error: bad syntax')
-                        continue
-
-                    arg = line[i + 1:].strip()
+                    arg = arg.strip()
                     # Remote SMTP servers can send us UTF-8 content despite
                     # whether they've declared to do so or not.  Some old
                     # servers can send 8-bit data.  Use surrogateescape so
@@ -304,7 +365,7 @@ class SMTP(asyncio.StreamReaderProtocol):
                     # RFC3207 part 4
                     await self.push('530 Must issue a STARTTLS command first')
                     continue
-                method = getattr(self, 'smtp_' + command, None)
+                method = self._smtp_methods.get(command)
                 if method is None:
                     await self.push(
                         '500 Error: command "%s" not recognized' % command)
@@ -333,6 +394,31 @@ class SMTP(asyncio.StreamReaderProtocol):
                         status = '500 Error: Cannot describe error'
                     await self.push(status)
 
+    async def check_helo_needed(self, helo: str = "HELO") -> bool:
+        """
+        Check if HELO/EHLO is needed.
+
+        :param helo: The actual string of HELO/EHLO
+        :return: True if HELO/EHLO is needed
+        """
+        if not self.session.host_name:
+            await self.push(f'503 Error: send {helo} first')
+            return True
+        return False
+
+    async def check_auth_needed(self, caller_method: str) -> bool:
+        """
+        Check if AUTH is needed.
+
+        :param caller_method: The SMTP method needing a check (for logging)
+        :return: True if AUTH is needed
+        """
+        if self._auth_required and not self.authenticated:
+            log.info(f'{caller_method}: Authentication required')
+            await self.push('530 5.7.0 Authentication required')
+            return True
+        return False
+
     # SMTP and ESMTP commands
     @syntax('HELO hostname')
     async def smtp_HELO(self, hostname):
@@ -369,6 +455,10 @@ class SMTP(asyncio.StreamReaderProtocol):
             warn('Use handler.handle_EHLO() instead of .ehlo_hook()',
                  DeprecationWarning)
             await self.ehlo_hook()
+        if not self._auth_require_tls or self._tls_protocol:
+            await self.push(
+                "250-AUTH " + " ".join(sorted(self._auth_methods.keys()))
+            )
         status = await self._call_handler_hook('EHLO', hostname)
         if status is MISSING:
             self.session.host_name = hostname
@@ -401,6 +491,7 @@ class SMTP(asyncio.StreamReaderProtocol):
             return
         await self.push('220 Ready to start TLS')
         # Create SSL layer.
+        # noinspection PyTypeChecker
         self._tls_protocol = sslproto.SSLProtocol(
             self.loop,
             self,
@@ -417,6 +508,149 @@ class SMTP(asyncio.StreamReaderProtocol):
         self.transport = self._tls_protocol._app_transport
         self._tls_protocol.connection_made(self._original_transport)
 
+    @syntax("AUTH <mechanism>")
+    async def smtp_AUTH(self, arg: str) -> None:
+        if await self.check_helo_needed("EHLO"):
+            return
+        elif not self.session.extended_smtp:
+            await self.push("500 Error: command 'AUTH' not recognized")
+        elif self._auth_require_tls and not self._tls_protocol:
+            await self.push("538 5.7.11 Encryption required for requested "
+                            "authentication mechanism")
+        elif self.authenticated:
+            await self.push('503 Already authenticated')
+        elif not arg:
+            await self.push('501 Not enough value')
+        else:
+            args = arg.split()
+            if len(args) > 2:
+                await self.push('501 Too many values')
+                return
+
+            mechanism = args[0]
+            if mechanism not in self._auth_methods:
+                await self.push('504 5.5.4 Unrecognized authentication type')
+                return
+
+            status = await self._call_handler_hook('AUTH', args)
+            if status is MISSING:
+                method = self._auth_methods[mechanism]
+                if method.is_builtin:
+                    log.debug(f"Using builtin auth_ hook for {mechanism}")
+                else:
+                    log.debug(f"Using handler auth_ hook for {mechanism}")
+                # Pass 'self' to method so external methods can leverage this
+                # class's helper methods such as push()
+                login_data = await method.method(self, args)
+                log.debug(f"auth_{mechanism} returned {login_data}")
+                if login_data is None:
+                    # None means there's an error already handled by method and
+                    # we don't need to do anything more
+                    return
+                elif login_data is MISSING:
+                    # MISSING means no error in AUTH process, but credentials
+                    # is rejected / not valid
+                    status = '535 5.7.8 Authentication credentials invalid'
+                else:
+                    self.authenticated = True
+                    self.session.login_data = login_data
+                    status = '235 2.7.0 Authentication successful'
+            if status is not None:  # pragma: no branch
+                await self.push(status)
+
+    async def _auth_interact(self, server_message) -> _TriStateType:
+        blob: bytes
+        await self.push(server_message)
+        line = await self._reader.readline()
+        blob = line.strip()
+        # '=' and '*' handling are in accordance with RFC4954
+        if blob == b"=":
+            log.debug("User responded with '='")
+            return None
+        if blob == b"*":
+            log.warning("User requested abort with '*'")
+            await self.push("501 Auth aborted")
+            return MISSING
+        try:
+            decoded_blob = b64decode(blob, validate=True)
+        except binascii.Error:
+            await self.push("501 5.5.2 Can't decode base64")
+            return MISSING
+        return decoded_blob
+
+    # IMPORTANT NOTES FOR THE auth_* METHODS
+    #
+    # 1. For internal methods, due to how they are called, we must ignore
+    #    the first arg
+    # 2. All auth_* methods can return one of three values:
+    #    - None: An error happened and handled;
+    #            smtp_AUTH should do nothing more
+    #    - MISSING: No error during SMTP AUTH process, but authentication
+    #               failed
+    #    - [Any]: Authentication succeeded and this is the 'identity' of
+    #             the SMTP user
+    #      - 'identity' is not always username, depending on the auth mecha-
+    #        nism. Might be a session key, a one-time user ID, or any kind of
+    #        object, actually.
+    #      - If the client provides "=" for username during interaction, the
+    #        method MUST return b"" (empty bytes)
+    # 3. Auth credentials checking is performed in the auth_* methods because
+    #    more advanced auth mechanism might not return login+password pair
+    #    (see #2 above)
+
+    async def auth_PLAIN(self, _, args: List[str]):
+        loginpassword: _TriStateType
+        if len(args) == 1:
+            # Trailing space is MANDATORY
+            # See https://tools.ietf.org/html/rfc4954#page-4
+            loginpassword = await self._auth_interact("334 ")
+            if loginpassword is MISSING:
+                return
+        else:
+            blob = args[1].encode()
+            if blob == b"=":
+                loginpassword = None
+            else:
+                try:
+                    loginpassword = b64decode(blob, validate=True)
+                except Exception:
+                    await self.push("501 5.5.2 Can't decode base64")
+                    return
+        if loginpassword is None:
+            login = password = None
+        else:
+            try:
+                _, login, password = loginpassword.split(b"\x00")
+            except ValueError:  # not enough args
+                await self.push("501 5.5.2 Can't split auth value")
+                return
+        if self._auth_callback("PLAIN", login, password):
+            if login is None:
+                login = EMPTYBYTES
+            return login
+        else:
+            return MISSING
+
+    async def auth_LOGIN(self, _, args: List[str]):
+        login: _TriStateType
+        # 'User Name\x00'
+        login = await self._auth_interact("334 VXNlciBOYW1lAA==")
+        if login is MISSING:
+            return
+
+        password: _TriStateType
+        # 'Password\x00'
+        password = await self._auth_interact("334 UGFzc3dvcmQA")
+        if password is MISSING:
+            return
+
+        if self._auth_callback("LOGIN", login, password):
+            if login is None:  # pragma: no branch
+                login = EMPTYBYTES
+            return login
+        else:
+            return MISSING
+
     def _strip_command_keyword(self, keyword, arg):
         keylen = len(keyword)
         if arg[:keylen].upper() == keyword:
@@ -449,17 +683,19 @@ class SMTP(asyncio.StreamReaderProtocol):
         return result
 
     def _syntax_available(self, method):
-        if getattr(method, '__smtp_syntax__', None) is None:
+        if not hasattr(method, '__smtp_syntax__'):
             return False
         if method.__smtp_syntax_when__:
             return bool(getattr(self, method.__smtp_syntax_when__))
         return True
 
     @syntax('HELP [command]')
-    async def smtp_HELP(self, arg):
+    async def smtp_HELP(self, arg: str) -> None:
+        if await self.check_auth_needed("HELP"):
+            return
         code = 250
         if arg:
-            method = getattr(self, 'smtp_' + arg.upper(), None)
+            method = self._smtp_methods.get(arg.upper())
             if method and self._syntax_available(method):
                 help_str = method.__smtp_syntax__
                 if (self.session.extended_smtp
@@ -469,18 +705,17 @@ class SMTP(asyncio.StreamReaderProtocol):
                 return
             code = 501
         commands = []
-        for name in dir(self):
-            if not name.startswith('smtp_'):
-                continue
-            method = getattr(self, name)
+        for name, method in self._smtp_methods.items():
             if self._syntax_available(method):
-                commands.append(name.lstrip('smtp_'))
+                commands.append(name)
         commands.sort()
         await self.push(
             '{} Supported commands: {}'.format(code, ' '.join(commands)))
 
     @syntax('VRFY <address>')
-    async def smtp_VRFY(self, arg):
+    async def smtp_VRFY(self, arg: str) -> None:
+        if await self.check_auth_needed("VRFY"):
+            return
         if arg:
             try:
                 address, params = self._getaddr(arg)
@@ -498,9 +733,10 @@ class SMTP(asyncio.StreamReaderProtocol):
             await self.push('501 Syntax: VRFY <address>')
 
     @syntax('MAIL FROM: <address>', extended=' [SP <mail-parameters>]')
-    async def smtp_MAIL(self, arg):
-        if not self.session.host_name:
-            await self.push('503 Error: send HELO first')
+    async def smtp_MAIL(self, arg: str) -> None:
+        if await self.check_helo_needed():
+            return
+        if await self.check_auth_needed("MAIL"):
             return
         log.debug('===> MAIL %s', arg)
         syntaxerr = '501 Syntax: MAIL FROM: <address>'
@@ -565,9 +801,10 @@ class SMTP(asyncio.StreamReaderProtocol):
         await self.push(status)
 
     @syntax('RCPT TO: <address>', extended=' [SP <mail-parameters>]')
-    async def smtp_RCPT(self, arg):
-        if not self.session.host_name:
-            await self.push('503 Error: send HELO first')
+    async def smtp_RCPT(self, arg: str) -> None:
+        if await self.check_helo_needed():
+            return
+        if await self.check_auth_needed("RCPT"):
             return
         log.debug('===> RCPT %s', arg)
         if not self.envelope.mail_from:
@@ -584,9 +821,6 @@ class SMTP(asyncio.StreamReaderProtocol):
             await self.push(syntaxerr)
             return
         address, params = self._getaddr(arg)
-        if address is None:
-            await self.push(syntaxerr)
-            return
         if not address:
             await self.push(syntaxerr)
             return
@@ -625,9 +859,10 @@ class SMTP(asyncio.StreamReaderProtocol):
         await self.push('250 OK' if status is MISSING else status)
 
     @syntax('DATA')
-    async def smtp_DATA(self, arg):
-        if not self.session.host_name:
-            await self.push('503 Error: send HELO first')
+    async def smtp_DATA(self, arg: str) -> None:
+        if await self.check_helo_needed():
+            return
+        if await self.check_auth_needed("DATA"):
             return
         if not self.envelope.rcpt_tos:
             await self.push('503 Error: need RCPT command')
@@ -651,9 +886,9 @@ class SMTP(asyncio.StreamReaderProtocol):
             if line == b'.\r\n':
                 break
             num_bytes += len(line)
-            if (not size_exceeded and
-                    self.data_size_limit and
-                    num_bytes > self.data_size_limit):
+            if (not size_exceeded
+                    and self.data_size_limit
+                    and num_bytes > self.data_size_limit):
                 size_exceeded = True
                 await self.push('552 Error: Too much mail data')
             if not size_exceeded:
@@ -663,9 +898,8 @@ class SMTP(asyncio.StreamReaderProtocol):
             return
         # Remove extraneous carriage returns and de-transparency
         # according to RFC 5321, Section 4.5.2.
-        for i in range(len(data)):
-            text = data[i]
-            if text and text[:1] == b'.':
+        for i, text in enumerate(data):
+            if text.startswith(b'.'):
                 data[i] = text[1:]
         content = original_content = EMPTYBYTES.join(data)
         if self._decode_data:
@@ -684,7 +918,7 @@ class SMTP(asyncio.StreamReaderProtocol):
         self.envelope.content = content
         self.envelope.original_content = original_content
         # Call the new API first if it's implemented.
-        if hasattr(self.event_handler, 'handle_DATA'):
+        if "DATA" in self._handle_hooks:
             status = await self._call_handler_hook('DATA')
         else:
             # Backward compatibility.
diff --git a/tox.ini b/tox.ini
index 684f1ed..6c27f1e 100644
--- a/tox.ini
+++ b/tox.ini
@@ -7,8 +7,8 @@ commands =
     python -c "import sys; print(sys.version); print(sys.executable)"
     python -c "import os,pprint; os.makedirs('_dynamic', exist_ok=True); pprint.pprint(dict(os.environ), stream=open('_dynamic/ENV."{envname}"', 'wt'))"
     nocov: python -m nose2 -v {posargs}
-    {cov,diffcov}: python -m coverage run -m nose2
-    {cov,diffcov}: python -m coverage combine
+    cov: python -m coverage run -m nose2
+    cov: python -m coverage combine
     cov: python -m coverage html --directory=htmlcov/{envname}
     cov: python -m coverage report -m --fail-under=100
     diffcov: python -m coverage xml
@@ -33,6 +33,8 @@ setenv =
     py39: INTERP=py39
     pypy3: INTERP=pypy3
     nocov: PYTHONASYNCIODEBUG=1
+    !py39: PY_39=lt
+    py39: PY_39=ge
     PLATFORM={env:PLATFORM:linux}
 passenv =
     PYTHON*


Test patch:
diff --git a/aiosmtpd/testing/helpers.py b/aiosmtpd/testing/helpers.py
index a88b550..6999d32 100644
--- a/aiosmtpd/testing/helpers.py
+++ b/aiosmtpd/testing/helpers.py
@@ -8,6 +8,7 @@ import logging
 import warnings
 
 from contextlib import ExitStack
+from unittest import TestCase
 from unittest.mock import patch
 
 
@@ -58,3 +59,26 @@ def start(plugin):
         logging.getLogger('asyncio').setLevel(logging.DEBUG)
         logging.getLogger('mail.log').setLevel(logging.DEBUG)
         warnings.filterwarnings('always', category=ResourceWarning)
+
+
+def assert_auth_success(testcase: TestCase, code, response):
+    testcase.assertEqual(code, 235)
+    testcase.assertEqual(response, b"2.7.0 Authentication successful")
+
+
+def assert_auth_invalid(testcase: TestCase, code, response):
+    testcase.assertEqual(code, 535)
+    testcase.assertEqual(response, b'5.7.8 Authentication credentials invalid')
+
+
+def assert_auth_required(testcase: TestCase, code, response):
+    testcase.assertEqual(code, 530)
+    testcase.assertEqual(response, b'5.7.0 Authentication required')
+
+
+SUPPORTED_COMMANDS_TLS: bytes = (
+    b'Supported commands: AUTH DATA EHLO HELO HELP MAIL '
+    b'NOOP QUIT RCPT RSET STARTTLS VRFY'
+)
+
+SUPPORTED_COMMANDS_NOTLS = SUPPORTED_COMMANDS_TLS.replace(b" STARTTLS", b"")
diff --git a/aiosmtpd/tests/test_handlers.py b/aiosmtpd/tests/test_handlers.py
index dc5de7c..49b69f4 100644
--- a/aiosmtpd/tests/test_handlers.py
+++ b/aiosmtpd/tests/test_handlers.py
@@ -22,6 +22,11 @@ class DecodingController(Controller):
         return Server(self.handler, decode_data=True)
 
 
+class AUTHDecodingController(Controller):
+    def factory(self):
+        return Server(self.handler, decode_data=True, auth_require_tls=False)
+
+
 class DataHandler:
     def __init__(self):
         self.content = None
@@ -538,6 +543,12 @@ class DATAHandler:
         return '599 Not today'
 
 
+class AUTHHandler:
+    async def handle_AUTH(self, server, session, envelope, args):
+        server.authenticates = True
+        return '235 Authentication successful'
+
+
 class NoHooksHandler:
     pass
 
@@ -604,6 +615,16 @@ Yikes
             self.assertEqual(cm.exception.smtp_code, 599)
             self.assertEqual(cm.exception.smtp_error, b'Not today')
 
+    def test_auth_hook(self):
+        controller = AUTHDecodingController(AUTHHandler())
+        controller.start()
+        self.addCleanup(controller.stop)
+        with SMTP(controller.hostname, controller.port) as client:
+            client.ehlo('me')
+            code, response = client.login("test", "test")
+        self.assertEqual(code, 235)
+        self.assertEqual(response, b'Authentication successful')
+
     def test_no_hooks(self):
         controller = Controller(NoHooksHandler())
         controller.start()
@@ -633,6 +654,10 @@ class CapturingServer(Server):
 
 
 class CapturingController(Controller):
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.smtpd = None
+
     def factory(self):
         self.smtpd = CapturingServer(self.handler)
         return self.smtpd
@@ -671,6 +696,10 @@ class DeprecatedHookServer(Server):
 
 
 class DeprecatedHookController(Controller):
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.smtpd = None
+
     def factory(self):
         self.smtpd = DeprecatedHookServer(self.handler)
         return self.smtpd
diff --git a/aiosmtpd/tests/test_lmtp.py b/aiosmtpd/tests/test_lmtp.py
index d77816e..a3855e4 100644
--- a/aiosmtpd/tests/test_lmtp.py
+++ b/aiosmtpd/tests/test_lmtp.py
@@ -48,5 +48,5 @@ class TestLMTP(unittest.TestCase):
             code, response = client.docmd('HELP')
             self.assertEqual(code, 250)
             self.assertEqual(response,
-                             b'Supported commands: DATA HELP LHLO MAIL '
+                             b'Supported commands: AUTH DATA HELP LHLO MAIL '
                              b'NOOP QUIT RCPT RSET VRFY')
diff --git a/aiosmtpd/tests/test_main.py b/aiosmtpd/tests/test_main.py
index f7db9d8..0581080 100644
--- a/aiosmtpd/tests/test_main.py
+++ b/aiosmtpd/tests/test_main.py
@@ -1,14 +1,11 @@
 import os
-import signal
 import asyncio
 import logging
 import unittest
 
-from aiosmtpd.handlers import Debugging
 from aiosmtpd.main import main, parseargs
-from aiosmtpd.smtp import SMTP, __version__
+from aiosmtpd.smtp import __version__
 from contextlib import ExitStack
-from functools import partial
 from io import StringIO
 from unittest.mock import patch
 
@@ -133,11 +130,11 @@ class TestMain(unittest.TestCase):
     def test_debug_0(self):
         # For this test, the runner will have already set the log level so it
         # may not be logging.ERROR.
-        log = logging.getLogger('mail.log')
-        default_level = log.getEffectiveLevel()
-        with patch.object(log, 'info'):
+        _log = logging.getLogger('mail.log')
+        default_level = _log.getEffectiveLevel()
+        with patch.object(_log, 'info'):
             main(('-n',))
-            self.assertEqual(log.getEffectiveLevel(), default_level)
+            self.assertEqual(_log.getEffectiveLevel(), default_level)
 
     def test_debug_1(self):
         # Mock the logger to eliminate console noise.
@@ -159,91 +156,6 @@ class TestMain(unittest.TestCase):
             self.assertTrue(asyncio.get_event_loop().get_debug())
 
 
-class TestLoop(unittest.TestCase):
-    def setUp(self):
-        # We mock out so much of this, is it even worthwhile testing?  Well, it
-        # does give us coverage.
-        self.loop = asyncio.get_event_loop()
-        pfunc = partial(patch.object, self.loop)
-        resources = ExitStack()
-        self.addCleanup(resources.close)
-        self.create_server = resources.enter_context(pfunc('create_server'))
-        self.run_until_complete = resources.enter_context(
-            pfunc('run_until_complete'))
-        self.add_signal_handler = resources.enter_context(
-            pfunc('add_signal_handler'))
-        resources.enter_context(
-            patch.object(logging.getLogger('mail.log'), 'info'))
-        self.run_forever = resources.enter_context(pfunc('run_forever'))
-
-    def test_loop(self):
-        main(('-n',))
-        # create_server() is called with a partial as the factory, and a
-        # socket object.
-        self.assertEqual(self.create_server.call_count, 1)
-        positional, keywords = self.create_server.call_args
-        self.assertEqual(positional[0].func, SMTP)
-        self.assertEqual(len(positional[0].args), 1)
-        self.assertIsInstance(positional[0].args[0], Debugging)
-        self.assertEqual(positional[0].keywords, dict(
-            data_size_limit=None,
-            enable_SMTPUTF8=False))
-        self.assertEqual(sorted(keywords), ['host', 'port'])
-        # run_until_complete() was called once.  The argument isn't important.
-        self.assertTrue(self.run_until_complete.called)
-        # add_signal_handler() is called with two arguments.
-        self.assertEqual(self.add_signal_handler.call_count, 1)
-        signal_number, callback = self.add_signal_handler.call_args[0]
-        self.assertEqual(signal_number, signal.SIGINT)
-        self.assertEqual(callback, self.loop.stop)
-        # run_forever() was called once.
-        self.assertEqual(self.run_forever.call_count, 1)
-
-    def test_loop_keyboard_interrupt(self):
-        # We mock out so much of this, is it even a worthwhile test?  Well, it
-        # does give us coverage.
-        self.run_forever.side_effect = KeyboardInterrupt
-        main(('-n',))
-        # loop.run_until_complete() was still executed.
-        self.assertTrue(self.run_until_complete.called)
-
-    def test_s(self):
-        # We mock out so much of this, is it even a worthwhile test?  Well, it
-        # does give us coverage.
-        main(('-n', '-s', '3000'))
-        positional, keywords = self.create_server.call_args
-        self.assertEqual(positional[0].keywords, dict(
-            data_size_limit=3000,
-            enable_SMTPUTF8=False))
-
-    def test_size(self):
-        # We mock out so much of this, is it even a worthwhile test?  Well, it
-        # does give us coverage.
-        main(('-n', '--size', '3000'))
-        positional, keywords = self.create_server.call_args
-        self.assertEqual(positional[0].keywords, dict(
-            data_size_limit=3000,
-            enable_SMTPUTF8=False))
-
-    def test_u(self):
-        # We mock out so much of this, is it even a worthwhile test?  Well, it
-        # does give us coverage.
-        main(('-n', '-u'))
-        positional, keywords = self.create_server.call_args
-        self.assertEqual(positional[0].keywords, dict(
-            data_size_limit=None,
-            enable_SMTPUTF8=True))
-
-    def test_smtputf8(self):
-        # We mock out so much of this, is it even a worthwhile test?  Well, it
-        # does give us coverage.
-        main(('-n', '--smtputf8'))
-        positional, keywords = self.create_server.call_args
-        self.assertEqual(positional[0].keywords, dict(
-            data_size_limit=None,
-            enable_SMTPUTF8=True))
-
-
 class TestParseArgs(unittest.TestCase):
     def test_handler_from_cli(self):
         # Ignore the host:port positional argument.
@@ -334,3 +246,27 @@ class TestParseArgs(unittest.TestCase):
             parseargs(('-v',))
             self.assertEqual(cm.exception.code, 0)
         self.assertEqual(stdout.getvalue(), 'smtpd {}\n'.format(__version__))
+
+
+class TestSigint(unittest.TestCase):
+    def setUp(self):
+        default_loop = asyncio.get_event_loop()
+        self.loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(self.loop)
+        self.addCleanup(asyncio.set_event_loop, default_loop)
+
+    def test_keyboard_interrupt(self):
+        """
+        main() must close loop gracefully on Ctrl-C.
+        """
+
+        def interrupt():
+            raise KeyboardInterrupt
+        self.loop.call_later(1.5, interrupt)
+
+        try:
+            main(("-n",))
+        except Exception:
+            self.fail("main() should've closed cleanly without exceptions!")
+        else:
+            self.assertFalse(self.loop.is_running())
diff --git a/aiosmtpd/tests/test_smtp.py b/aiosmtpd/tests/test_smtp.py
index 7ed2d0f..84782b7 100644
--- a/aiosmtpd/tests/test_smtp.py
+++ b/aiosmtpd/tests/test_smtp.py
@@ -7,8 +7,15 @@ import unittest
 
 from aiosmtpd.controller import Controller
 from aiosmtpd.handlers import Sink
-from aiosmtpd.smtp import SMTP as Server, __ident__ as GREETING
-from aiosmtpd.testing.helpers import reset_connection
+from aiosmtpd.smtp import MISSING, SMTP as Server, __ident__ as GREETING
+from aiosmtpd.testing.helpers import (
+    SUPPORTED_COMMANDS_NOTLS,
+    assert_auth_invalid,
+    assert_auth_required,
+    assert_auth_success,
+    reset_connection,
+)
+from base64 import b64encode
 from contextlib import ExitStack
 from smtplib import (
     SMTP, SMTPDataError, SMTPResponseException, SMTPServerDisconnected)
@@ -18,9 +25,62 @@ CRLF = '\r\n'
 BCRLF = b'\r\n'
 
 
+def authenticator(mechanism, login, password):
+    if login and login.decode() == 'goodlogin':
+        return True
+    else:
+        return False
+
+
 class DecodingController(Controller):
     def factory(self):
-        return Server(self.handler, decode_data=True, enable_SMTPUTF8=True)
+        return Server(self.handler, decode_data=True, enable_SMTPUTF8=True,
+                      auth_require_tls=False, auth_callback=authenticator)
+
+
+class PeekerHandler:
+    def __init__(self):
+        self.session = None
+
+    async def handle_MAIL(
+            self, server, session, envelope, address, mail_options
+    ):
+        self.session = session
+        return "250 OK"
+
+    async def auth_NULL(
+            self, server, args
+    ):
+        return "NULL_login"
+
+    async def auth_DONT(
+            self, server, args
+    ):
+        return MISSING
+
+
+class PeekerAuth:
+    def __init__(self):
+        self.login = None
+        self.password = None
+
+    def authenticate(
+            self, mechanism: str, login: bytes, password: bytes
+    ) -> bool:
+        self.login = login
+        self.password = password
+        return True
+
+
+auth_peeker = PeekerAuth()
+
+
+class DecodingControllerPeekAuth(Controller):
+    def factory(self):
+        return Server(self.handler, decode_data=True, enable_SMTPUTF8=True,
+                      auth_require_tls=False,
+                      auth_callback=auth_peeker.authenticate,
+                      **self.server_kwargs)
 
 
 class NoDecodeController(Controller):
@@ -29,8 +89,17 @@ class NoDecodeController(Controller):
 
 
 class TimeoutController(Controller):
+    Delay: float = 2.0
+
+    def factory(self):
+        return Server(self.handler, timeout=self.Delay)
+
+
+class RequiredAuthDecodingController(Controller):
     def factory(self):
-        return Server(self.handler, timeout=0.1)
+        return Server(self.handler, decode_data=True, enable_SMTPUTF8=True,
+                      auth_require_tls=False, auth_callback=authenticator,
+                      auth_required=True)
 
 
 class ReceivingHandler:
@@ -243,10 +312,15 @@ class TestSMTP(unittest.TestCase):
             code, response = client.ehlo('example.com')
             self.assertEqual(code, 250)
             lines = response.splitlines()
-            self.assertEqual(lines[0], bytes(socket.getfqdn(), 'utf-8'))
-            self.assertEqual(lines[1], b'SIZE 33554432')
-            self.assertEqual(lines[2], b'SMTPUTF8')
-            self.assertEqual(lines[3], b'HELP')
+            expecteds = (
+                bytes(socket.getfqdn(), 'utf-8'),
+                b'SIZE 33554432',
+                b'SMTPUTF8',
+                b'AUTH LOGIN PLAIN',
+                b'HELP',
+            )
+            for actual, expected in zip(lines, expecteds):
+                self.assertEqual(actual, expected)
 
     def test_ehlo_duplicate(self):
         with SMTP(*self.address) as client:
@@ -305,9 +379,7 @@ class TestSMTP(unittest.TestCase):
             # Don't get tricked by smtplib processing of the response.
             code, response = client.docmd('HELP')
             self.assertEqual(code, 250)
-            self.assertEqual(response,
-                             b'Supported commands: DATA EHLO HELO HELP MAIL '
-                             b'NOOP QUIT RCPT RSET VRFY')
+            self.assertEqual(response, SUPPORTED_COMMANDS_NOTLS)
 
     def test_help_helo(self):
         with SMTP(*self.address) as client:
@@ -387,14 +459,18 @@ class TestSMTP(unittest.TestCase):
             self.assertEqual(code, 250)
             self.assertEqual(response, b'Syntax: VRFY <address>')
 
+    def test_help_auth(self):
+        with SMTP(*self.address) as client:
+            code, response = client.docmd('HELP', 'AUTH')
+            self.assertEqual(code, 250)
+            self.assertEqual(response, b'Syntax: AUTH <mechanism>')
+
     def test_help_bad_arg(self):
         with SMTP(*self.address) as client:
             # Don't get tricked by smtplib processing of the response.
             code, response = client.docmd('HELP me!')
             self.assertEqual(code, 501)
-            self.assertEqual(response,
-                             b'Supported commands: DATA EHLO HELO HELP MAIL '
-                             b'NOOP QUIT RCPT RSET VRFY')
+            self.assertEqual(response, SUPPORTED_COMMANDS_NOTLS)
 
     def test_expn(self):
         with SMTP(*self.address) as client:
@@ -695,6 +771,343 @@ class TestSMTP(unittest.TestCase):
                 response,
                 b'Error: command "FOOBAR" not recognized')
 
+    def test_auth_no_ehlo(self):
+        with SMTP(*self.address) as client:
+            code, response = client.docmd('AUTH')
+            self.assertEqual(code, 503)
+            self.assertEqual(response, b'Error: send EHLO first')
+
+    def test_auth_helo(self):
+        with SMTP(*self.address) as client:
+            client.helo('example.com')
+            code, response = client.docmd('AUTH')
+            self.assertEqual(code, 500)
+            self.assertEqual(response, b"Error: command 'AUTH' not recognized")
+
+    def test_auth_too_many_values(self):
+        with SMTP(*self.address) as client:
+            client.ehlo('example.com')
+            code, response = client.docmd('AUTH PLAIN NONE NONE')
+            self.assertEqual(code, 501)
+            self.assertEqual(response, b'Too many values')
+
+    def test_auth_not_enough_values(self):
+        with SMTP(*self.address) as client:
+            client.ehlo('example.com')
+            code, response = client.docmd('AUTH')
+            self.assertEqual(code, 501)
+            self.assertEqual(response, b'Not enough value')
+
+    def test_auth_not_supported_methods(self):
+        for method in ('GSSAPI', 'DIGEST-MD5', 'MD5', 'CRAM-MD5'):
+            with SMTP(*self.address) as client:
+                client.ehlo('example.com')
+                code, response = client.docmd('AUTH ' + method)
+                self.assertEqual(code, 504)
+                self.assertEqual(
+                    response, b'5.5.4 Unrecognized authentication type')
+
+    def test_auth_already_authenticated(self):
+        with SMTP(*self.address) as client:
+            client.ehlo('example.com')
+            code, response = client.docmd(
+                'AUTH PLAIN ' +
+                b64encode(b'\0goodlogin\0goodpasswd').decode()
+                )
+            assert_auth_success(self, code, response)
+            code, response = client.docmd('AUTH')
+            self.assertEqual(code, 503)
+            self.assertEqual(response, b'Already authenticated')
+
+    def test_auth_bad_base64_encoding(self):
+        with SMTP(*self.address) as client:
+            client.ehlo('example.com')
+            code, response = client.docmd('AUTH PLAIN not-b64')
+            self.assertEqual(code, 501)
+            self.assertEqual(response, b"5.5.2 Can't decode base64")
+
+    def test_auth_bad_base64_length(self):
+        with SMTP(*self.address) as client:
+            client.ehlo('example.com')
+            code, response = client.docmd(
+                'AUTH PLAIN ' + b64encode(b'\0onlylogin').decode())
+            self.assertEqual(code, 501)
+            self.assertEqual(response, b"5.5.2 Can't split auth value")
+
+    def test_auth_bad_credentials(self):
+        with SMTP(*self.address) as client:
+            client.ehlo('example.com')
+            code, response = client.docmd(
+                'AUTH PLAIN ' +
+                b64encode(b'\0badlogin\0badpasswd').decode()
+                )
+            assert_auth_invalid(self, code, response)
+
+    def test_auth_two_steps_good_credentials(self):
+        with SMTP(*self.address) as client:
+            client.ehlo('example.com')
+            code, response = client.docmd('AUTH PLAIN')
+            self.assertEqual(code, 334)
+            self.assertEqual(response, b'')
+            code, response = client.docmd(
+                b64encode(b'\0goodlogin\0goodpasswd').decode()
+            )
+            assert_auth_success(self, code, response)
+
+    def test_auth_two_steps_bad_credentials(self):
+        with SMTP(*self.address) as client:
+            client.ehlo('example.com')
+            code, response = client.docmd('AUTH PLAIN')
+            self.assertEqual(code, 334)
+            self.assertEqual(response, b'')
+            code, response = client.docmd(
+                b64encode(b'\0badlogin\0badpasswd').decode()
+            )
+            assert_auth_invalid(self, code, response)
+
+    def test_auth_two_steps_abort(self):
+        with SMTP(*self.address) as client:
+            client.ehlo('example.com')
+            code, response = client.docmd('AUTH PLAIN')
+            self.assertEqual(code, 334)
+            self.assertEqual(response, b'')
+            code, response = client.docmd('*')
+            self.assertEqual(code, 501)
+            self.assertEqual(response, b'Auth aborted')
+
+    def test_auth_two_steps_bad_base64_encoding(self):
+        with SMTP(*self.address) as client:
+            client.ehlo('example.com')
+            code, response = client.docmd('AUTH PLAIN')
+            self.assertEqual(code, 334)
+            code, response = client.docmd("ab@%")
+            self.assertEqual(code, 501)
+            self.assertEqual(response, b"5.5.2 Can't decode base64")
+
+    def test_auth_good_credentials(self):
+        with SMTP(*self.address) as client:
+            client.ehlo('example.com')
+            code, response = client.docmd(
+                'AUTH PLAIN ' +
+                b64encode(b'\0goodlogin\0goodpasswd').decode()
+            )
+            assert_auth_success(self, code, response)
+
+    def test_auth_no_credentials(self):
+        with SMTP(*self.address) as client:
+            client.ehlo('example.com')
+            code, response = client.docmd('AUTH PLAIN =')
+            assert_auth_invalid(self, code, response)
+
+    def test_auth_two_steps_no_credentials(self):
+        with SMTP(*self.address) as client:
+            client.ehlo('example.com')
+            code, response = client.docmd('AUTH PLAIN')
+            self.assertEqual(code, 334)
+            self.assertEqual(response, b'')
+            code, response = client.docmd('=')
+            assert_auth_invalid(self, code, response)
+
+    def test_auth_login_multisteps_no_credentials(self):
+        with SMTP(*self.address) as client:
+            client.ehlo("example.com")
+            code, response = client.docmd("AUTH LOGIN")
+            self.assertEqual(code, 334)
+            self.assertEqual(response, b"VXNlciBOYW1lAA==")
+            code, response = client.docmd('=')
+            self.assertEqual(code, 334)
+            self.assertEqual(response, b"UGFzc3dvcmQA")
+            code, response = client.docmd('=')
+            assert_auth_invalid(self, code, response)
+
+
+class TestSMTPAuth(unittest.TestCase):
+    def setUp(self):
+        self.handler = PeekerHandler()
+        controller = DecodingControllerPeekAuth(
+            self.handler, server_kwargs={"auth_exclude_mechanism": ["DONT"]}
+        )
+        controller.start()
+        self.addCleanup(controller.stop)
+        self.address = (controller.hostname, controller.port)
+
+    def test_ehlo(self):
+        with SMTP(*self.address) as client:
+            code, response = client.ehlo('example.com')
+            self.assertEqual(code, 250)
+            lines = response.splitlines()
+            expecteds = (
+                bytes(socket.getfqdn(), 'utf-8'),
+                b'SIZE 33554432',
+                b'SMTPUTF8',
+                b'AUTH LOGIN NULL PLAIN',
+                b'HELP',
+            )
+            for actual, expected in zip(lines, expecteds):
+                self.assertEqual(actual, expected)
+
+    def test_auth_plain_null_credential(self):
+        with SMTP(*self.address) as client:
+            client.ehlo("example.com")
+            code, response = client.docmd("AUTH PLAIN")
+            self.assertEqual(code, 334)
+            self.assertEqual(response, b"")
+            code, response = client.docmd('=')
+            assert_auth_success(self, code, response)
+            self.assertEqual(auth_peeker.login, None)
+            self.assertEqual(auth_peeker.password, None)
+            code, response = client.mail("alice@example.com")
+            self.assertEqual(self.handler.session.login_data, b"")
+
+    def test_auth_login_null_credential(self):
+        with SMTP(*self.address) as client:
+            client.ehlo("example.com")
+            code, response = client.docmd("AUTH LOGIN")
+            self.assertEqual(code, 334)
+            self.assertEqual(response, b"VXNlciBOYW1lAA==")
+            code, response = client.docmd('=')
+            self.assertEqual(code, 334)
+            self.assertEqual(response, b"UGFzc3dvcmQA")
+            code, response = client.docmd('=')
+            assert_auth_success(self, code, response)
+            self.assertEqual(auth_peeker.login, None)
+            self.assertEqual(auth_peeker.password, None)
+            code, response = client.mail("alice@example.com")
+            self.assertEqual(self.handler.session.login_data, b"")
+
+    def test_auth_login_abort_login(self):
+        with SMTP(*self.address) as client:
+            client.ehlo("example.com")
+            code, response = client.docmd("AUTH LOGIN")
+            self.assertEqual(code, 334)
+            self.assertEqual(response, b"VXNlciBOYW1lAA==")
+            code, response = client.docmd('*')
+            self.assertEqual(code, 501)
+            self.assertEqual(response, b"Auth aborted")
+
+    def test_auth_login_abort_password(self):
+        auth_peeker.return_val = False
+        with SMTP(*self.address) as client:
+            client.ehlo("example.com")
+            code, response = client.docmd("AUTH LOGIN")
+            self.assertEqual(code, 334)
+            self.assertEqual(response, b"VXNlciBOYW1lAA==")
+            code, response = client.docmd('=')
+            self.assertEqual(code, 334)
+            self.assertEqual(response, b"UGFzc3dvcmQA")
+            code, response = client.docmd('*')
+            self.assertEqual(code, 501)
+            self.assertEqual(response, b"Auth aborted")
+
+    def test_auth_custom_mechanism(self):
+        auth_peeker.return_val = False
+        with SMTP(*self.address) as client:
+            client.ehlo("example.com")
+            code, response = client.docmd("AUTH NULL")
+            assert_auth_success(self, code, response)
+
+    def test_auth_disabled_mechanism(self):
+        with SMTP(*self.address) as client:
+            client.ehlo("example.com")
+            code, response = client.docmd("AUTH DONT")
+            self.assertEqual(code, 504)
+            self.assertEqual(response,
+                             b"5.5.4 Unrecognized authentication type")
+
+
+class TestRequiredAuthentication(unittest.TestCase):
+    def setUp(self):
+        controller = RequiredAuthDecodingController(Sink)
+        controller.start()
+        self.addCleanup(controller.stop)
+        self.address = (controller.hostname, controller.port)
+
+    def test_help_unauthenticated(self):
+        with SMTP(*self.address) as client:
+            code, response = client.docmd('HELP')
+            assert_auth_required(self, code, response)
+
+    def test_vrfy_unauthenticated(self):
+        with SMTP(*self.address) as client:
+            code, response = client.docmd('VRFY <anne@example.com>')
+            assert_auth_required(self, code, response)
+
+    def test_mail_unauthenticated(self):
+        with SMTP(*self.address) as client:
+            client.ehlo('example.com')
+            code, response = client.docmd('MAIL FROM: <anne@example.com>')
+            assert_auth_required(self, code, response)
+
+    def test_rcpt_unauthenticated(self):
+        with SMTP(*self.address) as client:
+            client.ehlo('example.com')
+            code, response = client.docmd('RCPT TO: <anne@example.com>')
+            assert_auth_required(self, code, response)
+
+    def test_data_unauthenticated(self):
+        with SMTP(*self.address) as client:
+            client.ehlo('example.com')
+            code, response = client.docmd('DATA')
+            assert_auth_required(self, code, response)
+
+    def test_help_authenticated(self):
+        with SMTP(*self.address) as client:
+            client.ehlo('example.com')
+            code, response = client.docmd(
+                'AUTH PLAIN ' +
+                b64encode(b'\0goodlogin\0goodpasswd').decode()
+            )
+            assert_auth_success(self, code, response)
+            code, response = client.docmd('HELP')
+            self.assertEqual(code, 250)
+            self.assertEqual(response, SUPPORTED_COMMANDS_NOTLS)
+
+    def test_vrfy_authenticated(self):
+        with SMTP(*self.address) as client:
+            client.ehlo('example.com')
+            code, response = client.docmd(
+                'AUTH PLAIN ' +
+                b64encode(b'\0goodlogin\0goodpasswd').decode()
+            )
+            assert_auth_success(self, 235, response)
+            code, response = client.docmd('VRFY <anne@example.com>')
+            self.assertEqual(code, 252)
+            self.assertEqual(
+                response,
+                b'Cannot VRFY user, but will accept message and '
+                b'attempt delivery'
+            )
+
+    def test_mail_authenticated(self):
+        with SMTP(*self.address) as client:
+            client.ehlo('example.com')
+            code, response = client.docmd(
+                'AUTH PLAIN ' +
+                b64encode(b'\0goodlogin\0goodpasswd').decode()
+            )
+            assert_auth_success(self, code, response)
+            code, response = client.docmd('MAIL FROM: <anne@example.com>')
+            self.assertEqual(code, 250)
+            self.assertEqual(response, b'OK')
+
+    def test_rcpt_authenticated(self):
+        with SMTP(*self.address) as client:
+            client.ehlo('example.com')
+            code, response = client.docmd(
+                'AUTH PLAIN ' +
+                b64encode(b'\0goodlogin\0goodpasswd').decode()
+            )
+            assert_auth_success(self, code, response)
+            code, response = client.docmd('RCPT TO: <anne@example.com>')
+            self.assertEqual(code, 503)
+            self.assertEqual(response, b'Error: need MAIL command')
+
+    def test_data_authenticated(self):
+        with SMTP(*self.address) as client:
+            client.ehlo('example.com')
+            code, response = client.docmd('DATA')
+            assert_auth_required(self, code, response)
+
 
 class TestResetCommands(unittest.TestCase):
     """Test that sender and recipients are reset on RSET, HELO, and EHLO.
@@ -1188,5 +1601,5 @@ class TestTimeout(unittest.TestCase):
     def test_timeout(self):
         with SMTP(*self.address) as client:
             code, response = client.ehlo('example.com')
-            time.sleep(0.3)
+            time.sleep(0.1 + TimeoutController.Delay)
             self.assertRaises(SMTPServerDisconnected, client.getreply)
diff --git a/aiosmtpd/tests/test_starttls.py b/aiosmtpd/tests/test_starttls.py
index f7c1fde..4a53cfd 100644
--- a/aiosmtpd/tests/test_starttls.py
+++ b/aiosmtpd/tests/test_starttls.py
@@ -5,6 +5,10 @@ import pkg_resources
 from aiosmtpd.controller import Controller as BaseController
 from aiosmtpd.handlers import Sink
 from aiosmtpd.smtp import SMTP as SMTPProtocol
+from aiosmtpd.testing.helpers import (
+    SUPPORTED_COMMANDS_TLS,
+    assert_auth_invalid,
+)
 from email.mime.text import MIMEText
 from smtplib import SMTP
 
@@ -49,6 +53,15 @@ class TLSController(Controller):
             tls_context=get_tls_context())
 
 
+class RequireTLSAuthDecodingController(Controller):
+    def factory(self):
+        return SMTPProtocol(
+            self.handler,
+            decode_data=True,
+            auth_require_tls=True,
+            tls_context=get_tls_context())
+
+
 class HandshakeFailingHandler:
     def handle_STARTTLS(self, server, session, envelope):
         return False
@@ -111,9 +124,7 @@ class TestStartTLS(unittest.TestCase):
             # Don't get tricked by smtplib processing of the response.
             code, response = client.docmd('HELP')
             self.assertEqual(code, 250)
-            self.assertEqual(response,
-                             b'Supported commands: DATA EHLO HELO HELP MAIL '
-                             b'NOOP QUIT RCPT RSET STARTTLS VRFY')
+            self.assertEqual(response, SUPPORTED_COMMANDS_TLS)
 
 
 class TestTLSForgetsSessionData(unittest.TestCase):
@@ -199,3 +210,27 @@ class TestRequireTLS(unittest.TestCase):
             client.ehlo('example.com')
             code, response = client.docmd('DATA')
             self.assertEqual(code, 530)
+
+
+class TestRequireTLSAUTH(unittest.TestCase):
+    def setUp(self):
+        controller = RequireTLSAuthDecodingController(Sink)
+        controller.start()
+        self.addCleanup(controller.stop)
+        self.address = (controller.hostname, controller.port)
+
+    def test_auth_notls(self):
+        with SMTP(*self.address) as client:
+            client.ehlo('example.com')
+            code, response = client.docmd("AUTH ")
+            self.assertEqual(code, 538)
+            self.assertEqual(response,
+                             b"5.7.11 Encryption required for requested "
+                             b"authentication mechanism")
+
+    def test_auth_tls(self):
+        with SMTP(*self.address) as client:
+            client.starttls()
+            client.ehlo('example.com')
+            code, response = client.docmd('AUTH PLAIN AHRlc3QAdGVzdA==')
+            assert_auth_invalid(self, code, response)