From 9501efb3d753ac275fb831696235559ab2a886cf Mon Sep 17 00:00:00 2001 From: worldmozara Date: Tue, 6 Oct 2026 19:42:06 +0800 Subject: [PATCH] core/tools: block a denylisted program behind env wrappers and run launchers --- src/plyngent/tools/workspace.py | 40 +++++++++++-------- tests/test_tools/test_workspace.py | 62 ++++++++++++++++++++++++++++++ 2 files changed, 85 insertions(+), 17 deletions(-) diff --git a/src/plyngent/tools/workspace.py b/src/plyngent/tools/workspace.py index a9eb2da..64898d4 100644 --- a/src/plyngent/tools/workspace.py +++ b/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 - if binary not in policy.command_denylist: - return - if binary in policy.policy_allowed_commands: - return - hook = policy.policy_confirm_hook - if hook is not None: + scan = unwrap_command(argv) + for binary in (*scan.wrappers, scan.base): + if binary not in policy.command_denylist or binary in policy.policy_allowed_commands: + continue + hook = policy.policy_confirm_hook + 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: - 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)" - ) - raise WorkspaceError(msg) - msg = f"command denied by policy (basename {binary!r} is blocked)" - raise WorkspaceError(msg) + if not allowed: + msg = ( + f"command denied by policy (basename {binary!r} is blocked; " + f"user declined or timed out after {timeout:g}s)" + ) + raise WorkspaceError(msg) + policy.policy_allowed_commands.add(binary) diff --git a/tests/test_tools/test_workspace.py b/tests/test_tools/test_workspace.py index 3a03b42..1846776 100644 --- a/tests/test_tools/test_workspace.py +++ b/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