RHEL-13375: 1.28 Parse URL properly

src/rhsm/connection.py

@@ -164,11 +164,15 @@ class ConnectionOSErrorException(ConnectionException):
     """
 
     def __init__(self, host: str, port: int, handler: str, exc: OSError):
-        self.host = host
+        self._host = host
         self.port = port
         self.handler = handler
         self.exc = exc
 
+    @property
+    def host(self) -> str:
+        return normalized_host(self._host)
+
 
 class BaseConnection(object):
     def __init__(
@@ -305,8 +309,12 @@ def __init__(
         connection_description = ""
         if proxy_description:
             connection_description += proxy_description
-        connection_description += "host=%s port=%s handler=%s %s" % (self.host, self.ssl_port,
-                                                                     self.handler, auth_description)
+        connection_description += "host=%s port=%s handler=%s %s" % (
+            normalized_host(self.host),
+            safe_int(self.ssl_port),
+            self.handler,
+            auth_description,
+        )
         log.debug("Connection built: %s", connection_description)
 
 
@@ -679,6 +687,7 @@ def _print_debug_info_about_request(self, request_type, handler, final_headers,
 
         if 'SUBMAN_DEBUG_PRINT_REQUEST' in os.environ:
             yellow_col = '\033[93m'
+            magenta_col = "\033[95m"
             blue_col = '\033[94m'
             green_col = '\033[92m'
             red_col = '\033[91m'
@@ -687,9 +696,24 @@ def _print_debug_info_about_request(self, request_type, handler, final_headers,
                 msg = blue_col + "Making insecure request:" + end_col
             else:
                 msg = blue_col + "Making request:" + end_col
-            msg += red_col + " %s:%s %s %s" % (self.host, self.ssl_port, request_type, handler) + end_col
+            msg += (
+                    red_col +
+                    " https://" +
+                    f"{normalized_host(self.host)}:{safe_int(self.ssl_port)}{handler} {request_type}" +
+                    end_col
+            )
             if self.proxy_hostname and self.proxy_port:
-                msg += blue_col + " using proxy " + red_col + f"{self.proxy_hostname}:{self.proxy_port}" + end_col
+                # Note: using only https:// is not a mistake. We use only https for proxy connection.
+                msg += blue_col + " Using proxy: " + magenta_col + "https://"
+                # Print username and eventually password
+                if self.proxy_user:
+                    if self.proxy_user and self.proxy_password:
+                        msg += f"{self.proxy_user}:{self.proxy_password}@"
+                    elif self.proxy_user and not self.proxy_password:
+                        msg += f"{self.proxy_user}@"
+                # Print hostname and port
+                msg += f"{normalized_host(self.proxy_hostname)}:{safe_int(self.proxy_port)}"
+                msg += end_col
             if 'SUBMAN_DEBUG_PRINT_REQUEST_HEADER' in os.environ:
                 msg += blue_col + " %s" % final_headers + end_col
             if 'SUBMAN_DEBUG_PRINT_REQUEST_BODY' in os.environ and body is not None:
@@ -810,7 +834,7 @@ def _request(self, request_type, method, info=None, headers=None, cert_key_pairs
                     break  # this client cert worked, no need to try more
                 elif self.cert_dir:
                     log.debug("Unable to get valid response: %s from CDN: %s" %
-                              (result, self.host))
+                              (result, normalized_host(self.host)))
 
             except ssl.SSLError:
                 if self.cert_file and not self.cert_dir:
@@ -849,7 +873,7 @@ def _request(self, request_type, method, info=None, headers=None, cert_key_pairs
             if self.cert_dir:
                 raise NoValidEntitlement(
                     "Cannot access CDN content on: %s using any of entitlement cert-key pair: %s" %
-                    (self.host, cert_key_pairs)
+                    (normalized_host(self.host), cert_key_pairs)
                 )
 
         self._print_debug_info_about_response(result)

src/rhsm/utils.py

@@ -116,7 +116,6 @@ def parse_url(local_server_entry,
     :param default_port: default_port
     :return: a tuple of (username, password, hostname, port, path)
     """
-    # Adding http:// onto the front of the hostname
 
     if local_server_entry == "":
         raise ServerUrlParseErrorEmpty(local_server_entry)
