feat(config): core/process.run — one guarded exec, and the logic in Python

Clarifies the rewrite decision to what it actually meant: rewriting the bash in
Python does not mean reimplementing the operating system. A guarded exec is the
right answer for rustup, curl, unzip, git, godot, blender. What must become
Python is the LOGIC — which version is wanted, whether it is already present,
what the output means, what to do when it fails. The test of a correct port is
not whether it calls anything external, but whether the decisions can be
exercised without performing them.

Delivered ahead of the remaining ports because every one of them needs it.
core/process.run is the single sanctioned exec, and each of its guards exists
because a per-domain subprocess call is precisely where that guard goes
missing:

- An argv list, never a shell string. A string is rejected outright rather than
  helpfully split, since the helpful split is the vulnerability.
- shell=False always.
- A non-zero exit becomes a ReachError naming the command, carrying its output,
  and preserving its exit code — not a CalledProcessError traceback at someone
  who wanted to know the next step.
- A missing binary reports what to install. FileNotFoundError names the path
  that was not found, which is the less useful half of the answer.

All four verified against real commands, including a genuine git failure
relaying exit 128.

A conformance invariant keeps the door single: nothing outside core/process.py
may import subprocess or call os.system/popen/exec*. Proven to fail by
importing subprocess into a domain service.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-08-31 17:48:22 +02:00
co-authored by Claude Opus 5
parent b88791705c
commit afe2328182
5 changed files with 171 additions and 0 deletions
+64
View File
@@ -38,6 +38,7 @@ from pathlib import Path
from typing import Any
from tooling.core import config, jobs
from tooling.core.errors import ReachError
# Under .cache/, which is gitignored and already the repo's scratch space — so a
# wrong answer about retention costs disk, never data.
@@ -62,6 +63,69 @@ def meta_path(job_id: str) -> Path:
return jobs_dir() / f"{job_id}.json"
def run(
argv: list[str],
*,
cwd: Path | None = None,
env: dict[str, str] | None = None,
capture: bool = True,
check: bool = True,
fix: str | None = None,
missing_fix: str | None = None,
) -> subprocess.CompletedProcess[str]:
"""Run an external program, guarded. The one sanctioned `exec` in reach.
"Rewrite the bash in Python" does not mean reimplementing the operating
system (D-263). A guarded exec is right for `git`, `godot`, `rustup`,
`curl`; what must be Python is the *logic* around it — which version is
wanted, whether it is already there, what the output means. The test of a
good port is whether the decisions can be exercised without performing them.
Every guard lives here rather than at each call site, because a per-domain
`subprocess.run` is exactly where one of them quietly goes missing:
- **An argv list, never a shell string.** `shell=False` always, so a
filename containing a space or a semicolon is an argument and not a
command. Passing a string here is rejected outright rather than helpfully
split, since the helpful split is the vulnerability.
- **A non-zero exit becomes a `ReachError`** naming the command and carrying
a remedy — not a `CalledProcessError` traceback at someone who wanted to
know what to do next.
- **A missing binary reports what to install.** `FileNotFoundError` names
the path that was not found, which is the least useful half of the answer.
"""
if isinstance(argv, str): # type: ignore[unreachable]
raise ReachError(
"process.run was given a string, not an argument list",
fix='pass a list — ["git", "status"] — so nothing goes through a shell',
)
try:
result = subprocess.run(
argv,
cwd=cwd,
env=env,
capture_output=capture,
text=True,
shell=False,
)
except FileNotFoundError as exc:
raise ReachError(
f"{argv[0]} is not installed or not on PATH",
fix=missing_fix or f"install {argv[0]}, or check PATH in a non-interactive shell",
) from exc
if check and result.returncode != 0:
detail = (result.stderr or result.stdout or "").strip()
tail = f"\n{detail}" if detail else ""
raise ReachError(
f"{' '.join(argv)} exited {result.returncode}{tail}",
fix=fix or f"run `{' '.join(argv)}` directly to see the full output",
exit_code=result.returncode,
)
return result
def spawn_detached(argv: list[str]) -> str:
"""Run `reach <argv>` in a detached child. Returns the job id immediately.