From 22f52a0bd2fb7dd308200d1d014084c18beb8e99 Mon Sep 17 00:00:00 2001 From: worldmozara Date: Wed, 9 Sep 2026 22:19:23 +0800 Subject: [PATCH] config: [agent] allow_paths static pre-allow --- src/plyngent/config/models.py | 6 ++++++ src/plyngent/tools/__init__.py | 1 + src/plyngent/tools/access.py | 28 ++++++++++++++++++++++++- tests/test_config/test_agent_section.py | 28 +++++++++++++++++++++++++ tests/test_tools/test_access.py | 27 ++++++++++++++++++++++++ 5 files changed, 89 insertions(+), 1 deletion(-) diff --git a/src/plyngent/config/models.py b/src/plyngent/config/models.py index a5dc617..397791c 100644 --- a/src/plyngent/config/models.py +++ b/src/plyngent/config/models.py @@ -10,6 +10,9 @@ from msgspec import Struct, field # ``https://api.deepseek.com/anthropic``) type DeepSeekConvention = Literal["", "openai", "anthropic", "responses"] +# Static directory pre-allow mode for ``[agent].allow_paths`` (path → mode). +type AllowPathMode = Literal["read", "write", "exec"] + # Built-in persona when ``[agent].system_prompt`` is omitted. # Set ``system_prompt = ""`` to omit the persona block only. # Override with a multi-line TOML literal (prefer ''' so nested " is fine). @@ -111,6 +114,9 @@ class AgentConfig(Struct, omit_defaults=True): parallel_tools: bool = True confirm_destructive: bool = True path_denylist: list[str] = field(default_factory=list) + # Static directory pre-allow (resolved path → mode); never prompts. + # Denylist still applies, and the path must exist when chat starts. + allow_paths: dict[str, AllowPathMode] = field(default_factory=dict) max_context_tokens: int = 200_000 # How to inject todo stack nags into model context (see agent/todo_nag.py). diff --git a/src/plyngent/tools/__init__.py b/src/plyngent/tools/__init__.py index f87a669..912f6b1 100644 --- a/src/plyngent/tools/__init__.py +++ b/src/plyngent/tools/__init__.py @@ -8,6 +8,7 @@ from .access import get_directory_access_confirm_hook as get_directory_access_co from .access import grant_process_access as grant_process_access from .access import grant_session_access as grant_session_access from .access import request_directory_access as request_directory_access +from .access import set_config_access as set_config_access from .access import set_directory_access_confirm_hook as set_directory_access_confirm_hook from .catalog import ToolCatalog as ToolCatalog from .catalog import ToolSource as ToolSource diff --git a/src/plyngent/tools/access.py b/src/plyngent/tools/access.py index 5cd2023..505af0a 100644 --- a/src/plyngent/tools/access.py +++ b/src/plyngent/tools/access.py @@ -30,7 +30,7 @@ from plyngent.tools.workspace import ( ) if TYPE_CHECKING: - from collections.abc import Callable + from collections.abc import Callable, Mapping from plyngent.tools.context import InstanceState, SessionState @@ -122,6 +122,32 @@ def active_grant_count() -> int: return total +def set_config_access(entries: Mapping[str, str]) -> tuple[list[Path], list[str]]: + """Install static TOML pre-allow entries on the bound instance policy. + + Replaces any previous config grants. Returns ``(applied roots, skipped + keys)``; entries are skipped when the mode is unknown or the path does not + exist (the host warns about skipped keys). + """ + policy = require_bound_instance().workspace + policy.config_allow.clear() + applied: list[Path] = [] + skipped: list[str] = [] + for raw_path, raw_mode in entries.items(): + mode = parse_access_mode(raw_mode) + resolved: Path | None + try: + resolved = Path(raw_path).expanduser().resolve() + except OSError: + resolved = None + if mode is None or resolved is None or not resolved.exists(): + skipped.append(raw_path) + continue + policy.config_allow[resolved] = mode + applied.append(resolved) + return applied, skipped + + def _resolve_target(path: str) -> Path | str: """Resolve a requested path; return a model-facing error string on failure.""" try: diff --git a/tests/test_config/test_agent_section.py b/tests/test_config/test_agent_section.py index 1e630f5..bd8c9c7 100644 --- a/tests/test_config/test_agent_section.py +++ b/tests/test_config/test_agent_section.py @@ -26,6 +26,7 @@ def test_agent_section_defaults(tmp_path: Path) -> None: assert store.agent_config.parallel_tools is True assert store.agent_config.confirm_destructive is True assert store.agent_config.path_denylist == [] + assert store.agent_config.allow_paths == {} assert store.agent_config.max_context_tokens == 200_000 assert store.plugins_config.enable == [] assert store.plugins_config.disable == [] @@ -162,6 +163,33 @@ tool_directives = "" ) +def test_agent_allow_paths_parse(tmp_path: Path) -> None: + path = tmp_path / "c.toml" + _ = path.write_text( + """ +[agent] +allow_paths = { "/data/datasets" = "read", "/tmp/build" = "exec" } +""", + encoding="utf-8", + ) + store = load(path) + assert store.agent_config.allow_paths == {"/data/datasets": "read", "/tmp/build": "exec"} + + +def test_agent_allow_paths_invalid_mode_falls_back(tmp_path: Path) -> None: + path = tmp_path / "c.toml" + _ = path.write_text( + """ +[agent] +allow_paths = { "/data" = "sudo" } +""", + encoding="utf-8", + ) + store = load(path) + # Invalid field value resets the whole [agent] section to defaults. + assert store.agent_config.allow_paths == {} + + def test_agent_mcp_instructions_flag(tmp_path: Path) -> None: path = tmp_path / "c.toml" _ = path.write_text( diff --git a/tests/test_tools/test_access.py b/tests/test_tools/test_access.py index f9ca8d8..53774e7 100644 --- a/tests/test_tools/test_access.py +++ b/tests/test_tools/test_access.py @@ -12,11 +12,13 @@ from plyngent.tools import ( AccessDecision, AccessMode, SessionState, + WorkspaceError, active_workspace_policy, get_directory_access_confirm_hook, grant_session_access, request_directory_access, resolve_path, + set_config_access, set_directory_access_confirm_hook, set_path_denylist, ) @@ -234,3 +236,28 @@ async def test_grant_cap( assert session.access_grants == {} finally: policy.config_allow.clear() + + +def test_set_config_access_installs_and_skips( + workspace: object, + session: SessionState, + tmp_path_factory: pytest.TempPathFactory, +) -> None: + del workspace, session + outside = tmp_path_factory.mktemp("config-allow").resolve() + target = outside / "data.txt" + _ = target.write_text("x", encoding="utf-8") + missing = outside / "missing" + applied, skipped = set_config_access( + { + str(outside): "read", + str(missing): "read", + str(target): "sudo", + } + ) + assert applied == [outside] + assert sorted(skipped) == sorted([str(missing), str(target)]) + assert active_workspace_policy().config_allow == {outside: AccessMode.READ} + assert resolve_path(str(target), required=AccessMode.READ) == target + with pytest.raises(WorkspaceError, match="escapes"): + _ = resolve_path(str(target), required=AccessMode.WRITE)