Updating to 0.2.10, see CHANGELOG.txt

CHANGELOG.txt

@@ -1,3 +1,9 @@
+0.2.10:
+- Adding STARTTLS handling
+- Use EHLO instead of HELO
+- Refactorings to handle errors during EHLO and MAIL FROM commands
+- Updated dependencies
+
 0.2.9:
 - Adding debug command to validate_email for debugging
 

README.rst

@@ -45,6 +45,8 @@ Basic usage::
 
 The function :code:`validate_email_or_fail()` works exactly like :code:`validate_email`, except that it raises an exception in the case of validation failure instead of returning :code:`False`.
 
+The module will try to negotiate a TLS connection with STARTTLS, and silently fall back to an unencrypted SMTP connection if the server doesn't support it.
+
 Auto-updater
 ============================
 The package contains an auto-updater for downloading and updating the built-in blacklist.txt. It will run on each module load (and installation), but will try to update the content only if the file is older than 5 days, and if the content is not the same that's already downloaded.

setup.py

@@ -56,9 +56,9 @@ class BuildPyCommand(build_py):
 
 setup(
     name='py3-validate-email',
-    version='0.2.9',
+    version='0.2.10',
     packages=find_packages(exclude=['tests']),
-    install_requires=['dnspython~=1.16', 'idna~=2.8', 'filelock~=3.0'],
+    install_requires=['dnspython~=2.0', 'idna~=2.10', 'filelock~=3.0'],
     author='László Károlyi',
     author_email='laszlo@karolyi.hu',
     description=(

validate_email/mx_check.py

@@ -1,7 +1,7 @@
-from smtplib import SMTP, SMTPServerDisconnected
+from smtplib import SMTP, SMTPNotSupportedError, SMTPServerDisconnected
 from socket import error as SocketError
 from socket import gethostname
-from typing import Optional
+from typing import Optional, Tuple
 
 from dns.exception import Timeout
 from dns.rdatatype import MX as rdtype_mx
@@ -16,6 +16,10 @@ from .exceptions import (
     DomainNotFoundError, NoMXError, NoNameserverError, NoValidMXError)
 
 
+class ProtocolError(Exception):
+    'Raised when there is an error during the SMTP conversation.'
+
+
 def _get_mx_records(domain: str, timeout: int) -> list:
     """
     Return a list of hostnames in the MX record, raise an exception on
@@ -44,32 +48,82 @@ def _get_mx_records(domain: str, timeout: int) -> list:
     return result
 
 
-def _check_one_mx(
+def _smtp_ehlo_tls(smtp: SMTP, helo_host: str):
-        smtp: SMTP, error_messages: list, mx_record: str, helo_host: str,
-        from_address: EmailAddress, email_address: EmailAddress) -> bool:
     """
-    Check one MX server, return the `is_ambigious` boolean or raise
+    Try and start the TLS session, fall back to unencrypted when
-    `StopIteration` if this MX accepts the email.
+    unavailable.
     """
+    code, message = smtp.ehlo(name=helo_host)
+    if code >= 300:
+        # EHLO bails out, no further SMTP commands are acceptable
+        message = message.decode(errors='ignore')
+        raise ProtocolError(f'EHLO failed: {message}')
     try:
-        smtp.connect(host=mx_record)
+        smtp.starttls()
-        smtp.helo(name=helo_host)
+        code, message = smtp.ehlo(name=helo_host)
-        smtp.mail(sender=from_address.ace)
+    except SMTPNotSupportedError as exc:
-        code, message = smtp.rcpt(recip=email_address.ace)
+        print('XXX', exc)
-        smtp.quit()
+        # The server does not support the STARTTLS extension
-    except SMTPServerDisconnected:
+        pass
-        return True
+    except RuntimeError:
-    except SocketError as error:
+        # SSL/TLS support is not available to your Python interpreter
-        error_messages.append(f'{mx_record}: {error}')
+        pass
-        return False
+
+
+def _smtp_mail(smtp: SMTP, from_address: EmailAddress):
+    'Send and evaluate the `MAIL FROM` command.'
+    code, message = smtp.mail(sender=from_address.ace)
+    if code >= 300:
+        # RCPT TO bails out, no further SMTP commands are acceptable
+        message = message.decode(errors='ignore')
+        raise ProtocolError(f'RCPT TO failed: {message}')
+
+
+def _smtp_converse(
+    mx_record: str, smtp_timeout: int, debug: bool, helo_host: str,
+    from_address: EmailAddress, email_address: EmailAddress
+) -> Optional[Tuple[int, str]]:
+    """
+    Do the `SMTP` conversation, handle errors in the caller.
+
+    Return a `tuple(code, message)` when ambigious, or raise
+    `StopIteration` if the conversation points out an existing email.
+    """
+    smtp = SMTP(timeout=smtp_timeout, host=mx_record)
+    smtp.set_debuglevel(debuglevel=2 if debug else False)
+    _smtp_ehlo_tls(smtp=smtp, helo_host=helo_host)
+    _smtp_mail(smtp=smtp, from_address=from_address)
+    code, message = smtp.rcpt(recip=email_address.ace)
+    smtp.quit()
     if code == 250:
         raise StopIteration
     elif 400 <= code <= 499:
         # Ambigious return code, can be graylist, temporary problems,
         # quota or mailsystem error
+        return code, message.decode(errors='ignore')
+
+
+def _check_one_mx(
+        error_messages: list, mx_record: str, helo_host: str,
+        from_address: EmailAddress, email_address: EmailAddress,
+        smtp_timeout: int, debug: bool) -> bool:
+    """
+    Check one MX server, return the `is_ambigious` boolean or raise
+    `StopIteration` if this MX accepts the email.
+    """
+    try:
+        result = _smtp_converse(
+            mx_record=mx_record, smtp_timeout=smtp_timeout, debug=debug,
+            helo_host=helo_host, from_address=from_address,
+            email_address=email_address)
+    except SMTPServerDisconnected:
+        return True
+    except (SocketError, ProtocolError) as error:
+        error_messages.append(f'{mx_record}: {error}')
+        return False
+    if result:
+        error_messages.append(f'{mx_record}: {result[0]} {result[1]}')
         return True
-    message = message.decode(errors='ignore')
-    error_messages.append(f'{mx_record}: {code} {message}')
     return False
 
 
@@ -79,16 +133,15 @@ def _check_mx_records(
     debug: bool,
 ) -> Optional[bool]:
     'Check the mx records for a given email address.'
-    smtp = SMTP(timeout=smtp_timeout)
-    smtp.set_debuglevel(debuglevel=2 if debug else False)
     error_messages = []
     found_ambigious = False
     for mx_record in mx_records:
         try:
             found_ambigious |= _check_one_mx(
-                smtp=smtp, error_messages=error_messages, mx_record=mx_record,
+                error_messages=error_messages, mx_record=mx_record,
                 helo_host=helo_host, from_address=from_address,
-                email_address=email_address)
+                email_address=email_address, smtp_timeout=smtp_timeout,
+                debug=debug)
         except StopIteration:
             return True
     # If any of the mx servers behaved ambigious, return None, otherwise raise