From 8366ef798b47e0aebb5593d1817a7fb1cde1a528 Mon Sep 17 00:00:00 2001 From: rob Date: Mon, 20 Jul 2026 02:33:44 -0300 Subject: [PATCH] Harden M7.6/7.7: recursive restriction accumulation, MCP server enforcement, fast-fail authorization --- AGENTS.md | 8 +- CLAUDE.md | 11 ++- src/cmdforge/providers.py | 25 +++++- src/cmdforge/runner.py | 154 +++++++++++++++++++++++--------- src/cmdforge/tool.py | 55 ++++++++---- tests/test_providers.py | 39 ++++++++ tests/test_runner.py | 181 +++++++++++++++++++++++++++++++++++++- tests/test_tool.py | 29 ++++++ 8 files changed, 435 insertions(+), 67 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 7ff227d..669f190 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -20,9 +20,11 @@ cf # Interactive tool picker ## Architecture Quick Reference - `cli/` - Routes all subcommands (list, create, run, test, providers, registry, collections, deps, install, etc.) -- `tool.py` - Tool/step dataclasses, YAML loading, wrapper generation -- `runner.py` - Step execution, variable substitution (`{input}`, `{varname}`) -- `providers.py` - AI provider abstraction (subprocess, API, PTY types; auto-discovery; fallback chains) +- `tool.py` - Tool/step dataclasses, including delegated `ToolStep` context and `McpStep` +- `runner.py` - Step execution, variable substitution, nested authorization and delegation +- `providers.py` - AI providers, auto-discovery, fallback chains, and tool/MCP allowlists +- `skills.py` - Per-provider Agent Skills loading and validation +- `mcp_client.py`, `mcp_server.py` - Stdio MCP client/server support - `gui/` - PySide6 desktop GUI with page-based navigation - `web/` - Flask web UI and forum - `registry/` - Flask registry API (search, publish, moderation) diff --git a/CLAUDE.md b/CLAUDE.md index 1026a1c..5f75026 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -40,11 +40,13 @@ python -m cmdforge.cli # Alternative CLI invocation ### Core Modules (`src/cmdforge/`) - **cli/**: CLI commands entry points (`cmdforge` command). Routes subcommands: list, create, edit, delete, test, run, ui, docs, check, refresh, providers, registry, collections, deps, install, lock, verify, add, remove, init, config, settings, system-deps -- **tool.py**: Tool definition dataclasses (`Tool`, `ToolArgument`, `PromptStep`, `CodeStep`, `ToolStep`), YAML config loading/saving, wrapper script generation +- **tool.py**: Tool definition dataclasses (`Tool`, `ToolArgument`, `PromptStep`, `CodeStep`, `ToolStep`, `McpStep`), YAML config loading/saving, wrapper script generation - **runner.py**: Execution engine. Runs tool steps sequentially, handles variable substitution (`{input}`, `{varname}`), executes Python code steps via `exec()`, handles nested tool calls with depth limit (MAX_TOOL_DEPTH=10) - **resolver.py**: Tool resolution. `resolve_tool()` searches: project manifest → local tools → owner/name → registry. Returns `ResolvedTool` with path info - **collection.py**: Collection management (`Collection` dataclass, `resolve_tool_references()`, `classify_tool_reference()`), local collection storage in `~/.cmdforge/collections/` -- **providers.py**: Provider abstraction. Supports subprocess CLI tools, OpenAI-compatible HTTP APIs, and experimental PTY wrappers. Auto-discovers installed providers on first run. Config in `~/.cmdforge/providers.yaml` with versioned migration. +- **providers.py**: Provider abstraction. Supports subprocess CLI tools, OpenAI-compatible HTTP APIs, experimental PTY wrappers, fallback chains, and tool/MCP-server allowlists. Auto-discovers installed providers on first run. Config in `~/.cmdforge/providers.yaml` with versioned migration. +- **skills.py**: Validated Agent Skills loader for per-provider `SKILL.md` context under `~/.cmdforge/providers//skills/` +- **mcp_client.py / mcp_server.py**: Stdio MCP client/server integration, configuration, schema discovery, and exposure policy - **profiles.py**: AI persona profiles with system prompts, stored in `~/.cmdforge/profiles/` - **manifest.py**: Project manifest (`cmdforge.yaml`) for declaring tool dependencies with version constraints - **lockfile.py**: Lock file support for reproducible installs (`cmdforge.lock`) @@ -85,7 +87,8 @@ Tools are YAML configs with: - `max_tokens`: Max output tokens (provider-dependent, e.g., 4096 for haiku) - `plain_text`: Bypass structured output enforcement 2. **Code Step**: Executes Python code via `exec()`, captures specified variables (comma-separated for multiple outputs) -3. **Tool Step**: Calls another tool (meta-tools), supports `input_template`, `args` dict, and `provider` override. Dependencies resolved via `resolver.py` +3. **Tool Step**: Calls another tool (meta-tools), supports `input_template`, `args`, and recursive agent context (`provider`, `profile`, `skills`, `tools`). Provider and delegated allowlists are both enforced before resolution. +4. **MCP Step**: Calls a tool on a configured MCP server. Provider `mcp_servers` allowlists are enforced before connecting. ### Variable Flow @@ -132,6 +135,8 @@ Provider fields: - `tags`: List of strings (e.g. `["free", "code", "local"]`) - `pty_config`: Dict with `prompt_pattern`, `response_pattern`, `exit_command` for pty providers - `install`: Optional structured install metadata dict +- `tools`: Optional CmdForge tool allowlist; `null` allows all and `[]` denies all +- `mcp_servers`: Optional MCP server allowlist; `null` allows all and `[]` denies all The `mock` provider is built-in for testing without API calls. Use `--provider mock` or `--dry-run` flags when testing tools. diff --git a/src/cmdforge/providers.py b/src/cmdforge/providers.py index 8f01533..0ec457c 100644 --- a/src/cmdforge/providers.py +++ b/src/cmdforge/providers.py @@ -5,6 +5,7 @@ import re import shlex import subprocess import shutil +import sys from dataclasses import dataclass, field from pathlib import Path from typing import Optional, List @@ -215,6 +216,19 @@ class Provider: tools: Optional[List[str]] = None # Allowed CmdForge tools (None = all) mcp_servers: Optional[List[str]] = None # MCP servers available to this provider + def __post_init__(self) -> None: + for field_name, values in ( + ("tools", self.tools), + ("mcp_servers", self.mcp_servers), + ): + if values is not None and ( + not isinstance(values, list) + or not all(isinstance(value, str) and value for value in values) + ): + raise ValueError( + f"Provider {field_name} must be a list of non-empty strings or null" + ) + def to_dict(self) -> dict: d = { "name": self.name, @@ -238,9 +252,9 @@ class Provider: d["api_key_env"] = self.api_key_env if self.pty_config: d["pty_config"] = self.pty_config - if self.tools: + if self.tools is not None: d["tools"] = self.tools - if self.mcp_servers: + if self.mcp_servers is not None: d["mcp_servers"] = self.mcp_servers return d @@ -419,8 +433,11 @@ def load_providers() -> List[Provider]: if not data or "providers" not in data: return DEFAULT_PROVIDERS.copy() providers = [Provider.from_dict(p) for p in data["providers"]] - except Exception: - return DEFAULT_PROVIDERS.copy() + except Exception as exc: + # A present but malformed configuration must not silently fall back to + # unrestricted defaults, particularly when access policies are invalid. + print(f"Warning: Failed to load provider configuration: {exc}", file=sys.stderr) + return [] config_version = data.get("version", 1) if not isinstance(config_version, int): diff --git a/src/cmdforge/runner.py b/src/cmdforge/runner.py index b2e37a9..af97e43 100644 --- a/src/cmdforge/runner.py +++ b/src/cmdforge/runner.py @@ -1,9 +1,11 @@ """Tool execution engine.""" import argparse +import fnmatch import json import re import sys +from dataclasses import replace from pathlib import Path from typing import Optional, List @@ -737,7 +739,10 @@ def execute_tool_step( provider_override: Optional[str] = None, dry_run: bool = False, verbose: bool = False, - call_stack: Optional[list] = None + call_stack: Optional[list] = None, + agent_profile: Optional[str] = None, + agent_skills: Optional[List[str]] = None, + agent_tool_policies: Optional[List[List[str]]] = None, ) -> tuple[str, bool]: """ Execute a tool step by calling another tool. @@ -761,7 +766,15 @@ def execute_tool_step( _print_call_stack(call_stack, f"Maximum tool nesting depth ({MAX_TOOL_DEPTH}) exceeded") return "", False - # Resolve the tool reference + # Authorization must happen before resolution because resolution may fetch a + # missing tool from the registry. + effective_provider = step.provider or provider_override + if not _authorize_tool_call( + step.tool, effective_provider, agent_tool_policies, call_stack + ): + return "", False + + # Resolve the tool reference only after authorization succeeds. try: resolved = resolve_tool(step.tool) nested_tool = resolved.tool @@ -778,26 +791,6 @@ def execute_tool_step( for key, value in step.args.items(): custom_args[key] = substitute_variables(str(value), variables, warn_non_scalar=verbose) - # Determine effective provider (step override > parent override) - effective_provider = step.provider or provider_override - - # Check tool access if the provider restricts which tools it can call - if effective_provider: - from .providers import get_provider - provider_obj = get_provider(effective_provider) - if provider_obj and provider_obj.tools is not None: - allowed = provider_obj.tools - if step.tool not in allowed and not _tool_match_allowlist(step.tool, allowed): - _print_call_stack( - call_stack, - f"Provider '{effective_provider}' is not allowed to call tool '{step.tool}'", - ) - print( - f"Provider '{effective_provider}' tools allowlist: {allowed}", - file=sys.stderr, - ) - return "", False - if verbose: print(f"[verbose] Tool step: calling {step.tool}", file=sys.stderr) print(f"[verbose] Input length: {len(input_text)} chars", file=sys.stderr) @@ -817,8 +810,11 @@ def execute_tool_step( verbose=verbose, _depth=depth + 1, _call_stack=call_stack, - agent_profile=step.profile, - agent_skills=step.skills, + agent_profile=step.profile if step.profile is not None else agent_profile, + agent_skills=step.skills if step.skills is not None else agent_skills, + agent_tool_policies=_extend_agent_tool_policies( + agent_tool_policies, step.tools + ), ) return output, exit_code == 0 @@ -837,6 +833,7 @@ def run_tool( _call_stack: Optional[list] = None, agent_profile: Optional[str] = None, agent_skills: Optional[List[str]] = None, + agent_tool_policies: Optional[List[List[str]]] = None, ) -> tuple[str, int]: """ Execute a tool. @@ -1017,7 +1014,10 @@ def run_tool( provider_override=provider_override, dry_run=dry_run, verbose=verbose, - call_stack=step_stack + call_stack=step_stack, + agent_profile=agent_profile, + agent_skills=agent_skills, + agent_tool_policies=agent_tool_policies, ) if not success: return "", 3 @@ -1033,6 +1033,11 @@ def run_tool( if dry_run: variables[step.output_var] = f"[DRY RUN - would call mcp:{step.server}/{step.tool}]" else: + step_stack = _call_stack + [(tool.name, i + 1)] + if not _authorize_mcp_call( + step.server, provider_override, step_stack + ): + return "", 4 args = _substitute_mcp_args(step.arguments, variables) try: if mcp_manager is None: @@ -1227,27 +1232,96 @@ def main(): sys.exit(exit_code) -def _tool_match_allowlist(name: str, allowed: list) -> bool: - for pattern in allowed: - if pattern == name: - return True - if pattern.endswith("*") and name.startswith(pattern[:-1]): - return True - return False +def _matches_allowlist(name: str, allowed: List[str]) -> bool: + return any(fnmatch.fnmatchcase(name, pattern) for pattern in allowed) + + +def _authorize_tool_call( + name: str, + provider_name: Optional[str], + agent_tool_policies: Optional[List[List[str]]], + call_stack: list, +) -> bool: + """Enforce provider and delegated-agent tool policies.""" + if provider_name: + from .providers import get_provider + provider = get_provider(provider_name) + if provider is None: + _print_call_stack( + call_stack, + f"Provider '{provider_name}' is not configured; refusing tool call '{name}'", + ) + return False + if provider.tools is not None and not _matches_allowlist(name, provider.tools): + _print_call_stack( + call_stack, + f"Provider '{provider_name}' is not allowed to call tool '{name}'", + ) + print( + f"Provider '{provider_name}' tools allowlist: {provider.tools}", + file=sys.stderr, + ) + return False + + for policy in agent_tool_policies or []: + if not _matches_allowlist(name, policy): + _print_call_stack( + call_stack, + f"Delegated agent is not allowed to call tool '{name}'", + ) + print(f"Delegated agent tools allowlist: {policy}", file=sys.stderr) + return False + return True + + +def _extend_agent_tool_policies( + policies: Optional[List[List[str]]], + additional: Optional[List[str]], +) -> Optional[List[List[str]]]: + """Add a restriction without allowing descendants to broaden ancestors.""" + if additional is None: + return policies + return [*(policies or []), additional] + + +def _authorize_mcp_call( + server: str, + provider_name: Optional[str], + call_stack: list, +) -> bool: + """Enforce the effective provider's MCP-server policy.""" + if not provider_name: + return True + + from .providers import get_provider + provider = get_provider(provider_name) + if provider is None: + _print_call_stack( + call_stack, + f"Provider '{provider_name}' is not configured; refusing MCP server '{server}'", + ) + return False + if provider.mcp_servers is not None and not _matches_allowlist( + server, provider.mcp_servers + ): + _print_call_stack( + call_stack, + f"Provider '{provider_name}' is not allowed to use MCP server '{server}'", + ) + print( + f"Provider '{provider_name}' MCP server allowlist: {provider.mcp_servers}", + file=sys.stderr, + ) + return False + return True def _step_with_profile(step, profile_name): - from copy import copy - s = copy(step) - s.profile = profile_name - return s + return replace(step, profile=profile_name) def _step_with_skills(step, skill_names): - from copy import copy - s = copy(step) - s.skills = skill_names - return s + return replace(step, skills=skill_names) if __name__ == "__main__": diff --git a/src/cmdforge/tool.py b/src/cmdforge/tool.py index 5400d7b..f069d8a 100644 --- a/src/cmdforge/tool.py +++ b/src/cmdforge/tool.py @@ -17,6 +17,33 @@ TOOLS_DIR = Path.home() / ".cmdforge" BIN_DIR = Path.home() / ".local" / "bin" +def _validate_skill_selection(skills: Optional[List[str]]) -> None: + """Validate a step's optional Agent Skills selection.""" + if skills is None: + return + if not isinstance(skills, list): + raise ValueError("skills must be a list of skill names") + if any(not isinstance(name, str) for name in skills): + raise ValueError("skills entries must be strings") + if "*" in skills and skills != ["*"]: + raise ValueError("'*' must be the only entry when enabling all skills") + if len(skills) != len(set(skills)): + raise ValueError("skills must not contain duplicate names") + + from .skills import _validate_skill_name + for name in skills: + if name != "*": + _validate_skill_name(name) + + +def _validate_optional_patterns(patterns: Optional[List[str]], field_name: str) -> None: + if patterns is not None and ( + not isinstance(patterns, list) + or not all(isinstance(pattern, str) and pattern for pattern in patterns) + ): + raise ValueError(f"{field_name} must be a list of non-empty strings or null") + + @dataclass class SystemDependency: """A system-level package dependency (apt, brew, pacman, etc.).""" @@ -133,21 +160,7 @@ class PromptStep: skills: Optional[List[str]] = None # Skill names to enable for this step def __post_init__(self) -> None: - if self.skills is None: - return - if not isinstance(self.skills, list): - raise ValueError("skills must be a list of skill names") - if any(not isinstance(name, str) for name in self.skills): - raise ValueError("skills entries must be strings") - if "*" in self.skills and self.skills != ["*"]: - raise ValueError("'*' must be the only entry when enabling all skills") - if len(self.skills) != len(set(self.skills)): - raise ValueError("skills must not contain duplicate names") - - from .skills import _validate_skill_name - for name in self.skills: - if name != "*": - _validate_skill_name(name) + _validate_skill_selection(self.skills) def to_dict(self) -> dict: d = { @@ -244,6 +257,15 @@ class ToolStep: profile: Optional[str] = None # AI persona for the nested tool skills: Optional[List[str]] = None # Skills to enable in the nested tool name: Optional[str] = None # Optional display name for the step + tools: Optional[List[str]] = None # Tools the delegated agent may call + + def __post_init__(self) -> None: + _validate_skill_selection(self.skills) + _validate_optional_patterns(self.tools, "tools") + if self.profile is not None and ( + not isinstance(self.profile, str) or not self.profile + ): + raise ValueError("profile must be a non-empty string or null") def to_dict(self) -> dict: d = { @@ -261,6 +283,8 @@ class ToolStep: d["profile"] = self.profile if self.skills is not None: d["skills"] = self.skills + if self.tools is not None: + d["tools"] = self.tools if self.name: d["name"] = self.name return d @@ -275,6 +299,7 @@ class ToolStep: provider=data.get("provider"), profile=data.get("profile"), skills=data.get("skills"), + tools=data.get("tools"), name=data.get("name") ) diff --git a/tests/test_providers.py b/tests/test_providers.py index 7be65c7..071cfd7 100644 --- a/tests/test_providers.py +++ b/tests/test_providers.py @@ -172,6 +172,20 @@ class TestProviderPersistence: assert oct(temp_providers_file.stat().st_mode & 0o777) == "0o600" + def test_malformed_access_policy_fails_closed(self, temp_providers_file, capsys): + temp_providers_file.parent.mkdir(parents=True) + temp_providers_file.write_text(yaml.safe_dump({ + "version": 2, + "providers": [{ + "name": "claude", + "command": "claude -p", + "tools": "should-have-been-a-list", + }], + })) + + assert load_providers() == [] + assert "Failed to load provider configuration" in capsys.readouterr().err + def test_legacy_config_gains_missing_defaults_without_losing_custom_provider(self, temp_providers_file): temp_providers_file.parent.mkdir(parents=True) temp_providers_file.write_text(yaml.safe_dump({ @@ -810,3 +824,28 @@ class TestProviderAccessControl: provider = Provider("test", "cmd") assert "tools" not in provider.to_dict() assert "mcp_servers" not in provider.to_dict() + + def test_empty_allowlists_survive_roundtrip(self): + provider = Provider("locked", "cmd", tools=[], mcp_servers=[]) + serialized = provider.to_dict() + assert serialized["tools"] == [] + assert serialized["mcp_servers"] == [] + + restored = Provider.from_dict(serialized) + assert restored.tools == [] + assert restored.mcp_servers == [] + + @pytest.mark.parametrize( + ("field", "value"), + [ + ("tools", "tool-a"), + ("tools", [""]), + ("tools", [1]), + ("mcp_servers", "filesystem"), + ("mcp_servers", [""]), + ("mcp_servers", [1]), + ], + ) + def test_rejects_invalid_access_policies(self, field, value): + with pytest.raises(ValueError, match=field): + Provider("test", "cmd", **{field: value}) diff --git a/tests/test_runner.py b/tests/test_runner.py index 4d7b9f0..0bfdad8 100644 --- a/tests/test_runner.py +++ b/tests/test_runner.py @@ -1,18 +1,20 @@ """Tests for runner.py - Tool execution engine.""" import pytest +from types import SimpleNamespace from unittest.mock import patch, MagicMock from cmdforge.runner import ( substitute_variables, execute_prompt_step, execute_code_step, + execute_tool_step, run_tool, create_argument_parser, collect_custom_args ) -from cmdforge.tool import Tool, ToolArgument, PromptStep, CodeStep -from cmdforge.providers import ProviderResult +from cmdforge.tool import Tool, ToolArgument, PromptStep, CodeStep, ToolStep, McpStep +from cmdforge.providers import Provider, ProviderResult class TestSubstituteVariables: @@ -1010,3 +1012,178 @@ class TestCreateArgumentParser: "max_size": "50", "format": "json", } + + +class TestProviderExecutionPolicies: + def test_denied_tool_is_not_resolved(self, monkeypatch): + resolver = MagicMock() + monkeypatch.setattr("cmdforge.runner.resolve_tool", resolver) + monkeypatch.setattr( + "cmdforge.providers.get_provider", + lambda name: Provider(name, "cmd", tools=[]), + ) + + output, success = execute_tool_step( + ToolStep(tool="forbidden", output_var="out", provider="locked"), + {"input": ""}, + ) + + assert (output, success) == ("", False) + resolver.assert_not_called() + + def test_unknown_provider_is_rejected_before_resolution(self, monkeypatch): + resolver = MagicMock() + monkeypatch.setattr("cmdforge.runner.resolve_tool", resolver) + monkeypatch.setattr("cmdforge.providers.get_provider", lambda name: None) + + _, success = execute_tool_step( + ToolStep(tool="anything", output_var="out", provider="missing"), + {"input": ""}, + ) + + assert success is False + resolver.assert_not_called() + + def test_provider_and_delegated_allowlists_both_apply(self, monkeypatch): + leaf = Tool(name="safe-lint", output="completed") + resolver = MagicMock(return_value=SimpleNamespace(tool=leaf)) + monkeypatch.setattr("cmdforge.runner.resolve_tool", resolver) + monkeypatch.setattr( + "cmdforge.providers.get_provider", + lambda name: Provider(name, "cmd", tools=["safe-*"]), + ) + + output, success = execute_tool_step( + ToolStep(tool="safe-lint", output_var="out", provider="restricted"), + {"input": ""}, + agent_tool_policies=[["*-lint"]], + ) + + assert (output, success) == ("completed", True) + resolver.assert_called_once_with("safe-lint") + + def test_delegated_allowlist_can_further_restrict_provider(self, monkeypatch): + resolver = MagicMock() + monkeypatch.setattr("cmdforge.runner.resolve_tool", resolver) + monkeypatch.setattr( + "cmdforge.providers.get_provider", + lambda name: Provider(name, "cmd", tools=["*"]), + ) + + _, success = execute_tool_step( + ToolStep(tool="unsafe-delete", output_var="out", provider="broad"), + {"input": ""}, + agent_tool_policies=[["safe-*"]], + ) + + assert success is False + resolver.assert_not_called() + + def test_mcp_server_policy_denies_before_manager_creation(self, monkeypatch): + manager = MagicMock() + monkeypatch.setattr("cmdforge.runner.McpClientManager", manager) + monkeypatch.setattr( + "cmdforge.providers.get_provider", + lambda name: Provider(name, "cmd", mcp_servers=[]), + ) + tool = Tool( + name="mcp-tool", + steps=[McpStep(server="database", tool="query", output_var="result")], + output="{result}", + ) + + output, exit_code = run_tool( + tool, "", {}, provider_override="restricted" + ) + + assert (output, exit_code) == ("", 4) + manager.assert_not_called() + + def test_mcp_server_policy_allows_matching_server(self, monkeypatch): + class FakeManager: + def call_tool(self, server, tool, arguments, result_mode="auto"): + return "allowed" + + monkeypatch.setattr("cmdforge.runner.McpClientManager", FakeManager) + monkeypatch.setattr( + "cmdforge.providers.get_provider", + lambda name: Provider(name, "cmd", mcp_servers=["data-*"]), + ) + tool = Tool( + name="mcp-tool", + steps=[McpStep(server="data-local", tool="query", output_var="result")], + output="{result}", + ) + + assert run_tool(tool, "", {}, provider_override="restricted") == ( + "allowed", 0 + ) + + def test_agent_context_propagates_through_nested_tools(self, monkeypatch): + captured = {} + leaf = Tool( + name="leaf", + steps=[PromptStep("prompt", "original", "answer")], + output="{answer}", + ) + middle = Tool( + name="middle", + steps=[ToolStep(tool="leaf", output_var="child")], + output="{child}", + ) + root = Tool( + name="root", + steps=[ + ToolStep( + tool="middle", + output_var="delegated", + provider="delegate", + profile="architect", + skills=["python"], + tools=["leaf"], + ) + ], + output="{delegated}", + ) + + def resolve(name): + return SimpleNamespace(tool={"middle": middle, "leaf": leaf}[name]) + + def execute_prompt(step, *args, **kwargs): + captured["step"] = step + captured["provider_override"] = args[1] + return "done", True + + monkeypatch.setattr("cmdforge.runner.resolve_tool", resolve) + monkeypatch.setattr("cmdforge.runner.execute_prompt_step", execute_prompt) + monkeypatch.setattr( + "cmdforge.providers.get_provider", + lambda name: Provider(name, "cmd", tools=["middle", "leaf"]), + ) + + assert run_tool(root, "", {}) == ("done", 0) + assert captured["step"].profile == "architect" + assert captured["step"].skills == ["python"] + assert captured["provider_override"] == "delegate" + + def test_descendant_cannot_broaden_ancestor_tool_policy(self, monkeypatch): + resolver = MagicMock() + monkeypatch.setattr("cmdforge.runner.resolve_tool", resolver) + monkeypatch.setattr( + "cmdforge.providers.get_provider", + lambda name: Provider(name, "cmd", tools=["*"]), + ) + + _, success = execute_tool_step( + ToolStep( + tool="dangerous", + output_var="out", + provider="delegate", + tools=["*"], + ), + {"input": ""}, + agent_tool_policies=[["safe-*"]], + ) + + assert success is False + resolver.assert_not_called() diff --git a/tests/test_tool.py b/tests/test_tool.py index 6856eb7..cf6830f 100644 --- a/tests/test_tool.py +++ b/tests/test_tool.py @@ -609,6 +609,35 @@ class TestAgentContext: assert "skills" not in d assert "profile" not in d + def test_toolstep_tools_roundtrip(self): + step = ToolStep( + tool="my-tool", output_var="out", tools=["safe-*", "official/lint"] + ) + assert ToolStep.from_dict(step.to_dict()).tools == [ + "safe-*", "official/lint" + ] + + def test_toolstep_empty_tools_roundtrip(self): + step = ToolStep(tool="my-tool", output_var="out", tools=[]) + serialized = step.to_dict() + assert serialized["tools"] == [] + assert ToolStep.from_dict(serialized).tools == [] + + @pytest.mark.parametrize( + "data", + [ + {"skills": "python"}, + {"skills": ["*", "python"]}, + {"skills": ["Python"]}, + {"tools": "safe-tool"}, + {"tools": [""]}, + {"profile": ""}, + ], + ) + def test_toolstep_rejects_invalid_agent_context(self, data): + with pytest.raises(ValueError): + ToolStep(tool="my-tool", output_var="out", **data) + def test_promptstep_skills_roundtrip(self): step = PromptStep( prompt="Test", provider="claude", output_var="out",