Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
78 changes: 76 additions & 2 deletions tests/tools/test_delegate.py
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Expand Down Expand Up @@ -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 = {
Expand Down Expand Up @@ -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):
Expand Down
93 changes: 85 additions & 8 deletions tools/delegate_tool.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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):
Expand All @@ -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 = []
Expand Down Expand Up @@ -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": {
Expand Down Expand Up @@ -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)."
),
},
Expand Down