Method _tunnel

Lib/http/client.py:978–1008

(self)

        return ip

    def _tunnel(self):
        connect = b"CONNECT %s:%d %s\r\n" % (
            self._wrap_ipv6(self._tunnel_host.encode("idna")),
            self._tunnel_port,
            self._http_vsn_str.encode("ascii"))
        headers = [connect]
        for header, value in self._tunnel_headers.items():
            headers.append(f"{header}: {value}\r\n".encode("latin-1"))
        headers.append(b"\r\n")
        # Making a single send() call instead of one per line encourages
        # the host OS to use a more optimal packet size instead of
        # potentially emitting a series of small packets.
        self.send(b"".join(headers))
        del headers

        response = self.response_class(self.sock, method=self._method)
        try:
            (version, code, message) = response._read_status()

            self._raw_proxy_headers = _read_headers(response.fp, self.max_response_headers)

            if self.debuglevel > 0:
                for header in self._raw_proxy_headers:
                    print('header:', header.decode())

            if code != http.HTTPStatus.OK:
                self.close()
                raise OSError(f"Tunnel connection failed: {code} {message.strip()}")

        finally:
            response.close()

    def get_proxy_response_headers(self):
        """