From 6ed009d213baf1504b5acb999e8b23cc93d95458 Mon Sep 17 00:00:00 2001 From: Cassio Farias Machado Date: Tue, 27 Oct 2026 17:30:31 -0300 Subject: [PATCH 1/2] fix: reject path-traversal inputs in secret resolver --- src/sap_cloud_sdk/aicore/__init__.py | 13 ++- .../core/secret_resolver/_validation.py | 68 +++++++++++++ .../core/secret_resolver/resolver.py | 17 ++-- tests/agent_memory/unit/test_config.py | 37 +++++++ tests/aicore/unit/test_aicore.py | 37 +++++++ .../unit/test_secret_resolver.py | 99 +++++++++++++++++++ 6 files changed, 261 insertions(+), 10 deletions(-) create mode 100644 src/sap_cloud_sdk/core/secret_resolver/_validation.py diff --git a/src/sap_cloud_sdk/aicore/__init__.py b/src/sap_cloud_sdk/aicore/__init__.py index 83965190..b789cf53 100644 --- a/src/sap_cloud_sdk/aicore/__init__.py +++ b/src/sap_cloud_sdk/aicore/__init__.py @@ -11,6 +11,10 @@ from typing import Optional from sap_cloud_sdk.core.secret_resolver import resolve_base_mount +from sap_cloud_sdk.core.secret_resolver._validation import ( + _assert_within_base, + _validate_path_component, +) from sap_cloud_sdk.core.telemetry.metrics_decorator import record_metrics from sap_cloud_sdk.core.telemetry.module import Module from sap_cloud_sdk.core.telemetry.operation import Operation @@ -72,7 +76,9 @@ def _get_secret( """ resolved_base_path = resolve_base_mount() + _validate_path_component("instance_name", instance_name) secrets_base_path = f"{resolved_base_path}/aicore/{instance_name}" + _assert_within_base(f"{resolved_base_path}/aicore", secrets_base_path) secret_file_name = file_name if file_name else env_var_name secret_file_path = os.path.join(secrets_base_path, secret_file_name) @@ -108,7 +114,9 @@ def _get_aicore_base_url(instance_name: str = "aicore-instance") -> str: Base URL for AI Core service """ resolved_base_path = resolve_base_mount() + _validate_path_component("instance_name", instance_name) secrets_base_path = f"{resolved_base_path}/aicore/{instance_name}" + _assert_within_base(f"{resolved_base_path}/aicore", secrets_base_path) serviceurls_file = os.path.join(secrets_base_path, "serviceurls") # Try reading from serviceurls file @@ -301,7 +309,10 @@ def _configure_direct_mode(instance_name: str) -> None: def _get_secret_dir_mtime(instance_name: str = "aicore-instance") -> float: """Return the mtime of the AI Core secret directory, or 0.0 if it does not exist.""" - secret_dir = os.path.join(resolve_base_mount(), "aicore", instance_name) + _validate_path_component("instance_name", instance_name) + aicore_base = os.path.join(resolve_base_mount(), "aicore") + secret_dir = os.path.join(aicore_base, instance_name) + _assert_within_base(aicore_base, secret_dir) try: return os.stat(secret_dir).st_mtime except OSError: diff --git a/src/sap_cloud_sdk/core/secret_resolver/_validation.py b/src/sap_cloud_sdk/core/secret_resolver/_validation.py new file mode 100644 index 00000000..8c6017d7 --- /dev/null +++ b/src/sap_cloud_sdk/core/secret_resolver/_validation.py @@ -0,0 +1,68 @@ +"""Shared input validation for secret resolution. + +Lives in its own module so every resolver entry point can import the same +validators without risking an import cycle. +""" + +from __future__ import annotations + +import os +from pathlib import Path + + +def _validate_inputs(module: str, instance: str) -> None: + """Single validation entry point: empty-check + component rules for both inputs.""" + if not isinstance(module, str) or not module.strip(): + raise ValueError("module name cannot be empty") + if not isinstance(instance, str) or not instance.strip(): + raise ValueError("instance name cannot be empty") + _validate_path_component("module", module) + _validate_path_component("instance", instance) + + +def _validate_path_component(name: str, value: str) -> None: + """Reject any value that is not a single safe filesystem path component. + + Called automatically before any path assembly or environment-variable + fallback — a rejected value reads no files. + Allows hyphens and internal dots (e.g. "hana-agent-memory", "aicore-instance", + "my-tenant-us10"); rejects separators, absolute/UNC forms, '.'/'..', NUL, + control characters, and values exceeding 255 characters. + """ + if "\x00" in value: + raise ValueError(f"{name} must not contain null bytes") + if any(ord(c) < 32 for c in value): + raise ValueError(f"{name} must not contain control characters") + if len(value) > 255: + raise ValueError(f"{name} exceeds maximum identifier length (255 chars)") + if os.path.isabs(value): + raise ValueError(f"{name} must not be an absolute path") + if value.startswith("\\\\"): + raise ValueError(f"{name} must not be a UNC path") + # Normalise both separator styles; require exactly one non-traversal segment. + parts = [p for p in value.replace("\\", "/").split("/") if p] + if len(parts) != 1 or parts[0] in (".", ".."): + raise ValueError( + f"{name} must be a single path component " + f"(no path separators, drive letters, or '.'/'..' segments); got {value!r}" + ) + + +def _assert_within_base(base_volume_mount: str, candidate: str) -> None: + """Defense-in-depth: verify candidate resolves canonically inside base. + + Resolves symlinks on both sides. Silent when the path does not exist yet — + the subsequent path-existence check surfaces a clean error in that case. + Protects against TOCTOU / symlink-escape scenarios that pass component + validation but point outside the trusted volume root at open-time. + """ + try: + base_real = Path(base_volume_mount).resolve(strict=True) + cand_real = Path(candidate).resolve(strict=True) + except (FileNotFoundError, OSError): + return + if cand_real != base_real and not cand_real.is_relative_to(base_real): + raise ValueError( + f"resolved binding path escapes trusted root {base_volume_mount!r}: " + f"{candidate!r} → {cand_real}" + ) diff --git a/src/sap_cloud_sdk/core/secret_resolver/resolver.py b/src/sap_cloud_sdk/core/secret_resolver/resolver.py index c76bbc74..4d8f5b9d 100644 --- a/src/sap_cloud_sdk/core/secret_resolver/resolver.py +++ b/src/sap_cloud_sdk/core/secret_resolver/resolver.py @@ -6,6 +6,10 @@ from dataclasses import fields, is_dataclass from typing import Any, Dict, Tuple from .constants import BASE_MOUNT_PATH +from ._validation import ( + _validate_inputs, + _assert_within_base, +) def resolve_base_mount(base_volume_mount: str = BASE_MOUNT_PATH) -> str: @@ -26,14 +30,6 @@ def resolve_base_mount(base_volume_mount: str = BASE_MOUNT_PATH) -> str: return os.environ.get("SERVICE_BINDING_ROOT", base_volume_mount) -def _validate_inputs(module: str, instance: str) -> None: - """Validate module and instance inputs.""" - if not isinstance(module, str) or not module.strip(): - raise ValueError("module name cannot be empty") - if not isinstance(instance, str) or not instance.strip(): - raise ValueError("instance name cannot be empty") - - def _validate_path(path: str) -> None: """Validate that the given path exists and is a directory.""" try: @@ -112,6 +108,7 @@ def _load_from_mount( {base_volume_mount}/{module}/{instance}/{field_key} """ secret_dir = os.path.join(base_volume_mount, module, instance) + _assert_within_base(base_volume_mount, secret_dir) _load_from_path(secret_dir, target) @@ -169,7 +166,9 @@ def read_from_mount_and_fallback_to_env_var( # $ROOT// before the legacy $ROOT/// path. if os.environ.get("SERVICE_BINDING_ROOT") is not None: try: - _load_from_path(os.path.join(resolved_base_path, module), target) + flat_dir = os.path.join(resolved_base_path, module) + _assert_within_base(resolved_base_path, flat_dir) + _load_from_path(flat_dir, target) return except Exception as e: errors.append(f"mount failed: {e};") diff --git a/tests/agent_memory/unit/test_config.py b/tests/agent_memory/unit/test_config.py index 40e9b2ea..3bc21511 100644 --- a/tests/agent_memory/unit/test_config.py +++ b/tests/agent_memory/unit/test_config.py @@ -210,3 +210,40 @@ def test_raises_config_error_when_uaa_json_invalid(self, monkeypatch): with patch("os.stat", side_effect=FileNotFoundError("no mount")): with pytest.raises(AgentMemoryConfigError, match="Failed to parse uaa JSON"): _load_config_from_env() + + +# ── Path-traversal propagation ──────────────────────────────────────────────── + +class TestAgentMemoryPathTraversal: + """agent_memory inherits resolver path-traversal protection via ConfigFactory.""" + + _BAD = ["../default", "../../etc", "/etc/passwd", "foo/bar", "..", "a\x00b"] + + @pytest.mark.parametrize("bad", _BAD) + def test_bad_instance_raises_on_factory_call(self, bad): + from sap_cloud_sdk.agent_memory.config import _make_config_factory + + with pytest.raises(ValueError, match="instance"): + _make_config_factory(bad)() + + def test_bad_instance_reads_no_files(self): + from sap_cloud_sdk.agent_memory.config import _make_config_factory + + def _spy_open(*args, **kwargs): + raise AssertionError("open() must not be called for a rejected value") + + with patch("builtins.open", side_effect=_spy_open): + with pytest.raises(ValueError, match="instance"): + _make_config_factory("../../etc")() + + def test_valid_instance_passes_validation(self): + from sap_cloud_sdk.agent_memory.config import _make_config_factory + + # Valid instance passes component validation; fails later because no + # mount/env is present (not a ValueError about the instance name). + with patch("os.path.isdir", return_value=False), \ + patch("os.stat", side_effect=FileNotFoundError()), \ + patch.dict("os.environ", {}, clear=True): + with pytest.raises(Exception) as exc: + _make_config_factory("default")() + assert "instance must be a single path component" not in str(exc.value) diff --git a/tests/aicore/unit/test_aicore.py b/tests/aicore/unit/test_aicore.py index c7ebe5b4..f109244b 100644 --- a/tests/aicore/unit/test_aicore.py +++ b/tests/aicore/unit/test_aicore.py @@ -1052,3 +1052,40 @@ def test_destination_mode_still_calls_set_filtering(self): mock_create.return_value.get_destination.return_value = dest set_aicore_config() mock_filter.assert_called_once() + + +class TestAicoreInstanceValidation: + """Path-traversal validation for aicore instance_name.""" + + BAD = ["../other", "../../etc", "/etc/passwd", "foo/bar", "foo\\bar", ".", "..", "a\x00b"] + + @pytest.mark.parametrize("bad", BAD) + def test_get_secret_rejects_bad_instance(self, bad): + with pytest.raises(ValueError, match="instance_name"): + _get_secret("AICORE_CLIENT_ID", "clientid", instance_name=bad) + + @pytest.mark.parametrize("bad", BAD) + def test_get_aicore_base_url_rejects_bad_instance(self, bad): + with pytest.raises(ValueError, match="instance_name"): + _get_aicore_base_url(instance_name=bad) + + @pytest.mark.parametrize("bad", BAD) + def test_get_secret_dir_mtime_rejects_bad_instance(self, bad): + from sap_cloud_sdk.aicore import _get_secret_dir_mtime + + with pytest.raises(ValueError, match="instance_name"): + _get_secret_dir_mtime(instance_name=bad) + + def test_get_secret_dir_mtime_rejects_symlink_escape(self, tmp_path): + from sap_cloud_sdk.aicore import _get_secret_dir_mtime + + base = tmp_path / "root" + aicore = base / "aicore" + aicore.mkdir(parents=True) + outside = tmp_path / "outside" + outside.mkdir() + (aicore / "inst").symlink_to(outside, target_is_directory=True) + + with patch("sap_cloud_sdk.aicore.resolve_base_mount", return_value=str(base)): + with pytest.raises(ValueError, match="escapes trusted root"): + _get_secret_dir_mtime(instance_name="inst") diff --git a/tests/core/unit/secret_resolver/unit/test_secret_resolver.py b/tests/core/unit/secret_resolver/unit/test_secret_resolver.py index f02445ff..9bd69ffb 100644 --- a/tests/core/unit/secret_resolver/unit/test_secret_resolver.py +++ b/tests/core/unit/secret_resolver/unit/test_secret_resolver.py @@ -8,6 +8,39 @@ from sap_cloud_sdk.core.secret_resolver import read_from_mount_and_fallback_to_env_var +BAD_INSTANCE_VALUES = [ + "../default", + "../../etc", + "/etc/passwd", + "C:\\Windows", + "C:/Windows", + "\\\\server\\share", + "foo/bar", + "foo\\bar", + "foo\x00bar", + "foo\x01bar", + ".", + "..", + "a" * 256, +] + +BAD_MODULE_VALUES = [ + "../sibling", + "/abs/path", + "foo/bar", + "..", +] + +GOOD_VALUES = [ + "default", + "hana-agent-memory", + "aicore-instance", + "my-tenant-us10", + "foo.bar", + "hr-advisor-destination-instance", +] + + @dataclass class SampleConfig: username: str = field(default="", metadata={"secret": "user"}) @@ -255,3 +288,69 @@ def test_service_binding_root_flat_fails_falls_back_to_module_instance(self, moc assert config.username == "legacy_user" assert config.password == "legacy_pass" assert config.endpoint == "legacy_ep" + + +class TestPathComponentValidation: + + @pytest.mark.parametrize("bad", BAD_INSTANCE_VALUES) + def test_bad_instance_raises_value_error(self, bad): + config = SampleConfig() + with pytest.raises(ValueError, match="instance"): + read_from_mount_and_fallback_to_env_var("/path", "VAR", "module", bad, config) + + @pytest.mark.parametrize("bad", BAD_MODULE_VALUES) + def test_bad_module_raises_value_error(self, bad): + config = SampleConfig() + with pytest.raises(ValueError, match="module"): + read_from_mount_and_fallback_to_env_var("/path", "VAR", bad, "instance", config) + + @pytest.mark.parametrize("good", GOOD_VALUES) + def test_valid_identifier_passes_validation(self, good): + # Valid names pass component validation; the call still fails because + # /nonexistent has no mount and no env var is set (RuntimeError, not ValueError). + config = SampleConfig() + with patch("os.path.isdir", return_value=False), \ + patch("os.stat", side_effect=FileNotFoundError()), \ + patch.dict(os.environ, {}, clear=True): + with pytest.raises(RuntimeError): + read_from_mount_and_fallback_to_env_var("/nonexistent", "VAR", "module", good, config) + + @patch.dict(os.environ, { + "VAR_MODULE_X_USER": "leaked", + "VAR_MODULE_X_PASSWORD": "leaked", + "VAR_MODULE_X_ENDPOINT": "leaked", + }) + def test_rejected_value_reads_no_files_and_no_env_fallback(self): + # A rejected instance must short-circuit before any file read AND before + # the environment-variable fallback runs. + opened: list = [] + + def _spy_open(*args, **kwargs): + opened.append(args[0]) + raise AssertionError("open() must not be called for a rejected value") + + config = SampleConfig() + with patch("builtins.open", side_effect=_spy_open): + with pytest.raises(ValueError, match="instance"): + read_from_mount_and_fallback_to_env_var( + "/path", "VAR", "module", "../../x", config + ) + assert opened == [] + # Env fallback did not populate the config despite matching env vars. + assert config.username == "" + + def test_symlink_escaping_base_is_rejected(self, tmp_path): + # A symlinked instance dir that points outside the mount root must be rejected + # at open-time by the containment check. + base = tmp_path / "appfnd" + module_dir = base / "module" + module_dir.mkdir(parents=True) + outside = tmp_path / "outside" + outside.mkdir() + (module_dir / "default").symlink_to(outside, target_is_directory=True) + + config = SampleConfig() + with pytest.raises(RuntimeError, match="escapes trusted root"): + read_from_mount_and_fallback_to_env_var( + str(base), "VAR", "module", "default", config + ) From a39aafdd8dcc5164aacbf32af66eaac42c92d1f0 Mon Sep 17 00:00:00 2001 From: Cassio Farias Machado Date: Tue, 3 Nov 2026 15:26:49 -0300 Subject: [PATCH 2/2] refactor(secret-resolver): re-export validation helpers via package boundary --- src/sap_cloud_sdk/aicore/__init__.py | 4 ++-- src/sap_cloud_sdk/core/secret_resolver/__init__.py | 3 +++ 2 files changed, 5 insertions(+), 2 deletions(-) diff --git a/src/sap_cloud_sdk/aicore/__init__.py b/src/sap_cloud_sdk/aicore/__init__.py index b789cf53..9fd0f85c 100644 --- a/src/sap_cloud_sdk/aicore/__init__.py +++ b/src/sap_cloud_sdk/aicore/__init__.py @@ -10,8 +10,8 @@ import threading from typing import Optional -from sap_cloud_sdk.core.secret_resolver import resolve_base_mount -from sap_cloud_sdk.core.secret_resolver._validation import ( +from sap_cloud_sdk.core.secret_resolver import ( + resolve_base_mount, _assert_within_base, _validate_path_component, ) diff --git a/src/sap_cloud_sdk/core/secret_resolver/__init__.py b/src/sap_cloud_sdk/core/secret_resolver/__init__.py index 61779d3c..947079e3 100644 --- a/src/sap_cloud_sdk/core/secret_resolver/__init__.py +++ b/src/sap_cloud_sdk/core/secret_resolver/__init__.py @@ -23,9 +23,12 @@ class MyConfig: from .resolver import read_from_mount_and_fallback_to_env_var, resolve_base_mount from ._config_factory import ConfigFactory +from ._validation import _assert_within_base, _validate_path_component __all__ = [ "read_from_mount_and_fallback_to_env_var", "resolve_base_mount", "ConfigFactory", + "_assert_within_base", + "_validate_path_component", ]