mirrors/cpython
1
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-07-23 03:05:38 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 15

Issue #27136: Fix DNS static resolution; don't use it in getaddrinfo

Browse source 
Patch by A. Jesse Jiryu Davis
This commit is contained in:
Yury Selivanov 2016-06-08 12:33:31 -04:00
parent 7d7a11b5d7
commit f1c6fa9866
6 changed files with 116 additions and 163 deletions
Download patch file Download diff file Expand all files Collapse all files

91
Lib/test/test_asyncio/test_base_events.py
View file

@ -45,6 +45,7 @@ def mock_socket_module():
m_socket.socket = mock.MagicMock()
m_socket.socket.return_value = test_utils.mock_nonblocking_socket()
m_socket.getaddrinfo._is_coroutine = False
return m_socket
@ -56,14 +57,6 @@ def patch_socket(f):
class BaseEventTests(test_utils.TestCase):
def setUp(self):
super().setUp()
base_events._ipaddr_info.cache_clear()
def tearDown(self):
base_events._ipaddr_info.cache_clear()
super().tearDown()
def test_ipaddr_info(self):
UNSPEC = socket.AF_UNSPEC
INET = socket.AF_INET
@ -77,6 +70,10 @@ class BaseEventTests(test_utils.TestCase):
(INET, STREAM, TCP, '', ('1.2.3.4', 1)),
base_events._ipaddr_info('1.2.3.4', 1, INET, STREAM, TCP))
self.assertEqual(
(INET, STREAM, TCP, '', ('1.2.3.4', 1)),
base_events._ipaddr_info(b'1.2.3.4', 1, INET, STREAM, TCP))
self.assertEqual(
(INET, STREAM, TCP, '', ('1.2.3.4', 1)),
base_events._ipaddr_info('1.2.3.4', 1, UNSPEC, STREAM, TCP))
@ -116,8 +113,7 @@ class BaseEventTests(test_utils.TestCase):
base_events._ipaddr_info('::3', 1, INET, STREAM, TCP))
# IPv6 address with zone index.
self.assertEqual(
(INET6, STREAM, TCP, '', ('::3%lo0', 1)),
self.assertIsNone(
base_events._ipaddr_info('::3%lo0', 1, INET6, STREAM, TCP))
def test_port_parameter_types(self):
@ -169,31 +165,10 @@ class BaseEventTests(test_utils.TestCase):
@patch_socket
def test_ipaddr_info_no_inet_pton(self, m_socket):
del m_socket.inet_pton
self.test_ipaddr_info()
def test_check_resolved_address(self):
sock = socket.socket(socket.AF_INET)
with sock:
base_events._check_resolved_address(sock, ('1.2.3.4', 1))
sock = socket.socket(socket.AF_INET6)
with sock:
base_events._check_resolved_address(sock, ('::3', 1))
base_events._check_resolved_address(sock, ('::3%lo0', 1))
with self.assertRaises(ValueError):
base_events._check_resolved_address(sock, ('foo', 1))
def test_check_resolved_sock_type(self):
# Ensure we ignore extra flags in sock.type.
if hasattr(socket, 'SOCK_NONBLOCK'):
sock = socket.socket(type=socket.SOCK_STREAM | socket.SOCK_NONBLOCK)
with sock:
base_events._check_resolved_address(sock, ('1.2.3.4', 1))
if hasattr(socket, 'SOCK_CLOEXEC'):
sock = socket.socket(type=socket.SOCK_STREAM | socket.SOCK_CLOEXEC)
with sock:
base_events._check_resolved_address(sock, ('1.2.3.4', 1))
self.assertIsNone(base_events._ipaddr_info('1.2.3.4', 1,
socket.AF_INET,
socket.SOCK_STREAM,
socket.IPPROTO_TCP))
class BaseEventLoopTests(test_utils.TestCase):
@ -1042,11 +1017,6 @@ class BaseEventLoopWithSelectorTests(test_utils.TestCase):
self.loop = asyncio.new_event_loop()
self.set_event_loop(self.loop)
def tearDown(self):
# Clear mocked constants like AF_INET from the cache.
base_events._ipaddr_info.cache_clear()
super().tearDown()
@patch_socket
def test_create_connection_multiple_errors(self, m_socket):
@ -1195,10 +1165,7 @@ class BaseEventLoopWithSelectorTests(test_utils.TestCase):
if not allow_inet_pton:
del m_socket.inet_pton
def getaddrinfo(*args, **kw):
self.fail('should not have called getaddrinfo')
m_socket.getaddrinfo = getaddrinfo
m_socket.getaddrinfo = socket.getaddrinfo
sock = m_socket.socket.return_value
self.loop.add_reader = mock.Mock()
@ -1210,9 +1177,9 @@ class BaseEventLoopWithSelectorTests(test_utils.TestCase):
t, p = self.loop.run_until_complete(coro)
try:
sock.connect.assert_called_with(('1.2.3.4', 80))
m_socket.socket.assert_called_with(family=m_socket.AF_INET,
proto=m_socket.IPPROTO_TCP,
type=m_socket.SOCK_STREAM)
_, kwargs = m_socket.socket.call_args
self.assertEqual(kwargs['family'], m_socket.AF_INET)
self.assertEqual(kwargs['type'], m_socket.SOCK_STREAM)
finally:
t.close()
test_utils.run_briefly(self.loop) # allow transport to close
@ -1221,10 +1188,15 @@ class BaseEventLoopWithSelectorTests(test_utils.TestCase):
coro = self.loop.create_connection(asyncio.Protocol, '::2', 80)
t, p = self.loop.run_until_complete(coro)
try:
sock.connect.assert_called_with(('::2', 80))
m_socket.socket.assert_called_with(family=m_socket.AF_INET6,
proto=m_socket.IPPROTO_TCP,
type=m_socket.SOCK_STREAM)
# Without inet_pton we use getaddrinfo, which transforms ('::2', 80)
# to ('::0.0.0.2', 80, 0, 0). The last 0s are flow info, scope id.
[address] = sock.connect.call_args[0]
host, port = address[:2]
self.assertRegex(host, r'::(0\.)*2')
self.assertEqual(port, 80)
_, kwargs = m_socket.socket.call_args
self.assertEqual(kwargs['family'], m_socket.AF_INET6)
self.assertEqual(kwargs['type'], m_socket.SOCK_STREAM)
finally:
t.close()
test_utils.run_briefly(self.loop) # allow transport to close
@ -1256,6 +1228,21 @@ class BaseEventLoopWithSelectorTests(test_utils.TestCase):
self.assertRaises(
OSError, self.loop.run_until_complete, coro)
@patch_socket
def test_create_connection_bluetooth(self, m_socket):
# See http://bugs.python.org/issue27136, fallback to getaddrinfo when
# we can't recognize an address is resolved, e.g. a Bluetooth address.
addr = ('00:01:02:03:04:05', 1)
def getaddrinfo(host, port, *args, **kw):
assert (host, port) == addr
return [(999, 1, 999, '', (addr, 1))]
m_socket.getaddrinfo = getaddrinfo
sock = m_socket.socket()
coro = self.loop.sock_connect(sock, addr)
self.loop.run_until_complete(coro)
def test_create_connection_ssl_server_hostname_default(self):
self.loop.getaddrinfo = mock.Mock()
@ -1369,7 +1356,7 @@ class BaseEventLoopWithSelectorTests(test_utils.TestCase):
getaddrinfo = self.loop.getaddrinfo = mock.Mock()
getaddrinfo.return_value = []
f = self.loop.create_server(MyProto, '0.0.0.0', 0)
f = self.loop.create_server(MyProto, 'python.org', 0)
self.assertRaises(OSError, self.loop.run_until_complete, f)
@patch_socket