mirror of
https://github.com/NCBM/plyngent.git
synced 2026-11-07 07:17:54 +08:00
NCBM
4bab780603
The four kinds and the words each produces, at the registry level and through the four policy gates: a timeout names its deadline, a decline carries the comment, and neither is reported as the other. The existing denial assertions now expect the specific clause.
507 lines
20 KiB
Python
507 lines
20 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_escape_error_offers_a_temporary_workspace_for_a_write(workspace: object) -> None:
|
|
"""A refused write is often scratch, which never needs a grant."""
|
|
del workspace
|
|
with pytest.raises(WorkspaceError, match="new_temporary_workspace"):
|
|
_ = resolve_path("../scratch.txt", required=AccessMode.WRITE)
|
|
|
|
|
|
@pytest.mark.parametrize("required", [AccessMode.READ, AccessMode.EXEC])
|
|
def test_escape_error_hides_the_scratch_hint_from_other_modes(workspace: object, required: AccessMode) -> None:
|
|
del workspace
|
|
with pytest.raises(WorkspaceError) as refusal:
|
|
_ = resolve_path("../scratch.txt", required=required)
|
|
assert "new_temporary_workspace" not in str(refusal.value)
|
|
|
|
|
|
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_a_denylist_answer_is_recorded_as_a_user_process_approval(workspace: object) -> None:
|
|
"""The answer that let the call through is what the result has to say."""
|
|
del workspace
|
|
from plyngent.agent import Approval
|
|
from plyngent.tools.context import collect_approvals
|
|
from plyngent.tools.workspace import clear_policy_allowed_commands, set_policy_confirm_hook
|
|
|
|
set_policy_confirm_hook(lambda *_args: True)
|
|
clear_policy_allowed_commands()
|
|
try:
|
|
with collect_approvals() as recorded:
|
|
check_command_allowed(["sudo", "id"])
|
|
finally:
|
|
set_policy_confirm_hook(None)
|
|
clear_policy_allowed_commands()
|
|
assert recorded == [Approval("user", "process")]
|
|
|
|
|
|
def test_a_matching_denylist_grant_is_recorded_as_a_process_grant(workspace: object) -> None:
|
|
"""Nobody is asked the second time, and the result says so."""
|
|
del workspace
|
|
from plyngent.agent import Approval
|
|
from plyngent.tools.context import collect_approvals
|
|
from plyngent.tools.workspace import clear_policy_allowed_commands, grant_policy_command
|
|
|
|
clear_policy_allowed_commands()
|
|
grant_policy_command("sudo")
|
|
try:
|
|
with collect_approvals() as recorded:
|
|
check_command_allowed(["sudo", "id"])
|
|
finally:
|
|
clear_policy_allowed_commands()
|
|
assert recorded == [Approval("grant", "process")]
|
|
|
|
|
|
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="user declined"):
|
|
check_command_allowed(["sudo", "echo", "x"])
|
|
finally:
|
|
set_policy_confirm_hook(None)
|
|
clear_policy_allowed_commands()
|
|
|
|
|
|
def test_command_denylist_policy_timeout_names_the_deadline(workspace: object) -> None:
|
|
"""Silence is reported as the clock, never as the human's decision."""
|
|
del workspace
|
|
from plyngent.agent import Denial, DenialKind
|
|
from plyngent.tools.workspace import clear_policy_allowed_commands, set_policy_confirm_hook
|
|
|
|
set_policy_confirm_hook(lambda *_a: Denial(DenialKind.TIMEOUT, seconds=3.0))
|
|
clear_policy_allowed_commands()
|
|
try:
|
|
with pytest.raises(WorkspaceError, match="no answer within 3s") as refusal:
|
|
check_command_allowed(["sudo", "echo", "x"])
|
|
finally:
|
|
set_policy_confirm_hook(None)
|
|
clear_policy_allowed_commands()
|
|
assert "declined" not in str(refusal.value)
|
|
|
|
|
|
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()
|