runs.py
Rama main · réplica de lectura del repositorio original.
"""
What a command runs underneath, read from the files on this machine before it runs, so the gateway can check those
commands too: `make clean` is only as safe as the `clean` target's recipe.
make [TARGET...] the recipe lines of each target and the targets it depends on (Makefile, -f, -C)
npm run X / npm test / ... the package.json script, with its pre/post scripts (npm, pnpm, yarn, bun)
a script, however it's run its lines: bash x.sh, ./x.sh, /abs/x.sh, source x.sh, . x.sh, exec ./x.sh,
sudo bash x.sh, pwsh -File x.ps1 (shell and PowerShell scripts; any name or extension)
Used by the hooks (claude_hook.py, agent_hook.py), which send the result as metadata["runs"]:
[{"via": "Makefile target `clean`", "lines": ["rm -rf build/ ~/"]}]
The gateway reads each line like the command itself (server.py command_signals). It can only make a decision
stricter: what's found here is checked in addition to the command, never instead of it.
Only reads files, never runs anything. Bounded: MAX_LINES lines and MAX_FILE_BYTES per file; anything it can't read
is left out. Scripts in other languages (python x.py, node x.js, ruby x.rb, ...): the commands they hand to a shell,
when written out in the file: os.system("..."), subprocess.run("..." or [...]), child_process.exec/execSync("..."),
system("...") and `...` in Ruby, Perl and PHP. A command built at run time can't be read here; if the agent wrote or
changed the file in this conversation, running it waits for a person (server.written_then_run). Also not read:
recipes built at run time ($(shell ...), eval), make variables beyond simple `NAME = value` definitions.
`include` files are followed (a few levels deep, cycles skipped) and conditionals are read on BOTH sides, so a
recipe hidden in an included file or in either branch of an `ifeq` is still checked.
"""
from __future__ import annotations
import json
import os
import re
from pathlib import Path
import commands
MAX_LINES = 60
MAX_FILE_BYTES = 256 * 1024
MAKE_PROGRAMS = {"make", "gmake", "mingw32-make"}
NODE_PM = {"npm", "pnpm", "yarn", "bun"}
NPM_SHORTCUTS = {"test", "start", "stop", "restart"} # `npm test` runs scripts.test
SHELLS = {"bash", "sh", "zsh", "dash"}
_VAR_DEF = re.compile(r"^([A-Za-z_][A-Za-z0-9_.-]*)\s*(?:\?|:|::)?=\s*(.*)$")
_RULE = re.compile(r"^([^:=#\t][^:=#]*?)\s*::?(?!=)\s*(.*)$")
_VAR_USE = re.compile(r"\$\(([A-Za-z_][A-Za-z0-9_.-]*)\)|\$\{([A-Za-z_][A-Za-z0-9_.-]*)\}")
_COND = re.compile(r"^(ifeq|ifneq|ifdef|ifndef)\b") # both branches are read, see _parse
_COND_ELSE = re.compile(r"^else\b")
_COND_END = re.compile(r"^endif\b")
_INCLUDE = re.compile(r"^(-?)include\s+(.*)$")
MAX_INCLUDE_DEPTH = 3 # "at most a few levels deep"
def _read(path: Path) -> str | None:
try:
if path.is_file() and path.stat().st_size <= MAX_FILE_BYTES:
return path.read_text(encoding="utf-8", errors="replace")
except OSError:
pass
return None
# --------------------------------------------------------------------------- make
def _parse(text: str, folder: Path, depth: int, seen: set[Path]) -> tuple[dict[str, tuple[list[str], list[str]]], dict[str, str], str | None]:
"""-> (rules, simple variables, the default target). `include` files are followed and conditionals are read
on BOTH sides: `ifeq ... else ... endif` keeps the recipe lines of either branch rather than working out
which one make would pick. That can only make a gateway decision stricter — it holds what any branch
would run — never looser.
"""
text = re.sub(r"[ \t]*\\\r?\n[ \t]*", " ", text) # joined continuation lines
variables: dict[str, str] = {}
rules: dict[str, tuple[list[str], list[str]]] = {}
includes: list[str] = []
default, current = None, []
cond_depth = 0
for raw in text.splitlines():
if raw.startswith("\t"):
line = raw[1:].strip()
if current and line and not line.startswith("#"):
for t in current:
rules[t][1].append(line) # either branch is kept, see _parse
continue
line = raw.split("#", 1)[0].rstrip()
stripped = line.strip()
if not stripped:
continue
if _COND.match(stripped):
cond_depth += 1
continue
if _COND_END.match(stripped):
cond_depth = max(0, cond_depth - 1)
continue
if _COND_ELSE.match(stripped):
continue # the other branch keeps the same target
if cond_depth == 0 and (m := _INCLUDE.match(stripped)):
includes += m.group(2).split()
current = []
continue
if m := _VAR_DEF.match(stripped):
variables[m.group(1)] = m.group(2).strip()
current = []
continue
if m := _RULE.match(line):
targets = [t for t in m.group(1).split() if t]
deps, _, inline = m.group(2).partition(";")
current = [t for t in targets if not t.startswith(".")]
for t in current:
rules.setdefault(t, ([], []))
rules[t][0].extend(deps.split())
if inline.strip():
rules[t][1].append(inline.strip())
if default is None and current and "%" not in current[0]:
default = current[0]
continue
current = []
for name in includes:
if depth >= MAX_INCLUDE_DEPTH: # "at most a few levels deep"
break
path = folder / name
if not path.is_file(): # a missing -include is ignored by make too
continue
try:
resolved = path.resolve()
except OSError:
continue
if resolved in seen: # include cycle: stop, don't hang the hook
continue
seen.add(resolved)
included = _read(path)
if included is None:
continue
inc_rules, inc_vars, inc_default = _parse(included, path.parent, depth + 1, seen)
for t, (deps, recipe) in inc_rules.items(): # the includer's own rules win, both are kept
if t not in rules:
rules[t] = (deps, recipe)
else:
rules[t][0].extend(d for d in deps if d not in rules[t][0])
rules[t][1].extend(recipe)
variables.update(inc_vars)
if default is None and inc_default:
default = inc_default
return rules, variables, default
def parse_makefile(text: str, folder: Path) -> tuple[dict[str, tuple[list[str], list[str]]], str | None]:
"""-> ({target: (prerequisites, recipe lines)}, the default target). Simple variables are expanded.
`folder` is the Makefile's own directory: `include` paths are resolved against it.
"""
rules, variables, default = _parse(text, folder, 0, set())
def expand(s: str, depth: int = 0) -> str:
if depth > 5:
return s
out = _VAR_USE.sub(lambda m: expand(variables.get(m.group(1) or m.group(2), m.group(0)), depth + 1), s)
return out.replace("$$", "$")
return ({t: ([expand(d) for d in deps], [expand(r) for r in recipe]) for t, (deps, recipe) in rules.items()},
default)
def _recipe_line(line: str) -> str:
return line.lstrip("@-+ \t") # @ silent, - ignore errors, + always run
def _make(words: list[str], cwd: Path) -> list[dict]:
directory, makefile, targets, args = cwd, None, [], iter(words[1:])
for w in args:
if w in ("-C", "--directory"):
directory = directory / next(args, ".")
elif w.startswith("--directory="):
directory = directory / w.split("=", 1)[1]
elif w in ("-f", "--file", "--makefile"):
makefile = next(args, None)
elif w.startswith(("--file=", "--makefile=")):
makefile = w.split("=", 1)[1]
elif w.startswith("-C") and len(w) > 2:
directory = directory / w[2:]
elif w.startswith("-f") and len(w) > 2:
makefile = w[2:]
elif not w.startswith("-") and "=" not in w:
targets.append(w)
if makefile:
path = directory / makefile
else: # make's own order, by the names really on disk (Windows would match "makefile" to "Makefile")
try:
names = set(os.listdir(directory))
except OSError:
return []
path = next((directory / n for n in ("GNUmakefile", "makefile", "Makefile") if n in names), None)
if path is None:
return []
text = _read(path)
if text is None:
return []
rules, default = parse_makefile(text, path.parent)
found: list[dict] = []
seen: set[str] = set()
def visit(t: str, depth: int) -> None:
if t in seen or depth > 8 or t not in rules:
return
seen.add(t)
deps, recipe = rules[t]
for d in deps:
visit(d, depth + 1)
if recipe:
found.append({"via": f"{path.name} target `{t}`", "lines": [_recipe_line(r) for r in recipe]})
for t in targets or ([default] if default else []):
visit(t, 0)
return found
# --------------------------------------------------------------------------- package.json scripts
def _node(words: list[str], cwd: Path) -> list[dict]:
prog, rest = words[0], [w for w in words[1:] if not w.startswith("-")]
if (not rest and prog == "yarn") or rest[:1] in (["install"], ["i"], ["ci"]):
if "--ignore-scripts" in words:
return []
# installing runs the project's own lifecycle scripts (a dependency's can't be read before it's downloaded)
text = _read(cwd / "package.json")
try:
scripts = json.loads(text).get("scripts") if text else None
except (ValueError, AttributeError):
return []
return [{"via": f"package.json script `{k}` (runs on install)", "lines": [scripts[k]]}
for k in ("preinstall", "install", "postinstall", "prepare")
if isinstance(scripts, dict) and isinstance(scripts.get(k), str)]
if not rest:
return []
if rest[0] in ("run", "run-script"):
name = rest[1] if len(rest) > 1 else None
elif rest[0] in NPM_SHORTCUTS or prog in ("yarn", "pnpm", "bun"):
name = rest[0] # yarn build, pnpm lint, bun dev
else:
return []
text = _read(cwd / "package.json")
try:
scripts = json.loads(text).get("scripts") if text else None
except (ValueError, AttributeError):
return []
if not isinstance(scripts, dict) or not isinstance(scripts.get(name), str):
return []
found = []
for key in (f"pre{name}", name, f"post{name}"):
if isinstance(scripts.get(key), str):
found.append({"via": f"package.json script `{key}`", "lines": [scripts[key]]})
return found
# --------------------------------------------------------------------------- shell scripts
OTHER_LANGUAGES = re.compile(r"(?i)\.(py|pyw|js|mjs|cjs|ts|tsx|jsx|rb|pl|php|lua|r)$") # not read as shell lines
# A command handed to a shell, written out: os.system("curl URL | sh"), subprocess.run(["sh", "-c", "..."]), execSync(`...`)
_SHELL_CALL = re.compile(r"\b(?:os\.(?:system|popen)|subprocess\.(?:run|call|check_call|check_output|Popen|getoutput|"
r"getstatusoutput)|(?:child_process\.)?(?:execSync|exec|spawnSync|spawn|execFileSync|execFile)|"
r"system|shell_exec|passthru|popen|proc_open|Open3\.\w+|IO\.popen|Kernel\.system)\s*\(\s*"
r"(?:[frbu]{0,2}(?P<q>[\"'`])(?P<s>(?:\\.|(?!(?P=q)).)*)(?P=q)|\[(?P<l>[^\]]*)\])", re.S | re.I)
_STRING = re.compile(r"(?P<q>[\"'])(?P<s>(?:\\.|(?!(?P=q)).)*)(?P=q)")
_BACKTICKS = re.compile(r"`([^`\n]+)`|%x\{([^}\n]+)\}") # Ruby, Perl, PHP
def _shell_outs(path_word: str, cwd: Path) -> list[dict]:
"""The commands a Python, Node, Ruby, Perl or PHP file hands to a shell, when they're written out in it."""
p = Path(os.path.expanduser(path_word.strip("'\"")))
p = p if p.is_absolute() else cwd / p
text = _read(p)
if text is None:
return []
lines = []
for m in _SHELL_CALL.finditer(text):
if m.group("s") is not None:
lines.append(m.group("s"))
else: # ["sh", "-c", "curl URL | sh"] or ["curl", url]: the words as a line
words = [w.group("s") for w in _STRING.finditer(m.group("l"))]
lines.append(words[-1] if len(words) >= 3 and words[1] in ("-c", "/c") else " ".join(words))
if p.suffix.lower() in (".rb", ".pl", ".php"):
lines += [a or b for a, b in _BACKTICKS.findall(text)]
lines = [l.strip() for l in lines if l.strip()]
return ([{"via": f"commands {p.name} hands to a shell", "lines": lines}] if lines else []) + _api_delete(text, p.name)
def _api_delete(text: str, name: str) -> list[dict]:
"""A GraphQL delete in a file (requests.post / fetch with mutation { volumeDelete(...) }, a curl -d @body.json),
as a curl line the gateway holds. Only the matched words leave this machine, never the file (it may hold keys)."""
m = commands.API_DELETE.search(text)
if not m:
return []
# only the operation's name: the match can span whatever sits between `mutation` and `{` (variables, defaults)
op = re.search(r"\{\s*([$\w{}]+)\s*\($", m.group(0))
sent = f"mutation {{ {op.group(1)}(" if op else m.group(0)[:80].replace("'", "")
return [{"via": f"the API request in {name}", "lines": [f"curl -d '{sent}'"]}]
def _request_bodies(line: str, cwd: Path) -> list[dict]:
"""Files an HTTP call sends: curl -d @q.json, --data-binary @q.json, cat q.json | curl -d @-, curl -T q.json."""
try:
reading = commands.read(line)
except Exception:
return []
if not any(c.program in commands.HTTP_CLIS for c in reading.commands):
return []
out = []
for word in dict.fromkeys(w.lstrip("@") for c in reading.commands for w in c.words[1:]):
if not word or word.startswith("-") or "://" in word:
continue
p = Path(os.path.expanduser(word.strip("'\"")))
text = _read(p if p.is_absolute() else cwd / p)
out += _api_delete(text, p.name) if text else []
return out
def _script(path_word: str, cwd: Path) -> list[dict]:
p = Path(os.path.expanduser(path_word.strip("'\"")))
p = p if p.is_absolute() else cwd / p
text = _read(p)
if text is None:
return []
lines = [l.strip() for l in text.splitlines() if l.strip() and not l.strip().startswith("#")]
return [{"via": f"script {p.name}", "lines": lines}] if lines else []
# --------------------------------------------------------------------------- entry point
def expand(line: str, cwd: str | None = None) -> list[dict]:
"""What the command line runs underneath, as [{"via": ..., "lines": [...]}]. Empty when there's nothing to read."""
base = Path(cwd or os.getcwd())
out: list[dict] = []
try:
reading = commands.read(line)
except Exception:
return out
for cmd in reading.commands:
words = cmd.words
prog = cmd.program
try:
if prog in MAKE_PROGRAMS:
out += _make(words, base)
elif prog in NODE_PM:
out += _node(words, base)
except Exception:
continue
sourced = [c.words[1] for c in reading.commands if c.program in ("source", ".") and len(c.words) > 1]
for path_word in dict.fromkeys(reading.scripts + sourced):
out += _shell_outs(path_word, base) if OTHER_LANGUAGES.search(path_word) else _script(path_word, base)
for inner in [line] + [l for item in out if item["via"].startswith("script ") for l in item["lines"]]:
out += _request_bodies(inner, base)
total, kept = 0, []
for item in out: # bounded, in order
lines = [l[:1000] for l in item["lines"] if l][: MAX_LINES - total]
if lines:
kept.append({"via": item["via"][:200], "lines": lines})
total += len(lines)
if total >= MAX_LINES:
break
return kept