@@ -142,65 +141,37 @@ def parse_url(local_server_entry,
     if not good_url:
         good_url = "https://%s" % local_server_entry
 
-    #FIXME: need a try except here? docs
-    # don't seem to indicate any expected exceptions
+    # FIXME: need a try except here? docs
+    #        don't seem to indicate any expected exceptions
     result = six.moves.urllib.parse.urlparse(good_url)
-    username = default_username
-    password = default_password
-    #netloc = result[1].split(":")
-
-    # to support username and password, let's split on @
-    # since the format will be username:password@hostname:port
-    foo = result[1].split("@")
-
-    # handle username/password portion, then deal with host:port
-    # just in case someone passed in @hostname without
-    # a username,  we default to the default_username
-    if len(foo) > 1:
-        creds = foo[0].split(":")
-        netloc = foo[1].split(":")
-
-        if len(creds) > 1:
-            password = creds[1]
-        if creds[0] is not None and len(creds[0]) > 0:
-            username = creds[0]
-    else:
-        netloc = foo[0].split(":")
-
-    # in some cases, if we try the attr accessors, we'll
-    # get a ValueError deep down in urlparse, particular if
-    # port ends up being None
-    #
-    # So maybe check result.port/path/hostname for None, and
-    # throw an exception in those cases.
-    # adding the schem seems to avoid this though
-    port = default_port
-    if len(netloc) > 1:
-        if netloc[1] != "":
-            port = str(netloc[1])
-        else:
-            raise ServerUrlParseErrorPort(local_server_entry)
 
-    # path can be None?
-    prefix = default_prefix
-    if result[2] is not None:
-        if result[2] != '':
-            prefix = result[2]
+    username = result.username
+    if username is None or username == "":
+        username = default_username
 
-    hostname = default_hostname
-    if netloc[0] is not None:
-        if netloc[0] != "":
-            hostname = netloc[0]
+    password = result.password
+    if password is None:
+        password = default_password
+
+    hostname = result.hostname
+    if hostname is None:
+        hostname = default_hostname
 
     try:
-        if port:
-            int(port)
-    except TypeError:
-        raise ServerUrlParseErrorPort(local_server_entry)
+        port = result.port
     except ValueError:
         raise ServerUrlParseErrorPort(local_server_entry)
 
-    return (username, password, hostname, port, prefix)
+    if port is None:
+        port = default_port
+    else:
+        port = str(port)
+
+    prefix = result.path
+    if prefix is None or prefix == "":
+        prefix = default_prefix
+
+    return username, password, hostname, port, prefix
 
 
 def get_env_proxy_info():

src/subscription_manager/managercli.py

@@ -40,7 +40,7 @@
 import rhsm.config
 import rhsm.connection as connection
 from rhsm.connection import ProxyException, UnauthorizedException, ConnectionException, RemoteServerException, ConnectionOSErrorException
-from rhsm.utils import cmd_name, remove_scheme, ServerUrlParseError
+from rhsm.utils import cmd_name, ServerUrlParseError
 
 from subscription_manager import identity
 from subscription_manager.branding import get_branding
@@ -462,16 +462,15 @@ def main(self, args=None):
                 baseurl_server_prefix)
             config_changed = True
 
-        # support foo.example.com:3128 format
         if hasattr(self.options, "proxy_url") and self.options.proxy_url:
-            parts = remove_scheme(self.options.proxy_url).split(':')
-            self.proxy_hostname = parts[0]
-            # no ':'
-            if len(parts) > 1:
-                self.proxy_port = int(parts[1])
-            else:
-                # if no port specified, use the one from the config, or fallback to the default
-                self.proxy_port = conf['server'].get_int('proxy_port') or rhsm.config.DEFAULT_PROXY_PORT
+            default_proxy_port = conf["server"].get_int("proxy_port") or rhsm.config.DEFAULT_PROXY_PORT
+            proxy_user, proxy_pass, proxy_hostname, proxy_port, proxy_prefix = rhsm.utils.parse_url(
+                self.options.proxy_url, default_port=default_proxy_port
+            )
+            self.proxy_user = proxy_user
+            self.proxy_password = proxy_pass
+            self.proxy_hostname = proxy_hostname
+            self.proxy_port = int(proxy_port)
             config_changed = True
 
         if hasattr(self.options, "proxy_user") and self.options.proxy_user:
@@ -526,7 +525,7 @@ def main(self, args=None):
                     # this tries to actually connect to the server and ping it
                     if not is_valid_server_info(self.no_auth_cp):
                         system_exit(os.EX_UNAVAILABLE, _("Unable to reach the server at %s:%s%s") % (
-                            self.no_auth_cp.host,
+                            connection.normalized_host(self.no_auth_cp.host),
                             self.no_auth_cp.ssl_port,
                             self.no_auth_cp.handler
                         ))

src/subscription_manager/utils.py

@@ -97,6 +97,11 @@ def format_baseurl(hostname, port, prefix):
     if prefix == DEFAULT_CDN_PREFIX:
         prefix = prefix[:-1]
 
+    # Handle raw IPv6 addresses.
+    # RFC 1035 only allows characters `a-zA-Z0-9.`, we can do this.
+    if ":" in hostname:
+        hostname = f"[{hostname}]"
+
     # just so we match how we format this by
     # default
     if port == DEFAULT_CDN_PORT:

test/rhsm/unit/test_connection.py

@@ -165,7 +165,7 @@ def mock_config_without_proxy_settings(section, name):
     def test_https_proxy_info_allcaps(self):
         with patch.dict('os.environ', {'HTTPS_PROXY': 'http://u:p@host:4444'}):
             with patch.object(connection.config, 'get', self.mock_config_without_proxy_settings):
-                uep = UEPConnection(username="dummy", password="dummy",
+                uep = UEPConnection(host="dummy", username="dummy", password="dummy",
                     handler="/Test/", insecure=True)
                 self.assertEqual("u", uep.proxy_user)
                 self.assertEqual("p", uep.proxy_password)
@@ -177,7 +177,7 @@ def test_order(self):
         with patch.dict('os.environ', {'HTTPS_PROXY': 'http://u:p@host:4444',
                                        'http_proxy': 'http://notme:orme@host:2222'}):
             with patch.object(connection.config, 'get', self.mock_config_without_proxy_settings):
-                uep = UEPConnection(username="dummy", password="dummy",
+                uep = UEPConnection(host="dummy", username="dummy", password="dummy",
                     handler="/Test/", insecure=True)
                 self.assertEqual("u", uep.proxy_user)
                 self.assertEqual("p", uep.proxy_password)
@@ -187,7 +187,7 @@ def test_order(self):
     def test_no_port(self):
         with patch.dict('os.environ', {'HTTPS_PROXY': 'http://u:p@host'}):
             with patch.object(connection.config, 'get', self.mock_config_without_proxy_settings):
-                uep = UEPConnection(username="dummy", password="dummy",
+                uep = UEPConnection(host="dummy", username="dummy", password="dummy",
                     handler="/Test/", insecure=True)
                 self.assertEqual("u", uep.proxy_user)
                 self.assertEqual("p", uep.proxy_password)
@@ -197,7 +197,7 @@ def test_no_port(self):
     def test_no_user_or_password(self):
         with patch.dict('os.environ', {'HTTPS_PROXY': 'http://host:1111'}):
             with patch.object(connection.config, 'get', self.mock_config_without_proxy_settings):
-                uep = UEPConnection(username="dummy", password="dummy",
+                uep = UEPConnection(host="dummy", username="dummy", password="dummy",
                     handler="/Test/", insecure=True)
                 self.assertEqual(None, uep.proxy_user)
                 self.assertEqual(None, uep.proxy_password)

test/rhsm/unit/test_util.py

@@ -137,16 +137,24 @@ def test_bad_http_scheme(self):
                           local_url)
 
     def test_colon_but_no_port(self):
+        # This is correct URL
         local_url = "https://myhost.example.com:/myapp"
-        self.assertRaises(ServerUrlParseErrorPort,
-                          parse_url,
-                          local_url)
+        username, password, hostname, port, prefix = parse_url(local_url)
+        self.assertIsNone(username)
+        self.assertIsNone(password)
+        self.assertEqual(hostname, "myhost.example.com")
+        self.assertIsNone(port)
+        self.assertEqual(prefix, "/myapp")
 
     def test_colon_but_no_port_no_scheme(self):
+        # This is correct URL
         local_url = "myhost.example.com:/myapp"
-        self.assertRaises(ServerUrlParseErrorPort,
-                          parse_url,
-                          local_url)
+        username, password, hostname, port, prefix = parse_url(local_url)
+        self.assertIsNone(username)
+        self.assertIsNone(password)
+        self.assertEqual(hostname, "myhost.example.com")
+        self.assertIsNone(port)
+        self.assertEqual(prefix, "/myapp")
 
     def test_colon_slash_slash_but_nothing_else(self):
         local_url = "http://"
@@ -273,6 +281,24 @@ def test_bad(self):
 
 class TestParseUrl(unittest.TestCase):
 
+    def test_ipv4_url(self):
+        local_url = "http://user:[email protected]:3128/prefix"
+        (username, password, hostname, port, prefix) = parse_url(local_url)
+        self.assertEqual("user", username)
+        self.assertEqual("pass", password)
+        self.assertEqual("192.168.0.10", hostname)
+        self.assertEqual("3128", port)
+        self.assertEqual("/prefix", prefix)
+
+    def test_ipv6_url(self):
+        local_url = "http://user:pass@[2001:db8::dead:beef:1]:3128/prefix"
+        (username, password, hostname, port, prefix) = parse_url(local_url)
+        self.assertEqual("user", username)
+        self.assertEqual("pass", password)
+        self.assertEqual("2001:db8::dead:beef:1", hostname)
+        self.assertEqual("3128", port)
+        self.assertEqual("/prefix", prefix)
+
     def test_username_password(self):
         local_url = "http://user:pass@hostname:1111/prefix"
         (username, password, hostname, port, prefix) = parse_url(local_url)

test/test_managercli.py

@@ -2484,6 +2484,24 @@ def test_list_by_default_with_options_from_super_class(self):
         self.cc._validate_options()
         self.assertTrue(self.cc.options.list)
 
+    def test_proxy_user_and_pass_from_url_overridden_by_cli_options(self):
+        """
+        Test that --proxyuser and --proxypassword have higher priority than --proxy
+        """
+        self.cc.main(
+            [
+                "--proxy",
+                "https://foo:[email protected]",
+                "--proxyuser",
+                "other-user",
+                "--proxypassword",
+                "other-password",
+            ]
+        )
+        self.cc._validate_options()
+        self.assertEqual(self.cc.proxy_user, "other-user")
+        self.assertEqual(self.cc.proxy_password, "other-password")
+
     def test_add_with_multiple_colons(self):
         self.cc.main(["--repo", "x", "--add", "url:http://example.com"])
         self.cc._validate_options()