| |
| """工作区 Shell 工具。 |
| |
| 提供一个隔离的工作区目录，允许 Agent 通过受限的 shell 风格命令操作 |
| 文件。 |
| """ |
|
|
| from __future__ import annotations |
|
|
| import argparse |
| import gzip |
| import logging |
| import os |
| import re |
| import shlex |
| import shutil |
| import stat |
| from datetime import datetime |
| from pathlib import Path |
| from typing import Any, Callable, Dict, List, Never |
|
|
| from ..runtime.workspace import Workspace |
| from .base_tool import BaseTool, ToolMetadata |
| from .code_executor import LimitedWriter |
| from ..utils import log_exception |
|
|
| _logger = logging.getLogger(f"sr_harness.{__name__}") |
|
|
| |
| ALLOWED_COMMANDS = { |
| "ls", "cat", "head", "tail", "wc", "grep", |
| "sort", "cut", "cp", "mv", "rm", "mkdir", |
| "gunzip", "gzip", "unzip", "tar", |
| } |
|
|
|
|
| class CommandParseError(ValueError): |
| """Raised when the restricted command language cannot parse an invocation.""" |
|
|
|
|
| class CommandArgumentParser(argparse.ArgumentParser): |
| """An ``argparse`` parser that reports errors instead of exiting the process.""" |
|
|
| def __init__(self, command: str): |
| super().__init__(prog=command, add_help=False, allow_abbrev=False, exit_on_error=False) |
|
|
| def error(self, message: str) -> Never: |
| """Run the ``error`` operation. |
| |
| Args: |
| message: Message text or provider message payload. |
| |
| Returns: |
| Never: The operation result. |
| """ |
| raise CommandParseError(f"{self.prog}: {message}") |
|
|
| def exit(self, status: int = 0, message: str | None = None) -> Never: |
| """Run the ``exit`` operation. |
| |
| Args: |
| status: Run completion status. |
| message: Message text or provider message payload. |
| |
| Returns: |
| Never: The operation result. |
| """ |
| raise CommandParseError(message or f"{self.prog}: invalid arguments") |
|
|
|
|
| @BaseTool.register("workspace_shell") |
| class WorkspaceShellTool(BaseTool): |
| """Implementation of the workspace shell tool.""" |
|
|
| metadata = ToolMetadata(name="workspace_shell") |
|
|
| DEFAULT_OUTPUT_LIMIT_BYTES = 64 * 1024 |
| MAX_OUTPUT_LIMIT_BYTES = 1024 * 1024 |
|
|
| @classmethod |
| def get_doc(cls) -> dict[str, str]: |
| """Return documentation exposed as a runtime skill. |
| |
| Returns: |
| dict[str, str]: The operation result. |
| """ |
| return { |
| "name": "workspace-shell", |
| "description": ( |
| "Inspect and modify workspace files with a safe, restricted shell-like " |
| "command language." |
| ), |
| "content": ( |
| "# Workspace Shell\n\n" |
| "`workspace_shell` implements a small command language in Python. It does not " |
| "start a system shell or execute operating-system commands. Use it to inspect, " |
| "filter, copy, move, remove, and unpack files in the current workspace.\n\n" |
| "## Syntax\n\n" |
| "- Separate arguments with whitespace. Single quotes, double quotes, and " |
| "backslash escapes follow POSIX shell lexical rules.\n" |
| "- Join supported commands with `|`. A pipeline passes each command's text " |
| "output to the next command in memory; spaces around `|` are optional.\n" |
| "- Short flags may be combined or separated: `ls -la`, `ls -al`, " |
| "`ls -l -a`, and `ls -a -l` are equivalent. Options and operands may usually " |
| "be interspersed. Use `--` before a path beginning with `-`.\n" |
| "- Every path must resolve inside the workspace. Absolute paths and paths or " |
| "symbolic links that escape the workspace are rejected.\n" |
| "- Redirection and other shell operators (`>`, `<`, `;`, `&&`, `||`, `&`) are " |
| "not supported. There is no globbing, variable or tilde expansion, command " |
| "substitution, background execution, or invocation of unlisted commands.\n\n" |
| "## Supported commands\n\n" |
| "```text\n" |
| "ls [-a|--all] [-l] [PATH ...]\n" |
| "cat [-n|--number] [FILE ...]\n" |
| "head [-n N|--lines N|-N] [FILE]\n" |
| "tail [-n N|--lines N|-N] [FILE]\n" |
| "wc [-l] [-w] [-c] [-m] [FILE ...]\n" |
| "grep [-i|--ignore-case] [-v|--invert-match] [-n|--line-number]\n" |
| " [-c|--count] [-F|--fixed-strings] PATTERN [FILE]\n" |
| "sort [-r|--reverse] [-n|--numeric-sort] [-u|--unique] [FILE]\n" |
| "cut [-d DELIMITER|--delimiter DELIMITER] -f FIELDS [FILE]\n" |
| "cp [-r|-R|--recursive] [-f|--force] SOURCE DESTINATION\n" |
| "mv [-f|--force] SOURCE DESTINATION\n" |
| "rm [-r|-R|--recursive] [-f|--force] PATH ...\n" |
| "mkdir [-p|--parents] DIRECTORY ...\n" |
| "gzip [-k|--keep] [-f|--force] FILE\n" |
| "gunzip [-k|--keep] [-f|--force] FILE\n" |
| "unzip [-d DIRECTORY|--directory DIRECTORY] FILE\n" |
| "tar (-x|--extract) [-z|--gzip] (-f FILE|--file FILE)\n" |
| " [-C DIRECTORY|--directory DIRECTORY]\n" |
| "```\n\n" |
| "`grep` performs fixed-string matching; `-F` is accepted for familiar grep " |
| "syntax. `cut -f` accepts comma-separated, one-based field numbers. `tar` only " |
| "supports extraction and accepts grouped flags such as `-xf` and `-xzf`. When " |
| "a text-processing command has no file operand, it reads the previous pipeline " |
| "stage.\n\n" |
| "## Examples\n\n" |
| "```text\n" |
| "ls -la\n" |
| "cat 'experiment 1.csv' | head -5\n" |
| "cat data.csv|grep -i result|wc -l\n" |
| "cut data.csv -d, -f1,3 | sort -u\n" |
| "mkdir -p results/archive\n" |
| "cp -r results results-copy\n" |
| "tar -xzf results.tar.gz -C results/archive\n" |
| "cat -- -notes\n" |
| "```" |
| ), |
| } |
|
|
| def execute( |
| self, |
| command: str, |
| output_limit_bytes: int = DEFAULT_OUTPUT_LIMIT_BYTES, |
| ) -> Dict[str, Any]: |
| """Execute a restricted shell command in the workspace directory. |
| Supported commands: ls, cat, head, tail, wc, grep, sort, cut, cp, mv, rm, mkdir, |
| gunzip, gzip, unzip, tar. |
| All file paths are relative to the workspace root. Absolute paths and path traversal |
| (e.g., ../) are forbidden. |
| |
| Args: |
| command: A shell command string. |
| Examples: "ls", "cat data.csv | head -5", "gunzip data.csv.gz". |
| output_limit_bytes: Maximum stdout size returned by each command segment. |
| """ |
| if self.context.workspace is None: |
| return self._error("Workspace not initialized.") |
| output_limit_bytes = self._bounded_output_limit(output_limit_bytes) |
|
|
| try: |
| pipeline = self._parse_pipeline(command) |
| except CommandParseError as exc: |
| return self._error(f"Command parse error: {exc}") |
|
|
| stdin_text = "" |
| for tokens in pipeline: |
| try: |
| result = self._execute_tokens(tokens, stdin_text) |
| if not result.get("success", False): |
| return result |
| stdin_text = self._limit_output(result.get("stdout", ""), output_limit_bytes) |
| except Exception as e: |
| rendered = shlex.join(tokens) |
| _logger.error(f"Error executing command segment '{rendered}': {log_exception(e)}") |
| return self._error( |
| f"Error executing command segment '{rendered}', ask human for help: " |
| f"{self._safe_exception_text(e)}" |
| ) |
| return self._ok(stdin_text) |
|
|
| @classmethod |
| def format_result_dict(cls, result: Dict[str, Any]) -> str: |
| """Format a tool result for the language model. |
| |
| Args: |
| result: Result mapping to format or update. |
| |
| Returns: |
| str: The operation result. |
| """ |
| if not result.get("success", False): |
| detail = result.get("error") or result.get("stderr") or "No error detail was provided." |
| return f"Command failed: {detail}" |
| parts = [] |
| if result.get("stdout"): |
| parts.append(result["stdout"]) |
| if result.get("stderr"): |
| parts.append(f"Command stderr:\n{result['stderr']}") |
| return ( |
| "\n".join(parts) if parts else "Command completed successfully and produced no output." |
| ) |
|
|
| def execute_single(self, command: str, stdin_text: str = "") -> Dict[str, Any]: |
| """Execute one command segment after pipeline parsing. |
| |
| Args: |
| command: Parsed command and arguments. |
| stdin_text: Text received from the previous pipeline segment. |
| |
| Returns: |
| Structured command output and status. |
| """ |
| if self.context.workspace is None: |
| return self._error("Workspace not initialized.") |
| try: |
| parts = shlex.split(command) |
| except ValueError as e: |
| return self._error(f"Command parse error: {e}") |
| return self._execute_tokens(parts, stdin_text) |
|
|
| @staticmethod |
| def _parse_pipeline(command: str) -> list[list[str]]: |
| """Parse the deliberately small shell-like language into a pipeline AST. |
| |
| Quotes and backslash escapes follow POSIX shell lexical rules, but the only |
| executable operator is a single pipe. Shell evaluation, expansion, |
| redirection, conditionals, command substitution, and background jobs are |
| intentionally outside this language. |
| """ |
| try: |
| lexer = shlex.shlex(command, posix=True, punctuation_chars="|&;<>") |
| lexer.whitespace_split = True |
| lexer.commenters = "" |
| tokens = list(lexer) |
| except ValueError as exc: |
| raise CommandParseError(str(exc)) from exc |
|
|
| pipeline: list[list[str]] = [[]] |
| for token in tokens: |
| if token == "|": |
| if not pipeline[-1]: |
| raise CommandParseError("empty command before pipe") |
| pipeline.append([]) |
| elif token and all(char in "|&;<>" for char in token): |
| raise CommandParseError(f"operator {token!r} is not supported") |
| else: |
| pipeline[-1].append(token) |
| if not pipeline[-1]: |
| message = "empty command" if len(pipeline) == 1 else "empty command after pipe" |
| raise CommandParseError(message) |
| return pipeline |
|
|
| def _execute_tokens(self, parts: list[str], stdin_text: str) -> Dict[str, Any]: |
| if not parts: |
| return self._error("Empty command.") |
|
|
| cmd_name = parts[0] |
| args = parts[1:] |
|
|
| if cmd_name not in ALLOWED_COMMANDS: |
| return self._error( |
| f"Command '{cmd_name}' is not allowed. " |
| f"Allowed: {', '.join(sorted(ALLOWED_COMMANDS))}" |
| ) |
|
|
| handler = getattr(self, f"_cmd_{cmd_name}", None) |
| if handler is None: |
| return self._error(f"Command '{cmd_name}' is not implemented.") |
| try: |
| return handler(args, stdin_text) |
| except CommandParseError as exc: |
| return self._error(str(exc)) |
|
|
| @staticmethod |
| def _ok(stdout: str) -> Dict[str, Any]: |
| return {"success": True, "stdout": stdout, "stderr": "", "exit_code": 0} |
|
|
| @staticmethod |
| def _error(msg: str) -> Dict[str, Any]: |
| return {"success": False, "stdout": "", "stderr": "", "error": msg, "exit_code": 1} |
|
|
| @classmethod |
| def _bounded_output_limit(cls, value: Any) -> int: |
| try: |
| parsed = int(value) |
| except (TypeError, ValueError): |
| parsed = cls.DEFAULT_OUTPUT_LIMIT_BYTES |
| return max(1024, min(cls.MAX_OUTPUT_LIMIT_BYTES, parsed)) |
|
|
| @staticmethod |
| def _limit_output(stdout: str, output_limit_bytes: int) -> str: |
| writer = LimitedWriter(output_limit_bytes) |
| writer.write(stdout) |
| return writer.getvalue() |
|
|
| @staticmethod |
| def _parse_args( |
| command: str, |
| args: list[str], |
| configure: Callable[[CommandArgumentParser], None], |
| ) -> argparse.Namespace: |
| parser = CommandArgumentParser(command) |
| configure(parser) |
| try: |
| return parser.parse_intermixed_args(args) |
| except argparse.ArgumentError as exc: |
| raise CommandParseError(f"{command}: {exc}") from exc |
|
|
| def _read_text(self, path: Path) -> str: |
| if not path.is_file(): |
| raise CommandParseError(f"Not a regular file: {path.name}") |
| try: |
| with self._workspace.safe_open( |
| path, "r", encoding="utf-8", errors="replace" |
| ) as f: |
| return f.read() |
| except OSError as exc: |
| raise CommandParseError( |
| f"Cannot read file: {self._safe_exception_text(exc)}" |
| ) from exc |
|
|
| @property |
| def _workspace(self) -> Workspace: |
| workspace = self.context.workspace |
| if workspace is None: |
| raise RuntimeError("Workspace not initialized.") |
| return workspace |
|
|
| def _safe_exception_text(self, error: BaseException) -> str: |
| """Remove host workspace and mount locations from a user-visible error.""" |
| message = str(error) |
| workspace = self.context.workspace |
| if workspace is None: |
| return message |
| replacements = [ |
| (workspace.path, "."), |
| *( |
| (source, os.fspath(logical.relative_to(workspace.path))) |
| for logical, source in workspace.mount_map.items() |
| ), |
| ] |
| for host_path, visible_path in sorted( |
| replacements, |
| key=lambda pair: len(os.fspath(pair[0])), |
| reverse=True, |
| ): |
| message = message.replace(os.fspath(host_path), visible_path) |
| return message |
|
|
| def _copy_file(self, source: str | os.PathLike, destination: str) -> str: |
| """Copy one regular file while enforcing safe source and destination access.""" |
| ws = self._workspace |
| with ws.safe_open(source, "rb") as input_stream: |
| with ws.open_for_write(destination, "wb") as output_stream: |
| shutil.copyfileobj(input_stream, output_stream) |
| return os.fspath(destination) |
|
|
| @staticmethod |
| def _reject_tree_symlinks(directory: str, names: list[str]) -> set[str]: |
| """Reject recursive copies containing symbolic links instead of following them.""" |
| symbolic_links = [name for name in names if (Path(directory) / name).is_symlink()] |
| if symbolic_links: |
| raise ValueError( |
| "recursive copies do not follow symbolic links: " |
| + ", ".join(symbolic_links) |
| ) |
| return set() |
|
|
| @staticmethod |
| def _find_tree_symlink(root: Path) -> Path | None: |
| """Return the first symbolic link below a directory without following links.""" |
| for directory, directories, filenames in os.walk(root, followlinks=False): |
| parent = Path(directory) |
| for name in (*directories, *filenames): |
| candidate = parent / name |
| if candidate.is_symlink(): |
| return candidate |
| return None |
|
|
| @staticmethod |
| def _nonnegative_int(value: str) -> int: |
| try: |
| parsed = int(value) |
| except ValueError as exc: |
| raise argparse.ArgumentTypeError("must be an integer") from exc |
| if parsed < 0: |
| raise argparse.ArgumentTypeError("must be non-negative") |
| return parsed |
|
|
| def _cmd_ls(self, args: list, stdin: str) -> Dict[str, Any]: |
| ws = self._workspace |
| def configure(parser): |
| parser.add_argument("-a", "--all", action="store_true") |
| parser.add_argument("-l", action="store_true", dest="long") |
| parser.add_argument("paths", nargs="*") |
|
|
| options = self._parse_args("ls", args, configure) |
| targets = options.paths or ["."] |
| sections = [] |
| for target in targets: |
| if (path := ws.resolve(target)) is None: |
| return self._error(f"Invalid path: {target}") |
| if not path.exists(): |
| return self._error(f"No such file or directory: {target}") |
| if path.is_dir(): |
| entries = sorted( |
| ( |
| entry |
| for entry in path.iterdir() |
| if options.all or not entry.name.startswith(".") |
| ), |
| key=lambda entry: entry.name, |
| ) |
| names = [ |
| entry.name + ("@" if entry.is_symlink() else "/" if entry.is_dir() else "") |
| for entry in entries |
| ] |
| if options.long: |
| rendered = [] |
| for name, entry in zip(names, entries): |
| info = entry.lstat() |
| modified = datetime.fromtimestamp(info.st_mtime).strftime("%Y-%m-%d %H:%M") |
| rendered.append( |
| f"{stat.filemode(info.st_mode)} {info.st_size:>10} {modified} {name}" |
| ) |
| output = "\n".join(rendered) |
| else: |
| output = "\n".join(names) |
| if len(targets) > 1: |
| sections.append(f"{target}:\n" + output) |
| else: |
| sections.append(output) |
| else: |
| if options.long: |
| info = path.stat() |
| modified = datetime.fromtimestamp(info.st_mtime).strftime("%Y-%m-%d %H:%M") |
| sections.append( |
| f"{stat.filemode(info.st_mode)} {info.st_size:>10} {modified} {path.name}" |
| ) |
| else: |
| sections.append(path.name) |
| return self._ok("\n\n".join(sections)) |
|
|
| def _cmd_cat(self, args: list, stdin: str) -> Dict[str, Any]: |
| ws = self._workspace |
| def configure(parser): |
| parser.add_argument("-n", "--number", action="store_true") |
| parser.add_argument("files", nargs="*") |
|
|
| options = self._parse_args("cat", args, configure) |
| if not options.files: |
| text = stdin |
| if options.number: |
| text = "\n".join( |
| f"{index:6}\t{line}" for index, line in enumerate(text.splitlines(), 1) |
| ) |
| return self._ok(text) |
| parts: List[str] = [] |
| for arg in options.files: |
| if (path := ws.resolve(arg)) is None: |
| return self._error(f"Invalid path: {arg}") |
| if not path.exists(): |
| return self._error(f"No such file: {arg}") |
| try: |
| parts.append(self._read_text(path)) |
| except Exception as e: |
| return self._error(f"Cannot read file: {self._safe_exception_text(e)}") |
| text = "".join(parts) |
| if options.number: |
| text = "\n".join( |
| f"{index:6}\t{line}" for index, line in enumerate(text.splitlines(), 1) |
| ) |
| return self._ok(text) |
|
|
| def _cmd_head(self, args: list, stdin: str) -> Dict[str, Any]: |
| ws = self._workspace |
| normalized = [ |
| part |
| for arg in args |
| for part in (["-n", arg[1:]] if re.fullmatch(r"-\d+", arg) else [arg]) |
| ] |
|
|
| def configure(parser): |
| parser.add_argument("-n", "--lines", type=self._nonnegative_int, default=10) |
| parser.add_argument("files", nargs="*") |
|
|
| options = self._parse_args("head", normalized, configure) |
| if len(options.files) > 1: |
| return self._error("head: at most one file is supported") |
| if options.files: |
| path = ws.resolve(options.files[0]) |
| if path is None: |
| return self._error(f"Invalid path: {options.files[0]}") |
| if not path.exists(): |
| return self._error(f"No such file: {options.files[0]}") |
| text = self._read_text(path) |
| else: |
| text = stdin |
| lines = text.splitlines()[:options.lines] |
| return self._ok("\n".join(lines)) |
|
|
| def _cmd_tail(self, args: list, stdin: str) -> Dict[str, Any]: |
| ws = self._workspace |
| normalized = [ |
| part |
| for arg in args |
| for part in (["-n", arg[1:]] if re.fullmatch(r"-\d+", arg) else [arg]) |
| ] |
|
|
| def configure(parser): |
| parser.add_argument("-n", "--lines", type=self._nonnegative_int, default=10) |
| parser.add_argument("files", nargs="*") |
|
|
| options = self._parse_args("tail", normalized, configure) |
| if len(options.files) > 1: |
| return self._error("tail: at most one file is supported") |
| if options.files: |
| path = ws.resolve(options.files[0]) |
| if path is None: |
| return self._error(f"Invalid path: {options.files[0]}") |
| if not path.exists(): |
| return self._error(f"No such file: {options.files[0]}") |
| text = self._read_text(path) |
| else: |
| text = stdin |
| lines = text.splitlines()[-options.lines:] if options.lines else [] |
| return self._ok("\n".join(lines)) |
|
|
| def _cmd_wc(self, args: list, stdin: str) -> Dict[str, Any]: |
| ws = self._workspace |
| def configure(parser): |
| parser.add_argument("-l", "--lines", action="store_true") |
| parser.add_argument("-w", "--words", action="store_true") |
| parser.add_argument("-c", "--bytes", action="store_true") |
| parser.add_argument("-m", "--chars", action="store_true") |
| parser.add_argument("files", nargs="*") |
|
|
| options = self._parse_args("wc", args, configure) |
| selected = [options.lines, options.words, options.bytes, options.chars] |
| enabled_counts = selected if any(selected) else [True, True, True, False] |
| sources: list[tuple[str, str]] = [] |
| if options.files: |
| for name in options.files: |
| path = ws.resolve(name) |
| if path is None: |
| return self._error(f"Invalid path: {name}") |
| if not path.exists(): |
| return self._error(f"No such file: {name}") |
| sources.append((name, self._read_text(path))) |
| else: |
| sources.append(("", stdin)) |
|
|
| def counts(text: str) -> list[int]: |
| values = [ |
| len(text.splitlines()), |
| len(text.split()), |
| len(text.encode("utf-8")), |
| len(text), |
| ] |
| return [value for value, enabled in zip(values, enabled_counts) if enabled] |
|
|
| output = [ |
| " ".join(map(str, counts(text))) + (f" {name}" if name else "") |
| for name, text in sources |
| ] |
| if len(sources) > 1: |
| totals = [sum(values) for values in zip(*(counts(text) for _, text in sources))] |
| output.append(" ".join(map(str, totals)) + " total") |
| return self._ok("\n".join(output)) |
|
|
| def _cmd_grep(self, args: list, stdin: str) -> Dict[str, Any]: |
| ws = self._workspace |
| def configure(parser): |
| parser.add_argument("-i", "--ignore-case", action="store_true") |
| parser.add_argument("-v", "--invert-match", action="store_true") |
| parser.add_argument("-n", "--line-number", action="store_true") |
| parser.add_argument("-c", "--count", action="store_true") |
| parser.add_argument("-F", "--fixed-strings", action="store_true") |
| parser.add_argument("pattern") |
| parser.add_argument("files", nargs="*") |
|
|
| options = self._parse_args("grep", args, configure) |
| if len(options.files) > 1: |
| return self._error("grep: at most one file is supported") |
| if options.files: |
| path = ws.resolve(options.files[0]) |
| if path is None: |
| return self._error(f"Invalid path: {options.files[0]}") |
| if not path.exists(): |
| return self._error(f"No such file: {options.files[0]}") |
| text = self._read_text(path) |
| else: |
| text = stdin |
| pattern = options.pattern.casefold() if options.ignore_case else options.pattern |
| matched = [] |
| for number, line in enumerate(text.splitlines(), 1): |
| searchable = line.casefold() if options.ignore_case else line |
| include = pattern in searchable |
| if include == options.invert_match: |
| continue |
| matched.append( |
| f"{number}:{line}" if options.line_number and not options.count else line |
| ) |
| return self._ok(str(len(matched)) if options.count else "\n".join(matched)) |
|
|
| def _cmd_sort(self, args: list, stdin: str) -> Dict[str, Any]: |
| ws = self._workspace |
| def configure(parser): |
| parser.add_argument("-r", "--reverse", action="store_true") |
| parser.add_argument("-n", "--numeric-sort", action="store_true") |
| parser.add_argument("-u", "--unique", action="store_true") |
| parser.add_argument("files", nargs="*") |
|
|
| options = self._parse_args("sort", args, configure) |
| if len(options.files) > 1: |
| return self._error("sort: at most one file is supported") |
| if options.files: |
| path = ws.resolve(options.files[0]) |
| if path is None: |
| return self._error(f"Invalid path: {options.files[0]}") |
| if not path.exists(): |
| return self._error(f"No such file: {options.files[0]}") |
| text = self._read_text(path) |
| else: |
| text = stdin |
| lines = text.splitlines() |
| if options.unique: |
| lines = list(dict.fromkeys(lines)) |
| if options.numeric_sort: |
| try: |
| lines.sort(key=lambda line: float(line.strip()), reverse=options.reverse) |
| except ValueError: |
| return self._error("sort: non-numeric input with -n") |
| else: |
| lines.sort(reverse=options.reverse) |
| return self._ok("\n".join(lines)) |
|
|
| def _cmd_cut(self, args: list, stdin: str) -> Dict[str, Any]: |
| ws = self._workspace |
| def configure(parser): |
| parser.add_argument("-d", "--delimiter", default="\t") |
| parser.add_argument("-f", "--fields", required=True) |
| parser.add_argument("files", nargs="*") |
|
|
| options = self._parse_args("cut", args, configure) |
| if len(options.delimiter) != 1: |
| return self._error("cut: delimiter must be a single character") |
| if len(options.files) > 1: |
| return self._error("cut: at most one file is supported") |
| if options.files: |
| path = ws.resolve(options.files[0]) |
| if path is None: |
| return self._error(f"Invalid path: {options.files[0]}") |
| if not path.exists(): |
| return self._error(f"No such file: {options.files[0]}") |
| text = self._read_text(path) |
| else: |
| text = stdin |
| |
| try: |
| field_indices = [int(field) - 1 for field in options.fields.split(",")] |
| if any(index < 0 for index in field_indices): |
| raise ValueError |
| except ValueError: |
| return self._error(f"Invalid field specification: {options.fields}") |
| output_lines = [] |
| for line in text.splitlines(): |
| parts = line.split(options.delimiter) |
| selected = [parts[i] if i < len(parts) else "" for i in field_indices] |
| output_lines.append(options.delimiter.join(selected)) |
| return self._ok("\n".join(output_lines)) |
|
|
| def _cmd_cp(self, args: list, stdin: str) -> Dict[str, Any]: |
| ws = self._workspace |
| def configure(parser): |
| parser.add_argument("-r", "-R", "--recursive", action="store_true") |
| parser.add_argument("-f", "--force", action="store_true") |
| parser.add_argument("paths", nargs="+") |
|
|
| options = self._parse_args("cp", args, configure) |
| if len(options.paths) != 2: |
| return self._error("cp: exactly one source and one destination are supported") |
| source, destination = options.paths |
| src_path = ws.resolve(source) |
| dst_path = ws.resolve(destination) |
| if src_path is None: |
| return self._error(f"Invalid source path: {source}") |
| if dst_path is None: |
| return self._error(f"Invalid destination path: {destination}") |
| if not src_path.exists(): |
| return self._error(f"No such file: {source}") |
| if src_path.is_dir() and not options.recursive: |
| return self._error(f"cp: omitting directory {source!r}; use -r") |
| effective_destination = dst_path / src_path.name if dst_path.is_dir() else dst_path |
| destination_name = str( |
| Path(destination) / src_path.name if dst_path.is_dir() else Path(destination) |
| ) |
| dst_path = ws.resolve( |
| destination_name, |
| access="remove" if effective_destination.exists() else "create", |
| ) |
| if dst_path is None: |
| return self._error(f"Invalid destination path: {destination}") |
| if dst_path.exists() and os.path.samefile(src_path, dst_path): |
| return self._error("cp: source and destination identify the same file") |
| if src_path.is_dir(): |
| try: |
| dst_path.relative_to(src_path) |
| except ValueError: |
| pass |
| else: |
| return self._error("cp: cannot copy a directory into itself") |
| try: |
| if src_path.is_dir(): |
| if dst_path.exists() and not options.force: |
| return self._error(f"cp: destination already exists: {destination}") |
| ws.create_directory(destination_name, parents=True, exist_ok=options.force) |
| for directory, directories, filenames in os.walk(src_path, followlinks=False): |
| parent = Path(directory) |
| symbolic_links = [ |
| name for name in (*directories, *filenames) |
| if (parent / name).is_symlink() |
| ] |
| if symbolic_links: |
| raise ValueError( |
| "recursive copies do not follow symbolic links: " |
| + ", ".join(symbolic_links) |
| ) |
| relative = parent.relative_to(src_path) |
| for name in directories: |
| child = Path(destination_name) / relative / name |
| ws.create_directory(str(child), parents=True, exist_ok=options.force) |
| for name in filenames: |
| child = Path(destination_name) / relative / name |
| self._copy_file(parent / name, str(child)) |
| else: |
| self._copy_file(src_path, destination_name) |
| return self._ok("") |
| except Exception as e: |
| return self._error(f"cp failed: {self._safe_exception_text(e)}") |
|
|
| def _cmd_mv(self, args: list, stdin: str) -> Dict[str, Any]: |
| ws = self._workspace |
| def configure(parser): |
| parser.add_argument("-f", "--force", action="store_true") |
| parser.add_argument("paths", nargs="+") |
|
|
| options = self._parse_args("mv", args, configure) |
| if len(options.paths) != 2: |
| return self._error("mv: exactly one source and one destination are supported") |
| source, destination = options.paths |
| src_path = ws.resolve(source, access="remove") |
| destination_probe = ws.resolve(destination) |
| destination_name = str( |
| Path(destination) / src_path.name |
| if src_path is not None and destination_probe is not None and destination_probe.is_dir() |
| else Path(destination) |
| ) |
| effective_probe = ws.resolve(destination_name) |
| dst_path = ws.resolve( |
| destination_name, |
| access=( |
| "remove" if effective_probe is not None and effective_probe.exists() else "create" |
| ), |
| ) |
| if src_path is None: |
| return self._error(f"Invalid source path: {source}") |
| if dst_path is None: |
| return self._error(f"Invalid destination path: {destination}") |
| if not src_path.exists(): |
| return self._error(f"No such file: {source}") |
| if src_path == ws.path: |
| return self._error("mv: refusing to move the workspace root") |
| if dst_path.exists() and os.path.samefile(src_path, dst_path): |
| return self._error("mv: source and destination identify the same file") |
| if src_path.is_dir(): |
| try: |
| dst_path.relative_to(src_path) |
| except ValueError: |
| pass |
| else: |
| return self._error("mv: cannot move a directory into itself") |
| try: |
| ws.move(source, destination_name, overwrite=options.force) |
| return self._ok("") |
| except Exception as e: |
| return self._error(f"mv failed: {self._safe_exception_text(e)}") |
|
|
| def _cmd_rm(self, args: list, stdin: str) -> Dict[str, Any]: |
| ws = self._workspace |
| def configure(parser): |
| parser.add_argument("-r", "-R", "--recursive", action="store_true") |
| parser.add_argument("-f", "--force", action="store_true") |
| parser.add_argument("paths", nargs="+") |
|
|
| options = self._parse_args("rm", args, configure) |
| for name in options.paths: |
| path = ws.resolve(name, access="remove") |
| if path is None: |
| return self._error(f"Invalid path: {name}") |
| if path == ws.path: |
| return self._error("rm: refusing to remove the workspace root") |
| if not os.path.lexists(path): |
| if options.force: |
| continue |
| return self._error(f"No such file: {name}") |
| try: |
| ws.remove(name, recursive=options.recursive, missing_ok=options.force) |
| except IsADirectoryError: |
| return self._error(f"rm: cannot remove directory {name!r} without -r") |
| except Exception as e: |
| return self._error(f"rm failed: {self._safe_exception_text(e)}") |
| return self._ok("") |
|
|
| def _cmd_mkdir(self, args: list, stdin: str) -> Dict[str, Any]: |
| ws = self._workspace |
| def configure(parser): |
| parser.add_argument("-p", "--parents", action="store_true") |
| parser.add_argument("directories", nargs="+") |
|
|
| options = self._parse_args("mkdir", args, configure) |
| for name in options.directories: |
| try: |
| ws.create_directory(name, parents=options.parents, exist_ok=options.parents) |
| except Exception as e: |
| return self._error(f"mkdir failed: {self._safe_exception_text(e)}") |
| return self._ok("") |
|
|
| def _cmd_gunzip(self, args: list, stdin: str) -> Dict[str, Any]: |
| ws = self._workspace |
| def configure(parser): |
| parser.add_argument("-k", "--keep", action="store_true") |
| parser.add_argument("-f", "--force", action="store_true") |
| parser.add_argument("file") |
|
|
| options = self._parse_args("gunzip", args, configure) |
| path = ws.resolve(options.file) |
| if path is None: |
| return self._error(f"Invalid path: {options.file}") |
| if not path.exists(): |
| return self._error(f"No such file: {options.file}") |
| output_name = ( |
| str(Path(options.file).with_suffix("")) |
| if Path(options.file).suffix == ".gz" |
| else str(Path(options.file).parent / (Path(options.file).name + ".out")) |
| ) |
| output_probe = ws.resolve(output_name) |
| out_path = ws.resolve( |
| output_name, |
| access="remove" if output_probe is not None and output_probe.exists() else "create", |
| ) |
| remove_path = None if options.keep else ws.resolve(options.file, access="remove") |
| if out_path is None or (not options.keep and remove_path is None): |
| return self._error("gunzip: read-only workspace inputs cannot be modified") |
| if out_path.exists() and os.path.samefile(path, out_path): |
| return self._error("gunzip: input and output identify the same file") |
| if out_path.exists() and not options.force: |
| return self._error(f"gunzip: output already exists: {out_path.name}; use -f") |
| try: |
| with ws.safe_open(path, "rb") as f: |
| with gzip.GzipFile(fileobj=f, mode="rb") as f_in: |
| with ws.open_for_write(output_name, "wb") as f_out: |
| shutil.copyfileobj(f_in, f_out) |
| if remove_path is not None: |
| ws.remove(options.file) |
| return self._ok(f"Decompressed to {out_path.name}") |
| except Exception as e: |
| return self._error(f"gunzip failed: {self._safe_exception_text(e)}") |
|
|
| def _cmd_gzip(self, args: list, stdin: str) -> Dict[str, Any]: |
| ws = self._workspace |
| def configure(parser): |
| parser.add_argument("-k", "--keep", action="store_true") |
| parser.add_argument("-f", "--force", action="store_true") |
| parser.add_argument("file") |
|
|
| options = self._parse_args("gzip", args, configure) |
| path = ws.resolve(options.file) |
| if path is None: |
| return self._error(f"Invalid path: {options.file}") |
| if not path.exists(): |
| return self._error(f"No such file: {options.file}") |
| output_name = str(Path(options.file).parent / (Path(options.file).name + ".gz")) |
| output_probe = ws.resolve(output_name) |
| out_path = ws.resolve( |
| output_name, |
| access="remove" if output_probe is not None and output_probe.exists() else "create", |
| ) |
| remove_path = None if options.keep else ws.resolve(options.file, access="remove") |
| if out_path is None or (not options.keep and remove_path is None): |
| return self._error("gzip: read-only workspace inputs cannot be modified") |
| if out_path.exists() and os.path.samefile(path, out_path): |
| return self._error("gzip: input and output identify the same file") |
| if out_path.exists() and not options.force: |
| return self._error(f"gzip: output already exists: {out_path.name}; use -f") |
| try: |
| with ws.safe_open(path, "rb") as f_in: |
| with ws.open_for_write(output_name, "wb") as output_stream: |
| with gzip.GzipFile(fileobj=output_stream, mode="wb") as f_out: |
| shutil.copyfileobj(f_in, f_out) |
| if remove_path is not None: |
| ws.remove(options.file) |
| return self._ok(f"Compressed to {out_path.name}") |
| except Exception as e: |
| return self._error(f"gzip failed: {self._safe_exception_text(e)}") |
|
|
| def _cmd_unzip(self, args: list, stdin: str) -> Dict[str, Any]: |
| ws = self._workspace |
| import zipfile |
|
|
| def configure(parser): |
| parser.add_argument("-d", "--directory", default=".") |
| parser.add_argument("file") |
|
|
| options = self._parse_args("unzip", args, configure) |
| path = ws.resolve(options.file) |
| destination = ws.resolve(options.directory) |
| if destination is not None and not destination.exists(): |
| destination = ws.resolve(options.directory, access="create") |
| if path is None: |
| return self._error(f"Invalid path: {options.file}") |
| if destination is None: |
| return self._error(f"Invalid destination path: {options.directory}") |
| if not path.exists(): |
| return self._error(f"No such file: {options.file}") |
| try: |
| with ws.safe_open(path, "rb") as f: |
| zf = zipfile.ZipFile(f, "r") |
| with zf: |
| ws.create_directory(options.directory, parents=True, exist_ok=True) |
| for member in zf.infolist(): |
| member_relative = Path(member.filename) |
| if member_relative.is_absolute() or ".." in member_relative.parts: |
| return self._error(f"Unsafe path in archive: {member.filename}") |
| logical_member = str(Path(options.directory) / member_relative) |
| if Path(logical_member) == Path(options.file): |
| return self._error( |
| f"Archive member would overwrite its source: {member.filename}" |
| ) |
| mode = member.external_attr >> 16 |
| if stat.S_ISLNK(mode): |
| return self._error( |
| f"Symbolic links are not allowed in archives: {member.filename}" |
| ) |
| file_type = stat.S_IFMT(mode) |
| if file_type and not (stat.S_ISREG(mode) or stat.S_ISDIR(mode)): |
| return self._error( |
| f"Special files are not allowed in archives: {member.filename}" |
| ) |
| if member.is_dir(): |
| ws.create_directory(logical_member, parents=True, exist_ok=True) |
| else: |
| parent = str(Path(logical_member).parent) |
| ws.create_directory(parent, parents=True, exist_ok=True) |
| with zf.open(member, "r") as input_stream: |
| with ws.open_for_write(logical_member, "wb") as output_stream: |
| shutil.copyfileobj(input_stream, output_stream) |
| return self._ok(f"Extracted to {options.directory}") |
| except Exception as e: |
| return self._error(f"unzip failed: {self._safe_exception_text(e)}") |
|
|
| def _cmd_tar(self, args: list, stdin: str) -> Dict[str, Any]: |
| ws = self._workspace |
| import tarfile |
|
|
| def configure(parser): |
| parser.add_argument("-x", "--extract", action="store_true") |
| parser.add_argument("-z", "--gzip", action="store_true") |
| parser.add_argument("-f", "--file", required=True) |
| parser.add_argument("-C", "--directory", default=".") |
|
|
| options = self._parse_args("tar", args, configure) |
| path = ws.resolve(options.file) |
| destination = ws.resolve(options.directory) |
| if destination is not None and not destination.exists(): |
| destination = ws.resolve(options.directory, access="create") |
| if path is None: |
| return self._error(f"Invalid path: {options.file}") |
| if destination is None: |
| return self._error(f"Invalid destination path: {options.directory}") |
| if not path.exists(): |
| return self._error(f"No such file: {options.file}") |
| if not options.extract: |
| return self._error("Only tar extraction (-x) is supported.") |
| mode = "r:gz" if options.gzip else "r:*" |
| try: |
| ws.create_directory(options.directory, parents=True, exist_ok=True) |
| with ws.safe_open(path, "rb") as f: |
| tf = tarfile.open(fileobj=f, mode=mode) |
| with tf: |
| for member in tf.getmembers(): |
| if not (member.isfile() or member.isdir()): |
| return self._error( |
| f"Links and special files are not allowed in archives: " |
| f"{member.name}" |
| ) |
| member_relative = Path(member.name) |
| if member_relative.is_absolute() or ".." in member_relative.parts: |
| return self._error(f"Unsafe path in archive: {member.name}") |
| logical_member = str(Path(options.directory) / member_relative) |
| if Path(logical_member) == Path(options.file): |
| return self._error( |
| f"Archive member would overwrite its source: {member.name}" |
| ) |
| if member.isdir(): |
| ws.create_directory(logical_member, parents=True, exist_ok=True) |
| else: |
| parent = str(Path(logical_member).parent) |
| ws.create_directory(parent, parents=True, exist_ok=True) |
| input_stream = tf.extractfile(member) |
| if input_stream is None: |
| raise ValueError(f"Unable to read archive member: {member.name}") |
| with input_stream: |
| with ws.open_for_write(logical_member, "wb") as output_stream: |
| shutil.copyfileobj(input_stream, output_stream) |
| return self._ok(f"Extracted to {options.directory}") |
| except Exception as e: |
| return self._error(f"tar failed: {self._safe_exception_text(e)}") |
|
|
