codekingpro/portable-devtools

1"""2_http.py3websocket - WebSocket client library for Python4 5Copyright 2025 engn33r6 7Licensed under the Apache License, Version 2.0 (the "License");8you may not use this file except in compliance with the License.9You may obtain a copy of the License at10 11    http://www.apache.org/licenses/LICENSE-2.012 13Unless required by applicable law or agreed to in writing, software14distributed under the License is distributed on an "AS IS" BASIS,15WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.16See the License for the specific language governing permissions and17limitations under the License.18"""19 20import errno21import os22import socket23from base64 import encodebytes as base64encode24 25from ._exceptions import (26    WebSocketAddressException,27    WebSocketException,28    WebSocketProxyException,29)30from ._logging import debug, dump, trace31from ._socket import DEFAULT_SOCKET_OPTION, recv_line, send32from ._ssl_compat import HAVE_SSL, ssl33from ._url import get_proxy_info, parse_url34 35__all__ = ["proxy_info", "connect", "read_headers"]36 37try:38    from python_socks._errors import ProxyConnectionError, ProxyError, ProxyTimeoutError39    from python_socks._types import ProxyType40    from python_socks.sync import Proxy41 42    HAVE_PYTHON_SOCKS = True43except:44    HAVE_PYTHON_SOCKS = False45 46    class ProxyError(Exception):47        pass48 49    class ProxyTimeoutError(Exception):50        pass51 52    class ProxyConnectionError(Exception):53        pass54 55 56class proxy_info:57    def __init__(self, **options):58        self.proxy_host = options.get("http_proxy_host", None)59        if self.proxy_host:60            self.proxy_port = options.get("http_proxy_port", 0)61            self.auth = options.get("http_proxy_auth", None)62            self.no_proxy = options.get("http_no_proxy", None)63            self.proxy_protocol = options.get("proxy_type", "http")64            # Note: If timeout not specified, default python-socks timeout is 60 seconds65            self.proxy_timeout = options.get("http_proxy_timeout", None)66            if self.proxy_protocol not in [67                "http",68                "socks4",69                "socks4a",70                "socks5",71                "socks5h",72            ]:73                raise ProxyError(74                    "Only http, socks4, socks5 proxy protocols are supported"75                )76        else:77            self.proxy_port = 078            self.auth = None79            self.no_proxy = None80            self.proxy_protocol = "http"81 82 83def _start_proxied_socket(url: str, options, proxy) -> tuple:84    if not HAVE_PYTHON_SOCKS:85        raise WebSocketException(86            "Python Socks is needed for SOCKS proxying but is not available"87        )88 89    hostname, port, resource, is_secure = parse_url(url)90 91    if proxy.proxy_protocol == "socks4":92        rdns = False93        proxy_type = ProxyType.SOCKS494    # socks4a sends DNS through proxy95    elif proxy.proxy_protocol == "socks4a":96        rdns = True97        proxy_type = ProxyType.SOCKS498    elif proxy.proxy_protocol == "socks5":99        rdns = False100        proxy_type = ProxyType.SOCKS5101    # socks5h sends DNS through proxy102    elif proxy.proxy_protocol == "socks5h":103        rdns = True104        proxy_type = ProxyType.SOCKS5105 106    ws_proxy = Proxy.create(107        proxy_type=proxy_type,108        host=proxy.proxy_host,109        port=int(proxy.proxy_port),110        username=proxy.auth[0] if proxy.auth else None,111        password=proxy.auth[1] if proxy.auth else None,112        rdns=rdns,113    )114 115    sock = ws_proxy.connect(hostname, port, timeout=proxy.proxy_timeout)116 117    if is_secure:118        if HAVE_SSL:119            sock = _ssl_socket(sock, options.sslopt, hostname)120        else:121            raise WebSocketException("SSL not available.")122 123    return sock, (hostname, port, resource)124 125 126def connect(url: str, options, proxy, socket):127    # Use _start_proxied_socket() only for socks4 or socks5 proxy128    # Use _tunnel() for http proxy129    # TODO: Use python-socks for http protocol also, to standardize flow130    if proxy.proxy_host and not socket and proxy.proxy_protocol != "http":131        return _start_proxied_socket(url, options, proxy)132 133    hostname, port_from_url, resource, is_secure = parse_url(url)134 135    if socket:136        return socket, (hostname, port_from_url, resource)137 138    addrinfo_list, need_tunnel, auth = _get_addrinfo_list(139        hostname, port_from_url, is_secure, proxy140    )141    if not addrinfo_list:142        raise WebSocketException(f"Host not found.: {hostname}:{port_from_url}")143 144    sock = None145    try:146        sock = _open_socket(addrinfo_list, options.sockopt, options.timeout)147        if need_tunnel:148            sock = _tunnel(sock, hostname, port_from_url, auth)149 150        if is_secure:151            if HAVE_SSL:152                sock = _ssl_socket(sock, options.sslopt, hostname)153            else:154                raise WebSocketException("SSL not available.")155 156        return sock, (hostname, port_from_url, resource)157    except:158        if sock:159            sock.close()160        raise161 162 163def _get_addrinfo_list(hostname, port: int, is_secure: bool, proxy) -> tuple:164    phost, pport, pauth = get_proxy_info(165        hostname,166        is_secure,167        proxy.proxy_host,168        proxy.proxy_port,169        proxy.auth,170        proxy.no_proxy,171    )172    try:173        # when running on windows 10, getaddrinfo without socktype returns a socktype 0.174        # This generates an error exception: `_on_error: exception Socket type must be stream or datagram, not 0`175        # or `OSError: [Errno 22] Invalid argument` when creating socket. Force the socket type to SOCK_STREAM.176        if not phost:177            addrinfo_list = socket.getaddrinfo(178                hostname, port, 0, socket.SOCK_STREAM, socket.SOL_TCP179            )180            return addrinfo_list, False, None181        else:182            pport = pport and pport or 80183            # when running on windows 10, the getaddrinfo used above184            # returns a socktype 0. This generates an error exception:185            # _on_error: exception Socket type must be stream or datagram, not 0186            # Force the socket type to SOCK_STREAM187            addrinfo_list = socket.getaddrinfo(188                phost, pport, 0, socket.SOCK_STREAM, socket.SOL_TCP189            )190            return addrinfo_list, True, pauth191    except socket.gaierror as e:192        raise WebSocketAddressException(e)193 194 195def _open_socket(addrinfo_list, sockopt, timeout):196    err = None197    for addrinfo in addrinfo_list:198        family, socktype, proto = addrinfo[:3]199        sock = socket.socket(family, socktype, proto)200        sock.settimeout(timeout)201        for opts in DEFAULT_SOCKET_OPTION:202            sock.setsockopt(*opts)203        for opts in sockopt:204            sock.setsockopt(*opts)205 206        address = addrinfo[4]207        err = None208        while not err:209            try:210                sock.connect(address)211            except socket.error as error:212                sock.close()213                error.remote_ip = str(address[0])214                try:215                    eConnRefused = (216                        errno.ECONNREFUSED,217                        errno.WSAECONNREFUSED,218                        errno.ENETUNREACH,219                    )220                except AttributeError:221                    eConnRefused = (errno.ECONNREFUSED, errno.ENETUNREACH)222                if error.errno not in eConnRefused:223                    raise error224                err = error225                continue226            else:227                break228        else:229            continue230        break231    else:232        if err:233            raise err234 235    return sock236 237 238def _wrap_sni_socket(sock: socket.socket, sslopt: dict, hostname, check_hostname):239    context = sslopt.get("context", None)240    if not context:241        context = ssl.SSLContext(sslopt.get("ssl_version", ssl.PROTOCOL_TLS_CLIENT))242        # Non default context need to manually enable SSLKEYLOGFILE support by setting the keylog_filename attribute.243        # For more details see also:244        # * https://docs.python.org/3.8/library/ssl.html?highlight=sslkeylogfile#context-creation245        # * https://docs.python.org/3.8/library/ssl.html?highlight=sslkeylogfile#ssl.SSLContext.keylog_filename246        keylog_file = os.environ.get("SSLKEYLOGFILE")247        if keylog_file is not None:248            context.keylog_filename = keylog_file249 250        if sslopt.get("cert_reqs", ssl.CERT_NONE) != ssl.CERT_NONE:251            cafile = sslopt.get("ca_certs", None)252            capath = sslopt.get("ca_cert_path", None)253            if cafile or capath:254                try:255                    context.load_verify_locations(cafile=cafile, capath=capath)256                except (FileNotFoundError, ssl.SSLError, ValueError) as e:257                    raise WebSocketException(f"SSL CA certificate loading failed: {e}")258            elif hasattr(context, "load_default_certs"):259                try:260                    context.load_default_certs(ssl.Purpose.SERVER_AUTH)261                except ssl.SSLError as e:262                    raise WebSocketException(263                        f"SSL default certificate loading failed: {e}"264                    )265        if sslopt.get("certfile", None):266            try:267                context.load_cert_chain(268                    sslopt["certfile"],269                    sslopt.get("keyfile", None),270                    sslopt.get("password", None),271                )272            except (FileNotFoundError, ValueError) as e:273                raise WebSocketException(f"SSL client certificate loading failed: {e}")274            except ssl.SSLError as e:275                raise WebSocketException(f"SSL client certificate loading failed: {e}")276 277        # Python 3.10 switch to PROTOCOL_TLS_CLIENT defaults to "cert_reqs = ssl.CERT_REQUIRED" and "check_hostname = True"278        # If both disabled, set check_hostname before verify_mode279        # see https://github.com/liris/websocket-client/commit/b96a2e8fa765753e82eea531adb19716b52ca3ca#commitcomment-10803153280        if sslopt.get("cert_reqs", ssl.CERT_NONE) == ssl.CERT_NONE and not sslopt.get(281            "check_hostname", False282        ):283            context.check_hostname = False284            context.verify_mode = ssl.CERT_NONE285        else:286            context.check_hostname = sslopt.get("check_hostname", True)287            context.verify_mode = sslopt.get("cert_reqs", ssl.CERT_REQUIRED)288 289        if "ciphers" in sslopt:290            try:291                context.set_ciphers(sslopt["ciphers"])292            except ssl.SSLError as e:293                raise WebSocketException(f"SSL cipher configuration failed: {e}")294        if "cert_chain" in sslopt:295            try:296                cert_chain = sslopt["cert_chain"]297                if not isinstance(cert_chain, (tuple, list)) or len(cert_chain) != 3:298                    raise ValueError(299                        "cert_chain must be a tuple/list of (certfile, keyfile, password)"300                    )301                certfile, keyfile, password = cert_chain302                context.load_cert_chain(certfile, keyfile, password)303            except ValueError:304                raise305            except (FileNotFoundError, ssl.SSLError) as e:306                raise WebSocketException(307                    f"SSL client certificate configuration failed: {e}"308                )309        if "ecdh_curve" in sslopt:310            try:311                context.set_ecdh_curve(sslopt["ecdh_curve"])312            except ValueError as e:313                raise WebSocketException(f"SSL ECDH curve configuration failed: {e}")314 315    return context.wrap_socket(316        sock,317        do_handshake_on_connect=sslopt.get("do_handshake_on_connect", True),318        suppress_ragged_eofs=sslopt.get("suppress_ragged_eofs", True),319        server_hostname=hostname,320    )321 322 323def _ssl_socket(sock: socket.socket, user_sslopt: dict, hostname):324    sslopt: dict = {"cert_reqs": ssl.CERT_REQUIRED}325    sslopt.update(user_sslopt)326 327    cert_path = os.environ.get("WEBSOCKET_CLIENT_CA_BUNDLE")328    if (329        cert_path330        and os.path.isfile(cert_path)331        and user_sslopt.get("ca_certs", None) is None332    ):333        sslopt["ca_certs"] = cert_path334    elif (335        cert_path336        and os.path.isdir(cert_path)337        and user_sslopt.get("ca_cert_path", None) is None338    ):339        sslopt["ca_cert_path"] = cert_path340 341    if sslopt.get("server_hostname", None):342        hostname = sslopt["server_hostname"]343 344    check_hostname = sslopt.get("check_hostname", True)345    sock = _wrap_sni_socket(sock, sslopt, hostname, check_hostname)346 347    return sock348 349 350def _tunnel(sock: socket.socket, host, port: int, auth) -> socket.socket:351    debug("Connecting proxy...")352    connect_header = f"CONNECT {host}:{port} HTTP/1.1\r\n"353    connect_header += f"Host: {host}:{port}\r\n"354 355    # TODO: support digest auth.356    if auth and auth[0]:357        auth_str = auth[0]358        if auth[1]:359            auth_str += f":{auth[1]}"360        encoded_str = base64encode(auth_str.encode()).strip().decode().replace("\n", "")361        connect_header += f"Proxy-Authorization: Basic {encoded_str}\r\n"362    connect_header += "\r\n"363    dump("request header", connect_header)364 365    send(sock, connect_header)366 367    try:368        status, _, _ = read_headers(sock)369    except (socket.error, WebSocketException) as e:370        raise WebSocketProxyException(str(e))371 372    if status != 200:373        raise WebSocketProxyException(f"failed CONNECT via proxy status: {status}")374 375    return sock376 377 378def read_headers(sock: socket.socket) -> tuple:379    status = None380    status_message = None381    headers: dict = {}382    trace("--- response header ---")383 384    while True:385        line = recv_line(sock)386        line = line.decode("utf-8").strip()387        if not line:388            break389        trace(line)390        if not status:391            status_info = line.split(" ", 2)392            status = int(status_info[1])393            if len(status_info) > 2:394                status_message = status_info[2]395        else:396            kv = line.split(":", 1)397            if len(kv) != 2:398                raise WebSocketException("Invalid header")399            key, value = kv400            if key.lower() == "set-cookie" and headers.get("set-cookie"):401                existing_cookie = headers.get("set-cookie")402                if existing_cookie is not None:403                    headers["set-cookie"] = existing_cookie + "; " + value.strip()404                else:405                    headers["set-cookie"] = value.strip()406            else:407                headers[key.lower()] = value.strip()408 409    trace("-----------------------")410 411    return status, headers, status_message412