diff --git a/docs/safety/hooks.md b/docs/safety/hooks.md index 925a9cf0b..839276dc2 100644 --- a/docs/safety/hooks.md +++ b/docs/safety/hooks.md @@ -510,17 +510,23 @@ relative `--input` files against the execution cwd only; a missing or relative execution cwd degrades cleanly to "unresolved" rather than denying every command that carries one. -Every deny routes through one of five machine-readable triggers +Every deny routes through one of seven machine-readable triggers (`DenyTrigger` in `github_mutation_guard.py`), each with its own reason text: | Trigger | Meaning | |---------|---------| | `field_confusion` | The payload carries both the Bash and run_cmd command fields, or a stray `cwd` inside a Bash `tool_input` — which text executes is ambiguous. | | `malformed_command` | The command field is missing or not a string. | -| `unresolved_mutation` | Mutation cardinality or target cannot be statically proven safe (dynamic values, unresolved `--input`, repeatable/dispatch constructs reaching a possible GitHub exec). | +| `malformed_cwd` | The command execution cwd is non-string or relative, so path-sensitive inputs cannot be resolved safely. | +| `unresolved_mutation` | Mutation cardinality or target cannot be statically proven safe. The denial includes only a bounded static classifier code; internal prose, command text, and paths remain private. | +| `classifier_internal_error` | The classifier failed unexpectedly. This is distinct from expected uncertainty and exposes no exception detail. | | `multiple_mutations` | The command issues more than one GitHub mutation request. | | `review_mutation` | The command is a raw pull-request review publication. | +The guard is registered globally with `session_scope="any"` and permits only exactly one +proven non-review mutation. Both expected classifier uncertainty and unexpected internal +failure remain fail-closed. + ## Drift detection `cli/_doctor.py:_check_hook_registry_drift` calls `generate_hooks_json()` and diff --git a/src/autoskillit/hooks/_command_classification.py b/src/autoskillit/hooks/_command_classification.py index d1cae2db8..a8b9471b5 100644 --- a/src/autoskillit/hooks/_command_classification.py +++ b/src/autoskillit/hooks/_command_classification.py @@ -165,6 +165,7 @@ _REDIRECT_TOKEN_RE = re.compile(r"^(\d*)>{1,2}(.+)$") _REDIRECT_OP_ONLY_RE = re.compile(r"^(\d*)>{1,2}$") _FD_REDIRECT_RE = re.compile(r"^\d*>{1,2}&") +_FD_DUPLICATION_RE = re.compile(r"^\d*>&\d+$") _TRAILING_SHELL_CLOSERS = frozenset({")", "`", "}", "'", '"', ";", "&", "|"}) _SHELL_VAR_RE = re.compile(r"\$\{[A-Za-z_]|\$[A-Za-z_]") @@ -285,16 +286,91 @@ def _normalize_newlines_for_tokenize(command: str) -> str: return "".join(result) -def tokenize_command_segments(command: str) -> list[list[str]]: - """Split a shell command into segments of (verb, args...) token lists. +@dataclass(frozen=True, slots=True) +class _CommandSegment: + tokens: list[str] + redirect_syntax: list[bool] - Each segment is one logical command, separated by shell operators. - Returns [] on shlex parse error (unclosed quotes). - """ + +def _mark_unquoted_output_redirects(command: str) -> tuple[str, dict[str, str]]: + """Replace recognized redirect operators with shlex-stable placeholders.""" + rendered: list[str] = [] + redirects: dict[str, str] = {} + in_single = False + in_double = False + i = 0 + while i < len(command): + char = command[i] + if char == "\\" and not in_single and i + 1 < len(command): + rendered.extend((char, command[i + 1])) + i += 2 + continue + if char == "'" and not in_double: + in_single = not in_single + rendered.append(char) + i += 1 + continue + if char == '"' and not in_single: + in_double = not in_double + rendered.append(char) + i += 1 + continue + if in_single or in_double: + rendered.append(char) + i += 1 + continue + + start = i + if char.isdecimal() and (i == 0 or command[i - 1].isspace() or command[i - 1] in ";&|("): + while i < len(command) and command[i].isdecimal(): + i += 1 + if i >= len(command) or command[i] != ">": + rendered.append(command[start]) + i = start + 1 + continue + elif char != ">": + rendered.append(char) + i += 1 + continue + + operator_start = start + operator_end = i + 1 + if operator_end < len(command) and command[operator_end] == ">": + operator_end += 1 + if operator_end < len(command) and command[operator_end] == "(": + rendered.append(command[start]) + i = start + 1 + continue + if ( + operator_end < len(command) + and command[operator_end] == "&" + and (not command[operator_start:i] or command[operator_start:i].isdecimal()) + ): + fd_end = operator_end + 1 + while fd_end < len(command) and command[fd_end].isdecimal(): + fd_end += 1 + if fd_end == operator_end + 1: + rendered.append(command[start]) + i = start + 1 + continue + operator_end = fd_end + + marker = f"__AUTOSKILLIT_REDIRECT_{len(redirects)}__" + redirects[marker] = command[operator_start:operator_end] + rendered.extend((" ", marker, " ")) + i = operator_end + return ("".join(rendered), redirects) + + +def _tokenize_command_segments_with_redirects(command: str) -> list[_CommandSegment]: + """Tokenize commands while retaining which redirect-shaped tokens are syntax.""" try: stripped = _HEREDOC_MARKER_RE.sub(r"\2", strip_heredoc_bodies(command)) + marked, redirects = _mark_unquoted_output_redirects( + _normalize_newlines_for_tokenize(stripped) + ) lexer = shlex.shlex( - _normalize_newlines_for_tokenize(stripped), + marked, posix=True, punctuation_chars=";&|", ) @@ -303,89 +379,111 @@ def tokenize_command_segments(command: str) -> list[list[str]]: except (ValueError, TypeError): return [] - segments: list[list[str]] = [] - current: list[str] = [] + segments: list[_CommandSegment] = [] + current_tokens: list[str] = [] + current_redirect_syntax: list[bool] = [] for token in tokens: if token in _SHELL_OPERATORS: - if current: - segments.append(current) - current = [] + if current_tokens: + segments.append(_CommandSegment(current_tokens, current_redirect_syntax)) + current_tokens = [] + current_redirect_syntax = [] else: - current.append(token) - if current: - segments.append(current) + redirect = redirects.get(token) + current_tokens.append(redirect if redirect is not None else token) + current_redirect_syntax.append(redirect is not None) + if current_tokens: + segments.append(_CommandSegment(current_tokens, current_redirect_syntax)) return segments -def extract_redirect_targets(tokens: list[str], cwd: str = "") -> list[str]: - """Extract redirect target paths from shlex-tokenized command tokens. - - Returns resolved paths including pseudo-devices — caller filters. - Relative paths are resolved against cwd when provided. +def tokenize_command_segments(command: str) -> list[list[str]]: + """Split a shell command into segments of (verb, args...) token lists.""" + return [segment.tokens for segment in _tokenize_command_segments_with_redirects(command)] - Handles three redirect forms at depth 0 only: - - Separate: ['>', '/path'] or ['>>', '/path'] - - Split: ['2>', '/path'] (operator-only token + next token) - - Merged: ['2>/path'] or ['2>>/path'] - Tracks subshell nesting via '(' and ')' — both standalone and fused - with adjacent text (shlex merges '(' with following chars in POSIX mode). - """ +def _partition_output_redirects( + tokens: Sequence[str], + *, + cwd: str, + redirect_syntax: Sequence[bool] | None = None, +) -> tuple[list[str], list[str], int]: + """Separate depth-zero output control from executable argv.""" + syntax = redirect_syntax if redirect_syntax is not None else [True] * len(tokens) + executable: list[str] = [] targets: list[str] = [] + file_redirect_count = 0 depth = 0 i = 0 while i < len(tokens): - tok = tokens[i] - if tok == "(" or (tok.startswith("(") and len(tok) > 1): + token = tokens[i] + if token == "(" or (token.startswith("(") and len(token) > 1): depth += 1 - if tok.endswith(")") and len(tok) > 1: + if token.endswith(")") and len(token) > 1: depth -= 1 + executable.append(token) i += 1 continue - if tok == ")": + if token == ")": if depth > 0: depth -= 1 + executable.append(token) i += 1 continue - if tok.endswith(")") and len(tok) > 1: + if token.endswith(")") and len(token) > 1: if depth > 0: depth -= 1 + executable.append(token) i += 1 continue - if depth > 0: + if depth > 0 or not syntax[i]: + executable.append(token) i += 1 continue - if tok in (">", ">>"): - if i + 1 < len(tokens): - path = tokens[i + 1] - while path and path[-1] in _TRAILING_SHELL_CLOSERS: - path = path[:-1] - resolved = resolve_write_target(path, cwd) - if resolved is not None: - targets.append(resolved) - i += 2 - continue - elif _REDIRECT_OP_ONLY_RE.match(tok): - if i + 1 < len(tokens): - path = tokens[i + 1] - while path and path[-1] in _TRAILING_SHELL_CLOSERS: - path = path[:-1] - resolved = resolve_write_target(path, cwd) - if resolved is not None: - targets.append(resolved) + if _FD_DUPLICATION_RE.fullmatch(token): + i += 1 + continue + + target: str | None = None + if _REDIRECT_OP_ONLY_RE.fullmatch(token): + file_redirect_count += 1 + if i + 1 < len(tokens) and not ( + syntax[i + 1] + and ( + _REDIRECT_OP_ONLY_RE.fullmatch(tokens[i + 1]) + or _FD_DUPLICATION_RE.fullmatch(tokens[i + 1]) + ) + ): + target = tokens[i + 1] i += 2 - continue + else: + i += 1 else: - m = _REDIRECT_TOKEN_RE.match(tok) - if m: - path = m.group(2) - while path and path[-1] in _TRAILING_SHELL_CLOSERS: - path = path[:-1] - resolved = resolve_write_target(path, cwd) - if resolved is not None: - targets.append(resolved) - i += 1 - return targets + match = _REDIRECT_TOKEN_RE.fullmatch(token) + if match is None: + executable.append(token) + i += 1 + continue + file_redirect_count += 1 + target = match.group(2) + i += 1 + + if target is not None: + while target and target[-1] in _TRAILING_SHELL_CLOSERS: + target = target[:-1] + resolved = resolve_write_target(target, cwd) + if resolved is not None: + targets.append(resolved) + return (executable, targets, file_redirect_count) + + +def extract_redirect_targets(tokens: list[str], cwd: str = "") -> list[str]: + """Extract resolved redirect target paths from already-tokenized input. + + Returns resolved paths including pseudo-devices — caller filters. + Relative paths are resolved against cwd when provided. + """ + return _partition_output_redirects(tokens, cwd=cwd)[1] def _is_posix_assignment(token: str) -> bool: @@ -798,6 +896,17 @@ def extract_shell_command_payloads(command: str) -> list[str]: return payloads +def _segment_evaluates_shell_payload(tokens: list[str], payload: str) -> bool: + """Return whether *tokens* structurally evaluate *payload* as shell text.""" + verb, args = command_verb_and_args(tokens) + if _is_shell_interpreter(verb) and args and args[0] == "-c" and len(args) >= 2: + return args[1] == payload + if verb == "eval" and args: + return " ".join(args) == payload + rendered = " ".join(tokens) + return f"$({payload})" in rendered or f"`{payload}`" in rendered + + def tokenize_shell_payload_segments(command: str) -> list[list[str]] | None: """Return tokenized segments for every evaluated shell payload in *command*. @@ -1129,6 +1238,7 @@ class GitHubMutationAnalysis: mutations: tuple[GitHubMutationRecord, ...] request_count: int | None review_comment_count: int | None + reason_code: str reason: str @@ -1241,11 +1351,14 @@ def _none_github_analysis() -> GitHubMutationAnalysis: mutations=(), request_count=0, review_comment_count=None, + reason_code="", reason="", ) def _unresolved_github_analysis( + *, + reason_code: str, reason: str, mutations: Sequence[GitHubMutationRecord] = (), ) -> GitHubMutationAnalysis: @@ -1254,6 +1367,7 @@ def _unresolved_github_analysis( mutations=tuple(mutations), request_count=None, review_comment_count=None, + reason_code=reason_code, reason=reason, ) @@ -1301,24 +1415,36 @@ def _load_literal_github_input( value: str, *, cwd: str, -) -> tuple[dict[str, Any] | None, str]: +) -> tuple[dict[str, Any] | None, str, str]: if value == "-": - return (None, "GitHub --input stdin is unresolved") + return (None, "unsafe_input_provenance", "GitHub --input stdin is unresolved") if not value or _is_dynamic_shell_value(value): - return (None, "GitHub --input path is dynamic") + return (None, "dynamic_target", "GitHub --input path is dynamic") if os.path.isabs(value): path = os.path.normpath(value) else: if not cwd or not os.path.isabs(cwd): - return (None, "relative GitHub --input requires an absolute cwd") + return ( + None, + "cwd_unresolved", + "relative GitHub --input requires an absolute cwd", + ) path = os.path.normpath(os.path.join(cwd, value)) try: before = os.lstat(path) if stat.S_ISLNK(before.st_mode) or not stat.S_ISREG(before.st_mode): - return (None, "GitHub --input must be a regular non-symlink file") + return ( + None, + "input_inspection_failed", + "GitHub --input must be a regular non-symlink file", + ) if before.st_size > _GITHUB_INPUT_LIMIT: - return (None, "GitHub --input exceeds the inspection limit") + return ( + None, + "input_inspection_failed", + "GitHub --input exceeds the inspection limit", + ) flags = os.O_RDONLY if hasattr(os, "O_NOFOLLOW"): flags |= os.O_NOFOLLOW @@ -1329,7 +1455,11 @@ def _load_literal_github_input( after.st_dev, after.st_ino, ): - return (None, "GitHub --input file identity changed") + return ( + None, + "input_inspection_failed", + "GitHub --input file identity changed", + ) chunks: list[bytes] = [] remaining = _GITHUB_INPUT_LIMIT + 1 while remaining: @@ -1342,10 +1472,18 @@ def _load_literal_github_input( finally: os.close(fd) if len(raw) > _GITHUB_INPUT_LIMIT: - return (None, "GitHub --input exceeds the inspection limit") - return (_json_object_without_duplicate_keys(raw), "") + return ( + None, + "input_inspection_failed", + "GitHub --input exceeds the inspection limit", + ) + return (_json_object_without_duplicate_keys(raw), "", "") except (OSError, UnicodeError, json.JSONDecodeError, ValueError) as exc: - return (None, f"GitHub --input is not safely inspectable: {exc}") + return ( + None, + "input_inspection_failed", + f"GitHub --input is not safely inspectable: {exc}", + ) _INPUT_SAFE_PRIOR_COMMANDS: frozenset[str] = frozenset( @@ -1353,22 +1491,24 @@ def _load_literal_github_input( ) -def _segment_is_safe_before_literal_input(segment: Sequence[str], *, cwd: str) -> bool: +def _segment_is_safe_before_literal_input(segment: Sequence[str]) -> bool: """Return whether *segment* is proven unable to rewrite a later input file.""" - if extract_redirect_targets(list(segment), cwd): - return False verb, _ = command_verb_and_args(list(segment)) executable = _normalize_executable(verb) return executable in _INPUT_SAFE_PRIOR_COMMANDS -def _comment_count_from_payload(payload: dict[str, Any]) -> tuple[int | None, str]: +def _comment_count_from_payload(payload: dict[str, Any]) -> tuple[int | None, str, str]: if "comments" not in payload: - return (None, "") + return (None, "", "") comments = payload["comments"] if not isinstance(comments, list): - return (None, "GitHub review comments must be a JSON array") - return (len(comments), "") + return ( + None, + "invalid_input_payload", + "GitHub review comments must be a JSON array", + ) + return (len(comments), "", "") def _flag_value( @@ -1395,7 +1535,9 @@ def _analyze_gh_api( *, cwd: str, input_context_safe: bool, -) -> tuple[GitHubMutationRecord | None, str]: + resolved_redirect_targets: Sequence[str], + file_redirect_count: int, +) -> tuple[GitHubMutationRecord | None, str, str]: method: str | None = None route: str | None = None input_value: str | None = None @@ -1416,7 +1558,7 @@ def _analyze_gh_api( value, next_i, matched = _flag_value(args, i, long_name="--method", short_name="-X") if matched or token in {"--method", "-X"}: if not matched or value is None: - return (None, "GitHub API method is missing") + return (None, "missing_required_value", "GitHub API method is missing") method = value.upper() i = next_i continue @@ -1424,7 +1566,7 @@ def _analyze_gh_api( value, next_i, matched = _flag_value(args, i, long_name="--input") if matched or token == "--input": if not matched or value is None: - return (None, "GitHub --input path is missing") + return (None, "missing_required_value", "GitHub --input path is missing") input_value = value has_body_fields = True i = next_i @@ -1440,7 +1582,7 @@ def _analyze_gh_api( ) if matched or token in {long_name, short_name}: if not matched or value is None: - return (None, f"{long_name} value is missing") + return (None, "missing_required_value", f"{long_name} value is missing") field_values.append(value) has_body_fields = True i = next_i @@ -1455,7 +1597,7 @@ def _analyze_gh_api( continue if token in {"-H", "--header", "--hostname", "--cache"}: if i + 1 >= len(args): - return (None, f"{token} value is missing") + return (None, "missing_required_value", f"{token} value is missing") i += 2 continue if token.startswith(("--header=", "--hostname=", "--cache=")): @@ -1468,37 +1610,83 @@ def _analyze_gh_api( route = token i += 1 continue - return (None, "multiple GitHub API routes are unresolved") + return ( + None, + "request_cardinality_unresolved", + "multiple GitHub API routes are unresolved", + ) if route is None: if method is not None or has_body_fields: - return (None, "GitHub API route is missing") - return (None, "") + return (None, "missing_required_value", "GitHub API route is missing") + return (None, "", "") if _is_dynamic_shell_value(route): - return (None, "GitHub API route is dynamic") + return (None, "dynamic_target", "GitHub API route is dynamic") if method is not None and _is_dynamic_shell_value(method): - return (None, "GitHub API method is dynamic") + return (None, "dynamic_target", "GitHub API method is dynamic") payload: dict[str, Any] = {} query_from_literal_input = input_value is not None if input_value is not None: if not input_context_safe: - return (None, "a prior command may rewrite the inspected GitHub --input file") - loaded, reason = _load_literal_github_input(input_value, cwd=cwd) + return ( + None, + "unsafe_input_provenance", + "a prior command may rewrite the inspected GitHub --input file", + ) + loaded, reason_code, reason = _load_literal_github_input(input_value, cwd=cwd) if loaded is None: - return (None, reason) + return (None, reason_code, reason) + input_path = ( + os.path.normpath(input_value) + if os.path.isabs(input_value) + else os.path.normpath(os.path.join(cwd, input_value)) + ) + if file_redirect_count != len(resolved_redirect_targets): + return ( + None, + "unsafe_input_provenance", + "an output redirect may alias the inspected GitHub --input file", + ) + redirect_aliases_input = False + for target in resolved_redirect_targets: + if os.path.realpath(target) == os.path.realpath(input_path): + redirect_aliases_input = True + break + if not os.path.exists(target): + continue + try: + redirect_aliases_input = os.path.samefile(target, input_path) + except OSError: + return ( + None, + "unsafe_input_provenance", + "an output redirect alias could not be inspected safely", + ) + if redirect_aliases_input: + break + if redirect_aliases_input: + return ( + None, + "unsafe_input_provenance", + "an output redirect aliases the inspected GitHub --input file", + ) payload = loaded effective_method = method or ("POST" if has_body_fields else "GET") normalized_route = _normalize_github_route(route) if effective_method not in _GITHUB_WRITE_METHODS: - return (None, "") + return (None, "", "") if paginate: - return (None, "mutation request count is indeterminate with --paginate") + return ( + None, + "request_cardinality_unresolved", + "mutation request count is indeterminate with --paginate", + ) - comment_count, reason = _comment_count_from_payload(payload) + comment_count, reason_code, reason = _comment_count_from_payload(payload) if reason: - return (None, reason) + return (None, reason_code, reason) if graphql: query = payload.get("query") @@ -1511,9 +1699,9 @@ def _analyze_gh_api( if not isinstance(query, str) or ( not query_from_literal_input and _is_dynamic_shell_value(query) ): - return (None, "GraphQL mutation document is unresolved") + return (None, "dynamic_target", "GraphQL mutation document is unresolved") if not re.search(r"\bmutation\b", query): - return (None, "") + return (None, "", "") kind = ( GitHubMutationKind.GRAPHQL_REVIEW if any( @@ -1533,6 +1721,7 @@ def _analyze_gh_api( review_comment_count=comment_count, ), "", + "", ) @@ -1598,7 +1787,7 @@ def _is_static_issue_edit_target(value: str) -> bool: ) -def _issue_edit_request_count(args: Sequence[str]) -> tuple[int | None, str]: +def _issue_edit_request_count(args: Sequence[str]) -> tuple[int | None, str, str]: targets = 0 options_ended = False i = 0 @@ -1614,7 +1803,11 @@ def _issue_edit_request_count(args: Sequence[str]) -> tuple[int | None, str]: or token in _GH_ISSUE_EDIT_SHORT_VALUE_FLAGS ): if i + 1 >= len(args): - return (None, f"gh issue edit flag {token} is missing a value") + return ( + None, + "missing_required_value", + f"gh issue edit flag {token} is missing a value", + ) i += 2 continue if any(token.startswith(f"{flag}=") for flag in _GH_ISSUE_EDIT_LONG_VALUE_FLAGS): @@ -1627,15 +1820,19 @@ def _issue_edit_request_count(args: Sequence[str]) -> tuple[int | None, str]: i += 1 continue if token.startswith("-"): - return (None, f"gh issue edit flag {token} is unresolved") + return ( + None, + "unsupported_grammar", + f"gh issue edit flag {token} is unresolved", + ) if not _is_static_issue_edit_target(token): - return (None, "gh issue edit target is unresolved") + return (None, "dynamic_target", "gh issue edit target is unresolved") targets += 1 i += 1 if targets == 0: - return (None, "gh issue edit target is missing") - return (targets, "") + return (None, "missing_required_value", "gh issue edit target is missing") + return (targets, "", "") _GH_MUTATION_SUBCOMMANDS: dict[str, frozenset[str]] = { @@ -1683,13 +1880,15 @@ def _analyze_gh_segment( *, cwd: str, input_context_safe: bool, -) -> tuple[GitHubMutationRecord | None, str]: + resolved_redirect_targets: Sequence[str], + file_redirect_count: int, +) -> tuple[GitHubMutationRecord | None, str, str]: if not args: - return (None, "") + return (None, "", "") if _gh_args_have_bare_help_flag(args[1:]): - return (None, "") + return (None, "", "") if args[:2] == ["pr", "create"]: - return (None, "") + return (None, "", "") if args[:2] == ["pr", "review"]: return ( GitHubMutationRecord( @@ -1700,11 +1899,12 @@ def _analyze_gh_segment( review_comment_count=None, ), "", + "", ) if args[:2] == ["issue", "edit"]: - request_count, reason = _issue_edit_request_count(args[2:]) + request_count, reason_code, reason = _issue_edit_request_count(args[2:]) if request_count is None: - return (None, reason) + return (None, reason_code, reason) return ( GitHubMutationRecord( method="POST", @@ -1714,15 +1914,20 @@ def _analyze_gh_segment( review_comment_count=None, ), "", + "", ) noun = args[0] mutation_verbs = _GH_MUTATION_SUBCOMMANDS.get(noun) if mutation_verbs is not None and len(args) >= 2: verb = args[1] if verb in _GH_READ_ONLY_SUBCOMMANDS.get(noun, frozenset()): - return (None, "") + return (None, "", "") if verb not in mutation_verbs: - return (None, f"gh {noun} {verb} mutation classification is unresolved") + return ( + None, + "unsupported_grammar", + f"gh {noun} {verb} mutation classification is unresolved", + ) return ( GitHubMutationRecord( method="POST", @@ -1732,15 +1937,22 @@ def _analyze_gh_segment( review_comment_count=None, ), "", + "", ) if args[0] != "api": - return (None, "") - return _analyze_gh_api(args[1:], cwd=cwd, input_context_safe=input_context_safe) + return (None, "", "") + return _analyze_gh_api( + args[1:], + cwd=cwd, + input_context_safe=input_context_safe, + resolved_redirect_targets=resolved_redirect_targets, + file_redirect_count=file_redirect_count, + ) def _analyze_curl_segment( args: Sequence[str], -) -> tuple[list[GitHubMutationRecord], str]: +) -> tuple[list[GitHubMutationRecord], str, str]: method: str | None = None has_data = False force_get = False @@ -1761,14 +1973,14 @@ def _analyze_curl_segment( value, next_i, matched = _flag_value(args, i, long_name="--request", short_name="-X") if matched or token in {"--request", "-X"}: if not matched or value is None: - return ([], "curl method is missing") + return ([], "missing_required_value", "curl method is missing") method = value.upper() i = next_i continue value, next_i, matched = _flag_value(args, i, long_name="--url") if matched or token == "--url": if not matched or value is None: - return ([], "curl URL is missing") + return ([], "missing_required_value", "curl URL is missing") urls.append(value) i = next_i continue @@ -1790,7 +2002,7 @@ def _analyze_curl_segment( ) if matched or token == long_name or (short_name is not None and token == short_name): if not matched or value is None: - return ([], f"{token} value is missing") + return ([], "missing_required_value", f"{token} value is missing") has_data = True i = next_i matched_value_flag = True @@ -1806,7 +2018,7 @@ def _analyze_curl_segment( ) if matched or token == long_name or token == short_name: if not matched or value is None: - return ([], f"{token} value is missing") + return ([], "missing_required_value", f"{token} value is missing") i = next_i matched_value_flag = True break @@ -1819,21 +2031,25 @@ def _analyze_curl_segment( i += 1 if method is not None and _is_dynamic_shell_value(method): - return ([], "curl method is dynamic") + return ([], "dynamic_target", "curl method is dynamic") if any(_is_dynamic_shell_value(url) for url in urls): - return ([], "curl URL is dynamic") + return ([], "dynamic_target", "curl URL is dynamic") github_urls = [] for url in urls: hostname = urlsplit(url).hostname if hostname is not None and hostname.lower() in {"api.github.com", "github.com"}: github_urls.append(url) if not github_urls: - return ([], "") + return ([], "", "") effective_method = method or ("GET" if force_get else ("POST" if has_data else "GET")) if effective_method not in _GITHUB_WRITE_METHODS: - return ([], "") + return ([], "", "") if saw_next or len(github_urls) != 1 or len(urls) != 1: - return ([], "curl mutation request count is indeterminate") + return ( + [], + "request_cardinality_unresolved", + "curl mutation request count is indeterminate", + ) route = urlsplit(github_urls[0]).path or "/" return ( [ @@ -1846,6 +2062,7 @@ def _analyze_curl_segment( ) ], "", + "", ) @@ -1873,19 +2090,23 @@ def _analyze_github_segment( *, cwd: str, input_context_safe: bool = True, -) -> tuple[list[GitHubMutationRecord], str]: + resolved_redirect_targets: Sequence[str] = (), + file_redirect_count: int = 0, +) -> tuple[list[GitHubMutationRecord], str, str]: verb, args = command_verb_and_args(list(segment)) executable = _normalize_executable(verb) if executable == "gh": - record, reason = _analyze_gh_segment( + record, reason_code, reason = _analyze_gh_segment( args, cwd=_segment_cwd(segment, cwd), input_context_safe=input_context_safe, + resolved_redirect_targets=resolved_redirect_targets, + file_redirect_count=file_redirect_count, ) - return (([record] if record is not None else []), reason) + return (([record] if record is not None else []), reason_code, reason) if executable == "curl": return _analyze_curl_segment(args) - return ([], "") + return ([], "", "") def analyze_github_mutations( @@ -1898,85 +2119,210 @@ def analyze_github_mutations( return _none_github_analysis() records: list[GitHubMutationRecord] = [] - reasons: list[str] = [] - queue: list[tuple[str, str, int]] = [(command, cwd, 0)] - argv_payloads: list[tuple[list[str], str]] = [] + reasons: list[tuple[str, str]] = [] + queue: list[tuple[str, str, int, bool, tuple[str, ...], int]] = [ + (command, cwd, 0, True, (), 0) + ] + argv_payloads: list[tuple[list[str], str, bool, tuple[str, ...], int]] = [] while queue: - payload, payload_cwd, depth = queue.pop(0) + ( + payload, + payload_cwd, + depth, + inherited_input_safe, + outer_redirect_targets, + outer_file_redirect_count, + ) = queue.pop(0) if depth > 32: - reasons.append("nested mutation command depth is unresolved") + reasons.append( + ("shell_structure_unresolved", "nested mutation command depth is unresolved") + ) continue - segments = tokenize_command_segments(payload) - if not segments and payload.strip(): + tokenized_segments = _tokenize_command_segments_with_redirects(payload) + segments = [segment.tokens for segment in tokenized_segments] + if not tokenized_segments and payload.strip(): if _POSSIBLE_GITHUB_EXEC_RE.search(payload): - reasons.append("mutation-bearing shell payload could not be parsed") + reasons.append( + ( + "shell_parse_unresolved", + "mutation-bearing shell payload could not be parsed", + ) + ) continue current_cwd = payload_cwd - input_context_safe = True - for segment in segments: - verb, args = command_verb_and_args(segment) + input_context_safe = inherited_input_safe + nested_contexts: list[tuple[list[str], str, bool, tuple[str, ...], int]] = [] + for command_segment in tokenized_segments: + raw_segment = command_segment.tokens + executable_tokens, redirect_targets, file_redirect_count = _partition_output_redirects( + raw_segment, + cwd=current_cwd, + redirect_syntax=command_segment.redirect_syntax, + ) + active_redirect_targets = outer_redirect_targets + tuple(redirect_targets) + active_file_redirect_count = outer_file_redirect_count + file_redirect_count + segment_cwd = _segment_cwd(executable_tokens, current_cwd) + nested_contexts.append( + ( + raw_segment, + segment_cwd, + input_context_safe, + active_redirect_targets, + active_file_redirect_count, + ) + ) + verb, args = command_verb_and_args(executable_tokens) if _normalize_executable(verb) == "cd": + input_context_safe = input_context_safe and file_redirect_count == 0 if len(args) != 1 or _is_dynamic_shell_value(args[0]): - reasons.append("shell cwd transition is unresolved") + reasons.append(("cwd_unresolved", "shell cwd transition is unresolved")) elif os.path.isabs(args[0]): current_cwd = os.path.normpath(args[0]) elif current_cwd: current_cwd = os.path.normpath(os.path.join(current_cwd, args[0])) else: - reasons.append("relative shell cwd transition has no authority") + reasons.append( + ("cwd_unresolved", "relative shell cwd transition has no authority") + ) continue - found, reason = _analyze_github_segment( - segment, + found, reason_code, reason = _analyze_github_segment( + executable_tokens, cwd=current_cwd, input_context_safe=input_context_safe, + resolved_redirect_targets=active_redirect_targets, + file_redirect_count=active_file_redirect_count, ) records.extend(found) if reason: - reasons.append(reason) + reasons.append((reason_code, reason)) - interpreter_specs, has_unresolved = _extract_interpreter_segment_specs(segment) + interpreter_specs, has_unresolved = _extract_interpreter_segment_specs( + executable_tokens + ) if has_unresolved and _POSSIBLE_GITHUB_EXEC_RE.search(payload): - reasons.append("interpreter subprocess command or cwd is unresolved") + reasons.append( + ( + "interpreter_structure_unresolved", + "interpreter subprocess command or cwd is unresolved", + ) + ) for spec in interpreter_specs: - interpreter_cwd = current_cwd + interpreter_cwd = segment_cwd if spec.cwd is not None: if os.path.isabs(spec.cwd): interpreter_cwd = os.path.normpath(spec.cwd) elif current_cwd: interpreter_cwd = os.path.normpath(os.path.join(current_cwd, spec.cwd)) else: - reasons.append("relative interpreter cwd has no authority") + reasons.append( + ("cwd_unresolved", "relative interpreter cwd has no authority") + ) continue if isinstance(spec.payload, str): - queue.append((spec.payload, interpreter_cwd, depth + 1)) + queue.append( + ( + spec.payload, + interpreter_cwd, + depth + 1, + input_context_safe, + active_redirect_targets, + active_file_redirect_count, + ) + ) else: - argv_payloads.append((spec.payload, interpreter_cwd)) - - input_context_safe = input_context_safe and _segment_is_safe_before_literal_input( - segment, - cwd=current_cwd, + argv_payloads.append( + ( + spec.payload, + interpreter_cwd, + input_context_safe, + active_redirect_targets, + active_file_redirect_count, + ) + ) + + input_context_safe = ( + input_context_safe + and file_redirect_count == 0 + and _segment_is_safe_before_literal_input(executable_tokens) ) + remaining_nested_contexts = list(nested_contexts) for nested in extract_shell_command_payloads(payload): - queue.append((nested, current_cwd, depth + 1)) + matching_index = next( + ( + index + for index, context in enumerate(remaining_nested_contexts) + if _segment_evaluates_shell_payload(context[0], nested) + ), + None, + ) + matching_context: tuple[list[str], str, bool, tuple[str, ...], int] + if matching_index is None: + matching_context = ( + [], + payload_cwd, + inherited_input_safe, + outer_redirect_targets, + outer_file_redirect_count, + ) + else: + matching_context = remaining_nested_contexts.pop(matching_index) + _, nested_cwd, nested_input_safe, nested_targets, nested_count = matching_context + queue.append( + ( + nested, + nested_cwd, + depth + 1, + nested_input_safe, + nested_targets, + nested_count, + ) + ) if ( _REPEATABLE_SHELL_RE.search(payload) or _PROCESS_SUBSTITUTION_RE.search(payload) ) and _segments_have_possible_github_exec_token(segments): - reasons.append("shell loop or wrapper has unresolved mutation cardinality") + reasons.append( + ( + "shell_structure_unresolved", + "shell loop or wrapper has unresolved mutation cardinality", + ) + ) if _segments_have_dispatch_word_exec_risk(segments): - reasons.append("mutation cardinality is unresolved in a shell wrapper") + reasons.append( + ( + "shell_structure_unresolved", + "mutation cardinality is unresolved in a shell wrapper", + ) + ) - for argv, argv_cwd in argv_payloads: - found, reason = _analyze_github_segment(argv, cwd=argv_cwd) + for ( + argv, + argv_cwd, + input_context_safe, + inherited_redirect_targets, + redirect_count, + ) in argv_payloads: + found, reason_code, reason = _analyze_github_segment( + argv, + cwd=argv_cwd, + input_context_safe=input_context_safe, + resolved_redirect_targets=inherited_redirect_targets, + file_redirect_count=redirect_count, + ) records.extend(found) if reason: - reasons.append(reason) + reasons.append((reason_code, reason)) if reasons: - return _unresolved_github_analysis("; ".join(dict.fromkeys(reasons)), records) + unique_reasons = list(dict.fromkeys(reasons)) + return _unresolved_github_analysis( + reason_code=unique_reasons[0][0], + reason="; ".join(reason for _, reason in unique_reasons), + mutations=records, + ) request_count = sum(record.request_count for record in records) if request_count == 0: return _none_github_analysis() @@ -1986,6 +2332,7 @@ def analyze_github_mutations( mutations=tuple(records), request_count=request_count, review_comment_count=None, + reason_code="", reason="", ) record = records[0] @@ -1994,5 +2341,6 @@ def analyze_github_mutations( mutations=(record,), request_count=1, review_comment_count=record.review_comment_count, + reason_code="", reason="", ) diff --git a/src/autoskillit/hooks/guards/github_mutation_guard.py b/src/autoskillit/hooks/guards/github_mutation_guard.py index cba6be57f..1b276ebe7 100644 --- a/src/autoskillit/hooks/guards/github_mutation_guard.py +++ b/src/autoskillit/hooks/guards/github_mutation_guard.py @@ -64,6 +64,7 @@ class DenyTrigger(StrEnum): MALFORMED_COMMAND = "malformed_command" MALFORMED_CWD = "malformed_cwd" UNRESOLVED_MUTATION = "unresolved_mutation" + CLASSIFIER_INTERNAL_ERROR = "classifier_internal_error" MULTIPLE_MUTATIONS = "multiple_mutations" REVIEW_MUTATION = "review_mutation" @@ -73,6 +74,7 @@ class GuardDecision(NamedTuple): allow: bool trigger: DenyTrigger | None + reason_code: str _POST_PR_REVIEW_POINTER = ( @@ -96,9 +98,14 @@ class GuardDecision(NamedTuple): ), DenyTrigger.UNRESOLVED_MUTATION: ( "unresolved_mutation: this command's GitHub mutation cardinality or " - f"target cannot be statically proven safe. {_POST_PR_REVIEW_POINTER} " + "target cannot be statically proven safe. Rewrite it as exactly one literal " + "non-review mutation; use post_pr_review only for review publication. " "Unresolved mutation commands fail closed." ), + DenyTrigger.CLASSIFIER_INTERNAL_ERROR: ( + "classifier_internal_error: GitHub mutation classification failed " + "unexpectedly. The command was denied without exposing runtime details." + ), DenyTrigger.MULTIPLE_MUTATIONS: ( "multiple_mutations: this command issues more than one GitHub mutation " f"request. {_POST_PR_REVIEW_POINTER} Multiple writes fail closed." @@ -118,34 +125,49 @@ def decide(parsed: ParsedHookCommand) -> GuardDecision: are checked before any command-content classification runs. """ if parsed.tool_kind not in ("bash", "run_cmd"): - return GuardDecision(allow=True, trigger=None) + return GuardDecision(allow=True, trigger=None, reason_code="") if parsed.command is None: - return GuardDecision(allow=False, trigger=DenyTrigger.MALFORMED_COMMAND) + return GuardDecision( + allow=False, + trigger=DenyTrigger.MALFORMED_COMMAND, + reason_code="", + ) if PayloadAnomaly.FIELD_CONFUSION in parsed.anomalies: - return GuardDecision(allow=False, trigger=DenyTrigger.FIELD_CONFUSION) + return GuardDecision(allow=False, trigger=DenyTrigger.FIELD_CONFUSION, reason_code="") if any( anomaly in parsed.anomalies for anomaly in (PayloadAnomaly.NON_STRING_CWD, PayloadAnomaly.RELATIVE_CWD) ): - return GuardDecision(allow=False, trigger=DenyTrigger.MALFORMED_CWD) + return GuardDecision(allow=False, trigger=DenyTrigger.MALFORMED_CWD, reason_code="") analysis = analyze_github_mutations(parsed.command, cwd=parsed.execution_cwd) if analysis.status is GitHubMutationStatus.MULTIPLE: - return GuardDecision(allow=False, trigger=DenyTrigger.MULTIPLE_MUTATIONS) + return GuardDecision( + allow=False, + trigger=DenyTrigger.MULTIPLE_MUTATIONS, + reason_code="", + ) if analysis.status is GitHubMutationStatus.UNRESOLVED: - return GuardDecision(allow=False, trigger=DenyTrigger.UNRESOLVED_MUTATION) + return GuardDecision( + allow=False, + trigger=DenyTrigger.UNRESOLVED_MUTATION, + reason_code=analysis.reason_code, + ) if any(record.kind in _REVIEW_KINDS for record in analysis.mutations): - return GuardDecision(allow=False, trigger=DenyTrigger.REVIEW_MUTATION) - return GuardDecision(allow=True, trigger=None) + return GuardDecision(allow=False, trigger=DenyTrigger.REVIEW_MUTATION, reason_code="") + return GuardDecision(allow=True, trigger=None, reason_code="") -def _deny(trigger: DenyTrigger) -> NoReturn: +def _deny(trigger: DenyTrigger, reason_code: str) -> NoReturn: + reason = _DENY_MESSAGES[trigger] + if trigger is DenyTrigger.UNRESOLVED_MUTATION: + reason = f"{reason} classifier_code={reason_code or 'unclassified_uncertainty'}" payload = { "hookSpecificOutput": { "hookEventName": "PreToolUse", "permissionDecision": "deny", - "permissionDecisionReason": _DENY_MESSAGES[trigger], + "permissionDecisionReason": reason, } } sys.stdout.write(json.dumps(payload) + "\n") @@ -164,9 +186,9 @@ def main() -> None: decision = decide(parse_hook_command(loaded)) except Exception: _LOGGER.error("GitHub mutation classification failed", exc_info=True) - _deny(DenyTrigger.UNRESOLVED_MUTATION) + _deny(DenyTrigger.CLASSIFIER_INTERNAL_ERROR, "") if not decision.allow and decision.trigger is not None: - _deny(decision.trigger) + _deny(decision.trigger, decision.reason_code) raise SystemExit(0) diff --git a/tests/arch/test_subpackage_isolation.py b/tests/arch/test_subpackage_isolation.py index 075ff713c..33275a5a9 100644 --- a/tests/arch/test_subpackage_isolation.py +++ b/tests/arch/test_subpackage_isolation.py @@ -1127,7 +1127,7 @@ def test_data_directories_are_not_python_packages() -> None: "(issue #4479).", ), "hooks/_command_classification.py": ( - 2050, + 2350, "REQ-CNST-010-E10: shared command-classification primitive consumed by all " "command-inspecting guards — tokenization, shell-payload extraction, " "interpreter-write detection, protected-path reads, recursive payload " @@ -1140,7 +1140,9 @@ def test_data_directories_are_not_python_packages() -> None: "_segments_have_dispatch_word_exec_risk, and _gh_args_have_bare_help_flag " "must stay adjacent to the tokenizer they share. Bumped to 2050 so gh issue " "edit's target/flag grammar and statically proven fan-out count remain beside " - "the mutation aggregation authority they feed.", + "the mutation aggregation authority they feed. Bumped to 2350 for #4581's " + "quote-aware output-redirection partition, nested writer provenance, and " + "bounded diagnostic codes, which share that same mutation authority.", ), "session.py": ( 1060, diff --git a/tests/hooks/fixtures/session_replays/__init__.py b/tests/hooks/fixtures/session_replays/__init__.py index 5ec9e3261..d93a7c999 100644 --- a/tests/hooks/fixtures/session_replays/__init__.py +++ b/tests/hooks/fixtures/session_replays/__init__.py @@ -14,6 +14,7 @@ from pathlib import Path INCIDENT_TRANSCRIPT: str = "incident_transcript_v1.jsonl" +INTERACTIVE_GITHUB_MUTATION: str = "interactive_github_mutation_v1.jsonl" def fixture_path(name: str) -> Path: @@ -21,4 +22,4 @@ def fixture_path(name: str) -> Path: return Path(__file__).parent / name -__all__ = ["INCIDENT_TRANSCRIPT", "fixture_path"] +__all__ = ["INCIDENT_TRANSCRIPT", "INTERACTIVE_GITHUB_MUTATION", "fixture_path"] diff --git a/tests/hooks/fixtures/session_replays/interactive_github_mutation_v1.jsonl b/tests/hooks/fixtures/session_replays/interactive_github_mutation_v1.jsonl new file mode 100644 index 000000000..61ef7e70b --- /dev/null +++ b/tests/hooks/fixtures/session_replays/interactive_github_mutation_v1.jsonl @@ -0,0 +1,3 @@ +{"session_env": {"AUTOSKILLIT_HEADLESS": "", "AUTOSKILLIT_SESSION_TYPE": ""}, "state_setup": []} +{"payload": {"tool_name": "Bash", "tool_input": {"command": "gh issue edit 4581 --repo TalonT-Org/AutoSkillit --body-file /tmp/body > /tmp/out 2>&1"}, "cwd": "{{ORCHESTRATING_ROOT}}"}, "expectations": {"allowed": true, "max_severity": "none"}} +{"payload": {"tool_name": "Bash", "tool_input": {"command": "gh pr review 4581 --approve"}, "cwd": "{{ORCHESTRATING_ROOT}}"}, "expectations": {"allowed": false, "max_severity": "none"}} diff --git a/tests/hooks/test_command_classification.py b/tests/hooks/test_command_classification.py index 8186785c4..580d4cd53 100644 --- a/tests/hooks/test_command_classification.py +++ b/tests/hooks/test_command_classification.py @@ -3,6 +3,7 @@ from __future__ import annotations import json +import re import shlex from pathlib import Path @@ -151,6 +152,65 @@ def test_adjacent_background_ampersand(self): result = tokenize_command_segments("echo ok&pip install -e .") assert len(result) == 2 + @pytest.mark.parametrize("redirect", ["2>&1", "1>&2", ">&1"]) + def test_fd_duplication_does_not_create_a_command_boundary(self, redirect: str) -> None: + assert tokenize_command_segments(f"gh issue edit 23 --title x {redirect}") == [ + ["gh", "issue", "edit", "23", "--title", "x", redirect] + ] + + @pytest.mark.parametrize( + "command", + [ + "printf '%s' '2>/tmp/out'", + 'printf "%s" "2>/tmp/out"', + r"printf %s 2\>/tmp/out", + ], + ids=["single-quoted", "double-quoted", "escaped"], + ) + def test_quoted_or_escaped_redirect_shape_remains_literal_argv(self, command: str) -> None: + segments = command_classification._tokenize_command_segments_with_redirects(command) + + assert segments[0].tokens == ["printf", "%s", "2>/tmp/out"] + assert segments[0].redirect_syntax == [False, False, False] + + @pytest.mark.parametrize( + ("command", "expected_tokens", "expected_targets", "expected_count"), + [ + ("cmd > /tmp/out", ["cmd"], ["/tmp/out"], 1), + ("cmd 2>/tmp/err", ["cmd"], ["/tmp/err"], 1), + ("cmd >> /tmp/out", ["cmd"], ["/tmp/out"], 1), + ("cmd 2>>/tmp/err", ["cmd"], ["/tmp/err"], 1), + ("cmd >$OUT", ["cmd"], [], 1), + ("cmd >", ["cmd"], [], 1), + ("cmd 2>&1", ["cmd"], [], 0), + ("curl --output /tmp/out URL", ["curl", "--output", "/tmp/out", "URL"], [], 0), + ], + ids=[ + "separate", + "merged", + "append-separate", + "append-merged", + "dynamic", + "missing", + "fd-dup", + "curl-option", + ], + ) + def test_output_control_partition( + self, + command: str, + expected_tokens: list[str], + expected_targets: list[str], + expected_count: int, + ) -> None: + segment = command_classification._tokenize_command_segments_with_redirects(command)[0] + + assert command_classification._partition_output_redirects( + segment.tokens, + cwd="/work", + redirect_syntax=segment.redirect_syntax, + ) == (expected_tokens, expected_targets, expected_count) + def test_bare_newline_separates_segments(self): result = tokenize_command_segments("echo ok\npip install -e .") assert len(result) == 2 @@ -880,6 +940,7 @@ def test_read_only_command_has_exact_empty_analysis(self) -> None: mutations=(), request_count=0, review_comment_count=None, + reason_code="", reason="", ) ) @@ -902,6 +963,7 @@ def test_simple_rest_review_has_exact_record(self) -> None: ), request_count=1, review_comment_count=None, + reason_code="", reason="", ) @@ -923,9 +985,75 @@ def test_simple_non_review_mutation_has_exact_record(self) -> None: ), request_count=1, review_comment_count=None, + reason_code="", reason="", ) + @pytest.mark.parametrize( + ("baseline", "redirected"), + [ + ( + "gh issue edit 4581 --repo TalonT-Org/AutoSkillit --body-file /tmp/body", + "sleep 1 && gh issue edit 4581 --repo TalonT-Org/AutoSkillit " + "--body-file /tmp/body 2>&1 | head -c 4000", + ), + ( + "gh issue edit 4581 --repo TalonT-Org/AutoSkillit --body-file /tmp/body", + "gh issue edit 4581 --repo TalonT-Org/AutoSkillit " + "--body-file /tmp/body > /tmp/out 2>&1", + ), + ( + "gh api --method PATCH repos/TalonT-Org/AutoSkillit/issues/4581 -f title=x", + "gh api --method PATCH repos/TalonT-Org/AutoSkillit/issues/4581 " + "-f title=x > /tmp/out 2>&1", + ), + ( + "curl -X PATCH https://api.github.com/repos/o/r/issues/4581 -d '{}';", + "curl -X PATCH https://api.github.com/repos/o/r/issues/4581 " + "-d '{}' > /tmp/out 2>&1", + ), + ], + ids=["filing-pipeline", "filing-file", "gh-api", "curl"], + ) + def test_output_redirection_does_not_change_mutation_identity( + self, + baseline: str, + redirected: str, + ) -> None: + expected = analyze_github_mutations(baseline) + actual = analyze_github_mutations(redirected) + + assert actual.status is GitHubMutationStatus.SINGLE_RESOLVED + assert actual.request_count == 1 + assert actual.mutations == expected.mutations + + @pytest.mark.parametrize( + ("command", "expected_status"), + [ + ("gh issue edit 23 24 --title x > /tmp/out", GitHubMutationStatus.MULTIPLE), + ( + "gh api --method PATCH /repos/o/r/issues/23 /repos/o/r/issues/24 > /tmp/out", + GitHubMutationStatus.UNRESOLVED, + ), + ( + "curl -X PATCH https://api.github.com/repos/o/r/issues/23 " + "https://api.github.com/repos/o/r/issues/24 > /tmp/out", + GitHubMutationStatus.UNRESOLVED, + ), + ( + "gh api --method PATCH /repos/o/r/issues/23 > >(tee /tmp/out)", + GitHubMutationStatus.UNRESOLVED, + ), + ], + ids=["issue-targets", "api-routes", "curl-urls", "process-substitution"], + ) + def test_redirect_normalization_preserves_negative_controls( + self, + command: str, + expected_status: GitHubMutationStatus, + ) -> None: + assert analyze_github_mutations(command).status is expected_status + @pytest.mark.parametrize( "command,kind", [ @@ -1004,6 +1132,9 @@ def test_unresolved_mutations_report_reason(self, command: str) -> None: assert analysis.status is GitHubMutationStatus.UNRESOLVED assert analysis.request_count is None assert analysis.reason + assert analysis.reason_code + assert len(analysis.reason_code.encode("utf-8")) <= 64 + assert re.fullmatch(r"[a-z][a-z0-9_]*", analysis.reason_code) @pytest.mark.parametrize( "command", @@ -1092,6 +1223,32 @@ def test_identical_nested_mutation_payloads_are_counted_per_occurrence(self) -> assert analysis.request_count == 2 assert len(analysis.mutations) == 2 + def test_identical_nested_payloads_keep_per_occurrence_cwd(self, tmp_path: Path) -> None: + (tmp_path / "payload.json").write_text(json.dumps({"body": "x"}), encoding="utf-8") + nested = "gh api --method POST /repos/o/r/issues/7/comments --input payload.json" + command = ( + f"cd {shlex.quote(str(tmp_path))} && $({nested}) && " + f"cd {shlex.quote(str(tmp_path / 'missing'))} && $({nested})" + ) + + analysis = analyze_github_mutations(command, cwd=str(tmp_path)) + + assert analysis.status is GitHubMutationStatus.UNRESOLVED + assert analysis.reason_code == "unsafe_input_provenance" + + def test_nested_payload_uses_its_structural_segment_context(self, tmp_path: Path) -> None: + payload = tmp_path / "payload.json" + payload.write_text(json.dumps({"body": "x"}), encoding="utf-8") + nested = f"gh api --method POST /repos/o/r/issues/7/comments --input {payload}" + command = ( + f"echo {shlex.quote(nested)} && printf x > {payload} && bash -c {shlex.quote(nested)}" + ) + + analysis = analyze_github_mutations(command, cwd=str(tmp_path)) + + assert analysis.status is GitHubMutationStatus.UNRESOLVED + assert analysis.reason_code == "unsafe_input_provenance" + def test_prior_command_that_can_rewrite_literal_input_is_unresolved( self, tmp_path: Path, @@ -1108,6 +1265,146 @@ def test_prior_command_that_can_rewrite_literal_input_is_unresolved( assert analysis.status is GitHubMutationStatus.UNRESOLVED assert "prior command may rewrite" in analysis.reason + @pytest.mark.parametrize( + "prefix", + [ + "python3 -c 'print(1)' && ", + "printf x > prior.out && ", + "cd /tmp > /tmp/cd.out && ", + ], + ids=["non-allowlisted", "prior-writer", "cd-writer"], + ) + def test_prior_command_provenance_remains_fail_closed( + self, + prefix: str, + tmp_path: Path, + ) -> None: + payload = tmp_path / "payload.json" + payload.write_text(json.dumps({"body": "x"}), encoding="utf-8") + + analysis = analyze_github_mutations( + prefix + f"gh api --method POST /repos/o/r/issues/7/comments --input {payload}", + cwd=str(tmp_path), + ) + + assert analysis.status is GitHubMutationStatus.UNRESOLVED + assert analysis.reason_code == "unsafe_input_provenance" + + @pytest.mark.parametrize("redirect", ["2>&1", ">&1"]) + def test_fd_duplication_does_not_make_later_input_unsafe( + self, + redirect: str, + tmp_path: Path, + ) -> None: + payload = tmp_path / "payload.json" + payload.write_text(json.dumps({"body": "x"}), encoding="utf-8") + + analysis = analyze_github_mutations( + f"printf ok {redirect} && gh api --method POST /repos/o/r/issues/7/comments " + f"--input {payload}", + cwd=str(tmp_path), + ) + + assert analysis.status is GitHubMutationStatus.SINGLE_RESOLVED + + @pytest.mark.parametrize( + ("redirect", "expected_status", "expected_reason_code"), + [ + ("> different.out", GitHubMutationStatus.SINGLE_RESOLVED, ""), + ("> payload.json", GitHubMutationStatus.UNRESOLVED, "unsafe_input_provenance"), + ("> $OUT", GitHubMutationStatus.UNRESOLVED, "unsafe_input_provenance"), + ], + ids=["distinct", "same-path", "unresolved-target"], + ) + def test_current_input_redirect_alias_safety( + self, + redirect: str, + expected_status: GitHubMutationStatus, + expected_reason_code: str, + tmp_path: Path, + ) -> None: + (tmp_path / "payload.json").write_text(json.dumps({"body": "x"}), encoding="utf-8") + command = ( + "env -C nested gh api --method POST /repos/o/r/issues/7/comments " + f"--input ../payload.json {redirect}" + ) + (tmp_path / "nested").mkdir() + + analysis = analyze_github_mutations(command, cwd=str(tmp_path)) + + assert analysis.status is expected_status + assert analysis.reason_code == expected_reason_code + + def test_current_input_redirect_rejects_hard_link_alias(self, tmp_path: Path) -> None: + payload = tmp_path / "payload.json" + payload.write_text(json.dumps({"body": "x"}), encoding="utf-8") + alias = tmp_path / "alias.json" + alias.hardlink_to(payload) + + analysis = analyze_github_mutations( + f"gh api --method POST /repos/o/r/issues/7/comments --input {payload} > {alias}", + cwd=str(tmp_path), + ) + + assert analysis.status is GitHubMutationStatus.UNRESOLVED + assert analysis.reason_code == "unsafe_input_provenance" + + @pytest.mark.parametrize("wrapper", ["shell", "argv"]) + def test_parent_redirect_provenance_reaches_nested_mutation( + self, + wrapper: str, + tmp_path: Path, + ) -> None: + payload = tmp_path / "payload.json" + payload.write_text(json.dumps({"body": "x"}), encoding="utf-8") + nested = f"gh api --method POST /repos/o/r/issues/7/comments --input {payload}" + if wrapper == "shell": + command = f"bash -c {shlex.quote(nested)} > {payload}" + else: + argv = [ + "gh", + "api", + "--method", + "POST", + "/repos/o/r/issues/7/comments", + "--input", + str(payload), + ] + command = ( + "python3 -c " + + shlex.quote(f"import subprocess; subprocess.run({argv!r})") + + f" > {payload}" + ) + + analysis = analyze_github_mutations(command, cwd=str(tmp_path)) + + assert analysis.status is GitHubMutationStatus.UNRESOLVED + assert analysis.reason_code == "unsafe_input_provenance" + + def test_unresolved_reason_codes_are_distinct_by_failure_family(self, tmp_path: Path) -> None: + payload = tmp_path / "payload.json" + payload.write_text(json.dumps({"body": "x"}), encoding="utf-8") + analyses = { + "dynamic target": analyze_github_mutations( + "gh issue edit $ISSUE --title x" + ).reason_code, + "shell structure": analyze_github_mutations( + "for x in 1 2; do gh issue edit 1 --title x; done" + ).reason_code, + "unsafe input provenance": analyze_github_mutations( + f"python3 -c 'print(1)' && gh api --method POST /repos/o/r/issues/7/comments " + f"--input {payload}" + ).reason_code, + "cwd": analyze_github_mutations("cd $DIR && gh issue edit 1 --title x").reason_code, + } + + assert analyses == { + "dynamic target": "dynamic_target", + "shell structure": "shell_structure_unresolved", + "unsafe input provenance": "unsafe_input_provenance", + "cwd": "cwd_unresolved", + } + def test_literal_interpreter_cwd_is_used_for_input_resolution(self, tmp_path: Path) -> None: nested = tmp_path / "nested" nested.mkdir() @@ -1312,6 +1609,7 @@ def test_issue_edit_counts_each_static_target(self, command: str) -> None: assert analysis.status is GitHubMutationStatus.MULTIPLE assert analysis.request_count == 2 assert analysis.mutations[0].request_count == 2 + assert analysis.reason_code == "" @pytest.mark.parametrize( "command", diff --git a/tests/hooks/test_github_mutation_guard.py b/tests/hooks/test_github_mutation_guard.py index f22d63f1b..4608afc57 100644 --- a/tests/hooks/test_github_mutation_guard.py +++ b/tests/hooks/test_github_mutation_guard.py @@ -32,12 +32,21 @@ def _run_cmd_event(command: str, *, cwd: str | None = None) -> dict: return {"tool_name": _RUN_CMD_TOOL, "tool_input": tool_input} -def _run_hook(event: dict, monkeypatch: pytest.MonkeyPatch) -> dict: +def _run_hook( + event: dict, + monkeypatch: pytest.MonkeyPatch, + *, + headless: bool = True, +) -> dict: from autoskillit.hooks.guards.github_mutation_guard import main monkeypatch.setattr("sys.stdin", io.StringIO(json.dumps(event))) - monkeypatch.setenv("AUTOSKILLIT_HEADLESS", "1") - monkeypatch.setenv("AUTOSKILLIT_SESSION_TYPE", "skill") + if headless: + monkeypatch.setenv("AUTOSKILLIT_HEADLESS", "1") + monkeypatch.setenv("AUTOSKILLIT_SESSION_TYPE", "skill") + else: + monkeypatch.delenv("AUTOSKILLIT_HEADLESS", raising=False) + monkeypatch.delenv("AUTOSKILLIT_SESSION_TYPE", raising=False) stdout = io.StringIO() try: with redirect_stdout(stdout): @@ -307,14 +316,48 @@ def test_unexpected_classifier_error_fails_closed( from autoskillit.hooks.guards import github_mutation_guard def _raise(*_args, **_kwargs): - raise RuntimeError("classifier failed") + raise RuntimeError("classifier-private-sentinel") monkeypatch.setattr(github_mutation_guard, "analyze_github_mutations", _raise) result = _run_hook(_bash_event("gh issue edit 23 --title x", cwd=str(tmp_path)), monkeypatch) assert result["hookSpecificOutput"]["permissionDecision"] == "deny" - assert "unresolved_mutation" in result["hookSpecificOutput"]["permissionDecisionReason"] + reason = result["hookSpecificOutput"]["permissionDecisionReason"] + assert "classifier_internal_error" in reason + assert "classifier-private-sentinel" not in reason + + +@pytest.mark.parametrize( + "command", + [ + "gh pr edit 4581 --body-file /tmp/body", + ( + "sleep 1 && gh issue edit 4581 --repo TalonT-Org/AutoSkillit " + "--body-file /tmp/body 2>&1 | head -c 4000" + ), + ("gh issue edit 4581 --repo TalonT-Org/AutoSkillit --body-file /tmp/body > /tmp/out 2>&1"), + "gh api --method PATCH repos/TalonT-Org/AutoSkillit/issues/4581 -f title=x", + ], + ids=["pr-edit", "filing-pipeline", "filing-file", "rest-patch"], +) +def test_interactive_cook_allows_literal_single_non_review_mutation( + command: str, + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + from autoskillit.hooks.guards.github_mutation_guard import decide, parse_hook_command + + event = _bash_event(command, cwd=str(tmp_path)) + + assert decide(parse_hook_command(event)).allow + result = _run_hook( + event, + monkeypatch, + headless=False, + ) + + assert result == {} @pytest.mark.parametrize("raw", ["{bad", "[]"], ids=["malformed-json", "non-object"]) @@ -616,7 +659,11 @@ def test_unresolved_mutation_denies_with_unresolved_mutation_reason( result = _run_hook(event, monkeypatch) assert result["hookSpecificOutput"]["permissionDecision"] == "deny" - assert "unresolved_mutation" in result["hookSpecificOutput"]["permissionDecisionReason"] + reason = result["hookSpecificOutput"]["permissionDecisionReason"] + assert "unresolved_mutation" in reason + assert "dynamic_target" in reason + assert "$OWNER" not in reason + assert "GitHub API route is dynamic" not in reason def test_deny_messages_mapping_is_exhaustive() -> None: @@ -705,6 +752,7 @@ def test_guard_registration_is_exact() -> None: entry = entries[0] assert entry.event_type == "PreToolUse" assert entry.matcher == r"Bash|mcp__.*autoskillit.*__run_cmd" + assert entry.session_scope == "any" assert entry.mechanism == "deny" assert entry.codex_status == "works-as-is" assert entry.enforcement_strength == { diff --git a/tests/hooks/test_session_replay.py b/tests/hooks/test_session_replay.py index 86a97e04a..1c713c232 100644 --- a/tests/hooks/test_session_replay.py +++ b/tests/hooks/test_session_replay.py @@ -45,7 +45,11 @@ from autoskillit.hooks._capture_lifecycle import CaptureLifecycleStore from .conftest import _FAILURE_GRADE_RE -from .fixtures.session_replays import INCIDENT_TRANSCRIPT, fixture_path +from .fixtures.session_replays import ( + INCIDENT_TRANSCRIPT, + INTERACTIVE_GITHUB_MUTATION, + fixture_path, +) pytestmark = [pytest.mark.layer("hooks"), pytest.mark.medium] @@ -386,6 +390,28 @@ def test_incident_transcript_replay_is_clean_and_denies_genuine_mutation(tmp_pat ) +def test_interactive_github_mutation_replay_reaches_global_guard(tmp_path: Path) -> None: + orchestrating_root = tmp_path / "interactive-project" + orchestrating_root.mkdir() + + replayed = replay( + INTERACTIVE_GITHUB_MUTATION, + {"{{ORCHESTRATING_ROOT}}": str(orchestrating_root)}, + process_cwd=orchestrating_root, + ) + + assert_replay_clean(replayed) + assert len(replayed) == 2 + allow_event, allow_results = replayed[0] + review_event, review_results = replayed[1] + guard_name = "guards/github_mutation_guard.py" + assert allow_event.allowed + assert any(result.script == guard_name for result in allow_results) + assert not any(result.denied for result in allow_results) + assert not review_event.allowed + assert any(result.script == guard_name and result.denied for result in review_results) + + def test_incident_transcript_capture_wrapped_command_over_seeded_backlog_is_clean( tmp_path: Path, ) -> None: