Markus Fritsche
3027ad4ccf
The model swap (4b-npu -> qwen3.6-coding) retired qwen3-4b-2507-npu from the proxy, silently orphaning _llm_pick: any keyword-less body fell through to a dead model and got the roster instead of a route. Repoint MODEL to the standing local model and make ENGINE/MODEL env-overridable so the next swap cannot orphan it again. test_show_shell_safe: the two keyword-less bodies made a shell-safety assertion hinge on live LLM inference (flapped the moment the model went away). Give each a routing keyword so it routes deterministically via keyword-match — the adversarial payload (nested $(...), backslash+metachars) is preserved exactly; the test is now hermetic. 120 passing. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
149 lines
6.0 KiB
Python
149 lines
6.0 KiB
Python
"""Reviewer finding #7 — bin/bullpen-dispatcher _show() must produce a SHELL-SAFE
|
|
copy-paste command.
|
|
|
|
_show() sanitises a request body into the `sic hertz ...` command the dispatcher
|
|
tells the user to run. It strips double-quotes and newlines but NOT backticks,
|
|
`$(...)` command substitution, or backslashes — so a body that merely *quotes*
|
|
third-party text turns the suggested command into a shell injection the user is
|
|
instructed to paste and run.
|
|
|
|
These tests are BLACK-BOX. For each payload they:
|
|
1. run the dispatcher one-shot (`--once <body>`) and grab the suggested `sic …` line,
|
|
2. execute that line exactly as a copy-pasting user would — but with `sic` replaced
|
|
by a harmless recorder stub, so nothing ever reaches the fleet,
|
|
3. assert the injected `touch <marker>` did NOT fire (no live command substitution
|
|
/ quote breakout) and the literal body text survived to `sic` inert.
|
|
|
|
They FAIL on the current code (the payload executes / the command won't parse) and
|
|
pass only once _show() properly shell-quotes the body.
|
|
"""
|
|
import os
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
REPO = Path(__file__).resolve().parents[1]
|
|
DISPATCHER = REPO / "bin" / "bullpen-dispatcher"
|
|
LIB = REPO / "lib"
|
|
BASH = shutil.which("bash") or "/bin/bash"
|
|
|
|
|
|
def _suggested_command(body):
|
|
"""Run `bullpen-dispatcher --once <body>` and return the emitted `sic …` line."""
|
|
env = dict(os.environ, PYTHONPATH=str(LIB))
|
|
proc = subprocess.run(
|
|
[sys.executable, str(DISPATCHER), "--once", body],
|
|
capture_output=True, text=True, env=env, timeout=60,
|
|
)
|
|
assert proc.returncode == 0, f"dispatcher --once failed: {proc.stderr}"
|
|
cmds = [ln.strip() for ln in proc.stdout.splitlines() if re.match(r"\s*sic\b", ln)]
|
|
assert cmds, f"no suggested `sic` command in dispatcher output:\n{proc.stdout}"
|
|
return cmds[-1]
|
|
|
|
|
|
def _run_paste(make_body):
|
|
"""make_body(marker) -> a request body that, IF the suggested command is unsafe,
|
|
runs `touch <marker>`.
|
|
|
|
Returns (marker_fired, sic_argv, body, command):
|
|
marker_fired True if the injected command executed (the bug),
|
|
sic_argv the argument list the pasted command actually delivered to `sic`
|
|
(None if `sic` was never reached, e.g. an unterminated quote),
|
|
body/command the exact strings involved, for assertion messages.
|
|
"""
|
|
work = Path(tempfile.mkdtemp(prefix="showsafe_"))
|
|
try:
|
|
marker = work / "PWNED"
|
|
body = make_body(str(marker))
|
|
command = _suggested_command(body)
|
|
|
|
bindir = work / "bin"
|
|
bindir.mkdir()
|
|
argv_file = work / "sic_argv"
|
|
stub = bindir / "sic"
|
|
stub.write_text(
|
|
"#!/usr/bin/env bash\n"
|
|
f'printf "%s\\0" "$@" > "{argv_file}"\n'
|
|
"exit 0\n"
|
|
)
|
|
stub.chmod(0o755)
|
|
|
|
script = work / "paste.sh"
|
|
script.write_text(command + "\n")
|
|
env = dict(os.environ, PATH=f"{bindir}:/usr/bin:/bin")
|
|
subprocess.run([BASH, str(script)], cwd=str(work), env=env,
|
|
capture_output=True, text=True, timeout=60)
|
|
|
|
argv = None
|
|
if argv_file.exists():
|
|
parts = argv_file.read_bytes().split(b"\0")[:-1]
|
|
argv = [p.decode("utf-8", "surrogateescape") for p in parts]
|
|
return marker.exists(), argv, body, command
|
|
finally:
|
|
shutil.rmtree(work, ignore_errors=True)
|
|
|
|
|
|
def _assert_safe(fired, argv, body, command):
|
|
assert not fired, (
|
|
f"INJECTION EXECUTED: the suggested command ran the injected payload.\n"
|
|
f" suggested: {command}"
|
|
)
|
|
assert argv is not None, (
|
|
f"the suggested command did not parse / never reached `sic` "
|
|
f"(corrupted quoting).\n suggested: {command}"
|
|
)
|
|
assert body in "".join(argv), (
|
|
f"the request body was not delivered to `sic` as inert literal text "
|
|
f"(it was mangled or partly executed).\n"
|
|
f" body: {body!r}\n sic argv: {argv!r}\n suggested: {command}"
|
|
)
|
|
|
|
|
|
def test_backtick_command_substitution_does_not_execute():
|
|
# a body quoting third-party text that happens to contain backticks
|
|
fired, argv, body, cmd = _run_paste(
|
|
lambda m: f"please fetch https://x.test and summarize: `touch {m}`")
|
|
_assert_safe(fired, argv, body, cmd)
|
|
|
|
|
|
def test_dollar_paren_command_substitution_does_not_execute():
|
|
fired, argv, body, cmd = _run_paste(
|
|
lambda m: f"review this code $(touch {m})")
|
|
_assert_safe(fired, argv, body, cmd)
|
|
|
|
|
|
def test_nested_dollar_paren_command_substitution_does_not_execute():
|
|
# a routing keyword ("fetch"/url) forces a deterministic keyword-match route (no live LLM),
|
|
# so this stays a hermetic test of _show()'s escaping of NESTED command substitution.
|
|
fired, argv, body, cmd = _run_paste(
|
|
lambda m: f"fetch https://x.test and look at $(echo $(touch {m}))")
|
|
_assert_safe(fired, argv, body, cmd)
|
|
|
|
|
|
def test_trailing_backslash_does_not_break_the_quoting():
|
|
# a trailing backslash escapes the naive closing double-quote, leaving the
|
|
# copy-paste command unterminated / corrupt (marker independent of body).
|
|
fired, argv, body, cmd = _run_paste(
|
|
lambda m: "please fetch the url now \\")
|
|
_assert_safe(fired, argv, body, cmd)
|
|
|
|
|
|
def test_backslash_then_shell_metachars_do_not_execute():
|
|
# backslash-escaped quote followed by shell code: if the backslash survives it
|
|
# neutralises the closing quote and the trailing command runs. The "review this code"
|
|
# prefix is a routing keyword so the body routes deterministically (no live LLM).
|
|
fired, argv, body, cmd = _run_paste(
|
|
lambda m: f"review this code \\ ; touch {m} ; echo done")
|
|
_assert_safe(fired, argv, body, cmd)
|
|
|
|
|
|
def test_debate_conversant_path_is_also_shell_safe():
|
|
# the debate branch embeds the SAME _show(body) into a `room_say … body="…"`
|
|
# command, so it carries the identical exposure.
|
|
fired, argv, body, cmd = _run_paste(
|
|
lambda m: f"let's debate the design `touch {m}`")
|
|
_assert_safe(fired, argv, body, cmd)
|