From 82b0698e55f591173a5afbd3df6048de60a4d987 Mon Sep 17 00:00:00 2001 From: Antoine Martin Date: Tue, 2 Aug 2022 17:29:41 +0200 Subject: [PATCH] #2105 detect proxy settings using libproxy --- xpra/net/ssh.py | 2 +- xpra/scripts/main.py | 2 +- xpra/scripts/parsing.py | 81 ++++++++++++++--------------------------- 3 files changed, 30 insertions(+), 55 deletions(-) diff --git a/xpra/net/ssh.py b/xpra/net/ssh.py index 40c9dd931b..f869a7844e 100644 --- a/xpra/net/ssh.py +++ b/xpra/net/ssh.py @@ -264,7 +264,7 @@ def proxycommand_ended(proc): from xpra.net.socket_util import socket_connect if "proxy_host" in display_desc: proxy_host = display_desc["proxy_host"] - proxy_port = display_desc.get("proxy_port", 22) + proxy_port = int(display_desc.get("proxy_port", 22)) proxy_username = display_desc.get("proxy_username", username) proxy_password = display_desc.get("proxy_password", password) proxy_keys = get_keyfiles(host_config, "proxy_key") diff --git a/xpra/scripts/main.py b/xpra/scripts/main.py index 6b8a34d029..0e67a58d9a 100755 --- a/xpra/scripts/main.py +++ b/xpra/scripts/main.py @@ -895,7 +895,7 @@ def proxy_connect(options): proxy_type = { "SOCKS5" : socks.SOCKS5, "SOCKS4" : socks.SOCKS4, - }.get(ptype) + }.get(ptype, socks.SOCKS5) if not proxy_type: raise InitExit(EXIT_UNSUPPORTED, f"unsupported proxy type {ptype!r}") host = to.strget("proxy-host") diff --git a/xpra/scripts/parsing.py b/xpra/scripts/parsing.py index f1bfaf3308..7f50c95db3 100755 --- a/xpra/scripts/parsing.py +++ b/xpra/scripts/parsing.py @@ -190,58 +190,31 @@ def _sep_pos(display_name): return scpos return min(scpos, slpos) -def parse_proxy_attributes(display_name): - import re - # Notes: - # (1) this regex permits a "?" in the password or username (because not just splitting at "?"). - # It doesn't look for the next "?" until after the "@", where a "?" really indicates - # another field. - # (2) all characters including "@"s go to "userpass" until the *last* "@" after which it all goes - # to "hostport" - reout = re.search("\\?proxy=(?P

((?P.+)@)?(?P[^?]+))", display_name) - if not reout: - return display_name, {} + +def auto_proxy(scheme, host): try: - desc_tmp = {} - # This one should *always* return a host, and should end with an optional numeric port - hostport = reout.group("hostport") - hostport_match = re.match(r"(?P[^:]+)($|:(?P\d+)$)", hostport) - if not hostport_match: - raise RuntimeError("bad format for 'hostport': '%s'" % hostport) - host = hostport_match.group("host") - if not host: - raise RuntimeError("bad format: missing host in '%s'" % hostport) - desc_tmp["proxy_host"] = host - if hostport_match.group("port"): - port_str = hostport_match.group("port") - try: - desc_tmp["proxy_port"] = int(port_str) - except ValueError: - raise RuntimeError("bad format: proxy port '%s' is not a number" % port_str) from None - userpass = reout.group("userpass") - if userpass: - # The username ends at the first colon. This decision was not unique: I could have - # allowed one colon in username if there were two in the string. - userpass_match = re.match("(?P[^:]+)(:(?P.+))?", userpass) - if not userpass_match: - raise RuntimeError("bad format for 'userpass': '%s'" % userpass) - # If there is a "userpass" part, then it *must* have a username - username = userpass_match.group("username") - if not username: - raise RuntimeError("bad format: missing username in '%s'" % userpass) - desc_tmp["proxy_username"] = username - password = userpass_match.group("password") - if password: - desc_tmp["proxy_password"] = password - except RuntimeError: - from xpra.log import Logger - sshlog = Logger("ssh") - sshlog.error("bad proxy argument: " + reout.group(0)) - return display_name, {} - else: - # rip out the part we've processed - display_name = display_name[:reout.start()] + display_name[reout.end():] - return display_name, desc_tmp + from xpra.net.libproxy import ProxyFactory + except ImportError as e: + warn("Warning: unable to detect proxy settings") + warn(f" {e}") + return {} + p = ProxyFactory() + proxies = p.getProxies("%s://%s" % (scheme, host)) + if not proxies or proxies[0]=="direct://": + return {} + #for the time being, just try the first one: + from urllib.parse import urlparse + url = urlparse(proxies[0]) + if not url.scheme or not url.netloc: + return {} + options = {"proxy-host" : url.hostname} + if url.port: + options["proxy-port"] = url.port + if url.username: + options["proxy-username"] = url.username + if url.password: + options["proxy-password"] = url.password + return options def parse_remote_display(s): if not s: @@ -405,8 +378,6 @@ def parse_display_name(error_cb, opts, display_name, cmdline=(), find_session_by "display_name" : display_name, "cmdline" : cmdline, } - display_name, proxy_attrs = parse_proxy_attributes(display_name) - desc.update(proxy_attrs) pos = _sep_pos(display_name) if pos<0 or (display_name and display_name[0] in "0123456789"): @@ -650,6 +621,10 @@ def _parse_remote_display(s): if desc.get("strict-host-check") is False: ssl_options["server-verify-mode"] = "none" desc["ssl-options"] = ssl_options + proxy = desc.get("proxy") + if proxy=="auto": + proxy_options = auto_proxy(desc["host"], desc["port"]) + desc.update(proxy_options) return desc if protocol=="vsock":