diff --git a/.env.example b/.env.example index 72b0974..d9e6518 100644 --- a/.env.example +++ b/.env.example @@ -63,8 +63,17 @@ ADMIN_API_KEY= # by default. Single-tenant self-hosted operators who intentionally use code # steps can opt in; leave false on multi-tenant deployments. CODE_STEPS_ALLOW_UNTRUSTED=false +# Run "code" steps in a separate Python subprocess (out-of-process isolation) so +# a sandbox escape cannot reach the parent process memory (settings, DB session, +# other tenants' data). On by default; set false to use the legacy in-process path. +CODE_STEPS_OUT_OF_PROCESS=true # Bearer token for the Memory MCP server over streamable-http. Required to start # streamable-http outside local mode (it fails closed without it). stdio is # unaffected. Callers must send: Authorization: Bearer # MEMORY_MCP_TOKEN= +# Optional tenant scope prefix for the Memory MCP server. When set, every tool +# call (add/search/forget/list_memories) must operate within this prefix, so an +# authenticated token cannot read/write/delete another tenant's memories. Leave +# unset for single-tenant/local deployments (no cross-tenant enforcement). +# MEMORY_MCP_SCOPE_PREFIX= LOG_LEVEL=info diff --git a/.github/workflows/pr-tests.yml b/.github/workflows/pr-tests.yml index ddc6eac..c7dc93c 100644 --- a/.github/workflows/pr-tests.yml +++ b/.github/workflows/pr-tests.yml @@ -28,7 +28,9 @@ jobs: python: name: Python tests runs-on: ubuntu-latest - timeout-minutes: 15 + # ~17k tests run in ~13.5 min plus dashboard build and dependency install; + # 20 min leaves headroom over the total setup + run time. + timeout-minutes: 20 steps: - uses: actions/checkout@v4 diff --git a/CHANGELOG.md b/CHANGELOG.md index 10af2e3..f1aa90e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,28 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +## [0.40.3] - 2026-07-11 - "Deeper Isolation" + +Follow-up hardening that finishes the three deferred items from the 0.40.2 audit sweep. Defaults +are secure; existing behavior is preserved with per-feature opt-outs. + +### Security +- **Code steps now run out-of-process by default.** Validated `code` steps execute in a separate + Python subprocess with a secret-free environment and POSIX CPU/address-space limits, so a + sandbox escape cannot reach the parent process memory (`settings`, DB session factory, other + tenants' data). The subprocess is really killed on timeout (the in-process thread path could + not be). Set `CODE_STEPS_OUT_OF_PROCESS=false` to use the legacy in-process path; infrastructure + failures fall back to it automatically so a step is never lost. +- **HTTP steps pin the validated IP at connect time.** The SSRF pre-flight resolves the hostname + once, validates the address, and a custom transport dials that exact IP while preserving the + `Host` header and TLS SNI, closing the DNS-rebind TOCTOU. Non-resolvable/mocked hosts are + unaffected. +- **Memory MCP per-tenant scope enforcement.** With `MEMORY_MCP_SCOPE_PREFIX` set, every tool call + (`add`/`search`/`list_memories`/`forget`) must stay within the configured scope, so an + authenticated caller cannot reach another tenant's memories by supplying a different scope + string. `forget` requires and validates the owning `user_id` against the prefix. Unset preserves + today's single-tenant behavior. + ## [0.40.2] - 2026-07-11 - "Hardened Steps" Security and correctness patch from a full audit sweep (engine, API, CLI, sandbox, deps). No new diff --git a/README.md b/README.md index 3ae365b..e2a75d1 100644 --- a/README.md +++ b/README.md @@ -2,7 +2,7 @@ **Build once. Run anywhere.** Sandcastle is an open-source, production-ready orchestrator for AI agents. **Describe a workflow in plain English and it builds it** (or write the YAML yourself); run it on **any model** — Claude, GPT, Mistral, or a local model on your own box — and move between them with one line; deploy it **your way** — cloud, your own server, fully air-gapped, or EU-only; and it **gets better over time**, on its own. Local models run at `$0/run` with hard data-residency enforcement and a tamper-evident audit trail; the cloud is there too, with 7 providers and auto-failover, 22 step types, verified templates, and a full dashboard. Sovereign by default. European-built. -[![PyPI](https://img.shields.io/badge/PyPI-v0.40.2-blue?style=flat-square)](https://pypi.org/project/sandcastle-ai/0.40.2/) +[![PyPI](https://img.shields.io/badge/PyPI-v0.40.3-blue?style=flat-square)](https://pypi.org/project/sandcastle-ai/0.40.3/) [![License: BSL 1.1](https://img.shields.io/badge/License-BSL_1.1-blue.svg)](LICENSE) [![Python 3.12+](https://img.shields.io/badge/python-3.12+-blue.svg)](https://www.python.org/downloads/) [![Tests](https://img.shields.io/badge/tests-17900%2B%20passing-brightgreen?style=flat-square)](https://github.com/gizmax/Sandcastle/actions) @@ -722,7 +722,9 @@ All backends share the same `SandboxBackend` protocol - same YAML, same API, sam | Setting | Default | Meaning | |---|---|---| | `CODE_STEPS_ALLOW_UNTRUSTED` | `false` | In-process `code` steps are admin-only by default. Set `true` on single-tenant self-hosted deployments to let any authenticated caller run them; keep `false` for multi-tenant. | +| `CODE_STEPS_OUT_OF_PROCESS` | `true` | Run `code` steps in a separate Python subprocess so a sandbox escape cannot reach the parent process memory (settings, DB session, other tenants' data). The child is really killed on timeout. Set `false` to fall back to the legacy in-process path. | | `MEMORY_MCP_TOKEN` | — | Bearer token for the Memory MCP `streamable-http` transport. Required to start it outside local mode (fails closed); callers must send `Authorization: Bearer `. stdio is unaffected. | +| `MEMORY_MCP_SCOPE_PREFIX` | — | Optional tenant scope prefix for the Memory MCP server. When set, every tool call (`add`/`search`/`forget`/`list_memories`) must operate within this prefix, so an authenticated token cannot cross tenants. Unset preserves single-tenant behavior. | --- diff --git a/deploy/mcp-tunnel/memory-mcp/Chart.yaml b/deploy/mcp-tunnel/memory-mcp/Chart.yaml index 492a04f..c44384d 100644 --- a/deploy/mcp-tunnel/memory-mcp/Chart.yaml +++ b/deploy/mcp-tunnel/memory-mcp/Chart.yaml @@ -3,7 +3,7 @@ name: memory-mcp description: Sandcastle Memory MCP server behind a Cloudflare MCP tunnel (outbound-only, no inbound ingress required) type: application version: 0.1.0 -appVersion: "0.40.2" +appVersion: "0.40.3" keywords: - sandcastle - mcp diff --git a/src/sandcastle/__init__.py b/src/sandcastle/__init__.py index ff4ef35..a0143d2 100644 --- a/src/sandcastle/__init__.py +++ b/src/sandcastle/__init__.py @@ -1,6 +1,6 @@ """Sandcastle - Production-ready workflow orchestrator for AI agents.""" -__version__ = "0.40.2" +__version__ = "0.40.3" __all__ = ["SandcastleClient", "AsyncSandcastleClient", "__version__"] diff --git a/src/sandcastle/config.py b/src/sandcastle/config.py index b14c35e..ce0ef88 100644 --- a/src/sandcastle/config.py +++ b/src/sandcastle/config.py @@ -116,6 +116,13 @@ class Settings(BaseSettings): # intentionally use code steps can opt in via CODE_STEPS_ALLOW_UNTRUSTED=true. code_steps_allow_untrusted: bool = False + # Run "code" steps in a separate Python subprocess (out-of-process isolation) + # so a sandbox escape cannot reach the parent process memory (settings, DB + # session factory, other tenants' data). On by default. Operators can fall + # back to the legacy in-process path via CODE_STEPS_OUT_OF_PROCESS=false if + # the subprocess path misbehaves in their environment. + code_steps_out_of_process: bool = True + # Database (empty = local SQLite mode) database_url: str = "" diff --git a/src/sandcastle/engine/code_subprocess_runner.py b/src/sandcastle/engine/code_subprocess_runner.py new file mode 100644 index 0000000..98b4e88 --- /dev/null +++ b/src/sandcastle/engine/code_subprocess_runner.py @@ -0,0 +1,294 @@ +"""Out-of-process runner for workflow ``code`` steps. + +The in-process ``exec`` path in :mod:`sandcastle.engine.executor` runs +restricted-but-still-Python code inside the API/worker process. Even with the +admin gate, the AST validator, the regex blocklist, and the secret-free file +helpers, a novel sandbox escape would land in the parent's address space next to +``settings``, the DB session factory, and other tenants' in-flight data. + +This module runs the same restricted code in a *separate Python subprocess* that: + +- does NOT inherit the parent's environment secrets (a minimal ``env`` is passed); +- rebuilds the SAME restricted ``exec_globals`` and re-runs the SAME AST + regex + validation the parent uses, by importing ``_validate_code_step_ast``, + ``_CODE_STEP_BLOCKED_PATTERNS`` and ``make_code_sandbox_helpers`` (no forked copy + of the security checks); +- applies POSIX ``resource`` limits (CPU seconds, address space) before exec; +- can be *really killed* on timeout, which the in-process thread path could not do. + +The parent (:func:`run_code_in_subprocess`) enforces the wall-clock timeout and +kills the child if it overruns. The child protocol is a single JSON object on +stdout: ``{"status": "completed", "output": }`` or +``{"status": "failed", "error": ""}``. User ``print`` output is diverted +to stderr so it cannot corrupt the result channel. +""" + +from __future__ import annotations + +import json +import os +import sys +from pathlib import Path +from typing import Any + + +class SubprocessInfraError(RuntimeError): + """Raised for infrastructure failures (spawn / serialization) so the caller + + can fall back to the in-process path instead of failing the step. + """ + + +# --------------------------------------------------------------------------- +# Child side - runs inside the spawned subprocess +# --------------------------------------------------------------------------- + + +def _build_exec_globals(input_data: Any, step_outputs: Any, data_dir: str) -> dict[str, Any]: + """Rebuild the exact restricted globals the in-process path uses. + + Kept byte-for-byte in sync with ``_execute_code_step`` in executor.py: the + same ``__builtins__`` allowlist, the same ``json`` / ``base64`` modules, and + the same ``read_file_b64`` / ``save_file_b64`` helpers bound to ``data_dir``. + """ + import base64 as _b64_mod + + from sandcastle.engine.code_sandbox_helpers import make_code_sandbox_helpers + + resolved_data_dir = Path(data_dir).resolve() + read_file_b64, save_file_b64 = make_code_sandbox_helpers(resolved_data_dir) + + return { + "__builtins__": { + "len": len, + "int": int, + "float": float, + "str": str, + "bool": bool, + "list": list, + "dict": dict, + "set": set, + "tuple": tuple, + "range": range, + "enumerate": enumerate, + "zip": zip, + "map": map, + "filter": filter, + "sorted": sorted, + "min": min, + "max": max, + "sum": sum, + "abs": abs, + "round": round, + "isinstance": isinstance, + "print": print, + "None": None, + "True": True, + "False": False, + }, + "_input": input_data, + "_steps": step_outputs, + "json": json, + "base64": _b64_mod, + "read_file_b64": read_file_b64, + "save_file_b64": save_file_b64, + "result": None, + } + + +def _apply_resource_limits(cpu_seconds: int) -> None: + """Cap CPU time and address space in the child on POSIX. + + On non-POSIX platforms ``resource`` is unavailable, so we rely on the + parent's wall-clock timeout instead. Failures to set a limit are tolerated + (they must never prevent the code from running when limits are simply + unsupported). + """ + try: + import resource + except ImportError: + return + try: + soft = max(1, int(cpu_seconds)) + resource.setrlimit(resource.RLIMIT_CPU, (soft, soft + 1)) + except (ValueError, OSError): + pass + try: + # 4 GiB address-space cap. Generous on purpose: Python plus the imported + # security modules already use a few hundred MB, and we only want to stop + # pathological allocations, not break normal transforms. + cap = 4 * 1024 * 1024 * 1024 + resource.setrlimit(resource.RLIMIT_AS, (cap, cap)) + except (ValueError, OSError): + pass + + +def _child_run(payload: dict[str, Any]) -> dict[str, Any]: + """Validate then execute the code step inside the child process.""" + from sandcastle.engine.executor import ( + _CODE_STEP_BLOCKED_PATTERNS, + _validate_code_step_ast, + ) + + code = payload["code"] + + # Defense in depth: re-run the SAME checks the parent already ran, so the + # child never execs code that failed validation even if invoked directly. + blocked = _CODE_STEP_BLOCKED_PATTERNS.search(code) + if blocked: + return { + "status": "failed", + "error": f"Code step uses blocked pattern '{blocked.group()}'.", + } + ast_error = _validate_code_step_ast(code) + if ast_error: + return {"status": "failed", "error": ast_error} + + exec_globals = _build_exec_globals( + payload.get("_input"), + payload.get("_steps"), + payload["data_dir"], + ) + + _apply_resource_limits(int(payload.get("cpu_seconds", 30))) + + # Divert user print() output to stderr so it cannot corrupt the JSON result + # written to stdout. + real_stdout = sys.stdout + sys.stdout = sys.stderr + try: + exec(code, exec_globals) # noqa: S102 + except Exception as exc: # noqa: BLE001 - surface the message like the in-process path + sys.stdout = real_stdout + return {"status": "failed", "error": str(exc)} + finally: + sys.stdout = real_stdout + + return {"status": "completed", "output": exec_globals.get("result", None)} + + +def _child_main() -> int: + """Entry point when run as ``python -m sandcastle.engine.code_subprocess_runner``.""" + try: + raw = sys.stdin.read() + payload = json.loads(raw) + except Exception as exc: # noqa: BLE001 + sys.stdout.write( + json.dumps({"status": "failed", "error": f"bad runner payload: {exc}"}) + ) + return 0 + result = _child_run(payload) + # default=str keeps a non-JSON-serializable ``result`` from hard-failing the + # step, mirroring the intent that code steps return simple data. + sys.stdout.write(json.dumps(result, default=str)) + return 0 + + +# --------------------------------------------------------------------------- +# Parent side - called from the executor +# --------------------------------------------------------------------------- + + +def _child_env() -> dict[str, str]: + """Build a minimal environment for the child that excludes parent secrets. + + Only what a bare Python subprocess needs to import ``sandcastle`` is passed: + ``PYTHONPATH`` (so the package resolves in dev/editable installs) plus a few + OS-level, non-secret variables. Provider API keys, DB URLs, tokens, and every + other parent env var are intentionally dropped. + """ + env: dict[str, str] = {"PYTHONPATH": os.pathsep.join(sys.path)} + # Non-secret OS plumbing some interpreters/libraries need to start. + for key in ( + "PATH", "HOME", "SYSTEMROOT", "PATHEXT", "TEMP", "TMP", "TMPDIR", "LANG", "LC_ALL", + ): + val = os.environ.get(key) + if val: + env[key] = val + # Keep hash randomization deterministic-free but present; harmless. + return env + + +async def run_code_in_subprocess( + code: str, + input_data: Any, + step_outputs: Any, + data_dir: str, + timeout: int, +) -> dict[str, Any]: + """Run a code step in an isolated subprocess and return a result dict. + + Returns ``{"status": "completed", "output": }`` or + ``{"status": "failed", "error": ""}``. Raises + :class:`SubprocessInfraError` for spawn/serialization problems so the caller + can fall back to the in-process path (a genuine sandbox result never raises). + """ + import asyncio + + payload = { + "code": code, + "_input": input_data, + "_steps": step_outputs, + "data_dir": str(data_dir), + "cpu_seconds": int(timeout), + } + try: + payload_bytes = json.dumps(payload).encode("utf-8") + except (TypeError, ValueError) as exc: + raise SubprocessInfraError( + f"code step input is not JSON-serializable: {exc}" + ) from exc + + try: + proc = await asyncio.create_subprocess_exec( + sys.executable, + "-m", + "sandcastle.engine.code_subprocess_runner", + stdin=asyncio.subprocess.PIPE, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + env=_child_env(), + ) + except (OSError, ValueError) as exc: + raise SubprocessInfraError(f"could not spawn code subprocess: {exc}") from exc + + # Shield communicate() so a timeout kills the child and then lets the same + # coroutine finish draining and closing the pipes, instead of cancelling it + # mid-read (which leaks a pipe reader and emits a ResourceWarning). + comm = asyncio.ensure_future(proc.communicate(input=payload_bytes)) + try: + stdout, stderr = await asyncio.wait_for( + asyncio.shield(comm), + timeout=max(1, int(timeout)), + ) + except asyncio.TimeoutError: + # Real kill - the in-process thread path could not do this. + try: + proc.kill() + except ProcessLookupError: + pass + try: + await comm # drain and close the pipes after the kill + except Exception: # noqa: BLE001 - best effort reap + pass + return { + "status": "failed", + "error": f"Code step timed out after {int(timeout)}s", + } + + text = (stdout or b"").decode("utf-8", "replace").strip() + if not text: + err_tail = (stderr or b"").decode("utf-8", "replace").strip()[-500:] + raise SubprocessInfraError( + f"code subprocess produced no result (exit {proc.returncode}): {err_tail}" + ) + try: + return json.loads(text) + except json.JSONDecodeError as exc: + raise SubprocessInfraError( + f"code subprocess returned malformed result: {exc}: {text[:200]}" + ) from exc + + +if __name__ == "__main__": + raise SystemExit(_child_main()) diff --git a/src/sandcastle/engine/executor.py b/src/sandcastle/engine/executor.py index 3f901fc..a2d9c49 100644 --- a/src/sandcastle/engine/executor.py +++ b/src/sandcastle/engine/executor.py @@ -2841,6 +2841,36 @@ async def _execute_llm_step( ) +def _build_pinned_transport(host_to_ip: dict[str, str]) -> Any: + """Build an httpx transport that dials pre-validated IPs, not re-resolved DNS. + + ``host_to_ip`` maps a hostname to the single allowed IP address already + validated by ``_execute_http_step``'s SSRF pre-flight. The transport rewrites + each outgoing request's URL host to that pinned IP so the connection goes to + the validated address, while preserving the original ``Host`` header (already + set by httpx before the transport runs) and the TLS SNI (via the + ``sni_hostname`` request extension) so certificate validation still uses the + hostname. This closes the DNS-rebind TOCTOU between the pre-flight check and + the actual dial. + """ + import httpx + + class _PinnedHTTPTransport(httpx.AsyncHTTPTransport): + async def handle_async_request(self, request: Any) -> Any: + original_host = request.url.host + pinned_ip = host_to_ip.get(original_host) + if pinned_ip and pinned_ip != original_host: + # Dial the validated IP but keep the hostname for TLS SNI so the + # certificate still validates against the real host. + request.url = request.url.copy_with(host=pinned_ip) + extensions = dict(request.extensions or {}) + extensions.setdefault("sni_hostname", original_host) + request.extensions = extensions + return await super().handle_async_request(request) + + return _PinnedHTTPTransport() + + async def _execute_http_step( step: StepDefinition, context: RunContext, @@ -2862,14 +2892,20 @@ async def _execute_http_step( # SSRF prevention for HTTP steps - block private/internal networks. # Reuse the webhook dispatcher's blocked-range list (IPv4-mapped, CGNAT, # ULA, loopback, ...) so the ranges stay defined in one place. - # TODO(ssrf): this validates the resolved IPs but hands the *hostname* to - # httpx, which re-resolves at connect time. A TTL=0 rebind between this - # check and httpx's own resolution is therefore still theoretically - # possible. Closing that fully needs a custom httpx transport that pins and - # dials the validated address; that (and fail-closing on gaierror, which - # would change behavior for mocked-transport callers) is intentionally - # deferred. Redirects are blocked below to limit the remaining surface, - # mirroring the webhook dispatcher's accepted mitigation. + # + # We resolve the hostname ONCE here, validate every resolved IP against + # the blocked ranges, and record one allowed address. That validated + # address is then pinned at connect time via a custom httpx transport + # (see _build_pinned_transport below), so httpx never re-resolves the + # hostname. This closes the DNS-rebind TOCTOU: a TTL=0 record that flips + # to a private IP between this check and the actual dial can no longer be + # reached, because the dial goes to the address we already validated while + # the original Host header and TLS SNI are preserved. + # + # Compatibility: resolution failures (gaierror) are still swallowed and + # left for httpx to surface, and no pin is applied in that case, so mocked + # or non-resolvable test hosts keep working exactly as before. + _pinned_hosts: dict[str, str] = {} try: import ipaddress as _ipaddress import socket as _socket @@ -2887,6 +2923,7 @@ async def _execute_http_step( if _parsed.hostname: try: _resolved = _socket.getaddrinfo(_parsed.hostname, _parsed.port or 443) + _first_allowed_ip: str | None = None for _, _, _, _, _sockaddr in _resolved: _ip = _ipaddress.ip_address(_sockaddr[0]) for _network in _BLOCKED_NETWORKS: @@ -2897,6 +2934,11 @@ async def _execute_http_step( error=f"HTTP step URL resolves to blocked network ({_ip})", duration_seconds=time.monotonic() - started_at, ) + if _first_allowed_ip is None: + _first_allowed_ip = _sockaddr[0] + # Only pin when we actually resolved a public, allowed IP. + if _first_allowed_ip is not None: + _pinned_hosts[_parsed.hostname] = _first_allowed_ip except _socket.gaierror: pass # DNS resolution failure - let httpx surface it naturally except ImportError: @@ -2999,10 +3041,15 @@ def _load_file_ref(match: _re_file.Match) -> str: len(body) if body else 0, len(url), ) + # Pin the validated address at connect time so httpx cannot re-resolve + # the hostname to a rebound private IP. Transport is None (httpx default) + # when no public IP was pinned (e.g. gaierror path or mocked callers). + _pinned_transport = _build_pinned_transport(_pinned_hosts) if _pinned_hosts else None async with httpx.AsyncClient( timeout=step.timeout, follow_redirects=False, max_redirects=0, + transport=_pinned_transport, ) as client: resp = await client.request( method=cfg.method.upper(), @@ -4966,6 +5013,55 @@ async def _execute_code_step( from sandcastle.engine.code_sandbox_helpers import make_code_sandbox_helpers _code_data_dir = Path(_code_settings.data_dir).resolve() + + _CODE_STEP_TIMEOUT = min(step.timeout, 30) + + # Out-of-process isolation (default). Run the validated code in a separate + # Python subprocess so a sandbox escape cannot reach this process' memory + # (settings, DB session factory, other tenants' data). The subprocess is + # really killed on timeout, which the in-process thread path could not do. + # Operators can fall back to the in-process path via + # CODE_STEPS_OUT_OF_PROCESS=false. Infrastructure failures (spawn / + # serialization) fall back to the in-process path so a step is never lost. + if _code_settings.code_steps_out_of_process: + from sandcastle.engine.code_subprocess_runner import ( + SubprocessInfraError, + run_code_in_subprocess, + ) + + try: + sub = await run_code_in_subprocess( + code=code, + input_data=context.input, + step_outputs=context.step_outputs, + data_dir=str(_code_data_dir), + timeout=_CODE_STEP_TIMEOUT, + ) + except SubprocessInfraError as infra_exc: + logger.warning( + "Code step '%s': subprocess unavailable (%s); " + "falling back to in-process execution", + step.id, infra_exc, + ) + sub = None + + if sub is not None: + duration = time.monotonic() - started_at + if sub.get("status") == "completed": + return StepResult( + step_id=step.id, + output=sub.get("output"), + cost_usd=0.0, + duration_seconds=duration, + status="completed", + ) + return StepResult( + step_id=step.id, + status="failed", + error=sub.get("error", "code step failed"), + duration_seconds=duration, + ) + _read_file_b64, _save_file_b64 = make_code_sandbox_helpers(_code_data_dir) exec_globals: dict[str, Any] = { @@ -5016,8 +5112,6 @@ async def _execute_code_step( # dies with the process on exit. import concurrent.futures - _CODE_STEP_TIMEOUT = min(step.timeout, 30) - def _run_code(): exec(code, exec_globals) # noqa: S102 diff --git a/src/sandcastle/engine/memory_mcp_server.py b/src/sandcastle/engine/memory_mcp_server.py index 3e70dbd..1ed5753 100644 --- a/src/sandcastle/engine/memory_mcp_server.py +++ b/src/sandcastle/engine/memory_mcp_server.py @@ -107,6 +107,34 @@ def _load_memory_module() -> Any: # --------------------------------------------------------------------------- +#: Env var that, when set, pins every tool call to a single tenant scope. +_SCOPE_PREFIX_ENV = "MEMORY_MCP_SCOPE_PREFIX" + + +def _scope_prefix() -> str: + """Return the configured tenant scope prefix, or '' when unset.""" + return (os.environ.get(_SCOPE_PREFIX_ENV) or "").strip() + + +def _enforce_scope_prefix(user_id: str) -> None: + """Reject a ``user_id`` that falls outside ``MEMORY_MCP_SCOPE_PREFIX``. + + A no-op when the prefix is unset (single-tenant / local deployments keep + today's behavior). When set, ``user_id`` must equal the prefix exactly or + start with ``"/"`` so an authenticated caller cannot read/write/delete + another tenant's memories by supplying a different scope string. + """ + prefix = _scope_prefix() + if not prefix: + return + if user_id == prefix or user_id.startswith(prefix + "/"): + return + raise MemoryValidationError( + f"user_id '{user_id}' is outside the configured " + f"{_SCOPE_PREFIX_ENV} '{prefix}'" + ) + + def _validate_user_id(value: Any) -> str: """Validate ``user_id`` for any memory operation.""" if not isinstance(value, str): @@ -313,6 +341,16 @@ def _normalize_record(item: Any) -> dict[str, Any]: "description": "Memory row identifier returned by add() or search().", "minLength": 1, }, + "user_id": { + "type": "string", + "description": ( + "Owning user or scope identifier. Required when the server " + "is started with MEMORY_MCP_SCOPE_PREFIX so the delete stays " + "inside the caller's tenant; optional otherwise." + ), + "minLength": 1, + "maxLength": MAX_USER_ID_LENGTH, + }, }, "required": ["memory_id"], "additionalProperties": False, @@ -389,6 +427,7 @@ async def _tool_add( ) -> dict[str, Any]: text = _validate_text(text) user_id = _validate_user_id(user_id) + _enforce_scope_prefix(user_id) metadata = _validate_metadata(metadata) mem = _load_memory_module() @@ -415,6 +454,7 @@ async def _tool_search( if not isinstance(query, str) or not query.strip(): raise MemoryValidationError("query must be a non-empty string") user_id = _validate_user_id(user_id) + _enforce_scope_prefix(user_id) limit = _validate_limit(limit) mem = _load_memory_module() @@ -427,8 +467,28 @@ async def _tool_search( return {"results": [_normalize_record(r) for r in (records or [])]} -async def _tool_forget(memory_id: str) -> dict[str, Any]: +async def _tool_forget( + memory_id: str, + user_id: str | None = None, +) -> dict[str, Any]: memory_id = _validate_memory_id(memory_id) + # forget() deletes by id and the backend does not scope the delete. When a + # tenant prefix is configured, require the caller to also name the owning + # user_id and prove it is inside the prefix, so a token cannot delete another + # tenant's memory by guessing an id. Backward-compat: when no prefix is set, + # user_id stays optional and behavior is unchanged. + prefix = _scope_prefix() + if prefix: + if user_id is None: + raise MemoryValidationError( + "user_id is required to delete a memory when " + f"{_SCOPE_PREFIX_ENV} is set" + ) + user_id = _validate_user_id(user_id) + _enforce_scope_prefix(user_id) + elif user_id is not None: + # Validate for shape even when no prefix is enforced. + user_id = _validate_user_id(user_id) mem = _load_memory_module() try: deleted = await mem.delete_memory(memory_id) @@ -444,6 +504,7 @@ async def _tool_list_memories( limit: int = DEFAULT_LIST_LIMIT, ) -> dict[str, Any]: user_id = _validate_user_id(user_id) + _enforce_scope_prefix(user_id) limit = _validate_limit(limit) mem = _load_memory_module() try: diff --git a/tests/conftest.py b/tests/conftest.py index b929bfc..db4b020 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -21,6 +21,14 @@ os.close(_test_db_fd) os.environ["DATABASE_URL"] = f"sqlite+aiosqlite:///{_test_db_path}" +# Run code steps in-process during the whole test suite for speed. Production +# defaults to out-of-process isolation, but spawning a Python subprocess per code +# step (~170ms each) across ~17k tests pushes the suite past the CI job timeout. +# Set via env (before any sandcastle import) so child processes tests spawn - the +# CLI, workflow e2e - inherit it too, not just this process' settings singleton. +# The out-of-process path is covered explicitly by test_code_subprocess_runner.py. +os.environ.setdefault("CODE_STEPS_OUT_OF_PROCESS", "false") + @atexit.register def _cleanup_test_db() -> None: @@ -93,6 +101,35 @@ async def _create(): event.remove(engine.sync_engine, "connect", _ensure_schema) +@pytest.fixture(autouse=True) +def _in_process_code_steps(): + """Run code steps in-process during the test suite for speed. + + Production defaults to out-of-process isolation (CODE_STEPS_OUT_OF_PROCESS), + but spawning a Python subprocess per code step adds ~170ms each and, across + the whole suite, pushes it past the CI job timeout. The out-of-process path + is exercised explicitly by tests/test_code_subprocess_runner.py (which calls + the runner directly and flips this flag back on for its integration test). + """ + try: + from sandcastle.config import settings + + prev = settings.code_steps_out_of_process + settings.code_steps_out_of_process = False + except Exception: + prev = None + try: + yield + finally: + if prev is not None: + try: + from sandcastle.config import settings + + settings.code_steps_out_of_process = prev + except Exception: + pass + + @pytest.fixture(autouse=True) def _reset_module_globals(): """Clear in-memory module caches between tests. diff --git a/tests/test_code_subprocess_runner.py b/tests/test_code_subprocess_runner.py new file mode 100644 index 0000000..42ab8e0 --- /dev/null +++ b/tests/test_code_subprocess_runner.py @@ -0,0 +1,168 @@ +"""Tests for out-of-process execution of workflow ``code`` steps. + +These exercise ``code_subprocess_runner.run_code_in_subprocess`` directly (the +child really is a separate Python process) plus the ``_execute_code_step`` +integration that dispatches to it. +""" + +from __future__ import annotations + +import asyncio + +import pytest + +from sandcastle.engine.code_subprocess_runner import ( + SubprocessInfraError, + run_code_in_subprocess, +) + + +def _run(coro): + return asyncio.run(coro) + + +class TestRunCodeInSubprocess: + """Direct exercise of the subprocess runner.""" + + def test_normal_json_transform(self, tmp_path) -> None: + res = _run( + run_code_in_subprocess( + code="result = sum(_input['items']) + _steps['prev']['n']", + input_data={"items": [1, 2, 3]}, + step_outputs={"prev": {"n": 10}}, + data_dir=str(tmp_path), + timeout=15, + ) + ) + assert res["status"] == "completed" + assert res["output"] == 16 + + def test_blocklist_violation_rejected(self, tmp_path) -> None: + res = _run( + run_code_in_subprocess( + code="result = eval('1+1')", + input_data={}, + step_outputs={}, + data_dir=str(tmp_path), + timeout=15, + ) + ) + assert res["status"] == "failed" + assert "blocked pattern" in res["error"].lower() + + def test_import_rejected_by_ast(self, tmp_path) -> None: + res = _run( + run_code_in_subprocess( + code="import os\nresult = 1", + input_data={}, + step_outputs={}, + data_dir=str(tmp_path), + timeout=15, + ) + ) + assert res["status"] == "failed" + assert "import" in res["error"].lower() + + def test_child_has_no_parent_settings(self, tmp_path) -> None: + # The parent's `settings` object must not exist in the child globals. + res = _run( + run_code_in_subprocess( + code="result = settings", + input_data={}, + step_outputs={}, + data_dir=str(tmp_path), + timeout=15, + ) + ) + assert res["status"] == "failed" + assert "settings" in res["error"] + + def test_timeout_kills_child(self, tmp_path) -> None: + res = _run( + run_code_in_subprocess( + code="while True:\n x = 1", + input_data={}, + step_outputs={}, + data_dir=str(tmp_path), + timeout=2, + ) + ) + assert res["status"] == "failed" + assert "timed out" in res["error"].lower() + + def test_read_file_b64_helper_works(self, tmp_path) -> None: + src = tmp_path / "note.txt" + src.write_text("hello") + res = _run( + run_code_in_subprocess( + code=f"result = read_file_b64({str(src)!r})", + input_data={}, + step_outputs={}, + data_dir=str(tmp_path), + timeout=15, + ) + ) + assert res["status"] == "completed" + # base64 of "hello" + assert res["output"] == "aGVsbG8=" + + def test_save_file_b64_helper_works(self, tmp_path) -> None: + src = tmp_path / "big.bin" + src.write_bytes(b"data") + res = _run( + run_code_in_subprocess( + code=f"result = save_file_b64({str(src)!r}, 'out.txt')", + input_data={}, + step_outputs={}, + data_dir=str(tmp_path), + timeout=15, + ) + ) + assert res["status"] == "completed" + assert res["output"].startswith("@file:") + saved = tmp_path / "tmp" / "out.txt" + assert saved.exists() + + def test_non_serializable_input_raises_infra_error(self, tmp_path) -> None: + # A non-JSON-serializable input signals an infra failure so the caller + # can fall back to the in-process path. + with pytest.raises(SubprocessInfraError): + _run( + run_code_in_subprocess( + code="result = 1", + input_data={"bad": object()}, + step_outputs={}, + data_dir=str(tmp_path), + timeout=15, + ) + ) + + +class TestExecuteCodeStepIntegration: + """The executor dispatches code steps out-of-process by default.""" + + @pytest.mark.asyncio + async def test_code_step_runs_out_of_process(self, monkeypatch) -> None: + from sandcastle.config import settings + from sandcastle.engine.dag import CodeConfig, StepDefinition + from sandcastle.engine.executor import RunContext, _execute_code_step + + # The suite defaults code steps to in-process (conftest); force the + # out-of-process path here so this integration test actually exercises it. + monkeypatch.setattr(settings, "code_steps_out_of_process", True) + + step = StepDefinition( + id="t", + type="code", + code_config=CodeConfig(code="result = len(_input['text'])"), + ) + ctx = RunContext( + run_id="r1", + input={"text": "hello"}, + step_outputs={}, + admin_trusted=True, + ) + result = await _execute_code_step(step, ctx) + assert result.status == "completed" + assert result.output == 5 + assert result.cost_usd == 0.0 diff --git a/tests/test_http_ssrf_pinning.py b/tests/test_http_ssrf_pinning.py new file mode 100644 index 0000000..6ad4ada --- /dev/null +++ b/tests/test_http_ssrf_pinning.py @@ -0,0 +1,138 @@ +"""Tests for HTTP-step SSRF IP-pinning (DNS-rebind TOCTOU closure). + +The executor resolves and validates a hostname once, then pins the validated IP +at connect time via a custom httpx transport so httpx cannot re-resolve to a +rebound private IP. These tests prove the transport dials the exact validated IP +while preserving the Host header and TLS SNI. +""" + +from __future__ import annotations + +import httpx +import pytest + +from sandcastle.engine.dag import StepDefinition +from sandcastle.engine.dag import HttpConfig +from sandcastle.engine.executor import ( + RunContext, + _build_pinned_transport, + _execute_http_step, +) + +_PUBLIC_IP = "93.184.216.34" # example.com, public +_PRIVATE_IP = "169.254.169.254" # cloud metadata, blocked + + +class TestPinnedTransport: + """Unit tests for the pinning transport itself.""" + + @pytest.mark.asyncio + async def test_dials_pinned_ip_and_preserves_host_and_sni(self, monkeypatch) -> None: + captured: dict = {} + + async def _fake_parent(self, request): # noqa: ANN001 + captured["host"] = request.url.host + captured["header_host"] = request.headers.get("host") + captured["sni"] = request.extensions.get("sni_hostname") + return httpx.Response(200, json={"ok": True}, request=request) + + monkeypatch.setattr(httpx.AsyncHTTPTransport, "handle_async_request", _fake_parent) + + transport = _build_pinned_transport({"api.example.com": _PUBLIC_IP}) + request = httpx.Request("GET", "https://api.example.com/data") + await transport.handle_async_request(request) + + # The connection host is rewritten to the validated IP... + assert captured["host"] == _PUBLIC_IP + # ...while the Host header and TLS SNI still name the real hostname. + assert captured["header_host"] == "api.example.com" + assert captured["sni"] == "api.example.com" + + +class TestHttpStepPinning: + """End-to-end: _execute_http_step dials the validated IP, not re-resolved DNS.""" + + @pytest.mark.asyncio + async def test_request_dialed_to_validated_public_ip(self, monkeypatch) -> None: + # Pre-flight resolves the hostname to a public IP. + def _fake_getaddrinfo(host, port, *a, **k): # noqa: ANN001 + return [(2, 1, 6, "", (_PUBLIC_IP, 443))] + + monkeypatch.setattr("socket.getaddrinfo", _fake_getaddrinfo) + + captured: dict = {} + + async def _fake_parent(self, request): # noqa: ANN001 + captured["host"] = request.url.host + captured["sni"] = request.extensions.get("sni_hostname") + return httpx.Response(200, json={"ok": True}, request=request) + + monkeypatch.setattr(httpx.AsyncHTTPTransport, "handle_async_request", _fake_parent) + + step = StepDefinition( + id="h", + type="http", + http_config=HttpConfig(url="https://api.example.com/data", method="GET"), + ) + ctx = RunContext(run_id="r", input={}, step_outputs={}, admin_trusted=True) + result = await _execute_http_step(step, ctx) + + assert result.status == "completed" + assert captured["host"] == _PUBLIC_IP + assert captured["sni"] == "api.example.com" + + @pytest.mark.asyncio + async def test_rebind_to_private_ip_cannot_be_reached(self, monkeypatch) -> None: + # First resolution (pre-flight) yields a public IP and is validated+pinned. + # A subsequent rebind to a private IP is irrelevant: the transport dials + # the already-validated public IP and never re-resolves the hostname. + calls = {"n": 0} + + def _rebinding_getaddrinfo(host, port, *a, **k): # noqa: ANN001 + calls["n"] += 1 + if calls["n"] == 1: + return [(2, 1, 6, "", (_PUBLIC_IP, 443))] + return [(2, 1, 6, "", (_PRIVATE_IP, 443))] + + monkeypatch.setattr("socket.getaddrinfo", _rebinding_getaddrinfo) + + captured: dict = {} + + async def _fake_parent(self, request): # noqa: ANN001 + captured["host"] = request.url.host + return httpx.Response(200, json={"ok": True}, request=request) + + monkeypatch.setattr(httpx.AsyncHTTPTransport, "handle_async_request", _fake_parent) + + step = StepDefinition( + id="h", + type="http", + http_config=HttpConfig(url="https://api.example.com/data", method="GET"), + ) + ctx = RunContext(run_id="r", input={}, step_outputs={}, admin_trusted=True) + result = await _execute_http_step(step, ctx) + + assert result.status == "completed" + # The dial went to the validated public IP, never the rebound private IP. + assert captured["host"] == _PUBLIC_IP + assert captured["host"] != _PRIVATE_IP + + @pytest.mark.asyncio + async def test_preflight_blocks_private_resolution(self, monkeypatch) -> None: + # If the hostname resolves to a blocked network up front, the step fails + # before any request is dialed. + def _fake_getaddrinfo(host, port, *a, **k): # noqa: ANN001 + return [(2, 1, 6, "", (_PRIVATE_IP, 443))] + + monkeypatch.setattr("socket.getaddrinfo", _fake_getaddrinfo) + + step = StepDefinition( + id="h", + type="http", + http_config=HttpConfig(url="https://metadata.internal/data", method="GET"), + ) + ctx = RunContext(run_id="r", input={}, step_outputs={}, admin_trusted=True) + result = await _execute_http_step(step, ctx) + + assert result.status == "failed" + assert "blocked network" in result.error diff --git a/tests/test_memory_mcp_server.py b/tests/test_memory_mcp_server.py index 2dfd9b0..4335de6 100644 --- a/tests/test_memory_mcp_server.py +++ b/tests/test_memory_mcp_server.py @@ -216,6 +216,80 @@ def test_list_memories_default_limit_used(self, fake_memory: _FakeMemoryModule) assert call.kwargs.get("limit") == mms.DEFAULT_LIST_LIMIT +# --------------------------------------------------------------------------- +# Per-tenant scope enforcement (MEMORY_MCP_SCOPE_PREFIX) +# --------------------------------------------------------------------------- + + +class TestScopePrefixEnforcement: + """With a scope prefix set, tools must not cross tenants.""" + + def test_no_prefix_allows_any_user_id( + self, fake_memory: _FakeMemoryModule, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.delenv("MEMORY_MCP_SCOPE_PREFIX", raising=False) + # Any user_id works and no error is raised when the prefix is unset. + _run(mms._tool_add("hi", "tenant:other/user:9")) + _run(mms._tool_search("q", "tenant:other/user:9")) + _run(mms._tool_list_memories("tenant:other/user:9")) + _run(mms._tool_forget("mem_1")) + fake_memory.delete_memory.assert_awaited_once_with("mem_1") + + def test_add_in_prefix_allowed( + self, fake_memory: _FakeMemoryModule, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setenv("MEMORY_MCP_SCOPE_PREFIX", "tenant:acme") + # Exact prefix and a child scope are both allowed. + _run(mms._tool_add("hi", "tenant:acme")) + _run(mms._tool_add("hi", "tenant:acme/user:1")) + assert fake_memory.save_memory.await_count == 2 + + def test_add_out_of_prefix_rejected( + self, fake_memory: _FakeMemoryModule, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setenv("MEMORY_MCP_SCOPE_PREFIX", "tenant:acme") + with pytest.raises(mms.MemoryValidationError): + _run(mms._tool_add("hi", "tenant:evil/user:1")) + # A prefix that is only a string-prefix but not a scope boundary is rejected. + with pytest.raises(mms.MemoryValidationError): + _run(mms._tool_add("hi", "tenant:acme-evil")) + fake_memory.save_memory.assert_not_awaited() + + def test_search_and_list_out_of_prefix_rejected( + self, fake_memory: _FakeMemoryModule, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setenv("MEMORY_MCP_SCOPE_PREFIX", "tenant:acme") + with pytest.raises(mms.MemoryValidationError): + _run(mms._tool_search("q", "tenant:evil")) + with pytest.raises(mms.MemoryValidationError): + _run(mms._tool_list_memories("tenant:evil")) + fake_memory.load_memories.assert_not_awaited() + + def test_forget_requires_user_id_when_prefix_set( + self, fake_memory: _FakeMemoryModule, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setenv("MEMORY_MCP_SCOPE_PREFIX", "tenant:acme") + with pytest.raises(mms.MemoryValidationError): + _run(mms._tool_forget("mem_1")) + fake_memory.delete_memory.assert_not_awaited() + + def test_forget_rejects_out_of_prefix_owner( + self, fake_memory: _FakeMemoryModule, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setenv("MEMORY_MCP_SCOPE_PREFIX", "tenant:acme") + with pytest.raises(mms.MemoryValidationError): + _run(mms._tool_forget("mem_1", user_id="tenant:evil/user:1")) + fake_memory.delete_memory.assert_not_awaited() + + def test_forget_allows_in_prefix_owner( + self, fake_memory: _FakeMemoryModule, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setenv("MEMORY_MCP_SCOPE_PREFIX", "tenant:acme") + result = _run(mms._tool_forget("mem_1", user_id="tenant:acme/user:1")) + fake_memory.delete_memory.assert_awaited_once_with("mem_1") + assert result == {"memory_id": "mem_1", "deleted": True} + + # --------------------------------------------------------------------------- # Resources # ---------------------------------------------------------------------------