Makes NO_PROXY, HTTP_PROXY, HTTPS_PROXY, PYMANAGER_USERNAME and _PASSWORD values configurable in settings

src/manage/commands.py

@@ -256,6 +256,20 @@ def execute(self):
     # Typically configured to '%VIRTUAL_ENV%' to pick up the active environment
     "virtual_env": (str, None, "env", "path"),
 
+    # Overrides for the OS proxy configuration, typically set to '%HTTP_PROXY%'
+    # by default to pick up environment settings, but can be overridden to empty
+    # to disable use of the process environment.
+    "http_proxy": (str, None, "env", "sensitive"),
+    "https_proxy": (str, None, "env", "sensitive"),
+    "no_proxy": (str, None, "env", "sensitive"),
+
+    # Global auth for accessing sources, typically set to '%PYMANAGER_USERNAME%'
+    # and '%PYMANAGER_PASSWORD%'. These should really have been
+    # '%PYTHON_MANAGER_USERNAME%' rather than '%PYMANAGER' - the latter is meant
+    # for debugging/hidden options, not user-visible ones.
+    "username": (str, None, "env", "sensitive"),
+    "password": (str, None, "env", "sensitive"),
+
     "list": {
         "format": (str, None, "env"),
         "unmanaged": (config_bool, None, "env"),
@@ -361,6 +375,13 @@ class BaseCommand:
     shebang_templates = {}
     welcome_on_update = False
 
+    # These variables are mainly for capturing optional environment variables.
+    http_proxy = None
+    https_proxy = None
+    no_proxy = None
+    username = None
+    password = None
+
     log_file = None
     _create_log_file = True
     keep_log = True

src/manage/config.py

@@ -275,13 +275,13 @@ def resolve_config(cfg, source, relative_to, key_so_far="", schema=None, error_u
 
         if isinstance(subschema, dict):
             if not isinstance(v, dict):
-                raise InvalidConfigurationError(source, key_so_far + k, v)
+                raise InvalidConfigurationError(source, key_so_far + k)
             resolve_config(v, source, relative_to, f"{key_so_far}{k}.", subschema, environ=environ)
             continue
 
         kind, merge, *opts = subschema
         from_env = False
-        if "env" in opts and isinstance(v, str):
+        if "env" in opts and v and isinstance(v, str):
             try:
                 orig_v = v
                 v = _expand_vars(v, environ)
@@ -296,7 +296,7 @@ def resolve_config(cfg, source, relative_to, key_so_far="", schema=None, error_u
         try:
             v = kind(v)
         except (TypeError, ValueError):
-            raise InvalidConfigurationError(source, key_so_far + k, v)
+            raise InvalidConfigurationError(source, key_so_far + k, v, opts)
         if v and "path" in opts and not _is_valid_url(v):
             # Paths from the config file are relative to the config file.
             # Paths from the environment are relative to the current working dir
@@ -312,44 +312,54 @@ def resolve_config(cfg, source, relative_to, key_so_far="", schema=None, error_u
             else:
                 v = str(v)
             if not _is_valid_url(v):
-                raise InvalidConfigurationError(source, key_so_far + k, v)
+                raise InvalidConfigurationError(source, key_so_far + k, v, opts)
         cfg[k] = v
 
 
-def merge_config(into_cfg, from_cfg, schema, *, source="<unknown>", overwrite=False):
+def merge_config(into_cfg, from_cfg, schema, *, key_so_far="", source="<unknown>", overwrite=False):
+    if key_so_far and not key_so_far.endswith("."):
+        key_so_far = f"{key_so_far}."
+
     for k, v in from_cfg.items():
         if k.startswith("#"):
             continue
-        try:
-            into = into_cfg[k]
-        except LookupError:
-            LOGGER.debug("Setting config %s to %r", k, v)
-            into_cfg[k] = v
-            continue
 
         try:
             subschema = schema[k]
         except LookupError:
             # No schema information, so let's just replace
-            LOGGER.warn("Unknown configuration key %s in %s", k, source)
+            LOGGER.warn("Unknown configuration key %s in %s", key_so_far + k, source)
             into_cfg[k] = v
             continue
 
         if isinstance(subschema, dict):
+            into = into_cfg.setdefault(k, {})
             if isinstance(into, dict) and isinstance(v, dict):
-                LOGGER.debug("Recursively updating config %s", k)
-                merge_config(into, v, subschema, source=source, overwrite=overwrite)
+                LOGGER.debug("Recursively updating config %s", key_so_far + k)
+                merge_config(into, v, subschema, key_so_far=f"{key_so_far}{k}.",
+                             source=source, overwrite=overwrite)
             else:
                 # Source isn't recursing, so let's ignore
                 # Should have been validated earlier
-                LOGGER.warn("Invalid configuration key %s in %s", k, source)
+                LOGGER.warn("Invalid configuration key %s in %s", key_so_far + k, source)
             continue
 
-        _, merge, *_ = subschema
-        if not merge or overwrite:
-            LOGGER.debug("Updating config %s from %r to %r", k, into, v)
-            into_cfg[k] = v
+        _, merge, *opts = subschema
+
+        try:
+            into = into_cfg[k]
+        except KeyError:
+            into = None
         else:
-            v2 = merge(into, v)
-            LOGGER.debug("Updating config %s from %r to %r", k, into, v2)
-            into_cfg[k] = v2
+            if merge and not overwrite:
+                v = merge(into, v)
+
+        if v != into:
+            if "sensitive" in opts:
+                LOGGER.debug("Updating config %s", key_so_far + k)
+            elif "uri" in opts:
+                from .urlutils import sanitise_url
+                LOGGER.debug("Updating config %s to %r", key_so_far + k, sanitise_url(v))
+            else:
+                LOGGER.debug("Updating config %s to %r", key_so_far + k, v)
+            into_cfg[k] = v

src/manage/exceptions.py

@@ -51,7 +51,16 @@ def __init__(self, message, prefix=None):
 
 
 class InvalidConfigurationError(ValueError):
-    def __init__(self, file=None, argument=None, value=None):
+    def __init__(self, file=None, argument=None, value=None, opts=None):
+        if value and opts:
+            if "sensitive" in opts:
+                value = "*" * len(value)
+            elif "uri" in opts:
+                try:
+                    from .urlutils import sanitise_url
+                    value = sanitise_url(value)
+                except Exception:
+                    value = ""
         if value:
             msg = f"Invalid configuration value {value!r} for key {argument} in {file}"
         elif argument:

src/manage/install_command.py

@@ -15,7 +15,6 @@
 from .tagutils import install_matches_any, tag_or_range
 from .urlutils import (
     sanitise_url,
-    urlopen as _urlopen,
     urlretrieve as _urlretrieve,
     IndexDownloader,
 )
@@ -48,7 +47,7 @@ def _expand_versions_by_tag(versions):
                 yield {**v, "tag": t}
 
 
-def select_package(index_downloader, tag, platform=None, *, urlopen=_urlopen, by_id=False, allow_pre=True):
+def select_package(index_downloader, tag, platform=None, *, by_id=False, allow_pre=True):
     """Finds suitable package from index.json that looks like:
     {"versions": [
       {"id": ..., "company": ..., "tag": ..., "url": ..., "hash": {"sha256": hexdigest}},
@@ -96,7 +95,7 @@ def select_package(index_downloader, tag, platform=None, *, urlopen=_urlopen, by
     raise RuntimeError("End of select_package reached")
 
 
-def download_package(cmd, install, dest, cache, *, on_progress=None, urlopen=_urlopen, urlretrieve=_urlretrieve):
+def download_package(cmd, install, dest, cache, *, on_progress=None, urlretrieve=_urlretrieve):
     LOGGER.debug("Starting download package %s to %s", sanitise_url(install["url"]), dest)
 
     if not cmd.force and dest.is_file():
@@ -130,7 +129,7 @@ def _find_creds(url):
         if auth:
             LOGGER.debug("Found credentials in URL or configured source.")
             return auth
-        auth = os.getenv("PYMANAGER_USERNAME", ""), os.getenv("PYMANAGER_PASSWORD", "")
+        auth = cmd.username, (cmd.password or "")
         if auth[0]:
             LOGGER.debug("Found credentials in environment.")
             return auth
@@ -143,6 +142,7 @@ def _find_creds(url):
         on_progress=on_progress,
         on_auth_request=_find_creds,
         on_cancel=lambda: cmd.ask_yn("Abort download?"),
+        proxy_info=cmd,
     )
     LOGGER.debug("Downloaded to %s", dest)
     return dest

src/manage/urlutils.py

@@ -102,21 +102,34 @@ def _parse_proxy(value):
     return proxy, credentials
 
 
-def _proxy_settings_from_env():
-    if os.getenv("NO_PROXY"):
-        return _ProxySettings(mode=PROXY_MODE_DIRECT)
+def _proxy_settings(info):
+    if not info:
+        return _ProxySettings()
 
-    http_value = os.getenv("HTTP_PROXY")
-    https_value = os.getenv("HTTPS_PROXY")
-    if not http_value and not https_value:
+    if info.no_proxy:
+        if info.no_proxy.lower() in ("1", "*", "y", "yes", "t", "true"):
+            return _ProxySettings(mode=PROXY_MODE_DIRECT)
+        if info.no_proxy.lower() not in ("0", "n", "no", "f", "false"):
+            # TODO: Parse and filter based on info.no_proxy IF we can be sure it's
+            # secure, preferably by delegating responsibility for parsing to the OS.
+            # Claiming to support proper whitelisting of domains is very likely to
+            # attract reports of bypasses, none of which we would treat as a
+            # vulnerability (due to non-default configuration settings), but they
+            # historically create more noise than they're worth. Let's just avoid
+            # breaking into jail until we're sure we can do it right.
+            LOGGER.warn("Values for NO_PROXY other than 1, *, y[es], t[rue], "
+                "or the equivalent false values are not supported. "
+                "Other proxy settings will be used.")
+
+    if not info.http_proxy and not info.https_proxy:
         return _ProxySettings()
 
     http_proxy = http_credentials = None
     https_proxy = https_credentials = None
-    if http_value:
-        http_proxy, http_credentials = _parse_proxy(http_value)
-    if https_value:
-        https_proxy, https_credentials = _parse_proxy(https_value)
+    if info.http_proxy:
+        http_proxy, http_credentials = _parse_proxy(info.http_proxy)
+    if info.https_proxy:
+        https_proxy, https_credentials = _parse_proxy(info.https_proxy)
 
     proxy_list = " ".join(
         value for value in (
@@ -134,20 +147,27 @@ def _proxy_settings_from_env():
     )
 
 
+class _EnvProxyInfo:
+    """Default value for proxy_info arguments from environment variables."""
+    no_proxy = os.getenv("NO_PROXY")
+    http_proxy = os.getenv("HTTP_PROXY")
+    https_proxy = os.getenv("HTTPS_PROXY")
+
+
 class NoInternetError(Exception):
     pass
 
 
 class _Request:
-    def __init__(self, url, method="GET", headers={}, outfile=None):
+    def __init__(self, url, method="GET", headers={}, outfile=None, proxy_info=None):
         self.url = url
         self.method = method.upper()
         self.headers = dict(headers)
         self.chunksize = 64 * 1024
         self.username = None
         self.password = None
         self.outfile = Path(outfile) if outfile else None
-        self.proxy_settings = _proxy_settings_from_env()
+        self.proxy_settings = _proxy_settings(proxy_info)
         self._on_progress = None
         self._on_auth_request = None
         self._on_cancel = None
@@ -488,7 +508,8 @@ def _powershell_urlretrieve(request):
                 raise
 
 
-def urlopen(url, method="GET", headers={}, on_progress=None, on_auth_request=None):
+def urlopen(url, method="GET", headers={}, on_progress=None,
+            on_auth_request=None, proxy_info=_EnvProxyInfo):
     scheme, sep, path = url.partition("://")
     if not sep:
         scheme = "file"
@@ -500,7 +521,7 @@ def urlopen(url, method="GET", headers={}, on_progress=None, on_auth_request=Non
         with open(file_url_to_path(url), "rb") as f:
             return f.read()
 
-    request = _Request(url, method=method, headers=headers)
+    request = _Request(url, method=method, headers=headers, proxy_info=proxy_info)
     request._on_progress = on_progress
     request._on_auth_request = on_auth_request
 
@@ -561,7 +582,8 @@ def urlopen(url, method="GET", headers={}, on_progress=None, on_auth_request=Non
 
 
 def urlretrieve(url, outfile, method="GET", headers={}, chunksize=64 * 1024,
-                on_progress=None, on_auth_request=None, on_cancel=None):
+                on_progress=None, on_auth_request=None, on_cancel=None,
+                proxy_info=_EnvProxyInfo):
     scheme, sep, path = url.partition("://")
     if not sep:
         scheme = "file"
@@ -588,7 +610,7 @@ def on_progress(_): pass
             on_progress(100)
         return
 
-    request = _Request(url, method=method, headers=headers)
+    request = _Request(url, method=method, headers=headers, proxy_info=proxy_info)
     request.outfile = Path(outfile)
     request.chunksize = chunksize
     request._on_progress = on_progress
@@ -798,6 +820,7 @@ def urlopen_index(self, url):
                 "GET",
                 {"Accept": "application/json"},
                 on_auth_request=self.on_auth,
+                proxy_info=self.cmd,
             )
         except FileNotFoundError: # includes 404
             (LOGGER.verbose if self.quiet else LOGGER.error)(
@@ -848,6 +871,7 @@ def verify(self, url, data, params, show_settings=False):
                     "GET",
                     {"Accept": "application/octet-stream"},
                     on_auth_request=self.on_auth,
+                    proxy_info=self.cmd,
                 )
                 self._cache[url + ".cat"] = cat
             except OSError as ex:

src/pymanager.json

@@ -12,6 +12,14 @@
     "automatic_install": "%PYTHON_MANAGER_AUTOMATIC_INSTALL%",
     "include_unmanaged": "%PYTHON_MANAGER_INCLUDE_UNMANAGED%",
     "virtual_env": "%VIRTUAL_ENV%",
+
+    "http_proxy": "%HTTP_PROXY%",
+    "https_proxy": "%HTTPS_PROXY%",
+    "no_proxy": "%NO_PROXY%",
+
+    "username": "%PYMANAGER_USERNAME%",
+    "password": "%PYMANAGER_PASSWORD%",
+
     "shebang_can_run_anything": "%PYTHON_MANAGER_SHEBANG_CAN_RUN_ANYTHING%",
     "shebang_can_run_anything_silently": false,