1""" 2websocket - WebSocket client library for Python 3 4Copyright (C) 2010 Hiroki Ohtani(liris) 5 6 This library is free software; you can redistribute it and/or 7 modify it under the terms of the GNU Lesser General Public 8 License as published by the Free Software Foundation; either 9 version 2.1 of the License, or (at your option) any later version. 10 11 This library is distributed in the hope that it will be useful, 12 but WITHOUT ANY WARRANTY; without even the implied warranty of 13 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU 14 Lesser General Public License for more details. 15 16 You should have received a copy of the GNU Lesser General Public 17 License along with this library; if not, write to the Free Software 18 Foundation, Inc., 51 Franklin Street, Fifth Floor, 19 Boston, MA 02110-1335 USA 20 21""" 22 23import os 24import socket 25import struct 26 27from six.moves.urllib.parse import urlparse 28 29 30__all__ = ["parse_url", "get_proxy_info"] 31 32 33def parse_url(url): 34 """ 35 parse url and the result is tuple of 36 (hostname, port, resource path and the flag of secure mode) 37 38 url: url string. 39 """ 40 if ":" not in url: 41 raise ValueError("url is invalid") 42 43 scheme, url = url.split(":", 1) 44 45 parsed = urlparse(url, scheme="ws") 46 if parsed.hostname: 47 hostname = parsed.hostname 48 else: 49 raise ValueError("hostname is invalid") 50 port = 0 51 if parsed.port: 52 port = parsed.port 53 54 is_secure = False 55 if scheme == "ws": 56 if not port: 57 port = 80 58 elif scheme == "wss": 59 is_secure = True 60 if not port: 61 port = 443 62 else: 63 raise ValueError("scheme %s is invalid" % scheme) 64 65 if parsed.path: 66 resource = parsed.path 67 else: 68 resource = "/" 69 70 if parsed.query: 71 resource += "?" + parsed.query 72 73 return hostname, port, resource, is_secure 74 75 76DEFAULT_NO_PROXY_HOST = ["localhost", "127.0.0.1"] 77 78 79def _is_ip_address(addr): 80 try: 81 socket.inet_aton(addr) 82 except socket.error: 83 return False 84 else: 85 return True 86 87 88def _is_subnet_address(hostname): 89 try: 90 addr, netmask = hostname.split("/") 91 return _is_ip_address(addr) and 0 <= int(netmask) < 32 92 except ValueError: 93 return False 94 95 96def _is_address_in_network(ip, net): 97 ipaddr = struct.unpack('I', socket.inet_aton(ip))[0] 98 netaddr, bits = net.split('/') 99 netmask = struct.unpack('I', socket.inet_aton(netaddr))[0] & ((2 << int(bits) - 1) - 1) 100 return ipaddr & netmask == netmask 101 102 103def _is_no_proxy_host(hostname, no_proxy): 104 if not no_proxy: 105 v = os.environ.get("no_proxy", "").replace(" ", "") 106 no_proxy = v.split(",") 107 if not no_proxy: 108 no_proxy = DEFAULT_NO_PROXY_HOST 109 110 if hostname in no_proxy: 111 return True 112 elif _is_ip_address(hostname): 113 return any([_is_address_in_network(hostname, subnet) for subnet in no_proxy if _is_subnet_address(subnet)]) 114 115 return False 116 117 118def get_proxy_info( 119 hostname, is_secure, proxy_host=None, proxy_port=0, proxy_auth=None, 120 no_proxy=None): 121 """ 122 try to retrieve proxy host and port from environment 123 if not provided in options. 124 result is (proxy_host, proxy_port, proxy_auth). 125 proxy_auth is tuple of username and password 126 of proxy authentication information. 127 128 hostname: websocket server name. 129 130 is_secure: is the connection secure? (wss) 131 looks for "https_proxy" in env 132 before falling back to "http_proxy" 133 134 options: "http_proxy_host" - http proxy host name. 135 "http_proxy_port" - http proxy port. 136 "http_no_proxy" - host names, which doesn't use proxy. 137 "http_proxy_auth" - http proxy auth information. 138 tuple of username and password. 139 default is None 140 """ 141 if _is_no_proxy_host(hostname, no_proxy): 142 return None, 0, None 143 144 if proxy_host: 145 port = proxy_port 146 auth = proxy_auth 147 return proxy_host, port, auth 148 149 env_keys = ["http_proxy"] 150 if is_secure: 151 env_keys.insert(0, "https_proxy") 152 153 for key in env_keys: 154 value = os.environ.get(key, None) 155 if value: 156 proxy = urlparse(value) 157 auth = (proxy.username, proxy.password) if proxy.username else None 158 return proxy.hostname, proxy.port, auth 159 160 return None, 0, None 161