host = host[:i]
else:
port = self.default_port
- if host and host[0] == '[' and host[-1] == ']':
- host = host[1:-1]
+ if host and host[0] == '[' and host[-1] == ']':
+ host = host[1:-1]
return (host, port)
def set_debuglevel(self, level):
self.debuglevel = level
+ def _wrap_ipv6(self, ip):
+ if b':' in ip and ip[0] != b'['[0]:
+ return b"[" + ip + b"]"
+ return ip
+
def _tunnel(self):
connect = b"CONNECT %s:%d HTTP/1.0\r\n" % (
- self._tunnel_host.encode("ascii"), self._tunnel_port)
+ self._wrap_ipv6(self._tunnel_host.encode("ascii")),
+ self._tunnel_port)
headers = [connect]
for header, value in self._tunnel_headers.items():
headers.append(f"{header}: {value}\r\n".encode("latin-1"))
# As per RFC 273, IPv6 address should be wrapped with []
# when used as Host header
-
+ host_enc = self._wrap_ipv6(host_enc)
if ":" in host:
- host_enc = b'[' + host_enc + b']'
host_enc = _strip_ipv6_iface(host_enc)
if port == self.default_port:
self.assertIn(b'CONNECT destination.com', self.conn.sock.data)
self.assertIn(b'Host: destination.com', self.conn.sock.data)
+ def test_connect_put_request_ipv6(self):
+ self.conn.set_tunnel('[1:2:3::4]', 1234)
+ self.conn.request('PUT', '/', '')
+ self.assertEqual(self.conn.sock.host, self.host)
+ self.assertEqual(self.conn.sock.port, client.HTTP_PORT)
+ self.assertIn(b'CONNECT [1:2:3::4]:1234', self.conn.sock.data)
+ self.assertIn(b'Host: [1:2:3::4]:1234', self.conn.sock.data)
+
+ def test_connect_put_request_ipv6_port(self):
+ self.conn.set_tunnel('[1:2:3::4]:1234')
+ self.conn.request('PUT', '/', '')
+ self.assertEqual(self.conn.sock.host, self.host)
+ self.assertEqual(self.conn.sock.port, client.HTTP_PORT)
+ self.assertIn(b'CONNECT [1:2:3::4]:1234', self.conn.sock.data)
+ self.assertIn(b'Host: [1:2:3::4]:1234', self.conn.sock.data)
+
def test_tunnel_debuglog(self):
expected_header = 'X-Dummy: 1'
response_text = 'HTTP/1.0 200 OK\r\n{}\r\n\r\n'.format(expected_header)