Sanitize AUTH Logging (#264)

* Sanitize AUTH logging
  By overriding the standard __repr__ of AuthResult and LoginPassword
* Test for password leaks
* Bump version to 1.4.2a2 and update NEWS.rst

aiosmtpd/__init__.py

@@ -1,4 +1,4 @@
 # Copyright 2014-2021 The aiosmtpd Developers
 # SPDX-License-Identifier: Apache-2.0
 
-__version__ = "1.4.2a1"
+__version__ = "1.4.2a2"

aiosmtpd/docs/NEWS.rst

@@ -16,7 +16,8 @@ Fixed/Improved
   (See #262)
 * Timeout messages in ``Controller.start()`` gets more details and a mention about the
   ``ready_timeout`` parameter. (See #262)
-
+* Prevent sensitive AUTH information leak by sanitizing the repr()
+  of AuthResult and LoginPassword.
 
 
 1.4.1 (2021-03-04)

aiosmtpd/smtp.py

@@ -11,8 +11,6 @@
 import re
 import socket
 import ssl
-from aiosmtpd import __version__
-from aiosmtpd.proxy_protocol import get_proxy, ProxyData
 from base64 import b64decode, b64encode
 from email._header_value_parser import get_addr_spec, get_angle_addr
 from email.errors import HeaderParseError
@@ -34,6 +32,9 @@
 import attr
 from public import public
 
+from aiosmtpd import __version__
+from aiosmtpd.proxy_protocol import ProxyData, get_proxy
+
 
 # region #### Custom Data Types #######################################################
 
@@ -121,9 +122,12 @@ class AuthResult:
     """
 
     message: Optional[str] = attr.ib(kw_only=True, default=None)
-    """Optional message for additional handling by smtp_AUTH"""
+    """
+    Optional message for additional handling by smtp_AUTH.
+    Applicable only if handled == False.
+    """
 
-    auth_data: Optional[Any] = attr.ib(kw_only=True, default=None)
+    auth_data: Optional[Any] = attr.ib(kw_only=True, default=None, repr=lambda x: "...")
     """
     Optional free-form authentication data. For the built-in mechanisms, it is usually
     an instance of LoginPassword. Other implementations are free to use any data
@@ -136,6 +140,12 @@ class LoginPassword(NamedTuple):
     login: bytes
     password: bytes
 
+    def __str__(self) -> str:
+        return f"LoginPassword(login='{self.login}', password=...)"
+
+    def __repr__(self) -> str:
+        return str(self)
+
 
 @public
 class Session:
@@ -915,14 +925,15 @@ async def smtp_AUTH(self, arg: str) -> None:
         status = await self._call_handler_hook('AUTH', args)
         if status is MISSING:
             auth_method = self._auth_methods[mechanism]
-            if auth_method.is_builtin:
-                log.debug(f"Using builtin auth_ hook for {mechanism}")
-            else:
-                log.debug(f"Using handler auth_ hook for {mechanism}")
+            log.debug(
+                "Using %s auth_ hook for %r",
+                "builtin" if auth_method.is_builtin else "handler",
+                mechanism
+            )
             # Pass 'self' to method so external methods can leverage this
             # class's helper methods such as push()
             auth_result = await auth_method.method(self, args)
-            log.debug(f"auth_{mechanism} returned {auth_result}")
+            log.debug("auth_%s returned %r", mechanism, auth_result)
 
             # New system using `authenticator` and AuthResult
             if isinstance(auth_result, AuthResult):
@@ -952,7 +963,6 @@ async def smtp_AUTH(self, arg: str) -> None:
                 # is rejected / not valid
                 status = CODE_INVALID
             else:
-
                 self.session.login_data = auth_result
                 status = CODE_SUCCESS
 

aiosmtpd/testing/statuscodes.py

@@ -13,19 +13,21 @@ def __call__(self, *args: bytes) -> "StatusCode":
         nmsg = self.mesg % args
         return StatusCode(self.code, nmsg)
 
-    def to_bytes(self) -> bytes:
+    def to_bytes(self, crlf: bool = False) -> bytes:
         """
         Returns code + mesg as bytes.
         WARNING: This is NOT identical to __str()__.encode()!
         """
-        return str(self.code).encode() + b" " + self.mesg
+        _crlf = b"\r\n" if crlf else b""
+        return str(self.code).encode() + b" " + self.mesg + _crlf
 
-    def to_str(self) -> str:
+    def to_str(self, crlf: bool = False) -> str:
         """
         Returns code + mesg as a string.
         WARNING: This is NOT identical to __str__()!
         """
-        return str(self.code) + " " + self.mesg.decode()
+        _crlf = "\r\n" if crlf else ""
+        return str(self.code) + " " + self.mesg.decode() + _crlf
 
 
 _COMMON_COMMANDS = [