chore: shared e2e ssh harness

tests/e2e/harness/ec2helpers.go

@@ -3,10 +3,8 @@
 package harness
 
 import (
-	"bytes"
 	"context"
 	"fmt"
-	"os/exec"
 	"strconv"
 	"strings"
 	"testing"
@@ -72,25 +70,7 @@ func InstancePrivateIP(t *testing.T, c *AWSClient, instanceID string) string {
 func LsblkRootGiB(t *testing.T, tgt SSHTarget) int {
 	t.Helper()
 	cmd := `SRC=$(findmnt -n -o SOURCE /); PKN=$(lsblk -n -o PKNAME "$SRC" 2>/dev/null | head -1); DEV=${PKN:-$(basename "$SRC")}; lsblk -b -d -n -o SIZE "/dev/$DEV"`
-	args := []string{
-		"-o", "StrictHostKeyChecking=no",
-		"-o", "UserKnownHostsFile=/dev/null",
-		"-o", "LogLevel=ERROR",
-		"-o", "ConnectTimeout=5",
-		"-o", "BatchMode=yes",
-		"-p", strconv.Itoa(tgt.Port),
-		"-i", tgt.KeyPath,
-		tgt.User + "@" + tgt.Host,
-		cmd,
-	}
-	var stdout, stderr bytes.Buffer
-	sshCmd := exec.Command("ssh", args...)
-	sshCmd.Stdout = &stdout
-	sshCmd.Stderr = &stderr
-	if err := sshCmd.Run(); err != nil {
-		t.Fatalf("ssh lsblk %s@%s:%d failed: %v\nstderr: %s", tgt.User, tgt.Host, tgt.Port, err, stderr.String())
-	}
-	raw := strings.TrimSpace(stdout.String())
+	raw := strings.TrimSpace(RunSSH(t, tgt, cmd))
 	bytesN, err := strconv.ParseInt(raw, 10, 64)
 	if err != nil {
 		t.Fatalf("ssh lsblk %s@%s:%d: parse %q: %v", tgt.User, tgt.Host, tgt.Port, raw, err)

tests/e2e/harness/guest_data.go

@@ -54,18 +54,7 @@ func GuestExec(tgt SSHTarget, cmd string) (string, error) {
 func GuestExecTimeout(tgt SSHTarget, cmd string, timeout time.Duration) (string, error) {
 	ctx, cancel := context.WithTimeout(context.Background(), timeout)
 	defer cancel()
-	args := []string{
-		"-o", "StrictHostKeyChecking=no",
-		"-o", "UserKnownHostsFile=/dev/null",
-		"-o", "LogLevel=ERROR",
-		"-o", "ConnectTimeout=5",
-		"-o", "BatchMode=yes",
-		"-p", strconv.Itoa(tgt.Port),
-		"-i", tgt.KeyPath,
-		tgt.User + "@" + tgt.Host,
-		cmd,
-	}
-	out, err := exec.CommandContext(ctx, "ssh", args...).CombinedOutput()
+	out, err := exec.CommandContext(ctx, "ssh", guestSSHArgs(tgt, 5, cmd)...).CombinedOutput()
 	if err != nil && ctx.Err() == context.DeadlineExceeded {
 		return string(out), fmt.Errorf("guest command exceeded its %s timeout (still running when the deadline hit): %w: %w", timeout, context.DeadlineExceeded, err)
 	}

tests/e2e/harness/ssh.go

@@ -3,14 +3,21 @@
 package harness
 
 import (
+	"bytes"
 	"context"
 	"fmt"
 	"net"
 	"os"
 	"os/exec"
 	"strconv"
+	"strings"
+	"sync/atomic"
+	"testing"
 	"time"
 
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/ec2"
+
 	"golang.org/x/crypto/ssh"
 )
 
@@ -28,24 +35,181 @@ type SSH interface {
 // RunGuestSSH executes cmd in a guest addressed by an SSHTarget. It returns
 // combined output so callers can retry assertions without terminating a test.
 func RunGuestSSH(ctx context.Context, target SSHTarget, cmd string) ([]byte, error) {
-	args := []string{
-		"-o", "StrictHostKeyChecking=no",
-		"-o", "UserKnownHostsFile=/dev/null",
-		"-o", "LogLevel=ERROR",
-		"-o", "ConnectTimeout=5",
-		"-o", "BatchMode=yes",
-		"-p", strconv.Itoa(target.Port),
-		"-i", target.KeyPath,
-		target.User + "@" + target.Host,
-		cmd,
-	}
-	out, err := exec.CommandContext(ctx, "ssh", args...).CombinedOutput()
+	out, err := exec.CommandContext(ctx, "ssh", guestSSHArgs(target, 5, cmd)...).CombinedOutput()
 	if err != nil {
 		return out, fmt.Errorf("guest ssh %s@%s:%d: %w: %s", target.User, target.Host, target.Port, err, out)
 	}
 	return out, nil
 }
 
+// guestSSHArgs is the non-interactive, no-host-key-check ssh(1) argument list
+// every guest probe uses; connectTimeout is in seconds.
+func guestSSHArgs(tgt SSHTarget, connectTimeout int, cmd string) []string {
+	return []string{
+		"-o", "StrictHostKeyChecking=no",
+		"-o", "UserKnownHostsFile=/dev/null",
+		"-o", "LogLevel=ERROR",
+		"-o", "ConnectTimeout=" + strconv.Itoa(connectTimeout),
+		"-o", "BatchMode=yes",
+		"-p", strconv.Itoa(tgt.Port),
+		"-i", tgt.KeyPath,
+		tgt.User + "@" + tgt.Host,
+		cmd,
+	}
+}
+
+// RunSSH runs command in the guest and returns stdout. It t.Fatals on a
+// non-zero exit so callers can chain assertions on the output.
+func RunSSH(t *testing.T, tgt SSHTarget, command string) string {
+	t.Helper()
+	var stdout, stderr bytes.Buffer
+	cmd := exec.Command("ssh", guestSSHArgs(tgt, 5, command)...)
+	cmd.Stdout = &stdout
+	cmd.Stderr = &stderr
+	if err := cmd.Run(); err != nil {
+		t.Fatalf("ssh %s@%s:%d %q failed: %v\nstderr: %s",
+			tgt.User, tgt.Host, tgt.Port, command, err, stderr.String())
+	}
+	return stdout.String()
+}
+
+// RunSSHCombined runs command in the guest and returns combined stdout+stderr
+// regardless of exit status, for probes (ping, curl) where a non-zero exit is
+// an expected outcome.
+func RunSSHCombined(tgt SSHTarget, command string) (string, error) {
+	out, err := exec.Command("ssh", guestSSHArgs(tgt, 5, command)...).CombinedOutput()
+	return string(out), err
+}
+
+// SSHReadyBudget bounds the first SSH-handshake pass. Baremetal boots q35
+// guests slower (OVMF + NBD disks), so the harness widens it via
+// SPINIFEX_SSH_READY_TIMEOUT (Go duration); default 3m bounds shared runners.
+var SSHReadyBudget = durationEnv("SPINIFEX_SSH_READY_TIMEOUT", 3*time.Minute)
+
+// sshReprimeBudget bounds the second SSH pass after an ARP re-prime; short
+// because a helpful re-prime lands quickly. SPINIFEX_SSH_REPRIME_TIMEOUT.
+var sshReprimeBudget = durationEnv("SPINIFEX_SSH_REPRIME_TIMEOUT", 60*time.Second)
+
+// wanBridge is the host bridge an EIP is presented on, i.e. the L2 segment the
+// runner ARPs to reach a guest's public IP. Override via SPINIFEX_WAN_BRIDGE.
+var wanBridge = stringEnv("SPINIFEX_WAN_BRIDGE", "br-wan")
+
+func durationEnv(key string, def time.Duration) time.Duration {
+	if v := os.Getenv(key); v != "" {
+		if d, err := time.ParseDuration(v); err == nil && d > 0 {
+			return d
+		}
+	}
+	return def
+}
+
+func stringEnv(key, def string) string {
+	if v := strings.TrimSpace(os.Getenv(key)); v != "" {
+		return v
+	}
+	return def
+}
+
+// SSHHealth is a suite's sticky SSH-datapath verdict: the first handshake
+// timeout marks it broken and later SSH-dependent tests skip rather than
+// re-run the same multi-minute wait. Each suite keeps its own value.
+type SSHHealth struct {
+	broken atomic.Bool
+}
+
+// Require skips t if an earlier SSH probe in this suite has failed.
+func (h *SSHHealth) Require(t *testing.T) {
+	t.Helper()
+	if h.broken.Load() {
+		t.Skipf("skipping: earlier SSH probe failed; " +
+			"not retrying to keep suite time bounded")
+	}
+}
+
+// WaitReady waits for a full SSH handshake to host:port. TCP reachability is
+// not enough: sshd accepts while pam/cloud-init finish, and the first real
+// command then times out during banner exchange.
+func (h *SSHHealth) WaitReady(t *testing.T, host string, port int, keyPath string) {
+	t.Helper()
+	h.Require(t)
+	addr := net.JoinHostPort(host, strconv.Itoa(port))
+	Step(t, "waiting for SSH handshake %s", addr)
+	if h.TryReady(host, port, keyPath, SSHReadyBudget) {
+		return
+	}
+	// OVN emits no GARP on a same-chassis EIP rebind, so a decayed host neigh
+	// entry can blackhole host->guest SSH past the budget. Re-prime and retry
+	// once before declaring the datapath broken.
+	Step(t, "SSH handshake %s missed %s budget; re-priming ARP + retrying", addr, SSHReadyBudget)
+	reprimeSSHReachability(t, host)
+	if h.TryReady(host, port, keyPath, sshReprimeBudget) {
+		return
+	}
+	h.broken.Store(true)
+	t.Fatalf("Eventually: condition not met within %s (+%s after ARP re-prime): "+
+		"[SSH handshake %s never completed] "+
+		"(sticky-skip enabled for downstream)", SSHReadyBudget, sshReprimeBudget, addr)
+}
+
+// TryReady is WaitReady without the re-prime or t.Fatal: it reports whether
+// the handshake completed within budget and never marks the suite broken.
+func (h *SSHHealth) TryReady(host string, port int, keyPath string, budget time.Duration) bool {
+	if h.broken.Load() {
+		return false
+	}
+	tgt := SSHTarget{User: "ubuntu", Host: host, Port: port, KeyPath: keyPath}
+	deadline := time.Now().Add(budget)
+	for {
+		if exec.Command("ssh", guestSSHArgs(tgt, 3, "true")...).Run() == nil {
+			return true
+		}
+		if time.Now().After(deadline) {
+			return false
+		}
+		time.Sleep(2 * time.Second)
+	}
+}
+
+// reprimeSSHReachability best-effort flushes the host ARP entry for host on
+// the WAN bridge so the next pass re-resolves it. Needs ip(8) and passwordless
+// sudo; non-fatal so a missing tool never masks the real result.
+func reprimeSSHReachability(t *testing.T, host string) {
+	t.Helper()
+	if _, err := exec.LookPath("ip"); err != nil {
+		Step(t, "skip ARP re-prime: ip(8) unavailable (%v)", err)
+		return
+	}
+	if err := exec.Command("sudo", "-n", "true").Run(); err != nil {
+		Step(t, "skip ARP re-prime: passwordless sudo unavailable (%v)", err)
+		return
+	}
+	Step(t, "flushing host neigh for %s dev %s", host, wanBridge)
+	out, err := exec.Command("sudo", "-n", "ip", "neigh", "flush", "to", host, "dev", wanBridge).CombinedOutput()
+	if err != nil {
+		Step(t, "ARP re-prime flush failed (best-effort): %v: %s", err, strings.TrimSpace(string(out)))
+	}
+}
+
+// WaitForInstanceStateSoft is the cleanup-time analogue of
+// WaitForInstanceState: it polls and returns an error instead of t.Fatal.
+func WaitForInstanceStateSoft(c *AWSClient, id, target string, wait time.Duration) error {
+	deadline := time.Now().Add(wait)
+	for {
+		out, err := c.EC2.DescribeInstances(&ec2.DescribeInstancesInput{
+			InstanceIds: []*string{aws.String(id)},
+		})
+		if err == nil && len(out.Reservations) > 0 && len(out.Reservations[0].Instances) > 0 {
+			if aws.StringValue(out.Reservations[0].Instances[0].State.Name) == target {
+				return nil
+			}
+		}
+		if time.Now().After(deadline) {
+			return fmt.Errorf("instance %s did not reach %s within %s", id, target, wait)
+		}
+		time.Sleep(2 * time.Second)
+	}
+}
+
 // sshClient is the production SSH transport, backed by
 // golang.org/x/crypto/ssh. One client is opened per (user, key, node) tuple
 // on first use and cached for the harness lifetime.

tests/e2e/iam/datapath_helpers_test.go

@@ -5,205 +5,18 @@ package iam
 import (
 	"bufio"
 	"bytes"
-	"fmt"
-	"net"
 	"os"
-	"os/exec"
 	"path/filepath"
-	"strconv"
 	"strings"
-	"sync/atomic"
 	"testing"
-	"time"
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/ec2"
 	"github.com/mulgadc/spinifex/tests/e2e/harness"
 )
 
-// SSH / datapath helpers ported from the single suite so the IMDS and
-// instance-profile tests run self-contained inside iam. The single suite keeps
-// its own copy — other single tests still use these. They are separate test
-// binaries (separate processes), so the sticky-skip state below is per-suite.
-
-// sshDatapathBroken is set the first time an SSH probe in this suite times out.
-// Downstream SSH-dependent assertions call requireSSHHealthy(t) and skip rather
-// than re-running the same multi-minute timeout against the same broken datapath.
-var sshDatapathBroken atomic.Bool
-
-// sshReadyBudget bounds the first SSH-handshake pass. Widen via
-// SPINIFEX_SSH_READY_TIMEOUT (Go duration); default 3m bounds shared runners.
-var sshReadyBudget = resolveDurationEnv("SPINIFEX_SSH_READY_TIMEOUT", 3*time.Minute)
-
-// sshReprimeBudget bounds the second SSH pass after an ARP re-prime. Short by
-// design — if the re-prime helps, the handshake lands quickly. Override via
-// SPINIFEX_SSH_REPRIME_TIMEOUT.
-var sshReprimeBudget = resolveDurationEnv("SPINIFEX_SSH_REPRIME_TIMEOUT", 60*time.Second)
-
-// wanBridge is the host bridge an EIP is presented on — the L2 segment the
-// runner ARPs to reach a guest's public IP. Default br-wan; override for
-// non-standard pool fixtures via SPINIFEX_WAN_BRIDGE.
-var wanBridge = resolveStringEnv("SPINIFEX_WAN_BRIDGE", "br-wan")
-
-func resolveDurationEnv(key string, def time.Duration) time.Duration {
-	if v := os.Getenv(key); v != "" {
-		if d, err := time.ParseDuration(v); err == nil && d > 0 {
-			return d
-		}
-	}
-	return def
-}
-
-func resolveStringEnv(key, def string) string {
-	if v := strings.TrimSpace(os.Getenv(key)); v != "" {
-		return v
-	}
-	return def
-}
-
-// requireSSHHealthy skips the calling test if a prior SSH probe in this suite
-// has already failed, cutting the cascade cost of a broken datapath.
-func requireSSHHealthy(t *testing.T) {
-	t.Helper()
-	if sshDatapathBroken.Load() {
-		t.Skipf("skipping: earlier SSH probe failed; " +
-			"not retrying to keep suite time bounded")
-	}
-}
-
-// waitForSSHReady probes a full SSH handshake (BatchMode + ConnectTimeout)
-// against host:port, retrying until the daemon completes banner exchange.
-// TCP-port reachability alone is insufficient — sshd accepts the connect
-// while pam/cloud-init are still finishing, and the first real runSSH then
-// hits "Connection timed out during banner exchange".
-func waitForSSHReady(t *testing.T, host string, port int, keyPath string) {
-	t.Helper()
-	requireSSHHealthy(t)
-	addr := net.JoinHostPort(host, strconv.Itoa(port))
-	harness.Step(t, "waiting for SSH handshake %s", addr)
-	if trySSHReady(host, port, keyPath, sshReadyBudget) {
-		return
-	}
-	// First budget elapsed. OVN emits no GARP on a same-chassis EIP rebind, so a
-	// decayed host neigh entry for the EIP can blackhole host->guest SSH until it
-	// ages out — past our budget. Re-prime the ARP entry and retry once before
-	// declaring the datapath broken, riding out the transient learning gap.
-	harness.Step(t, "SSH handshake %s missed %s budget; re-priming ARP + retrying", addr, sshReadyBudget)
-	reprimeSSHReachability(t, host)
-	if trySSHReady(host, port, keyPath, sshReprimeBudget) {
-		return
-	}
-	sshDatapathBroken.Store(true)
-	t.Fatalf("Eventually: condition not met within %s (+%s after ARP re-prime): "+
-		"[SSH handshake %s never completed] "+
-		"(sticky-skip enabled for downstream)", sshReadyBudget, sshReprimeBudget, addr)
-}
-
-// reprimeSSHReachability best-effort flushes the host ARP entry for the EIP on
-// the WAN bridge, forcing a fresh broadcast re-resolution before the next SSH
-// pass. Flush (not replace) is mode-agnostic: it needs no MAC and can't install
-// a wrong centralised-mode entry. Gated on ip + passwordless sudo like the IMDS
-// datapath dump; non-fatal so a missing tool never masks the real result.
-func reprimeSSHReachability(t *testing.T, host string) {
-	t.Helper()
-	if _, err := exec.LookPath("ip"); err != nil {
-		harness.Step(t, "skip ARP re-prime: ip(8) unavailable (%v)", err)
-		return
-	}
-	if err := exec.Command("sudo", "-n", "true").Run(); err != nil {
-		harness.Step(t, "skip ARP re-prime: passwordless sudo unavailable (%v)", err)
-		return
-	}
-	harness.Step(t, "flushing host neigh for %s dev %s", host, wanBridge)
-	out, err := exec.Command("sudo", "-n", "ip", "neigh", "flush", "to", host, "dev", wanBridge).CombinedOutput()
-	if err != nil {
-		harness.Step(t, "ARP re-prime flush failed (best-effort): %v: %s", err, strings.TrimSpace(string(out)))
-	}
-}
-
-// waitForSSHHandshake is an alias kept for the imds_test.go call site.
-func waitForSSHHandshake(t *testing.T, host string, port int, keyPath string) {
-	waitForSSHReady(t, host, port, keyPath)
-}
-
-// trySSHReady mirrors waitForSSHReady but returns reachability as a bool
-// instead of t.Fatal-ing on timeout.
-func trySSHReady(host string, port int, keyPath string, timeout time.Duration) bool {
-	if sshDatapathBroken.Load() {
-		return false
-	}
-	deadline := time.Now().Add(timeout)
-	for {
-		cmd := exec.Command("ssh",
-			"-o", "StrictHostKeyChecking=no",
-			"-o", "UserKnownHostsFile=/dev/null",
-			"-o", "LogLevel=ERROR",
-			"-o", "ConnectTimeout=3",
-			"-o", "BatchMode=yes",
-			"-p", strconv.Itoa(port),
-			"-i", keyPath,
-			"ubuntu@"+host,
-			"true")
-		if cmd.Run() == nil {
-			return true
-		}
-		if time.Now().After(deadline) {
-			return false
-		}
-		time.Sleep(2 * time.Second)
-	}
-}
-
-// runSSH is a thin wrapper around `ssh` matching the option set used by
-// harness.LsblkRootGiB. Returns stdout. t.Fatal on non-zero exit so callers
-// can chain assertions on the output without nil-checking err.
-func runSSH(t *testing.T, tgt harness.SSHTarget, command string) string {
-	t.Helper()
-	args := []string{
-		"-o", "StrictHostKeyChecking=no",
-		"-o", "UserKnownHostsFile=/dev/null",
-		"-o", "LogLevel=ERROR",
-		"-o", "ConnectTimeout=5",
-		"-o", "BatchMode=yes",
-		"-p", strconv.Itoa(tgt.Port),
-		"-i", tgt.KeyPath,
-		tgt.User + "@" + tgt.Host,
-		command,
-	}
-	var stdout, stderr bytes.Buffer
-	cmd := exec.Command("ssh", args...)
-	cmd.Stdout = &stdout
-	cmd.Stderr = &stderr
-	if err := cmd.Run(); err != nil {
-		t.Fatalf("ssh %s@%s:%d %q failed: %v\nstderr: %s",
-			tgt.User, tgt.Host, tgt.Port, command, err, stderr.String())
-	}
-	return stdout.String()
-}
-
-// runSSHCombined runs `command` over SSH against tgt and returns combined
-// stdout+stderr regardless of exit status. Unlike runSSH it does not t.Fatal
-// on non-zero exit — needed for probes (e.g. curl status codes) where a
-// non-zero exit is an expected outcome.
-func runSSHCombined(tgt harness.SSHTarget, command string) (string, error) {
-	args := []string{
-		"-o", "StrictHostKeyChecking=no",
-		"-o", "UserKnownHostsFile=/dev/null",
-		"-o", "LogLevel=ERROR",
-		"-o", "ConnectTimeout=5",
-		"-o", "BatchMode=yes",
-		"-p", strconv.Itoa(tgt.Port),
-		"-i", tgt.KeyPath,
-		tgt.User + "@" + tgt.Host,
-		command,
-	}
-	var combined bytes.Buffer
-	cmd := exec.Command("ssh", args...)
-	cmd.Stdout = &combined
-	cmd.Stderr = &combined
-	err := cmd.Run()
-	return combined.String(), err
-}
+// sshHealth is this suite's sticky SSH-datapath verdict; see harness.SSHHealth.
+var sshHealth harness.SSHHealth
 
 // primaryENI returns the NetworkInterfaceId of an instance's first ENI.
 // t.Fatal if the instance has no ENI — every running EC2 instance must.
@@ -219,26 +32,6 @@ func primaryENI(t *testing.T, inst *ec2.Instance) string {
 	return eni
 }
 
-// waitForInstanceStateSoft is the cleanup-time analogue of
-// harness.WaitForInstanceState — no t.Fatal, just polls and returns.
-func waitForInstanceStateSoft(c *harness.AWSClient, id, target string, timeout time.Duration) error {
-	deadline := time.Now().Add(timeout)
-	for {
-		out, err := c.EC2.DescribeInstances(&ec2.DescribeInstancesInput{
-			InstanceIds: []*string{aws.String(id)},
-		})
-		if err == nil && len(out.Reservations) > 0 && len(out.Reservations[0].Instances) > 0 {
-			if aws.StringValue(out.Reservations[0].Instances[0].State.Name) == target {
-				return nil
-			}
-		}
-		if time.Now().After(deadline) {
-			return fmt.Errorf("instance %s did not reach %s within %s", id, target, timeout)
-		}
-		time.Sleep(2 * time.Second)
-	}
-}
-
 // detectPoolMode reads external_mode from spinifex.toml. Defaults to false
 // (dev_networking) which is the single-node CI fixture. Any non-empty
 // external_mode value ("pool" / "nat") means external IPAM is in play, which

tests/e2e/iam/imds_test.go

@@ -104,7 +104,7 @@ type imdsCredDoc struct {
 func runIMDS(t *testing.T, fix *Fixture) {
 	harness.Phase(t, "Single — IMDSv2 Host-Served Instance Metadata")
 	harness.SkipIfNoOVN(t)
-	requireSSHHealthy(t)
+	sshHealth.Require(t)
 
 	adminAccount := harness.IAMAccountID(t, fix.AWS)
 	keyName, keyPath := needKeyPair(t, fix)
@@ -228,7 +228,7 @@ func runIMDS(t *testing.T, fix *Fixture) {
 			"public-hostname must be the AWS-shaped EC2 name for public-ipv4")
 
 		harness.Step(t, "resolve IMDS public-hostname %s from the guest", gotHost)
-		resolved := runSSH(t, tgtX, "getent ahostsv4 "+gotHost)
+		resolved := harness.RunSSH(t, tgtX, "getent ahostsv4 "+gotHost)
 		require.Containsf(t, strings.Fields(resolved), pubIP,
 			"public-hostname %s did not resolve to public-ipv4 %s: %s", gotHost, pubIP, resolved)
 	} else {
@@ -420,7 +420,7 @@ func imdsProbe(t *testing.T, fix *Fixture, keyPath string, spec imdsVMSpec) (str
 		_, _ = fix.AWS.EC2.TerminateInstances(&ec2.TerminateInstancesInput{
 			InstanceIds: []*string{aws.String(id)},
 		})
-		_ = waitForInstanceStateSoft(fix.AWS, id, "terminated", 5*time.Minute)
+		_ = harness.WaitForInstanceStateSoft(fix.AWS, id, "terminated", 5*time.Minute)
 	})
 
 	inst := harness.WaitForInstanceState(t, fix.AWS, id, "running")
@@ -430,7 +430,7 @@ func imdsProbe(t *testing.T, fix *Fixture, keyPath string, spec imdsVMSpec) (str
 
 	host, port := harness.InstancePublicSSHHost(t, inst)
 	harness.Step(t, "wait for %s SSH at %s:%d", id, host, port)
-	waitForSSHHandshake(t, host, port, keyPath)
+	sshHealth.WaitReady(t, host, port, keyPath)
 	return id, priv, eni, harness.SSHTarget{User: "ubuntu", Host: host, Port: port, KeyPath: keyPath}
 }
 
@@ -540,34 +540,34 @@ func imdsAssertBootFromIMDS(t *testing.T, tgt harness.SSHTarget, privIP string)
 	t.Helper()
 	harness.Step(t, "boot-from-IMDS: cloud-init selected the aws (Ec2) datasource")
 	ds := strings.ToLower(strings.TrimSpace(
-		runSSH(t, tgt, "cloud-id 2>/dev/null || cloud-init query --format '{{datasource}}'")))
+		harness.RunSSH(t, tgt, "cloud-id 2>/dev/null || cloud-init query --format '{{datasource}}'")))
 	require.Truef(t, strings.Contains(ds, "aws") || strings.Contains(ds, "ec2"),
 		"cloud-init must report the aws/Ec2 datasource (got %q) — not the retired NoCloud seed", ds)
 
 	harness.Step(t, "boot-from-IMDS: no NoCloud cidata seed device attached")
-	labels := strings.ToLower(runSSH(t, tgt, "lsblk -no LABEL"))
+	labels := strings.ToLower(harness.RunSSH(t, tgt, "lsblk -no LABEL"))
 	require.NotContainsf(t, labels, "cidata",
 		"no cidata-labelled device may be attached — the seed ISO is retired (lsblk LABELs: %q)", labels)
 
 	harness.Step(t, "boot-from-IMDS: login is the AMI stock default user %q", tgt.User)
-	require.Equal(t, tgt.User, strings.TrimSpace(runSSH(t, tgt, "id -un")),
+	require.Equal(t, tgt.User, strings.TrimSpace(harness.RunSSH(t, tgt, "id -un")),
 		"in-guest login must be the AMI stock default user, not a Spinifex-forced account")
 
 	harness.Step(t, "boot-from-IMDS: AWS-form hostname ip-<dashed-ip>")
 	wantHost := "ip-" + strings.ReplaceAll(privIP, ".", "-")
-	gotHost := strings.TrimSpace(runSSH(t, tgt, "hostname"))
+	gotHost := strings.TrimSpace(harness.RunSSH(t, tgt, "hostname"))
 	require.Truef(t, strings.HasPrefix(gotHost, wantHost),
 		"hostname must be the AWS form %q rendered from IMDS local-hostname (got %q)", wantHost, gotHost)
 
 	harness.Step(t, "boot-from-IMDS: primary NIC up with the VPC IP %s", privIP)
-	addrs := runSSH(t, tgt, "ip -4 -o addr show scope global")
+	addrs := harness.RunSSH(t, tgt, "ip -4 -o addr show scope global")
 	require.Containsf(t, addrs, privIP,
 		"the Ec2 datasource must render the primary NIC with the VPC IP %s from IMDS (ip addr: %q)", privIP, addrs)
 
 	harness.Step(t, "boot-from-IMDS: user-data runcmd executed")
 	deadline := time.Now().Add(90 * time.Second)
 	for {
-		out, _ := runSSHCombined(tgt, "cat "+imdsUDDoneFile+" 2>/dev/null")
+		out, _ := harness.RunSSHCombined(tgt, "cat "+imdsUDDoneFile+" 2>/dev/null")
 		if strings.Contains(out, imdsUDMarker) {
 			break
 		}
@@ -777,7 +777,7 @@ func imdsAwaitToken(t *testing.T, fix *Fixture, tgt harness.SSHTarget, subnetID,
 	for {
 		cmd := fmt.Sprintf(
 			`curl -sf -X PUT "%s/latest/api/token" -H "X-aws-ec2-metadata-token-ttl-seconds: 120"`, metaURL)
-		out, err := runSSHCombined(tgt, cmd)
+		out, err := harness.RunSSHCombined(tgt, cmd)
 		switch {
 		case err != nil:
 			lastErr = fmt.Errorf("token PUT failed: %w (out=%q)", err, out)
@@ -797,11 +797,11 @@ func imdsAwaitToken(t *testing.T, fix *Fixture, tgt harness.SSHTarget, subnetID,
 
 // imdsGet fetches a token-gated path from inside the guest and returns the
 // trimmed body. -sf turns any HTTP >= 400 into a non-zero exit, so a 4xx/5xx
-// fails the test loudly with the server status visible via runSSH's stderr.
+// fails the test loudly with the server status visible via harness.RunSSH's stderr.
 func imdsGet(t *testing.T, tgt harness.SSHTarget, token, path string) string {
 	t.Helper()
 	cmd := fmt.Sprintf(`curl -sf -H "X-aws-ec2-metadata-token: %s" %s%s`, token, metaURL, path)
-	return strings.TrimSpace(runSSH(t, tgt, cmd))
+	return strings.TrimSpace(harness.RunSSH(t, tgt, cmd))
 }
 
 // imdsExpectedPubKey derives the OpenSSH public key (type + base64) the IMDS
@@ -826,7 +826,7 @@ func imdsExpectedPubKey(t *testing.T, pemPath string) (keyType, base64Key string
 // argument (empty for the tokenless probe).
 func imdsCode(tgt harness.SSHTarget, headerArg, path string) string {
 	cmd := fmt.Sprintf(`curl -s -o /dev/null -w '%%{http_code}' %s %s%s`, headerArg, metaURL, path)
-	out, _ := runSSHCombined(tgt, cmd)
+	out, _ := harness.RunSSHCombined(tgt, cmd)
 	return strings.TrimSpace(out)
 }
 

tests/e2e/single/createimage_test.go

@@ -145,10 +145,10 @@ func runBootVerifyAMI(t *testing.T, fix *Fixture, amiID, label string) {
 	probeInst := harness.WaitForInstanceState(t, fix.AWS, probeInstanceID, "running")
 
 	host, port := harness.InstancePublicSSHHost(t, probeInst)
-	waitForSSHReady(t, host, port, keyPath)
+	sshHealth.WaitReady(t, host, port, keyPath)
 
 	tgt := harness.SSHTarget{User: "ubuntu", Host: host, Port: port, KeyPath: keyPath}
-	idOut := runSSH(t, tgt, "id")
+	idOut := harness.RunSSH(t, tgt, "id")
 	assert.Containsf(t, idOut, "ubuntu", "boot-verify %s: ssh id should report ubuntu\n%s", label, idOut)
 
 	harness.Detail(t, "boot_verify_"+label, "ok")

tests/e2e/single/datapath_test.go

@@ -105,7 +105,7 @@ func sgDatapathRevokeRounds() int {
 func runSGPolicyDatapath(t *testing.T, fix *Fixture) {
 	harness.Phase(t, "Single — Security Group Policy on a Real Datapath (OVN)")
 	harness.SkipIfNoOVN(t)
-	requireSSHHealthy(t)
+	sshHealth.Require(t)
 
 	// Bootstrap every prereq up front. runSGEInstance / primaryENI use
 	// `fix.AMIID / InstanceType / KeyName / KeyPath` indirectly; resolve
@@ -257,7 +257,7 @@ func runSGPolicyDatapath(t *testing.T, fix *Fixture) {
 		// Non-fatal probe so a timeout dumps the guest console + OVN/datapath
 		// state before Fatal. A full 2min unreachable window on a fresh public-IP
 		// VM is the flake signature; capture it from CI artifacts alone.
-		if !trySSHReady(clientHost, clientPort, keyPath, 2*time.Minute) {
+		if !sshHealth.TryReady(clientHost, clientPort, keyPath, 2*time.Minute) {
 			harness.DumpVPCFlowDiagnostics(t, fix.AWS, clientID,
 				fmt.Sprintf("8e-3 client-vm SSH timeout — vpc=%s sg=%s pub=%s", def.VPCID, clientSG, clientHost),
 				harness.VPCDiagnosticsOpts{
@@ -326,7 +326,7 @@ func runSGPolicyDatapath(t *testing.T, fix *Fixture) {
 			// so the only variable left is ovn-controller installing the
 			// chassis flows. Bash used 30 attempts at 2s; keep that budget.
 			harness.EventuallyErr(t, func() error {
-				out, err := runSSHCombined(clientTgt, curlCmd)
+				out, err := harness.RunSSHCombined(clientTgt, curlCmd)
 				if err != nil {
 					return fmt.Errorf("client -> target:8080 failed: %w (out=%q)", err, out)
 				}
@@ -343,7 +343,7 @@ func runSGPolicyDatapath(t *testing.T, fix *Fixture) {
 	// merits even if the 8080 path above failed.
 	t.Run("DeniedTraffic", func(t *testing.T) {
 		harness.Step(t, "8e-5 denied traffic client -> target:22 (no SSH ingress)")
-		if _, err := runSSHCombined(clientTgt, fmt.Sprintf("nc -z -w 5 %s 22", targetPriv)); err == nil {
+		if _, err := harness.RunSSHCombined(clientTgt, fmt.Sprintf("nc -z -w 5 %s 22", targetPriv)); err == nil {
 			t.Fatalf("FAIL: client reached target:22 — default-deny ACL not enforced")
 		}
 		harness.Detail(t, "step5", "denied_traffic_ok")
@@ -449,7 +449,7 @@ func runSGPolicyDatapath(t *testing.T, fix *Fixture) {
 				derived, clientLL, iface)
 
 			targetLL := linkLocalFromMAC(eniMAC(t, fix, targetENI)).String()
-			out, err := runSSHCombined(clientTgt, fmt.Sprintf("ping -6 -c 3 -W 3 %s%%%s", targetLL, iface))
+			out, err := harness.RunSSHCombined(clientTgt, fmt.Sprintf("ping -6 -c 3 -W 3 %s%%%s", targetLL, iface))
 			require.Errorf(t, err, "FAIL: client reached target over IPv6 at %s\n%s", targetLL, out)
 
 			// A non-zero exit alone also covers a broken ssh session, a missing
@@ -498,7 +498,7 @@ func runSGPolicyDatapath(t *testing.T, fix *Fixture) {
 				// Fresh TCP connection — conntrack does not affect new
 				// connections. Bash treats a single curl success as failure.
 				harness.Step(t, "8e-6 verify client -> target:8080 now blocked")
-				if _, err := runSSHCombined(clientTgt, curlCmd); err == nil {
+				if _, err := harness.RunSSHCombined(clientTgt, curlCmd); err == nil {
 					t.Fatalf("FAIL: client still reached target:8080 after revoke — propagation not immediate")
 				}
 				harness.Detail(t, "step6", "revoke_blocked_ok")
@@ -529,7 +529,7 @@ func runSGPolicyDatapath(t *testing.T, fix *Fixture) {
 				require.NoError(t, err, "re-authorize tcp/8080 from %s", clientSG)
 
 				harness.EventuallyErr(t, func() error {
-					out, err := runSSHCombined(clientTgt, curlCmd)
+					out, err := harness.RunSSHCombined(clientTgt, curlCmd)
 					if err != nil {
 						return fmt.Errorf("client -> target:8080 still blocked after re-add: %w (out=%q)", err, out)
 					}
@@ -706,12 +706,12 @@ func linkLocalFromMAC(mac net.HardwareAddr) net.IP {
 // link-local address configured on it.
 func guestLinkLocal(t *testing.T, tgt harness.SSHTarget) (string, string) {
 	t.Helper()
-	iface, err := runSSHCombined(tgt, "ip -o -4 route show default | awk '{print $5; exit}'")
+	iface, err := harness.RunSSHCombined(tgt, "ip -o -4 route show default | awk '{print $5; exit}'")
 	require.NoError(t, err, "guest could not report its default-route interface")
 	iface = strings.TrimSpace(iface)
 	require.NotEmpty(t, iface, "guest has no default-route interface")
 
-	ll, err := runSSHCombined(tgt,
+	ll, err := harness.RunSSHCombined(tgt,
 		fmt.Sprintf("ip -6 -o addr show dev %s scope link | awk '{print $4; exit}' | cut -d/ -f1", iface))
 	require.NoError(t, err, "guest could not report its link-local address")
 	ll = strings.TrimSpace(ll)

tests/e2e/single/guestchurn_test.go

@@ -120,13 +120,13 @@ func runGuestChurnDurability(t *testing.T, fix *Fixture) {
 
 		addr := net.JoinHostPort(host, strconv.Itoa(port))
 		harness.Step(t, "waiting for SSH handshake %s", addr)
-		waitForSSHHandshake(t, host, port, keyPath)
+		sshHealth.WaitReady(t, host, port, keyPath)
 		_ = addr
 
 		tgt := harness.SSHTarget{User: "ubuntu", Host: host, Port: port, KeyPath: keyPath}
 
 		harness.Step(t, "ssh id")
-		idOut := runSSH(t, tgt, "id")
+		idOut := harness.RunSSH(t, tgt, "id")
 		assert.Containsf(t, idOut, "ubuntu", "ssh id should report ubuntu\n%s", idOut)
 
 		harness.Step(t, "lsblk root-volume cross-check vs API")
@@ -151,7 +151,7 @@ func runGuestChurnDurability(t *testing.T, fix *Fixture) {
 		harness.Detail(t, "guest_gib", guestGiB, "api_gib", apiGiB)
 
 		harness.Step(t, "ssh hostname")
-		hn := strings.TrimSpace(runSSH(t, tgt, "hostname"))
+		hn := strings.TrimSpace(harness.RunSSH(t, tgt, "hostname"))
 		// Bash uses `spinifex-vm-<first 8 hex chars of instance ID>` and treats
 		// a missing prefix as a non-fatal warning. Replicate the soft check via
 		// t.Logf rather than asserting — the spx hostname format is not part of
@@ -251,7 +251,7 @@ func runGuestChurnRound(t *testing.T, fix *Fixture, instanceID, origType, keyPat
 
 	t.Run("HotplugENI", func(t *testing.T) {
 		harness.SkipIfNoOVN(t)
-		requireSSHHealthy(t)
+		sshHealth.Require(t)
 
 		tgt := resolveGuestSSHTarget(t, fix, instanceID, keyPath)
 		def := harness.EnsureDefaultVPC(t, fix.Harness)
@@ -475,11 +475,11 @@ func runGuestChurnRound(t *testing.T, fix *Fixture, instanceID, origType, keyPat
 		host, port := harness.InstancePublicSSHHost(t, runInst)
 		harness.Detail(t, "ssh_host", host, "ssh_port", port)
 
-		waitForSSHReady(t, host, port, keyPath)
+		sshHealth.WaitReady(t, host, port, keyPath)
 		tgt := harness.SSHTarget{User: "ubuntu", Host: host, Port: port, KeyPath: keyPath}
 
 		harness.Step(t, "ssh nproc")
-		nprocOut := strings.TrimSpace(runSSH(t, tgt, "nproc"))
+		nprocOut := strings.TrimSpace(harness.RunSSH(t, tgt, "nproc"))
 		vmVCPUs, err := strconv.ParseInt(nprocOut, 10, 64)
 		require.NoErrorf(t, err, "parse nproc output %q", nprocOut)
 		require.Equalf(t, expectedVCPUs, vmVCPUs,
@@ -487,7 +487,7 @@ func runGuestChurnRound(t *testing.T, fix *Fixture, instanceID, origType, keyPat
 		harness.Detail(t, "vm_vcpus", vmVCPUs)
 
 		harness.Step(t, "ssh MemTotal")
-		memOut := strings.TrimSpace(runSSH(t, tgt, "awk '/MemTotal/ {print $2}' /proc/meminfo"))
+		memOut := strings.TrimSpace(harness.RunSSH(t, tgt, "awk '/MemTotal/ {print $2}' /proc/meminfo"))
 		vmMemKB, err := strconv.ParseInt(memOut, 10, 64)
 		require.NoErrorf(t, err, "parse MemTotal output %q", memOut)
 		vmMemMiB := vmMemKB / 1024
@@ -522,7 +522,7 @@ func runGuestChurnRound(t *testing.T, fix *Fixture, instanceID, origType, keyPat
 		runInst := harness.WaitForInstanceState(t, fix.AWS, instanceID, "running")
 
 		host, port := harness.InstancePublicSSHHost(t, runInst)
-		waitForSSHReady(t, host, port, keyPath)
+		sshHealth.WaitReady(t, host, port, keyPath)
 		tgt := harness.SSHTarget{User: "ubuntu", Host: host, Port: port, KeyPath: keyPath}
 		gotSha := harness.GuestReadSentinelSha(t, tgt, "/dev/disk/by-label/"+guestChurnSentinelLabel, guestChurnSentinelLabel)
 		require.Equalf(t, wantSha, gotSha, "sha256 mismatch after stop/start")
@@ -532,12 +532,12 @@ func runGuestChurnRound(t *testing.T, fix *Fixture, instanceID, origType, keyPat
 	t.Run("Reboot", func(t *testing.T) {
 		inst := describeSingletonInstance(t, fix, instanceID)
 		host, port := harness.InstancePublicSSHHost(t, inst)
-		waitForSSHReady(t, host, port, keyPath)
+		sshHealth.WaitReady(t, host, port, keyPath)
 
 		// The boot ID is what proves a reboot happened. Uptime cannot on its
 		// own: a reboot is asynchronous, so the guest answering shortly after
 		// the call may still be the one that has not shut down yet.
-		preBootID := strings.TrimSpace(runSSH(t,
+		preBootID := strings.TrimSpace(harness.RunSSH(t,
 			harness.SSHTarget{User: "ubuntu", Host: host, Port: port, KeyPath: keyPath},
 			"cat /proc/sys/kernel/random/boot_id"))
 		require.NotEmpty(t, preBootID, "guest returned an empty boot ID before the reboot")
@@ -581,7 +581,7 @@ func runGuestChurnRound(t *testing.T, fix *Fixture, instanceID, origType, keyPat
 		require.NotEmpty(t, descPost.Reservations[0].Instances, "no instances post-reboot")
 		postInst := descPost.Reservations[0].Instances[0]
 		host, port = harness.InstancePublicSSHHost(t, postInst)
-		waitForSSHReady(t, host, port, keyPath)
+		sshHealth.WaitReady(t, host, port, keyPath)
 
 		tgt := harness.SSHTarget{User: "ubuntu", Host: host, Port: port, KeyPath: keyPath}
 		waitForNewBootID(t, tgt, preBootID)
@@ -589,7 +589,7 @@ func runGuestChurnRound(t *testing.T, fix *Fixture, instanceID, origType, keyPat
 		// Now that the guest is known to be on a new boot, uptime is a second
 		// reading of the same fact rather than a guess at it.
 		harness.Step(t, "ssh uptime")
-		uptimeOut := strings.TrimSpace(runSSH(t, tgt, "cat /proc/uptime | cut -d. -f1"))
+		uptimeOut := strings.TrimSpace(harness.RunSSH(t, tgt, "cat /proc/uptime | cut -d. -f1"))
 		uptimeSecs, err := strconv.ParseInt(uptimeOut, 10, 64)
 		require.NoErrorf(t, err, "parse uptime output %q", uptimeOut)
 		require.LessOrEqualf(t, uptimeSecs, int64(120),
@@ -667,7 +667,7 @@ func resolveGuestSSHTarget(t *testing.T, fix *Fixture, instanceID, keyPath strin
 	t.Helper()
 	inst := describeSingletonInstance(t, fix, instanceID)
 	host, port := harness.InstancePublicSSHHost(t, inst)
-	waitForSSHReady(t, host, port, keyPath)
+	sshHealth.WaitReady(t, host, port, keyPath)
 	return harness.SSHTarget{User: "ubuntu", Host: host, Port: port, KeyPath: keyPath}
 }
 

tests/e2e/single/instance_test.go

@@ -3,9 +3,6 @@
 package single
 
 import (
-	"bytes"
-	"os/exec"
-	"strconv"
 	"testing"
 
 	"github.com/aws/aws-sdk-go/aws"
@@ -85,30 +82,3 @@ func runConsoleOutput(t *testing.T, fix *Fixture) {
 		"has_output", out.Output != nil && aws.StringValue(out.Output) != "",
 	)
 }
-
-// runSSH is a thin wrapper around `ssh` matching the option set used by
-// harness.LsblkRootGiB. Returns stdout. t.Fatal on non-zero exit so callers
-// can chain assertions on the output without nil-checking err.
-func runSSH(t *testing.T, tgt harness.SSHTarget, command string) string {
-	t.Helper()
-	args := []string{
-		"-o", "StrictHostKeyChecking=no",
-		"-o", "UserKnownHostsFile=/dev/null",
-		"-o", "LogLevel=ERROR",
-		"-o", "ConnectTimeout=5",
-		"-o", "BatchMode=yes",
-		"-p", strconv.Itoa(tgt.Port),
-		"-i", tgt.KeyPath,
-		tgt.User + "@" + tgt.Host,
-		command,
-	}
-	var stdout, stderr bytes.Buffer
-	cmd := exec.Command("ssh", args...)
-	cmd.Stdout = &stdout
-	cmd.Stderr = &stderr
-	if err := cmd.Run(); err != nil {
-		t.Fatalf("ssh %s@%s:%d %q failed: %v\nstderr: %s",
-			tgt.User, tgt.Host, tgt.Port, command, err, stderr.String())
-	}
-	return stdout.String()
-}

tests/e2e/single/launchtemplate_test.go

@@ -134,7 +134,7 @@ func runLaunchTemplateBoot(t *testing.T, fix *Fixture) {
 
 	host, port := harness.InstancePublicSSHHost(t, inst)
 	harness.Detail(t, "instance_id", instanceID, "ssh_host", host, "ssh_port", port)
-	waitForSSHReady(t, host, port, keyPath)
+	sshHealth.WaitReady(t, host, port, keyPath)
 	tgt := harness.SSHTarget{User: "ubuntu", Host: host, Port: port, KeyPath: keyPath}
 
 	harness.Step(t, "guest: template user-data ran")

tests/e2e/single/lifecycle_test.go

@@ -3,13 +3,7 @@
 package single
 
 import (
-	"net"
-	"os"
-	"os/exec"
-	"strconv"
-	"sync/atomic"
 	"testing"
-	"time"
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/ec2"
@@ -17,36 +11,8 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
-// sshDatapathBroken is set the first time a single-node SSH probe times
-// out. Downstream SSH-dependent tests call requireSSHHealthy(t) and skip
-// rather than re-running the same 3-minute timeout against the same
-// broken datapath.
-var sshDatapathBroken atomic.Bool
-
-// sshReadyBudget bounds the single-node SSH-handshake probes. Baremetal boots
-// q35 guests slower (OVMF + NBD disks), so the harness widens it via
-// SPINIFEX_SSH_READY_TIMEOUT (Go duration); default 3m bounds shared runners.
-var sshReadyBudget = resolveSSHReadyBudget(3 * time.Minute)
-
-func resolveSSHReadyBudget(def time.Duration) time.Duration {
-	if v := os.Getenv("SPINIFEX_SSH_READY_TIMEOUT"); v != "" {
-		if d, err := time.ParseDuration(v); err == nil && d > 0 {
-			return d
-		}
-	}
-	return def
-}
-
-// requireSSHHealthy skips the calling test if a prior SSH probe on this
-// single-node VM has already failed. Cuts the cascade cost of a broken
-// datapath from 4×3min down to one fail + four skips.
-func requireSSHHealthy(t *testing.T) {
-	t.Helper()
-	if sshDatapathBroken.Load() {
-		t.Skipf("skipping: earlier single-node SSH probe failed; " +
-			"not retrying to keep suite time bounded")
-	}
-}
+// sshHealth is this suite's sticky SSH-datapath verdict; see harness.SSHHealth.
+var sshHealth harness.SSHHealth
 
 // runStopStart stops fix.Instance and waits for it to reach the
 // "stopped" state, then asserts that rebooting a stopped instance is
@@ -157,58 +123,3 @@ func runAttachToStoppedError(t *testing.T, fix *Fixture) {
 	require.NoError(t, err, "delete-volume %s", stoppedVolID)
 	cleaned = true
 }
-
-// waitForSSHReady probes a full SSH handshake (BatchMode + ConnectTimeout)
-// against host:port, retrying until the daemon completes banner exchange.
-// TCP-port reachability alone is insufficient — sshd accepts the connect
-// while pam/cloud-init are still finishing, and the first real runSSH then
-// hits "Connection timed out during banner exchange" (CI run 26034322018).
-func waitForSSHReady(t *testing.T, host string, port int, keyPath string) {
-	t.Helper()
-	requireSSHHealthy(t)
-	addr := net.JoinHostPort(host, strconv.Itoa(port))
-	harness.Step(t, "waiting for SSH handshake %s", addr)
-	if !trySSHReady(host, port, keyPath, sshReadyBudget) {
-		sshDatapathBroken.Store(true)
-		t.Fatalf("Eventually: condition not met within %s: "+
-			"[SSH handshake %s never completed] "+
-			"(sticky-skip enabled for downstream)", sshReadyBudget, addr)
-	}
-}
-
-// waitForSSHHandshake is an alias kept for guestchurn_test.go's call site.
-func waitForSSHHandshake(t *testing.T, host string, port int, keyPath string) {
-	waitForSSHReady(t, host, port, keyPath)
-}
-
-// trySSHReady mirrors waitForSSHReady but returns reachability as a bool
-// instead of t.Fatal-ing on timeout. Matches run-e2e.sh ~2850–2860, which
-// treats SSH-via-public-IP timeout as a WARN ("bridge not ready") rather
-// than a hard test failure — new-subnet IGW SNAT/DNAT
-// flow installation is timing-sensitive on shared CI runners and not part
-// of the EC2 surface contract this test exists to validate.
-func trySSHReady(host string, port int, keyPath string, timeout time.Duration) bool {
-	if sshDatapathBroken.Load() {
-		return false
-	}
-	deadline := time.Now().Add(timeout)
-	for {
-		cmd := exec.Command("ssh",
-			"-o", "StrictHostKeyChecking=no",
-			"-o", "UserKnownHostsFile=/dev/null",
-			"-o", "LogLevel=ERROR",
-			"-o", "ConnectTimeout=3",
-			"-o", "BatchMode=yes",
-			"-p", strconv.Itoa(port),
-			"-i", keyPath,
-			"ubuntu@"+host,
-			"true")
-		if cmd.Run() == nil {
-			return true
-		}
-		if time.Now().After(deadline) {
-			return false
-		}
-		time.Sleep(2 * time.Second)
-	}
-}

tests/e2e/single/rebootliveness_test.go

@@ -34,10 +34,10 @@ func runEarlyRebootLiveness(t *testing.T, fix *Fixture) {
 	instanceID := launchBaselineInstance(t, fix, amiID, instType, keyName, vpc.SubnetID, []string{vpc.SGID})
 	inst := describeSingletonInstance(t, fix, instanceID)
 	host, port := harness.InstancePublicSSHHost(t, inst)
-	waitForSSHReady(t, host, port, keyPath)
+	sshHealth.WaitReady(t, host, port, keyPath)
 	tgt := harness.SSHTarget{User: "ubuntu", Host: host, Port: port, KeyPath: keyPath}
 
-	bootID := strings.TrimSpace(runSSH(t, tgt, "cat /proc/sys/kernel/random/boot_id"))
+	bootID := strings.TrimSpace(harness.RunSSH(t, tgt, "cat /proc/sys/kernel/random/boot_id"))
 	require.NotEmpty(t, bootID, "fresh guest returned an empty boot ID")
 	waitForCloudInitTarget(t, tgt)
 

tests/e2e/single/securitygroup_test.go

@@ -3,10 +3,7 @@
 package single
 
 import (
-	"bytes"
-	"os/exec"
 	"regexp"
-	"strconv"
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/ec2"
@@ -40,28 +37,3 @@ func authorizeAllowAllEgress(c *harness.AWSClient, sgID string) error {
 	})
 	return err
 }
-
-// runSSHCombined runs `command` over SSH against tgt and returns combined
-// stdout+stderr regardless of exit status. Unlike runSSH (instance_test.go)
-// it does not t.Fatal on non-zero exit — needed because `ping` legitimately
-// exits non-zero when packets are dropped, which is the expected outcome of
-// the revoke test.
-func runSSHCombined(tgt harness.SSHTarget, command string) (string, error) {
-	args := []string{
-		"-o", "StrictHostKeyChecking=no",
-		"-o", "UserKnownHostsFile=/dev/null",
-		"-o", "LogLevel=ERROR",
-		"-o", "ConnectTimeout=5",
-		"-o", "BatchMode=yes",
-		"-p", strconv.Itoa(tgt.Port),
-		"-i", tgt.KeyPath,
-		tgt.User + "@" + tgt.Host,
-		command,
-	}
-	var combined bytes.Buffer
-	cmd := exec.Command("ssh", args...)
-	cmd.Stdout = &combined
-	cmd.Stderr = &combined
-	err := cmd.Run()
-	return combined.String(), err
-}

tests/e2e/single/sgreachability_test.go

@@ -85,7 +85,7 @@ func runSGReachabilityPolicy(t *testing.T, fix *Fixture) {
 		// No ingress rules yet; egress is allow-all by default so only inbound
 		// is gated. Probe a short window to confirm the default-deny ACL is
 		// applied and stable. This overlaps guest boot, and Authorize below
-		// still pays the full boot wait via trySSHReady, so a longer window
+		// still pays the full boot wait via sshHealth.TryReady, so a longer window
 		// buys little extra coverage.
 		harness.Step(t, "asserting tcp/22 stays blocked under default-deny SG")
 		deadline := time.Now().Add(10 * time.Second)
@@ -99,12 +99,12 @@ func runSGReachabilityPolicy(t *testing.T, fix *Fixture) {
 	authorizeOK := t.Run("Authorize", func(t *testing.T) {
 		harness.Step(t, "authorizing tcp/22 ingress, expecting reachability")
 		harness.AuthorizeSSHIngress(t, fix.AWS, sgID)
-		require.Truef(t, trySSHReady(pubIP, 22, keyPath, sshReadyBudget),
+		require.Truef(t, sshHealth.TryReady(pubIP, 22, keyPath, harness.SSHReadyBudget),
 			"tcp/22 to %s never became reachable after authorizing ingress — "+
 				"default subnet egress/IGW datapath is broken", pubIP)
 
 		tgt := harness.SSHTarget{User: "ubuntu", Host: pubIP, Port: 22, KeyPath: keyPath}
-		idOut := runSSH(t, tgt, "id")
+		idOut := harness.RunSSH(t, tgt, "id")
 		assert.Containsf(t, idOut, "ubuntu", "ssh id after authorize\n%s", idOut)
 	})
 	if !authorizeOK {
@@ -193,7 +193,7 @@ func runSGReachabilityPolicy(t *testing.T, fix *Fixture) {
 	harness.Detail(t, "probe_wan", wanIP)
 
 	probeWAN := func() string {
-		out, _ := runSSHCombined(tgt, fmt.Sprintf("ping -c 3 -W 2 %s", wanIP))
+		out, _ := harness.RunSSHCombined(tgt, fmt.Sprintf("ping -c 3 -W 2 %s", wanIP))
 		return out
 	}
 

tests/e2e/single/smtpblock_test.go

@@ -66,7 +66,7 @@ func runWANEgressSMTPBlock(t *testing.T, fix *Fixture) {
 
 	harness.Step(t, "authorizing tcp/22 ingress, expecting reachability")
 	harness.AuthorizeSSHIngress(t, fix.AWS, sgID)
-	require.Truef(t, trySSHReady(pubIP, 22, keyPath, sshReadyBudget),
+	require.Truef(t, sshHealth.TryReady(pubIP, 22, keyPath, harness.SSHReadyBudget),
 		"tcp/22 to %s never became reachable after authorizing ingress", pubIP)
 	tgt := harness.SSHTarget{User: "ubuntu", Host: pubIP, Port: 22, KeyPath: keyPath}
 

tests/e2e/single/vpcegress_test.go

@@ -268,7 +268,7 @@ func runVPCEgressPaths(t *testing.T, fix *Fixture) {
 		_, _ = c.EC2.TerminateInstances(&ec2.TerminateInstancesInput{
 			InstanceIds: []*string{aws.String(pubInstanceID)},
 		})
-		_ = waitForInstanceStateSoft(c, pubInstanceID, "terminated", 5*time.Minute)
+		_ = harness.WaitForInstanceStateSoft(c, pubInstanceID, "terminated", 5*time.Minute)
 	})
 	harness.Detail(t, "instance", pubInstanceID)
 
@@ -281,7 +281,7 @@ func runVPCEgressPaths(t *testing.T, fix *Fixture) {
 	harness.Detail(t, "public_ip", pubIP, "private_ip", pubPrivIP)
 
 	pubOK := t.Run("PublicSubnetEgress", func(t *testing.T) {
-		if !trySSHReady(pubIP, 22, keyPath, sshReadyBudget) {
+		if !sshHealth.TryReady(pubIP, 22, keyPath, harness.SSHReadyBudget) {
 			harness.DumpVPCFlowDiagnostics(t, c, pubInstanceID,
 				fmt.Sprintf("PublicSubnetEgress SSH timeout — vpc=%s igw=%s pub=%s", vpcID, igwID, pubIP),
 				harness.VPCDiagnosticsOpts{
@@ -289,13 +289,13 @@ func runVPCEgressPaths(t *testing.T, fix *Fixture) {
 					LogicalIP:   pubPrivIP,
 					ArtifactDir: fix.ArtifactDir(t),
 				})
-			t.Fatalf("SSH handshake %s:22 never completed within %s (see diagnostics above)", pubIP, sshReadyBudget)
+			t.Fatalf("SSH handshake %s:22 never completed within %s (see diagnostics above)", pubIP, harness.SSHReadyBudget)
 		}
 
 		tgt := harness.SSHTarget{User: "ubuntu", Host: pubIP, Port: 22, KeyPath: keyPath}
 
 		harness.Step(t, "ssh id (smoke)")
-		idOut := runSSH(t, tgt, "id")
+		idOut := harness.RunSSH(t, tgt, "id")
 		require.Containsf(t, idOut, "ubuntu", "ssh id did not report ubuntu\n%s", idOut)
 
 		// Verify external connectivity through the IGW + 0.0.0.0/0 route. ping
@@ -405,7 +405,7 @@ func runVPCEgressPaths(t *testing.T, fix *Fixture) {
 		harness.Detail(t, "instance", instanceID, "public_ip", rbPubIP)
 
 		harness.Step(t, "expecting external SSH to instance in fresh-VPC public subnet")
-		if !trySSHReady(rbPubIP, 22, keyPath, sshReadyBudget) {
+		if !sshHealth.TryReady(rbPubIP, 22, keyPath, harness.SSHReadyBudget) {
 			harness.DumpVPCFlowDiagnostics(t, c, instanceID,
 				fmt.Sprintf("RouteBeforeSubnet SSH timeout — vpc=%s igw=%s pub=%s", vID, iID, rbPubIP),
 				harness.VPCDiagnosticsOpts{
@@ -419,7 +419,7 @@ func runVPCEgressPaths(t *testing.T, fix *Fixture) {
 		}
 
 		tgt := harness.SSHTarget{User: "ubuntu", Host: rbPubIP, Port: 22, KeyPath: keyPath}
-		idOut := runSSH(t, tgt, "id")
+		idOut := harness.RunSSH(t, tgt, "id")
 		assert.Containsf(t, idOut, "ubuntu", "ssh id in fresh-VPC subnet\n%s", idOut)
 	})
 
@@ -442,7 +442,7 @@ func runVPCEgressPaths(t *testing.T, fix *Fixture) {
 		_, _ = c.EC2.TerminateInstances(&ec2.TerminateInstancesInput{
 			InstanceIds: []*string{aws.String(privInstanceID)},
 		})
-		_ = waitForInstanceStateSoft(c, privInstanceID, "terminated", 5*time.Minute)
+		_ = harness.WaitForInstanceStateSoft(c, privInstanceID, "terminated", 5*time.Minute)
 	})
 
 	privInst := harness.WaitForInstanceState(t, c, privInstanceID, "running")
@@ -456,7 +456,7 @@ func runVPCEgressPaths(t *testing.T, fix *Fixture) {
 	// Copy the keypair to the bastion so it can hop into the private guest.
 	harness.Step(t, "scp keypair -> bastion:/tmp/key.pem")
 	scpKey(t, keyPath, pubIP)
-	_ = runSSH(t, bastionTgt, "chmod 600 /tmp/key.pem")
+	_ = harness.RunSSH(t, bastionTgt, "chmod 600 /tmp/key.pem")
 
 	privProbe := fmt.Sprintf(
 		"ssh -o StrictHostKeyChecking=no -o UserKnownHostsFile=/dev/null "+
@@ -727,7 +727,7 @@ func runVPCEgressPaths(t *testing.T, fix *Fixture) {
 		}
 
 		harness.Step(t, "ssh to guest via %s after restart", newIP)
-		if !trySSHReady(newIP, 22, keyPath, sshReadyBudget) {
+		if !sshHealth.TryReady(newIP, 22, keyPath, harness.SSHReadyBudget) {
 			harness.DumpVPCFlowDiagnostics(t, c, pubInstanceID,
 				fmt.Sprintf("StopStartAddress SSH timeout — was=%s now=%s instance=%s", pubIP, newIP, pubInstanceID),
 				harness.VPCDiagnosticsOpts{
@@ -735,9 +735,9 @@ func runVPCEgressPaths(t *testing.T, fix *Fixture) {
 					LogicalIP:   pubPrivIP,
 					ArtifactDir: fix.ArtifactDir(t),
 				})
-			t.Fatalf("guest unreachable via %s within %s after restart (see diagnostics above)", newIP, sshReadyBudget)
+			t.Fatalf("guest unreachable via %s within %s after restart (see diagnostics above)", newIP, harness.SSHReadyBudget)
 		}
-		out := runSSH(t, harness.SSHTarget{User: "ubuntu", Host: newIP, Port: 22, KeyPath: keyPath}, "id")
+		out := harness.RunSSH(t, harness.SSHTarget{User: "ubuntu", Host: newIP, Port: 22, KeyPath: keyPath}, "id")
 		require.Containsf(t, out, "ubuntu", "ssh after restart did not report ubuntu\n%s", out)
 		harness.Detail(t, "datapath", "reachable_on_new_address_ok")
 
@@ -789,7 +789,7 @@ func runVPCEgressPaths(t *testing.T, fix *Fixture) {
 		// association publishes; poll the SSH handshake rather than
 		// asserting immediately.
 		harness.Step(t, "ssh to guest via EIP %s", eipIP)
-		if !trySSHReady(eipIP, 22, keyPath, sshReadyBudget) {
+		if !sshHealth.TryReady(eipIP, 22, keyPath, harness.SSHReadyBudget) {
 			harness.DumpVPCFlowDiagnostics(t, c, pubInstanceID,
 				fmt.Sprintf("EIP SSH timeout — eip=%s instance=%s", eipIP, pubInstanceID),
 				harness.VPCDiagnosticsOpts{
@@ -797,10 +797,10 @@ func runVPCEgressPaths(t *testing.T, fix *Fixture) {
 					LogicalIP:   pubPrivIP,
 					ArtifactDir: fix.ArtifactDir(t),
 				})
-			t.Fatalf("guest unreachable via EIP %s within %s (see diagnostics above)", eipIP, sshReadyBudget)
+			t.Fatalf("guest unreachable via EIP %s within %s (see diagnostics above)", eipIP, harness.SSHReadyBudget)
 		}
 		tgt := harness.SSHTarget{User: "ubuntu", Host: eipIP, Port: 22, KeyPath: keyPath}
-		idOut := runSSH(t, tgt, "id")
+		idOut := harness.RunSSH(t, tgt, "id")
 		require.Containsf(t, idOut, "ubuntu", "ssh via EIP id did not report ubuntu\n%s", idOut)
 		harness.Detail(t, "datapath", "eip_reachable_ok")
 
@@ -851,7 +851,7 @@ func runVPCEgressPaths(t *testing.T, fix *Fixture) {
 				"instance came back on a different address than the Elastic IP associated to it")
 
 			harness.Step(t, "ssh to guest via EIP %s after restart", eipIP)
-			if !trySSHReady(eipIP, 22, keyPath, sshReadyBudget) {
+			if !sshHealth.TryReady(eipIP, 22, keyPath, harness.SSHReadyBudget) {
 				harness.DumpVPCFlowDiagnostics(t, c, pubInstanceID,
 					fmt.Sprintf("EIP SSH timeout after restart — eip=%s instance=%s", eipIP, pubInstanceID),
 					harness.VPCDiagnosticsOpts{
@@ -859,9 +859,9 @@ func runVPCEgressPaths(t *testing.T, fix *Fixture) {
 						LogicalIP:   aws.StringValue(started.PrivateIpAddress),
 						ArtifactDir: fix.ArtifactDir(t),
 					})
-				t.Fatalf("guest unreachable via EIP %s within %s after restart (see diagnostics above)", eipIP, sshReadyBudget)
+				t.Fatalf("guest unreachable via EIP %s within %s after restart (see diagnostics above)", eipIP, harness.SSHReadyBudget)
 			}
-			out := runSSH(t, harness.SSHTarget{User: "ubuntu", Host: eipIP, Port: 22, KeyPath: keyPath}, "id")
+			out := harness.RunSSH(t, harness.SSHTarget{User: "ubuntu", Host: eipIP, Port: 22, KeyPath: keyPath}, "id")
 			require.Containsf(t, out, "ubuntu", "ssh via EIP after restart did not report ubuntu\n%s", out)
 			harness.Detail(t, "datapath", "eip_reachable_after_restart_ok")
 		})
@@ -958,26 +958,6 @@ func waitForNATGatewayStateSoft(c *harness.AWSClient, id, target string, timeout
 	}
 }
 
-// waitForInstanceStateSoft is the cleanup-time analogue of
-// harness.WaitForInstanceState — no t.Fatal, just polls and returns.
-func waitForInstanceStateSoft(c *harness.AWSClient, id, target string, timeout time.Duration) error {
-	deadline := time.Now().Add(timeout)
-	for {
-		out, err := c.EC2.DescribeInstances(&ec2.DescribeInstancesInput{
-			InstanceIds: []*string{aws.String(id)},
-		})
-		if err == nil && len(out.Reservations) > 0 && len(out.Reservations[0].Instances) > 0 {
-			if aws.StringValue(out.Reservations[0].Instances[0].State.Name) == target {
-				return nil
-			}
-		}
-		if time.Now().After(deadline) {
-			return fmt.Errorf("instance %s did not reach %s within %s", id, target, timeout)
-		}
-		time.Sleep(2 * time.Second)
-	}
-}
-
 // scpKey copies the harness PEM to /tmp/key.pem on the bastion so it can
 // hop into a guest that has no direct SSH ingress of its own.
 func scpKey(t *testing.T, keyPath, host string) {
@@ -1004,7 +984,7 @@ func scpKey(t *testing.T, keyPath, host string) {
 // EventuallyErr budget. The ctx deadline kills ssh regardless of which phase hangs.
 const sshQuietHardTimeout = 30 * time.Second
 
-// runSSHQuiet is the EventuallyErr-friendly variant of runSSH: it returns
+// runSSHQuiet is the EventuallyErr-friendly variant of harness.RunSSH: it returns
 // stdout+stderr and the error rather than calling t.Fatal, so polling
 // loops can iterate on transient failures (cloud-init not done, OVN
 // flow not installed yet, etc).