Add a TLD check to validate_address

flanker/addresslib/address.py

@@ -42,6 +42,7 @@
 from ply.lex import LexError
 from ply.yacc import YaccError
 from six.moves.urllib_parse import urlparse
+from tld import get_tld
 
 from flanker.addresslib.lexer import lexer
 from flanker.addresslib.parser import (Mailbox, Url, mailbox_parser,
@@ -50,7 +51,7 @@
                                        addr_spec_parser, url_parser)
 from flanker.addresslib.quote import smart_unquote, smart_quote
 from flanker.addresslib.validate import (mail_exchanger_lookup,
-                                         preparse_address, plugin_for_esp)
+                                         plugin_for_esp)
 from flanker.mime.message.headers.encodedword import mime_to_unicode
 from flanker.mime.message.headers.encoding import encode_string
 from flanker.utils import is_pure_ascii, metrics_wrapper
@@ -256,7 +257,7 @@ def parse_list(address_list, strict=False, as_tuple=False, metrics=False):
 
 
 @metrics_wrapper()
-def validate_address(addr_spec, metrics=False):
+def validate_address(addr_spec, metrics=False, skip_remote_checks=False):
     """
     Given an addr-spec, runs the pre-parser, the parser, DNS MX checks,
     MX existence checks, and if available, ESP specific grammar for the
@@ -276,6 +277,7 @@ def validate_address(addr_spec, metrics=False):
         user.1234@gmail.com
     """
     mtimes = {'parsing': 0,
+              'tld_lookup': 0,
               'mx_lookup': 0,
               'dns_lookup': 0,
               'mx_conn':0 ,
@@ -284,23 +286,30 @@ def validate_address(addr_spec, metrics=False):
     # sanity check
     if addr_spec is None:
         return None, mtimes
-
-    # preparse address into its parts and perform any ESP specific pre-parsing
-    addr_parts = preparse_address(addr_spec)
-    if addr_parts is None:
-        _log.warning('failed preparse check for %s', addr_spec)
+    if '@' not in addr_spec:
         return None, mtimes
 
     # run parser against address
     bstart = time()
-    paddr = parse('@'.join(addr_parts), addr_spec_only=True, strict=True)
+    paddr = parse(addr_spec, addr_spec_only=True, strict=True)
     mtimes['parsing'] = time() - bstart
     if paddr is None:
         _log.warning('failed parse check for %s', addr_spec)
         return None, mtimes
 
+    # lookup the TLD
+    bstart = time()
+    tld = get_tld(paddr.hostname, fail_silently=True, fix_protocol=True)
+    mtimes['tld_lookup'] = time() - bstart
+    if tld is None:
+        _log.warning('failed tld check for %s', addr_spec)
+        return None, mtimes
+
+    if skip_remote_checks:
+        return paddr, mtimes
+
     # lookup if this domain has a mail exchanger
-    exchanger, mx_metrics = mail_exchanger_lookup(addr_parts[-1], metrics=True)
+    exchanger, mx_metrics = mail_exchanger_lookup(paddr.hostname, metrics=True)
     mtimes['mx_lookup'] = mx_metrics['mx_lookup']
     mtimes['dns_lookup'] = mx_metrics['dns_lookup']
     mtimes['mx_conn'] = mx_metrics['mx_conn']

flanker/addresslib/validate.py

@@ -87,7 +87,6 @@ def suggest_alternate(addr_spec):
     if addr_spec is None:
         return None
 
-    # preparse address into its parts and perform any ESP specific preparsing
     addr_parts = preparse_address(addr_spec)
     if addr_parts is None:
         return None

setup.py

@@ -28,6 +28,7 @@
           'ply>=3.10',
           'regex>=0.1.20110315',
           'six',
+          'tld',
           'WebOb>=0.9.8'],
       extras_require={
           'validator': [

tests/addresslib/validator_test.py

@@ -334,3 +334,10 @@ def test_mx_aol_manage_flag_toggle():
     addr_obj = address.parse(mailbox)
     unmanaged = validate.aol.unmanaged_email(addr_obj.hostname)
     assert_equal(unmanaged, True)
+
+def test_bad_tld():
+    # con is not a valid TLD, we get a lot of messages to some.user@gmail.con
+    mailbox = 'test@example.con'
+    addr_obj, metrics = address.validate_address(mailbox, skip_remote_checks=True, metrics=True)
+    assert_equal(addr_obj, None)
+    assert_not_equal(metrics['tld_lookup'], 0)