diff --git a/src/plyngent/tools/__init__.py b/src/plyngent/tools/__init__.py index 0251020..91a36d0 100644 --- a/src/plyngent/tools/__init__.py +++ b/src/plyngent/tools/__init__.py @@ -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 diff --git a/src/plyngent/tools/danger.py b/src/plyngent/tools/danger.py new file mode 100644 index 0000000..cf976f0 --- /dev/null +++ b/src/plyngent/tools/danger.py @@ -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] diff --git a/src/plyngent/tools/workspace.py b/src/plyngent/tools/workspace.py index aba8366..81584ff 100644 --- a/src/plyngent/tools/workspace.py +++ b/src/plyngent/tools/workspace.py @@ -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) diff --git a/tests/test_tools/test_danger.py b/tests/test_tools/test_danger.py new file mode 100644 index 0000000..da05955 --- /dev/null +++ b/tests/test_tools/test_danger.py @@ -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 diff --git a/tests/test_tools/test_workspace.py b/tests/test_tools/test_workspace.py index b66bd1e..d260d33 100644 --- a/tests/test_tools/test_workspace.py +++ b/tests/test_tools/test_workspace.py @@ -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)