diff --git a/src/agent_comms/plan_phases.py b/src/agent_comms/plan_phases.py index a3ba290a5b..2ca44b5d8e 100644 --- a/src/agent_comms/plan_phases.py +++ b/src/agent_comms/plan_phases.py @@ -230,7 +230,7 @@ class RecordPlanePhase(CommanderNativePhase): class ReviewQueuePhase(CommanderNativePhase): """A verdict per PR head gates merging through main's required review status. - The owner's admin bypass stays; its hook demands a recorded verdict on that head. + QueueBypass declares each bypass's pin and recorded-verdict obligations. No GitHub merge queue is required. """ diff --git a/src/agent_comms/shell_effects.py b/src/agent_comms/shell_effects.py index 7748b9d036..2f9a714cff 100644 --- a/src/agent_comms/shell_effects.py +++ b/src/agent_comms/shell_effects.py @@ -45,8 +45,8 @@ def accepts(cls, name: str) -> bool: def matches(args: Sequence[str], prefixes: Sequence[tuple[str, ...]]) -> bool: return any(tuple(args[:len(prefix)]) == prefix for prefix in prefixes) - def invocation(self, decoded: ShellCommand, cwd: Path) -> ShellInvocation: - return ShellInvocation(decoded.words, cwd) + def following_directory(self, args: list[str], cwd: Path) -> Path: + return cwd def effects(self, args: list[str], cwd: Path) -> tuple[StorageEffect, ...]: """Unknown programs have no declared effects; capabilities compose through super().""" @@ -158,25 +158,29 @@ class ShellInvocation(ShellCommand): @classmethod def parse(cls, command: str, cwd: Path) -> Iterator[ShellInvocation]: try: - decoded_commands = tuple(ShellCommand.split(command)) + yield from cls.at_directory(ShellCommand.parsed(command), cwd) except ValueError as error: raise StorageDenied(f'StorageRule: the shell command cannot be decoded ({error}); ' 'no tool was started') from error - for decoded in decoded_commands: - invocation = ShellProgram.named(decoded.program).invocation(decoded, cwd) - cwd = invocation.cwd - yield invocation + + @classmethod + def at_directory(cls, commands: Iterator[ShellCommand], cwd: Path) -> Iterator[ShellInvocation]: + for decoded in commands: + yield cls(decoded.words, cwd, redirections=decoded.redirections) + # Child execution inherits this directory; its changes die with that child. + yield from cls.at_directory(decoded.nested(), cwd) + cwd = ShellProgram.named(decoded.program).following_directory(decoded.args, cwd) class ChangeDirectoryProgram(ShellProgram, names=frozenset({'cd'})): - def invocation(self, decoded: ShellCommand, cwd: Path) -> ShellInvocation: + def following_directory(self, args: list[str], cwd: Path) -> Path: try: - _, targets = getopt.getopt(decoded.words[1:], '') + _, targets = getopt.getopt(args, '') except getopt.GetoptError as error: raise StorageDenied(f'StorageRule: unknown cd option: {error}') from error if targets: cwd = physical_path(targets[0], cwd) - return ShellInvocation(decoded.words, cwd) + return cwd @dataclass(frozen=True) @@ -238,15 +242,22 @@ def effects(self, args: list[str], cwd: Path) -> tuple[StorageEffect, ...]: class WritesTargets(ShellProgram): - targets: ClassVar[slice] = slice(None) - minimum: ClassVar[int] = 1 + """The declaring CLI owns option arity; only decoded operands become paths.""" + + short_options: ClassVar[str] = '' + long_options: ClassVar[tuple[str, ...]] = ('help', 'version') + + def destinations(self, options: list[tuple[str, str]], arguments: list[str]) -> tuple[str, ...]: + return tuple(arguments) def effects(self, args: list[str], cwd: Path) -> tuple[StorageEffect, ...]: - paths = [arg for arg in args if not arg.startswith('-')] - selected = tuple( - WriteStorageEffect(physical_path(path, cwd)) for path in paths[self.targets] - ) if len(paths) >= self.minimum else () - return super().effects(args, cwd) + selected + try: + options, arguments = getopt.gnu_getopt(args, self.short_options, self.long_options) + except getopt.GetoptError as error: + raise StorageDenied(f'StorageRule: incomplete or unknown file-program option: {error}') from error + paths = () if any(option in ('--help', '--version') for option, _ in options) else self.destinations(options, arguments) + return super().effects(args, cwd) + tuple( + WriteStorageEffect(physical_path(path, cwd)) for path in paths) class WritesWorkingTree(ShellProgram): @@ -348,22 +359,121 @@ class ProfileProgram(AllocatesOnCommands, names=frozenset({'py-spy'})): prefixes = (('record',),) -class FileWriterProgram(WritesTargets, names=frozenset({'mkdir', 'touch', 'tee'})): - pass +class MakeDirectoryProgram(WritesTargets, names=frozenset({'mkdir'})): + short_options = 'm:pvZ' + long_options = (*WritesTargets.long_options, 'mode=', 'parents', 'verbose', 'context=?') + + +class TouchProgram(WritesTargets, names=frozenset({'touch'})): + short_options = 'acd:fhmr:t:' + long_options = (*WritesTargets.long_options, 'no-create', 'date=', 'no-dereference', 'reference=', 'time=') + + def destinations(self, options: list[tuple[str, str]], arguments: list[str]) -> tuple[str, ...]: + # '-' names the selected stdout descriptor, not a file named '-'. + return tuple(path for path in arguments if path != '-') + + +class TeeProgram(WritesTargets, names=frozenset({'tee'})): + short_options = 'aip' + long_options = (*WritesTargets.long_options, 'append', 'ignore-interrupts', 'output-error=?') + + +class CopyProgram(WritesTargets): + """Copy-like CLIs share the final destination operand, not their option meanings.""" + + def destinations(self, options: list[tuple[str, str]], arguments: list[str]) -> tuple[str, ...]: + return tuple(arguments[-1:]) if len(arguments) >= 2 else () + + +class TargetDirectoryCopyProgram(CopyProgram): + short_options = 't:S:' + long_options = (*CopyProgram.long_options, 'target-directory=', 'suffix=') + + def destinations(self, options: list[tuple[str, str]], arguments: list[str]) -> tuple[str, ...]: + directories = tuple(value for option, value in options if option in ('-t', '--target-directory')) + return directories or super().destinations(options, arguments) + + +class CopyFilesProgram(TargetDirectoryCopyProgram, names=frozenset({'cp'})): + short_options = TargetDirectoryCopyProgram.short_options + 'abdfiHLPlnpRrsTuvxZ' + long_options = (*TargetDirectoryCopyProgram.long_options, 'archive', 'attributes-only', 'backup=?', 'copy-contents', + 'no-dereference', 'debug', 'force', 'interactive', 'dereference', 'keep-directory-symlink', + 'link', 'no-clobber', 'preserve=?', 'no-preserve=', 'parents', 'recursive', 'reflink=?', + 'remove-destination', 'sparse=', 'strip-trailing-slashes', 'symbolic-link', + 'no-target-directory', 'update=?', 'verbose', 'one-file-system', 'context=?') + + +class MoveProgram(TargetDirectoryCopyProgram, names=frozenset({'mv'})): + short_options = TargetDirectoryCopyProgram.short_options + 'bfinTuvZ' + long_options = (*TargetDirectoryCopyProgram.long_options, 'backup=?', 'debug', 'exchange', 'force', 'interactive', + 'no-clobber', 'no-copy', 'strip-trailing-slashes', 'no-target-directory', + 'update=?', 'verbose', 'context') + + def destinations(self, options: list[tuple[str, str]], arguments: list[str]) -> tuple[str, ...]: + # Moving also removes source directory entries; destination-only admission hid that write. + return tuple(dict.fromkeys((*super().destinations(options, arguments), *arguments))) + + +class InstallFilesProgram(TargetDirectoryCopyProgram, names=frozenset({'install'})): + short_options = TargetDirectoryCopyProgram.short_options + 'bcCdDg:m:o:psTvZ' + long_options = (*TargetDirectoryCopyProgram.long_options, 'backup=?', 'compare', 'directory', 'debug', 'group=', + 'mode=', 'owner=', 'preserve-timestamps', 'strip', 'strip-program=', + 'no-target-directory', 'verbose', 'preserve-context', 'context=?') + + def destinations(self, options: list[tuple[str, str]], arguments: list[str]) -> tuple[str, ...]: + if any(option in ('-d', '--directory') for option, _ in options): + return tuple(arguments) + return super().destinations(options, arguments) + + +class LinkProgram(TargetDirectoryCopyProgram, names=frozenset({'ln'})): + short_options = TargetDirectoryCopyProgram.short_options + 'bdFfiLnPrsTv' + long_options = (*TargetDirectoryCopyProgram.long_options, 'backup=?', 'directory', 'force', 'interactive', 'logical', + 'no-dereference', 'physical', 'relative', 'symbolic', 'no-target-directory', 'verbose') + + def destinations(self, options: list[tuple[str, str]], arguments: list[str]) -> tuple[str, ...]: + selected = super().destinations(options, arguments) + return selected or (('.',) if len(arguments) == 1 else ()) + +class RsyncProgram(CopyProgram, names=frozenset({'rsync'})): + # Remote/native expansion is not inferred. Unknown options are explicit refusals. + short_options = '0468ACDEFGHIJKLNOPRSUVWXM:B:T:e:f:abcdghiklmnopqrstuvxyz@:' + long_options = (*WritesTargets.long_options, 'verbose', 'quiet', 'archive', 'recursive', 'relative', + 'checksum', 'links', 'copy-links', 'perms', 'owner', 'group', 'times', 'dry-run', + 'delete', 'delete-before', 'delete-during', 'delete-after', 'progress', 'stats', + 'human-readable', 'itemize-changes', 'compress', 'one-file-system', 'inplace', + 'partial', 'numeric-ids', 'ignore-existing', 'size-only', 'remove-source-files', + 'info=', 'debug=', 'stderr=', 'suffix=', 'backup-dir=', 'chmod=', 'chown=', + 'block-size=', 'rsh=', 'rsync-path=', 'timeout=', 'contimeout=', 'modify-window=', + 'temp-dir=', 'partial-dir=', 'compare-dest=', 'copy-dest=', 'link-dest=', + 'filter=', 'exclude=', 'exclude-from=', 'include=', 'include-from=', 'files-from=', + 'out-format=', 'log-file=', 'log-file-format=', 'bwlimit=', 'max-size=', 'min-size=', + 'write-batch=', 'only-write-batch=', 'read-batch=', 'remote-option=') + receiver_writes: ClassVar[frozenset[str]] = frozenset({ + '-T', '--temp-dir', '--partial-dir', '--backup-dir'}) + auxiliary_writes: ClassVar[frozenset[str]] = receiver_writes | frozenset({ + '--log-file', '--write-batch', '--only-write-batch'}) -class CopyProgram(WritesTargets, names=frozenset({'cp', 'mv', 'rsync', 'install', 'ln'})): - targets = slice(-1, None) - minimum = 2 + def destinations(self, options: list[tuple[str, str]], arguments: list[str]) -> tuple[str, ...]: + paths = super().destinations(options, arguments) + for option, value in options: + if option in self.receiver_writes and not Path(value).is_absolute(): + raise StorageDenied('StorageRule: use an absolute rsync receiver-side destination') + paths += tuple(value for option, value in options if option in self.auxiliary_writes) + if any(option == '--remove-source-files' for option, _ in options): + paths += tuple(arguments[:-1]) + if any(':' in path for path in paths): + raise StorageDenied('StorageRule: remote rsync destinations require a separately admitted transport') + return tuple(dict.fromkeys(paths)) def shell_storage_effects(command: str, cwd: Path) -> tuple[StorageEffect, ...]: """Decode known shell operations, not arbitrary interpreter programs.""" effects: list[StorageEffect] = [] for invocation in ShellInvocation.parse(command, cwd): - args = invocation.args - for index, word in enumerate(args[:-1]): - if word in ('>', '>>'): - effects.append(WriteStorageEffect(physical_path(args[index + 1], invocation.cwd))) - effects.extend(ShellProgram.named(invocation.program).effects(args, invocation.cwd)) + for redirection in invocation.redirections: + effects.extend(WriteStorageEffect(physical_path(path, invocation.cwd)) + for path in redirection.written_paths()) + effects.extend(ShellProgram.named(invocation.program).effects(invocation.args, invocation.cwd)) return tuple(effects) diff --git a/src/agent_comms/storage_rules.py b/src/agent_comms/storage_rules.py index 932353015a..d720a9db5a 100644 --- a/src/agent_comms/storage_rules.py +++ b/src/agent_comms/storage_rules.py @@ -689,14 +689,17 @@ def require_write(self, path: Path) -> None: owner, = owners resolved = path.resolve() sources = Checkout.git_roots(resolved, *resolved.parents) + scratch = self.rule.roots_of(owner).scratch.resolve() + # The registry's scratch allocation is more specific than an enclosing + # source tree. A repository rooted in that allocation is still source. + if resolved.is_relative_to(scratch) and not any( + root.is_relative_to(scratch) for root in sources): + return store = self.comms.goals.todos allowed = frozenset(boundary for root in sources for allocation in CheckoutAllocation.for_path(self.rule, owner, root) for boundary in allocation.edit_boundary(self.rule, owner, root, store, self.reads)) - # A thread registered at a directory holding its scratch (unit-fieldplan at $HOME) still - # owns that scratch: the rule gate's audit lives there, and refusing it deadlocked the unit. - in_project = (resolved.is_relative_to(owner.project_path.resolve()) - and not resolved.is_relative_to(self.rule.roots_of(owner).scratch.resolve())) + in_project = resolved.is_relative_to(owner.project_path.resolve()) if (sources or path.is_relative_to(self.rule.checkout_base) or in_project) and not any( resolved.is_relative_to(root) for root in allowed): self.rule.deny(f'{path} is a read/launch source, not allocated editable work of {owner.name}.') diff --git a/src/agent_comms_hooks/bounded_waits.py b/src/agent_comms_hooks/bounded_waits.py index 1909b99dba..0198420d6b 100644 --- a/src/agent_comms_hooks/bounded_waits.py +++ b/src/agent_comms_hooks/bounded_waits.py @@ -18,7 +18,7 @@ from .shell_commands import ( ShellCommand, - shell_commands, + TimeoutRunner, ) from .watch import SessionHookProgram @@ -31,9 +31,7 @@ class Timeout: """`timeout [OPTION]... DURATION COMMAND`: whether it bounds the command it runs, and that command. Any other command runs itself, unbounded.""" - program: ClassVar[str] = "timeout" - valued: ClassVar[frozenset[str]] = frozenset(("-k", "-s", "--kill-after", "--signal")) - """Options whose value is the next word.""" + program: ClassVar[str] = TimeoutRunner.names[0] duration: ClassVar[re.Pattern[str]] = re.compile(r"(\d+(?:\.\d*)?)([smhd]?)") unit: ClassVar[dict[str, int]] = {"": 1, "s": 1, "m": 60, "h": 3600, "d": 86400} max_seconds: ClassVar[int] = 3600 @@ -45,9 +43,7 @@ class Timeout: def around(cls, command: ShellCommand) -> Timeout: if command.program != cls.program: return cls(False, command.words) - rest = command.words[1:] - while rest[:1] and rest[0].startswith("-"): - rest = rest[2:] if rest[0] in cls.valued else rest[1:] + rest = tuple(TimeoutRunner.arguments(command.args)) limit = cls.duration.fullmatch(rest[0]) if rest else None seconds = float(limit.group(1)) * cls.unit[limit.group(2)] if limit else 0 return cls(0 < seconds <= cls.max_seconds, rest[1:]) @@ -93,39 +89,28 @@ class WaitCommand(Bash): """The word that ends a loop's condition.""" openers: ClassVar[frozenset[str]] = frozenset((*body, "then", "!", "{")) """Words after which the next word is again a command.""" - shell: ClassVar[re.Pattern[str]] = re.compile(r"(?:ba)?sh") - script_flag: ClassVar[re.Pattern[str]] = re.compile(r"-\w*c") - @classmethod def commands(cls, segment: Words) -> Iterator[ShellCommand]: - """What one segment runs: its command, or the command its `timeout` runs.""" + """Execution descendants derive from the same runner/group declarations.""" for command in ShellCommand.of(dropwhile(cls.openers.__contains__, segment)): - yield from Timeout.around(command).commands() + yield from command.run_commands() def waits(self) -> Iterator[Wait]: - """Every loop this command runs, at command position or in a script a shell runs with - -c. Only such a script can be bounded: `timeout` runs programs, not shell keywords.""" - segments = tuple(map(tuple, shell_commands(self.command))) - for index, segment in enumerate(segments): - for command in ShellCommand.of(dropwhile(self.openers.__contains__, segment)): - timeout = Timeout.around(command) - for run in timeout.commands(): - match (run.program, *run.args): - case (keyword, *condition) if keyword in self.loops: - yield Wait( - False, - ( - tuple(condition), - *takewhile( - lambda later: later[:1] != self.body, segments[index + 1 :] - ), - ), - ) - case (program, flag, script, *_) if self.shell.fullmatch( - program - ) and self.script_flag.fullmatch(flag): - for wait in WaitCommand(script).waits(): - yield replace(wait, bounded=wait.bounded or timeout.bounded) + return self.in_scope(tuple(ShellCommand.parsed(self.command)), False) + + @classmethod + def in_scope(cls, commands: tuple[ShellCommand, ...], bounded: bool) -> Iterator[Wait]: + """Shell source owns control words; executable argv does not become shell syntax.""" + segments = tuple(command.source_words for command in commands) + for index, decoded in enumerate(commands): + command = replace(decoded, words=tuple(dropwhile(cls.openers.__contains__, decoded.words))) + timeout = Timeout.around(command) + for source in command.source_commands(): + for run in Timeout.around(source).commands(): + if run.program in cls.loops: + yield Wait(bounded, (tuple(run.args), *takewhile( + lambda later: later[:1] != cls.body, segments[index + 1:]))) + yield from cls.in_scope(tuple(command.nested()), bounded or timeout.bounded) def unbounded_wait(self) -> tuple[str, ...]: """Why this command's wait loop could run forever; empty when it holds no loop.""" diff --git a/src/agent_comms_hooks/queue_bypass.py b/src/agent_comms_hooks/queue_bypass.py index 1eb6cf8156..d9ba9988da 100644 --- a/src/agent_comms_hooks/queue_bypass.py +++ b/src/agent_comms_hooks/queue_bypass.py @@ -144,7 +144,7 @@ def matching(cls, command: ShellCommand) -> tuple[QueueBypass, ...]: class AdminMerge(QueueBypass): - """GitHub's head precondition and Core's verdict name the same commit.""" + """A pinned admin invocation supplies its review without a saved Core verdict.""" reason = "`gh pr merge --admin` bypasses required checks." prefix = ("pr", "merge") diff --git a/src/agent_comms_hooks/shell_commands.py b/src/agent_comms_hooks/shell_commands.py index 252b1246a3..276291dfb0 100644 --- a/src/agent_comms_hooks/shell_commands.py +++ b/src/agent_comms_hooks/shell_commands.py @@ -1,58 +1,419 @@ from __future__ import annotations +import getopt import io +import re import shlex -from collections.abc import ( - Iterable, - Iterator, -) -from dataclasses import dataclass -from itertools import groupby +from collections.abc import Iterable, Iterator +from dataclasses import dataclass, field, replace from pathlib import Path -from typing import ( - ClassVar, - Self, -) - - -def without_here_documents(command: str) -> str: - """The command's own text: one lexer reads each quoted word whole across newlines, - stops at the end of the line that opens here-documents, and resumes after their - bodies, so a body is never lexed as shell.""" - kept = [] - while command: - text = io.StringIO(command) - lex = shlex.shlex(text, posix=True, punctuation_chars="<>") - lex.whitespace_split = True - delimiters: list[str] = [] - previous, line_end = "", len(command) - for word in lex: - if previous in ("<<", "<<-"): - if not delimiters: - found = command.find("\n", text.tell() - 1) - line_end = len(command) if found < 0 else found - lex.instream = io.StringIO(command[text.tell():line_end + 1]) - delimiters.append(word) - previous = word - lines = iter(command[line_end + 1:].splitlines(keepends=True)) - kept.append(command[:line_end + 1]) - for delimiter in delimiters: - for line in lines: - if line.rstrip("\r\n").lstrip("\t") == delimiter: - break - command = "".join(lines) if delimiters else "" - return "".join(kept) - - -def shell_commands(command: str) -> Iterator[list[str]]: - """Shell words, not matches in quoted code or here-document bodies.""" - lex = shlex.shlex(without_here_documents(command), posix=True, punctuation_chars=";&|><()\n") - lex.whitespace = " \t\r" - lex.whitespace_split = True - separators = set(";&|()\n") - for separator, group in groupby(lex, key=lambda word: bool(word) and set(word) <= separators): - if not separator: - yield list(group) +from typing import ClassVar, Never, Self + + +class ShellText(io.StringIO): + """shlex skips comments with readline; the shell still needs their newline.""" + + def readline(self, size: int = -1) -> str: + line = super().readline(size) + if line.endswith('\n'): + self.seek(self.tell() - 1) + return line + + def document_lines(self) -> Iterator[str]: + while line := super().readline(): + yield line + + +@dataclass(frozen=True) +class Redirection: + target: str + + def written_paths(self) -> tuple[str, ...]: + return () + + def defer(self, lexer: ShellLexer) -> None: + pass + + def read_body(self, lexer: ShellLexer) -> None: + pass + + +class OutputRedirection(Redirection): + def written_paths(self) -> tuple[str, ...]: + return (self.target,) + + +class HereDocument(Redirection): + def defer(self, lexer: ShellLexer) -> None: + lexer.expect_document(self) + + def line_value(self, line: str) -> str: + return line.rstrip('\r\n') + + def check_line(self, line: str) -> None: + pass + + def read_body(self, lexer: ShellLexer) -> None: + lexer.resume_at_boundary() + for line in lexer.text.document_lines(): + if self.line_value(line) == self.target: + return + self.check_line(line) + raise ValueError(f'Unterminated here-document {self.target!r}') + + +class ExpandedHereDocument(HereDocument): + def check_line(self, line: str) -> None: + if '$(' in line or '`' in line: + raise ValueError('Command substitution in an expanded here-document is not admitted') + + +class TabStrippedHereDocument(HereDocument): + def line_value(self, line: str) -> str: + return super().line_value(line).lstrip('\t') + + +class TabStrippedExpandedHereDocument(TabStrippedHereDocument, ExpandedHereDocument): + pass + + +@dataclass(frozen=True) +class ShellWord: + value: str + start: int + end: int + spelling: str + + @classmethod + def decoded(cls, value: str, lexer: ShellLexer, end: int) -> ShellWord: + return cls(value, lexer.word_start, end, lexer.command[lexer.word_start:end]) + + def consume(self, parser: ShellParser) -> Iterator[ShellCommand]: + parser.word(self) + return iter(()) + + descriptor_syntax: ClassVar[re.Pattern[str]] = re.compile(r'[0-9]+\Z') + + def descriptor_before(self, position: int) -> bool: + return (self.end == position and self.descriptor_syntax.fullmatch(self.value) is not None + and self.spelling == self.value) + + def descriptor_value(self) -> bool: + return self.value == '-' or self.descriptor_syntax.fullmatch(self.value.removesuffix('-')) is not None + + assignment_syntax: ClassVar[re.Pattern[str]] = re.compile(r'[a-zA-Z_][a-zA-Z_0-9]*=') + + def assignment(self) -> bool: + return self.assignment_syntax.match(self.spelling) is not None + + def command_type(self) -> type[ShellCommand]: + return SourceCommand + + def document(self) -> HereDocument: + return ExpandedHereDocument(self.value) + + def tabbed_document(self) -> HereDocument: + return TabStrippedExpandedHereDocument(self.value) + + +class QuotedWord(ShellWord): + def command_type(self) -> type[ShellCommand]: + return ShellCommand + + def document(self) -> HereDocument: + return HereDocument(self.value) + + def tabbed_document(self) -> HereDocument: + return TabStrippedHereDocument(self.value) + + +class ShellLexer(shlex.shlex): + """Retain shlex's own punctuation decision, before POSIX quote removal erases it. + + No second quoting scanner: native state transitions distinguish punctuation from + quoted/escaped words. The pushback queue determines the actual source boundary. + """ + + def __init__(self, command: str): + self.text = ShellText(command) + self.command = command + self._state: str | None = None + self.word_start = 0 + self.word_type: type[ShellWord] = ShellWord + self.documents: list[HereDocument] = [] + super().__init__(self.text, posix=True, punctuation_chars=';&|><()\n') + self.whitespace = ' \t\r' + self.whitespace_split = True + + @property + def state(self) -> str | None: + return self._state + + @state.setter + def state(self, value: str | None) -> None: + if self._state == ' ' and value not in (' ', None): + self.word_start = self.position - 1 + if value == 'c': + self.word_type = Operator + elif value in ("'", '"', '\\'): + self.word_type = QuotedWord + self._state = value + + @property + def position(self) -> int: + return self.text.tell() - len(self._pushback_chars) + + def take_adjacent(self, character: str) -> bool: + if self.command[self.position:self.position + 1] != character: + return False + if self._pushback_chars: + self._pushback_chars.pop() + else: + self.text.read(1) + return True + + def resume_at_boundary(self, position: int | None = None) -> None: + self.text.seek(self.position if position is None else position) + self._pushback_chars.clear() + self.state = ' ' + + def expect_document(self, document: HereDocument) -> None: + self.documents.append(document) + + def read_documents(self) -> None: + documents, self.documents = self.documents, [] + for document in documents: + document.read_body(self) + + def next_word(self) -> ShellWord | None: + self.word_type = ShellWord + value = super().read_token() + if value is None: + return None + end = self.position + if self.state == ' ' and self.command[end - 1:end] in self.whitespace: + end -= 1 + return self.word_type.decoded(value, self, end) + + +class Operator(ShellWord): + """The grammar's declarations own operator membership and behavior.""" + + members: ClassVar[dict[str, type[Operator]]] = {} + + def __init_subclass__(cls, *, spellings: tuple[str, ...] = ()) -> None: + super().__init_subclass__() + for spelling in spellings: + cls.members[spelling] = cls + + @classmethod + def decoded(cls, value: str, lexer: ShellLexer, end: int) -> ShellWord: + spelling = next((key for key in sorted(cls.members, key=len, reverse=True) + if value.startswith(key)), '') + if not spelling: + raise ValueError(f'Unsupported shell operator {value!r}') + end = lexer.word_start + len(spelling) + selected = cls.members[spelling](spelling, lexer.word_start, end, spelling) + # Admit one operator before recursive parsing; native lookahead stays here. + lexer.resume_at_boundary(end) + return selected + + def consume(self, parser: ShellParser) -> Iterator[ShellCommand]: + return parser.operator(self) + + def apply(self, parser: ShellParser) -> Iterator[ShellCommand]: + parser.finish() + return iter(()) + + +class Separator(Operator, spellings=(';', ';;', ';&', ';;&', '&', '&&', '|', '||', '|&')): + pass + + +class OpenGroup(Operator, spellings=('(',)): + def apply(self, parser: ShellParser) -> Iterator[ShellCommand]: + parser.open_group() + return iter(()) + + +class CloseGroup(Operator, spellings=(')',)): + def apply(self, parser: ShellParser) -> Iterator[ShellCommand]: + parser.close_group() + + +class Newline(Operator, spellings=('\n',)): + def apply(self, parser: ShellParser) -> Iterator[ShellCommand]: + parser.finish() + parser.lexer.read_documents() + return parser.flush() + + +class Redirect(Operator): + redirection: ClassVar[type[Redirection]] = Redirection + + def bind(self, word: ShellWord) -> Redirection: + return self.redirection(word.value) + + def consume_descriptor(self, parser: ShellParser) -> None: + pass + + def apply(self, parser: ShellParser) -> Iterator[ShellCommand]: + parser.target.require_word() + self.consume_descriptor(parser) + parser.target = RedirectionTarget(self) + return iter(()) + + +class NumberedRedirect(Redirect, spellings=('<', '<<<')): + """This grammar admits an optional adjacent, unquoted ASCII IO number.""" + + def consume_descriptor(self, parser: ShellParser) -> None: + if parser.words and parser.words[-1].descriptor_before(self.start): + parser.words.pop() + + +class WriteRedirect(Redirect): + redirection = OutputRedirection + + +class NumberedWriteRedirect(NumberedRedirect, WriteRedirect, spellings=('>', '>>', '>|', '<>')): + pass + + +class CombinedWriteRedirect(WriteRedirect, spellings=('&>', '&>>')): + pass + + +class ReadDescriptorRedirect(NumberedRedirect, spellings=('<&',)): + def bind(self, word: ShellWord) -> Redirection: + if not word.descriptor_value(): + raise ValueError('Input descriptor redirection requires a literal descriptor') + return super().bind(word) + + +class WriteDescriptorRedirect(NumberedRedirect, spellings=('>&',)): + def bind(self, word: ShellWord) -> Redirection: + if word.descriptor_value(): + return Redirection(word.value) + return OutputRedirection(word.value) + + +class DocumentRedirect(NumberedRedirect, spellings=('<<',)): + def bind(self, word: ShellWord) -> Redirection: + return word.document() + + def apply(self, parser: ShellParser) -> Iterator[ShellCommand]: + super().apply(parser) + if parser.lexer.take_adjacent('-'): + parser.target = RedirectionTarget(TabbedDocumentRedirect( + '<<-', self.start, self.end + 1, '<<-')) + return iter(()) + + +class TabbedDocumentRedirect(DocumentRedirect): + def bind(self, word: ShellWord) -> Redirection: + return word.tabbed_document() + + +class ArgumentTarget: + def take(self, parser: ShellParser, word: ShellWord) -> None: + parser.words.append(word) + + def require_word(self) -> None: + pass + + +@dataclass(frozen=True) +class RedirectionTarget(ArgumentTarget): + operator: Redirect + + def take(self, parser: ShellParser, word: ShellWord) -> None: + redirection = self.operator.bind(word) + parser.redirections.append(redirection) + redirection.defer(parser.lexer) + parser.target = ArgumentTarget() + + def require_word(self) -> None: + raise ValueError('Shell redirection requires a word operand') + + +class ShellParser: + def __init__(self, lexer: ShellLexer): + self.lexer = lexer + self.pending = ShellCommand(()) + self.words: list[ShellWord] = [] + self.redirections: list[Redirection] = [] + self.target = ArgumentTarget() + self.line: list[ShellCommand] = [] + + def word(self, word: ShellWord) -> None: + self.target.take(self, word) + + def operator(self, operator: Operator) -> Iterator[ShellCommand]: + return operator.apply(self) + + def open_group(self) -> None: + self.target.require_word() + if self.words or self.redirections: + raise ValueError('A parenthesized command requires its own command position') + self.pending.require_group_boundary() + self.pending = CommandGroup((), body=tuple(GroupParser(self.lexer).commands())) + + def close_group(self) -> Never: + raise ValueError('Unexpected closing parenthesis') + + def finish(self) -> None: + self.target.require_word() + while self.words and self.words[0].assignment(): + self.words.pop(0) + self.line.extend(self.pending.completed(tuple(self.words), tuple(self.redirections))) + self.pending = ShellCommand(()) + self.words.clear() + self.redirections.clear() + + def flush(self) -> Iterator[ShellCommand]: + completed = tuple(self.line) + self.line.clear() + return iter(completed) + + def commands(self) -> Iterator[ShellCommand]: + while word := self.lexer.next_word(): + yield from word.consume(self) + self.finish() + self.lexer.read_documents() + yield from self.flush() + + +class GroupClosed(Exception): + """The group's own closing delimiter ends its shared-source parse.""" + + +class GroupParser(ShellParser): + def close_group(self) -> Never: + raise GroupClosed() + + def commands(self) -> Iterator[ShellCommand]: + try: + yield from super().commands() + except GroupClosed: + self.finish() + yield from self.flush() + return + raise ValueError('Unterminated parenthesized command') + + +class NotStandalone(ValueError): + """Literal recovery does not admit this shell syntax.""" + + +class LiteralShellParser(ShellParser): + def word(self, word: ShellWord) -> None: + if set(word.spelling) & {'$', '`', '\n'}: + raise NotStandalone('Recovery requires literal, single-line words') + super().word(word) + + def operator(self, operator: Operator) -> Iterator[ShellCommand]: + raise NotStandalone('Recovery does not admit shell operators') @dataclass(frozen=True) @@ -67,98 +428,185 @@ def runs(self, command: ShellCommand) -> bool: class ConnectCommand: - """The Claude bridge's connect declaration, shared by its CLI and outage permission.""" - - program: ClassVar[Program] = Program("agent-comms-claude-bridge") + program: ClassVar[Program] = Program('agent-comms-claude-bridge') @classmethod def command_name(cls) -> str: - return ConnectCommand.__name__.removesuffix("Command").lower() + return ConnectCommand.__name__.removesuffix('Command').lower() @classmethod def permits(cls, command: str) -> bool: - """Recovery runs alone, not beside another command or shell redirection.""" commands = tuple(ShellCommand.split(command)) return (len(commands) == 1 and commands[0].invokes(cls.program, cls.command_name()) - and not commands[0].args[1:]) + and not commands[0].args[1:] and not commands[0].redirections) @dataclass(frozen=True) class ShellCommand: - """One simple command: its words after any leading `NAME=value` assignments, so its - program is the first of them, named by its basename however its path is written.""" + """Executable words and syntax operands remain distinct after boundary decoding.""" words: tuple[str, ...] + redirections: tuple[Redirection, ...] = field(default=(), kw_only=True) @property def program(self) -> str: - return Path(self.words[0]).name + return Path(self.words[0]).name if self.words else '' @property def args(self) -> list[str]: return list(self.words[1:]) def invokes(self, program: Program, verb: str) -> bool: - """The executable and first argument identify one declared invocation.""" return program.runs(self) and self.args[:1] == [verb] def value_options(self, allowed: frozenset[str]) -> bool: - """After its verb, this command contains only named option/value pairs.""" options = self.args[1:] return len(options) % 2 == 0 and set(options[::2]) <= allowed @classmethod - def standalone(cls, command: str) -> Iterator[Self]: - """One literal invocation, never shell operators, substitution or redirection.""" - if set(command) & set(";|&$`<>()\n"): + def standalone(cls, command: str) -> Iterator[ShellCommand]: + try: + yield from LiteralShellParser(ShellLexer(command)).commands() + except NotStandalone: return - words = tuple(shlex.split(command)) - if words: - yield cls(words) @classmethod def of(cls, words: Iterable[str]) -> Iterator[Self]: - """The command these words run; none when they only assign variables.""" - words = list(words) - while words and "=" in words[0]: - words.pop(0) + words = tuple(words) if words: - yield cls(tuple(words)) + yield cls(words) - shells: ClassVar[frozenset[str]] = frozenset({"bash", "sh", "zsh"}) - # Programs that run their arguments as a command, after their flags, env's assignments - # and this many operands (timeout's duration): each tool's own command-line contract. - runners: ClassVar[dict[str, int]] = {"timeout": 1, "env": 0, "nohup": 0, "stdbuf": 0} + @classmethod + def parsed(cls, command: str) -> Iterator[ShellCommand]: + yield from ShellParser(ShellLexer(command)).commands() @classmethod - def split(cls, command: str) -> Iterator[Self]: - """Every simple command the text runs, including those a shell runs from its `-c` - string, so `bash -c '...'` hides nothing from a reader of the commands.""" - for words in shell_commands(command): - for simple in cls.of(words): - yield from simple.run_commands() - - def run_commands(self) -> Iterator[Self]: - """This command and every command it runs: a `-c` string, or the command a runner - such as `timeout` or `env` starts.""" + def split(cls, command: str) -> Iterator[ShellCommand]: + for simple in cls.parsed(command): + yield from simple.run_commands() + + def source_commands(self) -> Iterator[ShellCommand]: + """Arguments launched as an executable are not shell control syntax.""" + return iter(()) + + @property + def source_words(self) -> tuple[str, ...]: + return tuple(word for command in self.source_commands() for word in command.words) + + def completed(self, words: tuple[ShellWord, ...], redirections: tuple[Redirection, ...]) -> Iterator[ShellCommand]: + if words or redirections: + command_type = words[0].command_type() if words else ShellCommand + yield command_type(tuple(word.value for word in words), redirections=redirections) + + def require_group_boundary(self) -> None: + pass + + def nested(self) -> Iterator[ShellCommand]: + runner = Runner.named(self.program) + yield from runner.inline(self.args) + yield from runner.started(self.args) + + def run_commands(self) -> Iterator[ShellCommand]: yield self - yield from self.inline_commands() - for command in self.started(): + for command in self.nested(): yield from command.run_commands() - def started(self) -> Iterator[Self]: - """The command a runner starts; none for any other program.""" - if self.program not in self.runners: - return - words = list(self.args) - for _ in range(len(words)): - if not words or not (words[0].startswith("-") or "=" in words[0]): - break - words.pop(0) - yield from self.of(words[self.runners[self.program] :]) - - def inline_commands(self) -> Iterator[Self]: - """The commands this shell runs from its `-c` string; none for any other program.""" - args = self.args - if self.program in self.shells and "-c" in args[:-1]: - yield from self.split(args[args.index("-c") + 1]) + +class SourceCommand(ShellCommand): + def source_commands(self) -> Iterator[ShellCommand]: + return iter((self,)) + + +@dataclass(frozen=True) +class CommandGroup(ShellCommand): + body: tuple[ShellCommand, ...] = field(kw_only=True) + + def completed(self, words: tuple[ShellWord, ...], redirections: tuple[Redirection, ...]) -> Iterator[Self]: + if words or not self.body: + raise ValueError('A parenthesized command requires a nonempty body and no trailing words') + yield replace(self, redirections=redirections) + + def require_group_boundary(self) -> None: + raise ValueError('Adjacent groups require a command separator') + + def nested(self) -> Iterator[ShellCommand]: + return iter(self.body) + + +class Runner(Program): + members: ClassVar[dict[str, type[Runner]]] = {} + names: ClassVar[tuple[str, ...]] = () + short_options: ClassVar[str] = '' + long_options: ClassVar[tuple[str, ...]] = ('help', 'version') + operands: ClassVar[int] = 0 + + def __init_subclass__(cls, *, names: tuple[str, ...] = ()) -> None: + super().__init_subclass__() + cls.names = names + for name in names: + cls.members[name] = cls + + @classmethod + def named(cls, name: str) -> Runner: + return cls.members.get(name, Runner)(name) + + @classmethod + def options(cls, args: list[str]) -> tuple[list[tuple[str, str]], list[str]]: + try: + return getopt.getopt(args, cls.short_options, cls.long_options) + except getopt.GetoptError as error: + raise ValueError(f'Incomplete or unknown runner option: {error}') from error + + @classmethod + def arguments(cls, args: list[str]) -> list[str]: + return cls.options(args)[1] + + def started(self, args: list[str]) -> Iterator[ShellCommand]: + return iter(()) + + def inline(self, args: list[str]) -> Iterator[ShellCommand]: + return iter(()) + + +class ArgumentRunner(Runner): + def started(self, args: list[str]) -> Iterator[ShellCommand]: + yield from ShellCommand.of(self.arguments(args)[self.operands:]) + + +class TimeoutRunner(ArgumentRunner, names=('timeout',)): + short_options = 'k:s:v' + long_options = ('kill-after=', 'signal=', 'foreground', 'preserve-status', 'verbose', 'help', 'version') + operands = 1 + + +class EnvironmentRunner(ArgumentRunner, names=('env',)): + short_options = 'i0u:' + long_options = ('ignore-environment', 'null', 'unset=', 'help', 'version') + + def started(self, args: list[str]) -> Iterator[ShellCommand]: + arguments = self.arguments(args) + while arguments and ShellWord.assignment_syntax.match(arguments[0]): + arguments.pop(0) + yield from ShellCommand.of(arguments) + + +class NoHangupRunner(ArgumentRunner, names=('nohup',)): + pass + + +class BufferRunner(ArgumentRunner, names=('stdbuf',)): + short_options = 'i:o:e:' + long_options = ('input=', 'output=', 'error=', 'help', 'version') + + +class ShellRunner(Runner, names=('bash', 'sh', 'zsh')): + short_options = 'c:o:O:abefhiklmnprstuvxBCEHPT' + long_options = ('noprofile', 'norc', 'posix', 'restricted', 'verbose', 'login', 'help', 'version') + + def inline(self, args: list[str]) -> Iterator[ShellCommand]: + options, remaining = self.options(args) + scripts = tuple(value for option, value in options if option == '-c') + if not scripts and remaining and remaining[0].startswith('+'): + raise ValueError('Shell +options are not admitted; use explicit -options') + for script in scripts: + yield from ShellCommand.parsed(script) diff --git a/tests/test_queue_bypass_hook.py b/tests/test_queue_bypass_hook.py index c517d170ff..2862a7cb61 100644 --- a/tests/test_queue_bypass_hook.py +++ b/tests/test_queue_bypass_hook.py @@ -1,4 +1,4 @@ -"""The real PreToolUse hook requires Core's exact-head verdict for a commander bypass.""" +"""The real PreToolUse hook preserves each bypass's pin and review contract.""" import json import subprocess @@ -13,7 +13,7 @@ from hook_session_fixture import resident_hook, resident_session from agent_comms.todos import QualifiedHeadVerdict, ReviewedHeadVerdict, RejectedHeadVerdict from agent_comms_hooks.queue_bypass import MergeTarget, MergeTargetUnreadable, QueueBypass -from agent_comms_hooks.shell_commands import ShellCommand, shell_commands +from agent_comms_hooks.shell_commands import HereDocument, ShellCommand from test_head_review import ACCEPTED, QUALIFIED, REASON, invoke, opened pytest_plugins = ("test_head_review",) @@ -108,40 +108,36 @@ def test_option_data_does_not_select_a_merge_family(command, github_cli): assert QueueBypass.matching(simple) == () -def test_commander_admin_and_api_merges_require_the_pinned_recorded_ok( +def test_commander_api_merge_requires_the_pinned_recorded_ok( head_bus, reviewed_source, tmp_path): comms, github, request = head_bus transcript = tmp_path / "s.jsonl" - def commands(head): - return ( - f"gh pr merge {request.pr} --admin --merge --repo {request.repo.to_wire()} --match-head-commit {head}", - f"gh api -X PUT repos/{request.repo.to_wire()}/pulls/{request.pr}/merge -f merge_method=merge -f sha={head}", - ) + + def command(head): + return (f"gh api -X PUT repos/{request.repo.to_wire()}/pulls/{request.pr}/merge" + f" -f merge_method=merge -f sha={head}") with resident_session(comms, tmp_path / "state", "lead", "s") as bridge: def refused(head): - for command in commands(head): - result = resident_tool_use(command, transcript, bridge) - assert result.status == 0, result.stderr - decision = json.loads(result.stdout)["hookSpecificOutput"] - assert decision["permissionDecision"] == "deny" - assert f"#{request.pr}" in decision["permissionDecisionReason"] - assert head in decision["permissionDecisionReason"] - assert "ok verdict" in decision["permissionDecisionReason"] + result = resident_tool_use(command(head), transcript, bridge) + assert result.status == 0, result.stderr + decision = json.loads(result.stdout)["hookSpecificOutput"] + assert decision["permissionDecision"] == "deny" + assert f"#{request.pr}" in decision["permissionDecisionReason"] + assert head in decision["permissionDecisionReason"] + assert "ok verdict" in decision["permissionDecisionReason"] refused(request.head) invoke(comms, request, ReviewedHeadVerdict(ACCEPTED)) - for command in commands(request.head): - result = resident_tool_use(command, transcript, bridge) - assert result.status == 0, result.stderr - assert result.stdout == "" + result = resident_tool_use(command(request.head), transcript, bridge) + assert result.status == 0, result.stderr + assert result.stdout == "" moved = opened(github, reviewed_source, request.pr, "b" * 40) refused(moved.head) # Core authorizes the saved pin; GitHub's precondition refuses that old pin after a push. - for command in commands(request.head): - result = resident_tool_use(command, transcript, bridge) - assert result.status == 0, result.stderr - assert result.stdout == "" + result = resident_tool_use(command(request.head), transcript, bridge) + assert result.status == 0, result.stderr + assert result.stdout == "" opened(github, reviewed_source, request.pr, request.head) invoke(comms, request, RejectedHeadVerdict(REASON)) refused(request.head) @@ -150,10 +146,9 @@ def refused(head): @pytest.mark.parametrize("actor,allowed", (("lead", True), ("worker", False))) -def test_merge_authority_comes_from_the_admitted_actor_not_a_changed_launcher( +def test_pinned_admin_merge_needs_no_recorded_ok_and_uses_the_admitted_actor( head_bus, tmp_path, actor, allowed): comms, _, request = head_bus - invoke(comms, request, ReviewedHeadVerdict(ACCEPTED)) command = (f"gh pr merge {request.pr} --admin --merge --repo {request.repo.to_wire()}" f" --match-head-commit {request.head}") with resident_session(comms, tmp_path / "state", actor, "s") as bridge: @@ -170,15 +165,26 @@ def test_merge_authority_comes_from_the_admitted_actor_not_a_changed_launcher( assert "bound commander" in decision["permissionDecisionReason"] +def test_commander_admin_merge_still_requires_its_head_pin(head_bus, tmp_path): + comms, _, request = head_bus + command = f"gh pr merge {request.pr} --admin --merge --repo {request.repo.to_wire()}" + with resident_session(comms, tmp_path / "state", "lead", "s") as bridge: + result = resident_tool_use(command, tmp_path / "s.jsonl", bridge) + assert result.status == 0, result.stderr + assert json.loads(result.stdout)["hookSpecificOutput"]["permissionDecision"] == "deny" + + def test_escaped_quote_inside_double_quotes_is_one_word(): command = """git grep -n "'watch'\\|\\"watch\\"" -- tools | cut -c1-9""" - assert list(shell_commands(command)) == [ + assert [list(simple.words) for simple in ShellCommand.parsed(command)] == [ ["git", "grep", "-n", """'watch'\\|"watch\"""", "--", "tools"], ["cut", "-c1-9"], ] def test_here_document_body_after_escaped_quote_is_not_shell(): command = 'echo "a\\"b"; cat <<\'EOF\'\nit\'s "not shell\nEOF\necho done\n' - assert list(shell_commands(command)) == [ - ["echo", 'a"b'], ["cat", "<<", "EOF"], ["echo", "done"], + parsed = tuple(ShellCommand.parsed(command)) + assert [list(simple.words) for simple in parsed] == [ + ["echo", 'a"b'], ["cat"], ["echo", "done"], ] + assert parsed[1].redirections == (HereDocument('EOF'),) diff --git a/tests/test_shell_decoding.py b/tests/test_shell_decoding.py new file mode 100644 index 0000000000..ae3c5b31e7 --- /dev/null +++ b/tests/test_shell_decoding.py @@ -0,0 +1,209 @@ +"""Data-only shell boundary controls: no commands, subprocesses or filesystem writes.""" +from __future__ import annotations + +from pathlib import Path +import shlex +import unittest + +from agent_comms.shell_effects import ( + ForbiddenStorageEffect, ShellInvocation, StorageDenied, WriteStorageEffect, + shell_storage_effects, +) +from agent_comms_hooks.bounded_waits import WaitCommand +from agent_comms_hooks.shell_commands import ConnectCommand, HereDocument, ShellCommand + + +CWD = Path('/shell-decoder-control') + + +class ShellDecoding(unittest.TestCase): + def test_words_are_never_promoted_to_operators(self): + cases = ( + ("echo '>' picked", ('echo', '>', 'picked')), + ('echo ">>" picked', ('echo', '>>', 'picked')), + (r'echo \> picked', ('echo', '>', 'picked')), + ("echo '<<' EOF", ('echo', '<<', 'EOF')), + ("echo ';' '|' '&' '(' ')'; echo done", ('echo', ';', '|', '&', '(', ')')), + ('echo "a\\\"b"', ('echo', 'a"b')), + ("echo prefix'<<'suffix", ('echo', 'prefix< picked it off the wheel')\nPY\nmkdir target", + "python - <<'PY' # a header comment\nprint('> picked')\nPY\nmkdir target", + "cat <<'A' <<'B'\nuv sync\nA\npython -m venv env\nB\nmkdir target", + "cat <<-'A'\n\t'not shell\n\tA\nmkdir target", + "cat <>> '''not shell\nEOF\nmkdir target", + "cat <<'EOF'; echo header\n>& 'not shell\nEOF\nmkdir target", + shlex.join(('bash', '-c', "python - <<'PY'\nprint(\"it's > picked\")\nPY\nmkdir target")), + ) + for text in cases: + with self.subTest(text=text): + self.assertEqual(shell_storage_effects(text, CWD), (WriteStorageEffect(CWD / 'target'),)) + + def test_document_scope_and_unsupported_expansion_are_explicit(self): + for text in ( + 'cat <', + ): + with self.subTest(text=text): + with self.assertRaises(StorageDenied): + shell_storage_effects(text, CWD) + + def test_redirection_operands_are_not_arguments(self): + cases = ( + ('echo value > target', ('echo', 'value'), ('target',)), + ('echo 2> target', ('echo',), ('target',)), + ("echo '2'> target", ('echo', '2'), ('target',)), + ('echo 2 > target', ('echo', '2'), ('target',)), + ('echo value 2>&1', ('echo', 'value'), ()), + ('echo ٢> target', ('echo', '٢'), ('target',)), + ('echo >&٢', ('echo',), ('٢',)), + ('echo value >& target', ('echo', 'value'), ('target',)), + ('cat < input', ('cat',), ()), + ('cat 2 target', ('cat',), ('target',)), + ('cat <<< text', ('cat',), ()), + ('cat 2<< target', (), ('target',)), + ('echo &>> target', ('echo',), ('target',)), + ) + for number in ('2', '700'): + retained = (f"'{number}'", f'"{number}"', '\\' + number, number + ' ') + for operator in ('&>', '&>>'): + cases += tuple((f'mkdir {prefix}{operator}/owned/log', ('mkdir', number), + ('/owned/log', number)) for prefix in (number, *retained)) + for operator in ('>', '>>', '>|', '<>'): + cases += ((f'mkdir {number}{operator}/owned/log', ('mkdir',), ('/owned/log',)),) + cases += tuple((f'mkdir {prefix}{operator}/owned/log', ('mkdir', number), + ('/owned/log', number)) for prefix in retained) + for text, words, paths in cases: + with self.subTest(text=text): + command, = ShellCommand.parsed(text) + self.assertEqual(command.words, words) + self.assertEqual(shell_storage_effects(text, CWD), tuple( + WriteStorageEffect(CWD / path) for path in paths)) + + def test_option_operands_belong_to_the_declaring_program(self): + cases = ( + ('mkdir -m 700 target', ('target',)), + ('mkdir target --mode=700', ('target',)), + ('mkdir -pm700 target other', ('target', 'other')), + ('mkdir -- -m 700', ('-m', '700')), + ('mkdir --context target', ('target',)), + ('mkdir --context=label target', ('target',)), + ('touch -d tomorrow -r reference target', ('target',)), + ('touch -t 202611010100 target', ('target',)), + ('touch -', ()), + ('tee --output-error=warn -a target', ('target',)), + ('cp --reflink=auto --sparse always source target', ('target',)), + ('cp source -t target', ('target',)), + ('mv -t target source', ('target', 'source')), + ('install -m 700 -g group source target', ('target',)), + ('install -dm700 one two', ('one', 'two')), + ('ln -s source', ('.',)), + ('ln -s -t target source', ('target',)), + ('rsync -avt -e "ssh -p 2222" --exclude pattern source target', ('target',)), + ('rsync --log-file=log source target', ('target', 'log')), + ('mkdir --help', ()), + ('mkdir -- --help', ('--help',)), + ) + for text, paths in cases: + with self.subTest(text=text): + self.assertEqual(shell_storage_effects(text, CWD), tuple( + WriteStorageEffect((CWD / path).resolve()) for path in paths)) + + def test_bad_or_unknown_option_operands_are_not_guessed(self): + for text in ('mkdir -m', 'touch -r', 'cp --invented path source target', + 'rsync source host:target', 'rsync --partial-dir relative source target', + 'bash +e -c "uv sync"', 'env -C elsewhere mkdir target'): + with self.subTest(text=text): + with self.assertRaises(StorageDenied): + shell_storage_effects(text, CWD) + + def test_redirections_use_execution_directory_before_cd(self): + text = 'cd next > before; mkdir -m 700 after' + self.assertEqual(shell_storage_effects(text, CWD), ( + WriteStorageEffect(CWD / 'before'), WriteStorageEffect(CWD / 'next' / 'after'))) + self.assertEqual([invocation.cwd for invocation in ShellInvocation.parse(text, CWD)], + [CWD, CWD / 'next']) + + def test_literal_recovery_uses_the_same_operator_boundary(self): + valid = 'agent-comms-claude-bridge watch --state "quoted;path"' + command, = ShellCommand.standalone(valid) + self.assertEqual(command.args[-1], 'quoted;path') + for text in ('echo ok; echo no', 'echo "$HOME"', 'echo `date`', 'echo ok > file', + 'echo ok\necho no', '(echo ok)'): + with self.subTest(text=text): + self.assertEqual(tuple(ShellCommand.standalone(text)), ()) + self.assertTrue(ConnectCommand.permits('agent-comms-claude-bridge connect')) + self.assertFalse(ConnectCommand.permits('agent-comms-claude-bridge connect > file')) + + def test_child_execution_owns_its_directory_lifetime(self): + for text, paths in ( + ('bash -c "cd child; mkdir inner"; mkdir outer', ('child/inner', 'outer')), + ('(cd child; mkdir inner); mkdir outer', ('child/inner', 'outer')), + ('((cd child; mkdir inner)); mkdir outer', ('child/inner', 'outer')), + ('timeout -s INT 5 env -u OLD bash -c "cd child; mkdir inner"; mkdir outer', ('child/inner', 'outer')), + ('(cd child && mkdir inner) > group.log; mkdir outer', ('group.log', 'child/inner', 'outer')), + ("(cat <<'EOF') > group.log\n> not shell\nEOF\nmkdir outer", ('group.log', 'outer')), + ): + with self.subTest(text=text): + self.assertEqual(shell_storage_effects(text, CWD), tuple( + WriteStorageEffect(CWD / path) for path in paths)) + for text in ('(mkdir target', ')', '()', '(mkdir a)(mkdir b)', '(mkdir a) trailing'): + with self.subTest(text=text), self.assertRaises(StorageDenied): + shell_storage_effects(text, CWD) + + def test_waits_follow_source_scope_and_runner_ownership(self): + script = 'while kill -0 123; do sleep 1; done' + for text, bounded in ( + (script, False), + ('(' + script + ')', False), + (shlex.join(('bash', '-o', 'pipefail', '-c', script)), False), + (shlex.join(('timeout', '-s', 'INT', '5', 'bash', '-c', script)), True), + (shlex.join(('env', '-u', 'OLD', 'timeout', '5', 'bash', '-c', '(' + script + ')')), True), + ): + with self.subTest(text=text): + waits = tuple(WaitCommand(text).waits()) + self.assertEqual(len(waits), 1) + self.assertEqual(waits[0].bounded, bounded) + self.assertEqual(tuple(WaitCommand('"while" kill -0 123').waits()), ()) + polling = shlex.join(('timeout', '5', 'bash', '-c', + 'while env -u OLD pgrep -f worker; do sleep 1; done')) + self.assertTrue(tuple(WaitCommand(polling).waits())[0].polls_pgrep_f()) + + def test_document_capabilities_compose(self): + parsed, = ShellCommand.parsed("cat <<'EOF'\nliteral\nEOF\n") + self.assertEqual(parsed.redirections, (HereDocument('EOF'),)) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/test_storage_rules.py b/tests/test_storage_rules.py index 38b3e59b4e..5b50cdb83f 100644 --- a/tests/test_storage_rules.py +++ b/tests/test_storage_rules.py @@ -282,19 +282,44 @@ def test_equal_repository_spellings_allocate_one_owned_checkout(checkout_comms: assert path in rule.checkouts_of(thread, RepositoryReads()) -def test_a_thread_registered_above_its_scratch_may_write_its_scratch_and_only_that(checkout_comms: Comms): - """unit-fieldplan is registered at $HOME: its audit path, which the rule gate requires, sat - inside its project and was refused as source it does not hold, so nothing could run.""" +@pytest.mark.parametrize('enclosing_source', (False, True)) +def test_a_thread_registered_above_its_scratch_may_write_its_scratch_and_only_that( + checkout_comms: Comms, enclosing_source: bool): + """An enclosing project or Git tree cannot take the registry's scratch allocation.""" comms = checkout_comms rule = comms.registry.storage_rule above = rule.scratch_base.parent + if enclosing_source: + Repo.git(above, 'init', '-q') thread = comms.registry.declare(Thread('homed', frozenset(), Checkout(above))) projection = StorageProjection(rule, thread, comms) - projection.require_write(rule.roots_of(thread).scratch / 'compaction-deal.audit.json') + scratch = rule.roots_of(thread).scratch + projection.require_write(scratch / 'compaction-deal.audit.json') with pytest.raises(StorageDenied, match='not allocated editable work'): projection.require_write(above / 'notes.txt') with pytest.raises(StorageDenied, match='foreign scratch'): projection.require_write(rule.scratch_base / 'worker' / 'notes.txt') + with pytest.raises(StorageDenied, match='artifacts'): + projection.require_write(scratch / '.artifacts' / 'output') + scratch.mkdir(parents=True) + (scratch / 'project-link').symlink_to(above, target_is_directory=True) + with pytest.raises(StorageDenied, match='not allocated editable work'): + projection.require_write(scratch / 'project-link' / 'notes.txt') + + +@pytest.mark.parametrize('source_path', (Path(), Path('nested-source'))) +def test_git_source_inside_scratch_still_requires_an_edit_allocation( + checkout_comms: Comms, source_path: Path): + comms = checkout_comms + rule = comms.registry.storage_rule + above = rule.scratch_base.parent + thread = comms.registry.declare(Thread('homed', frozenset(), Checkout(above))) + scratch = rule.roots_of(thread).scratch + source = scratch / source_path + source.mkdir(parents=True) + Repo.git(source, 'init', '-q') + with pytest.raises(StorageDenied, match='not allocated editable work'): + StorageProjection(rule, thread, comms).require_write(source / 'implementation.py') def test_three_small_checkouts_are_admitted(checkout_comms: Comms):