mirror of
https://github.com/NCBM/plyngent.git
synced 2026-11-07 07:17:54 +08:00
NCBM
8c289db545
What the scan resolves (destination, options, jammed `-p2222`, mosh's `--`, a hop behind `sudo`), what it leaves alone (a client with no command), and both gates: `check_command_allowed` now refuses `ssh prod rm -rf /` and prompts for the `sudo` behind the hop, while harmless remote commands stay allowed; the review prints a remote `bash -c` code block and always asks under `BatchMode=yes`.
440 lines
17 KiB
Python
440 lines
17 KiB
Python
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from plyngent.tools import (
|
|
AccessMode,
|
|
WorkspaceError,
|
|
active_workspace_policy,
|
|
check_command_allowed,
|
|
get_trash_enabled,
|
|
get_workspace_root,
|
|
image_input_refusal,
|
|
mode_covers,
|
|
parse_access_mode,
|
|
resolve_path,
|
|
set_command_denylist,
|
|
set_image_in,
|
|
set_path_denylist,
|
|
set_trash_enabled,
|
|
)
|
|
from plyngent.tools.context import SessionState, bind_session
|
|
|
|
|
|
def test_resolve_relative_and_absolute(workspace: object) -> None:
|
|
from pathlib import Path
|
|
|
|
assert isinstance(workspace, Path)
|
|
_ = (workspace / "a.txt").write_text("x", encoding="utf-8")
|
|
assert resolve_path("a.txt") == workspace / "a.txt"
|
|
assert resolve_path(workspace / "a.txt") == workspace / "a.txt"
|
|
|
|
|
|
def test_escape_rejected(workspace: object) -> None:
|
|
del workspace
|
|
with pytest.raises(WorkspaceError, match="escapes"):
|
|
_ = resolve_path("../outside")
|
|
|
|
|
|
def test_escape_error_hints_request_tool(workspace: object) -> None:
|
|
del workspace
|
|
with pytest.raises(WorkspaceError, match="request_directory_access"):
|
|
_ = resolve_path("../outside")
|
|
|
|
|
|
@pytest.mark.parametrize("required", [AccessMode.READ, AccessMode.WRITE, AccessMode.EXEC])
|
|
def test_escape_error_names_the_mode_to_request(workspace: object, required: AccessMode) -> None:
|
|
"""The refusal carries the lowest access mode that would cover the call."""
|
|
del workspace
|
|
with pytest.raises(WorkspaceError, match=f"mode='{required.name.lower()}'"):
|
|
_ = resolve_path("../outside", required=required)
|
|
|
|
|
|
def test_access_mode_parse_and_covers() -> None:
|
|
assert parse_access_mode(" Read ") is AccessMode.READ
|
|
assert parse_access_mode("EXEC") is AccessMode.EXEC
|
|
assert parse_access_mode("bogus") is None
|
|
assert mode_covers(AccessMode.EXEC, AccessMode.WRITE)
|
|
assert mode_covers(AccessMode.WRITE, AccessMode.WRITE)
|
|
assert not mode_covers(AccessMode.READ, AccessMode.WRITE)
|
|
|
|
|
|
def test_trash_switch(workspace: object) -> None:
|
|
"""Deletes go to the trash until a host clears the switch."""
|
|
del workspace
|
|
assert get_trash_enabled() is True
|
|
set_trash_enabled(enabled=False)
|
|
assert get_trash_enabled() is False
|
|
set_trash_enabled(enabled=True)
|
|
assert get_trash_enabled() is True
|
|
|
|
|
|
def test_image_switch(workspace: object) -> None:
|
|
"""A picture is refused once the host says the model cannot take one."""
|
|
del workspace
|
|
assert image_input_refusal() is None
|
|
|
|
set_image_in(image_in=False)
|
|
unnamed = image_input_refusal()
|
|
assert unnamed is not None
|
|
assert "does not accept image input" in unnamed
|
|
|
|
set_image_in(image_in=False, provider="ds", model="text-only")
|
|
named = image_input_refusal()
|
|
assert named is not None
|
|
assert "model 'text-only' does not accept image input" in named
|
|
assert '[providers.ds.models."text-only"]' in named
|
|
|
|
set_image_in(image_in=True, provider="ds", model="vision")
|
|
assert image_input_refusal() is None
|
|
|
|
|
|
def test_the_image_switch_needs_a_declaration(workspace: object) -> None:
|
|
"""No host verdict (``None``) sends the picture: only ``False`` refuses."""
|
|
del workspace
|
|
policy = active_workspace_policy()
|
|
assert policy.image_in is None
|
|
assert image_input_refusal() is None
|
|
|
|
|
|
def test_config_grant_modes(workspace: object, tmp_path_factory: pytest.TempPathFactory) -> None:
|
|
del workspace
|
|
outside = tmp_path_factory.mktemp("granted")
|
|
target = outside / "data.txt"
|
|
_ = target.write_text("x", encoding="utf-8")
|
|
policy = active_workspace_policy()
|
|
policy.config_allow[outside] = AccessMode.READ
|
|
try:
|
|
assert resolve_path(str(target), required=AccessMode.READ) == target
|
|
with pytest.raises(WorkspaceError, match="escapes"):
|
|
_ = resolve_path(str(target), required=AccessMode.WRITE)
|
|
policy.config_allow[outside] = AccessMode.WRITE
|
|
assert resolve_path(str(target), required=AccessMode.WRITE) == target
|
|
with pytest.raises(WorkspaceError, match="escapes"):
|
|
_ = resolve_path(str(target), required=AccessMode.EXEC)
|
|
finally:
|
|
policy.config_allow.clear()
|
|
|
|
|
|
def test_session_grant_covers_subtree(workspace: object, tmp_path_factory: pytest.TempPathFactory) -> None:
|
|
del workspace
|
|
outside = tmp_path_factory.mktemp("session")
|
|
nested = outside / "a" / "b"
|
|
nested.mkdir(parents=True)
|
|
with bind_session(SessionState(access_grants={outside: AccessMode.EXEC})):
|
|
assert resolve_path(str(nested), required=AccessMode.EXEC) == nested
|
|
with pytest.raises(WorkspaceError, match="escapes"):
|
|
_ = resolve_path(str(nested), required=AccessMode.READ)
|
|
|
|
|
|
def test_file_grant_is_exact_path(workspace: object, tmp_path_factory: pytest.TempPathFactory) -> None:
|
|
del workspace
|
|
outside = tmp_path_factory.mktemp("file-grant")
|
|
granted = outside / "one.txt"
|
|
sibling = outside / "two.txt"
|
|
_ = granted.write_text("1", encoding="utf-8")
|
|
_ = sibling.write_text("2", encoding="utf-8")
|
|
policy = active_workspace_policy()
|
|
policy.config_allow[granted] = AccessMode.READ
|
|
try:
|
|
assert resolve_path(str(granted), required=AccessMode.READ) == granted
|
|
with pytest.raises(WorkspaceError, match="escapes"):
|
|
_ = resolve_path(str(sibling), required=AccessMode.READ)
|
|
# A file grant does not cover its parent directory.
|
|
with pytest.raises(WorkspaceError, match="escapes"):
|
|
_ = resolve_path(str(outside), required=AccessMode.READ)
|
|
finally:
|
|
policy.config_allow.clear()
|
|
|
|
|
|
def test_denylist_wins_over_grant(workspace: object, tmp_path_factory: pytest.TempPathFactory) -> None:
|
|
del workspace
|
|
outside = tmp_path_factory.mktemp("denied")
|
|
secret = outside / "secret"
|
|
secret.mkdir()
|
|
key = secret / "key"
|
|
_ = key.write_text("k", encoding="utf-8")
|
|
policy = active_workspace_policy()
|
|
policy.config_allow[outside] = AccessMode.WRITE
|
|
set_path_denylist(["/secret/"])
|
|
try:
|
|
with pytest.raises(WorkspaceError, match="matched '/secret/'"):
|
|
_ = resolve_path(str(key), required=AccessMode.READ)
|
|
finally:
|
|
set_path_denylist(None)
|
|
policy.config_allow.clear()
|
|
|
|
|
|
def test_path_denylist(workspace: object) -> None:
|
|
from pathlib import Path
|
|
|
|
assert isinstance(workspace, Path)
|
|
secrets = workspace / "secrets"
|
|
secrets.mkdir()
|
|
_ = (secrets / "key").write_text("k", encoding="utf-8")
|
|
set_path_denylist(["/secrets/"])
|
|
with pytest.raises(WorkspaceError, match="matched '/secrets/'"):
|
|
_ = resolve_path("secrets/key")
|
|
set_path_denylist(None)
|
|
|
|
|
|
def test_command_denylist(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()
|
|
with pytest.raises(WorkspaceError, match="basename 'rm' is blocked"):
|
|
check_command_allowed(["rm", "-rf", "/"])
|
|
check_command_allowed(["echo", "ok"])
|
|
set_command_denylist(None)
|
|
|
|
|
|
def test_command_denylist_policy_confirm_allow(workspace: object) -> None:
|
|
del workspace
|
|
from plyngent.tools.workspace import (
|
|
clear_policy_allowed_commands,
|
|
set_policy_confirm_hook,
|
|
)
|
|
|
|
calls: list[tuple[str, float]] = []
|
|
|
|
def hook(basename: str, argv: object, timeout: float) -> bool:
|
|
del argv
|
|
calls.append((basename, timeout))
|
|
return basename == "sudo"
|
|
|
|
set_policy_confirm_hook(hook)
|
|
clear_policy_allowed_commands()
|
|
try:
|
|
check_command_allowed(["sudo", "echo", "test"])
|
|
assert calls == [("sudo", 30.0)]
|
|
# Session grant: second call does not re-prompt.
|
|
check_command_allowed(["sudo", "id"])
|
|
assert len(calls) == 1
|
|
finally:
|
|
set_policy_confirm_hook(None)
|
|
clear_policy_allowed_commands()
|
|
|
|
|
|
def test_command_denylist_policy_confirm_deny(workspace: object) -> None:
|
|
del workspace
|
|
from plyngent.tools.workspace import (
|
|
clear_policy_allowed_commands,
|
|
set_policy_confirm_hook,
|
|
)
|
|
|
|
set_policy_confirm_hook(lambda *_a: False)
|
|
clear_policy_allowed_commands()
|
|
try:
|
|
with pytest.raises(WorkspaceError, match="declined or timed out"):
|
|
check_command_allowed(["sudo", "echo", "x"])
|
|
finally:
|
|
set_policy_confirm_hook(None)
|
|
clear_policy_allowed_commands()
|
|
|
|
|
|
def test_command_denylist_policy_confirm_timeout_value(workspace: object) -> None:
|
|
del workspace
|
|
from plyngent.tools.workspace import (
|
|
clear_policy_allowed_commands,
|
|
set_policy_confirm_hook,
|
|
set_policy_confirm_timeout,
|
|
)
|
|
|
|
seen: list[float] = []
|
|
|
|
def hook(basename: str, argv: object, timeout: float) -> bool:
|
|
del basename, argv
|
|
seen.append(timeout)
|
|
return False
|
|
|
|
set_policy_confirm_timeout(5.0)
|
|
set_policy_confirm_hook(hook)
|
|
clear_policy_allowed_commands()
|
|
try:
|
|
with pytest.raises(WorkspaceError):
|
|
check_command_allowed(["rm", "x"])
|
|
assert seen == [5.0]
|
|
finally:
|
|
set_policy_confirm_timeout(30.0)
|
|
set_policy_confirm_hook(None)
|
|
clear_policy_allowed_commands()
|
|
|
|
|
|
def test_command_denylist_sees_the_program_behind_value_free_flags(workspace: object) -> None:
|
|
"""`exec -c` / `command -p` take no value, so the program stays visible."""
|
|
del workspace
|
|
from plyngent.tools.workspace import clear_policy_allowed_commands, set_policy_confirm_hook
|
|
|
|
set_policy_confirm_hook(None)
|
|
clear_policy_allowed_commands()
|
|
set_command_denylist(None)
|
|
try:
|
|
for argv in (
|
|
["exec", "-c", "rm", "-rf", "/"],
|
|
["command", "-p", "rm", "-rf", "/"],
|
|
["command", "-v", "rm"],
|
|
):
|
|
with pytest.raises(WorkspaceError, match="basename 'rm' is blocked"):
|
|
check_command_allowed(list(argv))
|
|
# ``init`` here is a git subcommand, not the denylisted program: the
|
|
# command resolves to ``git`` and stays allowed.
|
|
check_command_allowed(["exec", "-c", "git", "-c", "init.defaultBranch=main", "init"])
|
|
finally:
|
|
clear_policy_allowed_commands()
|
|
set_command_denylist(None)
|
|
|
|
|
|
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_reaches_the_far_end_of_ssh(workspace: object) -> None:
|
|
"""A remote command is judged by the program that runs there, not by ``ssh``."""
|
|
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(["ssh", "prod", "rm", "-rf", "/"])
|
|
with pytest.raises(WorkspaceError, match="basename 'sudo' is blocked"):
|
|
check_command_allowed(["ssh", "-p", "2222", "prod", "sudo", "systemctl", "restart", "web"])
|
|
# A remote shell (no command) and harmless remote commands stay allowed.
|
|
check_command_allowed(["ssh", "prod"])
|
|
check_command_allowed(["ssh", "-o", "BatchMode=yes", "prod", "uptime"])
|
|
check_command_allowed(["mosh", "prod", "--", "htop"])
|
|
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_match_command_denylist_entries() -> None:
|
|
from plyngent.tools import match_command_denylist
|
|
|
|
denylist = frozenset({"rm", "systemctl reboot poweroff", "loginctl kill-session"})
|
|
hits = match_command_denylist(["sudo", "rm", "-rf", "/"], denylist)
|
|
assert [(hit.program, hit.keyword) for hit in hits] == [("rm", None)]
|
|
assert [hit.program for hit in match_command_denylist(["env", "FOO=1", "rm", "x"], denylist)] == ["rm"]
|
|
hits = match_command_denylist(["sudo", "sudo", "rm", "x"], frozenset({"sudo", "rm"}))
|
|
assert [hit.program for hit in hits] == ["sudo", "rm"]
|
|
# Subcommand entries match the program's own arguments, targets included.
|
|
hits = match_command_denylist(["systemctl", "reboot"], denylist)
|
|
assert [(hit.program, hit.keyword, hit.entry) for hit in hits] == [
|
|
("systemctl", "reboot", "systemctl reboot poweroff")
|
|
]
|
|
assert match_command_denylist(["systemctl", "start", "poweroff.target"], denylist)[0].keyword == "poweroff"
|
|
assert match_command_denylist(["pdm", "run", "loginctl", "kill-session"], denylist)[0].program == "loginctl"
|
|
# Read-only and unrelated calls stay clean.
|
|
assert match_command_denylist(["systemctl", "status", "sshd"], denylist) == []
|
|
assert match_command_denylist(["loginctl", "list-sessions"], denylist) == []
|
|
|
|
|
|
def test_command_denylist_power_subcommands(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 'systemctl' is blocked: 'reboot'"):
|
|
check_command_allowed(["systemctl", "reboot"])
|
|
with pytest.raises(WorkspaceError, match="basename 'systemctl' is blocked: 'reboot'"):
|
|
check_command_allowed(["systemctl", "start", "reboot.target"])
|
|
with pytest.raises(WorkspaceError, match="basename 'systemctl' is blocked: 'poweroff'"):
|
|
check_command_allowed(["systemctl", "--user", "poweroff"])
|
|
with pytest.raises(WorkspaceError, match="basename 'systemctl' is blocked: 'emergency'"):
|
|
check_command_allowed(["timeout", "3", "systemctl", "emergency"])
|
|
with pytest.raises(WorkspaceError, match="basename 'loginctl' is blocked: 'kill-session'"):
|
|
check_command_allowed(["loginctl", "kill-session", "3"])
|
|
with pytest.raises(WorkspaceError, match="basename 'pm-suspend' is blocked"):
|
|
check_command_allowed(["pm-suspend"])
|
|
with pytest.raises(WorkspaceError, match="basename 'run0' is blocked"):
|
|
check_command_allowed(["run0", "id"])
|
|
with pytest.raises(WorkspaceError, match="basename 'systemd-run' is blocked"):
|
|
check_command_allowed(["systemd-run", "--user", "echo", "hi"])
|
|
# The outermost denylisted program is reported first.
|
|
with pytest.raises(WorkspaceError, match="basename 'sudo' is blocked"):
|
|
check_command_allowed(["sudo", "systemctl", "reboot"])
|
|
# Read-only systemd calls stay allowed.
|
|
check_command_allowed(["systemctl", "status", "sshd"])
|
|
check_command_allowed(["systemctl", "list-units", "--type=service"])
|
|
check_command_allowed(["systemctl", "show", "-p", "CanReboot"])
|
|
check_command_allowed(["loginctl", "list-sessions"])
|
|
finally:
|
|
clear_policy_allowed_commands()
|
|
|
|
|
|
def test_root_required() -> None:
|
|
from plyngent.tools.context import InstanceState, bind_instance
|
|
|
|
with bind_instance(InstanceState()), pytest.raises(WorkspaceError, match="workspace root is not set"):
|
|
_ = get_workspace_root()
|
|
|
|
|
|
def test_unbound_instance_errors() -> None:
|
|
with pytest.raises(WorkspaceError, match="instance state is not bound"):
|
|
_ = get_workspace_root()
|