diff --git a/.claude-plugin/plugin.json b/.claude-plugin/plugin.json index 318244d..47d4989 100644 --- a/.claude-plugin/plugin.json +++ b/.claude-plugin/plugin.json @@ -1,6 +1,6 @@ { "name": "agentsmd", - "version": "17.4.2", + "version": "17.4.3", "description": "Portable agent rules, Project Direction, Delivery System profiles, owned workflow Skills, and deterministic version control.", "author": { "name": "lukaszmaj" diff --git a/.codex-plugin/plugin.json b/.codex-plugin/plugin.json index aaad0b3..5772072 100644 --- a/.codex-plugin/plugin.json +++ b/.codex-plugin/plugin.json @@ -1,6 +1,6 @@ { "name": "agentsmd", - "version": "17.4.2", + "version": "17.4.3", "description": "Portable agent rules, Project Direction, Delivery System profiles, owned workflow Skills, and deterministic version control.", "author": { "name": "lukaszmaj", diff --git a/.grok-plugin/plugin.json b/.grok-plugin/plugin.json index 1f94bd5..509a80d 100644 --- a/.grok-plugin/plugin.json +++ b/.grok-plugin/plugin.json @@ -1,6 +1,6 @@ { "name": "agentsmd", - "version": "17.4.2", + "version": "17.4.3", "description": "Portable agent rules, Project Direction, Delivery System profiles, owned workflow Skills, and deterministic version control.", "author": { "name": "lukaszmaj" diff --git a/.muse-plugin/plugin.json b/.muse-plugin/plugin.json index 31bd368..8442ca5 100644 --- a/.muse-plugin/plugin.json +++ b/.muse-plugin/plugin.json @@ -2,7 +2,7 @@ "schemaVersion": 1, "name": "agentsmd", "displayName": "AgentsMD", - "version": "17.4.2", + "version": "17.4.3", "description": "Portable agent rules, Project Direction, Delivery System profiles, owned workflow Skills, and deterministic version control.", "compat": { "source": "native", diff --git a/CHANGELOG.md b/CHANGELOG.md index b67b3d0..2eb7897 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,6 +3,12 @@ All completed repository versions are recorded here. Release entries follow SemVer and identify one user-consumable repository state. +## [17.4.3] - 2026-10-19 + +### Changed + +- Gate gh new aliases and visible creates inside shell wrappers and shell-fed scripts + ## [17.4.2] - 2026-10-19 ### Changed diff --git a/README.md b/README.md index e9b3b61..d84ab37 100644 --- a/README.md +++ b/README.md @@ -336,7 +336,7 @@ read the procedure it links in full. [The #174 benchmark](docs/work/174-routing- and [the #182 sweep](docs/work/182-routing-sweep/results.md) record its effect. **The Elon gate.** On every harness, Issue and PR creation through -`gh issue create`, `gh pr create`, or REST `gh api` POST requests to +`gh issue create|new`, `gh pr create|new`, or REST `gh api` POST requests to `repos/OWNER/REPO/issues` or `repos/OWNER/REPO/pulls` whose body lacks a non-empty field for **Requirements and who asked**, **Deleted**, **Bottleneck**, or **Checked myself** is blocked (exit 2), and the missing @@ -349,6 +349,78 @@ with `cat > FILE <<'EOF'` or `tee FILE <<'EOF'`. For `gh api`, the hook reads `--input -`); when `--input` is present, only its JSON body counts. GraphQL, MCP tools, and API clients other than `gh api` are not checked. +The `new` aliases use the same body checks and bare-help exemption as `create`. +Only a single plain invocation with one help flag and optionally one repo +selector is exempt; mixed flags and shell syntax remain gated. +The parser checks each visible create separately, including commands inside +`$(…)`, backticks and `(…)`, compound commands, literal scripts passed to +`sh|bash|zsh|dash|ksh -c`, and scripts fed to those shells by heredocs or +here-strings, including shell positional arguments and `/dev/stdin`. Shell +options with separate values (`-o`, `-O`, `+o`, `--init-file`, `--rcfile`) are +consumed before deciding whether a script file was supplied. +A shell consuming anything other than a fully parsed literal `-c` script +activates a conservative whole-command check. It joins backslash-newline +continuations, removes quoting and uses the same gh flag-value classification +as bare commands. Every visible create not already checked as its own create +blocks, regardless of pipes, compounds, keywords or redirection descriptors. +Direct literal heredocs and here-strings can receive normal record checks. +Script input through producer pipes, compounds, prefixes or `/dev/fd/N` may +block even with a record; their data flow is deliberately not modeled. + +Producer modeling remains only for body sources and file writes: plain `cat` +without operands, `tee`, flag-free single-operand `echo` without backslashes, +and `printf '%s'` / `printf '%s\n'`. Only the word at `executable_index` is the +producer; a producer name among another program's arguments cannot authorize +its output. Unknown body-source content blocks. Unknown escape pairs are +possible word boundaries in the conservative script check, preserving zsh echo +coverage without interpreting producer output. Harmless shell input with no +visible create stays allowed. +An exact double-quoted `"$(cat <<'EOF' … EOF)"` body is read as its heredoc content, with +quoted or unquoted delimiters; other substituted bodies remain unreadable. +Each create uses its own inline body, body file or stdin heredoc; a record on +another invocation cannot authorize it. Repeated body options block instead of +risking checking a different body from the one gh submits. Wrapped help gets no +bare-help exemption. + +Quoted arguments to `git commit`, `echo`, `printf`, read-only searches and +quoted-delimiter heredoc bodies are data. Unquoted heredoc bodies execute +`$(…)` and backticks: only those substitutions are scanned, using double-quote +expansion rules. Literal body text is not scanned as a script unless fed to a +shell. Array and arithmetic expressions stay data while their actual command +substitutions are checked. Visible function bodies are scanned like subshells, +including definitions that are never called. +One shared routine scans every command-word position for gh and shells, so +keywords, prefixes and leading redirections cannot hide them. Producers use +only the actual executable position. Unquoted data may conservatively match a +create. Whole quoted examples remain data during normal classification; the +whole-command guard can block even unrelated quoted examples when the same +command also invokes a shell outside a fully parsed literal `-c` script. +A record-checked create's own body remains data. + +Limits: this is an accidental-create guard, not a shell interpreter or sandbox. +It does not resolve `eval` or non-visible script generators, +split variables or `$IFS`, parameter expansion that constructs commands, brace +expansion, ANSI-C/locale word expansion, function call/argument resolution, +aliases, unseen input to process substitution such as `>(sh)`, variable-supplied script text +such as `echo "$s" | sh`, scripts written in earlier commands and run later +(including `cat s.txt | sh`), or +commands assembled from xargs input such as `echo 'issue create …' | xargs gh`. +Other limits are `env -S`, zsh `=gh`, `. /dev/stdin` or `source /dev/stdin`, +`tcsh -c`, format-assembled scripts, the separated `$((gh ...) )` form, and +ANSI-C `$'...'` quoting that desynchronizes the scanner. `printf --` and +unmodeled formats are unknown producers: visible create signatures feeding a +shell block, while generated signatures absent from the input remain a limit. +Visible process-substitution signatures such as `bash <(echo 'gh issue create …')` +block under the whole-command check; only non-visible generators remain a limit. +A visible `xargs gh issue create …` or `find … -exec gh issue create …` is checked. +Unquoted `cat <]*>$") HEADING = re.compile(r"^\s*#{1,6}\s") SEPARATORS = ("&&", "||", "|&", ";", "|", "\n") VALUE_FLAGS = ("-R", "--repo", "--hostname") +# Create flags may precede the verb. Share global flag-value knowledge while +# keeping API parsing independent of Issue/PR-specific options. +CREATE_VALUE_FLAGS = VALUE_FLAGS + ( + "-a", "--assignee", "--attach", "--blocked-by", "--blocking", "-B", "--base", + "-b", "--body", "-F", "--body-file", "-H", "--head", "-l", "--label", + "-m", "--milestone", "--parent", "-p", "--project", "--recover", + "-r", "--reviewer", "-T", "--template", "-t", "--title", "--type", +) # Tools whose `command` is patch data, and tools whose `command` a shell runs. PATCH_TOOLS = ("apply_patch",) SHELL_TOOLS = ("Bash", "bash", "shell", "local_shell", "exec_command", "run_terminal_command") @@ -709,7 +717,55 @@ def field_value(body: str, field: str) -> str: return "" -def shell_segments(command: str) -> list[tuple[str, str | None, bool]]: +class ShellCommand: + def __init__(self, text: str, *, children: list[list["ShellCommand"]]): + self.text = text + self.stdin: str | None = None + self.stdin_visible: str | None = None + self.heredoc_text: list[str] = [] + self.piped = False + self.children = children + self.literals: dict[str, str] = {} + + +SUBSTITUTION = "__agentsmd_unknown_substitution__" + + +def shell_segments(command: str) -> list[ShellCommand]: + return shell_region(command, 0, None)[0] + + +def heredoc_word(command: str, start: int) -> tuple[str, bool, int] | None: + """Read one shell delimiter word, then remove its quotes without expansion.""" + index = start + while index < len(command) and command[index] in " \t": + index += 1 + begin, quote, quoted = index, "", False + while index < len(command): + char = command[index] + if char == "\\" and quote != "'": + quoted = True + index += 2 + continue + if quote: + if char == quote: + quote = "" + elif char in "'\"": + quote, quoted = char, True + elif char.isspace() or char in ";&|<>()": + break + index += 1 + if quote or index == begin: + return None + try: + words = shlex.split(command[begin:index], posix=True) + except ValueError: + return None + return (words[0], quoted, index) if len(words) == 1 else None + + +def shell_region(command: str, start: int, + closing: str | None, *, data_quote: bool = False) -> tuple[list[ShellCommand], int]: """Split a shell command into simple commands, each with its heredoc text and whether a pipe feeds it from the previous one. @@ -718,42 +774,107 @@ def shell_segments(command: str) -> list[tuple[str, str | None, bool]]: declares it, even when later segments share its line. As in the shell, only the last heredoc on stdin reaches the command; `3< None: - segments.append("".join(current)) + segment = ShellCommand("".join(current), children=children.copy()) + segment.literals = literals.copy() + segments.append(segment) + literals.clear() current.clear() + children.clear() + + def finish(end: int) -> tuple[list[ShellCommand], int]: + close() + for i, segment in enumerate(segments): + segment.stdin_visible = heredocs.get(i) + segment.stdin = None if i in dynamic_heredocs else segment.stdin_visible + segment.piped = i in piped + return segments, end while index < length: char = command[index] - if quote: + if quote == "'": current.append(char) - if char == "\\" and quote == '"' and index + 1 < length: - current.append(command[index + 1]) - index += 2 - continue if char == quote: quote = "" index += 1 continue + if char == "\\" and index + 1 < length: + if command[index + 1] != "\n": + current.append(command[index:index + 2]) + index += 2 + continue + if data_quote and closing == "))": + if char == "(": + arithmetic_depth += 1 + elif char == ")": + if arithmetic_depth: + arithmetic_depth -= 1 + elif command.startswith("))", index): + return finish(index + 2) + if closing and char == closing and not quote: + return finish(index + 1) + if command.startswith("$((", index): + arithmetic, index = shell_region(command, index + 3, "))", data_quote=True) + # Arithmetic text is data. Only its actual substitutions execute. + children.extend(child for entry in arithmetic for child in entry.children) + current.append(SUBSTITUTION) + continue + if command.startswith("$(", index) or char == "`": + offset, end = (2, ")") if char == "$" else (1, "`") + child, index = shell_region(command, index + offset, end) + children.append(child) + marker = SUBSTITUTION + str(index) + literal = literal_cat(child) if offset == 2 and quote == '"' else None + if literal is not None: + literals[marker] = literal + current.append(marker) + continue + if quote: + current.append(char) + if char == quote and not data_quote: + quote = "" + index += 1 + continue if char in "'\"": quote = char current.append(char) index += 1 continue - if char == "\\" and index + 1 < length: - # Bash removes an unquoted backslash-LF before recognizing words - # and comments. CRLF is not a continuation. - if command[index + 1] != "\n": - current.append(command[index:index + 2]) - index += 2 + # Scan visible function bodies conservatively, just like subshells; + # resolving calls and parameters remains outside this parser. + if not "".join(current).strip(): + function = re.match(r"(?:function +)?[A-Za-z_][A-Za-z0-9_]* *\( *\) *\{", command[index:]) + if function: + child, index = shell_region(command, index + function.end(), "}") + children.append(child) + current.append(SUBSTITUTION) + continue + if char == "(": + child, index = shell_region(command, index + 1, ")") + if re.search(r"(?:^|\s)[A-Za-z_][A-Za-z0-9_]*=$", "".join(current)): + # Array entries are data, but substitutions in them execute. + children.extend(grandchild for entry in child for grandchild in entry.children) + else: + children.append(child) + current.append(SUBSTITUTION) + continue + if char == ")": + # Outside a substitution/subshell this terminates a case pattern. + current.append(" ") + index += 1 continue if char == "#" and (not current or current[-1].isspace()): # Only an unquoted word starting with # opens a shell comment. @@ -767,8 +888,9 @@ def shell_segments(command: str) -> list[tuple[str, str | None, bool]]: index += 3 continue if command.startswith("<<", index): - match = re.match(r"<<(-?)\s*(['\"]?)([A-Za-z0-9_]+)\2", command[index:]) - if match: + strip_tabs = command.startswith("<<-", index) + delimiter = heredoc_word(command, index + (3 if strip_tabs else 2)) + if delimiter: # Digits that start a word before `<<` name the fd it feeds. digits = 0 while digits < len(current) and current[-1 - digits].isdigit(): @@ -778,14 +900,14 @@ def shell_segments(command: str) -> list[tuple[str, str | None, bool]]: if digits and (digits == len(current) or current[-1 - digits].isspace()): stdin = fd == "0" del current[len(current) - digits:] - pending.append((match.group(3), match.group(1) == "-", len(segments), stdin)) - index += match.end() + word, quoted, index = delimiter + pending.append((word, strip_tabs, len(segments), stdin, quoted)) continue if char == "\n" and pending: # Heredoc bodies start on the next line. close() index += 1 - for word, strip_tabs, owner, stdin in pending: + for word, strip_tabs, owner, stdin, quoted in pending: body: list[str] = [] while index < length: newline = command.find("\n", index) @@ -793,9 +915,24 @@ def shell_segments(command: str) -> list[tuple[str, str | None, bool]]: index = length if newline < 0 else newline + 1 if (line.lstrip("\t") if strip_tabs else line) == word: break - body.append(line) + body.append(line.lstrip("\t") if strip_tabs else line) + content = "".join(line + "\n" for line in body) + dynamic = False + visible = content + if not quoted: + # Quotes in the body are literal. Only substitutions run, + # with the same backslash rules as double-quoted text. + data, _ = shell_region(content, 0, None, data_quote=True) + segments[owner].children.extend(child for entry in data for child in entry.children) + dynamic = any(entry.children for entry in data) + visible = "".join(entry.text for entry in data) + segments[owner].heredoc_text.append(visible) if stdin: - heredocs[owner] = "".join(line + "\n" for line in body) + heredocs[owner] = content + if dynamic: + dynamic_heredocs.add(owner) + else: + dynamic_heredocs.discard(owner) pending.clear() continue separator = next((s for s in SEPARATORS if command.startswith(s, index)), None) @@ -813,8 +950,22 @@ def shell_segments(command: str) -> list[tuple[str, str | None, bool]]: continue current.append(char) index += 1 - close() - return [(text, heredocs.get(i), i in piped) for i, text in enumerate(segments)] + return finish(index) + + +def literal_cat(segments: list[ShellCommand]) -> str | None: + """Resolve only an exact cat + stdin heredoc substitution, never execute it.""" + commands = [segment for segment in segments if segment.text.strip()] + if len(commands) != 1: + return None + command = commands[0] + if command.children or command.stdin is None or command.piped: + return None + try: + words = shlex.split(command.text) + except ValueError: + return None + return command.stdin.rstrip("\n") if words == ["cat"] else None OUTPUTS = (">>", ">|", ">&", ">") @@ -841,11 +992,6 @@ def redirections(words: list[str]) -> tuple[list[str], list[tuple[str, str, str] return operands, redirects -def stdin_redirected(redirects: list[tuple[str, str, str]]) -> bool: - """Return whether `< FILE`, `<<< WORD` or a dup replaces a heredoc on stdin.""" - return any(fd in ("", "0") and op not in OUTPUTS for fd, op, _ in redirects) - - def track_writes(words: list[str], stdin: str | None, directory: Path, written: dict[str, str | None]) -> str | None: """Record what one simple command leaves in the files it redirects to, and @@ -859,7 +1005,15 @@ def track_writes(words: list[str], stdin: str | None, directory: Path, operands, redirects = redirections(words) name = os.path.basename(words[0]) if words else "" copies = name == "tee" or (name == "cat" and not operands) - content = stdin if copies and not stdin_redirected(redirects) else None + content = stdin if copies else None + if (name == "echo" and len(operands) == 1 + and not operands[0].startswith("-") and "\\" not in operands[0]): + content = " ".join(operands) + "\n" + elif name == "printf" and operands: + if operands[0] in ("%s", "%s\\n"): + content = ("\n" if operands[0].endswith("n") else "").join(operands[1:]) + if operands[0].endswith("n"): + content += "\n" files = [(fd, op, target) for fd, op, target in redirects if op in OUTPUTS and not (op == ">&" and (target.isdigit() or target == "-"))] stdout_files = [target for fd, _, target in files if fd in ("", "1", "&")] @@ -904,7 +1058,7 @@ def option_value(words: list[str], index: int, long: str, short: str) -> tuple[s def gh_create_help(words: list[str]) -> bool: """Match only bare create help, optionally with one owner/repo selector.""" if (len(words) < 4 or os.path.basename(words[0]) != "gh" - or words[1] not in ("issue", "pr") or words[2] != "create"): + or words[1] not in ("issue", "pr") or words[2] not in ("create", "new")): return False args = words[3:] repo = None @@ -924,26 +1078,51 @@ def gh_create_help(words: list[str]) -> bool: ) +def executable_index(words: list[str]) -> int | None: + """Locate a literal executable after ordinary assignment/exec prefixes.""" + index = 0 + while index < len(words): + word = words[index] + if re.match(r"[A-Za-z_][A-Za-z0-9_]*=", word): + index += 1 + continue + name = os.path.basename(word) + if name not in ("env", "command", "exec"): + return index + index += 1 + while index < len(words) and words[index].startswith("-"): + flag = words[index] + if (name == "command" and flag in ("-v", "-V")) or flag in ("-S", "--split-string"): + return None + index += 2 if flag in ("-u", "--unset", "-C", "--chdir", "-a") else 1 + return None + + +def gh_positionals(words: list[str]) -> list[str]: + """Read gh positional words without mistaking option values for verbs.""" + rest: list[str] = [] + skip = False + operands, _ = redirections(words) + for word in operands: + if skip: + skip = False + elif word == "--": + break + elif word in CREATE_VALUE_FLAGS: + skip = True + elif not word.startswith("-"): + rest.append(word) + return rest + + def gh_create(words: list[str], *, allow_help: bool = False) -> str | None: - """Return `issue` or `pr` when words run `gh issue|pr create`.""" + """Return `issue` or `pr` when words run `gh issue|pr create|new`.""" if allow_help and gh_create_help(words): return None - for index, word in enumerate(words): - if os.path.basename(word) != "gh": - continue - rest: list[str] = [] - skip = False - operands, _ = redirections(words[index:]) - for word in operands: - if skip: - skip = False - elif word == "--": - break - elif word in VALUE_FLAGS: - skip = True - elif not word.startswith("-"): - rest.append(word) - if len(rest) >= 2 and rest[0] in ("issue", "pr") and rest[1] == "create": + index = executable_index(words) + if index is not None and os.path.basename(words[index]) == "gh": + rest = gh_positionals(words[index:]) + if len(rest) >= 2 and rest[0] in ("issue", "pr") and rest[1] in ("create", "new"): return rest[0] return None @@ -976,28 +1155,29 @@ def create_body(words: list[str], stdin: str | None, directory: Path, written: dict[str, str | None]) -> tuple[str | None, str]: """Return the body a `gh issue|pr create` call supplies, if the hook can see it, and a note when a `--body-file` cannot be read.""" - body = None + sources: list[tuple[bool, str | None]] = [] index = 0 while index < len(words): value, used = option_value(words, index, "--body-file", "-F") if used: - if value is None: - return None, "" - return body_file(value, stdin, directory, written) - value, used = option_value(words, index, "--body", "-b") - if used: - body = value - index += used - continue - index += 1 - return body, "" + sources.append((True, value)) + else: + value, used = option_value(words, index, "--body", "-b") + if used: + sources.append((False, value)) + index += used or 1 + if len(sources) > 1: + return None, "Repeated --body/--body-file options are not supported; pass one body source. " + if not sources or sources[0][1] is None or SUBSTITUTION in sources[0][1]: + return None, "" + file, value = sources[0] + return body_file(value, stdin, directory, written) if file else (value, "") def gh_api_create(words: list[str]) -> str | None: """Return `issue` or `pr` when words POST a new Issue or PR through `gh api`.""" - for index, word in enumerate(words): - if os.path.basename(word) != "gh": - continue + index = executable_index(words) + if index is not None and os.path.basename(words[index]) == "gh": rest = words[index + 1:] start = 0 while start < len(rest) and rest[start].startswith("-"): @@ -1084,43 +1264,179 @@ def command_strings(event_input: dict[str, object]) -> list[str]: return [text for text in strings if not patch_data(text)] -def elon_gate(event_input: dict[str, object], cwd: str) -> str | None: - """Return a block message when a `gh` call creating an Issue or PR lacks the Elon record.""" - commands = command_strings(event_input) - for text in commands: - if "gh" not in text or ("create" not in text and "api" not in text): +SHELLS = {"sh", "bash", "zsh", "dash", "ksh"} + + +def resolved_stdin(words: list[str], stdin: str | None) -> str | None: + for fd, op, target in redirections(["__redirects__", *words])[1]: + if fd in ("", "0") and op not in OUTPUTS: + stdin = target + "\n" if op == "<<<" and SUBSTITUTION not in target else None + return stdin + + +def command_positions(words: list[str]) -> list[tuple[int, list[str]]]: + """One conservative word-position scan for commands and their producers. + + Whole quoted examples stay one word. Only known read-only consumers + exempt their argument data; arbitrary prefixes never hide a command. + """ + executable = executable_index(words) + if executable is not None: + name = os.path.basename(words[executable]) + args = gh_positionals(words[executable:]) if name == "gh" else [] + if name in ("rg", "grep") or (len(args) >= 2 + and args[0] in ("issue", "pr") + and args[1] in ("list", "view", "search")): + return [] + return [(index, words[index:]) for index in range(len(words))] + + +def shell_script(args: list[str], stdin: str | None) -> str | None: + """Find a literal -c script or stdin after consuming shell option values.""" + index = 0 + use_stdin = False + while index < len(args): + arg = args[index] + if arg in ("-o", "-O", "+o", "+O", "--init-file", "--rcfile"): + index += 2 continue - directory = Path(cwd) - # What the command itself leaves in files before `gh` runs; None is unknown. - written: dict[str, str | None] = {} - stdout: str | None = None - segments = shell_segments(text) - # Help is exempt only for one plain invocation whose raw characters - # are all explicitly allowed. Unknown characters never authorize help. - allow_help = (len(commands) == 1 and len(segments) == 1 - and re.fullmatch(r"[-A-Za-z0-9 _./=:,@+]+", text) is not None) - for segment, heredoc, piped in segments: - try: - words = shlex.split(segment, posix=True) - except ValueError: - words = segment.split() - allow_help = False - # A heredoc is stdin; otherwise a pipe from a `cat <<'EOF'` may be. - stdin = heredoc if heredoc is not None else (stdout if piped else None) - if len(words) >= 2 and words[0] == "cd": - target = Path(words[1]).expanduser() - directory = target if target.is_absolute() else directory / target - stdout = None + if arg == "--": + index += 1 + break + if not arg.startswith(("-", "+")): + break + # Bundled short flags such as -euo still give -o a separate value. + if not arg.startswith("--") and any(flag in arg[1:] for flag in "oO"): + index += 1 + if arg.startswith("-") and not arg.startswith("--") and "c" in arg: + index += 1 + if index < len(args) and args[index] == "--": + index += 1 + return args[index] if index < len(args) else None + if arg.startswith("-") and not arg.startswith("--") and "s" in arg: + use_stdin = True + index += 1 + if use_stdin or index >= len(args) or args[index] == "/dev/stdin": + return stdin + return None + + +def shell_words(text: str) -> list[str]: + """Use one word splitter for commands and visible unknown script input.""" + try: + return shlex.split(text, posix=True) + except ValueError: + return text.split() + + +class ShellInspection: + """Whole-command evidence left after checking each executable create.""" + def __init__(self) -> None: + self.shell_input = False + self.unchecked: list[str] = [] + + +def visible_script_create(text: str) -> bool: + """Classify visible gh words with the same option arities as bare creates.""" + text = text.replace("\\\n", "") + # Unknown producers may turn an escape into a word boundary (zsh echo's + # \n is a required case). Do not decode output; conservatively check both + # the literal words and possible boundaries with the same gh classifier. + quoted = text.translate(str.maketrans({"'": " ", '"': " "})) + for visible in (text, quoted, re.sub(r"\\.", " ", quoted)): + words = shell_words(visible) + if any(os.path.basename(word) == "gh" and gh_create(words[index:]) is not None + for index, word in enumerate(words)): + return True + return False + + +def check_shell_script(segments: list[ShellCommand], directory: Path, + written: dict[str, str | None], inspection: ShellInspection, + *, allow_help: bool = False) -> str | None: + stdout: str | None = None + for segment in segments: + literal_data: list[tuple[int, int]] = [] + for child in segment.children: + begin = len(inspection.unchecked) + blocked = check_shell_script(child, directory, written, inspection) + if blocked: + return blocked + if literal_cat(child) is not None: + literal_data.append((begin, len(inspection.unchecked))) + words = shell_words(segment.text) + if not words: + stdout = None + continue + words = [segment.literals.get(word, word) for word in words] + words = ["--body=" + segment.literals[word[7:]] + if word.startswith("--body=") and word[7:] in segment.literals else word + for word in words] + own_stdin = segment.stdin_visible is not None + stdin = segment.stdin if own_stdin else (stdout if segment.piped else None) + stdin = resolved_stdin(words, stdin) + if words[0] == "cd" and len(words) >= 2: + target = Path(words[1]).expanduser() + directory = target if target.is_absolute() else directory / target + stdout = None + inspection.unchecked.extend([" ".join(words), *segment.heredoc_text]) + continue + positions = command_positions(words) + executable = executable_index(words) + # Unclassified wrappers still make redirected file contents unknown. + producer = words[executable:] if executable is not None else words + stdout = track_writes(producer, stdin, directory, written) + residual = words.copy() + data = segment.heredoc_text.copy() + found = [] + for index, candidate in positions: + name = os.path.basename(candidate[0]) + if name in SHELLS: + args = redirections(candidate)[0] + literal = shell_script(args, None) + literal_c = literal is not None and SUBSTITUTION not in literal + if not literal_c: + inspection.shell_input = True + # Script pipes and compound stdin are deliberately unmodeled. + # Only the actual executable may consume known direct input. + direct = resolved_stdin(words, segment.stdin if own_stdin else None) + script = literal if literal_c else (shell_script(args, direct) + if index == executable else None) + if script is not None and SUBSTITUTION not in script: + blocked = check_shell_script(shell_segments(script), directory, written, inspection) + if blocked: + return blocked + # The parsed script contributes its own unchecked words; + # do not count the original quoted copy a second time. + target = script if literal_c else script.removesuffix("\n") + for position, word in enumerate(residual): + prefix = word[:-len(target)] if target and word.endswith(target) else "" + if word == target or (not literal_c and re.fullmatch(r"\d*<<<", prefix)): + residual[position] = "" + break + if not literal_c: + for position, content in enumerate(data): + if content.replace("\\\n", "") == script.replace("\\\n", ""): + del data[position] + break + if name != "gh": continue - stdout = track_writes(words, stdin, directory, written) - kind = gh_create(words, allow_help=allow_help) + kind = gh_create(candidate, allow_help=allow_help and index == 0) api = kind is None if api: - kind = gh_api_create(words) - if kind is None: - continue - if stdin_redirected(redirections(words)[1]): - stdin = None + kind = gh_api_create(candidate) + if kind is not None: + found.append((index, candidate, kind, api)) + if not found: + inspection.unchecked.extend([" ".join(residual), *data]) + continue + # Every found suffix below is checked independently. Preserve prefix + # data (e.g. awk's program) instead of authorizing an entire segment. + for begin, end in reversed(literal_data): + del inspection.unchecked[begin:end] + first_create = min(entry[0] for entry in found) + inspection.unchecked.append(" ".join(residual[:first_create])) + for _, words, kind, api in found: read = api_body if api else create_body body, note = read(words, stdin, directory, written) missing = [f for f in ELON_FIELDS if not field_value(body or "", f)] @@ -1145,6 +1461,28 @@ def elon_gate(event_input: dict[str, object], cwd: str) -> str | None: ) return None + +def elon_gate(event_input: dict[str, object], cwd: str) -> str | None: + """Record-check each visible create; literals and heredoc bodies stay data.""" + commands = command_strings(event_input) + for text in commands: + segments = shell_segments(text) + allow_help = (len(commands) == 1 and len(segments) == 1 + and not segments[0].children + and re.fullmatch(r"[-A-Za-z0-9 _./=:,@+]+", text) is not None) + inspection = ShellInspection() + blocked = check_shell_script(segments, Path(cwd), {}, inspection, allow_help=allow_help) + if blocked: + return blocked + if inspection.shell_input and visible_script_create("\n".join(inspection.unchecked)): + return ("AgentsMD blocked shell-fed gh create: a shell executes input " + "outside a fully parsed literal -c script, and the whole command " + "contains a visible gh issue|pr create|new that has not received " + "its own record check. Use a literal -c script or direct literal " + "heredoc/here-string with a verifiable record.") + return None + + def report_input_error(detail: str) -> int: json.dump( { diff --git a/tests/test_elon_gate.py b/tests/test_elon_gate.py index a8867db..84968d0 100644 --- a/tests/test_elon_gate.py +++ b/tests/test_elon_gate.py @@ -10,6 +10,7 @@ import subprocess import tempfile import unittest +from unittest import mock from pathlib import Path ROOT = Path(__file__).resolve().parents[1] @@ -90,6 +91,732 @@ def test_other_tool_calls_pass_silently(self) -> None: result = self.hook(tool_input) self.assertEqual((result.returncode, result.stdout, result.stderr), (0, "", "")) + def bash_argv(self, command: str, *, succeeds: bool = True, capture_stdin: bool = False) -> list[list[str]]: + """Run real Bash with only gh replaced at the external-effect boundary.""" + binary = self.cwd / "fake-bin" + binary.mkdir(exist_ok=True) + gh = binary / "gh" + gh.write_text( + "#!/usr/bin/env python3\n" + "import json, os, sys\n" + "with open(os.environ['GH_ARGV_LOG'], 'a') as log:\n" + " log.write(json.dumps(sys.argv[1:]) + '\\n')\n" + "if os.environ.get('GH_STDIN_LOG'):\n" + " with open(os.environ['GH_STDIN_LOG'], 'w') as log:\n" + " log.write(sys.stdin.read())\n", + encoding="utf-8", + ) + gh.chmod(0o755) + log = self.cwd / "argv.jsonl" + log.write_text("", encoding="utf-8") + pinned_path = str(binary) + os.pathsep + self.env["PATH"] + guard = ("export PATH=" + shlex.quote(pinned_path) + "; " + "[ \"$(command -v gh)\" = " + shlex.quote(str(gh)) + " ] || exit 97; ") + if command.startswith("zsh "): + args = shlex.split(command) + args[-1] = guard + args[-1] + command = shlex.join(args) + command = guard + command + result = subprocess.run( + ["bash", "--noprofile", "--norc", "-c", command], cwd=self.cwd, + env={**self.env, "PATH": str(binary) + os.pathsep + self.env["PATH"], + "GH_ARGV_LOG": str(log), + "GH_STDIN_LOG": str(self.cwd / "stdin.log") if capture_stdin else ""}, capture_output=True, text=True, timeout=10, + ) + if succeeds: + self.assertEqual(result.returncode, 0, result.stderr) + else: + self.assertNotEqual(result.returncode, 0, result.stderr) + return [json.loads(line) for line in log.read_text().splitlines()] + + def test_new_aliases_are_gated_and_accept_a_record(self) -> None: + for kind in ("issue", "pr"): + command = f"gh {kind} new --title x --body hello" + with self.subTest(kind=kind): + self.assertEqual(self.bash_argv(command), + [[kind, "new", "--title", "x", "--body", "hello"]]) + self.assertBlocked(self.hook({"command": command})) + self.assertEqual(self.hook({"command": f"gh {kind} new --body " + + shlex.quote(RECORD)}).returncode, 0) + + @staticmethod + def wrappers(call: str) -> tuple[str, ...]: + return (f"URL=$({call})", f'URL="$({call})"', f"URL=`{call}`", + f"({call})", "sh -c " + shlex.quote(call), + "bash -c " + shlex.quote(call), "zsh -f -c " + shlex.quote(call)) + + def test_confident_wrappers_check_their_own_body(self) -> None: + for kind in ("issue", "pr"): + for verb in ("create", "new"): + for complete in (True, False): + record = RECORD if complete else "hello\n" + (self.cwd / "f.md").write_text(record, encoding="utf-8") + for args in ("--body-file f.md", "-F f.md", + "--body " + shlex.quote(record), + "--body-file - <<'EOF'\n" + record + "EOF\n"): + call = f"gh {kind} {verb} --title t " + args + for command in self.wrappers(call): + with self.subTest(command=command, complete=complete): + if not command.startswith("zsh") or shutil.which("zsh"): + argv = self.bash_argv(command) + self.assertEqual(argv[0][:4], [kind, verb, "--title", "t"]) + result = self.hook({"command": command}) + if complete: + self.assertEqual(result.returncode, 0, result.stderr) + else: + self.assertBlocked(result) + + def test_plain_commit_echo_and_printf_mentions_are_data(self) -> None: + for command in (r'printf %s "gh issue create"', + r'printf "%s\n" "gh issue create"', + r'git commit -m "mention \"gh issue create\""', + r'echo "\$(gh issue create --body hello)"', + r'printf "%s" "\`gh issue create --body hello\`"'): + with self.subTest(command=command): + self.assertEqual(self.hook({"command": command}).returncode, 0) + if command.startswith(("echo ", "printf ")): + self.assertEqual(self.bash_argv(command), []) + for data in ("mention gh issue create", "mention gh pr new", + "$(gh issue create --body hello)", "`gh pr create --body hello`", + "sh -c 'gh issue create'", "{gh,} issue create", "gh $'issue' create"): + commands = ("git commit -m " + shlex.quote(data), + "echo " + shlex.quote(data), "printf '%s' " + shlex.quote(data)) + if "$" not in data and "`" not in data and '"' not in data: + commands += ('git commit -m "' + data + '"', 'echo "' + data + '"', + 'printf "%s" "' + data + '"') + for command in commands: + with self.subTest(command=command): + self.assertEqual(self.hook({"command": command}).returncode, 0) + if command.startswith(("echo ", "printf ")): + self.assertEqual(self.bash_argv(command), []) + if data.startswith("mention"): + for command in (f'git commit -m "{data}"', f'echo "{data}"', + f'printf "%s" "{data}"'): + self.assertEqual(self.hook({"command": command}).returncode, 0) + if command.startswith(("echo ", "printf ")): + self.assertEqual(self.bash_argv(command), []) + + def test_nested_commands_check_each_visible_create(self) -> None: + (self.cwd / "f.md").write_text(RECORD, encoding="utf-8") + for call in ('echo $(gh pr create --body-file f.md)', + '(gh pr create --body-file f.md)'): + for command in self.wrappers(call): + self.assertEqual(self.hook({"command": command}).returncode, 0) + for command in ('git commit -m "$(gh issue create --body hello)"', + 'echo "$(gh issue create --body hello)"', + 'gh pr create --body-file f.md --title $(gh issue create --body hello)'): + self.assertBlocked(self.hook({"command": command})) + + def test_outer_blank_lines_preserve_wrappers_and_literal_data(self) -> None: + for command in ('git commit -m "gh issue create"', + 'echo "gh issue create"', 'printf "%s\\n" "gh issue create"'): + self.assertEqual(self.hook({"command": "\n\t" + command + "\n"}).returncode, 0) + for complete in (True, False): + (self.cwd / "f.md").write_text(RECORD if complete else "hello", encoding="utf-8") + command = "\nURL=$(gh pr create --title t --body-file f.md)\n" + self.assertEqual(self.bash_argv(command), [["pr", "create", "--title", "t", "--body-file", "f.md"]]) + result = self.hook({"command": command}) + if complete: + self.assertEqual(result.returncode, 0, result.stderr) + else: + self.assertBlocked(result) + + def test_data_and_compound_wrappers_pass(self) -> None: + (self.cwd / "f.md").write_text(RECORD, encoding="utf-8") + for command in ('echo "$HOME gh issue create"', + "echo 'gh issue create' > log", + "echo 'gh issue create'; echo done", + "git -C repo commit -m 'gh issue create'", + "bash -lc 'gh pr create --body-file f.md'", + "URL=$(gh\tpr create --body-file f.md)", + "URL=$(gh pr create --body-file - < None: + for kind in ("issue", "pr"): + for verb in ("create", "new"): + call = f"gh {kind} {verb} --title x --body hello" + forms = ( + f"({call})", f"( ( {call} ) )", f"echo $({call})", + f'echo "$({call})"', f"echo `{call}`", f'echo "`{call}`"', + f"echo $(echo $({call}))", f"(echo $({call}))", + f"echo $( ({call}) )", f"sh -c '{call}'", + f"bash -c '{call}'", + f"s'h' -c '{call}'", + f"g'h' '{kind}' '{verb}' --title x --body hello", + f'gh "{kind}" "{verb}" --title x --body hello', + f'gh "{kind[:1]}\\\n{kind[1:]}" {verb} --title x --body hello', + f'gh {kind} "{verb[:1]}\\\n{verb[1:]}" --title x --body hello', + ) + for command in forms: + with self.subTest(command=command): + self.assertIn([kind, verb, "--title", "x", "--body", "hello"], + self.bash_argv(command)) + self.assertBlocked(self.hook({"command": command})) + + def test_zsh_c_is_gated(self) -> None: + command = "zsh -f -c 'gh issue create --body hello'" + self.assertBlocked(self.hook({"command": command})) + if shutil.which("zsh"): + self.assertEqual(self.bash_argv(command), [["issue", "create", "--body", "hello"]]) + + def test_250_bypasses_still_block_with_bash_argv_proof(self) -> None: + for kind in ("issue", "pr"): + for verb in ("create", "new"): + call = f"gh {kind} {verb} --title x --body hello" + cases = ( + (call + " # --help", [kind, verb, "--title", "x", "--body", "hello"]), + (call + " \\\n# --help", [kind, verb, "--title", "x", "--body", "hello"]), + (f"gh {kind} {verb} --help <<< foo\n{call}", + [kind, verb, "--title", "x", "--body", "hello"]), + (f"gh {kind} {verb} -dt --help -b x", [kind, verb, "-dt", "--help", "-b", "x"]), + (f"gh {kind} {verb} -fb --help", [kind, verb, "-fb", "--help"]), + (f"gh {kind} {verb} --title '--help'", [kind, verb, "--title", "--help"]), + ) + for command, argv in cases: + with self.subTest(command=command): + self.assertIn(argv, self.bash_argv(command)) + self.assertBlocked(self.hook({"command": command})) + # Bash's exec interface rejects NUL before execution. It must still be + # blocked by the hook, rather than receiving the bare-help exemption. + self.assertBlocked(self.hook({"command": "gh issue create --body hello\x00 --help"})) + + def test_each_visible_create_needs_its_own_record(self) -> None: + for call in ("(gh issue create --body hello)", + "$(gh issue new --body hello)", + "`gh pr create --body hello`", + "sh -c 'gh pr new --body hello'"): + with self.subTest(call=call): + command = "gh issue create --body " + shlex.quote(RECORD) + "; " + call + self.assertBlocked(self.hook({"command": command})) + command = "(gh pr create --body " + shlex.quote(RECORD) + ")" + self.assertEqual(self.hook({"command": command}).returncode, 0) + + def test_literal_create_mentions_remain_data_in_plain_commands(self) -> None: + for data in ("gh issue create", "gh pr new"): + for command in ("cat <<'EOF'\n" + data + "\nEOF", + "gh issue create --body " + shlex.quote(RECORD + data)): + with self.subTest(command=command): + self.assertEqual(self.hook({"command": command}).returncode, 0) + self.assertEqual(self.hook({"command": "printf '%s' " + shlex.quote(data)}).returncode, 0) + + def test_separate_body_file_avoids_dynamic_data_false_positive(self) -> None: + data = RECORD + "\nExample: $(gh issue create) and sh -c 'gh pr new'\n" + (self.cwd / "record.md").write_text(data, encoding="utf-8") + self.assertEqual(self.hook({"command": "gh issue create --body-file record.md"}).returncode, 0) + self.assertEqual(self.hook({"command": "gh pr new --body-file record.md"}).returncode, 0) + + def test_wrapped_help_never_uses_the_bare_help_exemption(self) -> None: + for kind in ("issue", "pr"): + for verb in ("create", "new"): + call = f"gh {kind} {verb} --help" + for command in (f"({call})", f"echo $({call})", f"echo `{call}`", + f"sh -c '{call}'"): + with self.subTest(command=command): + self.assertIn([kind, verb, "--help"], self.bash_argv(command)) + self.assertBlocked(self.hook({"command": command})) + + def test_dynamic_looking_data_stays_literal(self) -> None: + for data in ("$(gh issue create)", "`gh pr new`", "{gh,} issue create", + "gh $'issue' create", "sh -c 'gh issue create'"): + for command in ("printf '%s' " + shlex.quote(data), + "cat <<'EOF'\n" + data + "\nEOF", + "gh issue create --body " + shlex.quote(RECORD + data)): + with self.subTest(command=command): + result = self.hook({"command": command}) + self.assertEqual(result.returncode, 0, result.stderr) + + def test_shell_fed_scripts_record_check_and_real_argv(self) -> None: + for complete in (False, True): + body = RECORD if complete else "hello" + for kind, verb in (("issue", "create"), ("pr", "new")): + call = f"gh {kind} {verb} --body " + shlex.quote(body) + commands = [] + for shell in ("sh", "bash"): + for flags in ("", " -s"): + for operator in ("< None: + body = RECORD + "Examples: `gh issue create` $HOME $(gh pr new) (parentheses)\n" + for command in ("gh pr create --body-file - <<'EOF'\n" + body + "EOF\n", + "URL=$(gh pr create --body-file - <<'EOF'\n" + body + "EOF\n) && echo \"$URL\""): + self.assertEqual(self.bash_argv(command), [["pr", "create", "--body-file", "-"]]) + self.assertEqual(self.hook({"command": command}).returncode, 0) + + def test_compound_url_capture_checks_record(self) -> None: + for complete in (False, True): + (self.cwd / "f.md").write_text(RECORD if complete else "hello") + command = 'URL=$(gh pr create --title t --body-file f.md) && echo "$URL"' + self.assertEqual(self.bash_argv(command), [["pr", "create", "--title", "t", "--body-file", "f.md"]]) + self.assertEqual(self.hook({"command": command}).returncode, 0 if complete else 2) + + def test_read_only_searches_are_data(self) -> None: + for command in ("rg gh issue create", "grep 'gh issue create' README.md", + "gh issue list --search 'gh issue create'", + "gh pr list --search 'gh pr create'", + "gh issue view 1 --json body", "gh pr search 'gh issue create'"): + self.assertEqual(self.hook({"command": command}).returncode, 0, command) + + def test_repeated_body_options_block(self) -> None: + (self.cwd / "good.md").write_text(RECORD) + (self.cwd / "bad.md").write_text("hello") + for options in ("--body-file good.md --body-file bad.md", + "--body-file bad.md --body-file good.md", + "--body " + shlex.quote(RECORD) + " --body hello", + "--body-file good.md --body hello"): + command = "gh pr create " + options + self.assertEqual(self.bash_argv(command), [["pr", "create"] + shlex.split(options)]) + self.assertBlocked(self.hook({"command": command})) + + def test_dynamic_shell_limits_are_not_inferred_from_raw_words(self) -> None: + (self.cwd / "earlier.sh").write_text("gh issue create\n") + for command in ("{gh,} issue create", "gh${IFS}issue${IFS}create", + "cmd=gh; $cmd issue create", "gh issue ${verb:-create}", + "eval '$SCRIPT'", + "alias c='gh issue create'; c", + "echo 'issue create --body hello' | xargs gh", + "sh earlier.sh"): + self.assertEqual(self.hook({"command": command}).returncode, 0, command) + self.assertBlocked(self.hook({"command": "cat > later.sh <<'EOF'\ngh issue create\nEOF\nsh later.sh"})) + + def test_literal_scripts_for_all_named_shells(self) -> None: + for shell in ("sh", "bash", "zsh", "dash", "ksh"): + for complete in (False, True): + call = "gh pr new --body " + shlex.quote(RECORD if complete else "hello") + command = shell + " -c " + shlex.quote(call) + self.assertEqual(self.hook({"command": command}).returncode, 0 if complete else 2) + + @staticmethod + def prefixed_creates(call: str) -> tuple[str, ...]: + return tuple(prefix + call for prefix in + ("time ", "nohup ", "nice ", "timeout 2 ", "sudo ", + "echo x | xargs ", "find . -maxdepth 0 -exec ", "! ", "{ ", "then ", "do ", "else ", + "if ", "while ", "until ", "case x in x) ", "unknown-prefix ")) + + def test_prefix_positions_cannot_hide_creates(self) -> None: + for kind in ("issue", "pr"): + for verb in ("create", "new"): + for command in self.prefixed_creates(f"gh {kind} {verb} --body hello"): + with self.subTest(command=command): + self.assertBlocked(self.hook({"command": command})) + for command in self.prefixed_creates(f"gh {kind} {verb} --body " + shlex.quote(RECORD)): + self.assertEqual(self.hook({"command": command}).returncode, 0) + for command in ("time gh issue create --body hello", + "nohup gh issue create --body hello", + "nice gh issue create --body hello", + "timeout 2 gh issue create --body hello", + "{ gh issue create --body hello; }", + "if true; then gh issue create --body hello; fi", + "for x in a; do gh issue create --body hello; done", + "case x in x) gh issue create --body hello;; esac", + "case x in x)gh issue create --body hello;; esac", + "echo x | xargs gh issue create --body hello", + "i=0; while test $i = 0; do i=1; gh issue create --body hello; done", + "i=0; until test $i = 1; do i=1; gh issue create --body hello; done", + "while gh issue create --body hello; do break; done", + "until gh issue create --body hello; do break; done", + "find . -maxdepth 0 -exec gh issue create --body hello \\;"): + prefix = shlex.split(command)[0] + if prefix in ("{", "if", "for", "case", "while", "until") or "=" in prefix or shutil.which(prefix): + self.assertEqual(self.bash_argv(command)[0][:2], ["issue", "create"]) + self.assertBlocked(self.hook({"command": command})) + + def test_prefix_oracle_keeps_hooks_when_timeout_is_unavailable(self) -> None: + real_which, real_oracle = shutil.which, self.bash_argv + observed = [] + def oracle(command: str) -> list[list[str]]: + self.assertFalse(command.startswith("timeout ")) + observed.append(command) + return real_oracle(command) + with mock.patch("shutil.which", side_effect=lambda name: None if name == "timeout" else real_which(name)): + with mock.patch.object(self, "bash_argv", side_effect=oracle): + self.test_prefix_positions_cannot_hide_creates() + self.assertTrue(observed) + + def test_literal_cat_heredoc_substitution_is_a_body(self) -> None: + for complete in (False, True): + body = RECORD if complete else "hello\n" + for delimiter in ("EOF", "'EOF'", '\"EOF\"'): + for quotes in ('"', ''): + command = ('gh pr create --title x --body ' + quotes + + '$(cat <<' + delimiter + '\n' + body + 'EOF\n)' + quotes) + with self.subTest(command=command): + self.assertEqual(self.hook({"command": command}).returncode, + 0 if complete and quotes else 2) + if quotes: + self.assertEqual(self.bash_argv(command), + [["pr", "create", "--title", "x", "--body", body.rstrip('\n')]]) + + def test_shell_script_arguments_and_literal_pipes(self) -> None: + for complete in (False, True): + body = RECORD if complete else "hello" + call = "gh issue create --body " + shlex.quote(body) + commands = [prefix + " <<'EOF'\n" + call + "\nEOF\n" + for prefix in ("bash -s foo", "bash -s -- foo", "bash /dev/stdin")] + commands += ["bash -c -- " + shlex.quote(call)] + for shell in ("sh", "bash"): + commands += ["echo " + shlex.quote(call) + " | " + shell, + "printf '%s' " + shlex.quote(call) + " | " + shell, + "printf '%s\\n' " + shlex.quote(call) + " | " + shell] + for command in commands: + with self.subTest(command=command): + self.assertEqual(self.bash_argv(command), [["issue", "create", "--body", body]]) + expected = 0 if complete and not command.startswith(("echo ", "printf ")) else 2 + self.assertEqual(self.hook({"command": command}).returncode, expected) + + def test_array_literals_are_data_but_substitutions_execute(self) -> None: + command = 'arr=(gh issue create); echo "${arr[@]}"' + self.assertEqual(self.bash_argv(command), []) + self.assertEqual(self.hook({"command": command}).returncode, 0) + self.assertBlocked(self.hook({"command": 'arr=($(gh issue create --body hello))'})) + + @staticmethod + def prefixed_shell_scripts(call: str) -> tuple[str, ...]: + script = shlex.quote(call) + return ("if true; then bash -c " + script + "; fi", + "time bash -c " + script, "nohup bash -c " + script, + "nice bash -c " + script, "sudo -n bash -c " + script, + "{ bash -c " + script + "; }", "! bash -c " + script + "; true", + ">/dev/null bash -c " + script, + "<<< " + script + " bash", + "find . -maxdepth 0 -exec sh -c " + script + " \\;", + "nice sh <<'EOF'\n" + call + "\nEOF\n", + "if true; then echo " + script + " | sh; fi") + + def test_prefixed_shells_use_the_same_positions_as_gh(self) -> None: + for complete in (False, True): + (self.cwd / "f.md").write_text(RECORD if complete else "hello") + call = "gh issue create --body-file f.md" + for command in self.prefixed_shell_scripts(call): + with self.subTest(command=command, complete=complete): + prefix = shlex.split(command)[0] + if prefix != "sudo" and (prefix in ("if", "{", "!", ">/dev/null", "<<<") or shutil.which(prefix)): + self.assertEqual(self.bash_argv(command), [["issue", "create", "--body-file", "f.md"]]) + conservative = command.startswith(("nice sh ", "<<< ")) or " | sh" in command + self.assertEqual(self.hook({"command": command}).returncode, + 0 if complete and not conservative else 2) + + def test_function_bodies_are_scanned_like_subshells(self) -> None: + for complete in (False, True): + (self.cwd / "f.md").write_text(RECORD if complete else "hello") + command = "f() { gh issue create --body-file f.md; }; f" + self.assertEqual(self.bash_argv(command), [["issue", "create", "--body-file", "f.md"]]) + self.assertEqual(self.hook({"command": command}).returncode, 0 if complete else 2) + self.assertBlocked(self.hook({"command": "f() { gh issue create; }; f"})) + + def test_unquoted_heredoc_substitutions_execute_but_quoted_are_data(self) -> None: + for complete in (False, True): + (self.cwd / "f.md").write_text(RECORD if complete else "hello") + call = "gh pr new --body-file f.md" + for delimiter in ("EOF", "'EOF'", '\"EOF\"'): + for body in ("$(" + call + ")", "`" + call + "`", "'$(" + call + ")'", + '\"$(' + call + ')\"'): + command = "cat <<" + delimiter + "\n" + body + "\nEOF\n" + with self.subTest(command=command, complete=complete): + unquoted = delimiter == "EOF" + self.assertEqual(self.bash_argv(command), + [["pr", "new", "--body-file", "f.md"]] if unquoted else []) + self.assertEqual(self.hook({"command": command}).returncode, + 2 if unquoted and not complete else 0) + for body in ("gh issue create", r"\$(gh issue create)", r"\`gh issue create\`"): + command = "cat < tuple[str, ...]: + return tuple("bash " + flags + " < None: + for complete in (False, True): + (self.cwd / "f.md").write_text(RECORD if complete else "hello") + commands = list(self.shell_option_scripts("gh issue create --body-file f.md")) + commands.append("bash -eco pipefail 'gh issue create --body-file f.md'") + for command in commands: + with self.subTest(command=command, complete=complete): + self.assertEqual(self.bash_argv(command), [["issue", "create", "--body-file", "f.md"]]) + self.assertEqual(self.hook({"command": command}).returncode, 0 if complete else 2) + + def test_arithmetic_is_data_but_nested_substitutions_execute(self) -> None: + for command in ("echo $((gh issue create))", "echo $(( gh issue create ))"): + with self.subTest(command=command): + self.assertEqual(self.bash_argv(command, succeeds=False), []) + self.assertEqual(self.hook({"command": command}).returncode, 0) + command = "echo $(( (1+2) * (4+5) ))" + self.assertEqual(self.bash_argv(command), []) + self.assertEqual(self.hook({"command": command}).returncode, 0) + for complete in (False, True): + (self.cwd / "f.md").write_text(RECORD if complete else "hello") + command = "echo $((0$(gh issue create --body-file f.md)))" + self.assertEqual(self.bash_argv(command), [["issue", "create", "--body-file", "f.md"]]) + self.assertEqual(self.hook({"command": command}).returncode, 0 if complete else 2) + + @staticmethod + def round_four_commands() -> list[str]: + return [ + "echo " + shlex.quote("x\\c\n" + RECORD) + " | gh issue create -t x -F -", + "gh issue --title T create -b y", + "gh pr -B main new -t x -b y", + "cat < None: + for flag in ("", "-n ", "-e ", "-E "): + body = "x\\c\n" + RECORD if not flag else RECORD + command = "echo " + flag + shlex.quote(body) + " | gh issue create -t x -F -" + with self.subTest(command=command): + self.assertBlocked(self.hook({"command": command})) + self.assertEqual(self.bash_argv(command), [["issue", "create", "-t", "x", "-F", "-"]]) + if shutil.which("zsh"): + command = "zsh -f -c " + shlex.quote(self.round_four_commands()[0]) + self.assertBlocked(self.hook({"command": command})) + self.assertEqual(self.bash_argv(command, capture_stdin=True), [["issue", "create", "-t", "x", "-F", "-"]]) + self.assertEqual((self.cwd / "stdin.log").read_text(), "x") + + def test_create_flags_before_verb_do_not_hide_creates(self) -> None: + for prefix in ("gh issue --title T", "gh pr -B main", "gh issue --title create"): + for verb in ("create", "new"): + for body, expected in (("hello", 2), (RECORD, 0)): + command = prefix + " " + verb + " -b " + shlex.quote(body) + with self.subTest(command=command): + self.assertEqual(self.bash_argv(command), [shlex.split(command)[1:]]) + self.assertEqual(self.hook({"command": command}).returncode, expected) + + def test_full_heredoc_delimiter_words(self) -> None: + for raw, delimiter in (("BODY-X", "BODY-X"), ("EOF.", "EOF."), + ("'PR-BODY'", "PR-BODY"), ('E"OF"', "EOF"), + (r"E\OF", "EOF"), ('"PR-BODY"', "PR-BODY")): + for body, expected in (("hello\n", 2), (RECORD, 0)): + for prefix in ("gh pr create --body-file -", "bash"): + content = body if prefix != "bash" else "gh pr new -b " + shlex.quote(body) + "\n" + command = prefix + " <<" + raw + "\n" + content + delimiter + "\n" + with self.subTest(command=command): + argv = (["pr", "create", "--body-file", "-"] if prefix != "bash" + else ["pr", "new", "-b", body]) + self.assertEqual(self.bash_argv(command), [argv]) + self.assertEqual(self.hook({"command": command}).returncode, expected) + command = "cat <<" + raw + "\ndata\n" + delimiter + "\ngh issue create -b y" + with self.subTest(command=command): + self.assertEqual(self.bash_argv(command), [["issue", "create", "-b", "y"]]) + self.assertBlocked(self.hook({"command": command})) + + @staticmethod + def unknown_producer_scripts(body: str) -> list[str]: + call = "gh issue create --body " + shlex.quote(body) + quoted = shlex.quote(call) + rows = ["echo " + flag + " " + quoted + " | sh" + for flag in ("-n", "-e", "-E", "-ne")] + rows += ["echo 'gh issue' " + shlex.quote("create --body " + shlex.quote(body)) + " | bash", + "echo " + shlex.quote("true\\n" + call) + " | bash", + "printf -- '%s\\n' " + quoted + " | sh", + "printf " + quoted + " | sh", + "printf " + shlex.quote(call + "\\n") + " | sh"] + rows += [prefix + " <<'EOF' | " + shell + "\n" + call + "\nEOF\n" + for prefix in ("cat -", "cat --", "cat -u", "cat -s", "grep issue", "head -1") + for shell in ("sh", "bash")] + rows += ["echo -n " + quoted + " | head -1 | sh", + "if true; then echo -n " + quoted + " | sh; fi", + "echo -n " + quoted + " | nice sh"] + return rows + + def test_unmodeled_producers_with_visible_creates_fail_closed(self) -> None: + for body in ("hello", RECORD): + for command in self.unknown_producer_scripts(body): + with self.subTest(command=command): + self.assertBlocked(self.hook({"command": command})) + # Backslash echo is shell-dependent; zsh covers it below. + if "true\\n" not in command: + truncated = body == RECORD and ("grep issue" in command or "head -1" in command) + if truncated: + self.assertEqual(self.bash_argv(command, succeeds=False), []) + else: + self.assertEqual(self.bash_argv(command), [["issue", "create", "--body", body]]) + for command in ("echo -n hello | sh", "grep foo file | sh"): + with self.subTest(command=command): + self.assertEqual(self.hook({"command": command}).returncode, 0) + if shutil.which("zsh"): + command = "zsh -f -c " + shlex.quote("echo " + shlex.quote("true\\ngh issue create --body hello") + " | bash") + self.assertBlocked(self.hook({"command": command})) + self.assertEqual(self.bash_argv(command), [["issue", "create", "--body", "hello"]]) + + @staticmethod + def unreadable_shell_scripts(body: str) -> list[tuple[str, list[str]]]: + call = "gh issue create -b " + shlex.quote(body) + return [ + ("echo -n " + shlex.quote("gh issue --title T create -b " + shlex.quote(body)) + " | sh", + ["issue", "--title", "T", "create", "-b", body]), + ("echo -n " + shlex.quote("gh -R o/r issue create -b " + shlex.quote(body)) + " | sh", + ["-R", "o/r", "issue", "create", "-b", body]), + ("cat -s <<'EOF' | sh\ngh issue \\\ncreate -b " + shlex.quote(body) + "\nEOF\n", + ["issue", "create", "-b", body]), + ('bash <<< "$(echo -n ' + shlex.quote(call) + ')"', ["issue", "create", "-b", body]), + ("bash < None: + for body in ("hello", RECORD): + for command, argv in self.unreadable_shell_scripts(body): + with self.subTest(command=command): + self.assertEqual(self.bash_argv(command), [argv]) + self.assertBlocked(self.hook({"command": command})) + + def test_unreadable_shell_scripts_without_visible_creates_are_allowed(self) -> None: + for command in ('bash <<< "$(echo hello)"', "bash < list[str]: + call = "gh issue create -b hello" + awk = "awk " + shlex.quote('BEGIN{print "' + call + '"}') + pipe = "printf '%s\\n' " + shlex.quote(call) + return [awk + " echo hi | sh", "echo hi | " + awk + " cat | sh", + pipe + " | (sh)", pipe + " | if true; then sh; fi", + "{ bash; } <<'EOF'\n" + call + "\nEOF\n", + "(bash) <<'EOF'\n" + call + "\nEOF\n", + "bash /dev/fd/3 3<<'EOF'\n" + call + "\nEOF\n"] + + def test_producer_arguments_cannot_authorize_a_body(self) -> None: + command = "awk 'BEGIN{print \"hello\"}' echo " + shlex.quote("Title\n" + RECORD) + " | gh issue create -F -" + self.assertEqual(self.bash_argv(command, capture_stdin=True), [["issue", "create", "-F", "-"]]) + self.assertEqual((self.cwd / "stdin.log").read_text(), "hello\n") + self.assertBlocked(self.hook({"command": command})) + (self.cwd / "f.md").write_text(RECORD) + command = "command -v cat > f.md; gh issue create -F f.md" + self.assertBlocked(self.hook({"command": command})) + self.assertEqual(self.bash_argv(command), [["issue", "create", "-F", "f.md"]]) + + def test_shell_feeding_checks_the_whole_command(self) -> None: + for command in self.whole_command_shell_feeds(): + with self.subTest(command=command): + self.assertEqual(self.bash_argv(command), [["issue", "create", "-b", "hello"]]) + self.assertBlocked(self.hook({"command": command})) + + def test_shell_feeding_without_unchecked_creates_stays_allowed(self) -> None: + (self.cwd / "f.md").write_text(RECORD) + commands = ["curl -fsSL https://x/install.sh | sh", "bash <<'EOF'\necho setup\nEOF\n", + "rg foo | sh -c 'wc -l'", 'bash < list[str]: + call = "gh issue create -b hello" + return ["sh <<< 'hi' <<'EOF'\n" + call + "\nEOF\n", + "cat <<< 'hi' <<'EOF' | sh\n" + call + "\nEOF\n", + "cat <<'A' <<'B' | sh\n" + call + "\nA\ntrue\nB\n", + "bash <(echo " + shlex.quote(call) + ")"] + + def test_whole_command_keeps_all_redirected_script_text(self) -> None: + for index, command in enumerate(self.competing_shell_redirects()): + with self.subTest(command=command): + expected = [] if index == 2 else [["issue", "create", "-b", "hello"]] + self.assertEqual(self.bash_argv(command), expected) + self.assertBlocked(self.hook({"command": command})) + zsh = "zsh -f -c " + shlex.quote(command) + self.assertBlocked(self.hook({"command": zsh})) + if shutil.which("zsh"): + self.assertEqual(self.bash_argv(zsh), [["issue", "create", "-b", "hello"]]) + + def test_main_differential_has_no_unintended_allowances(self) -> None: + # The exact shipped classifier is an independent safety baseline. + baseline = self.cwd / "baseline" + baseline.mkdir() + for name in ("project-direction", "agentsmd_instructions.py"): + path = baseline / name + path.write_bytes(subprocess.check_output( + ["git", "show", "15b1dd5:bin/" + name], cwd=ROOT)) + path.chmod(0o755) + corpus = self.round_four_commands() + self.unknown_producer_scripts("hello") + list(self.prefixed_creates("gh issue create --body hello")) + corpus += ["gh issue create --body hello", "gh issue create --help", + "gh issue create --title --help", "gh issue create # --help", + "gh pr create --body-file missing.md", + "gh -R view issue create --body hello", + "gh --repo search pr create --body hello", + "gh view gh issue create --body hello", + "case x in x)gh issue create --body hello;; esac", + "i=0; while test $i = 0; do i=1; gh issue create --body hello; done", + "i=0; until test $i = 1; do i=1; gh issue create --body hello; done", + "while gh issue create --body hello; do break; done", + "until gh issue create --body hello; do break; done", + "find . -maxdepth 0 -exec gh issue create --body hello \\;", + "nice gh api -X POST repos/o/r/issues -f body=hello", + "gh pr create --body " + shlex.quote(RECORD)] + corpus += self.competing_shell_redirects() + corpus += ["zsh -f -c " + shlex.quote(command) for command in self.competing_shell_redirects()] + corpus += self.whole_command_shell_feeds() + corpus += ["cat > f.md <<'BODY'\n" + RECORD + "BODY\ncommand -v cat > f.md; gh issue create -F f.md"] + corpus += [command for command, _ in self.unreadable_shell_scripts("hello")] + corpus += ['bash <<< "$(echo hello)"', "bash < None: for kind in ("issue", "pr"): for help_flag in ("--help", "-h", "--help=true", "--help=TRUE", "--help=True", "--help=1", "--help=t", "--help=T"): @@ -100,9 +827,13 @@ def test_create_help_passes_without_a_record(self) -> None: f"{kind} create {help_flag} --repo o/r", f"{kind} create --repo=o/r {help_flag}", f"{kind} create {help_flag} --repo=o/r"): - with self.subTest(args=args): - result = self.hook({"command": "gh " + args}) - self.assertEqual((result.returncode, result.stdout, result.stderr), (0, "", "")) + for verb in ("create", "new"): + command = "gh " + args.replace(" create ", " " + verb + " ") + with self.subTest(command=command): + self.assertEqual(self.bash_argv(command), + [shlex.split(command)[1:]]) + result = self.hook({"command": command}) + self.assertEqual((result.returncode, result.stdout, result.stderr), (0, "", "")) def test_help_does_not_exempt_another_create_in_the_command(self) -> None: for operator in (";", "&&", "||", "|", "|&", "&", "\n"): diff --git a/tools/versionctl/pyproject.toml b/tools/versionctl/pyproject.toml index 9661b58..b278431 100644 --- a/tools/versionctl/pyproject.toml +++ b/tools/versionctl/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "toolboxmd-versionctl" -version = "17.4.2" +version = "17.4.3" description = "Deterministic repository version mechanics for coding agents" readme = "README.md" requires-python = ">=3.11"