diff --git a/src/autoskillit/hooks/__init__.py b/src/autoskillit/hooks/__init__.py index 7f04230fc..4feef35f1 100644 --- a/src/autoskillit/hooks/__init__.py +++ b/src/autoskillit/hooks/__init__.py @@ -11,7 +11,11 @@ HookDef, generate_hooks_json, ) -from autoskillit.hooks._command_classification import _INTERPRETER_LINE_RE, _WRITE_APIS_RE +from autoskillit.hooks._command_classification import ( + _INTERPRETER_LINE_RE, + _WRITE_APIS_RE, + command_has_blocked_protected_path_read, +) from autoskillit.hooks.formatters._fmt_primitives import _HOOK_CONFIG_PATH_COMPONENTS from autoskillit.hooks.guards.branch_protection_guard import BRANCH_PROTECTION_DENY_TRIGGER from autoskillit.hooks.guards.review_loop_gate import REVIEW_LOOP_DENY_TRIGGER @@ -30,5 +34,6 @@ "_HOOK_CONFIG_PATH_COMPONENTS", "_INTERPRETER_LINE_RE", "_WRITE_APIS_RE", + "command_has_blocked_protected_path_read", "generate_hooks_json", ] diff --git a/src/autoskillit/hooks/_command_classification.py b/src/autoskillit/hooks/_command_classification.py index 4c338df56..bd70d7cd6 100644 --- a/src/autoskillit/hooks/_command_classification.py +++ b/src/autoskillit/hooks/_command_classification.py @@ -11,6 +11,7 @@ import re import shlex from collections.abc import Sequence +from typing import Protocol _INTERPRETER_RE = re.compile( r"(?:^|&&|\|\||;)\s*(?:env\s+)?(?:python3?|perl|ruby|node)\s+" @@ -74,6 +75,71 @@ re.DOTALL, ) +_PROTECTED_PATH_METADATA_GIT_SUBCOMMANDS: frozenset[str] = frozenset({"add", "diff", "status"}) +# Command wrappers whose only effect is to invoke the next command with +# adjusted environment/priority. The verb is the token after the wrapper. +# 'xargs' is intentionally excluded: it dispatches a downstream reader and +# adding it would let `xargs cat src/.../foo.yaml` reach the reader check, +# weakening xargs-chain bypass detection (see D1 design decision). +_COMMAND_WRAPPERS: frozenset[str] = frozenset({"command", "nice", "time", "sudo", "nohup"}) +# Wrappers that consume a mandatory DURATION as their first non-wrapper token. +_WRAPPERS_WITH_DURATION: frozenset[str] = frozenset({"timeout"}) +# Wrappers that take a single short flag as their first non-wrapper token +# (e.g. 'stdbuf -o0', 'stdbuf -i0', 'stdbuf -e0'). +_WRAPPERS_WITH_SHORT_FLAG: frozenset[str] = frozenset({"stdbuf"}) +_GIT_ADD_CONTENT_FLAGS: frozenset[str] = frozenset( + { + "-p", + "--patch", + "-e", + "--edit", + "-i", + "--interactive", + "--pathspec-from-file", + # Content-staging flags: -A/--all stages all changes (incl. content); + # --force/--no-ignore-removal/--no-all are the no-restriction variants. + # Without these, `git add -A -- src/.../foo.yaml` is classified as + # metadata but actually stages content for indirect read via + # `git diff --staged`. + "-A", + "--all", + "--force", + "--no-ignore-removal", + "--no-all", + } +) +_GIT_STATUS_CONTENT_FLAGS: frozenset[str] = frozenset({"-v", "--verbose"}) +_GIT_DIFF_CONTENT_FLAGS: frozenset[str] = frozenset( + { + "-p", + "--patch", + "--patch-with-stat", + "--patch-with-raw", + "--binary", + "--text", + "--word-diff", + "--color-words", + } +) +_GIT_DIFF_METADATA_FLAGS: frozenset[str] = frozenset( + { + "--name-only", + "--name-status", + "--stat", + "--shortstat", + "--numstat", + "--summary", + } +) +_SHELL_SUBSTITUTION_RE = re.compile(r"\$\(|`|[<>]\(") +_SHELL_STATE_VAR_RE = re.compile(r"\$(?:_|[A-Za-z][A-Za-z0-9_]*|\{[^}]+\})") +_PROTECTED_READ_SHELL_OPS: frozenset[str] = frozenset({"&&", "||", ";", "|", "&"}) +_WC_FLAG_RE = re.compile(r"-l+|--lines$") + + +class SearchPattern(Protocol): + def search(self, string: str, /): ... + def strip_heredoc_bodies(command: str) -> str: """Strip heredoc body content, preserving the opening line and terminator. @@ -195,16 +261,46 @@ def extract_redirect_targets(tokens: list[str], cwd: str = "") -> list[str]: return targets -def command_verb(segment: list[str]) -> str: - """Return the command verb from a segment, skipping 'env' prefix.""" +def _command_start_index(segment: list[str]) -> int | None: if not segment: - return "" + return None start = 0 - if segment[0] == "env" and len(segment) > 1: - start = 1 - while start < len(segment) and (segment[start].startswith("-") or "=" in segment[start]): + # Iteratively strip 'env', command wrappers, and any wrapper-required + # argument. This handles nested forms like + # 'env FOO=BAR command git diff' and 'command env FOO=BAR git diff' + # consistently — the real command verb is the first token that is not a + # known env/wrapper prefix. 'xargs' is intentionally not a wrapper + # (see _COMMAND_WRAPPERS docstring). + while start < len(segment): + token = segment[start] + if token == "env" and start + 1 < len(segment): start += 1 - return segment[start] if start < len(segment) else "" + while start < len(segment) and ( + segment[start].startswith("-") or "=" in segment[start] + ): + start += 1 + continue + if token in _COMMAND_WRAPPERS: + start += 1 + continue + if token in _WRAPPERS_WITH_DURATION and start + 1 < len(segment): + start += 2 + continue + if ( + token in _WRAPPERS_WITH_SHORT_FLAG + and start + 1 < len(segment) + and segment[start + 1].startswith("-") + ): + start += 2 + continue + break + return start if start < len(segment) else None + + +def command_verb(segment: list[str]) -> str: + """Return the command verb from a segment, skipping 'env' prefix.""" + start = _command_start_index(segment) + return segment[start] if start is not None else "" def is_gh_command(segment: list[str]) -> bool: @@ -233,12 +329,13 @@ def extract_git_subcommand_and_flags( then returns (subcommand, remaining_tokens). Returns None if the segment is not a git command or has no subcommand. """ - if not segment: + start = _command_start_index(segment) + if start is None: return None - verb = segment[0] + verb = segment[start] if verb != "git" and not verb.endswith("/git"): return None - i = 1 + i = start + 1 while i < len(segment): token = segment[i] if token in _GIT_GLOBAL_FLAGS_WITH_VALUE: @@ -257,6 +354,110 @@ def extract_git_subcommand_and_flags( return None +def _is_allowed_wc_flag(token: str) -> bool: + """Return True when *token* is a wc flag that does not reveal file contents. + + Allows ``-l``, repeated ``-l`` (e.g. ``-ll``), and the long form ``--lines`` + only. Any value-bearing variant (``--lines=10``) or compound form + (``-lL``) is rejected because those are not used for metadata-only reads. + """ + return bool(_WC_FLAG_RE.fullmatch(token)) + + +def is_allowed_protected_path_metadata_command(segment: list[str]) -> bool: + """Return True for protected-path commands that inspect metadata or VCS state. + + Protected recipe/skill/agent paths are normally deny-by-default because most + commands that mention them are content reads. These narrow exceptions support + legitimate pipeline work on files already in scope. + """ + verb = command_verb(segment) + if verb == "git" or verb.endswith("/git"): + git_parts = extract_git_subcommand_and_flags(segment) + if git_parts is None: + return False + subcommand, flags = git_parts + if subcommand not in _PROTECTED_PATH_METADATA_GIT_SUBCOMMANDS: + return False + if subcommand == "add": + return not any( + flag in _GIT_ADD_CONTENT_FLAGS or flag.startswith("--pathspec-from-file=") + for flag in flags + ) + if subcommand == "status": + return not any(flag in _GIT_STATUS_CONTENT_FLAGS for flag in flags) + if subcommand == "diff": + if any( + flag in _GIT_DIFF_CONTENT_FLAGS + or flag.startswith("-U") + or flag.startswith("--unified") + or flag.startswith("--word-diff") + or flag.startswith("--color-words") + or flag.startswith("--patch-with-stat") + or flag.startswith("--patch-with-raw") + for flag in flags + ): + return False + return any( + flag in _GIT_DIFF_METADATA_FLAGS or flag.startswith("--stat=") for flag in flags + ) + return False + if verb == "wc" or verb.endswith("/wc"): + start = _command_start_index(segment) + if start is None: + return False + flags = [token for token in segment[start + 1 :] if token.startswith("-")] + return bool(flags) and all(_is_allowed_wc_flag(token) for token in flags) + return False + + +def _tokenize_protected_read_segments(command: str) -> list[list[str]]: + try: + lexer = shlex.shlex(command.replace("\n", " ; "), posix=True, punctuation_chars=";&|()") + lexer.whitespace_split = True + tokens = list(lexer) + except (ValueError, TypeError): + return [] + + segments: list[list[str]] = [] + current: list[str] = [] + for token in tokens: + if token in _PROTECTED_READ_SHELL_OPS: + if current: + segments.append(current) + current = [] + else: + current.append(token) + if current: + segments.append(current) + return segments + + +def command_has_blocked_protected_path_read( + command: str, protected_path_patterns: Sequence[SearchPattern] +) -> bool: + """Return True when a command reads a protected recipe/skill/agent path.""" + if not any(pattern.search(command) for pattern in protected_path_patterns): + return False + + if "<<" in command or _SHELL_SUBSTITUTION_RE.search(command): + return True + + segments = _tokenize_protected_read_segments(command) + if not segments: + return True + + if len(segments) > 1 and _SHELL_STATE_VAR_RE.search(command): + return True + + for segment in segments: + segment_text = " ".join(segment) + if any(pattern.search(segment_text) for pattern in protected_path_patterns): + if not is_allowed_protected_path_metadata_command(segment): + return True + return False + + def has_interpreter_write(command: str) -> bool: if not _INTERPRETER_RE.search(command): return False diff --git a/src/autoskillit/hooks/guards/recipe_read_guard.py b/src/autoskillit/hooks/guards/recipe_read_guard.py index ccaae59d6..5957d014a 100755 --- a/src/autoskillit/hooks/guards/recipe_read_guard.py +++ b/src/autoskillit/hooks/guards/recipe_read_guard.py @@ -13,6 +13,15 @@ import os import re import sys +from importlib import import_module +from pathlib import Path + +_HOOKS_DIR = str(Path(__file__).resolve().parent.parent) +if _HOOKS_DIR not in sys.path: + sys.path.insert(0, _HOOKS_DIR) +command_has_blocked_protected_path_read = import_module( + "_command_classification" +).command_has_blocked_protected_path_read RECIPE_READ_DENY_TRIGGER: str = "must not read recipe/skill/agent files directly" @@ -43,13 +52,12 @@ def main() -> None: if tool == "run_cmd" or tool_name == "Bash": cmd_key = "command" if tool_name == "Bash" else "cmd" cmd: str = tool_input.get(cmd_key, "") - for pattern in _CMD_PATH_PATTERNS: - if pattern.search(cmd): - _deny( - f"run_cmd/Bash {RECIPE_READ_DENY_TRIGGER}. " - "Use load_recipe to recall step definitions or the Skill tool " - "for skill instructions." - ) + if command_has_blocked_protected_path_read(cmd, _CMD_PATH_PATTERNS): + _deny( + f"run_cmd/Bash {RECIPE_READ_DENY_TRIGGER}. " + "Use load_recipe to recall step definitions or the Skill tool " + "for skill instructions." + ) if tool == "run_python": callable_name: str = tool_input.get("callable", "") diff --git a/src/autoskillit/server/_guards.py b/src/autoskillit/server/_guards.py index e8facbcd9..1dfcaaa2e 100644 --- a/src/autoskillit/server/_guards.py +++ b/src/autoskillit/server/_guards.py @@ -19,6 +19,7 @@ is_path_like_token, session_type, ) +from autoskillit.hooks import command_has_blocked_protected_path_read from autoskillit.pipeline import gate_error_result, headless_error_result if TYPE_CHECKING: @@ -157,13 +158,12 @@ def _check_recipe_read_prohibition( 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 command_has_blocked_protected_path_read(cmd, _RECIPE_READ_CMD_PATTERNS): + 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( diff --git a/tests/infra/test_recipe_read_guard.py b/tests/infra/test_recipe_read_guard.py index a8c38f41a..647077fa8 100644 --- a/tests/infra/test_recipe_read_guard.py +++ b/tests/infra/test_recipe_read_guard.py @@ -97,6 +97,102 @@ def test_allows_non_recipe_yaml(self): ) assert not out.strip() + @pytest.mark.parametrize( + "cmd", + [ + "git add -- src/autoskillit/recipes/remediation.yaml", + "git diff --stat -- src/autoskillit/recipes/remediation.yaml", + "git diff --name-only -- src/autoskillit/recipes/remediation.yaml", + "git status -- src/autoskillit/recipes/remediation.yaml", + "wc -l src/autoskillit/recipes/remediation.yaml", + ], + ) + def test_allows_recipe_path_vcs_and_metadata_commands(self, cmd): + out = _run_guard( + { + "tool_name": "mcp__mcp-autoskillit__run_cmd", + "tool_input": {"cmd": cmd}, + } + ) + assert not out.strip() + + def test_allows_chain_of_safe_recipe_metadata_commands(self): + out = _run_guard( + { + "tool_name": "mcp__mcp-autoskillit__run_cmd", + "tool_input": { + "cmd": ( + "git add -- src/autoskillit/recipes/remediation.yaml " + "&& git status --short -- src/autoskillit/recipes/remediation.yaml" + ) + }, + } + ) + assert not out.strip() + + def test_denies_chained_recipe_read_after_allowed_git_add(self): + out = _run_guard( + { + "tool_name": "mcp__mcp-autoskillit__run_cmd", + "tool_input": { + "cmd": ( + "git add -- src/autoskillit/recipes/remediation.yaml " + "&& cat src/autoskillit/recipes/remediation.yaml" + ) + }, + } + ) + assert _is_denied(out) + + @pytest.mark.parametrize( + "cmd", + [ + "git diff -- src/autoskillit/recipes/remediation.yaml", + "git status -v -- src/autoskillit/recipes/remediation.yaml", + "git add -p -- src/autoskillit/recipes/remediation.yaml", + "git add --pathspec-from-file=src/autoskillit/recipes/remediation.yaml", + "git add --pathspec-from-file src/autoskillit/recipes/remediation.yaml", + "git diff --stat --patch-with-stat -- src/autoskillit/recipes/remediation.yaml", + "git diff --stat --patch-with-raw -- src/autoskillit/recipes/remediation.yaml", + "git diff --stat --binary -- src/autoskillit/recipes/remediation.yaml", + "git diff --check -- src/autoskillit/recipes/remediation.yaml", + ("python3 <<'PY'\nprint(open('src/autoskillit/recipes/remediation.yaml').read())\nPY"), + ( + "git add -- src/autoskillit/recipes/remediation.yaml\n" + "cat src/autoskillit/recipes/remediation.yaml" + ), + ( + "git add -- src/autoskillit/recipes/remediation.yaml " + "& cat src/autoskillit/recipes/remediation.yaml" + ), + 'git add -- "$(cat src/autoskillit/recipes/remediation.yaml)"', + "git add -- src/autoskillit/recipes/remediation.yaml && cat $_", + ( + "git add -- src/autoskillit/recipes/remediation.yaml" + "&&cat src/autoskillit/recipes/remediation.yaml" + ), + # Input redirection: `<` is not in punctuation_chars; path is in + # the segment text, pattern matches, verb `cat` is not git/wc. + "cat < src/autoskillit/recipes/remediation.yaml", + # Content-reading git subcommands not in the metadata allowlist. + "git show HEAD:src/autoskillit/recipes/remediation.yaml", + "git log -p -- src/autoskillit/recipes/remediation.yaml", + "git blame src/autoskillit/recipes/remediation.yaml", + "git grep pattern -- src/autoskillit/recipes/remediation.yaml", + # -A/--all/--force stage content even when restricted to a path. + "git add -A -- src/autoskillit/recipes/remediation.yaml", + "git add --all -- src/autoskillit/recipes/remediation.yaml", + ], + ) + def test_denies_protected_path_content_bypasses(self, cmd): + out = _run_guard( + { + "tool_name": "mcp__mcp-autoskillit__run_cmd", + "tool_input": {"cmd": cmd}, + } + ) + assert _is_denied(out) + class TestRunPythonBlocking: """Verify run_python calls to recipe module are blocked.""" @@ -223,3 +319,14 @@ def test_allows_bash_unrelated(self): } ) assert not out.strip() + + def test_allows_bash_git_diff_recipe_path(self): + out = _run_guard( + { + "tool_name": "Bash", + "tool_input": { + "command": "git diff --stat -- .autoskillit/recipes/remediation.yaml" + }, + } + ) + assert not out.strip() diff --git a/tests/server/test_tools_run_cmd_invariants.py b/tests/server/test_tools_run_cmd_invariants.py index 93b25fadf..e6523747d 100644 --- a/tests/server/test_tools_run_cmd_invariants.py +++ b/tests/server/test_tools_run_cmd_invariants.py @@ -69,6 +69,92 @@ async def test_allows_non_recipe_python_file(self, tool_ctx_kitchen_open): result = json.loads(await run_cmd(cmd="cat src/autoskillit/core/__init__.py", cwd="/tmp")) assert result["success"] is True + @pytest.mark.anyio + @pytest.mark.parametrize( + "cmd", + [ + "git add -- src/autoskillit/recipes/remediation.yaml", + "git diff --stat -- src/autoskillit/recipes/remediation.yaml", + "git diff --name-only -- src/autoskillit/recipes/remediation.yaml", + "git status -- src/autoskillit/recipes/remediation.yaml", + "wc -l src/autoskillit/recipes/remediation.yaml", + ], + ) + async def test_allows_recipe_path_vcs_and_metadata_commands(self, tool_ctx_kitchen_open, cmd): + tool_ctx_kitchen_open.runner.push(_make_result(0, "", "")) + result = json.loads(await run_cmd(cmd=cmd, cwd="/tmp")) + assert result["success"] is True + assert "subtype" not in result + + @pytest.mark.anyio + async def test_allows_chain_of_safe_recipe_metadata_commands(self, tool_ctx_kitchen_open): + tool_ctx_kitchen_open.runner.push(_make_result(0, "", "")) + result = json.loads( + await run_cmd( + cmd=( + "git add -- src/autoskillit/recipes/remediation.yaml " + "&& git status --short -- src/autoskillit/recipes/remediation.yaml" + ), + cwd="/tmp", + ) + ) + assert result["success"] is True + + @pytest.mark.anyio + async def test_denies_chained_recipe_read_after_allowed_git_add(self, tool_ctx_kitchen_open): + result = json.loads( + await run_cmd( + cmd=( + "git add -- src/autoskillit/recipes/remediation.yaml " + "&& cat src/autoskillit/recipes/remediation.yaml" + ), + cwd="/tmp", + ) + ) + assert result["success"] is False + assert result["subtype"] == "gate_error" + assert RECIPE_READ_DENY_TRIGGER in result["result"] + + @pytest.mark.anyio + @pytest.mark.parametrize( + "cmd", + [ + "git diff -- src/autoskillit/recipes/remediation.yaml", + "git status -v -- src/autoskillit/recipes/remediation.yaml", + "git add -p -- src/autoskillit/recipes/remediation.yaml", + "git add --pathspec-from-file=src/autoskillit/recipes/remediation.yaml", + "git add --pathspec-from-file src/autoskillit/recipes/remediation.yaml", + "git diff --stat --patch-with-stat -- src/autoskillit/recipes/remediation.yaml", + "git diff --stat --patch-with-raw -- src/autoskillit/recipes/remediation.yaml", + "git diff --stat --binary -- src/autoskillit/recipes/remediation.yaml", + "git diff --check -- src/autoskillit/recipes/remediation.yaml", + ("python3 <<'PY'\nprint(open('src/autoskillit/recipes/remediation.yaml').read())\nPY"), + ( + "git add -- src/autoskillit/recipes/remediation.yaml\n" + "cat src/autoskillit/recipes/remediation.yaml" + ), + ( + "git add -- src/autoskillit/recipes/remediation.yaml " + "& cat src/autoskillit/recipes/remediation.yaml" + ), + # Command substitution $(...) bypass — exercises the _SHELL_SUBSTITUTION_RE + # branch in command_has_blocked_protected_path_read. + "git add -- $(cat src/autoskillit/recipes/remediation.yaml)", + # State-var-via-history $_ bypass — exercises the + # _SHELL_STATE_VAR_RE branch gated on len(segments) > 1. + "git add -- src/autoskillit/recipes/remediation.yaml && cat $_", + ( + "git add -- src/autoskillit/recipes/remediation.yaml" + "&&cat src/autoskillit/recipes/remediation.yaml" + ), + ], + ) + async def test_denies_protected_path_content_bypasses(self, tool_ctx_kitchen_open, cmd): + result = json.loads(await run_cmd(cmd=cmd, 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_allows_in_non_headless(self, tool_ctx_kitchen_open, monkeypatch): monkeypatch.delenv("AUTOSKILLIT_HEADLESS", raising=False)