diff --git a/src/sap_cloud_sdk/aicore/__init__.py b/src/sap_cloud_sdk/aicore/__init__.py index 7fb10094..3c9a580e 100644 --- a/src/sap_cloud_sdk/aicore/__init__.py +++ b/src/sap_cloud_sdk/aicore/__init__.py @@ -9,7 +9,11 @@ import os 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 @@ -51,7 +55,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) @@ -87,7 +93,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 diff --git a/src/sap_cloud_sdk/core/secret_resolver/__init__.py b/src/sap_cloud_sdk/core/secret_resolver/__init__.py index 0ab34ca6..b32904cc 100644 --- a/src/sap_cloud_sdk/core/secret_resolver/__init__.py +++ b/src/sap_cloud_sdk/core/secret_resolver/__init__.py @@ -46,6 +46,10 @@ get_sdk_config, get_resolver, ) +from sap_cloud_sdk.core.secret_resolver._validation import ( + _assert_within_base, + _validate_path_component, +) __all__ = [ # Class-based API @@ -61,4 +65,7 @@ # Legacy function-based API "read_from_mount_and_fallback_to_env_var", "resolve_base_mount", + # Internal helpers (re-exported for SDK-internal cross-package use) + "_assert_within_base", + "_validate_path_component", ] 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..627e2793 --- /dev/null +++ b/src/sap_cloud_sdk/core/secret_resolver/_validation.py @@ -0,0 +1,70 @@ +"""Shared input validation for secret resolution. + +Lives in its own module (imported by both ``resolvers.py`` and +``mount_resolver.py``) to avoid an import cycle: ``resolver.py`` already imports +from ``mount_resolver.py``, so these helpers cannot live in ``resolver.py`` +without creating ``resolver → mount_resolver → resolver``. +""" + +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/mount_resolver.py b/src/sap_cloud_sdk/core/secret_resolver/mount_resolver.py index f24dde78..86bf967c 100644 --- a/src/sap_cloud_sdk/core/secret_resolver/mount_resolver.py +++ b/src/sap_cloud_sdk/core/secret_resolver/mount_resolver.py @@ -4,6 +4,7 @@ from typing import Any from sap_cloud_sdk.core.secret_resolver._mapping import _get_field_map +from sap_cloud_sdk.core.secret_resolver._validation import _assert_within_base from sap_cloud_sdk.core.secret_resolver.constants import ( BASE_MOUNT_PATH, SERVICE_BINDING_ROOT, @@ -59,6 +60,7 @@ def _load_from_mount( Sets string attributes directly on the dataclass instance. """ secret_dir = os.path.join(base_volume_mount, module, instance) + _assert_within_base(base_volume_mount, secret_dir) _validate_path(secret_dir) field_map = _get_field_map(target) diff --git a/src/sap_cloud_sdk/core/secret_resolver/resolver.py b/src/sap_cloud_sdk/core/secret_resolver/resolver.py index a8cd4d2f..2999cf2e 100644 --- a/src/sap_cloud_sdk/core/secret_resolver/resolver.py +++ b/src/sap_cloud_sdk/core/secret_resolver/resolver.py @@ -9,6 +9,7 @@ _load_from_mount, ) from sap_cloud_sdk.core.secret_resolver.sdk_env_resolver import _load_from_env +from sap_cloud_sdk.core.secret_resolver._validation import _validate_inputs def read_from_mount_and_fallback_to_env_var( @@ -67,11 +68,3 @@ def read_from_mount_and_fallback_to_env_var( raise RuntimeError( f"module={module} instance={instance} failed to read secrets: {errors} {guidance}" ) - - -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") diff --git a/src/sap_cloud_sdk/core/secret_resolver/resolvers.py b/src/sap_cloud_sdk/core/secret_resolver/resolvers.py index e74b6cfe..35b6e785 100644 --- a/src/sap_cloud_sdk/core/secret_resolver/resolvers.py +++ b/src/sap_cloud_sdk/core/secret_resolver/resolvers.py @@ -17,6 +17,8 @@ from dataclasses import fields, is_dataclass from typing import Any, Protocol, runtime_checkable +from sap_cloud_sdk.core.secret_resolver._validation import _validate_inputs + @runtime_checkable class Resolver(Protocol): @@ -67,6 +69,7 @@ def __init__( def resolve(self, module: str, instance: str, target: Any) -> None: """Try each resolver in order; raise on total failure.""" + _validate_inputs(module, instance) if not is_dataclass(target) or isinstance(target, type): raise TypeError("target must be a dataclass instance") for f in fields(target): diff --git a/tests/agent_memory/unit/test_config.py b/tests/agent_memory/unit/test_config.py index ce297514..e732350c 100644 --- a/tests/agent_memory/unit/test_config.py +++ b/tests/agent_memory/unit/test_config.py @@ -285,3 +285,24 @@ def test_loads_from_env_vars_for_named_instance(self, monkeypatch): assert config.base_url == "https://acme-corp.memory.example.com" assert config.client_id == "acme-client" + + +class TestAgentMemoryPathTraversal: + """agent_memory inherits resolver path-traversal protection.""" + + _BAD = ["../default", "../../etc", "/etc/passwd", "foo/bar", "..", "a\x00b"] + + @pytest.mark.parametrize("bad", _BAD) + def test_bad_instance_rejected(self, bad): + # Resolver raises ValueError("instance ..."); config wraps it in + # AgentMemoryConfigError. Either way, the bad value is rejected. + with pytest.raises((ValueError, AgentMemoryConfigError), match="instance"): + _load_config_for_instance(bad) + + def test_bad_instance_reads_no_files(self): + 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, AgentMemoryConfigError), match="instance"): + _load_config_for_instance("../../etc") diff --git a/tests/aicore/unit/test_aicore.py b/tests/aicore/unit/test_aicore.py index 5439329c..9c2b2cc1 100644 --- a/tests/aicore/unit/test_aicore.py +++ b/tests/aicore/unit/test_aicore.py @@ -4,6 +4,8 @@ import os from unittest.mock import mock_open, patch +import pytest + from sap_cloud_sdk.aicore import ( _get_aicore_base_url, @@ -710,3 +712,31 @@ def test_set_config_decorated_with_record_metrics(self): # Function should complete without errors even with decorator # The actual telemetry recording is tested in telemetry tests + + +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) + + def test_get_secret_rejects_symlink_escape(self, tmp_path): + 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("AICORE_CLIENT_ID", "clientid", 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 18c5a52f..201c6dda 100644 --- a/tests/core/unit/secret_resolver/unit/test_secret_resolver.py +++ b/tests/core/unit/secret_resolver/unit/test_secret_resolver.py @@ -504,3 +504,103 @@ def test_get_resolver_default_chain_contains_mount_and_env(self): # Verify it contains MountResolver and SdkEnvVarResolver by attempting resolution # (the internal _resolvers list is private, so we check behaviour via the protocol) assert hasattr(resolver, "resolve") + + +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", +] + + +class TestPathTraversalLegacy: + """Path-traversal validation via the legacy function entry point.""" + + @pytest.mark.parametrize("bad", BAD_INSTANCE_VALUES) + def test_bad_instance_raises(self, bad): + with pytest.raises(ValueError, match="instance"): + read_from_mount_and_fallback_to_env_var("/path", "VAR", "module", bad, SampleConfig()) + + @pytest.mark.parametrize("bad", BAD_MODULE_VALUES) + def test_bad_module_raises(self, bad): + with pytest.raises(ValueError, match="module"): + read_from_mount_and_fallback_to_env_var("/path", "VAR", bad, "instance", SampleConfig()) + + +class TestPathTraversalChainedResolver: + """Path-traversal validation via the default class-based entry point.""" + + def _chain(self): + return ChainedResolver([MountResolver(), SdkEnvVarResolver()]) + + @pytest.mark.parametrize("bad", BAD_INSTANCE_VALUES) + def test_bad_instance_raises(self, bad): + with pytest.raises(ValueError, match="instance"): + self._chain().resolve("module", bad, SampleConfig()) + + @pytest.mark.parametrize("bad", BAD_MODULE_VALUES) + def test_bad_module_raises(self, bad): + with pytest.raises(ValueError, match="module"): + self._chain().resolve(bad, "instance", SampleConfig()) + + @patch.dict(os.environ, { + "CLOUD_SDK_CFG_MODULE_X_USER": "leaked", + }) + def test_rejected_value_reads_no_files_and_no_env_fallback(self): + def _spy_open(*args, **kwargs): + 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"): + self._chain().resolve("module", "../../x", config) + assert config.username == "" + + @pytest.mark.parametrize("good", GOOD_VALUES) + def test_valid_identifier_passes_validation(self, good): + # Valid names pass validation; the chain still fails (no mount, no env). + with patch.dict(os.environ, {}, clear=True): + with pytest.raises(RuntimeError): + ChainedResolver([MountResolver("/nonexistent"), SdkEnvVarResolver()]).resolve( + "module", good, SampleConfig() + ) + + +class TestPathTraversalMountContainment: + """Containment (symlink-escape) protection in the mount resolver.""" + + def test_symlink_escaping_base_is_rejected(self, tmp_path): + 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) + + with pytest.raises(ValueError, match="escapes trusted root"): + MountResolver(str(base)).resolve("module", "default", SampleConfig())