mirror of
https://github.com/NCBM/plyngent.git
synced 2026-11-07 07:17:54 +08:00
config: [agent] allow_paths static pre-allow
This commit is contained in:
@@ -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).
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -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)
|
||||
|
||||
Reference in New Issue
Block a user