Prevent secrets from being fetched as remote relative paths

CHANGELOG.rst

@@ -37,6 +37,10 @@ Fixed
   nothing has been typed and the type accepts values other than the choices,
   e.g. ``int | SomeEnum`` (`#976
   <https://github.com/mauvilsa/jsonargparse/pull/976>`__).
+- Secrets in a config read from a URL or fsspec being requested as relative path
+  from the remote, leaking the secret. Types that include ``SecretStr`` in a
+  union now only resolve relative paths locally (`#977
+  <https://github.com/mauvilsa/jsonargparse/pull/977>`__).
 
 Changed
 ^^^^^^^

DOCUMENTATION.rst

@@ -547,7 +547,9 @@ Types can be nested with any complexity. Notes about the support:
   serialized. ``jsonargparse.typing.SecretStr`` gives the same behavior without
   the pydantic dependency. Dumps only have the mask ``**********``, and parsing
   this mask as a secret fails, so that a config bootstrapped with
-  ``--print_config`` is not used with the mask as the secret.
+  ``--print_config`` is not used with the mask as the secret. In a union these
+  types also keep the secret from being fetched as a path, see
+  :ref:`parsing-urls`.
 
 - ``pydantic.FilePath`` and ``pydantic.DirectoryPath`` run the corresponding
   pydantic validation when parsing. Arguments with these types also get file and
@@ -913,6 +915,14 @@ So a tool that takes a config file can also get it from a URL:
     ``s3://bucket/config.yaml``, its parsed absolute path becomes
     ``s3://bucket/model/state_dict.pt``.
 
+.. warning::
+
+    Checking a path means accessing it, so any value in a remote config that a
+    type accepts as a path is requested from the remote, a secret given inline
+    included. To prevent this, add ``SecretStr`` to the type, e.g.
+    ``path_type('fsr') | SecretStr``. Relative paths are then only resolved
+    locally, so just values with an explicit scheme are fetched.
+
 
 .. _boolean-arguments:
 

jsonargparse/_paths.py

@@ -19,6 +19,7 @@
 
 _current_path_dir: ContextVar[str | None] = ContextVar("_current_path_dir", default=None)
 _initial_cwd: ContextVar[str | None] = ContextVar("_initial_cwd", default=None)
+_remote_relative_disabled: ContextVar[bool] = ContextVar("_remote_relative_disabled", default=False)
 
 
 class _CachedStdin(StringIO):
@@ -71,6 +72,22 @@ def _resolve_relative_path(path: str) -> str:
     return "/".join(resolved)
 
 
+def _log_remote_relative_skip(cwd_url_data: _UrlData) -> None:
+    """Debug logs that a relative path was kept local because the type includes a secret.
+
+    The path is not included in the message, since it could be the secret itself.
+    """
+    from ._common import parent_parser
+
+    parser = parent_parser.get()
+    if parser:
+        parent = cwd_url_data.scheme + cwd_url_data.url_path
+        parser.logger.debug(
+            f"Relative path not resolved against remote parent {parent!r} because the type includes a secret, "
+            "give an absolute path to fetch it"
+        )
+
+
 def _known_to_fsspec(path: str) -> bool:
     import_fsspec("_known_to_fsspec")
     from fsspec.registry import known_implementations
@@ -152,8 +169,11 @@ def __init__(
             is_absolute = _is_absolute_path(abs_path)
             url_data = _parse_url(abs_path)
             cwd_url_data = _parse_url(cwd or _current_path_dir.get() or os.getcwd())
-            if ("u" in mode or "s" in mode) and (url_data or (cwd_url_data and not is_absolute)):
-                if cwd_url_data and not is_absolute:
+            remote_relative = not is_absolute and not _remote_relative_disabled.get()
+            if ("u" in mode or "s" in mode) and cwd_url_data and not is_absolute and not remote_relative:
+                _log_remote_relative_skip(cwd_url_data)
+            if ("u" in mode or "s" in mode) and (url_data or (cwd_url_data and remote_relative)):
+                if cwd_url_data and remote_relative:
                     abs_path = _resolve_relative_path(cwd_url_data.url_path + "/" + path)
                     abs_path = cwd_url_data.scheme + abs_path
                     url_data = _parse_url(abs_path)
@@ -346,6 +366,23 @@ def _check_mode(mode: str):
             raise ValueError('Both modes "d" and "s" not possible.')
 
 
+@contextmanager
+def disable_remote_relative_paths(disable: bool = True) -> Iterator[None]:
+    """A context manager that keeps relative paths from being resolved against a remote parent.
+
+    Used when a value could be a secret, so that it is never sent to a remote
+    filesystem or server just to check whether it is an existing path.
+    """
+    if not disable:
+        yield
+        return
+    token = _remote_relative_disabled.set(True)
+    try:
+        yield
+    finally:
+        _remote_relative_disabled.reset(token)
+
+
 @contextmanager
 def change_to_path_dir(path: Path | str | None) -> Iterator[str | None]:
     """A context manager for running code in the directory of a path."""

jsonargparse/_typehints.py

@@ -103,7 +103,7 @@
     typing_extensions_import,
     validate_annotated,
 )
