core/tools: block a denylisted program behind env wrappers and run launchers

src/plyngent/tools/workspace.py

@@ -5,6 +5,8 @@ from enum import IntEnum
 from pathlib import Path
 from typing import TYPE_CHECKING, cast
 
+from plyngent.tools.command_scan import unwrap_command
+
 if TYPE_CHECKING:
     from collections.abc import Callable, Mapping, Sequence
 
@@ -404,7 +406,12 @@ def argv_shape_error(argv: object) -> str | None:
 
 
 def check_command_allowed(argv: list[str]) -> None:
-    """Raise if argv is empty or the executable basename is denylisted.
+    """Raise if argv is empty or a basename in the command chain is denylisted.
+
+    The chain is the program that really runs plus every wrapper it is reached
+    through (``env FOO=1 rm``, ``pdm run rm``, ``sudo -u x rm``), so a denylisted
+    program cannot hide behind one. Basenames are compared lower-cased, so
+    ``RM`` and ``rm.exe`` count as ``rm``.
 
     Denylisted basenames are not hard-rejected when a policy confirm hook is
     installed: the human is asked (with a timeout; default deny). Session grants
@@ -418,25 +425,24 @@ def check_command_allowed(argv: list[str]) -> None:
         msg = "command argv must not be empty"
         raise WorkspaceError(msg)
     policy = active_workspace_policy()
-    binary = Path(argv[0]).name
+    scan = unwrap_command(argv)
-    if binary not in policy.command_denylist:
+    for binary in (*scan.wrappers, scan.base):
-        return
+        if binary not in policy.command_denylist or binary in policy.policy_allowed_commands:
-    if binary in policy.policy_allowed_commands:
+            continue
-        return
         hook = policy.policy_confirm_hook
-    if hook is not None:
+        if hook is None:
+            msg = f"command denied by policy (basename {binary!r} is blocked)"
+            raise WorkspaceError(msg)
         timeout = policy.policy_confirm_timeout_seconds
         try:
             allowed = bool(hook(binary, list(argv), timeout))
         except Exception as exc:
             msg = f"command denied by policy (basename {binary!r}; confirm failed: {exc})"
             raise WorkspaceError(msg) from exc
-        if allowed:
+        if not allowed:
-            policy.policy_allowed_commands.add(binary)
-            return
             msg = (
-            f"command denied by policy (basename {binary!r} is blocked; user declined or timed out after {timeout:g}s)"
+                f"command denied by policy (basename {binary!r} is blocked; "
+                f"user declined or timed out after {timeout:g}s)"
             )
             raise WorkspaceError(msg)
-    msg = f"command denied by policy (basename {binary!r} is blocked)"
+        policy.policy_allowed_commands.add(binary)
-    raise WorkspaceError(msg)

tests/test_tools/test_workspace.py

@@ -215,6 +215,68 @@ def test_command_denylist_policy_confirm_timeout_value(workspace: object) -> Non
         clear_policy_allowed_commands()
 
 
+def test_command_denylist_reaches_through_wrappers(workspace: object) -> None:
+    del workspace
+    from plyngent.tools.workspace import clear_policy_allowed_commands, set_policy_confirm_hook
+
+    set_policy_confirm_hook(None)
+    clear_policy_allowed_commands()
+    try:
+        with pytest.raises(WorkspaceError, match="basename 'rm' is blocked"):
+            check_command_allowed(["env", "FOO=1", "rm", "-rf", "/"])
+        with pytest.raises(WorkspaceError, match="basename 'rm' is blocked"):
+            check_command_allowed(["pdm", "run", "rm", "-rf", "/"])
+        with pytest.raises(WorkspaceError, match="basename 'rm' is blocked"):
+            check_command_allowed(["timeout", "5", "rm", "-rf", "/"])
+        # The outermost denylisted program is the one named (``sudo`` here).
+        with pytest.raises(WorkspaceError, match="basename 'sudo' is blocked"):
+            check_command_allowed(["sudo", "-u", "root", "rm", "-rf", "/"])
+        # Wrapped programs that are not denylisted stay allowed.
+        check_command_allowed(["pdm", "run", "pytest", "-x"])
+        check_command_allowed(["env", "FOO=1", "echo", "ok"])
+    finally:
+        clear_policy_allowed_commands()
+
+
+def test_command_denylist_matches_lowercased_basename(workspace: object) -> None:
+    del workspace
+    from plyngent.tools.workspace import clear_policy_allowed_commands, set_policy_confirm_hook
+
+    set_policy_confirm_hook(None)
+    clear_policy_allowed_commands()
+    try:
+        with pytest.raises(WorkspaceError, match="basename 'rm' is blocked"):
+            check_command_allowed(["RM", "-rf", "/"])
+        with pytest.raises(WorkspaceError, match="basename 'rm' is blocked"):
+            check_command_allowed(["C:\\Tools\\rm.exe", "-rf", "/"])
+    finally:
+        clear_policy_allowed_commands()
+
+
+def test_command_denylist_prompts_for_the_wrapped_program(workspace: object) -> None:
+    del workspace
+    from plyngent.tools.workspace import clear_policy_allowed_commands, set_policy_confirm_hook
+
+    asked: list[str] = []
+
+    def hook(basename: str, argv: object, timeout: float) -> bool:
+        del argv, timeout
+        asked.append(basename)
+        return True
+
+    set_policy_confirm_hook(hook)
+    clear_policy_allowed_commands()
+    try:
+        check_command_allowed(["pdm", "run", "rm", "-rf", "build"])
+        assert asked == ["rm"]
+        # Session grant: the same wrapped program does not re-prompt.
+        check_command_allowed(["pdm", "run", "rm", "-rf", "dist"])
+        assert asked == ["rm"]
+    finally:
+        set_policy_confirm_hook(None)
+        clear_policy_allowed_commands()
+
+
 def test_root_required() -> None:
     from plyngent.tools.context import InstanceState, bind_instance