diff --git a/tests/tools/test_delegate.py b/tests/tools/test_delegate.py index 0fa8a965cb6c..9f29e012287c 100644 --- a/tests/tools/test_delegate.py +++ b/tests/tools/test_delegate.py @@ -33,9 +33,54 @@ _resolve_child_credential_pool, _resolve_delegation_credentials, _inherit_parent_base_url, + _SUPPORTED_ACP_COMMANDS, + _validate_acp_command, ) +class TestAcpCommandAllowlist(unittest.TestCase): + """The model-supplied acp_command override must be restricted to the + supported ACP transport — it otherwise reaches subprocess.Popen as an + arbitrary host binary (CWE-78).""" + + def test_allows_bare_supported_name(self): + # Only the bare command name (+ Windows exe suffix, any case) is allowed. + for value in ("copilot", "copilot.exe", "Copilot.exe", "COPILOT", + "copilot.cmd", " copilot "): + self.assertIsNone(_validate_acp_command(value), value) + + def test_rejects_any_path(self): + # A model-supplied PATH whose basename is "copilot" must be rejected — + # the model must not point the transport at an arbitrary host binary it + # merely named "copilot" (e.g. /tmp/copilot). Custom paths are operator + # config only. + for value in ("/tmp/copilot", "/usr/local/bin/copilot", + "/opt/copilot/bin/copilot", "C:\\tools\\copilot", + "C:\\tools\\copilot.exe", "./copilot", "..\\copilot.exe"): + err = _validate_acp_command(value) + self.assertIsNotNone(err, value) + self.assertIn("path", err.lower()) + + def test_empty_is_allowed(self): + self.assertIsNone(_validate_acp_command(None)) + self.assertIsNone(_validate_acp_command("")) + self.assertIsNone(_validate_acp_command(" ")) + + def test_rejects_arbitrary_binaries(self): + for value in ("python", "powershell", "bash", "claude", + "copilot-evil", "copilot.evil", "notcopilot.exe"): + err = _validate_acp_command(value) + self.assertIsNotNone(err, value) + self.assertIn("Unsupported", err) + + def test_field_name_propagates(self): + err = _validate_acp_command("python", field="tasks[1].acp_command") + self.assertIn("tasks[1].acp_command", err) + + def test_allowlist_is_copilot_only(self): + self.assertEqual(set(_SUPPORTED_ACP_COMMANDS), {"copilot"}) + + def _make_mock_parent(depth=0): """Create a mock parent agent with the fields delegate_task expects.""" parent = MagicMock() @@ -216,6 +261,33 @@ def test_task_missing_goal(self): result = json.loads(delegate_task(tasks=[{"context": "no goal here"}], parent_agent=parent)) self.assertIn("error", result) + def test_rejects_model_top_level_acp_command(self): + """A model-supplied top-level acp_command is rejected before any child + is built — no subprocess is spawned for an arbitrary binary.""" + parent = _make_mock_parent() + with patch("tools.delegate_tool._build_child_agent") as build: + result = json.loads( + delegate_task(goal="test", acp_command="python", parent_agent=parent) + ) + self.assertIn("error", result) + self.assertIn("Unsupported acp_command='python'", result["error"]) + build.assert_not_called() + + def test_rejects_model_per_task_acp_command(self): + """A per-task acp_command override is validated too, with the offending + task index named.""" + parent = _make_mock_parent() + with patch("tools.delegate_tool._build_child_agent") as build: + result = json.loads( + delegate_task( + tasks=[{"goal": "ok"}, {"goal": "bad", "acp_command": "claude"}], + parent_agent=parent, + ) + ) + self.assertIn("error", result) + self.assertIn("tasks[1].acp_command", result["error"]) + build.assert_not_called() + @patch("tools.delegate_tool._run_single_child") def test_single_task_mode(self, mock_run): mock_run.return_value = { @@ -2250,12 +2322,14 @@ def test_acp_args_forwarded(self, mock_creds, mock_cfg): delegate_task( goal="test", - acp_command="claude", + # "copilot" is the only acp_command the allowlist permits; + # arbitrary binaries are rejected (see TestAcpCommandAllowlist). + acp_command="copilot", acp_args=["--acp", "--stdio"], parent_agent=parent, ) _, kwargs = mock_build.call_args - self.assertEqual(kwargs["override_acp_command"], "claude") + self.assertEqual(kwargs["override_acp_command"], "copilot") self.assertEqual(kwargs["override_acp_args"], ["--acp", "--stdio"]) class TestDelegateEventEnum(unittest.TestCase): diff --git a/tools/delegate_tool.py b/tools/delegate_tool.py index 8a5a060fd48c..979b24b9d72c 100644 --- a/tools/delegate_tool.py +++ b/tools/delegate_tool.py @@ -2344,6 +2344,66 @@ def _recover_tasks_from_json_string( return parsed, None +# Allowlist of ACP CLIs the subprocess transport can actually drive. Kept +# narrow on purpose: agent/copilot_acp_client.py is wired specifically to the +# GitHub Copilot CLI ("copilot --acp --stdio"). Add an entry here only when a +# new ACP-compatible CLI is genuinely supported. +_SUPPORTED_ACP_COMMANDS = frozenset({"copilot"}) + + +def _validate_acp_command( + value: Optional[str], *, field: str = "acp_command" +) -> Optional[str]: + """Reject a model-supplied acp_command the ACP transport cannot safely drive. + + ``acp_command`` / ``acp_args`` are free-form, model-controllable tool-call + arguments that propagate unvalidated into ``subprocess.Popen`` (see + agent/copilot_acp_client.py). An unrestricted value (e.g. ``python``, + ``powershell``, ``bash``) is therefore arbitrary host-process execution, + bypassing the terminal backend's sandboxing and command approvals. Restrict + the model-facing override to the supported command IDENTITY only — the bare + name ``copilot`` (optionally with a Windows executable suffix), compared + case-insensitively. + + A PATH is NOT accepted from the model. Accepting any value whose basename is + ``copilot`` (e.g. ``/tmp/copilot``) would let the model point the ACP + transport at an arbitrary host binary it merely *named* ``copilot``, which + then runs via ``subprocess.Popen`` outside the terminal sandbox/approvals. + Custom install locations belong in TRUSTED operator config + (``delegation.command`` / ``HERMES_COPILOT_ACP_COMMAND``), which is never + routed through this check. + + Returns an error string if blocked, else ``None`` (valid). + """ + if not value: + return None + v = str(value).strip() + if not v: + return None + if "/" in v or "\\" in v: + return ( + f"Unsupported {field}={value!r}: a path is not accepted from the " + f"model. Only the bare command name {sorted(_SUPPORTED_ACP_COMMANDS)} " + f"is allowed; set a custom install path via operator config " + f"(delegation.command), not the model-facing override." + ) + # Compare case-insensitively and tolerate a Windows executable suffix + # (the installed binary is `copilot.exe` on Windows). Only KNOWN suffixes + # are stripped, so "copilot.evil" / "copilot-evil" stay blocked. + name = v.lower() + for _suffix in (".exe", ".cmd", ".bat", ".com"): + if name.endswith(_suffix): + name = name[: -len(_suffix)] + break + if name in _SUPPORTED_ACP_COMMANDS: + return None + return ( + f"Unsupported {field}={value!r}. The ACP subprocess transport only " + f"supports the GitHub Copilot CLI ({sorted(_SUPPORTED_ACP_COMMANDS)}). " + f"Do not set {field} to any other binary or to a path." + ) + + def delegate_task( goal: Optional[str] = None, context: Optional[str] = None, @@ -2463,6 +2523,17 @@ def delegate_task( if not task_list: return tool_error("No tasks provided.") + # SECURITY: restrict the model-supplied acp_command override to the + # supported ACP transport BEFORE any child agent is built. acp_command / + # acp_args flow unvalidated into subprocess.Popen, so an arbitrary value + # (python/powershell/...) is host-process execution outside the terminal + # sandbox and command approvals. Only the model-facing top-level and + # per-task overrides are checked here; operator-configured + # delegation.command (creds["command"]) is trusted and unaffected. + _acp_err = _validate_acp_command(acp_command) + if _acp_err: + return tool_error(_acp_err) + # Validate each task has a goal for i, task in enumerate(task_list): if not isinstance(task, dict): @@ -2471,6 +2542,11 @@ def delegate_task( ) if not task.get("goal", "").strip(): return tool_error(f"Task {i} is missing a 'goal'.") + _task_acp_err = _validate_acp_command( + task.get("acp_command"), field=f"tasks[{i}].acp_command" + ) + if _task_acp_err: + return tool_error(_task_acp_err) overall_start = time.monotonic() results = [] @@ -3400,9 +3476,10 @@ def _build_dynamic_schema_overrides() -> dict: "acp_command": { "type": "string", "description": ( - "Per-task ACP command override (e.g. 'copilot'). " - "Overrides the top-level acp_command for this task only. " - "Do NOT set unless the user explicitly told you an ACP CLI is installed." + "Per-task ACP command override; overrides the top-level " + "acp_command for this task only. Only 'copilot' is supported " + "(any other value is rejected). Do NOT set unless the user " + "explicitly told you the GitHub Copilot CLI is installed." ), }, "acp_args": { @@ -3443,13 +3520,13 @@ def _build_dynamic_schema_overrides() -> dict: "acp_command": { "type": "string", "description": ( - "Override ACP command for child agents (e.g. 'copilot'). " - "When set, children use ACP subprocess transport instead of inheriting " - "the parent's transport. Requires an ACP-compatible CLI " - "(currently GitHub Copilot CLI via 'copilot --acp --stdio'). " + "Override ACP command for child agents. Only 'copilot' (the " + "GitHub Copilot CLI, 'copilot --acp --stdio') is supported; " + "any other value is rejected. When set, children use the ACP " + "subprocess transport instead of inheriting the parent's. " "See agent/copilot_acp_client.py for the implementation. " "IMPORTANT: Do NOT set this unless the user has explicitly told you " - "a specific ACP-compatible CLI is installed and configured. " + "the GitHub Copilot CLI is installed and configured. " "Leave empty to use the parent's default transport (Hermes subagents)." ), },