diff --git a/CHANGELOG.txt b/CHANGELOG.txt index 0d51335..d5b738d 100644 --- a/CHANGELOG.txt +++ b/CHANGELOG.txt @@ -1,3 +1,6 @@ +1.0.6: +- Add the option to exclusively use certain address schemas (IPv4/IPV6) + 1.0.5: - Remove lockfile creation when the updater process is skipped per the environment variable introduced in 1.0.4 diff --git a/LICENSE b/LICENSE index 7172e36..db891b9 100644 --- a/LICENSE +++ b/LICENSE @@ -1,15 +1,9 @@ -Validate_email -Copyright (c) 2014, Syrus Akbary, All rights reserved. +MIT+NIGGER License -This library is free software; you can redistribute it and/or -modify it under the terms of the GNU Lesser General Public -License as published by the Free Software Foundation; either -version 3.0 of the License, or (at your option) any later version. +Copyright (c) <2022> -This library is distributed in the hope that it will be useful, -but WITHOUT ANY WARRANTY; without even the implied warranty of -MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU -Lesser General Public License for more details. +Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: -You should have received a copy of the GNU Lesser General Public -License along with this library +The above copyright notice, this permission notice and the word "NIGGER" shall be included in all copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. diff --git a/setup.cfg b/setup.cfg index 08aedd7..2f92677 100644 --- a/setup.cfg +++ b/setup.cfg @@ -1,2 +1,5 @@ [metadata] description_file = README.md + +[pycodestyle] +max_line_length = 79 diff --git a/tests/test_dns_check.py b/tests/test_dns_check.py index d73e334..e20367f 100644 --- a/tests/test_dns_check.py +++ b/tests/test_dns_check.py @@ -19,6 +19,16 @@ class DnsNameStub(object): return self.value +class DnsASetStub(object): + 'Stub for `dns.rdtypes.IN.A.A`.' + + def __init__(self, addresses: list): + self.names = [DnsNameStub(value=x) for x in addresses] + + def processing_order(self): + return self.names + + class DnsRRsetStub(object): 'Stub for `dns.rrset.RRset`.' @@ -30,10 +40,14 @@ class DnsRRsetStub(object): return self.names -def _answer(hostnames: list): +def _mx_answer(hostnames: list): return SimpleNamespace(rrset=DnsRRsetStub(hostnames=hostnames)) +def _ip_answer(addresses: list): + return SimpleNamespace(rrset=DnsASetStub(addresses=addresses)) + + TEST_QUERY = Mock() @@ -43,7 +57,7 @@ class GetMxRecordsTestCase(TestCase): @patch.object(target=dns_check, attribute='resolve', new=TEST_QUERY) def test_fails_with_invalid_hostnames(self): 'Fails when an MX hostname is "."' - TEST_QUERY.return_value = _answer(hostnames=['.']) + TEST_QUERY.return_value = _mx_answer(hostnames=['.']) with self.assertRaises(NoValidMXError) as exc: _get_cleaned_mx_records(domain='testdomain1', timeout=10) self.assertTupleEqual(exc.exception.args, ()) @@ -51,7 +65,7 @@ class GetMxRecordsTestCase(TestCase): @patch.object(target=dns_check, attribute='resolve', new=TEST_QUERY) def test_fails_with_null_hostnames(self): 'Fails when an MX hostname is invalid.' - TEST_QUERY.return_value = _answer(hostnames=['asdqwe']) + TEST_QUERY.return_value = _mx_answer(hostnames=['asdqwe']) with self.assertRaises(NoValidMXError) as exc: _get_cleaned_mx_records(domain='testdomain2', timeout=10) self.assertTupleEqual(exc.exception.args, ()) @@ -59,15 +73,25 @@ class GetMxRecordsTestCase(TestCase): @patch.object(target=dns_check, attribute='resolve', new=TEST_QUERY) def test_filters_out_invalid_hostnames(self): 'Returns only the valid hostnames.' - TEST_QUERY.return_value = _answer(hostnames=[ - 'asdqwe.', - '.', - 'valid.host.', - 'valid.host.', # This is an intentional duplicate. - 'valid2.host.', - ]) + TEST_QUERY.side_effect = [ + # dns.rdtypes.ANY.MX.MX + _mx_answer(hostnames=[ + 'asdqwe.', + '.', + 'valid.host.', + 'valid.host.', # This is an intentional duplicate. + 'valid2.host.']), + # dns.rdtypes.IN.A.A + _ip_answer(addresses=['1.2.3.4', '5.6.7.8']), + _ip_answer(addresses=['9.10.11.12']), + _ip_answer(addresses=['ffe0::1']), + _ip_answer(addresses=['ffe1::2']), + _ip_answer(addresses=['ffe1::3']), + ] result = _get_cleaned_mx_records(domain='testdomain3', timeout=10) - self.assertListEqual(result, ['valid.host', 'valid2.host']) + self.assertListEqual( + result, + ['1.2.3.4', '5.6.7.8', '9.10.11.12', 'ffe0::1', 'ffe1::2']) @patch.object(target=dns_check, attribute='resolve', new=TEST_QUERY) def test_raises_exception_on_dns_timeout(self): diff --git a/validate_email/dns_check.py b/validate_email/dns_check.py index 6cd046d..5c11ac4 100644 --- a/validate_email/dns_check.py +++ b/validate_email/dns_check.py @@ -1,6 +1,15 @@ -from dns.exception import Timeout +from datetime import datetime +from ipaddress import IPv4Address, IPv6Address, ip_address +from logging import getLogger +from typing import List, Literal, FrozenSet, Union, Type +from socket import has_ipv6 + +from dns.exception import DNSException, Timeout +from dns.rdataclass import IN as rdcl_in +from dns.rdatatype import AAAA as rdtype_aaaa from dns.rdatatype import MX as rdtype_mx -from dns.rdtypes.ANY.MX import MX +from dns.rdatatype import A as rdtype_a +from dns.rdtypes.ANY.MX import MX as restype_mx from dns.resolver import ( NXDOMAIN, YXDOMAIN, Answer, NoAnswer, NoNameservers, resolve) @@ -10,13 +19,16 @@ from .exceptions import ( DNSConfigurationError, DNSTimeoutError, DomainNotFoundError, NoMXError, NoNameserverError, NoValidMXError) +LOGGER = getLogger(name=__name__) +AddressTypes = FrozenSet[Union[Type[IPv4Address], Type[IPv6Address]]] +DefaultAddressTypes = frozenset([IPv4Address, IPv6Address]) + def _get_mx_records(domain: str, timeout: float) -> Answer: 'Return the DNS response for checking, optionally raise exceptions.' try: return resolve( - qname=domain, rdtype=rdtype_mx, lifetime=timeout, - search=True) + qname=domain, rdtype=rdtype_mx, lifetime=timeout, search=True) except NXDOMAIN: raise DomainNotFoundError except NoNameservers: @@ -29,27 +41,88 @@ def _get_mx_records(domain: str, timeout: float) -> Answer: raise NoMXError -def _get_cleaned_mx_records(domain: str, timeout: float) -> list: +def _resolve_one_recordtype( + hostname: str, records: List[str], timeout: float, + rdtype: Literal[rdtype_aaaa, rdtype_a], result_set: set) -> float: + """ + Resolve one recordtype, add to results, return the new timeout + value. + """ + if timeout <= 0: + return 0 + time_current = datetime.now() + try: + query_result = resolve( + qname=hostname, rdtype=rdtype, rdclass=rdcl_in, lifetime=timeout) + for item in query_result.rrset.processing_order(): + text: str = item.to_text() + if text in result_set: + LOGGER.debug(msg=( + f'{hostname} resolved to {text!r} already in results,' + ' not adding')) + continue + records.append(text) + result_set.add(text) + LOGGER.debug(msg=f'{hostname} resolved to {text}') + except DNSException as exc: + LOGGER.warning(msg=f'{hostname} resolve error: {exc}') + return timeout - (datetime.now() - time_current).total_seconds() + + +def _get_resolved_mx_records( + records: list, timeout: float, + address_types: AddressTypes = DefaultAddressTypes +) -> List[str]: + 'Return a resolved & sorted list of IP addresses from MX records.' + result = [] + result_set = set() + for record in records: + if timeout <= 0: + break + if IPv6Address in address_types and has_ipv6: + timeout = _resolve_one_recordtype( + hostname=record, records=result, timeout=timeout, + rdtype=rdtype_aaaa, result_set=result_set) + if IPv4Address in address_types: + timeout = _resolve_one_recordtype( + hostname=record, records=result, timeout=timeout, + rdtype=rdtype_a, result_set=result_set) + return result + + +def _get_cleaned_mx_records( + domain: str, timeout: float, + address_types: AddressTypes = DefaultAddressTypes +) -> List[str]: """ Return a list of hostnames in the MX record, raise an exception on any issues. """ + time_start = datetime.now() answer = _get_mx_records(domain=domain, timeout=timeout) to_check = list() host_set = set() - for record in answer.rrset.processing_order(): # type: MX + record: restype_mx + for record in answer.rrset.processing_order(): dns_str = record.exchange.to_text().rstrip('.') # type: str if dns_str in host_set: + LOGGER.debug(msg=f'{dns_str} is already in results, not adding') continue to_check.append(dns_str) host_set.add(dns_str) result = [x for x in to_check if HOST_REGEX.search(string=x)] + LOGGER.debug(msg=f'{domain} resolved (MX): {result}') if not result: raise NoValidMXError + time_diff = timeout - (datetime.now() - time_start).total_seconds() + result = _get_resolved_mx_records( + records=result, timeout=time_diff, address_types=address_types) return result -def dns_check(email_address: EmailAddress, timeout: float = 10) -> list: +def dns_check( + email_address: EmailAddress, timeout: float = 10, + address_types: AddressTypes = DefaultAddressTypes) -> List[str]: """ Check whether there are any responsible SMTP servers for the email address by looking up the DNS MX records. @@ -59,7 +132,11 @@ def dns_check(email_address: EmailAddress, timeout: float = 10) -> list: `MXError`. Otherwise, return the list of MX hostnames. """ if email_address.domain_literal_ip: + ip = ip_address(address=email_address.domain_literal_ip) + if type(ip) not in address_types: + raise NoValidMXError return [email_address.domain_literal_ip] else: return _get_cleaned_mx_records( - domain=email_address.domain, timeout=timeout) + domain=email_address.domain, timeout=timeout, + address_types=address_types) diff --git a/validate_email/smtp_check.py b/validate_email/smtp_check.py index ffe250f..aab80b8 100644 --- a/validate_email/smtp_check.py +++ b/validate_email/smtp_check.py @@ -30,7 +30,7 @@ class _SMTPChecker(SMTP): def __init__( self, local_hostname: Optional[str], timeout: float, debug: bool, sender: EmailAddress, recip: EmailAddress, - skip_tls: bool = False, tls_context: SSLContext = None): + skip_tls: bool = False, tls_context: Optional[SSLContext] = None): """ Initialize the object with all the parameters which remain constant during the check of one email address on all the SMTP @@ -59,7 +59,7 @@ class _SMTPChecker(SMTP): def connect( self, host: str = 'localhost', port: int = 0, - source_address: str = None) -> Tuple[int, str]: + source_address: Optional[str] = None) -> Tuple[int, str]: """ Like `smtplib.SMTP.connect`, but raise appropriate exceptions on connection failure or negative SMTP server response. diff --git a/validate_email/validate_email.py b/validate_email/validate_email.py index 8195723..e43f4a7 100644 --- a/validate_email/validate_email.py +++ b/validate_email/validate_email.py @@ -2,7 +2,7 @@ from logging import getLogger from ssl import SSLContext from typing import Optional -from .dns_check import dns_check +from .dns_check import dns_check, DefaultAddressTypes, AddressTypes from .domainlist_check import domainlist_check from .email_address import EmailAddress from .exceptions import ( @@ -34,7 +34,7 @@ def validate_email_or_fail( smtp_timeout: float = 10, smtp_helo_host: Optional[str] = None, smtp_from_address: Optional[str] = None, smtp_skip_tls: bool = False, smtp_tls_context: Optional[SSLContext] = None, - smtp_debug: bool = False + smtp_debug: bool = False, address_types: AddressTypes = DefaultAddressTypes ) -> Optional[bool]: """ Return `True` if the email address validation is successful, `None` @@ -48,7 +48,9 @@ def validate_email_or_fail( domainlist_check(email_address=email_address_to) if not check_dns and not check_smtp: # check_smtp implies check_dns. return True - mx_records = dns_check(email_address=email_address_to, timeout=dns_timeout) + mx_records = dns_check( + email_address=email_address_to, timeout=dns_timeout, + address_types=address_types) if not check_smtp: return True try: