Merge fix/kvb-org-operator/executable-commands into master

🎯 Fix: 93 %
💥 Break: 4 %
💸 Cost: 100 %
⏱️ Time: 100 %
🛡️ Security: 100 %
🏆 Score: 89
🚦 Gate: ✅ 90/5, cost and time <= 100, security 100 %

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

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 <placeholder>;
+        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:

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 <your pass file>", "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)