Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 7 additions & 0 deletions src/sap_cloud_sdk/agent_memory/user-guide.md
Original file line number Diff line number Diff line change
Expand Up @@ -182,6 +182,13 @@ across create, read, and search calls is the implementer's responsibility.

- **Further reading:** N/A

> **Security note (path safety):** When a per-tenant `instance` value is resolved
> (e.g. from a JWT claim or HTTP header in a multitenant deployment), the SDK
> validates it as a single-component identifier before building any filesystem path.
> A crafted value such as `"../default"` cannot select another binding's credentials.
> This protection is automatic for all SDK consumers — no extra validation is needed
> in agent or application code.

## Semantic Search: A Brief Primer

Texts with different words — or even different languages — can have the same meaning.
Expand Down
9 changes: 9 additions & 0 deletions src/sap_cloud_sdk/aicore/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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.resolver 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
Expand Down Expand Up @@ -71,8 +75,10 @@ def _get_secret(
instance_name: Name of the aicore instance defined in app.yaml. Defaults to aicore-instance

"""
_validate_path_component("instance_name", instance_name)
resolved_base_path = resolve_base_mount()
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)

Expand Down Expand Up @@ -107,8 +113,10 @@ def _get_aicore_base_url(instance_name: str = "aicore-instance") -> str:
Returns:
Base URL for AI Core service
"""
_validate_path_component("instance_name", instance_name)
resolved_base_path = resolve_base_mount()
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
Expand Down Expand Up @@ -301,6 +309,7 @@ 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."""
_validate_path_component("instance_name", instance_name)
secret_dir = os.path.join(resolve_base_mount(), "aicore", instance_name)
try:
return os.stat(secret_dir).st_mtime
Expand Down
11 changes: 11 additions & 0 deletions src/sap_cloud_sdk/aicore/user-guide.md
Original file line number Diff line number Diff line change
Expand Up @@ -45,6 +45,12 @@ from sap_cloud_sdk.aicore import set_aicore_config
set_aicore_config(instance_name="aicore-production")
```

> **Security note:** `instance_name` is validated as a safe single path component
> before any file is read. Traversal sequences (`../`), absolute paths (`/etc/…`),
> UNC paths, embedded separators, NUL / control characters, and values exceeding
> 255 characters raise `ValueError`. All standard BTP instance names such as
> `aicore-instance` and `aicore-prod` are accepted unchanged.

---

## Routing Modes
Expand Down Expand Up @@ -628,6 +634,11 @@ If credentials from the wrong instance are loaded:
set_aicore_config(instance_name="correct-instance-name")
```

If you see a `ValueError: instance_name must be a single path component`, the
value contains a path separator, traversal segment, or other disallowed
character. Use a plain BTP instance name (letters, digits, hyphens, underscores,
internal dots — no slashes or `..`).

---

## Integration with Telemetry
Expand Down
56 changes: 55 additions & 1 deletion src/sap_cloud_sdk/core/secret_resolver/resolver.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@

import os
from dataclasses import fields, is_dataclass
from pathlib import Path
from typing import Any, Dict, Tuple
from .constants import BASE_MOUNT_PATH

Expand Down Expand Up @@ -32,6 +33,56 @@ def _validate_inputs(module: str, instance: str) -> None:
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 _validate_path() call 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}"
)


def _validate_path(path: str) -> None:
Expand Down Expand Up @@ -112,6 +163,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)


Expand Down Expand Up @@ -169,7 +221,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};")
Expand Down
33 changes: 33 additions & 0 deletions src/sap_cloud_sdk/core/secret_resolver/user-guide.md
Original file line number Diff line number Diff line change
Expand Up @@ -106,6 +106,39 @@ export DB_DATABASE_PRIMARY_PASSWORD="secret123"

---

## Input Validation & Security

Both `module` and `instance` are validated as **safe single-component path
identifiers** before any filesystem path is assembled or any environment
variable is consulted. A rejected value raises `ValueError` immediately and
reads no files.

**Allowed:** alphanumeric characters, hyphens, underscores, and internal dots.
Examples: `"default"`, `"hana-agent-memory"`, `"aicore-instance"`,
`"my-tenant-us10"`.

**Rejected** (raises `ValueError`):

| Pattern | Example |
|---------|---------|
| Path traversal | `"../default"`, `"../../etc"` |
| Absolute POSIX path | `"/etc/passwd"` |
| Windows absolute path | `"C:\\Windows"`, `"C:/Windows"` |
| UNC path | `"\\\\server\\share"` |
| Embedded path separator | `"foo/bar"`, `"foo\\bar"` |
| Dot components | `"."`, `".."` |
| NUL / control characters | `"foo\x00bar"` |
| Exceeds 255 characters | `"a" * 256` |

After component validation, the resolved canonical path is compared against the
trusted root using `pathlib.Path.resolve()`. This defense-in-depth check
prevents symlink-based escape even when component validation passes.

**This protection is automatic** — no caller needs to pre-validate inputs. Every
agent or application using the SDK is protected without any code change.

---

## Usage Examples

### ObjectStore Configuration
Expand Down
74 changes: 74 additions & 0 deletions tests/aicore/unit/test_aicore.py
Original file line number Diff line number Diff line change
Expand Up @@ -1052,3 +1052,77 @@ 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()


# ---------------------------------------------------------------------------
# Path-traversal security regression tests (HASI2026203-281)
# ---------------------------------------------------------------------------


class TestAicoreInstanceValidation:
"""Verify that _get_secret, _get_aicore_base_url, and _get_secret_dir_mtime
reject traversal/absolute instance_name values before touching the filesystem."""

from sap_cloud_sdk.aicore import _get_secret_dir_mtime

BAD_NAMES = [
"../default",
"../../etc",
"/etc/passwd",
"C:/Windows",
"\\\\server\\share",
"foo/bar",
"foo\x00bar",
".",
"..",
"a/b",
]

GOOD_NAMES = [
"aicore-instance",
"aicore-prod",
"my-instance",
"custom-aicore",
]

@pytest.mark.parametrize("bad", BAD_NAMES)
def test_get_secret_rejects_traversal(self, bad):
with pytest.raises(ValueError):
_get_secret("ENV_VAR", instance_name=bad)

@pytest.mark.parametrize("bad", BAD_NAMES)
def test_get_aicore_base_url_rejects_traversal(self, bad):
with pytest.raises(ValueError):
_get_aicore_base_url(instance_name=bad)

@pytest.mark.parametrize("bad", BAD_NAMES)
def test_get_secret_dir_mtime_rejects_traversal(self, bad):
from sap_cloud_sdk.aicore import _get_secret_dir_mtime
with pytest.raises(ValueError):
_get_secret_dir_mtime(instance_name=bad)

@pytest.mark.parametrize("good", GOOD_NAMES)
def test_valid_instance_name_passes_get_secret(self, good):
with (
patch("os.path.exists", return_value=False),
patch.dict("os.environ", {}, clear=True),
):
result = _get_secret("MISSING_VAR", instance_name=good)
assert result == ""

@pytest.mark.parametrize("good", GOOD_NAMES)
def test_valid_instance_name_passes_get_aicore_base_url(self, good):
with (
patch("os.path.exists", return_value=False),
patch.dict("os.environ", {}, clear=True),
):
result = _get_aicore_base_url(instance_name=good)
assert result == ""

def test_get_secret_does_not_read_file_on_traversal(self):
"""Rejected instance_name must not trigger any file access."""
opened = []
with patch("builtins.open", lambda *a, **kw: opened.append(a)):
with pytest.raises(ValueError):
_get_secret("ENV_VAR", instance_name="../default")
assert opened == [], "traversal must not open any file"
83 changes: 83 additions & 0 deletions tests/core/unit/secret_resolver/unit/test_secret_resolver.py
Original file line number Diff line number Diff line change
Expand Up @@ -255,3 +255,86 @@ 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"


# ---------------------------------------------------------------------------
# Path-traversal security regression tests (HASI2026203-281)
# ---------------------------------------------------------------------------

BAD_INSTANCE_VALUES = [
"../default", # relative traversal (the primary attack)
"../../etc", # multi-hop traversal
"/etc/passwd", # absolute POSIX path
"C:\\Windows", # Windows-style absolute (backslash)
"C:/Windows", # Windows-style absolute (forward slash)
"\\\\server\\share", # UNC path
"foo/bar", # embedded forward separator
"foo\\bar", # embedded backslash
"foo\x00bar", # NUL byte
"foo\x01bar", # control character
".", # dot component
"..", # parent component
"a" * 256, # exceeds maximum length
]

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


class TestPathComponentValidation:

@pytest.mark.parametrize("bad", BAD_INSTANCE_VALUES)
def test_bad_instance_raises_value_error(self, bad):
with pytest.raises(ValueError):
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_value_error(self, bad):
with pytest.raises(ValueError):
read_from_mount_and_fallback_to_env_var(
"/path", "VAR", bad, "instance", SampleConfig()
)

@pytest.mark.parametrize("good", GOOD_VALUES)
def test_valid_identifier_passes_validation(self, good):
# Validation passes; RuntimeError expected because /nonexistent doesn't exist.
with pytest.raises(RuntimeError):
read_from_mount_and_fallback_to_env_var(
"/nonexistent", "VAR", "module", good, SampleConfig()
)

def test_rejected_value_reads_no_files_and_no_env_fallback(self, monkeypatch):
"""A traversal value must never reach the filesystem or env-var strategy."""
opened = []
monkeypatch.setattr("builtins.open", lambda *a, **kw: opened.append(a))
monkeypatch.setenv("VAR_MODULE_X_USER", "leak")
with pytest.raises(ValueError):
read_from_mount_and_fallback_to_env_var(
"/path", "VAR", "module", "../x", SampleConfig()
)
assert opened == [], "bad instance must not open any file"

def test_symlink_escaping_base_is_rejected(self, tmp_path):
"""A symlink pointing outside the trusted root must be rejected."""
base = tmp_path / "appfnd"
(base / "mod").mkdir(parents=True)
outside = tmp_path / "outside"
outside.mkdir()
(base / "mod" / "inst").symlink_to(outside, target_is_directory=True)
# "inst" is a valid single-component name, passes _validate_path_component,
# but _assert_within_base detects the symlink escapes and raises ValueError
# which is aggregated into RuntimeError by read_from_mount_and_fallback_to_env_var.
with pytest.raises(RuntimeError, match="escapes trusted root"):
read_from_mount_and_fallback_to_env_var(
str(base), "VAR", "mod", "inst", SampleConfig()
)
Loading