dongarandClaude Opus 5.5
761ebdb857
CI / checks (push) Skipped
CI / checks (pull_request) Failing after 55s
execution.SSHHost.execute is now the only way the agent runs anything (L13). Commands travel as a script on SSH's stdin, so the secrets handed to a call are never on a command line, and every value handed out is redacted from the output. Timeouts kill the remote process group, and an unreachable host is reported as such, not as a failed command. Workspace and GitRepo run on the build VM. The clone uses the repository's SSH URL, so no credential is stored in it (#89); fetches and pushes get the push key over stdin, in a short-lived ssh-agent, and PRs are still opened from the agent's host with its API token. The bench runs on the build host too. Tests reach "the build VM" through a throwaway local sshd (nisse_dev .testing); CI installs openssh-server. The live push test now pushes with the sandbox's deploy key (NISSE_PUSH_KEY). Fixes #89. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01PPQ14L3LikyefCMFBqEbKE
78 lines
2.9 KiB
Python
78 lines
2.9 KiB
Python
"""Running commands on a tool VM over SSH (L13), against a throwaway sshd."""
|
|
|
|
import os
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from nisse_dev.execution import REDACTED, HostUnavailable, SSHHost, redact
|
|
|
|
|
|
def test_exit_codes_and_streams_come_back(ssh_host: SSHHost) -> None:
|
|
result = ssh_host.execute("echo out; echo err >&2; exit 3")
|
|
assert (result.exit_code, result.stdout, result.stderr) == (3, "out\n", "err\n")
|
|
assert not result.ok
|
|
|
|
|
|
def test_cwd_and_stdin(ssh_host: SSHHost, tmp_path: Path) -> None:
|
|
result = ssh_host.execute("pwd; cat", cwd=str(tmp_path), stdin="line one\nline two")
|
|
assert result.stdout == f"{tmp_path}\nline one\nline two\n"
|
|
|
|
|
|
def test_stdin_that_looks_like_a_delimiter_survives(ssh_host: SSHHost) -> None:
|
|
payload = "NISSE_EOF\nEOF\n'quoted' $notexpanded `nor this`"
|
|
assert ssh_host.execute("cat", stdin=payload).stdout == payload + "\n"
|
|
|
|
|
|
def test_without_stdin_a_command_cannot_read_the_script(ssh_host: SSHHost) -> None:
|
|
# The script arrives on the same channel; the command must see /dev/null.
|
|
result = ssh_host.execute("cat; echo done")
|
|
assert result.stdout == "done\n"
|
|
|
|
|
|
def test_a_missing_cwd_fails_before_running(ssh_host: SSHHost, tmp_path: Path) -> None:
|
|
marker = tmp_path / "ran"
|
|
result = ssh_host.execute(f"touch {marker}", cwd=str(tmp_path / "absent"))
|
|
assert result.exit_code == 125
|
|
assert not marker.exists()
|
|
|
|
|
|
def test_a_command_exiting_255_is_not_mistaken_for_ssh(ssh_host: SSHHost) -> None:
|
|
assert ssh_host.execute("exit 255").exit_code == 254
|
|
|
|
|
|
def test_an_unreachable_host_raises(tmp_path: Path) -> None:
|
|
host = SSHHost("127.0.0.1", key=tmp_path / "k", state_dir=tmp_path / "s", port=1)
|
|
with pytest.raises(HostUnavailable):
|
|
host.execute("true")
|
|
|
|
|
|
def test_a_timeout_is_reported(ssh_host: SSHHost) -> None:
|
|
result = ssh_host.execute("sleep 30", timeout_s=1)
|
|
assert result.timed_out
|
|
|
|
|
|
def test_secrets_reach_the_command_and_never_its_output(ssh_host: SSHHost) -> None:
|
|
value = "s-" + os.urandom(12).hex()
|
|
result = ssh_host.execute(
|
|
'echo "$API_KEY"; echo "$API_KEY" >&2', secrets={"API_KEY": value}
|
|
)
|
|
assert result.ok
|
|
assert value not in result.stdout + result.stderr
|
|
assert result.stdout == f"{REDACTED}\n"
|
|
|
|
|
|
def test_secrets_are_never_on_a_command_line(ssh_host: SSHHost, tmp_path: Path) -> None:
|
|
# Made up here, so no command line in this test run can contain it except
|
|
# by leaking it.
|
|
value = "s-" + os.urandom(12).hex()
|
|
seen = tmp_path / "ps.txt"
|
|
ssh_host.execute(f"ps -eo args > {seen}", secrets={"API_KEY": value})
|
|
assert "bash -s" in seen.read_text() # the listing did include our shell
|
|
assert value not in seen.read_text()
|
|
|
|
|
|
def test_redact_masks_longest_values_first() -> None:
|
|
assert redact("abc-abcdef", ["abcdef", "abc-abcdef"]) == REDACTED
|
|
assert redact("short ab stays", ["ab"]) == "short ab stays" # too short to mask
|