-from ._paths import Path, PathError, change_to_path_dir
+from ._paths import Path, PathError, change_to_path_dir, disable_remote_relative_paths
 from ._required import clear_required
 from ._subcommands import find_action, find_parent_action, parse_kwargs
 from ._type_checking import ArgumentParser
@@ -120,7 +120,7 @@
     parse_value_or_config,
     warning,
 )
-from .typing import _LazyInitBaseClass, get_registered_type, is_pydantic_type
+from .typing import _LazyInitBaseClass, get_registered_type, is_pydantic_type, is_secret_type
 
 NotRequired = typing_extensions_import("NotRequired")
 ReadOnly = typing_extensions_import("ReadOnly")
@@ -1522,19 +1522,22 @@ def adapt_typehints(
     elif typehint_origin == Union:
         vals = []
         sorted_subtypes = sort_subtypes_for_union(subtypehints, val, prev_val, append)
-        for subtype in sorted_subtypes:
-            try:
-                # a pristine value, since adapting can modify it in place
-                subtype_val = adapt_typehints(pristine_value(val), subtype, **adapt_kwargs)
-            except Exception as ex:
-                if subtype is str and not isinstance(val, str) and isinstance(orig_val, str):
-                    vals.append(orig_val)
+        # a secret must not be sent to a remote filesystem or server just to check whether it is a path
+        has_secret = any(is_secret_type(s) for s in sorted_subtypes)
+        with disable_remote_relative_paths(has_secret):
+            for subtype in sorted_subtypes:
+                try:
+                    # a pristine value, since adapting can modify it in place
+                    subtype_val = adapt_typehints(pristine_value(val), subtype, **adapt_kwargs)
+                except Exception as ex:
+                    if subtype is str and not isinstance(val, str) and isinstance(orig_val, str):
+                        vals.append(orig_val)
+                        continue
+                    vals.append(ex)
                     continue
-                vals.append(ex)
-                continue
-            vals.append(subtype_val)
-            if not sub_defaults_invalidate_value(subtype_val, subtype, sorted_subtypes, adapt_kwargs):
-                break
+                vals.append(subtype_val)
+                if not sub_defaults_invalidate_value(subtype_val, subtype, sorted_subtypes, adapt_kwargs):
+                    break
         if all(isinstance(v, Exception) for v in vals):
             raise_union_unexpected_value(sorted_subtypes, val, vals)
         val = next((v for v in reversed(vals) if not isinstance(v, Exception)))

jsonargparse/typing.py

@@ -9,7 +9,14 @@
 from collections.abc import Callable
 from typing import Any, TypeAlias, get_type_hints
 
-from ._common import ClassType, get_settings_logger, is_final_class, is_subclass, path_dump_preserve_relative
+from ._common import (
+    ClassType,
+    get_settings_logger,
+    get_unaliased_type,
+    is_final_class,
+    is_subclass,
+    path_dump_preserve_relative,
+)
 from ._namespace import Namespace
 from ._optionals import final, is_alias_type, pydantic_support
 from ._paths import Path, change_to_path_dir
@@ -704,6 +711,16 @@ def pydantic_secret_str_deserializer(value):
 register_type_on_first_use("pydantic.SecretStr", deserializer=pydantic_secret_str_deserializer)
 
 
+def is_secret_type(typehint) -> bool:
+    """Whether the type holds a secret, i.e. a subclass of jsonargparse's or pydantic's ``SecretStr``."""
+    typehint = get_unaliased_type(typehint)
+    if is_subclass(typehint, SecretStr):
+        return True
+    return inspect.isclass(typehint) and any(
+        t.__module__.startswith("pydantic") and t.__name__ == "SecretStr" for t in inspect.getmro(typehint)
+    )
+
+
 def pydantic_deserializer(class_type):
     from pydantic import create_model  # pylint: disable=no-name-in-module
 

jsonargparse_tests/test_paths.py

@@ -12,11 +12,12 @@
 
 import pytest
 
-from jsonargparse import ArgumentError, Namespace, set_parsing_settings
+from jsonargparse import ArgumentError, ArgumentParser, Namespace, set_parsing_settings
 from jsonargparse._optionals import fsspec_support, url_support
 from jsonargparse._paths import _current_path_dir, _parse_url
-from jsonargparse.typing import Path, Path_drw, Path_fc, Path_fr, path_type
+from jsonargparse.typing import Path, Path_drw, Path_fc, Path_fr, SecretStr, path_type
 from jsonargparse_tests.conftest import (
+    capture_logs,
     get_parser_help,
     is_posix,
     json_or_yaml_dump,
@@ -459,6 +460,129 @@ def test_relative_path_context_fsspec(tmp_cwd, subtests):
         assert _current_path_dir.get() is None
 
 
+# secret types in union tests
+
+
+Path_fsr = path_type("fsr")
+
+
+def secrets_config_parser(password_type):
+    parser = ArgumentParser(exit_on_error=False)
+    parser.add_argument("--cfg", action="config")
+    parser.add_argument("--password", type=password_type)
+    return parser
+
+
+@pytest.fixture
+def fsspec_secrets_config():
+    """A remote config whose password value collides with the name of a sibling remote file."""
+    set_parsing_settings(config_read_mode_fsspec_enabled=True)
+    with fsspec.open("memory://secrets/item/PASSWORD", "w") as f:
+        f.write("sibling file content")
+    config_path = "memory://secrets/item/config.yaml"
+    with fsspec.open(config_path, "w") as f:
+        f.write(json_or_yaml_dump({"password": "PASSWORD"}))
+    return config_path
+
+
+@skip_if_fsspec_unavailable
+@patch_parsing_settings
+def test_secret_in_union_relative_path_not_resolved_as_remote(fsspec_secrets_config):
+    parser = secrets_config_parser(Union[Path_fsr, SecretStr])
+    cfg = parser.parse_args([f"--cfg={fsspec_secrets_config}"])
+    assert isinstance(cfg.password, SecretStr)
+    assert "PASSWORD" == cfg.password.get_secret_value()
+
+
+@skip_if_fsspec_unavailable
+@patch_parsing_settings
+def test_secret_in_union_relative_path_skip_logged(fsspec_secrets_config, logger):
+    with fsspec.open(fsspec_secrets_config, "w") as f:
+        f.write(json_or_yaml_dump({"password": "SECRET_VALUE"}))
+
+    parser = secrets_config_parser(Union[Path_fsr, SecretStr])
+    parser.logger = logger
+    with capture_logs(logger) as logs:
+        cfg = parser.parse_args([f"--cfg={fsspec_secrets_config}"])
+
+    assert isinstance(cfg.password, SecretStr)
+    skip_logs = [x for x in logs.getvalue().split("\n") if "not resolved against remote parent" in x]
+    assert 1 == len(skip_logs)
+    assert "Relative path not resolved against remote parent 'memory://secrets/item'" in skip_logs[0]
+    assert "SECRET_VALUE" not in skip_logs[0]
+
+
+@skip_if_fsspec_unavailable
+@patch_parsing_settings
+def test_no_secret_in_union_relative_path_skip_not_logged(fsspec_secrets_config, logger):
+    parser = secrets_config_parser(Union[Path_fsr, str])
+    parser.logger = logger
+    with capture_logs(logger) as logs:
+        parser.parse_args([f"--cfg={fsspec_secrets_config}"])
+    assert "not resolved against remote parent" not in logs.getvalue()
+
+
+@skip_if_fsspec_unavailable
+@patch_parsing_settings
+def test_secret_in_union_absolute_remote_path_resolved(fsspec_secrets_config):
+    with fsspec.open(fsspec_secrets_config, "w") as f:
+        f.write(json_or_yaml_dump({"password": "memory://secrets/item/PASSWORD"}))
+
+    parser = secrets_config_parser(Union[Path_fsr, SecretStr])
+    cfg = parser.parse_args([f"--cfg={fsspec_secrets_config}"])
+    assert isinstance(cfg.password, Path_fsr)
+    assert "sibling file content" == cfg.password.read_text()
+
+
+@skip_if_fsspec_unavailable
+@patch_parsing_settings
+def test_secret_in_union_local_relative_path_unaffected(fsspec_secrets_config, tmp_cwd):
+    (tmp_cwd / "PASSWORD").write_text("local file content")
+
+    parser = secrets_config_parser(Union[Path_fsr, SecretStr])
+    cfg = parser.parse_args([f"--cfg={fsspec_secrets_config}"])
+    assert isinstance(cfg.password, Path_fsr)
+    assert "local file content" == cfg.password.read_text()
+
+
+@skip_if_fsspec_unavailable
+@patch_parsing_settings
+def test_no_secret_in_union_relative_path_resolved_as_remote(fsspec_secrets_config):
+    parser = secrets_config_parser(Union[Path_fsr, str])
+    cfg = parser.parse_args([f"--cfg={fsspec_secrets_config}"])
+    assert isinstance(cfg.password, Path_fsr)
+    assert "memory://secrets/item/PASSWORD" == cfg.password.absolute
+
+
+@skip_if_fsspec_unavailable
+@patch_parsing_settings
+def test_secret_in_nested_union_relative_path_not_resolved_as_remote(fsspec_secrets_config):
+    with fsspec.open(fsspec_secrets_config, "w") as f:
+        f.write(json_or_yaml_dump({"password": ["PASSWORD"]}))
+
+    parser = secrets_config_parser(List[Union[Path_fsr, SecretStr]])
+    cfg = parser.parse_args([f"--cfg={fsspec_secrets_config}"])
+    assert isinstance(cfg.password[0], SecretStr)
+    assert "PASSWORD" == cfg.password[0].get_secret_value()
+
+
+@skip_if_responses_unavailable
+@responses_activate
+@patch_parsing_settings
+def test_secret_in_union_relative_url_not_resolved_as_remote():
+    set_parsing_settings(config_read_mode_urls_enabled=True)
+    config_url = "http://example.com/item/config.yaml"
+    body = json_or_yaml_dump({"password": "PASSWORD"})
+    responses.add(responses.GET, config_url, status=200, body=body)
+    responses.add(responses.HEAD, config_url, status=200)
+    responses.add(responses.HEAD, "http://example.com/item/PASSWORD", status=200)
+
+    parser = secrets_config_parser(Union[path_type("fur"), SecretStr])
+    cfg = parser.parse_args([f"--cfg={config_url}"])
+    assert isinstance(cfg.password, SecretStr)
+    assert "PASSWORD" == cfg.password.get_secret_value()
+
+
 # path types tests
 
 

jsonargparse_tests/test_pydantic.py

@@ -19,13 +19,16 @@
     typing_extensions_import,
 )
 from jsonargparse._signatures import convert_to_dict
+from jsonargparse.typing import path_type
 from jsonargparse_tests.conftest import (
     capture_logs,
     get_parse_args_stdout,
     get_parser_help,
     json_or_yaml_dump,
     json_or_yaml_load,
+    patch_parsing_settings,
     skip_if_docstring_parser_unavailable,
+    skip_if_fsspec_unavailable,
 )
 
 if pydantic_support:
@@ -60,7 +63,7 @@
     parser.add_argument("--password", type=pydantic.SecretStr)
     cfg = parser.parse_args(["--password=secret"])
     assert isinstance(cfg.password, pydantic.SecretStr)
     assert cfg.password.get_secret_value() == "secret"
     assert "secret" not in parser.dump(cfg)
 
 
@@ -72,6 +75,28 @@
         parser.parse_string(dumped)
 
 
+@skip_if_fsspec_unavailable
+@skip_if_pydantic_v1_on_v2
+@patch_parsing_settings
+def test_pydantic_secret_str_in_union_relative_path_not_resolved_as_remote():
+    import fsspec
+
+    set_parsing_settings(config_read_mode_fsspec_enabled=True)
+    with fsspec.open("memory://pydantic_secrets/item/PASSWORD", "w") as f:
+        f.write("sibling file content")
+    config_path = "memory://pydantic_secrets/item/config.yaml"
+    with fsspec.open(config_path, "w") as f:
+        f.write(json_or_yaml_dump({"password": "PASSWORD"}))
+
+    parser = ArgumentParser(exit_on_error=False)
+    parser.add_argument("--cfg", action="config")
+    parser.add_argument("--password", type=Union[path_type("fsr"), pydantic.SecretStr])
+
+    cfg = parser.parse_args([f"--cfg={config_path}"])
+    assert isinstance(cfg.password, pydantic.SecretStr)
+    assert "PASSWORD" == cfg.password.get_secret_value()
+
+
 if annotated and pydantic_support > 1:
 
     @pydantic.dataclasses.dataclass(frozen=True)