diff --git a/skills/kvb-org-operator/scripts/operator.py b/skills/kvb-org-operator/scripts/operator.py index dd84185..79f73d1 100755 --- a/skills/kvb-org-operator/scripts/operator.py +++ b/skills/kvb-org-operator/scripts/operator.py @@ -27,7 +27,11 @@ Args: decision a decision with no command to run: listed in the table only, never in a block or `run`; only `done` closes it. Paths are absolute; no argument holds a shell metacharacter; the - command is one line without control characters. + command is one line without control characters. Unless the check is + decision, the command starts, after VAR=value words, with a program + on PATH, a path to one or a shell builtin and holds no ; + an alias, a slash command or prose is refused, and an older row + holding one stays in the table only. list [--json]: run the check of every queued or blocked operator row and set the passing ones done. run: walk the open rows that hold a directory and a command, in list @@ -129,6 +133,9 @@ OPEN = ("queued", "blocked") META = re.compile(r"[\s|&;()<>$`\\\"'*?\[\]{}~!#]") NAME = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._+-]*$") COMMIT = re.compile(r"\bcommit\b") +ASSIGNMENT = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*=") +SLOT = re.compile(r"<[A-Za-z][^<>]*>") +BUILTINS = frozenset({".", ":", "cd", "exec", "exit", "export", "set", "source", "umask"}) HEADER = "| # | Directory | Command | Why | Check |" GIT = ["git", "-c", "core.fsmonitor=false", "-c", "core.hooksPath=/dev/null", "-c", "log.showSignature=false", "--no-pager", "--no-optional-locks"] DIFF = [*GIT, "diff", "--no-ext-diff", "--no-textconv", "--ignore-submodules=all"] @@ -188,6 +195,19 @@ def problem(check: str) -> str | None: return path_problem(arg, kind) +def unrunnable(command: str, directory: str) -> str | None: + if SLOT.search(command): + return "holds a <...> slot to fill in" + try: + words = [word for word in shlex.split(command) if not ASSIGNMENT.match(word)] + except ValueError: + return "does not parse as a shell command" + program = os.path.expanduser(next((word.lstrip("(") for word in words if word.lstrip("(")), "")) + if program in BUILTINS or shutil.which(os.path.join(directory, program) if "/" in program else program): + return None + return f"names no program on PATH ({program!r})" + + def git_env() -> dict[str, str]: return {**os.environ, "GIT_NO_LAZY_FETCH": "1"} @@ -249,6 +269,9 @@ def cmd_add(rasci: ModuleType, args: argparse.Namespace) -> int: fail("--command and --reason must not be empty") if not command.isprintable(): fail(f"--command {visible(command, '')!r} holds a control character; record one line") + refused = None if args.check == "decision" else unrunnable(command, args.dir) + if refused: + fail(f"--command {command!r} {refused}; write the real command, or record the step with --check decision") council = module("council", rasci.COUNCIL) sid, _ = council.caller(args.session) party = council.live_sessions().get(sid) or sid @@ -332,7 +355,10 @@ def cmd_list(rasci: ModuleType, args: argparse.Namespace) -> int: def runnable(row: dict) -> bool: - return bool(row["directory"] and row["command"]) and row["command"].isprintable() and row["check"] != "decision" + return ( + bool(row["directory"] and row["command"]) and row["command"].isprintable() and row["check"] != "decision" + and not unrunnable(row["command"], row["directory"]) + ) def visible(text: str, keep: str) -> str: diff --git a/tests/test_operator.py b/tests/test_operator.py index e5b9685..90b1f0a 100644 --- a/tests/test_operator.py +++ b/tests/test_operator.py @@ -126,6 +126,9 @@ class TestOperator(unittest.TestCase): return json.loads((self.root / "rasci.json").read_text(encoding="utf-8")) def test_checks_close_rows_once_their_condition_holds(self): + for name in ("pipx", "git-sign-push"): + (self.bin / name).write_text("#!/bin/sh\n", encoding="utf-8") + (self.bin / name).chmod(0o755) repo = self._repo("repo", upstream=True) src, dst = self.base / "src", self.base / "dst" (src / "sub").mkdir(parents=True) @@ -227,6 +230,25 @@ class TestOperator(unittest.TestCase): [HEADER, "|---|---|---|---|---|", "| 1 | - | sudo reboot | - | manual |"], ) + def test_add_refuses_a_command_no_shell_can_run_unless_it_is_a_decision(self): + for command in ("Edit pcs.yml: add the group", "/kvb-skill-cleanup", "deploy -p ", "gipsn"): + result = self._run("add", "x", "--dir", str(self.base), "--command", command, "--reason", "r", + "--check", "manual") + self.assertEqual(result.returncode, 2, command) + self.assertIn("record the step with --check decision", result.stderr) + self._add("subshell", self.base, "(umask 077; touch secret)", "r", "manual") + self._add("env", self.base, "GIT_CONFIG_COUNT=1 git status", "r", "manual") + self._add("ask", self.base, "Edit pcs.yml: add the group", "r", "decision") + self.assertEqual(self._open(), ["operator/subshell", "operator/env", "operator/ask"]) + + def test_an_older_prose_row_stays_out_of_blocks_and_run(self): + result = self._run("set", "operator/prose", "--title", "Rotate the credential", "--workdir", str(self.base), + "--status", "queued", script=RASCI) + self.assertEqual(result.returncode, 0, result.stderr) + self._add("go", self.base, "true", "runnable", "manual") + self.assertEqual(self._out("list")[4:], [*self._block(2, "go", self.base, "true"), *RUNNER]) + self.assertNotIn("Rotate", self._feed("y\n").stdout) + def test_bash_block_skips_legacy_rows_without_directory(self): result = self._run("set", "operator/legacy", "--title", "Rotate leaked credentials", "--status", "queued", script=RASCI)