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

@@ -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")

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)"],

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