| """ |
| websocket - WebSocket client library for Python |
| |
| Copyright (C) 2010 Hiroki Ohtani(liris) |
| |
| This library is free software; you can redistribute it and/or |
| modify it under the terms of the GNU Lesser General Public |
| License as published by the Free Software Foundation; either |
| version 2.1 of the License, or (at your option) any later version. |
| |
| This library is distributed in the hope that it will be useful, |
| but WITHOUT ANY WARRANTY; without even the implied warranty of |
| MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU |
| Lesser General Public License for more details. |
| |
| You should have received a copy of the GNU Lesser General Public |
| License along with this library; if not, write to the Free Software |
| Foundation, Inc., 51 Franklin Street, Fifth Floor, |
| Boston, MA 02110-1335 USA |
| |
| """ |
| |
| from six.moves.urllib.parse import urlparse |
| import os |
| |
| __all__ = ["parse_url", "get_proxy_info"] |
| |
| |
| def parse_url(url): |
| """ |
| parse url and the result is tuple of |
| (hostname, port, resource path and the flag of secure mode) |
| |
| url: url string. |
| """ |
| if ":" not in url: |
| raise ValueError("url is invalid") |
| |
| scheme, url = url.split(":", 1) |
| |
| parsed = urlparse(url, scheme="ws") |
| 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 += "?" + parsed.query |
| |
| return (hostname, port, resource, is_secure) |
| |
| |
| DEFAULT_NO_PROXY_HOST = ["localhost", "127.0.0.1"] |
| |
| |
| def _is_no_proxy_host(hostname, no_proxy): |
| if not no_proxy: |
| v = os.environ.get("no_proxy", "").replace(" ", "") |
| no_proxy = v.split(",") |
| if not no_proxy: |
| no_proxy = DEFAULT_NO_PROXY_HOST |
| |
| return hostname in no_proxy |
| |
| |
| def get_proxy_info(hostname, is_secure, |
| proxy_host=None, proxy_port=0, proxy_auth=None, no_proxy=None): |
| """ |
| 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. |
| |
| hostname: websocket server name. |
| |
| is_secure: is the connection secure? (wss) |
| looks for "https_proxy" in env |
| before falling back to "http_proxy" |
| |
| options: "http_proxy_host" - http proxy host name. |
| "http_proxy_port" - http proxy port. |
| "http_no_proxy" - host names, which doesn't use proxy. |
| "http_proxy_auth" - http proxy auth information. |
| tuple of username and password. |
| default is None |
| """ |
| if _is_no_proxy_host(hostname, no_proxy): |
| return None, 0, None |
| |
| if proxy_host: |
| port = proxy_port |
| auth = proxy_auth |
| return proxy_host, port, auth |
| |
| env_keys = ["http_proxy"] |
| if is_secure: |
| env_keys.insert(0, "https_proxy") |
| |
| for key in env_keys: |
| value = os.environ.get(key, None) |
| if value: |
| proxy = urlparse(value) |
| auth = (proxy.username, proxy.password) if proxy.username else None |
| return proxy.hostname, proxy.port, auth |
| |
| return None, 0, None |
