diff --git a/src/autoskillit/server/_guards.py b/src/autoskillit/server/_guards.py index b713e6005..e8facbcd9 100644 --- a/src/autoskillit/server/_guards.py +++ b/src/autoskillit/server/_guards.py @@ -6,9 +6,12 @@ from pathlib import Path from typing import TYPE_CHECKING, Any +import regex as re + from autoskillit.core import ( InputContractResolver, SessionType, + extract_bash_write_targets, extract_path_arg, extract_positional_args, extract_skill_name, @@ -24,6 +27,17 @@ logger = get_logger(__name__) +RECIPE_READ_DENY_TRIGGER: str = "must not read recipe/skill/agent files directly" + +_RECIPE_READ_CMD_PATTERNS: list[re.Pattern[str]] = [ + re.compile(r"(?:\.autoskillit|src/autoskillit)/recipes/.*\.ya?ml"), + re.compile(r"src/autoskillit/skills(?:_extended)?/.*/SKILL\.md"), + re.compile(r"src/autoskillit/agents/.*\.md"), +] + +_RECIPE_READ_CALLABLE_PATTERN: re.Pattern[str] = re.compile(r"autoskillit\.recipe\.(?!_cmd_rpc)") + + def _get_ctx(): # type: ignore[return] from autoskillit.server._state import _get_ctx as _ctx_fn # circular-break @@ -132,6 +146,57 @@ def _require_enabled() -> str | None: return None +def _check_recipe_read_prohibition( + *, cmd: str | None = None, callable_name: str | None = None +) -> str | None: + """Deny recipe/skill/agent file reads and recipe module callables. + + Headless-only: interactive sessions bypass this guard. + Returns gate_error_result JSON on match, None on pass. + """ + if os.environ.get("AUTOSKILLIT_HEADLESS") != "1": + return None + if cmd is not None: + for pattern in _RECIPE_READ_CMD_PATTERNS: + if pattern.search(cmd): + return gate_error_result( + f"run_cmd {RECIPE_READ_DENY_TRIGGER}. " + "Use load_recipe to recall step definitions or the Skill tool " + "for skill instructions." + ) + if callable_name is not None: + if _RECIPE_READ_CALLABLE_PATTERN.match(callable_name): + return gate_error_result( + f"run_python {RECIPE_READ_DENY_TRIGGER}. " + "Use load_recipe to recall step definitions." + ) + return None + + +def _check_write_target_boundary( + cmd: str, cwd: str, allowed_prefixes: tuple[str, ...] +) -> str | None: + """Deny run_cmd writes outside allowed prefix directories. + + Fail-open: returns None when allowed_prefixes is empty (no write scope configured). + Returns gate_error_result JSON when a write target falls outside all prefixes. + """ + if not allowed_prefixes: + return None + targets = extract_bash_write_targets(cmd, cwd) + if not targets: + return None + normalized = tuple(os.path.realpath(p).rstrip("/") + "/" for p in allowed_prefixes) + for target in targets: + resolved = os.path.realpath(target) + if not any(resolved.startswith(pfx) for pfx in normalized): + return gate_error_result( + f"run_cmd write target {target!r} is outside allowed write prefixes: " + f"{', '.join(allowed_prefixes)}" + ) + return None + + def _validate_skill_command(skill_command: str) -> str | None: """Return error JSON if skill_command does not start with '/'. diff --git a/src/autoskillit/server/tools/tools_execution.py b/src/autoskillit/server/tools/tools_execution.py index b145c3ac1..1597d7366 100644 --- a/src/autoskillit/server/tools/tools_execution.py +++ b/src/autoskillit/server/tools/tools_execution.py @@ -42,6 +42,8 @@ from autoskillit.server._guards import ( _check_dry_walkthrough, _check_input_contracts, + _check_recipe_read_prohibition, + _check_write_target_boundary, _require_enabled, _require_orchestrator_or_higher, _validate_skill_command, @@ -283,6 +285,22 @@ def _has_active_locks(order_id: str) -> bool: return any(v is False for steps in locked_steps.values() for v in steps.values()) +def _derive_run_cmd_write_prefixes() -> tuple[str, ...]: + """Read allowed write prefixes from environment. + + Mirrors the env-var resolution logic in hooks/guards/write_guard.py: + AUTOSKILLIT_ALLOWED_WRITE_PREFIXES (colon-separated) takes precedence over + AUTOSKILLIT_ALLOWED_WRITE_PREFIX (single value). + """ + multi = os.environ.get("AUTOSKILLIT_ALLOWED_WRITE_PREFIXES", "") + if multi: + return tuple(p for p in multi.split(":") if p) + single = os.environ.get("AUTOSKILLIT_ALLOWED_WRITE_PREFIX", "") + if single: + return (single,) + return () + + @mcp.tool(tags={"autoskillit", "kitchen", "kitchen-core"}, annotations={"readOnlyHint": True}) @_cancellation_shield(result_type="run_cmd") @track_response_size("run_cmd") @@ -307,8 +325,18 @@ async def run_cmd( return tier_gate if (gate := _require_enabled()) is not None: return gate + if (gate := _check_recipe_read_prohibition(cmd=cmd)) is not None: + return gate + if ( + gate := _check_write_target_boundary(cmd, cwd, _derive_run_cmd_write_prefixes()) + ) is not None: + return gate try: with structlog.contextvars.bound_contextvars(tool="run_cmd", cwd=cwd): + if not _derive_run_cmd_write_prefixes(): + logger.debug( + "run_cmd: no write prefixes configured — write boundary guard inactive" + ) logger.info("run_cmd", cmd=cmd[:80], cwd=cwd) await _notify( ctx, "info", f"run_cmd: {cmd[:80]}", "autoskillit.run_cmd", extra={"cwd": cwd} @@ -398,6 +426,8 @@ async def run_python( return tier_gate if (gate := _require_enabled()) is not None: return gate + if (gate := _check_recipe_read_prohibition(callable_name=callable)) is not None: + return gate try: with structlog.contextvars.bound_contextvars(tool="run_python"): logger.info("run_python", callable=callable, timeout=timeout) diff --git a/tests/_test_filter.py b/tests/_test_filter.py index cccdeb176..861e242ae 100644 --- a/tests/_test_filter.py +++ b/tests/_test_filter.py @@ -254,7 +254,7 @@ class ImportContext(enum.StrEnum): "_type_tradition_manifest": frozenset({"core"}), "_step_context": frozenset({"core", "execution", "pipeline", "server"}), "_execution_marker": frozenset({"core", "execution", "fleet", "server"}), - "bash_write_targets": frozenset({"core", "execution"}), + "bash_write_targets": frozenset({"core", "execution", "server"}), } # Narrow per-module cascade for execution/. Modules not listed here fall through diff --git a/tests/arch/test_subpackage_isolation.py b/tests/arch/test_subpackage_isolation.py index 80403cd54..436b216fe 100644 --- a/tests/arch/test_subpackage_isolation.py +++ b/tests/arch/test_subpackage_isolation.py @@ -982,12 +982,13 @@ def test_data_directories_are_not_python_packages() -> None: "; supports_quota_check bool at call sites replaces resolve_provider (+3 net lines)", ), "tools_execution.py": ( - 1130, + 1150, "REQ-CNST-010-E8: execution tool handlers — run_cmd/run_python/run_skill are the " "three primary execution paths; fail-closed existence gate, empty-closure gate " "for fabricated skill name rejection, _check_backend_compat fail-closed gate " "with resolver-absent fallback via extract_skill_name, and fix-required hook " - "dispatch gate add defense-in-depth checks", + "dispatch gate add defense-in-depth checks; server-side recipe-read prohibition " + "and write-target boundary guards add defense-in-depth gate checks", ), "execution/backends/codex.py": ( 1114, diff --git a/tests/hooks/test_hook_sync.py b/tests/hooks/test_hook_sync.py index dc5ae92c5..2d2f12ad2 100644 --- a/tests/hooks/test_hook_sync.py +++ b/tests/hooks/test_hook_sync.py @@ -100,3 +100,13 @@ def test_ingredient_lock_deny_trigger_sync(): f"server={INGREDIENT_LOCK_DENY_PREFIX!r} vs " f"hook={INGREDIENT_LOCK_DENY_TRIGGER!r}" ) + + +def test_recipe_read_deny_trigger_sync(): + """RECIPE_READ_DENY_TRIGGER must be identical in server guards and hook guard.""" + from autoskillit.hooks.guards.recipe_read_guard import RECIPE_READ_DENY_TRIGGER as _HOOK + from autoskillit.server._guards import RECIPE_READ_DENY_TRIGGER as _SERVER + + assert _SERVER == _HOOK, ( + f"Recipe read deny trigger mismatch: server={_SERVER!r} vs hook={_HOOK!r}" + ) diff --git a/tests/server/AGENTS.md b/tests/server/AGENTS.md index 2000f4b03..24d54cb2a 100644 --- a/tests/server/AGENTS.md +++ b/tests/server/AGENTS.md @@ -115,6 +115,8 @@ Server tool handler unit tests — kitchen, execution, CI, clone, workspace tool | `test_tools_recipe.py` | Tests for autoskillit server validate_recipe tool and recipe docstring contracts | | `test_tools_report_bug.py` | Tests for report_bug MCP tool handler and supporting helpers (_parse_fingerprint, _extract_block, _parse_prepare_result, _parse_enrich_result) | | `test_tools_run_cmd.py` | Tests for run_cmd and run_python MCP tool handlers | +| `test_tools_run_cmd_invariants.py` | Server-side invariant tests for run_cmd: recipe-read prohibition and write-target boundary | +| `test_tools_run_python_invariants.py` | Server-side invariant tests for run_python: recipe-read prohibition | | `test_tools_run_cmd_unit.py` | Unit tests for run_cmd: observability, timing, and headless gate enforcement | | `test_tools_run_python.py` | Unit tests for run_python: observability and headless gate enforcement | | `test_tools_run_python_cwd.py` | Tests for run_python work_dir path resolution: anchors relative output_dir to work_dir | diff --git a/tests/server/test_guards_module.py b/tests/server/test_guards_module.py index 459147caa..53368720b 100644 --- a/tests/server/test_guards_module.py +++ b/tests/server/test_guards_module.py @@ -30,3 +30,19 @@ def test_require_fleet_doc_mentions_l3(): assert "L3" in doc assert "L1" in doc assert "L2" in doc + + +def test_check_recipe_read_prohibition_importable(): + from autoskillit.server._guards import _check_recipe_read_prohibition + + doc = _check_recipe_read_prohibition.__doc__ or "" + assert "recipe" in doc.lower() + assert "headless" in doc.lower() + + +def test_check_write_target_boundary_importable(): + from autoskillit.server._guards import _check_write_target_boundary + + doc = _check_write_target_boundary.__doc__ or "" + assert "write" in doc.lower() + assert "prefix" in doc.lower() diff --git a/tests/server/test_tools_run_cmd_invariants.py b/tests/server/test_tools_run_cmd_invariants.py new file mode 100644 index 000000000..93b25fadf --- /dev/null +++ b/tests/server/test_tools_run_cmd_invariants.py @@ -0,0 +1,114 @@ +"""Server-side invariant tests for run_cmd: recipe-read prohibition and write-target boundary.""" + +from __future__ import annotations + +import json + +import pytest + +from autoskillit.server._guards import RECIPE_READ_DENY_TRIGGER +from autoskillit.server.tools.tools_execution import run_cmd +from tests.conftest import _make_result + +pytestmark = [pytest.mark.layer("server"), pytest.mark.small] + + +class TestRecipeReadProhibitionCmd: + """run_cmd denies recipe/skill/agent file access in headless sessions.""" + + @pytest.fixture(autouse=True) + def _headless(self, monkeypatch): + monkeypatch.setenv("AUTOSKILLIT_HEADLESS", "1") + monkeypatch.setenv("AUTOSKILLIT_SESSION_TYPE", "orchestrator") + + @pytest.mark.anyio + async def test_denies_recipe_yaml_path(self, tool_ctx_kitchen_open): + result = json.loads(await run_cmd(cmd="cat .autoskillit/recipes/deploy.yaml", cwd="/tmp")) + assert result["success"] is False + assert result["subtype"] == "gate_error" + assert RECIPE_READ_DENY_TRIGGER in result["result"] + + @pytest.mark.anyio + async def test_denies_src_recipe_yaml(self, tool_ctx_kitchen_open): + result = json.loads(await run_cmd(cmd="head src/autoskillit/recipes/base.yml", cwd="/tmp")) + assert result["subtype"] == "gate_error" + + @pytest.mark.anyio + async def test_denies_skill_md_path(self, tool_ctx_kitchen_open): + result = json.loads( + await run_cmd(cmd="cat src/autoskillit/skills/open-kitchen/SKILL.md", cwd="/tmp") + ) + assert result["subtype"] == "gate_error" + + @pytest.mark.anyio + async def test_denies_skills_extended_skill_md(self, tool_ctx_kitchen_open): + result = json.loads( + await run_cmd( + cmd="cat src/autoskillit/skills_extended/audit-arch/SKILL.md", + cwd="/tmp", + ) + ) + assert result["subtype"] == "gate_error" + + @pytest.mark.anyio + async def test_denies_agent_md_path(self, tool_ctx_kitchen_open): + result = json.loads( + await run_cmd(cmd="cat src/autoskillit/agents/explorer.md", cwd="/tmp") + ) + assert result["subtype"] == "gate_error" + + @pytest.mark.anyio + async def test_allows_benign_cmd(self, tool_ctx_kitchen_open): + tool_ctx_kitchen_open.runner.push(_make_result(0, "hello\n", "")) + result = json.loads(await run_cmd(cmd="echo hello", cwd="/tmp")) + assert result["success"] is True + + @pytest.mark.anyio + async def test_allows_non_recipe_python_file(self, tool_ctx_kitchen_open): + tool_ctx_kitchen_open.runner.push(_make_result(0, "", "")) + result = json.loads(await run_cmd(cmd="cat src/autoskillit/core/__init__.py", cwd="/tmp")) + assert result["success"] is True + + @pytest.mark.anyio + async def test_allows_in_non_headless(self, tool_ctx_kitchen_open, monkeypatch): + monkeypatch.delenv("AUTOSKILLIT_HEADLESS", raising=False) + tool_ctx_kitchen_open.runner.push(_make_result(0, "", "")) + result = json.loads(await run_cmd(cmd="cat .autoskillit/recipes/deploy.yaml", cwd="/tmp")) + assert result["success"] is True + + +class TestWriteTargetBoundaryCmd: + """run_cmd denies writes outside allowed prefixes; fails open when unset.""" + + @pytest.fixture(autouse=True) + def _headless(self, monkeypatch): + monkeypatch.setenv("AUTOSKILLIT_HEADLESS", "1") + monkeypatch.setenv("AUTOSKILLIT_SESSION_TYPE", "orchestrator") + + @pytest.mark.anyio + async def test_denies_write_outside_allowed_prefix(self, tool_ctx_kitchen_open, monkeypatch): + monkeypatch.setenv("AUTOSKILLIT_ALLOWED_WRITE_PREFIXES", "/safe/dir") + result = json.loads(await run_cmd(cmd="echo x > /forbidden/file.txt", cwd="/tmp")) + assert result["success"] is False + assert result["subtype"] == "gate_error" + assert "outside allowed write prefixes" in result["result"] + + @pytest.mark.anyio + async def test_allows_write_inside_prefix(self, tool_ctx_kitchen_open, monkeypatch): + monkeypatch.setenv("AUTOSKILLIT_ALLOWED_WRITE_PREFIXES", "/safe/dir") + tool_ctx_kitchen_open.runner.push(_make_result(0, "", "")) + result = json.loads(await run_cmd(cmd="echo x > /safe/dir/file.txt", cwd="/tmp")) + assert result["success"] is True + + @pytest.mark.anyio + async def test_allows_any_write_when_no_prefix(self, tool_ctx_kitchen_open): + tool_ctx_kitchen_open.runner.push(_make_result(0, "", "")) + result = json.loads(await run_cmd(cmd="echo x > /anywhere/file.txt", cwd="/tmp")) + assert result["success"] is True + + @pytest.mark.anyio + async def test_allows_read_only_cmd_with_prefix(self, tool_ctx_kitchen_open, monkeypatch): + monkeypatch.setenv("AUTOSKILLIT_ALLOWED_WRITE_PREFIXES", "/safe/dir") + tool_ctx_kitchen_open.runner.push(_make_result(0, "", "")) + result = json.loads(await run_cmd(cmd="ls /forbidden/", cwd="/tmp")) + assert result["success"] is True diff --git a/tests/server/test_tools_run_python_invariants.py b/tests/server/test_tools_run_python_invariants.py new file mode 100644 index 000000000..6b4b5ab3d --- /dev/null +++ b/tests/server/test_tools_run_python_invariants.py @@ -0,0 +1,63 @@ +"""Server-side invariant tests for run_python: recipe-read prohibition.""" + +from __future__ import annotations + +import json +from unittest.mock import AsyncMock + +import pytest + +from autoskillit.server._guards import RECIPE_READ_DENY_TRIGGER +from autoskillit.server.tools.tools_execution import run_python + +pytestmark = [pytest.mark.layer("server"), pytest.mark.small] + + +class TestRecipeReadProhibitionCallable: + """run_python denies autoskillit.recipe.* callables in headless sessions.""" + + @pytest.fixture(autouse=True) + def _headless(self, monkeypatch): + monkeypatch.setenv("AUTOSKILLIT_HEADLESS", "1") + monkeypatch.setenv("AUTOSKILLIT_SESSION_TYPE", "orchestrator") + + @pytest.mark.anyio + async def test_denies_recipe_callable(self, tool_ctx_kitchen_open): + result = json.loads(await run_python(callable="autoskillit.recipe.load_recipe")) + assert result["success"] is False + assert result["subtype"] == "gate_error" + assert RECIPE_READ_DENY_TRIGGER in result["result"] + + @pytest.mark.anyio + async def test_denies_recipe_submodule_callable(self, tool_ctx_kitchen_open): + result = json.loads(await run_python(callable="autoskillit.recipe.schema.validate")) + assert result["subtype"] == "gate_error" + + @pytest.mark.anyio + async def test_allows_cmd_rpc_callable(self, tool_ctx_kitchen_open, monkeypatch): + mock = AsyncMock(return_value={"success": True, "result": "ok"}) + monkeypatch.setattr("autoskillit.server.tools.tools_execution._import_and_call", mock) + result = json.loads(await run_python(callable="autoskillit.recipe._cmd_rpc")) + assert result["success"] is True + + @pytest.mark.anyio + async def test_allows_cmd_rpc_batch_callable(self, tool_ctx_kitchen_open, monkeypatch): + mock = AsyncMock(return_value={"success": True, "result": "ok"}) + monkeypatch.setattr("autoskillit.server.tools.tools_execution._import_and_call", mock) + result = json.loads(await run_python(callable="autoskillit.recipe._cmd_rpc_batch")) + assert result["success"] is True + + @pytest.mark.anyio + async def test_allows_non_recipe_callable(self, tool_ctx_kitchen_open, monkeypatch): + mock = AsyncMock(return_value={"success": True, "result": "ok"}) + monkeypatch.setattr("autoskillit.server.tools.tools_execution._import_and_call", mock) + result = json.loads(await run_python(callable="autoskillit.core.paths.pkg_root")) + assert result["success"] is True + + @pytest.mark.anyio + async def test_allows_in_non_headless(self, tool_ctx_kitchen_open, monkeypatch): + monkeypatch.delenv("AUTOSKILLIT_HEADLESS", raising=False) + mock = AsyncMock(return_value={"success": True, "result": "ok"}) + monkeypatch.setattr("autoskillit.server.tools.tools_execution._import_and_call", mock) + result = json.loads(await run_python(callable="autoskillit.recipe.load_recipe")) + assert result["success"] is True