diff --git a/lib/py/src/transport/sslcompat.py b/lib/py/src/transport/sslcompat.py index 90012de719..1d505d0c37 100644 --- a/lib/py/src/transport/sslcompat.py +++ b/lib/py/src/transport/sslcompat.py @@ -64,6 +64,16 @@ def legacy_validate_callback(cert, hostname): % (hostname, cert)) +def _unmap_ipv4(address): + """Reduce an IPv4-mapped IPv6 address to the IPv4 address it carries. + + A dual-stack listener reports an IPv4 peer as ::ffff:127.0.0.1, while a + certificate normally carries the plain 127.0.0.1. They are the same + address and have to compare equal. + """ + return getattr(address, 'ipv4_mapped', None) or address + + def match_peer_ipaddress(cert, hostname): """Check that a peer's certificate covers the IP address it connected from. @@ -88,7 +98,7 @@ def match_peer_ipaddress(cert, hostname): 'No SSL certificate found from %s' % hostname) try: - peer = ipaddress.ip_address(hostname) + peer = _unmap_ipv4(ipaddress.ip_address(hostname)) except ValueError: raise TTransportException( TTransportException.NOT_OPEN, @@ -104,7 +114,7 @@ def match_peer_ipaddress(cert, hostname): if kind != 'IP Address': continue try: - if ipaddress.ip_address(value) == peer: + if _unmap_ipv4(ipaddress.ip_address(value)) == peer: return except ValueError: continue diff --git a/lib/py/test/test_sslsocket.py b/lib/py/test/test_sslsocket.py index a40afdd11d..e03e328934 100644 --- a/lib/py/test/test_sslsocket.py +++ b/lib/py/test/test_sslsocket.py @@ -450,6 +450,21 @@ def test_peer_address_matcher_does_not_pass_on_a_name(self): with self.assertRaises(Exception): match_peer_ipaddress(cert, 'localhost') + def test_peer_address_matcher_unmaps_ipv4(self): + # A dual-stack listener reports an IPv4 peer as ::ffff:127.0.0.1, + # which is the address the certificate carries as 127.0.0.1. + from thrift.transport.sslcompat import match_peer_ipaddress + mapped = {'subjectAltName': (('IP Address', '::ffff:127.0.0.1'),)} + plain = {'subjectAltName': (('IP Address', '127.0.0.1'),)} + match_peer_ipaddress(plain, '::ffff:127.0.0.1') + match_peer_ipaddress(mapped, '127.0.0.1') + with self.assertRaises(Exception): + match_peer_ipaddress( + {'subjectAltName': (('IP Address', '127.0.0.2'),)}, + '::ffff:127.0.0.1') + with self.assertRaises(Exception): + match_peer_ipaddress(plain, '::1') + # Add a dummy test because starting from python 3.12, if all tests in a test # file are skipped that's considered an error.