Implementation Plan: T5-P5-A3-WP1 Server-Side Recipe-Read Prohibition & Write-Target Boundary

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 '/'.
 

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)

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

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,

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

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 |

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

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

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