# coding: utf-8 """Tests for the eventlet.support.greendns module""" import os import socket import tempfile import time from dns.resolver import NoAnswer, Answer, Resolver from eventlet.support import greendns from eventlet.support.greendns import dns import tests import tests.mock def _make_host_resolver(): """Returns a HostResolver instance The hosts file will be empty but accessible as a py.path.local instance using the ``hosts`` attribute. """ hosts = tempfile.NamedTemporaryFile() hr = greendns.HostsResolver(fname=hosts.name) hr.hosts = hosts hr._last_stat = 0 return hr class TestHostsResolver(tests.LimitedTestCase): def test_default_fname(self): hr = greendns.HostsResolver() assert os.path.exists(hr.fname) def test_readlines_lines(self): hr = _make_host_resolver() hr.hosts.write(b'line0\n') hr.hosts.flush() assert list(hr._readlines()) == ['line0'] hr._last_stat = 0 hr.hosts.write(b'line1\n') hr.hosts.flush() assert list(hr._readlines()) == ['line0', 'line1'] # Test reading of varied newline styles hr._last_stat = 0 hr.hosts.seek(0) hr.hosts.truncate() hr.hosts.write(b'\naa\r\nbb\r cc \n\n\tdd ee') hr.hosts.flush() assert list(hr._readlines()) == ['aa', 'bb', 'cc', 'dd ee'] # Test comments, including inline comments hr._last_stat = 0 hr.hosts.seek(0) hr.hosts.truncate() hr.hosts.write(b'''\ # First couple lines # are comments. line1 #comment line2 # inline comment ''') hr.hosts.flush() assert list(hr._readlines()) == ['line1', 'line2'] def test_readlines_missing_file(self): hr = _make_host_resolver() hr.hosts.close() hr._last_stat = 0 assert list(hr._readlines()) == [] def test_load_no_contents(self): hr = _make_host_resolver() hr._load() assert not hr._v4 assert not hr._v6 assert not hr._aliases def test_load_v4_v6_cname_aliases(self): hr = _make_host_resolver() hr.hosts.write(b'1.2.3.4 v4.example.com v4\n' b'dead:beef::1 v6.example.com v6\n') hr.hosts.flush() hr._load() assert hr._v4 == {'v4.example.com': '1.2.3.4', 'v4': '1.2.3.4'} assert hr._v6 == {'v6.example.com': 'dead:beef::1', 'v6': 'dead:beef::1'} assert hr._aliases == {'v4': 'v4.example.com', 'v6': 'v6.example.com'} def test_load_v6_link_local(self): hr = _make_host_resolver() hr.hosts.write(b'fe80:: foo\n' b'fe80:dead:beef::1 bar\n') hr.hosts.flush() hr._load() assert not hr._v4 assert not hr._v6 def test_query_A(self): hr = _make_host_resolver() hr._v4 = {'v4.example.com': '1.2.3.4'} ans = hr.query('v4.example.com') assert ans[0].address == '1.2.3.4' ans = hr.query(u'v4.example.com') assert ans[0].address == '1.2.3.4' ans = hr.query(b'v4.example.com') assert ans[0].address == '1.2.3.4' def test_query_ans_types(self): # This assumes test_query_A above succeeds hr = _make_host_resolver() hr._v4 = {'v4.example.com': '1.2.3.4'} hr._last_stat = time.time() ans = hr.query('v4.example.com') assert isinstance(ans, greendns.dns.resolver.Answer) assert ans.response is None assert ans.qname == dns.name.from_text('v4.example.com') assert ans.rdtype == dns.rdatatype.A assert ans.rdclass == dns.rdataclass.IN assert ans.canonical_name == dns.name.from_text('v4.example.com') assert ans.expiration assert isinstance(ans.rrset, dns.rrset.RRset) assert ans.rrset.rdtype == dns.rdatatype.A assert ans.rrset.rdclass == dns.rdataclass.IN ttl = greendns.HOSTS_TTL assert ttl - 1 <= ans.rrset.ttl <= ttl + 1 rr = ans.rrset[0] assert isinstance(rr, greendns.dns.rdtypes.IN.A.A) assert rr.rdtype == dns.rdatatype.A assert rr.rdclass == dns.rdataclass.IN assert rr.address == '1.2.3.4' def test_query_AAAA(self): hr = _make_host_resolver() hr._v6 = {'v6.example.com': 'dead:beef::1'} ans = hr.query('v6.example.com', dns.rdatatype.AAAA) assert ans[0].address == 'dead:beef::1' def test_query_unknown_raises(self): hr = _make_host_resolver() with tests.assert_raises(greendns.dns.resolver.NoAnswer): hr.query('example.com') def test_query_unknown_no_raise(self): hr = _make_host_resolver() ans = hr.query('example.com', raise_on_no_answer=False) assert isinstance(ans, greendns.dns.resolver.Answer) assert ans.response is None assert ans.qname == dns.name.from_text('example.com') assert ans.rdtype == dns.rdatatype.A assert ans.rdclass == dns.rdataclass.IN assert ans.canonical_name == dns.name.from_text('example.com') assert ans.expiration assert isinstance(ans.rrset, greendns.dns.rrset.RRset) assert ans.rrset.rdtype == dns.rdatatype.A assert ans.rrset.rdclass == dns.rdataclass.IN assert len(ans.rrset) == 0 def test_query_CNAME(self): hr = _make_host_resolver() hr._aliases = {'host': 'host.example.com'} ans = hr.query('host', dns.rdatatype.CNAME) assert ans[0].target == dns.name.from_text('host.example.com') assert str(ans[0].target) == 'host.example.com.' def test_query_unknown_type(self): hr = _make_host_resolver() with tests.assert_raises(greendns.dns.resolver.NoAnswer): hr.query('example.com', dns.rdatatype.MX) def test_getaliases(self): hr = _make_host_resolver() hr._aliases = {'host': 'host.example.com', 'localhost': 'host.example.com'} res = set(hr.getaliases('host')) assert res == set(['host.example.com', 'localhost']) def test_getaliases_unknown(self): hr = _make_host_resolver() assert hr.getaliases('host.example.com') == [] def test_getaliases_fqdn(self): hr = _make_host_resolver() hr._aliases = {'host': 'host.example.com'} res = set(hr.getaliases('host.example.com')) assert res == set(['host']) def test_hosts_case_insensitive(self): name = 'example.com' hr = _make_host_resolver() hr.hosts.write(b'1.2.3.4 ExAmPlE.CoM\n') hr.hosts.flush() hr._load() ans = hr.query(name) rr = ans.rrset[0] assert isinstance(rr, greendns.dns.rdtypes.IN.A.A) assert rr.rdtype == dns.rdatatype.A assert rr.rdclass == dns.rdataclass.IN assert rr.address == '1.2.3.4' def _make_mock_base_resolver(): """A mocked base resolver class""" class RR(object): pass class Resolver(object): aliases = ['cname.example.com'] raises = None rr = RR() rr6 = RR() def query(self, *args, **kwargs): self.args = args self.kwargs = kwargs if self.raises: raise self.raises() if hasattr(self, 'rrset'): rrset = self.rrset else: if self.rr6 and self.args[1] == dns.rdatatype.AAAA: rrset = [self.rr6] else: rrset = [self.rr] return greendns.HostsAnswer('foo', 1, 1, rrset, False) def getaliases(self, *args, **kwargs): return self.aliases return Resolver class TestUdp(tests.LimitedTestCase): def setUp(self): # Store this so we can reuse it for each test self.query = greendns.dns.message.Message() self.query.flags = greendns.dns.flags.QR self.query_wire = self.query.to_wire() super(TestUdp, self).setUp() def test_udp_ipv4(self): with tests.mock.patch('eventlet.support.greendns.socket.socket.recvfrom', return_value=(self.query_wire, ('127.0.0.1', 53))): greendns.udp(self.query, '127.0.0.1') def test_udp_ipv4_timeout(self): with tests.mock.patch('eventlet.support.greendns.socket.socket.recvfrom', side_effect=socket.timeout): with tests.assert_raises(dns.exception.Timeout): greendns.udp(self.query, '127.0.0.1', timeout=0.1) def test_udp_ipv4_wrong_addr_ignore(self): with tests.mock.patch('eventlet.support.greendns.socket.socket.recvfrom', side_effect=socket.timeout): with tests.assert_raises(dns.exception.Timeout): greendns.udp(self.query, '127.0.0.1', timeout=0.1, ignore_unexpected=True) def test_udp_ipv4_wrong_addr(self): with tests.mock.patch('eventlet.support.greendns.socket.socket.recvfrom', return_value=(self.query_wire, ('127.0.0.2', 53))): with tests.assert_raises(dns.query.UnexpectedSource): greendns.udp(self.query, '127.0.0.1') def test_udp_ipv6(self): with tests.mock.patch('eventlet.support.greendns.socket.socket.recvfrom', return_value=(self.query_wire, ('::1', 53, 0, 0))): greendns.udp(self.query, '::1') def test_udp_ipv6_timeout(self): with tests.mock.patch('eventlet.support.greendns.socket.socket.recvfrom', side_effect=socket.timeout): with tests.assert_raises(dns.exception.Timeout): greendns.udp(self.query, '::1', timeout=0.1) def test_udp_ipv6_addr_zeroes(self): with tests.mock.patch('eventlet.support.greendns.socket.socket.recvfrom', return_value=(self.query_wire, ('0:00:0000::1', 53, 0, 0))): greendns.udp(self.query, '::1') def test_udp_ipv6_wrong_addr_ignore(self): with tests.mock.patch('eventlet.support.greendns.socket.socket.recvfrom', side_effect=socket.timeout): with tests.assert_raises(dns.exception.Timeout): greendns.udp(self.query, '::1', timeout=0.1, ignore_unexpected=True) def test_udp_ipv6_wrong_addr(self): with tests.mock.patch('eventlet.support.greendns.socket.socket.recvfrom', return_value=(self.query_wire, ('ffff:0000::1', 53, 0, 0))): with tests.assert_raises(dns.query.UnexpectedSource): greendns.udp(self.query, '::1') class TestProxyResolver(tests.LimitedTestCase): def test_clear(self): rp = greendns.ResolverProxy() assert rp._cached_resolver is None resolver = rp._resolver assert resolver is not None rp.clear() assert rp._resolver is not None assert rp._resolver != resolver def _make_mock_hostsresolver(self): """A mocked HostsResolver""" base_resolver = _make_mock_base_resolver() base_resolver.rr.address = '1.2.3.4' return base_resolver() def _make_mock_resolver(self): """A mocked Resolver""" base_resolver = _make_mock_base_resolver() base_resolver.rr.address = '5.6.7.8' return base_resolver() def test_hosts(self): hostsres = self._make_mock_hostsresolver() rp = greendns.ResolverProxy(hostsres) ans = rp.query('host.example.com') assert ans[0].address == '1.2.3.4' ans = rp.query(u'host.example.com') assert ans[0].address == '1.2.3.4' ans = rp.query(b'host.example.com') assert ans[0].address == '1.2.3.4' def test_hosts_noanswer(self): hostsres = self._make_mock_hostsresolver() res = self._make_mock_resolver() rp = greendns.ResolverProxy(hostsres) rp._resolver = res hostsres.raises = greendns.dns.resolver.NoAnswer ans = rp.query('host.example.com') assert ans[0].address == '5.6.7.8' def test_resolver(self): res = self._make_mock_resolver() rp = greendns.ResolverProxy() rp._resolver = res ans = rp.query('host.example.com') assert ans[0].address == '5.6.7.8' assert isinstance(res.args[0], dns.name.Name) ans = rp.query(u'host.example.com') assert ans[0].address == '5.6.7.8' assert isinstance(res.args[0], dns.name.Name) ans = rp.query(b'host.example.com') assert ans[0].address == '5.6.7.8' assert isinstance(res.args[0], dns.name.Name) def test_noanswer(self): res = self._make_mock_resolver() rp = greendns.ResolverProxy() rp._resolver = res res.raises = greendns.dns.resolver.NoAnswer with tests.assert_raises(greendns.dns.resolver.NoAnswer): rp.query('host.example.com') def test_nxdomain(self): res = self._make_mock_resolver() rp = greendns.ResolverProxy() rp._resolver = res res.raises = greendns.dns.resolver.NXDOMAIN with tests.assert_raises(greendns.dns.resolver.NXDOMAIN): rp.query('host.example.com') def test_noanswer_hosts(self): hostsres = self._make_mock_hostsresolver() res = self._make_mock_resolver() rp = greendns.ResolverProxy(hostsres) rp._resolver = res hostsres.raises = greendns.dns.resolver.NoAnswer res.raises = greendns.dns.resolver.NoAnswer with tests.assert_raises(greendns.dns.resolver.NoAnswer): rp.query('host.example.com') def _make_mock_resolver_aliases(self): class RR(object): target = 'host.example.com' class Resolver(object): call_count = 0 exc_type = greendns.dns.resolver.NoAnswer def query(self, *args, **kwargs): self.args = args self.kwargs = kwargs self.call_count += 1 if self.call_count < 2: return greendns.HostsAnswer(args[0], 1, 5, [RR()], False) else: raise self.exc_type() return Resolver() def test_getaliases(self): aliases_res = self._make_mock_resolver_aliases() rp = greendns.ResolverProxy() rp._resolver = aliases_res aliases = set(rp.getaliases('alias.example.com')) assert aliases == set(['host.example.com']) def test_getaliases_fqdn(self): aliases_res = self._make_mock_resolver_aliases() rp = greendns.ResolverProxy() rp._resolver = aliases_res rp._resolver.call_count = 1 assert rp.getaliases('host.example.com') == [] def test_getaliases_nxdomain(self): aliases_res = self._make_mock_resolver_aliases() rp = greendns.ResolverProxy() rp._resolver = aliases_res rp._resolver.call_count = 1 rp._resolver.exc_type = greendns.dns.resolver.NXDOMAIN assert rp.getaliases('host.example.com') == [] class TestResolve(tests.LimitedTestCase): def setUp(self): base_resolver = _make_mock_base_resolver() base_resolver.rr.address = '1.2.3.4' self._old_resolver = greendns.resolver greendns.resolver = base_resolver() def tearDown(self): greendns.resolver = self._old_resolver def test_A(self): ans = greendns.resolve('host.example.com', socket.AF_INET) assert ans[0].address == '1.2.3.4' assert greendns.resolver.args == ('host.example.com', dns.rdatatype.A) def test_AAAA(self): greendns.resolver.rr6.address = 'dead:beef::1' ans = greendns.resolve('host.example.com', socket.AF_INET6) assert ans[0].address == 'dead:beef::1' assert greendns.resolver.args == ('host.example.com', dns.rdatatype.AAAA) def test_unknown_rdtype(self): with tests.assert_raises(socket.gaierror): greendns.resolve('host.example.com', socket.AF_INET6 + 1) def test_timeout(self): greendns.resolver.raises = greendns.dns.exception.Timeout with tests.assert_raises(socket.gaierror): greendns.resolve('host.example.com') def test_exc(self): greendns.resolver.raises = greendns.dns.exception.DNSException with tests.assert_raises(socket.gaierror): greendns.resolve('host.example.com') def test_noraise_noanswer(self): greendns.resolver.rrset = None ans = greendns.resolve('example.com', raises=False) assert not ans.rrset def test_noraise_nxdomain(self): greendns.resolver.raises = greendns.dns.resolver.NXDOMAIN ans = greendns.resolve('example.com', raises=False) assert not ans.rrset class TestResolveCname(tests.LimitedTestCase): def setUp(self): base_resolver = _make_mock_base_resolver() base_resolver.rr.target = 'cname.example.com' self._old_resolver = greendns.resolver greendns.resolver = base_resolver() def tearDown(self): greendns.resolver = self._old_resolver def test_success(self): cname = greendns.resolve_cname('alias.example.com') assert cname == 'cname.example.com' def test_timeout(self): greendns.resolver.raises = greendns.dns.exception.Timeout with tests.assert_raises(socket.gaierror): greendns.resolve_cname('alias.example.com') def test_nodata(self): greendns.resolver.raises = greendns.dns.exception.DNSException with tests.assert_raises(socket.gaierror): greendns.resolve_cname('alias.example.com') def test_no_answer(self): greendns.resolver.raises = greendns.dns.resolver.NoAnswer assert greendns.resolve_cname('host.example.com') == 'host.example.com' def _make_mock_resolve(): """A stubbed out resolve function This monkeypatches the greendns.resolve() function with a mock. You must give it answers by calling .add(). """ class MockAnswer(list): pass class MockResolve(object): def __init__(self): self.answers = {} def __call__(self, name, family=socket.AF_INET, raises=True, _proxy=None, use_network=True): qname = dns.name.from_text(name) try: rrset = self.answers[name][family] except KeyError: if raises: raise greendns.dns.resolver.NoAnswer() rrset = dns.rrset.RRset(qname, 1, 1) ans = MockAnswer() ans.qname = qname ans.rrset = rrset ans.extend(rrset.items) return ans def add(self, name, addr): """Add an address to a name and family""" try: rdata = dns.rdtypes.IN.A.A(dns.rdataclass.IN, dns.rdatatype.A, addr) family = socket.AF_INET except (socket.error, dns.exception.SyntaxError): rdata = dns.rdtypes.IN.AAAA.AAAA(dns.rdataclass.IN, dns.rdatatype.AAAA, addr) family = socket.AF_INET6 family_dict = self.answers.setdefault(name, {}) rrset = family_dict.get(family) if not rrset: family_dict[family] = rrset = dns.rrset.RRset( dns.name.from_text(name), rdata.rdclass, rdata.rdtype) rrset.add(rdata) resolve = MockResolve() return resolve class TestGetaddrinfo(tests.LimitedTestCase): def _make_mock_resolve_cname(self): """A stubbed out cname function""" class ResolveCname(object): qname = None cname = 'cname.example.com' def __call__(self, host): self.qname = host return self.cname resolve_cname = ResolveCname() return resolve_cname def setUp(self): self._old_resolve = greendns.resolve self._old_resolve_cname = greendns.resolve_cname self._old_orig_getaddrinfo = greendns.socket.getaddrinfo def tearDown(self): greendns.resolve = self._old_resolve greendns.resolve_cname = self._old_resolve_cname greendns.socket.getaddrinfo = self._old_orig_getaddrinfo def test_getaddrinfo(self): greendns.resolve = _make_mock_resolve() greendns.resolve.add('example.com', '127.0.0.2') greendns.resolve.add('example.com', '::1') res = greendns.getaddrinfo('example.com', 'domain') addr = ('127.0.0.2', 53) tcp = (socket.AF_INET, socket.SOCK_STREAM, socket.IPPROTO_TCP, addr) udp = (socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP, addr) addr = ('::1', 53, 0, 0) tcp6 = (socket.AF_INET6, socket.SOCK_STREAM, socket.IPPROTO_TCP, addr) udp6 = (socket.AF_INET6, socket.SOCK_DGRAM, socket.IPPROTO_UDP, addr) filt_res = [ai[:3] + (ai[4],) for ai in res] assert tcp in filt_res assert udp in filt_res assert tcp6 in filt_res assert udp6 in filt_res def test_getaddrinfo_idn(self): greendns.resolve = _make_mock_resolve() idn_name = u'евентлет.com' greendns.resolve.add(idn_name.encode('idna').decode('ascii'), '127.0.0.2') res = greendns.getaddrinfo(idn_name, 'domain') addr = ('127.0.0.2', 53) tcp = (socket.AF_INET, socket.SOCK_STREAM, socket.IPPROTO_TCP, addr) udp = (socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP, addr) filt_res = [ai[:3] + (ai[4],) for ai in res] assert tcp in filt_res assert udp in filt_res def test_getaddrinfo_inet(self): greendns.resolve = _make_mock_resolve() greendns.resolve.add('example.com', '127.0.0.2') res = greendns.getaddrinfo('example.com', 'domain', socket.AF_INET) addr = ('127.0.0.2', 53) tcp = (socket.AF_INET, socket.SOCK_STREAM, socket.IPPROTO_TCP, addr) udp = (socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP, addr) assert tcp in [ai[:3] + (ai[4],) for ai in res] assert udp in [ai[:3] + (ai[4],) for ai in res] def test_getaddrinfo_inet6(self): greendns.resolve = _make_mock_resolve() greendns.resolve.add('example.com', '::1') res = greendns.getaddrinfo('example.com', 'domain', socket.AF_INET6) addr = ('::1', 53, 0, 0) tcp = (socket.AF_INET6, socket.SOCK_STREAM, socket.IPPROTO_TCP, addr) udp = (socket.AF_INET6, socket.SOCK_DGRAM, socket.IPPROTO_UDP, addr) assert tcp in [ai[:3] + (ai[4],) for ai in res] assert udp in [ai[:3] + (ai[4],) for ai in res] def test_getaddrinfo_only_a_ans(self): greendns.resolve = _make_mock_resolve() greendns.resolve.add('example.com', '1.2.3.4') res = greendns.getaddrinfo('example.com', None) addr = [('1.2.3.4', 0)] * len(res) assert addr == [ai[-1] for ai in res] def test_getaddrinfo_only_aaaa_ans(self): greendns.resolve = _make_mock_resolve() greendns.resolve.add('example.com', 'dead:beef::1') res = greendns.getaddrinfo('example.com', None) addr = [('dead:beef::1', 0, 0, 0)] * len(res) assert addr == [ai[-1] for ai in res] def test_getaddrinfo_hosts_only_ans_with_timeout(self): def clear_raises(res_self): res_self.raises = None return greendns.dns.resolver.NoAnswer() hostsres = _make_mock_base_resolver() hostsres.raises = clear_raises hostsres.rr.address = '1.2.3.4' greendns.resolver = greendns.ResolverProxy(hostsres()) res = _make_mock_base_resolver() res.raises = greendns.dns.exception.Timeout greendns.resolver._resolver = res() result = greendns.getaddrinfo('example.com', None, 0) addr = [('1.2.3.4', 0)] * len(result) assert addr == [ai[-1] for ai in result] def test_getaddrinfo_hosts_only_ans_with_error(self): def clear_raises(res_self): res_self.raises = None return greendns.dns.resolver.NoAnswer() hostsres = _make_mock_base_resolver() hostsres.raises = clear_raises hostsres.rr.address = '1.2.3.4' greendns.resolver = greendns.ResolverProxy(hostsres()) res = _make_mock_base_resolver() res.raises = greendns.dns.exception.DNSException greendns.resolver._resolver = res() result = greendns.getaddrinfo('example.com', None, 0) addr = [('1.2.3.4', 0)] * len(result) assert addr == [ai[-1] for ai in result] def test_getaddrinfo_bytes(self): greendns.resolve = _make_mock_resolve() greendns.resolve.add('example.com', '1.2.3.4') res = greendns.getaddrinfo(b'example.com', b'0') addr = [('1.2.3.4', 0)] * len(res) assert addr == [ai[-1] for ai in res] def test_getaddrinfo_hosts_only_timeout(self): hostsres = _make_mock_base_resolver() hostsres.raises = greendns.dns.resolver.NoAnswer greendns.resolver = greendns.ResolverProxy(hostsres()) res = _make_mock_base_resolver() res.raises = greendns.dns.exception.Timeout greendns.resolver._resolver = res() with tests.assert_raises(socket.gaierror): greendns.getaddrinfo('example.com', None, 0) def test_getaddrinfo_hosts_only_dns_error(self): hostsres = _make_mock_base_resolver() hostsres.raises = greendns.dns.resolver.NoAnswer greendns.resolver = greendns.ResolverProxy(hostsres()) res = _make_mock_base_resolver() res.raises = greendns.dns.exception.DNSException greendns.resolver._resolver = res() with tests.assert_raises(socket.gaierror): greendns.getaddrinfo('example.com', None, 0) def test_canonname(self): greendns.resolve = _make_mock_resolve() greendns.resolve.add('host.example.com', '1.2.3.4') greendns.resolve_cname = self._make_mock_resolve_cname() res = greendns.getaddrinfo('host.example.com', None, 0, 0, 0, socket.AI_CANONNAME) assert res[0][3] == 'cname.example.com' def test_host_none(self): res = greendns.getaddrinfo(None, 80) for addr in set(ai[-1] for ai in res): assert addr in [('127.0.0.1', 80), ('::1', 80, 0, 0)] def test_host_none_passive(self): res = greendns.getaddrinfo(None, 80, 0, 0, 0, socket.AI_PASSIVE) for addr in set(ai[-1] for ai in res): assert addr in [('0.0.0.0', 80), ('::', 80, 0, 0)] def test_v4mapped(self): greendns.resolve = _make_mock_resolve() greendns.resolve.add('example.com', '1.2.3.4') res = greendns.getaddrinfo('example.com', 80, socket.AF_INET6, 0, 0, socket.AI_V4MAPPED) addrs = set(ai[-1] for ai in res) assert addrs == set([('::ffff:1.2.3.4', 80, 0, 0)]) def test_v4mapped_all(self): greendns.resolve = _make_mock_resolve() greendns.resolve.add('example.com', '1.2.3.4') greendns.resolve.add('example.com', 'dead:beef::1') res = greendns.getaddrinfo('example.com', 80, socket.AF_INET6, 0, 0, socket.AI_V4MAPPED | socket.AI_ALL) addrs = set(ai[-1] for ai in res) for addr in addrs: assert addr in [('::ffff:1.2.3.4', 80, 0, 0), ('dead:beef::1', 80, 0, 0)] def test_numericserv(self): greendns.resolve = _make_mock_resolve() greendns.resolve.add('example.com', '1.2.3.4') with tests.assert_raises(socket.gaierror): greendns.getaddrinfo('example.com', 'www', 0, 0, 0, socket.AI_NUMERICSERV) def test_numerichost(self): greendns.resolve = _make_mock_resolve() greendns.resolve.add('example.com', '1.2.3.4') with tests.assert_raises(socket.gaierror): greendns.getaddrinfo('example.com', 80, 0, 0, 0, socket.AI_NUMERICHOST) def test_noport(self): greendns.resolve = _make_mock_resolve() greendns.resolve.add('example.com', '1.2.3.4') ai = greendns.getaddrinfo('example.com', None) assert ai[0][-1][1] == 0 def test_AI_ADDRCONFIG(self): # When the users sets AI_ADDRCONFIG but only has an IPv4 # address configured we will iterate over the results, but the # call for the IPv6 address will fail rather then return an # empty list. In that case we should catch the exception and # only return the ones which worked. def getaddrinfo(addr, port, family, socktype, proto, aiflags): if addr == '127.0.0.1': return [(socket.AF_INET, 1, 0, '', ('127.0.0.1', 0))] elif addr == '::1' and aiflags & socket.AI_ADDRCONFIG: raise socket.error(socket.EAI_ADDRFAMILY, 'Address family for hostname not supported') elif addr == '::1' and not aiflags & socket.AI_ADDRCONFIG: return [(socket.AF_INET6, 1, 0, '', ('::1', 0, 0, 0))] greendns.socket.getaddrinfo = getaddrinfo greendns.resolve = _make_mock_resolve() greendns.resolve.add('localhost', '127.0.0.1') greendns.resolve.add('localhost', '::1') res = greendns.getaddrinfo('localhost', None, 0, 0, 0, socket.AI_ADDRCONFIG) assert res == [(socket.AF_INET, 1, 0, '', ('127.0.0.1', 0))] def test_AI_ADDRCONFIG_noaddr(self): # If AI_ADDRCONFIG is used but there is no address we need to # get an exception, not an empty list. def getaddrinfo(addr, port, family, socktype, proto, aiflags): raise socket.error(socket.EAI_ADDRFAMILY, 'Address family for hostname not supported') greendns.socket.getaddrinfo = getaddrinfo greendns.resolve = _make_mock_resolve() try: greendns.getaddrinfo('::1', None, 0, 0, 0, socket.AI_ADDRCONFIG) except socket.error as e: assert e.errno == socket.EAI_ADDRFAMILY class TestIsIpAddr(tests.LimitedTestCase): def test_isv4(self): assert greendns.is_ipv4_addr('1.2.3.4') def test_isv4_false(self): assert not greendns.is_ipv4_addr('260.0.0.0') def test_isv6(self): assert greendns.is_ipv6_addr('dead:beef::1') def test_isv6_invalid(self): assert not greendns.is_ipv6_addr('foobar::1') def test_v4(self): assert greendns.is_ip_addr('1.2.3.4') def test_v4_illegal(self): assert not greendns.is_ip_addr('300.0.0.1') def test_v6_addr(self): assert greendns.is_ip_addr('::1') def test_isv4_none(self): assert not greendns.is_ipv4_addr(None) def test_isv6_none(self): assert not greendns.is_ipv6_addr(None) def test_none(self): assert not greendns.is_ip_addr(None) class TestGethostbyname(tests.LimitedTestCase): def setUp(self): self._old_resolve = greendns.resolve greendns.resolve = _make_mock_resolve() def tearDown(self): greendns.resolve = self._old_resolve def test_ipaddr(self): assert greendns.gethostbyname('1.2.3.4') == '1.2.3.4' def test_name(self): greendns.resolve.add('host.example.com', '1.2.3.4') assert greendns.gethostbyname('host.example.com') == '1.2.3.4' class TestGetaliases(tests.LimitedTestCase): def _make_mock_resolver(self): base_resolver = _make_mock_base_resolver() resolver = base_resolver() resolver.aliases = ['cname.example.com'] return resolver def setUp(self): self._old_resolver = greendns.resolver greendns.resolver = self._make_mock_resolver() def tearDown(self): greendns.resolver = self._old_resolver def test_getaliases(self): assert greendns.getaliases('host.example.com') == ['cname.example.com'] class TestGethostbyname_ex(tests.LimitedTestCase): def _make_mock_getaliases(self): class GetAliases(object): aliases = ['cname.example.com'] def __call__(self, *args, **kwargs): return self.aliases getaliases = GetAliases() return getaliases def setUp(self): self._old_resolve = greendns.resolve greendns.resolve = _make_mock_resolve() self._old_getaliases = greendns.getaliases def tearDown(self): greendns.resolve = self._old_resolve greendns.getaliases = self._old_getaliases def test_ipaddr(self): res = greendns.gethostbyname_ex('1.2.3.4') assert res == ('1.2.3.4', [], ['1.2.3.4']) def test_name(self): greendns.resolve.add('host.example.com', '1.2.3.4') greendns.getaliases = self._make_mock_getaliases() greendns.getaliases.aliases = [] res = greendns.gethostbyname_ex('host.example.com') assert res == ('host.example.com', [], ['1.2.3.4']) def test_multiple_addrs(self): greendns.resolve.add('host.example.com', '1.2.3.4') greendns.resolve.add('host.example.com', '1.2.3.5') greendns.getaliases = self._make_mock_getaliases() greendns.getaliases.aliases = [] res = greendns.gethostbyname_ex('host.example.com') assert res == ('host.example.com', [], ['1.2.3.4', '1.2.3.5']) class TinyDNSTests(tests.LimitedTestCase): def test_raise_dns_tcp(self): # https://github.com/eventlet/eventlet/issues/499 # None means we don't want the server to find the IP with tests.dns_tcp_server(None) as dnsaddr: resolver = Resolver() resolver.nameservers = [dnsaddr[0]] resolver.nameserver_ports[dnsaddr[0]] = dnsaddr[1] with self.assertRaises(NoAnswer): resolver.query('host.example.com', 'a', tcp=True) def test_noraise_dns_tcp(self): # https://github.com/eventlet/eventlet/issues/499 expected_ip = "192.168.1.1" with tests.dns_tcp_server(expected_ip) as dnsaddr: resolver = Resolver() resolver.nameservers = [dnsaddr[0]] resolver.nameserver_ports[dnsaddr[0]] = dnsaddr[1] response = resolver.query('host.example.com', 'a', tcp=True) self.assertIsInstance(response, Answer) self.assertEqual(list(response.rrset.items)[0].address, expected_ip) def test_reverse_name(): tests.run_isolated('greendns_from_address_203.py') def test_proxy_resolve_unqualified(): # https://github.com/eventlet/eventlet/issues/363 rp = greendns.ResolverProxy(filename=None) rp._resolver.search.append(dns.name.from_text('example.com')) with tests.mock.patch('dns.resolver.Resolver.query', side_effect=dns.resolver.NoAnswer) as m: try: rp.query('machine') assert False, 'Expected NoAnswer exception' except dns.resolver.NoAnswer: pass assert any(call[0][0] == dns.name.from_text('machine') for call in m.call_args_list) assert any(call[0][0] == dns.name.from_text('machine.') for call in m.call_args_list) def test_hosts_priority(): name = 'example.com' addr_from_ns = '1.0.2.0' hr = _make_host_resolver() rp = greendns.ResolverProxy(hosts_resolver=hr, filename=None) base = _make_mock_base_resolver() base.rr.address = addr_from_ns rp._resolver = base() # Default behavior rrns = greendns.resolve(name, _proxy=rp).rrset[0] assert rrns.address == addr_from_ns # Hosts result must shadow that from nameservers hr.hosts.write(b'1.2.3.4 example.com\ndead:beef::1 example.com\n') hr.hosts.flush() hr._load() rrs4 = greendns.resolve(name, family=socket.AF_INET, _proxy=rp).rrset assert len(rrs4) == 1 assert rrs4[0].address == '1.2.3.4', rrs4[0].address rrs6 = greendns.resolve(name, family=socket.AF_INET6, _proxy=rp).rrset assert len(rrs6) == 1 assert rrs6[0].address == 'dead:beef::1', rrs6[0].address def test_hosts_no_network(): name = 'example.com' addr_from_ns = '1.0.2.0' addr6_from_ns = 'dead:beef::1' hr = _make_host_resolver() rp = greendns.ResolverProxy(hosts_resolver=hr, filename=None) base = _make_mock_base_resolver() base.rr.address = addr_from_ns base.rr6.address = addr6_from_ns rp._resolver = base() with tests.mock.patch.object(greendns, 'resolver', new_callable=tests.mock.PropertyMock(return_value=rp)): res = greendns.getaddrinfo('example.com', 'domain', socket.AF_UNSPEC) # Default behavior addr = (addr_from_ns, 53) tcp = (socket.AF_INET, socket.SOCK_STREAM, socket.IPPROTO_TCP, addr) udp = (socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP, addr) addr = (addr6_from_ns, 53, 0, 0) tcp6 = (socket.AF_INET6, socket.SOCK_STREAM, socket.IPPROTO_TCP, addr) udp6 = (socket.AF_INET6, socket.SOCK_DGRAM, socket.IPPROTO_UDP, addr) filt_res = [ai[:3] + (ai[4],) for ai in res] assert tcp in filt_res assert udp in filt_res assert tcp6 in filt_res assert udp6 in filt_res # Hosts result must shadow that from nameservers hr = _make_host_resolver() hr.hosts.write(b'1.2.3.4 example.com') hr.hosts.flush() hr._load() greendns.resolver._hosts = hr res = greendns.getaddrinfo('example.com', 'domain', socket.AF_UNSPEC) filt_res = [ai[:3] + (ai[4],) for ai in res] # Make sure that only IPv4 entry from hosts is present. assert tcp not in filt_res assert udp not in filt_res assert tcp6 not in filt_res assert udp6 not in filt_res addr = ('1.2.3.4', 53) tcp = (socket.AF_INET, socket.SOCK_STREAM, socket.IPPROTO_TCP, addr) udp = (socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP, addr) assert tcp in filt_res assert udp in filt_res def test_import_rdtypes_then_eventlet(): # https://github.com/eventlet/eventlet/issues/479 tests.run_isolated('greendns_import_rdtypes_then_eventlet.py')