From 32aea8b9cec72aae22af3f27c88c2e563a98db02 Mon Sep 17 00:00:00 2001 From: lostlight530 <13957086206@163.com> Date: Tue, 27 Oct 2026 07:47:56 +0800 Subject: [PATCH 1/2] test(tool-cli): reject output collisions with input evidence and request files --- tests/test_tool_cli.py | 40 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 40 insertions(+) diff --git a/tests/test_tool_cli.py b/tests/test_tool_cli.py index fc12bca..39f2d32 100644 --- a/tests/test_tool_cli.py +++ b/tests/test_tool_cli.py @@ -80,6 +80,46 @@ def test_cli_invalid_arguments_machine_failure(self): self.assertEqual(parsed["error"]["code"], "INVALID_ARGUMENT") + def test_cli_output_cannot_replace_source_or_request_inputs(self): + with tempfile.TemporaryDirectory() as temp: + source = Path(temp) / "source.json" + source.write_text("{\"profile\":\"sci-render-kit/figure-evidence\"}", encoding="utf-8") + original = source.read_bytes() + arguments = {"figure_evidence": str(source)} + request_file = Path(temp) / "request.json" + request_file.write_text(json.dumps(arguments), encoding="utf-8") + request_original = request_file.read_bytes() + base = [sys.executable, str(ROOT / "tool_cli.py"), "invoke", "communication_transfer"] + for options, destination in ( + (["--args", json.dumps(arguments)], source), + (["--args-file", str(request_file)], request_file), + ): + completed = subprocess.run( + base + options + ["--output", str(destination)], + cwd=ROOT, text=True, capture_output=True, + ) + self.assertEqual(completed.returncode, 2, completed.stderr) + self.assertIn("must not overwrite an input file", completed.stderr) + self.assertEqual(source.read_bytes(), original) + self.assertEqual(request_file.read_bytes(), request_original) + + plan_file = Path(temp) / "plan.json" + plan_file.write_text(json.dumps({"steps": [ + {"tool": "communication_transfer", "arguments": arguments} + ]}), encoding="utf-8") + plan_original = plan_file.read_bytes() + preview = subprocess.run( + [sys.executable, str(ROOT / "tool_cli.py"), "plan", + "--plan-file", str(plan_file), "--output", str(plan_file)], + cwd=ROOT, text=True, capture_output=True, + ) + self.assertEqual(preview.returncode, 2, preview.stderr) + self.assertIn("must not overwrite an input file", preview.stderr) + self.assertEqual(plan_file.read_bytes(), plan_original) + self.assertEqual(source.read_bytes(), original) + + + class TestRepoIntegration(unittest.TestCase): def test_real_recipe_audit_remains_nonrendering(self): value = execute_tool_call("figure_audit", From c458fd223c9206ef0958158ace4dad173a5722dc Mon Sep 17 00:00:00 2001 From: lostlight530 <13957086206@163.com> Date: Tue, 27 Oct 2026 07:48:22 +0800 Subject: [PATCH 2/2] fix(tool-cli): protect input evidence and request files from output replacement --- tool_cli.py | 28 +++++++++++++++++++++++++--- 1 file changed, 25 insertions(+), 3 deletions(-) diff --git a/tool_cli.py b/tool_cli.py index 946c974..aa79157 100644 --- a/tool_cli.py +++ b/tool_cli.py @@ -79,8 +79,20 @@ def _html_report(payload: dict) -> str: """ -def _write_output(path: str, content: str) -> None: +_INPUT_FILE_ARGUMENTS = frozenset({"artifact_record", "claim_audit", "figure_evidence", "graph", "recipe"}) + + +def _source_file_arguments(arguments: Any) -> list[str]: + if not isinstance(arguments, dict): + return [] + return [value for key, value in arguments.items() + if key in _INPUT_FILE_ARGUMENTS and isinstance(value, str) and value.strip()] + + +def _write_output(path: str, content: str, *, protected_inputs: tuple[str, ...] | list[str] = ()) -> None: target = Path(path) + if any(target.resolve() == Path(source).resolve() for source in protected_inputs): + raise ValueError("--output must not overwrite an input file") target.parent.mkdir(parents=True, exist_ok=True) fd, temporary = tempfile.mkstemp(prefix="." + target.name + ".", dir=str(target.parent)) try: @@ -129,6 +141,7 @@ def main(argv: list[str] | None = None) -> int: cmd.add_argument("--output", help="write formatted result to a local file") cmd.add_argument("--no-color", action="store_true") a = parser.parse_args(argv) + protected_inputs: list[str] = [] if a.command == "list": result = {"profile": PROFILE, "tool": None, "status": "ok", "result": {"tools": tool_definitions()}, "error": None, @@ -140,17 +153,26 @@ def main(argv: list[str] | None = None) -> int: "error": None if info else {"code": "UNKNOWN_TOOL", "message": "tool is not allowlisted"}, "scientific_validity_claim": False, "remote_provider_called": False} elif a.command == "plan": + if a.plan_file != "-": + protected_inputs.append(a.plan_file) try: from agent_plan import run_plan plan = _read_arguments(argparse.Namespace(args_file=a.plan_file, args="{}")) + if isinstance(plan.get("steps"), list): + for step in plan["steps"]: + if isinstance(step, dict): + protected_inputs.extend(_source_file_arguments(step.get("arguments"))) result = run_plan(plan, approved=a.approve) except (OSError, ValueError, json.JSONDecodeError) as exc: result = {"profile": PROFILE, "tool": "agent_plan", "status": "error", "result": None, "error": {"code": "INVALID_PLAN", "message": str(exc)}, "scientific_validity_claim": False, "remote_provider_called": False} else: + if a.args_file not in (None, "-"): + protected_inputs.append(a.args_file) try: parsed = _read_arguments(a) + protected_inputs.extend(_source_file_arguments(parsed)) result = execute_tool_call(a.tool, parsed) except (OSError, ValueError, json.JSONDecodeError) as exc: result = {"profile": PROFILE, "tool": a.tool, "status": "error", @@ -166,8 +188,8 @@ def main(argv: list[str] | None = None) -> int: rendered = _html_report(result) if a.output: try: - _write_output(a.output, rendered) - except OSError as exc: + _write_output(a.output, rendered, protected_inputs=protected_inputs) + except (OSError, ValueError) as exc: print(f"OUTPUT_WRITE_FAILED: {exc}", file=sys.stderr) return 2 else: