diff options
author | robot-piglet <robot-piglet@yandex-team.com> | 2024-12-09 18:25:21 +0300 |
---|---|---|
committer | robot-piglet <robot-piglet@yandex-team.com> | 2024-12-09 19:18:57 +0300 |
commit | 13374e0884578812cda7697d0c5680122db59a37 (patch) | |
tree | 30a022eb841035299deb2b8c393b2902f0c21735 /contrib/python/websocket-client/py3/websocket/_url.py | |
parent | c7ade6d3bf7cd492235a61b77153351e422a28f3 (diff) | |
download | ydb-13374e0884578812cda7697d0c5680122db59a37.tar.gz |
Intermediate changes
commit_hash:034150f557268506d7bc0cbd8b5becf65f765593
Diffstat (limited to 'contrib/python/websocket-client/py3/websocket/_url.py')
-rw-r--r-- | contrib/python/websocket-client/py3/websocket/_url.py | 190 |
1 files changed, 0 insertions, 190 deletions
diff --git a/contrib/python/websocket-client/py3/websocket/_url.py b/contrib/python/websocket-client/py3/websocket/_url.py deleted file mode 100644 index 902131710ba..00000000000 --- a/contrib/python/websocket-client/py3/websocket/_url.py +++ /dev/null @@ -1,190 +0,0 @@ -import os -import socket -import struct -from typing import Optional -from urllib.parse import unquote, urlparse -from ._exceptions import WebSocketProxyException - -""" -_url.py -websocket - WebSocket client library for Python - -Copyright 2024 engn33r - -Licensed under the Apache License, Version 2.0 (the "License"); -you may not use this file except in compliance with the License. -You may obtain a copy of the License at - - http://www.apache.org/licenses/LICENSE-2.0 - -Unless required by applicable law or agreed to in writing, software -distributed under the License is distributed on an "AS IS" BASIS, -WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -See the License for the specific language governing permissions and -limitations under the License. -""" - -__all__ = ["parse_url", "get_proxy_info"] - - -def parse_url(url: str) -> tuple: - """ - parse url and the result is tuple of - (hostname, port, resource path and the flag of secure mode) - - Parameters - ---------- - url: str - url string. - """ - if ":" not in url: - raise ValueError("url is invalid") - - scheme, url = url.split(":", 1) - - parsed = urlparse(url, scheme="http") - if parsed.hostname: - hostname = parsed.hostname - else: - raise ValueError("hostname is invalid") - port = 0 - if parsed.port: - port = parsed.port - - is_secure = False - if scheme == "ws": - if not port: - port = 80 - elif scheme == "wss": - is_secure = True - if not port: - port = 443 - else: - raise ValueError("scheme %s is invalid" % scheme) - - if parsed.path: - resource = parsed.path - else: - resource = "/" - - if parsed.query: - resource += f"?{parsed.query}" - - return hostname, port, resource, is_secure - - -DEFAULT_NO_PROXY_HOST = ["localhost", "127.0.0.1"] - - -def _is_ip_address(addr: str) -> bool: - try: - socket.inet_aton(addr) - except socket.error: - return False - else: - return True - - -def _is_subnet_address(hostname: str) -> bool: - try: - addr, netmask = hostname.split("/") - return _is_ip_address(addr) and 0 <= int(netmask) < 32 - except ValueError: - return False - - -def _is_address_in_network(ip: str, net: str) -> bool: - ipaddr: int = struct.unpack("!I", socket.inet_aton(ip))[0] - netaddr, netmask = net.split("/") - netaddr: int = struct.unpack("!I", socket.inet_aton(netaddr))[0] - - netmask = (0xFFFFFFFF << (32 - int(netmask))) & 0xFFFFFFFF - return ipaddr & netmask == netaddr - - -def _is_no_proxy_host(hostname: str, no_proxy: Optional[list]) -> bool: - if not no_proxy: - if v := os.environ.get("no_proxy", os.environ.get("NO_PROXY", "")).replace( - " ", "" - ): - no_proxy = v.split(",") - if not no_proxy: - no_proxy = DEFAULT_NO_PROXY_HOST - - if "*" in no_proxy: - return True - if hostname in no_proxy: - return True - if _is_ip_address(hostname): - return any( - [ - _is_address_in_network(hostname, subnet) - for subnet in no_proxy - if _is_subnet_address(subnet) - ] - ) - for domain in [domain for domain in no_proxy if domain.startswith(".")]: - if hostname.endswith(domain): - return True - return False - - -def get_proxy_info( - hostname: str, - is_secure: bool, - proxy_host: Optional[str] = None, - proxy_port: int = 0, - proxy_auth: Optional[tuple] = None, - no_proxy: Optional[list] = None, - proxy_type: str = "http", -) -> tuple: - """ - Try to retrieve proxy host and port from environment - if not provided in options. - Result is (proxy_host, proxy_port, proxy_auth). - proxy_auth is tuple of username and password - of proxy authentication information. - - Parameters - ---------- - hostname: str - Websocket server name. - is_secure: bool - Is the connection secure? (wss) looks for "https_proxy" in env - instead of "http_proxy" - proxy_host: str - http proxy host name. - proxy_port: str or int - http proxy port. - no_proxy: list - Whitelisted host names that don't use the proxy. - proxy_auth: tuple - HTTP proxy auth information. Tuple of username and password. Default is None. - proxy_type: str - Specify the proxy protocol (http, socks4, socks4a, socks5, socks5h). Default is "http". - Use socks4a or socks5h if you want to send DNS requests through the proxy. - """ - if _is_no_proxy_host(hostname, no_proxy): - return None, 0, None - - if proxy_host: - if not proxy_port: - raise WebSocketProxyException("Cannot use port 0 when proxy_host specified") - port = proxy_port - auth = proxy_auth - return proxy_host, port, auth - - env_key = "https_proxy" if is_secure else "http_proxy" - value = os.environ.get(env_key, os.environ.get(env_key.upper(), "")).replace( - " ", "" - ) - if value: - proxy = urlparse(value) - auth = ( - (unquote(proxy.username), unquote(proxy.password)) - if proxy.username - else None - ) - return proxy.hostname, proxy.port, auth - - return None, 0, None |