bpo-30378: Fix the problem that SysLogHandler can't handle IPv6 addresses (#1676) (#1903)
diff --git a/Lib/logging/handlers.py b/Lib/logging/handlers.py
index 2356f8d..2f934b3 100644
--- a/Lib/logging/handlers.py
+++ b/Lib/logging/handlers.py
@@ -827,11 +827,26 @@
self.unixsocket = False
if socktype is None:
socktype = socket.SOCK_DGRAM
- self.socket = socket.socket(socket.AF_INET, socktype)
- if socktype == socket.SOCK_STREAM:
- self.socket.connect(address)
+ host, port = address
+ ress = socket.getaddrinfo(host, port, 0, socktype)
+ if not ress:
+ raise OSError("getaddrinfo returns an empty list")
+ for res in ress:
+ af, socktype, proto, _, sa = res
+ err = sock = None
+ try:
+ sock = socket.socket(af, socktype, proto)
+ if socktype == socket.SOCK_STREAM:
+ sock.connect(sa)
+ break
+ except OSError as exc:
+ err = exc
+ if sock is not None:
+ sock.close()
+ if err is not None:
+ raise err
+ self.socket = sock
self.socktype = socktype
- self.formatter = None
def _connect_unixsocket(self, address):
use_socktype = self.socktype
@@ -870,7 +885,7 @@
priority = self.priority_names[priority]
return (facility << 3) | priority
- def close (self):
+ def close(self):
"""
Closes the socket.
"""
diff --git a/Lib/test/test_logging.py b/Lib/test/test_logging.py
index 0a2584c..474affd 100644
--- a/Lib/test/test_logging.py
+++ b/Lib/test/test_logging.py
@@ -1663,7 +1663,7 @@
server.ready.wait()
hcls = logging.handlers.SysLogHandler
if isinstance(server.server_address, tuple):
- self.sl_hdlr = hcls(('localhost', server.port))
+ self.sl_hdlr = hcls((server.server_address[0], server.port))
else:
self.sl_hdlr = hcls(server.server_address)
self.log_output = ''
@@ -1723,6 +1723,24 @@
SysLogHandlerTest.tearDown(self)
support.unlink(self.address)
+@unittest.skipUnless(support.IPV6_ENABLED,
+ 'IPv6 support required for this test.')
+@unittest.skipUnless(threading, 'Threading required for this test.')
+class IPv6SysLogHandlerTest(SysLogHandlerTest):
+
+ """Test for SysLogHandler with IPv6 host."""
+
+ server_class = TestUDPServer
+ address = ('::1', 0)
+
+ def setUp(self):
+ self.server_class.address_family = socket.AF_INET6
+ super(IPv6SysLogHandlerTest, self).setUp()
+
+ def tearDown(self):
+ self.server_class.address_family = socket.AF_INET
+ super(IPv6SysLogHandlerTest, self).tearDown()
+
@unittest.skipUnless(threading, 'Threading required for this test.')
class HTTPHandlerTest(BaseTest):
"""Test for HTTPHandler."""
@@ -4378,7 +4396,7 @@
QueueHandlerTest, ShutdownTest, ModuleLevelMiscTest, BasicConfigTest,
LoggerAdapterTest, LoggerTest, SMTPHandlerTest, FileHandlerTest,
RotatingFileHandlerTest, LastResortTest, LogRecordTest,
- ExceptionTest, SysLogHandlerTest, HTTPHandlerTest,
+ ExceptionTest, SysLogHandlerTest, IPv6SysLogHandlerTest, HTTPHandlerTest,
NTEventLogHandlerTest, TimedRotatingFileHandlerTest,
UnixSocketHandlerTest, UnixDatagramHandlerTest, UnixSysLogHandlerTest,
MiscTestCase