core/tools: classify destructive tools and clarify policy denies

Add classify_danger for delete/move/overwrite paths, get_path_denylist,
and clearer path/command denial messages.
This commit is contained in:
2026-07-14 23:30:02 +08:00
parent 81a30d0e4d
commit 510dd67a54
5 changed files with 88 additions and 5 deletions
+2
View File
@@ -1,3 +1,4 @@
from .danger import classify_danger as classify_danger
from .file import FILE_TOOLS as FILE_TOOLS
from .file import copy_path as copy_path
from .file import delete_path as delete_path
@@ -21,6 +22,7 @@ from .workspace import WorkspaceError as WorkspaceError
from .workspace import check_command_allowed as check_command_allowed
from .workspace import clear_workspace_root as clear_workspace_root
from .workspace import get_command_denylist as get_command_denylist
from .workspace import get_path_denylist as get_path_denylist
from .workspace import get_workspace_root as get_workspace_root
from .workspace import resolve_path as resolve_path
from .workspace import set_command_denylist as set_command_denylist
+46
View File
@@ -0,0 +1,46 @@
from __future__ import annotations
from typing import TYPE_CHECKING
from plyngent.tools.workspace import WorkspaceError, resolve_path
if TYPE_CHECKING:
from collections.abc import Mapping
def _write_file_reason(args: Mapping[str, object]) -> str | None:
path = args.get("path")
if not isinstance(path, str) or not path:
return None
try:
target = resolve_path(path)
except WorkspaceError:
return None
if target.exists() or target.is_symlink():
return f"overwrite existing file {path!r}"
return None
def classify_danger(name: str, args: Mapping[str, object]) -> str | None:
"""Return a short reason if ``name``/``args`` need user confirm, else ``None``.
Hard denylists (paths/commands) still raise independently. This only covers
soft confirms for mutating tools that policy otherwise allows.
"""
reasons: dict[str, str | None] = {
"delete_path": (
f"delete path {args.get('path', '')!r} recursively"
if bool(args.get("recursive", False))
else f"delete path {args.get('path', '')!r}"
),
"move_path": f"move {args.get('src', '')!r} -> {args.get('dst', '')!r}",
"copy_path": (
f"copy with overwrite {args.get('src', '')!r} -> {args.get('dst', '')!r}"
if bool(args.get("overwrite", False))
else None
),
"write_file": _write_file_reason(args),
}
if name not in reasons:
return None
return reasons[name]
+8 -3
View File
@@ -68,6 +68,11 @@ def set_path_denylist(patterns: list[str] | tuple[str, ...] | None) -> None:
_state.path_denylist = tuple(patterns or ())
def get_path_denylist() -> tuple[str, ...]:
"""Return the current path substring denylist."""
return _state.path_denylist
def set_command_denylist(names: list[str] | tuple[str, ...] | frozenset[str] | None) -> None:
"""Set denied command basenames (None restores defaults)."""
_state.command_denylist = DEFAULT_COMMAND_DENYLIST if names is None else frozenset(names)
@@ -87,12 +92,12 @@ def resolve_path(path: str | Path) -> Path:
try:
_ = resolved.relative_to(root)
except ValueError as exc:
msg = f"path escapes workspace root: {path}"
msg = f"path escapes workspace root ({root}): {path}"
raise WorkspaceError(msg) from exc
resolved_str = str(resolved)
for pattern in _state.path_denylist:
if pattern and pattern in resolved_str:
msg = f"path denied by policy: {path}"
msg = f"path denied by policy (matched {pattern!r}): {path}"
raise WorkspaceError(msg)
return resolved
@@ -104,5 +109,5 @@ def check_command_allowed(argv: list[str]) -> None:
raise WorkspaceError(msg)
binary = Path(argv[0]).name
if binary in _state.command_denylist:
msg = f"command denied by policy: {binary}"
msg = f"command denied by policy (basename {binary!r} is blocked)"
raise WorkspaceError(msg)
+30
View File
@@ -0,0 +1,30 @@
from __future__ import annotations
from pathlib import Path
from plyngent.tools.danger import classify_danger
def test_classify_delete_and_move() -> None:
assert classify_danger("delete_path", {"path": "a.txt"}) == "delete path 'a.txt'"
assert "recursively" in (classify_danger("delete_path", {"path": "d", "recursive": True}) or "")
assert "move" in (classify_danger("move_path", {"src": "a", "dst": "b"}) or "")
def test_classify_copy_overwrite_only() -> None:
assert classify_danger("copy_path", {"src": "a", "dst": "b"}) is None
assert "overwrite" in (
classify_danger("copy_path", {"src": "a", "dst": "b", "overwrite": True}) or ""
)
def test_classify_write_file_overwrite(workspace: object) -> None:
assert isinstance(workspace, Path)
_ = (workspace / "x.txt").write_text("old", encoding="utf-8")
assert "overwrite" in (classify_danger("write_file", {"path": "x.txt", "content": "n"}) or "")
assert classify_danger("write_file", {"path": "new.txt", "content": "n"}) is None
def test_classify_safe_tools() -> None:
assert classify_danger("read_file", {"path": "a"}) is None
assert classify_danger("run_command", {"command": ["echo", "hi"]}) is None
+2 -2
View File
@@ -36,14 +36,14 @@ def test_path_denylist(workspace: object) -> None:
secrets.mkdir()
_ = (secrets / "key").write_text("k", encoding="utf-8")
set_path_denylist(["/secrets/"])
with pytest.raises(WorkspaceError, match="denied"):
with pytest.raises(WorkspaceError, match="matched '/secrets/'"):
_ = resolve_path("secrets/key")
set_path_denylist(None)
def test_command_denylist(workspace: object) -> None:
del workspace
with pytest.raises(WorkspaceError, match="denied"):
with pytest.raises(WorkspaceError, match="basename 'rm' is blocked"):
check_command_allowed(["rm", "-rf", "/"])
check_command_allowed(["echo", "ok"])
set_command_denylist(None)