From 8c289db5450cbaa108d5301a027e69c9d38feabe Mon Sep 17 00:00:00 2001 From: worldmozara Date: Fri, 23 Oct 2026 10:37:14 +0800 Subject: [PATCH] test: cover the far end of ssh 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`. --- tests/test_tools/test_command_scan.py | 56 ++++++++++++++++++++++++++- tests/test_tools/test_danger.py | 28 ++++++++++++++ tests/test_tools/test_workspace.py | 20 ++++++++++ 3 files changed, 103 insertions(+), 1 deletion(-) diff --git a/tests/test_tools/test_command_scan.py b/tests/test_tools/test_command_scan.py index a4d85d3..2567e80 100644 --- a/tests/test_tools/test_command_scan.py +++ b/tests/test_tools/test_command_scan.py @@ -1,6 +1,6 @@ from __future__ import annotations -from plyngent.tools.command_scan import basename, escalators_in_chain, is_interpreter, unwrap_command +from plyngent.tools.command_scan import basename, dash_c_code, escalators_in_chain, is_interpreter, unwrap_command def test_basename() -> None: @@ -132,6 +132,60 @@ def test_unwrap_empty_argv() -> None: assert scan.command == () +def test_unwrap_remote_command() -> None: + """``ssh``/``mosh`` are unwrapped to the program they run on the far end.""" + scan = unwrap_command(["ssh", "prod", "rm", "-rf", "/"]) + assert scan.wrappers == ("ssh",) + assert scan.base == "rm" + assert scan.command == ("rm", "-rf", "/") + assert scan.remote is not None + assert (scan.remote.wrapper, scan.remote.host, scan.remote.batch) == ("ssh", "prod", False) + + scan = unwrap_command(["ssh", "-p", "2222", "-i", "~/.ssh/id", "user@prod", "ls", "-la"]) + assert scan.base == "ls" + assert scan.command == ("ls", "-la") + assert scan.remote is not None and scan.remote.host == "user@prod" + + # Jammed values, and a hop behind a local wrapper. + scan = unwrap_command(["ssh", "-p2222", "prod", "uptime"]) + assert scan.base == "uptime" + assert unwrap_command(["sudo", "ssh", "prod", "id"]).base == "id" + scan = unwrap_command(["mosh", "prod", "--", "htop"]) + assert scan.base == "htop" + assert scan.remote is not None and scan.remote.wrapper == "mosh" + + +def test_unwrap_remote_batch_mode() -> None: + """``-o BatchMode=yes`` is reported: the code runs with no prompt at either end.""" + for argv in ( + ["ssh", "-o", "BatchMode=yes", "prod", "deploy.sh"], + ["ssh", "-oBatchMode=yes", "prod", "deploy.sh"], + ["ssh", "-o", "batchmode=YES", "-p", "22", "prod", "deploy.sh"], + ): + scan = unwrap_command(argv) + assert scan.remote is not None, argv + assert scan.remote.batch is True, argv + scan = unwrap_command(["ssh", "-o", "BatchMode=no", "prod", "deploy.sh"]) + assert scan.remote is not None and scan.remote.batch is False + + +def test_unwrap_remote_client_alone() -> None: + """A client with no command opens a remote shell: nothing to unwrap.""" + for argv in (["ssh", "prod"], ["ssh", "-N", "prod"], ["ssh", "-G", "prod"], ["ssh"], ["mosh", "prod"]): + scan = unwrap_command(argv) + assert scan.wrappers == (), argv + assert scan.base == argv[0], argv + assert scan.remote is None, argv + + +def test_unwrap_remote_command_still_reviews_what_runs() -> None: + """The remote slice is parsed like a local command (code, escalation).""" + scan = unwrap_command(["ssh", "prod", "bash", "-c", "rm -rf /"]) + assert scan.base == "bash" + assert dash_c_code(scan.command) == (1, "rm -rf /") + assert escalators_in_chain(["ssh", "prod", "sudo", "rm", "-rf", "/"]) == ("sudo",) + + def test_is_interpreter_ignores_version_suffix() -> None: assert is_interpreter("python") assert is_interpreter("python3.12") diff --git a/tests/test_tools/test_danger.py b/tests/test_tools/test_danger.py index 851516b..7c841cb 100644 --- a/tests/test_tools/test_danger.py +++ b/tests/test_tools/test_danger.py @@ -160,6 +160,34 @@ def test_versioned_interpreter_is_still_reviewed() -> None: assert "import os" in reason +def test_a_remote_command_is_judged_by_what_runs_there() -> None: + """``ssh``/``mosh`` do not hide the program from the review.""" + # The remote slice decides, exactly like a local command. + plain = _reason("run_argv", {"argv": ["ssh", "prod", "uptime"]}) + assert plain is None + code = _reason("run_argv", {"argv": ["ssh", "prod", "bash", "-c", "rm -rf /"]}) + assert code is not None + assert "bash -c" in code + assert "rm -rf" in code + # And it is the same confirm through a PTY session. + pty = _reason("open_pty", {"command": ["ssh", "prod", "python3"]}) + assert pty is not None and "python3" in pty + + +def test_batch_mode_is_always_reviewed() -> None: + """``BatchMode=yes`` cannot prompt at either end, so the command is shown.""" + reason = _reason("run_argv", {"argv": ["ssh", "-o", "BatchMode=yes", "prod", "deploy.sh"]}) + assert reason is not None + assert "BatchMode=yes" in reason + assert "deploy.sh" in reason + assert _danger("run_argv", {"argv": ["ssh", "-oBatchMode=yes", "prod", "deploy.sh"]}).yolo_skip is True + # An escalation behind the hop still never skips the confirm under YOLO. + escalated = _danger("run_argv", {"argv": ["ssh", "prod", "sudo", "python3", "-c", "x=1"]}) + assert escalated.yolo_skip is False + assert "escalation sudo" in escalated.text + assert "interpreter 'python3'" in escalated.text + + def test_wrappers_reach_the_interpreter() -> None: for argv in ( ["env", "FOO=1", "python", "-c", "print(1)"], diff --git a/tests/test_tools/test_workspace.py b/tests/test_tools/test_workspace.py index 88e1c89..b9cccfd 100644 --- a/tests/test_tools/test_workspace.py +++ b/tests/test_tools/test_workspace.py @@ -312,6 +312,26 @@ def test_command_denylist_reaches_through_wrappers(workspace: object) -> None: 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