dongarandClaude Opus 5.5
e351d0a74e
A pure move; behaviour unchanged. - models (L9, L7): llm, catalog, probe, render, render_anthropic, context - log (L7, L6): canonical, views - state (L8): store (was state), reconcile, worker, cancel, dump - knowledge (L10, L6.4): vault, memory - conversation (L5): transport, matrix - execution (L13): ssh (was execution), testing - authority (P3, P4, L4): actions - gitea, oplog, envfile stay at the top until each gains a second module. Every import, monkeypatch target, script and doc reference rewritten. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01J5Utyfzw7gF7UhjRDEtjTk
109 lines
3.9 KiB
Python
109 lines
3.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_core.execution.ssh 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
|
|
|
|
|
|
def test_a_changed_host_key_is_its_own_failure_and_only_forgetting_clears_it(
|
|
ssh_host: SSHHost, tmp_path: Path
|
|
) -> None:
|
|
import subprocess
|
|
|
|
from nisse_core.execution.ssh import HostKeyChanged
|
|
|
|
host = SSHHost(
|
|
ssh_host.address,
|
|
key=ssh_host._key,
|
|
user=ssh_host.user,
|
|
port=ssh_host.port,
|
|
state_dir=tmp_path / "state",
|
|
)
|
|
assert host.execute("true").ok # first contact pins the key
|
|
host.close()
|
|
# Someone else's key where the pinned one was: a rebuilt VM, or not it.
|
|
subprocess.run(
|
|
["ssh-keygen", "-q", "-t", "ed25519", "-N", "", "-f", str(tmp_path / "other")],
|
|
check=True,
|
|
)
|
|
other = (tmp_path / "other.pub").read_text().split()
|
|
known = tmp_path / "state" / "known_hosts"
|
|
name = host.address if host.port == 22 else f"[{host.address}]:{host.port}"
|
|
known.write_text(f"{name} {other[0]} {other[1]}\n")
|
|
with pytest.raises(HostKeyChanged):
|
|
host.execute("true")
|
|
host.forget_host_key()
|
|
assert host.execute("true").ok # the real key pinned again
|