fix: reject path-traversal inputs in secret resolver

src/sap_cloud_sdk/aicore/__init__.py

@@ -10,7 +10,11 @@
 import threading
 from typing import Optional
 
-from sap_cloud_sdk.core.secret_resolver import resolve_base_mount
+from sap_cloud_sdk.core.secret_resolver import (
+    resolve_base_mount,
+    _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:

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",
 ]

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}"
+        )

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/<module>/<field> before the legacy $ROOT/<module>/<instance>/<field> 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};")

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)

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")

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
+            )