Email Verification

[amitsingh07s]

what else can done to reach every domain

@app.route('/<string:email>',methods=['GET','POST'])
def post(email):
    blocked_keywords = ["spamhaus",
            "proofpoint",
            "cloudmark",
            "banned",
            "blacklisted",
            "blocked",
            "block list",
            "denied"]
 
 
 
    proxy = {
        'socks4': socks.SOCKS4,
        'socks5': socks.SOCKS5,
        'http': socks.HTTP # does not guareentee it will work with HTTP
    }
 
    class UnknownProxyError(Exception):
        def __init__(self, proxy_type):
            self.msg = f"The proxy type {proxy_type} is not known\n Try one of socks4, socks5 or http"
 
    class EmailFormatError(Exception):
 
        def __init__(self, msg):
            self.msg = msg
 
    class SMTPRecepientException(Exception): # don't cover
 
        def __init__(self, code, response):
            self.code = code
            self.response = response
 
 
    ####
    # SMTP RCPT error handlers
    ####
    def handle_550(response):
        if any([keyword.encode() in response for keyword in blocked_keywords]):
            return dict(message="Blocked by mail server- couldn't verify", deliverable=False, host_exists=True)
        else:
            return dict(deliverable=False, host_exists=True)
 
 
 
 
    handle_error = {
        # 250 and 251 are not errors
        550: handle_550,
        551: lambda _: dict(deliverable=False, host_exists=True),
        552: lambda _: dict(deliverable=True, host_exists=True, full_inbox=True),
        553: lambda _: dict(deliverable=False, host_exists=True),
        450: lambda _: dict(deliverable=False, host_exists=True),
        451: lambda _: dict(deliverable=False, message="Local error processing, try again later."),
        452: lambda _: dict(deliverable=True, full_inbox=True),
        # Syntax errors
        # 500 (command not recognised)
        # 501 (parameter/argument not recognised)
        # 503 (bad command sequence)
        521: lambda _: dict(deliverable=False, host_exists=False),
        421: lambda _: dict(deliverable=False, host_exists=True, message="Service not available, try again later."),
        441: lambda _: dict(deliverable=True, full_inbox=True, host_exists=True)
    }
 
    handle_unrecognised = lambda a: dict(message=f"Unrecognised error: {a}", deliverable=False)
 
 
    # create a namedtuple to hold the email address
    Address = namedtuple("Address", ["name", "addr", "username", "domain"])
 
    class e_verifier:
 
        def __init__(self,
                    source_addr,
                    proxy_type = None,
                    proxy_addr = None,
                    proxy_port = None,
                    proxy_username = None ,
                    proxy_password = None):
 
 
            if proxy_type:
                try:
                    self.proxy_type = proxy[proxy_type.lower()]
                except KeyError as e:
                    raise UnknownProxyError(proxy_type)
            else:
                self.proxy_type = None
            self.source_addr = source_addr
            self.proxy_addr = proxy_addr
            self.proxy_port = proxy_port
            self.proxy_username = proxy_username
            self.proxy_password = proxy_password
 
        def _parse_address(self, email) -> Address:
 
            name, addr = parseaddr(email)
            if not addr:
                raise EmailFormatError(f"email does not contain address: {email}")
            try:
                domain = addr.split('@')[-1]
                username = addr.split('@')[:-1][0]
            except IndexError:
                raise EmailFormatError(f"address provided is invalid: {email}")
            return Address(name, addr, username, domain)
 
        def _random_email(self, domain):
 
            return f'{binascii.hexlify(os.urandom(20)).decode()}@{domain}'
 
        def _can_deliver(self,
                        exchange : str,
                        address : str):
 
            host_exists = False
            with SMTP(exchange[1],
                    proxy_type=self.proxy_type,
                    proxy_addr=self.proxy_addr,
                    proxy_port=self.proxy_port,
                    proxy_username=self.proxy_username,
                    proxy_password=self.proxy_password) as smtp:
                host_exists = True
                smtp.helo()
                smtp.mail(self.source_addr)
                test_resp = smtp.rcpt(address.addr)
                catch_all_resp = smtp.rcpt(self._random_email(address.domain))
                if test_resp[0] == 250:
                    deliverable = True
                    if catch_all_resp[0] == 250:
                        catch_all = True
                    else:
                        catch_all = False
                elif test_resp[0] >= 400:
                    raise SMTPRecepientException(*test_resp)
            return host_exists, deliverable, catch_all
 
        def verify(self, email):
 
            lookup = {
                'address': None,
                'valid_format': False,
                'deliverable': False,
                'full_inbox': False,
                'host_exists': False,
                'catch_all': False,
            }
            try:
                lookup['address'] = self._parse_address(email)
                lookup['valid_format'] = True
            except EmailFormatError:
                lookup['address'] = f"{email}"
                return lookup
 
            # look for mx record and create a list of mail exchanges
            try:
                mx_record = resolver.query(lookup['address'].domain, 'MX')
                mail_exchangers = [exchange.to_text().split() for exchange in mx_record]
                lookup['host_exists'] = True
            except (resolver.NoAnswer, resolver.NXDOMAIN, resolver.NoNameservers):
                lookup['host_exists'] = False
                return lookup
 
            for exchange in mail_exchangers:
                try:
                    host_exists, deliverable, catch_all = self._can_deliver(exchange, lookup['address'])
                    if deliverable:
                        lookup['host_exists'] = host_exists
                        lookup['deliverable'] = deliverable
                        lookup['catch_all'] = catch_all
                        break
                except SMTPRecepientException as err:
                    # Error handlers return a dict that is then merged with 'lookup'
                    kwargs = handle_error.get(err.code, handle_unrecognised)(err.response)
                    # This expression merges the lookup dict with kwargs
                    lookup = {**lookup, **kwargs}
 
                except smtplib.SMTPServerDisconnected as err:
                    lookup['message'] = "Internal Error"
                except smtplib.SMTPConnectError as err:
 
                    lookup['message'] = "Internal Error. Maybe blacklisted"
 
            return lookup
 
    v = e_verifier(source_addr='[e-mail]')
    email = email
    l = v.verify(email)
    return l  
 
if __name__ == "__main__":
    app.run(debug=True)