feat(e2e): more e2e coverage (#365)

spinifex/utils/utils_test.go

@@ -1182,23 +1182,6 @@ func TestHashMAC_IdSeparates(t *testing.T) {
 	assert.NotEqual(t, a, b)
 }
 
-func TestHashMAC_Distribution(t *testing.T) {
-	// 100k random ids — 0 collisions expected. 46-bit space, birthday-
-	// paradox 1% at ~1.2M; 100k is well below.
-	if testing.Short() {
-		t.Skip("skipping distribution test in -short mode")
-	}
-	seen := make(map[string]string, 100_000)
-	for range 100_000 {
-		id := GenerateResourceID("res")
-		mac := HashMAC(id)
-		if prev, ok := seen[mac]; ok {
-			t.Fatalf("collision on %s: id=%s prev=%s", mac, id, prev)
-		}
-		seen[mac] = id
-	}
-}
-
 // writeTempImage writes content to a file named name under t.TempDir() and
 // returns the absolute path.
 func writeTempImage(t *testing.T, name string, content []byte) string {

tests/e2e/eks/eks_test.go

@@ -17,6 +17,7 @@ import (
 	"github.com/aws/aws-sdk-go/aws/awserr"
 	"github.com/aws/aws-sdk-go/service/ec2"
 	"github.com/aws/aws-sdk-go/service/eks"
+	"github.com/aws/aws-sdk-go/service/iam"
 	"github.com/aws/aws-sdk-go/service/sts"
 	"github.com/mulgadc/spinifex/tests/e2e/harness"
 	"github.com/stretchr/testify/assert"
@@ -208,6 +209,11 @@ func TestEKS(t *testing.T) {
 		}, 3*time.Minute, 10*time.Second)
 	})
 
+	t.Run("IRSAWebIdentity", func(t *testing.T) {
+		requireClusterReady(t, fx)
+		runIRSAWebIdentity(t, c, env, artifacts, fx)
+	})
+
 	t.Run("DeleteCluster", func(t *testing.T) {
 		requireClusterReady(t, fx)
 		_, err := c.EKS.DeleteCluster(&eks.DeleteClusterInput{Name: aws.String(fx.ClusterName)})
@@ -217,6 +223,103 @@ func TestEKS(t *testing.T) {
 	})
 }
 
+// runIRSAWebIdentity exercises the IRSA token-exchange path at the API level
+// (no pod, no nodegroup): register the cluster's OIDC provider, create a role
+// trusting it, mint a real ServiceAccount token via `kubectl create token`
+// (signed by the cluster's OIDC key, JWKS published at CreateCluster), exchange
+// it through AssumeRoleWithWebIdentity, and prove the returned credentials are
+// usable via GetCallerIdentity.
+func runIRSAWebIdentity(t *testing.T, c *harness.AWSClient, env *harness.Env, artifacts string, fx *clusterFixture) {
+	require.NotNil(t, fx.Cluster.Identity, "ACTIVE cluster must expose Identity")
+	require.NotNil(t, fx.Cluster.Identity.Oidc, "ACTIVE cluster must expose Identity.Oidc")
+	issuer := aws.StringValue(fx.Cluster.Identity.Oidc.Issuer)
+	require.NotEmpty(t, issuer, "cluster OIDC issuer must be published")
+	t.Logf("OIDC issuer: %s", issuer)
+
+	// 1) Register the cluster's OIDC provider in the caller account so the STS
+	//    handler will accept tokens carrying this issuer.
+	oidcOut, err := c.IAM.CreateOpenIDConnectProvider(&iam.CreateOpenIDConnectProviderInput{
+		Url:            aws.String(issuer),
+		ClientIDList:   aws.StringSlice([]string{"sts.amazonaws.com"}),
+		ThumbprintList: aws.StringSlice([]string{"0000000000000000000000000000000000000000"}),
+	})
+	require.NoError(t, err, "create-open-id-connect-provider")
+	providerArn := aws.StringValue(oidcOut.OpenIDConnectProviderArn)
+	require.NotEmpty(t, providerArn, "OIDC provider ARN empty")
+	t.Cleanup(func() {
+		_, _ = c.IAM.DeleteOpenIDConnectProvider(&iam.DeleteOpenIDConnectProviderInput{
+			OpenIDConnectProviderArn: aws.String(providerArn),
+		})
+	})
+	t.Logf("OIDC provider: %s", providerArn)
+
+	// 2) Create a role whose trust policy federates the OIDC provider. No
+	//    Condition block — the Federated principal + AssumeRoleWithWebIdentity
+	//    action are sufficient to grant, which keeps the test independent of the
+	//    condition-key issuer-prefix format.
+	roleName := fmt.Sprintf("%s-irsa", fx.ClusterName)
+	trustPolicy := fmt.Sprintf(`{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Principal":{"Federated":%q},"Action":"sts:AssumeRoleWithWebIdentity"}]}`, providerArn)
+	roleOut, err := c.IAM.CreateRole(&iam.CreateRoleInput{
+		RoleName:                 aws.String(roleName),
+		AssumeRolePolicyDocument: aws.String(trustPolicy),
+		Description:              aws.String("E2E IRSA web-identity role"),
+	})
+	require.NoError(t, err, "create-role")
+	roleArn := aws.StringValue(roleOut.Role.Arn)
+	require.NotEmpty(t, roleArn, "role ARN empty")
+	t.Cleanup(func() {
+		_, _ = c.IAM.DeleteRole(&iam.DeleteRoleInput{RoleName: aws.String(roleName)})
+	})
+	t.Logf("role: %s", roleArn)
+
+	// 3) Mint a ServiceAccount token bound to sts.amazonaws.com. `kubectl create
+	//    token` uses the TokenRequest API — no pod required. The token's iss is
+	//    the cluster OIDC issuer and aud includes sts.amazonaws.com (k3s is wired
+	//    with --service-account-issuer / --api-audiences at CreateCluster).
+	kcPath := writeKubeconfig(t, artifacts, fx.Cluster)
+	kc := harness.NewKubectl(t, kcPath, getTokenEnv(t, env))
+	tokenOut, err := kc.Run(30*time.Second, "create", "token", "default",
+		"--namespace", "default", "--audience", "sts.amazonaws.com")
+	require.NoErrorf(t, err, "kubectl create token:\n%s", tokenOut)
+	token := strings.TrimSpace(tokenOut)
+	require.Equal(t, 2, strings.Count(token, "."), "web-identity token must be a JWT (3 dot-separated parts)")
+
+	// 4) Exchange the token. AssumeRoleWithWebIdentity is anonymous (the SDK
+	//    strips SigV4 for this op); the JWT is the identity.
+	const sessionName = "e2e-irsa"
+	var assumeOut *sts.AssumeRoleWithWebIdentityOutput
+	harness.EventuallyErr(t, func() error {
+		out, aerr := c.STS.AssumeRoleWithWebIdentity(&sts.AssumeRoleWithWebIdentityInput{
+			RoleArn:          aws.String(roleArn),
+			RoleSessionName:  aws.String(sessionName),
+			WebIdentityToken: aws.String(token),
+		})
+		if aerr != nil {
+			return fmt.Errorf("assume-role-with-web-identity: %w", aerr)
+		}
+		assumeOut = out
+		return nil
+	}, 60*time.Second, 5*time.Second)
+	require.NotNil(t, assumeOut.Credentials, "AssumeRoleWithWebIdentity returned no credentials")
+	assert.Equal(t, "system:serviceaccount:default:default",
+		aws.StringValue(assumeOut.SubjectFromWebIdentityToken), "subject must be the default SA")
+	require.True(t, strings.HasPrefix(aws.StringValue(assumeOut.Credentials.AccessKeyId), "ASIA"),
+		"web-identity credentials must be temporary (ASIA…)")
+
+	// 5) The returned temporary credentials must be usable: GetCallerIdentity
+	//    must resolve to the assumed-role principal.
+	sessClient := harness.NewAWSClientWithSessionCreds(t, env,
+		aws.StringValue(assumeOut.Credentials.AccessKeyId),
+		aws.StringValue(assumeOut.Credentials.SecretAccessKey),
+		aws.StringValue(assumeOut.Credentials.SessionToken))
+	ident, err := sessClient.STS.GetCallerIdentity(&sts.GetCallerIdentityInput{})
+	require.NoError(t, err, "get-caller-identity with web-identity creds")
+	assert.Equal(t, fx.AccountID, aws.StringValue(ident.Account), "caller account must match")
+	assert.Contains(t, aws.StringValue(ident.Arn), "assumed-role/"+roleName,
+		"caller ARN must reflect the assumed IRSA role")
+	t.Logf("GetCallerIdentity (web-identity creds): %s", aws.StringValue(ident.Arn))
+}
+
 // --- Fixture --------------------------------------------------------------
 
 type clusterFixture struct {

tests/e2e/harness/aws.go

@@ -13,6 +13,7 @@ import (
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/aws/credentials"
 	"github.com/aws/aws-sdk-go/aws/session"
+	"github.com/aws/aws-sdk-go/service/acm"
 	"github.com/aws/aws-sdk-go/service/ec2"
 	"github.com/aws/aws-sdk-go/service/ec2/ec2iface"
 	"github.com/aws/aws-sdk-go/service/eks"
@@ -33,6 +34,7 @@ type AWSClient struct {
 	IAM     *iam.IAM
 	STS     *sts.STS
 	EKS     *eks.EKS
+	ACM     *acm.ACM
 }
 
 // NewAWSClient builds clients pointed at the spinifex gateway using the
@@ -128,6 +130,7 @@ func newAWSClient(t *testing.T, env *Env, accessKey, secretKey, sessionToken str
 		IAM:     iam.New(sess),
 		STS:     sts.New(sess),
 		EKS:     eks.New(sess),
+		ACM:     acm.New(sess),
 	}
 }
 
@@ -141,5 +144,6 @@ func (c *AWSClient) IgnoreCertErrors() {
 	c.EKS.Config.HTTPClient = hc
 	c.IAM.Config.HTTPClient = hc
 	c.STS.Config.HTTPClient = hc
+	c.ACM.Config.HTTPClient = hc
 	_ = (*x509.CertPool)(nil) // silence unused-import on hardened paths
 }

tests/e2e/harness/cert.go

@@ -3,15 +3,21 @@
 package harness
 
 import (
+	"crypto/ecdsa"
+	"crypto/elliptic"
+	"crypto/rand"
 	"crypto/x509"
+	"crypto/x509/pkix"
 	"encoding/pem"
 	"fmt"
+	"math/big"
 	"net"
 	"os"
 	"os/exec"
 	"path/filepath"
 	"strings"
 	"testing"
+	"time"
 )
 
 // ResolveCACert walks the standard config locations for the Spinifex CA cert
@@ -121,6 +127,50 @@ func OpenSSLVerify(t *testing.T, caFile, host string, port int) error {
 	return fmt.Errorf("openssl verify %s failed: %v\noutput:\n%s", target, err, out)
 }
 
+// GenerateSelfSignedCertPEM mints an ECDSA P-256 self-signed leaf for host
+// (an IP or DNS name) and returns the cert PEM, the PKCS#8 key PEM, and the
+// parsed leaf. Used by the ACM ImportCertificate → HTTPS listener e2e to feed
+// a real keypair through the import → HAProxy termination path.
+func GenerateSelfSignedCertPEM(host string) (certPEM, keyPEM []byte, leaf *x509.Certificate, err error) {
+	key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
+	if err != nil {
+		return nil, nil, nil, fmt.Errorf("generate key: %w", err)
+	}
+	serial, err := rand.Int(rand.Reader, new(big.Int).Lsh(big.NewInt(1), 128))
+	if err != nil {
+		return nil, nil, nil, fmt.Errorf("serial: %w", err)
+	}
+	tmpl := &x509.Certificate{
+		SerialNumber:          serial,
+		Subject:               pkix.Name{CommonName: host},
+		NotBefore:             time.Now().Add(-time.Hour),
+		NotAfter:              time.Now().Add(24 * time.Hour),
+		KeyUsage:              x509.KeyUsageDigitalSignature | x509.KeyUsageKeyEncipherment,
+		ExtKeyUsage:           []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth},
+		BasicConstraintsValid: true,
+	}
+	if ip := net.ParseIP(host); ip != nil {
+		tmpl.IPAddresses = []net.IP{ip}
+	} else {
+		tmpl.DNSNames = []string{host}
+	}
+	der, err := x509.CreateCertificate(rand.Reader, tmpl, tmpl, &key.PublicKey, key)
+	if err != nil {
+		return nil, nil, nil, fmt.Errorf("create certificate: %w", err)
+	}
+	keyDER, err := x509.MarshalPKCS8PrivateKey(key)
+	if err != nil {
+		return nil, nil, nil, fmt.Errorf("marshal key: %w", err)
+	}
+	certPEM = pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: der})
+	keyPEM = pem.EncodeToMemory(&pem.Block{Type: "PRIVATE KEY", Bytes: keyDER})
+	leaf, err = x509.ParseCertificate(der)
+	if err != nil {
+		return nil, nil, nil, fmt.Errorf("parse leaf: %w", err)
+	}
+	return certPEM, keyPEM, leaf, nil
+}
+
 // FingerprintMatches returns true if the two PEM certs share the SHA-256
 // fingerprint that openssl x509 -fingerprint prints.
 func FingerprintMatches(a, b *x509.Certificate) bool {

tests/e2e/harness/ec2helpers.go

@@ -8,6 +8,7 @@ import (
 	"strconv"
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/aws/awserr"
@@ -213,6 +214,59 @@ func EnsureDefaultSGOpen(t *testing.T, c *AWSClient) {
 	AuthorizeICMPIngress(t, c, sgID)
 }
 
+// AttachVolumeWait attaches volID to instanceID at device and blocks until the
+// volume reaches "in-use". Used by the data-durability tests where the attach
+// must complete before the guest can format the disk.
+func AttachVolumeWait(t *testing.T, c *AWSClient, volID, instanceID, device string) {
+	t.Helper()
+	_, err := c.EC2.AttachVolume(&ec2.AttachVolumeInput{
+		VolumeId:   aws.String(volID),
+		InstanceId: aws.String(instanceID),
+		Device:     aws.String(device),
+	})
+	if err != nil {
+		t.Fatalf("attach-volume %s -> %s as %s: %v", volID, instanceID, device, err)
+	}
+	WaitForVolumeState(t, c, volID, ec2.VolumeStateInUse, WithPoll(500*time.Millisecond))
+}
+
+// DetachVolumeWait detaches volID and blocks until it reaches "available".
+func DetachVolumeWait(t *testing.T, c *AWSClient, volID string) {
+	t.Helper()
+	_, err := c.EC2.DetachVolume(&ec2.DetachVolumeInput{VolumeId: aws.String(volID)})
+	if err != nil {
+		t.Fatalf("detach-volume %s: %v", volID, err)
+	}
+	WaitForVolumeState(t, c, volID, "available", WithPoll(500*time.Millisecond))
+}
+
+// RegisterVolumeTeardown best-effort force-detaches then deletes volID at test
+// end. Non-fatal — it polls for "available" within a bounded window and never
+// calls t.Fatal so a cleanup hiccup doesn't mask the test's real result.
+func RegisterVolumeTeardown(t *testing.T, c *AWSClient, volID string) {
+	t.Helper()
+	t.Cleanup(func() {
+		_, _ = c.EC2.DetachVolume(&ec2.DetachVolumeInput{
+			VolumeId: aws.String(volID),
+			Force:    aws.Bool(true),
+		})
+		deadline := time.Now().Add(90 * time.Second)
+		for time.Now().Before(deadline) {
+			out, err := c.EC2.DescribeVolumes(&ec2.DescribeVolumesInput{
+				VolumeIds: []*string{aws.String(volID)},
+			})
+			if err != nil || len(out.Volumes) == 0 {
+				return
+			}
+			if aws.StringValue(out.Volumes[0].State) == "available" {
+				break
+			}
+			time.Sleep(2 * time.Second)
+		}
+		_, _ = c.EC2.DeleteVolume(&ec2.DeleteVolumeInput{VolumeId: aws.String(volID)})
+	})
+}
+
 // asErr is a thin errors.As that lets the helper stay testify-free.
 func asErr(err error, target *awserr.Error) bool {
 	if a, ok := err.(awserr.Error); ok {

tests/e2e/harness/eks_mgmt.go

@@ -11,6 +11,7 @@ import (
 	"github.com/mulgadc/spinifex/spinifex/config"
 	handlers_eks "github.com/mulgadc/spinifex/spinifex/handlers/eks"
 	"github.com/mulgadc/spinifex/spinifex/utils"
+	"github.com/spf13/viper"
 )
 
 // ControlPlaneMgmtIP resolves the EKS control-plane VM's br-mgmt address —
@@ -56,7 +57,7 @@ func natsConn(t *testing.T, env *Env) (host, token, ca string) {
 	}
 
 	cfgPath := filepath.Join(env.ConfigDir, "spinifex.toml")
-	cc, err := config.LoadConfig(cfgPath)
+	cc, err := loadClusterConfig(cfgPath)
 	if err != nil {
 		t.Fatalf("load %s for NATS settings: %v", cfgPath, err)
 	}
@@ -67,6 +68,23 @@ func natsConn(t *testing.T, env *Env) (host, token, ca string) {
 	return dialableNATSHost(node.NATS.Host), node.NATS.ACL.Token, node.NATS.CACert
 }
 
+// loadClusterConfig reads spinifex.toml via a private viper instance with no
+// AutomaticEnv, so the harness's own SPINIFEX_* vars (the multinode workflow
+// exports SPINIFEX_NODES) cannot shadow the [nodes.*] table and blank the parse.
+func loadClusterConfig(path string) (*config.ClusterConfig, error) {
+	v := viper.New()
+	v.SetConfigFile(path)
+	v.SetConfigType("toml")
+	if err := v.ReadInConfig(); err != nil {
+		return nil, err
+	}
+	var cc config.ClusterConfig
+	if err := v.Unmarshal(&cc); err != nil {
+		return nil, err
+	}
+	return &cc, nil
+}
+
 // dialableNATSHost rewrites a wildcard bind address to loopback. NATS.Host is a
 // bind address (commonly 0.0.0.0:4222); the TLS serving cert SANs the loopback
 // and real node IPs but not 0.0.0.0, so dial 127.0.0.1 instead.

tests/e2e/harness/guest_data.go

@@ -0,0 +1,159 @@
+//go:build e2e
+
+package harness
+
+import (
+	"context"
+	"fmt"
+	"os/exec"
+	"regexp"
+	"strconv"
+	"strings"
+	"testing"
+	"time"
+)
+
+// Guest-side data-durability helpers shared by the volume-durability,
+// snapshot-restore, and CreateImage e2e tests. They drive a real guest over SSH
+// to format, write, sync, and checksum bytes so the assembled
+// QEMU↔viperblock↔predastore I/O path is exercised end-to-end — something the
+// per-layer unit tests cannot cover.
+
+// guestSentinelFile is the file written into a freshly formatted data volume.
+const guestSentinelFile = "e2e-sentinel.bin"
+
+// guestExecTimeout bounds a single guest command (mkfs/dd are sub-second on the
+// small payloads used here; the ceiling only guards against a hung SSH).
+const guestExecTimeout = 2 * time.Minute
+
+// sha256RE matches a bare sha256 digest so the checksum can be lifted out of
+// command output that may also carry the file path (`<sha>␠␠<path>`).
+var sha256RE = regexp.MustCompile(`\b[0-9a-f]{64}\b`)
+
+// GuestExec runs cmd over SSH against tgt and returns combined stdout+stderr
+// plus the run error. It never calls t.Fatal, so callers can branch on an
+// expected non-zero exit.
+func GuestExec(tgt SSHTarget, cmd string) (string, error) {
+	ctx, cancel := context.WithTimeout(context.Background(), guestExecTimeout)
+	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()
+	return string(out), err
+}
+
+// guestDiskSet returns the set of whole-disk device names (lsblk TYPE=disk)
+// currently visible in the guest, keyed by bare name (e.g. "vda", "vdc").
+func guestDiskSet(tgt SSHTarget) (map[string]struct{}, error) {
+	out, err := GuestExec(tgt, "lsblk -dn -o NAME,TYPE")
+	if err != nil {
+		return nil, fmt.Errorf("lsblk: %w\n%s", err, out)
+	}
+	set := map[string]struct{}{}
+	for _, line := range strings.Split(strings.TrimSpace(out), "\n") {
+		f := strings.Fields(line)
+		if len(f) >= 2 && f[1] == "disk" {
+			set[f[0]] = struct{}{}
+		}
+	}
+	return set, nil
+}
+
+// GuestDiskSet is the t.Fatal-on-error wrapper around guestDiskSet, used to
+// snapshot the guest's disk inventory before an AttachVolume.
+func GuestDiskSet(t *testing.T, tgt SSHTarget) map[string]struct{} {
+	t.Helper()
+	set, err := guestDiskSet(tgt)
+	if err != nil {
+		t.Fatalf("GuestDiskSet: %v", err)
+	}
+	return set
+}
+
+// WaitForNewGuestDisk polls the guest until a whole-disk device absent from
+// `before` appears, returning its bare name (e.g. "vdc"). Proves a hotplugged
+// volume actually reached the guest kernel, not just the EC2 control plane.
+func WaitForNewGuestDisk(t *testing.T, tgt SSHTarget, before map[string]struct{}, timeout time.Duration) string {
+	t.Helper()
+	var found string
+	EventuallyErr(t, func() error {
+		now, err := guestDiskSet(tgt)
+		if err != nil {
+			return err
+		}
+		for name := range now {
+			if _, ok := before[name]; !ok {
+				found = name
+				return nil
+			}
+		}
+		names := make([]string, 0, len(now))
+		for name := range now {
+			names = append(names, name)
+		}
+		return fmt.Errorf("no new disk yet (visible: %s)", strings.Join(names, ","))
+	}, timeout, 2*time.Second)
+	return found
+}
+
+// GuestFormatWriteSentinel formats /dev/<dev> as ext4 with the given label,
+// mounts it, writes sizeMiB of random data to the sentinel file, fsyncs, and
+// returns the file's sha256. Unmounts before returning so the volume can be
+// detached cleanly.
+func GuestFormatWriteSentinel(t *testing.T, tgt SSHTarget, dev, label string, sizeMiB int) string {
+	t.Helper()
+	mnt := "/mnt/" + label
+	script := strings.Join([]string{
+		fmt.Sprintf("sudo mkfs.ext4 -F -L %s /dev/%s >/dev/null 2>&1", label, dev),
+		fmt.Sprintf("sudo mkdir -p %s", mnt),
+		fmt.Sprintf("sudo mount /dev/%s %s", dev, mnt),
+		fmt.Sprintf("sudo dd if=/dev/urandom of=%s/%s bs=1M count=%d conv=fsync status=none", mnt, guestSentinelFile, sizeMiB),
+		"sync",
+		fmt.Sprintf("sudo sha256sum %s/%s", mnt, guestSentinelFile),
+		fmt.Sprintf("sudo umount %s", mnt),
+	}, " && ")
+	out, err := GuestExec(tgt, script)
+	if err != nil {
+		t.Fatalf("GuestFormatWriteSentinel(%s,%s): %v\n%s", dev, label, err, out)
+	}
+	return mustSha(t, out)
+}
+
+// GuestReadSentinelSha mounts source (a device path such as "/dev/vdc" or a
+// "/dev/disk/by-label/<label>" path) at a temp mountpoint, returns the
+// sentinel file's sha256, and unmounts.
+func GuestReadSentinelSha(t *testing.T, tgt SSHTarget, source, label string) string {
+	t.Helper()
+	mnt := "/mnt/" + label
+	script := strings.Join([]string{
+		fmt.Sprintf("sudo mkdir -p %s", mnt),
+		fmt.Sprintf("sudo mount %s %s", source, mnt),
+		fmt.Sprintf("sudo sha256sum %s/%s", mnt, guestSentinelFile),
+		fmt.Sprintf("sudo umount %s", mnt),
+	}, " && ")
+	out, err := GuestExec(tgt, script)
+	if err != nil {
+		t.Fatalf("GuestReadSentinelSha(%s): %v\n%s", source, err, out)
+	}
+	return mustSha(t, out)
+}
+
+// mustSha extracts the single sha256 digest from command output, failing the
+// test if none is present.
+func mustSha(t *testing.T, out string) string {
+	t.Helper()
+	sha := sha256RE.FindString(out)
+	if sha == "" {
+		t.Fatalf("no sha256 digest in guest output:\n%s", out)
+	}
+	return sha
+}

tests/e2e/harness/jetstream.go

@@ -0,0 +1,39 @@
+//go:build e2e
+
+package harness
+
+import (
+	"testing"
+
+	"github.com/mulgadc/spinifex/spinifex/utils"
+)
+
+// KVReplicaFactors connects to the cluster NATS and returns the configured
+// replica factor for each named KV bucket, read from JetStream stream metadata
+// (the "KV_" stream prefix is applied internally). The replica count lives in
+// cluster-global Raft metadata, so any node's NATS reports the same value.
+// Fatals if NATS is unreachable or a bucket's stream cannot be introspected.
+func KVReplicaFactors(t *testing.T, env *Env, buckets ...string) map[string]int {
+	t.Helper()
+	host, token, ca := natsConn(t, env)
+	nc, err := utils.ConnectNATS(host, token, ca)
+	if err != nil {
+		t.Fatalf("connect NATS %s: %v", host, err)
+	}
+	defer nc.Close()
+
+	js, err := nc.JetStream()
+	if err != nil {
+		t.Fatalf("jetstream context: %v", err)
+	}
+
+	out := make(map[string]int, len(buckets))
+	for _, b := range buckets {
+		info, err := js.StreamInfo("KV_" + b)
+		if err != nil {
+			t.Fatalf("stream info for KV bucket %q: %v", b, err)
+		}
+		out[b] = info.Config.Replicas
+	}
+	return out
+}

tests/e2e/lb/lb_test.go

@@ -5,6 +5,7 @@ package lb
 import (
 	"bytes"
 	"context"
+	"crypto/x509"
 	_ "embed"
 	"encoding/base64"
 	"encoding/json"
@@ -18,6 +19,7 @@ import (
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/aws/awserr"
+	"github.com/aws/aws-sdk-go/service/acm"
 	"github.com/aws/aws-sdk-go/service/ec2"
 	"github.com/aws/aws-sdk-go/service/elbv2"
 	"github.com/mulgadc/spinifex/tests/e2e/harness"
@@ -88,6 +90,15 @@ func TestLoadBalancer(t *testing.T) {
 		}
 		runLBSuite(t, client, fixture, kindNLB, "internet-facing", ssh, peer)
 	})
+	t.Run("InternetFacing_ALB_HTTPS", func(t *testing.T) {
+		if peer == "" {
+			// HTTPS handshake is driven from the test runner against the LB
+			// public IP; gate on the same peer-available signal the sibling
+			// internet-facing subtests use, where driver→LB reachability holds.
+			t.Skip("no peer node available")
+		}
+		runHTTPSCertSuite(t, client, fixture)
+	})
 	// Gate remaining internal subtests on Internal_ALB: if the LB never reaches active,
 	// the rest will time out identically — fail fast instead of burning ~5min each.
 	albOK := t.Run("Internal_ALB", func(t *testing.T) {
@@ -527,6 +538,142 @@ func runLBSuite(t *testing.T, c *harness.AWSClient, f *sharedFixture, kind lbKin
 	runInternalTrafficViaClient(t, c, f, kind, priv)
 }
 
+// runHTTPSCertSuite exercises the ACM → HTTPS listener path end-to-end:
+// ImportCertificate a self-signed leaf → create an internet-facing ALB HTTPS
+// listener referencing the cert ARN → complete a TLS handshake against the LB
+// and assert HAProxy serves exactly the imported cert → drive one HTTPS GET
+// that terminates TLS at the LB and forwards to a healthy backend →
+// DeleteCertificate. The cert is minted for the LB's public IP (known only
+// after the LB is active) so the end-to-end GET can verify the chain by IP SAN.
+func runHTTPSCertSuite(t *testing.T, c *harness.AWSClient, f *sharedFixture) {
+	t.Helper()
+	const label = "ALB internet-facing HTTPS (ACM)"
+	const httpsPort int64 = 443
+
+	authorizeSGPort(t, c, f, "tcp", httpsPort)
+
+	tgArn := createTargetGroup(t, c, f, "lb-e2e-https-tg", "HTTP", httpPort, "/index.html")
+	t.Cleanup(func() { deleteTargetGroup(t, c, tgArn) })
+	registerTargets(t, c, tgArn, f.AppInstanceIDs)
+	t.Cleanup(func() { deregisterTargets(t, c, tgArn, f.AppInstanceIDs) })
+
+	lb := createLB(t, c, f, "lb-e2e-https", "application", "internet-facing")
+	t.Cleanup(func() { deleteLB(t, c, lb) })
+
+	harness.WaitForLBActive(t, c, lb.ARN, label, 5*time.Minute)
+	eni := lbENI(t, c, "app", lb)
+	pubIP := publicIP(eni)
+	require.NotEmpty(t, pubIP, label+" needs a public IP for the TLS handshake")
+
+	// Mint a self-signed leaf for the LB's public IP and import it into ACM.
+	certPEM, keyPEM, leaf, err := harness.GenerateSelfSignedCertPEM(pubIP)
+	require.NoError(t, err, "generate self-signed cert")
+	imp, err := c.ACM.ImportCertificate(&acm.ImportCertificateInput{
+		Certificate: certPEM,
+		PrivateKey:  keyPEM,
+	})
+	require.NoError(t, err, "acm import-certificate")
+	certArn := aws.StringValue(imp.CertificateArn)
+	require.NotEmpty(t, certArn, "ImportCertificate returned empty CertificateArn")
+	require.Contains(t, certArn, ":certificate/", "cert ARN must be an ACM certificate ARN")
+	t.Cleanup(func() {
+		if _, derr := c.ACM.DeleteCertificate(&acm.DeleteCertificateInput{
+			CertificateArn: aws.String(certArn),
+		}); derr != nil {
+			t.Logf("delete certificate %s: %v", certArn, derr)
+		}
+	})
+	t.Logf("imported cert: %s", certArn)
+
+	listener := createHTTPSListener(t, c, lb.ARN, httpsPort, tgArn, certArn)
+	t.Cleanup(func() { deleteListener(t, c, listener) })
+
+	harness.WaitForTargetsHealthy(t, c, tgArn, 2, label, 2*time.Minute)
+
+	// TLS handshake: HAProxy must present exactly the imported leaf. The
+	// lb-agent picks up the new listener cert on its reconcile tick, so poll
+	// the handshake until the served cert matches before asserting.
+	var served *x509.Certificate
+	harness.EventuallyErr(t, func() error {
+		got, ferr := harness.FetchServedCert(pubIP, int(httpsPort), 5*time.Second)
+		if ferr != nil {
+			return ferr
+		}
+		if !harness.FingerprintMatches(got, leaf) {
+			return fmt.Errorf("served cert does not match imported cert yet")
+		}
+		served = got
+		return nil
+	}, 90*time.Second, 3*time.Second)
+	require.NotNil(t, served, label+" never served the imported cert")
+	t.Logf("TLS handshake ok — HAProxy served the imported cert (CN=%s)", served.Subject.CommonName)
+
+	version, _, err := harness.FetchTLSPosture(pubIP, int(httpsPort), 5*time.Second)
+	require.NoError(t, err, "fetch TLS posture")
+	assert.GreaterOrEqualf(t, version, uint16(0x0303), "negotiated TLS version must be >= 1.2, got 0x%04x", version)
+
+	// End-to-end: one HTTPS GET that terminates TLS at the LB and forwards to a
+	// backend. Verify the chain against the imported cert (IP SAN matches pubIP).
+	pool := x509.NewCertPool()
+	require.True(t, pool.AppendCertsFromPEM(certPEM), "build CA pool from imported cert")
+	url := fmt.Sprintf("https://%s:%d/index.html", pubIP, httpsPort)
+	harness.EventuallyErr(t, func() error {
+		status, body, gerr := harness.HTTPSGet(url, pool, 10*time.Second)
+		if gerr != nil {
+			return gerr
+		}
+		if status != http.StatusOK {
+			return fmt.Errorf("status %d (body=%q)", status, string(body))
+		}
+		return nil
+	}, 90*time.Second, 3*time.Second)
+	t.Logf("HTTPS GET %s -> 200 (TLS terminated at LB, forwarded to backend)", url)
+}
+
+// createHTTPSListener creates an HTTPS listener referencing certArn. Distinct
+// from createListener because the ELBv2 handler requires Certificates for HTTPS
+// and rejects them for HTTP.
+func createHTTPSListener(t *testing.T, c *harness.AWSClient, lbArn string, port int64, tgArn, certArn string) string {
+	t.Helper()
+	// e2e:allow-create — the HTTPS listener is the subject under test (ACM cert -> termination).
+	out, err := c.ELBv2.CreateListener(&elbv2.CreateListenerInput{
+		LoadBalancerArn: aws.String(lbArn),
+		Protocol:        aws.String("HTTPS"),
+		Port:            aws.Int64(port),
+		Certificates:    []*elbv2.Certificate{{CertificateArn: aws.String(certArn)}},
+		DefaultActions: []*elbv2.Action{{
+			Type:           aws.String("forward"),
+			TargetGroupArn: aws.String(tgArn),
+		}},
+	})
+	require.NoError(t, err, "create-listener HTTPS")
+	require.NotEmpty(t, out.Listeners, "create-listener HTTPS returned no listeners")
+	arn := aws.StringValue(out.Listeners[0].ListenerArn)
+	t.Logf("listener HTTPS:%d -> %s (cert %s)", port, arn, certArn)
+	return arn
+}
+
+// authorizeSGPort idempotently opens proto/port from 0.0.0.0/0 on the shared
+// default SG, tolerating the Duplicate code on re-runs.
+func authorizeSGPort(t *testing.T, c *harness.AWSClient, f *sharedFixture, proto string, port int64) {
+	t.Helper()
+	_, err := c.EC2.AuthorizeSecurityGroupIngress(&ec2.AuthorizeSecurityGroupIngressInput{
+		GroupId: aws.String(f.SecurityGroup),
+		IpPermissions: []*ec2.IpPermission{{
+			IpProtocol: aws.String(proto),
+			FromPort:   aws.Int64(port),
+			ToPort:     aws.Int64(port),
+			IpRanges:   []*ec2.IpRange{{CidrIp: aws.String("0.0.0.0/0")}},
+		}},
+	})
+	if err != nil {
+		var aerr awserr.Error
+		if !errors.As(err, &aerr) || aerr.Code() != "InvalidPermission.Duplicate" {
+			t.Fatalf("authorize %s/%d: %v", proto, port, err)
+		}
+	}
+}
+
 // runUDPNLBSuite exercises an internal NLB with a UDP listener. The lb-agent probes
 // targets over TCP/9000 (nginx stream has no active health-check); once healthy, a
 // UDP datagram round-trip through the VIP proves the L4 data plane.

tests/e2e/multinode/jetstream_replicas_test.go

@@ -0,0 +1,44 @@
+//go:build e2e
+
+package multinode
+
+import (
+	"strings"
+	"testing"
+
+	"github.com/mulgadc/spinifex/spinifex/daemon"
+	"github.com/mulgadc/spinifex/tests/e2e/harness"
+	"github.com/stretchr/testify/require"
+)
+
+// jetStreamReplicaBuckets are the daemon-owned KV buckets created at replicas=1
+// during initJetStream and bumped to the cluster size by upgradeJetStreamReplicas.
+// A regression there leaves them stuck at 1, silently losing EC2/cluster state
+// on a single node loss — exactly what this guards (the upgrade path is 0% unit).
+var jetStreamReplicaBuckets = []string{
+	daemon.InstanceStateBucket,
+	daemon.ClusterStateBucket,
+	daemon.TerminatedInstanceBucket,
+}
+
+// runJetStreamReplicas asserts every daemon KV bucket's configured replica factor
+// matches the cluster node count, proving upgradeJetStreamReplicas ran end-to-end
+// (the assembled NATS path no unit test reaches). Read-only over NATS.
+func runJetStreamReplicas(t *testing.T, fix *Fixture) {
+	harness.Phase(t, "Multinode — JetStream KV Replica Factor")
+
+	want := len(fix.Cluster.Nodes)
+	if want < 2 {
+		t.Skipf("cluster has %d node(s); replica upgrade is a no-op below 2", want)
+	}
+	harness.Detail(t, "cluster_nodes", want, "buckets", strings.Join(jetStreamReplicaBuckets, ","))
+
+	harness.Step(t, "introspect KV stream replica factors")
+	got := harness.KVReplicaFactors(t, fix.Env, jetStreamReplicaBuckets...)
+	for _, b := range jetStreamReplicaBuckets {
+		harness.Detail(t, "bucket", b, "replicas", got[b])
+		require.Equalf(t, want, got[b],
+			"KV bucket %s replicas=%d want=%d — upgradeJetStreamReplicas did not match cluster size; a single node loss would lose this bucket",
+			b, got[b], want)
+	}
+}

tests/e2e/multinode/tests_test.go

@@ -34,6 +34,13 @@ func TestMultinodeInstanceDistribution(t *testing.T) {
 	runInstanceDistribution(t, requireMultiNodeFixture(t))
 }
 
+// TestMultinodeJetStreamReplicas is read-only over NATS; parallel so it resumes
+// after the sequential node-failure/recovery tests have restabilised the cluster.
+func TestMultinodeJetStreamReplicas(t *testing.T) {
+	t.Parallel()
+	runJetStreamReplicas(t, requireMultiNodeFixture(t))
+}
+
 // TestMultinodeVolumeLifecycle is sequential: touches predastore state.
 func TestMultinodeVolumeLifecycle(t *testing.T) {
 	runVolumeLifecycle(t, requireMultiNodeFixture(t))

tests/e2e/single/eip_test.go

@@ -0,0 +1,159 @@
+//go:build e2e
+
+package single
+
+import (
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/ec2"
+	"github.com/mulgadc/spinifex/tests/e2e/harness"
+	"github.com/stretchr/testify/require"
+)
+
+// runInstanceEIP exercises the instance-level Elastic IP datapath end to end:
+// AllocateAddress -> AssociateAddress to a running instance -> reach the guest
+// over SSH via the EIP -> DisassociateAddress -> assert the EIP no longer
+// reaches the guest -> ReleaseAddress. Proves the OVN DNAT/SNAT flows track the
+// association lifecycle, not just the control-plane record.
+//
+// Uses a dedicated throwaway instance, never the shared singleton: associating
+// an EIP rewrites the instance's public IP, so reusing the singleton would
+// corrupt its public-IP datapath for every downstream test. Pool-mode + OVN
+// gated — the EIP needs external IPAM and OVN flows to be observable.
+func runInstanceEIP(t *testing.T, fix *Fixture) {
+	if !fix.PoolMode {
+		t.Skip("instance EIP datapath requires pool-mode networking (external IPAM)")
+	}
+	harness.Phase(t, "Single — Instance Elastic IP Datapath")
+	harness.SkipIfNoOVN(t)
+
+	c := fix.AWS
+	amiID := needAMI(t, fix)
+	instType, _ := needInstanceTypeArch(t, fix)
+	keyName, keyPath := needKeyPair(t, fix)
+
+	def := harness.EnsureDefaultVPC(t, fix.Harness)
+	require.NotEmpty(t, def.SubnetID, "default subnet ID required")
+	require.NotEmpty(t, def.SGID, "default SG ID required")
+	harness.AuthorizeSSHIngress(t, c, def.SGID)
+
+	// --- Throwaway EIP subject --------------------------------------------
+	harness.Step(t, "run-instances (EIP subject)")
+	// e2e:allow-create — dedicated throwaway subject; reusing the singleton would corrupt its public IP.
+	runOut, err := c.EC2.RunInstances(&ec2.RunInstancesInput{
+		ImageId:      aws.String(amiID),
+		InstanceType: aws.String(instType),
+		KeyName:      aws.String(keyName),
+		SubnetId:     aws.String(def.SubnetID),
+		MinCount:     aws.Int64(1),
+		MaxCount:     aws.Int64(1),
+	})
+	require.NoError(t, err, "run-instances eip subject")
+	require.NotEmpty(t, runOut.Instances, "run-instances returned no Instances")
+	instID := aws.StringValue(runOut.Instances[0].InstanceId)
+	require.NotEmpty(t, instID, "InstanceId empty")
+	t.Cleanup(func() {
+		_, _ = c.EC2.TerminateInstances(&ec2.TerminateInstancesInput{
+			InstanceIds: []*string{aws.String(instID)},
+		})
+		_ = waitForInstanceStateSoft(c, instID, "terminated", 5*time.Minute)
+	})
+
+	inst := harness.WaitForInstanceState(t, c, instID, "running")
+	privIP := aws.StringValue(inst.PrivateIpAddress)
+	require.NotEmpty(t, privIP, "EIP subject has no PrivateIpAddress")
+	harness.Detail(t, "instance", instID, "boot_public_ip", aws.StringValue(inst.PublicIpAddress))
+
+	// --- AllocateAddress ---------------------------------------------------
+	harness.Step(t, "allocate-address (vpc)")
+	// e2e:allow-create — the Elastic IP is the resource under test.
+	eipOut, err := c.EC2.AllocateAddress(&ec2.AllocateAddressInput{Domain: aws.String("vpc")})
+	require.NoError(t, err, "allocate-address")
+	allocID := aws.StringValue(eipOut.AllocationId)
+	eipIP := aws.StringValue(eipOut.PublicIp)
+	require.NotEmpty(t, allocID, "AllocationId empty")
+	require.NotEmpty(t, eipIP, "EIP PublicIp empty")
+	eipReleased := false
+	t.Cleanup(func() {
+		if eipReleased {
+			return
+		}
+		_, _ = c.EC2.ReleaseAddress(&ec2.ReleaseAddressInput{AllocationId: aws.String(allocID)})
+	})
+	harness.Detail(t, "eip", eipIP, "alloc", allocID)
+
+	// --- AssociateAddress --------------------------------------------------
+	harness.Step(t, "associate-address %s -> %s", eipIP, instID)
+	assocOut, err := c.EC2.AssociateAddress(&ec2.AssociateAddressInput{
+		AllocationId: aws.String(allocID),
+		InstanceId:   aws.String(instID),
+	})
+	require.NoError(t, err, "associate-address")
+	assocID := aws.StringValue(assocOut.AssociationId)
+	require.NotEmpty(t, assocID, "AssociationId empty")
+	disassociated := false
+	t.Cleanup(func() {
+		if disassociated {
+			return
+		}
+		_, _ = c.EC2.DisassociateAddress(&ec2.DisassociateAddressInput{AssociationId: aws.String(assocID)})
+	})
+	harness.Detail(t, "assoc", assocID)
+
+	// --- Datapath: reach the guest via the EIP ----------------------------
+	// OVN needs a beat to install the DNAT/SNAT flows after the 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) {
+		harness.DumpVPCFlowDiagnostics(t, c, instID,
+			fmt.Sprintf("EIP SSH timeout — eip=%s instance=%s", eipIP, instID),
+			harness.VPCDiagnosticsOpts{
+				ExternalIP:  eipIP,
+				LogicalIP:   privIP,
+				ArtifactDir: fix.ArtifactDir(t),
+			})
+		t.Fatalf("guest unreachable via EIP %s within %s (see diagnostics above)", eipIP, sshReadyBudget)
+	}
+	tgt := harness.SSHTarget{User: "ec2-user", Host: eipIP, Port: 22, KeyPath: keyPath}
+	idOut := runSSH(t, tgt, "id")
+	require.Containsf(t, idOut, "ec2-user", "ssh via EIP id did not report ec2-user\n%s", idOut)
+	harness.Detail(t, "datapath", "eip_reachable_ok")
+
+	// --- DisassociateAddress + assert unreachable -------------------------
+	harness.Step(t, "disassociate-address %s", assocID)
+	_, err = c.EC2.DisassociateAddress(&ec2.DisassociateAddressInput{AssociationId: aws.String(assocID)})
+	require.NoError(t, err, "disassociate-address")
+	disassociated = true
+
+	// The EIP no longer maps to the guest, but DNAT teardown can lag and a single
+	// transient SSH error must not be read as teardown — require several
+	// consecutive unreachable probes before declaring success.
+	harness.Step(t, "verify guest unreachable via EIP %s after disassociate", eipIP)
+	const wantUnreachable = 3
+	consecutive := 0
+	harness.EventuallyErr(t, func() error {
+		if _, sErr := runSSHQuiet(tgt, "true"); sErr == nil {
+			consecutive = 0
+			return fmt.Errorf("EIP %s still reaches guest after disassociate", eipIP)
+		}
+		if consecutive++; consecutive < wantUnreachable {
+			return fmt.Errorf("EIP %s unreachable %d/%d — confirming it is not a transient blip",
+				eipIP, consecutive, wantUnreachable)
+		}
+		return nil
+	}, 90*time.Second, 5*time.Second)
+
+	// Rule out an instance-death false pass: the unreachability must be the EIP
+	// teardown, not the subject VM having gone away mid-poll.
+	harness.WaitForInstanceState(t, c, instID, "running", harness.WithTimeout(30*time.Second))
+	harness.Detail(t, "datapath", "eip_unreachable_after_disassociate_ok")
+
+	// --- ReleaseAddress ----------------------------------------------------
+	harness.Step(t, "release-address %s", allocID)
+	_, err = c.EC2.ReleaseAddress(&ec2.ReleaseAddressInput{AllocationId: aws.String(allocID)})
+	require.NoError(t, err, "release-address")
+	eipReleased = true
+}

tests/e2e/single/eni_test.go

@@ -0,0 +1,262 @@
+//go:build e2e
+
+package single
+
+import (
+	"fmt"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/ec2"
+	"github.com/mulgadc/spinifex/tests/e2e/harness"
+	"github.com/stretchr/testify/require"
+)
+
+// runENIHotplug exercises the secondary-ENI hotplug datapath end to end:
+// CreateNetworkInterface in the singleton's subnet, AttachNetworkInterface at
+// DeviceIndex 1 to the running guest, then assert the new virtio-net NIC
+// surfaces inside the guest kernel carrying the ENI's MAC. Proves the QMP
+// device_add path (gateway -> ec2.cmd.<id> daemon -> QEMU hot-plug) reaches
+// the guest, not just the control plane. OVN-gated; reuses the singleton VM
+// and restores it (detach + delete) before returning.
+func runENIHotplug(t *testing.T, fix *Fixture) {
+	harness.Phase(t, "Single — Secondary ENI Hotplug Datapath (OVN)")
+	harness.SkipIfNoOVN(t)
+	requireSSHHealthy(t)
+
+	c := fix.AWS
+	inst, _ := needInstance(t, fix)
+	instanceID := aws.StringValue(inst.InstanceId)
+	_, keyPath := needKeyPair(t, fix)
+
+	def := harness.EnsureDefaultVPC(t, fix.Harness)
+	require.NotEmpty(t, def.SubnetID, "default subnet ID required")
+	require.NotEmpty(t, def.SGID, "default SG ID required")
+	harness.AuthorizeSSHIngress(t, c, def.SGID)
+
+	// SSH to the singleton via its public IP — the in-guest `ip link` snapshot
+	// needs a live datapath. Fatal (with sticky-skip) if the handshake never
+	// completes; an unreachable singleton is a datapath bug, not test flake.
+	host, port := harness.InstancePublicSSHHost(t, inst)
+	waitForSSHReady(t, host, port, keyPath)
+	tgt := harness.SSHTarget{User: "ec2-user", Host: host, Port: port, KeyPath: keyPath}
+
+	// Snapshot the guest's MAC inventory before the hotplug so we detect the
+	// brand-new interface rather than match a NIC that already existed.
+	before := guestMACSet(t, tgt)
+	harness.Detail(t, "guest_macs_before", strings.Join(setKeys(before), ","))
+
+	// --- CreateNetworkInterface -------------------------------------------
+	harness.Step(t, "create-network-interface subnet=%s", def.SubnetID)
+	// e2e:allow-create — the secondary ENI is the subject hot-plugged under test.
+	cniOut, err := c.EC2.CreateNetworkInterface(&ec2.CreateNetworkInterfaceInput{
+		SubnetId:    aws.String(def.SubnetID),
+		Groups:      []*string{aws.String(def.SGID)},
+		Description: aws.String("e2e-eni-hotplug"),
+	})
+	require.NoError(t, err, "create-network-interface")
+	require.NotNil(t, cniOut.NetworkInterface, "create-network-interface returned nil ENI")
+	eniID := aws.StringValue(cniOut.NetworkInterface.NetworkInterfaceId)
+	eniMAC := strings.ToLower(aws.StringValue(cniOut.NetworkInterface.MacAddress))
+	eniIP := aws.StringValue(cniOut.NetworkInterface.PrivateIpAddress)
+	require.NotEmpty(t, eniID, "ENI NetworkInterfaceId empty")
+	require.NotEmpty(t, eniMAC, "ENI MacAddress empty")
+	require.NotEmpty(t, eniIP, "ENI PrivateIpAddress empty")
+	harness.Detail(t, "eni", eniID, "eni_mac", eniMAC, "eni_ip", eniIP)
+
+	// Delete cleanup runs last (LIFO); the detach cleanup registered next must
+	// flip the ENI back to "available" before this fires.
+	t.Cleanup(func() {
+		if _, derr := c.EC2.DeleteNetworkInterface(&ec2.DeleteNetworkInterfaceInput{
+			NetworkInterfaceId: aws.String(eniID),
+		}); derr != nil && !harness.ErrorCodeIs(derr, "InvalidNetworkInterfaceID.NotFound") {
+			t.Logf("WARNING: cleanup delete ENI %s: %v", eniID, derr)
+		}
+	})
+
+	// --- AttachNetworkInterface (hot-plug) --------------------------------
+	harness.Step(t, "attach-network-interface %s -> %s device-index=1", eniID, instanceID)
+	attachOut, err := c.EC2.AttachNetworkInterface(&ec2.AttachNetworkInterfaceInput{
+		NetworkInterfaceId: aws.String(eniID),
+		InstanceId:         aws.String(instanceID),
+		DeviceIndex:        aws.Int64(1),
+	})
+	require.NoError(t, err, "attach-network-interface")
+	attachmentID := aws.StringValue(attachOut.AttachmentId)
+	require.NotEmpty(t, attachmentID, "AttachmentId empty")
+	detached := false
+	t.Cleanup(func() {
+		if detached {
+			return
+		}
+		if _, derr := c.EC2.DetachNetworkInterface(&ec2.DetachNetworkInterfaceInput{
+			AttachmentId: aws.String(attachmentID),
+		}); derr != nil && !harness.ErrorCodeIs(derr, "InvalidAttachmentID.NotFound") {
+			t.Logf("WARNING: cleanup detach ENI %s: %v", eniID, derr)
+		}
+		_ = waitForENIStatusSoft(c, eniID, "available", 2*time.Minute)
+	})
+	harness.Detail(t, "attachment", attachmentID)
+
+	// --- Control plane: ENI shows attached to the singleton ---------------
+	harness.Step(t, "wait ENI %s in-use attached to %s", eniID, instanceID)
+	waitForENIAttached(t, c, eniID, instanceID)
+
+	// --- Datapath: new NIC surfaces in the guest carrying the ENI MAC -----
+	harness.Step(t, "assert guest NIC with MAC %s appears", eniMAC)
+	if _, existed := before[eniMAC]; existed {
+		t.Fatalf("ENI MAC %s was already present before attach — cannot prove a fresh hotplug", eniMAC)
+	}
+	harness.EventuallyErr(t, func() error {
+		now := guestMACSet(t, tgt)
+		if _, ok := now[eniMAC]; ok {
+			return nil
+		}
+		return fmt.Errorf("ENI MAC %s not yet visible in guest (have: %s)",
+			eniMAC, strings.Join(setKeys(now), ","))
+	}, 90*time.Second, 3*time.Second)
+	harness.Detail(t, "datapath", "guest_nic_hotplugged_ok")
+
+	// Cross-check the control-plane secondary IP is the one assigned to the
+	// hotplugged ENI. In-guest IP autoconfiguration of the secondary NIC is
+	// OS/cloud-init dependent, so the IP assertion stays at the API level.
+	eni, err := describeENI(c, eniID)
+	require.NoError(t, err, "describe ENI %s after attach", eniID)
+	require.Equal(t, eniIP, aws.StringValue(eni.PrivateIpAddress),
+		"attached ENI %s private IP drifted from create-time value", eniID)
+
+	// --- Detach + assert the NIC drops out of the guest -------------------
+	harness.Step(t, "detach-network-interface %s", attachmentID)
+	_, err = c.EC2.DetachNetworkInterface(&ec2.DetachNetworkInterfaceInput{
+		AttachmentId: aws.String(attachmentID),
+	})
+	require.NoError(t, err, "detach-network-interface")
+	waitForENIStatus(t, c, eniID, "available")
+	detached = true
+
+	// Guest NIC removal is a soft check: virtio-net unplug + udev cleanup can
+	// lag the control-plane detach, and a lingering link doesn't break the EC2
+	// contract the way a missing hotplug would. Poll briefly, warn if it stays.
+	harness.Step(t, "verify guest NIC for MAC %s disappears", eniMAC)
+	dropped := false
+	deadline := time.Now().Add(45 * time.Second)
+	for time.Now().Before(deadline) {
+		if _, ok := guestMACSet(t, tgt)[eniMAC]; !ok {
+			dropped = true
+			break
+		}
+		time.Sleep(3 * time.Second)
+	}
+	if dropped {
+		harness.Detail(t, "datapath", "guest_nic_removed_ok")
+	} else {
+		t.Logf("WARN: guest NIC for MAC %s still present after detach (udev lag?)", eniMAC)
+	}
+
+	// --- DeleteNetworkInterface -------------------------------------------
+	harness.Step(t, "delete-network-interface %s", eniID)
+	_, err = c.EC2.DeleteNetworkInterface(&ec2.DeleteNetworkInterfaceInput{
+		NetworkInterfaceId: aws.String(eniID),
+	})
+	require.NoError(t, err, "delete-network-interface")
+}
+
+// guestMACSet returns the set of lowercased MAC addresses currently present on
+// the guest's network interfaces, read from sysfs so the list tracks the live
+// kernel view (including hotplugged NICs). The all-zero loopback MAC is
+// dropped so it never masks a real match.
+func guestMACSet(t *testing.T, tgt harness.SSHTarget) map[string]struct{} {
+	t.Helper()
+	out, err := harness.GuestExec(tgt, "cat /sys/class/net/*/address")
+	if err != nil {
+		t.Fatalf("guest read MAC inventory: %v\n%s", err, out)
+	}
+	set := map[string]struct{}{}
+	for _, line := range strings.Split(strings.TrimSpace(out), "\n") {
+		mac := strings.ToLower(strings.TrimSpace(line))
+		if mac == "" || mac == "00:00:00:00:00:00" {
+			continue
+		}
+		set[mac] = struct{}{}
+	}
+	return set
+}
+
+// setKeys returns the keys of a string set as a slice for log/Detail output.
+func setKeys(m map[string]struct{}) []string {
+	keys := make([]string, 0, len(m))
+	for k := range m {
+		keys = append(keys, k)
+	}
+	return keys
+}
+
+// describeENI returns the single ENI record for eniID, erroring if absent.
+func describeENI(c *harness.AWSClient, eniID string) (*ec2.NetworkInterface, error) {
+	out, err := c.EC2.DescribeNetworkInterfaces(&ec2.DescribeNetworkInterfacesInput{
+		NetworkInterfaceIds: []*string{aws.String(eniID)},
+	})
+	if err != nil {
+		return nil, fmt.Errorf("describe-network-interfaces %s: %w", eniID, err)
+	}
+	if len(out.NetworkInterfaces) == 0 {
+		return nil, fmt.Errorf("ENI %s not found", eniID)
+	}
+	return out.NetworkInterfaces[0], nil
+}
+
+// waitForENIAttached polls until eniID reports Status=in-use with an
+// Attachment bound to instanceID.
+func waitForENIAttached(t *testing.T, c *harness.AWSClient, eniID, instanceID string) {
+	t.Helper()
+	harness.EventuallyErr(t, func() error {
+		eni, err := describeENI(c, eniID)
+		if err != nil {
+			return err
+		}
+		status := aws.StringValue(eni.Status)
+		if eni.Attachment == nil {
+			return fmt.Errorf("ENI %s status=%s has no attachment yet", eniID, status)
+		}
+		if got := aws.StringValue(eni.Attachment.InstanceId); got != instanceID {
+			return fmt.Errorf("ENI %s attached to %q want %q", eniID, got, instanceID)
+		}
+		if status != "in-use" {
+			return fmt.Errorf("ENI %s status=%s want in-use", eniID, status)
+		}
+		return nil
+	}, 90*time.Second, 3*time.Second)
+}
+
+// waitForENIStatus polls until eniID reaches target status (e.g. "available").
+func waitForENIStatus(t *testing.T, c *harness.AWSClient, eniID, target string) {
+	t.Helper()
+	harness.EventuallyErr(t, func() error {
+		eni, err := describeENI(c, eniID)
+		if err != nil {
+			return err
+		}
+		if got := aws.StringValue(eni.Status); got != target {
+			return fmt.Errorf("ENI %s status=%s want=%s", eniID, got, target)
+		}
+		return nil
+	}, 90*time.Second, 3*time.Second)
+}
+
+// waitForENIStatusSoft is the cleanup-time variant of waitForENIStatus: never
+// calls t.Fatal, just polls best-effort within timeout.
+func waitForENIStatusSoft(c *harness.AWSClient, eniID, target string, timeout time.Duration) error {
+	deadline := time.Now().Add(timeout)
+	for {
+		eni, err := describeENI(c, eniID)
+		if err == nil && aws.StringValue(eni.Status) == target {
+			return nil
+		}
+		if time.Now().After(deadline) {
+			return fmt.Errorf("ENI %s did not reach %s within %s", eniID, target, timeout)
+		}
+		time.Sleep(2 * time.Second)
+	}
+}

tests/e2e/single/tests_test.go

@@ -60,6 +60,13 @@ func TestRouteTableValidation(t *testing.T) {
 	runRouteTableValidation(t, requireSingleNodeFixture(t))
 }
 
+// TestReplaceRouteConvergence owns its own scratch VPCs/IGWs end to end and
+// never touches the singleton or default VPC, so it is parallel-safe.
+func TestReplaceRouteConvergence(t *testing.T) {
+	t.Parallel()
+	runReplaceRouteConvergence(t, requireSingleNodeFixture(t))
+}
+
 // --- Sequential: singleton VM lifecycle ---
 
 // TestClusterStatsCLI asserts 0-VM baseline state before any instance launches.
@@ -101,6 +108,13 @@ func TestConsoleOutput(t *testing.T) {
 	runConsoleOutput(t, requireSingleNodeFixture(t))
 }
 
+// TestENIHotplug hot-plugs a secondary ENI onto the freshly-booted singleton
+// and asserts the NIC reaches the guest, then restores it. Placed before the
+// stop/start churn so the singleton is known running + SSH-healthy.
+func TestENIHotplug(t *testing.T) {
+	runENIHotplug(t, requireSingleNodeFixture(t))
+}
+
 func TestVolumeLifecycle(t *testing.T) {
 	runVolumeLifecycle(t, requireSingleNodeFixture(t))
 }
@@ -155,6 +169,13 @@ func TestNATGateway(t *testing.T) {
 	runNATGateway(t, requireSingleNodeFixture(t))
 }
 
+// TestInstanceEIP allocates an Elastic IP, associates it to a throwaway VM,
+// and asserts the EIP datapath flips on/off with the association. Sequential:
+// it authorizes ingress on the shared default SG.
+func TestInstanceEIP(t *testing.T) {
+	runInstanceEIP(t, requireSingleNodeFixture(t))
+}
+
 func TestSGToSGDatapath(t *testing.T) {
 	runSGToSGDatapath(t, requireSingleNodeFixture(t))
 }

tests/e2e/single/vpc_test.go

@@ -573,3 +573,191 @@ func runRouteTableValidation(t *testing.T, fix *Fixture) {
 		}
 	})
 }
+
+// runReplaceRouteConvergence validates ReplaceRoute's control-plane contract:
+// a route installed against one IGW target re-points to a different IGW, with
+// the route record converging to the new gateway. It then walks the negative
+// matrix (no-such-route, local-route, missing-target, unknown-IGW, cross-VPC
+// IGW) so each rejection path stays pinned to its AWS error code.
+//
+// V1 scope note: ReplaceRoute updates the route record but publishes no vpc
+// event, so it does NOT reprogram the OVN datapath. There is no datapath to
+// assert here — the contract under test is control-plane convergence only.
+// When ReplaceRoute begins emitting add/delete-igw-route events, extend this
+// with a guest-egress assertion mirroring runNATGateway.
+//
+// Owns two scratch VPCs + three IGWs end to end, so it never touches the
+// singleton or default VPC and is safe in the parallel bucket. No pool mode or
+// OVN needed — every step is pure KV control plane.
+func runReplaceRouteConvergence(t *testing.T, fix *Fixture) {
+	harness.Phase(t, "Single — ReplaceRoute Convergence (control plane)")
+
+	c := fix.AWS
+
+	// --- Primary VPC + two attached IGWs -----------------------------------
+	vpcID := createScratchVPC(t, c, "10.77.0.0/16")
+	igw1 := createAttachedIGW(t, c, vpcID)
+	igw2 := createAttachedIGW(t, c, vpcID)
+
+	// --- Secondary VPC + its own IGW (cross-VPC rejection target) ----------
+	otherVPCID := createScratchVPC(t, c, "10.88.0.0/16")
+	igwOther := createAttachedIGW(t, c, otherVPCID)
+
+	// --- Custom route table in the primary VPC -----------------------------
+	harness.Step(t, "create-route-table (vpc=%s)", vpcID)
+	// e2e:allow-create — scratch route table owned by this ReplaceRoute test.
+	rtOut, err := c.EC2.CreateRouteTable(&ec2.CreateRouteTableInput{VpcId: aws.String(vpcID)})
+	require.NoError(t, err, "create-route-table")
+	require.NotNil(t, rtOut.RouteTable, "create-route-table returned nil RouteTable")
+	rtbID := aws.StringValue(rtOut.RouteTable.RouteTableId)
+	require.NotEmpty(t, rtbID, "RouteTableId empty")
+	t.Cleanup(func() {
+		_, _ = c.EC2.DeleteRouteTable(&ec2.DeleteRouteTableInput{RouteTableId: aws.String(rtbID)})
+	})
+
+	const dest = "192.0.2.0/24"
+
+	// --- CreateRoute -> igw1 -----------------------------------------------
+	harness.Step(t, "create-route %s -> %s", dest, igw1)
+	_, err = c.EC2.CreateRoute(&ec2.CreateRouteInput{
+		RouteTableId:         aws.String(rtbID),
+		DestinationCidrBlock: aws.String(dest),
+		GatewayId:            aws.String(igw1),
+	})
+	require.NoError(t, err, "create-route %s -> igw1", dest)
+	require.Equalf(t, igw1, routeGateway(t, c, rtbID, dest),
+		"route %s should target igw1 after create", dest)
+
+	// --- ReplaceRoute -> igw2 (genuine target convergence) -----------------
+	harness.Step(t, "replace-route %s -> %s", dest, igw2)
+	_, err = c.EC2.ReplaceRoute(&ec2.ReplaceRouteInput{
+		RouteTableId:         aws.String(rtbID),
+		DestinationCidrBlock: aws.String(dest),
+		GatewayId:            aws.String(igw2),
+	})
+	require.NoError(t, err, "replace-route %s -> igw2", dest)
+	require.Equalf(t, igw2, routeGateway(t, c, rtbID, dest),
+		"route %s did not converge to igw2 after replace-route", dest)
+	harness.Detail(t, "convergence", dest+": "+igw1+" -> "+igw2)
+
+	// --- Negative matrix ---------------------------------------------------
+	// Unknown destination CIDR — route doesn't exist in the table.
+	harness.Step(t, "replace-route on absent route (must fail)")
+	harness.ExpectError(t, "InvalidRoute.NotFound", func() error {
+		_, e := c.EC2.ReplaceRoute(&ec2.ReplaceRouteInput{
+			RouteTableId:         aws.String(rtbID),
+			DestinationCidrBlock: aws.String("198.51.100.0/24"),
+			GatewayId:            aws.String(igw2),
+		})
+		return e
+	})
+
+	// Local route is immutable.
+	harness.Step(t, "replace-route on local route (must fail)")
+	harness.ExpectError(t, "InvalidParameterValue", func() error {
+		_, e := c.EC2.ReplaceRoute(&ec2.ReplaceRouteInput{
+			RouteTableId:         aws.String(rtbID),
+			DestinationCidrBlock: aws.String("10.77.0.0/16"),
+			GatewayId:            aws.String(igw2),
+		})
+		return e
+	})
+
+	// Missing target — V1 requires a GatewayId.
+	harness.Step(t, "replace-route with no target (must fail)")
+	harness.ExpectError(t, "InvalidParameterValue", func() error {
+		_, e := c.EC2.ReplaceRoute(&ec2.ReplaceRouteInput{
+			RouteTableId:         aws.String(rtbID),
+			DestinationCidrBlock: aws.String(dest),
+		})
+		return e
+	})
+
+	// Unknown IGW target.
+	harness.Step(t, "replace-route to non-existent IGW (must fail)")
+	harness.ExpectError(t, "InvalidInternetGatewayID.NotFound", func() error {
+		_, e := c.EC2.ReplaceRoute(&ec2.ReplaceRouteInput{
+			RouteTableId:         aws.String(rtbID),
+			DestinationCidrBlock: aws.String(dest),
+			GatewayId:            aws.String("igw-00000000000000000"),
+		})
+		return e
+	})
+
+	// IGW attached to a different VPC than the route table.
+	harness.Step(t, "replace-route to cross-VPC IGW %s (must fail)", igwOther)
+	harness.ExpectError(t, "InvalidParameterValue", func() error {
+		_, e := c.EC2.ReplaceRoute(&ec2.ReplaceRouteInput{
+			RouteTableId:         aws.String(rtbID),
+			DestinationCidrBlock: aws.String(dest),
+			GatewayId:            aws.String(igwOther),
+		})
+		return e
+	})
+
+	// Each rejection left the route untouched at igw2.
+	require.Equalf(t, igw2, routeGateway(t, c, rtbID, dest),
+		"route %s drifted off igw2 after the rejection matrix", dest)
+}
+
+// createScratchVPC creates a VPC with the given CIDR and registers its deletion.
+func createScratchVPC(t *testing.T, c *harness.AWSClient, cidr string) string {
+	t.Helper()
+	harness.Step(t, "create-vpc %s", cidr)
+	// e2e:allow-create — scratch VPC owned end to end by the caller.
+	out, err := c.EC2.CreateVpc(&ec2.CreateVpcInput{CidrBlock: aws.String(cidr)})
+	require.NoError(t, err, "create-vpc %s", cidr)
+	require.NotNil(t, out.Vpc, "create-vpc %s returned nil Vpc", cidr)
+	id := aws.StringValue(out.Vpc.VpcId)
+	require.NotEmpty(t, id, "VpcId empty for %s", cidr)
+	t.Cleanup(func() {
+		_, _ = c.EC2.DeleteVpc(&ec2.DeleteVpcInput{VpcId: aws.String(id)})
+	})
+	return id
+}
+
+// createAttachedIGW creates an internet gateway, attaches it to vpcID, and
+// registers detach+delete cleanup (LIFO-safe: detach runs before the owning
+// VPC's delete).
+func createAttachedIGW(t *testing.T, c *harness.AWSClient, vpcID string) string {
+	t.Helper()
+	// e2e:allow-create — scratch IGW owned by the caller's ReplaceRoute test.
+	out, err := c.EC2.CreateInternetGateway(&ec2.CreateInternetGatewayInput{})
+	require.NoError(t, err, "create-internet-gateway")
+	require.NotNil(t, out.InternetGateway, "create-internet-gateway returned nil IGW")
+	igwID := aws.StringValue(out.InternetGateway.InternetGatewayId)
+	require.NotEmpty(t, igwID, "InternetGatewayId empty")
+	_, err = c.EC2.AttachInternetGateway(&ec2.AttachInternetGatewayInput{
+		InternetGatewayId: aws.String(igwID),
+		VpcId:             aws.String(vpcID),
+	})
+	require.NoError(t, err, "attach-internet-gateway %s -> %s", igwID, vpcID)
+	t.Cleanup(func() {
+		_, _ = c.EC2.DetachInternetGateway(&ec2.DetachInternetGatewayInput{
+			InternetGatewayId: aws.String(igwID),
+			VpcId:             aws.String(vpcID),
+		})
+		_, _ = c.EC2.DeleteInternetGateway(&ec2.DeleteInternetGatewayInput{
+			InternetGatewayId: aws.String(igwID),
+		})
+	})
+	return igwID
+}
+
+// routeGateway returns the GatewayId of the route matching destCidr in rtbID,
+// failing the test if the route is absent.
+func routeGateway(t *testing.T, c *harness.AWSClient, rtbID, destCidr string) string {
+	t.Helper()
+	out, err := c.EC2.DescribeRouteTables(&ec2.DescribeRouteTablesInput{
+		RouteTableIds: []*string{aws.String(rtbID)},
+	})
+	require.NoError(t, err, "describe-route-tables %s", rtbID)
+	require.NotEmpty(t, out.RouteTables, "route table %s not found", rtbID)
+	for _, r := range out.RouteTables[0].Routes {
+		if aws.StringValue(r.DestinationCidrBlock) == destCidr {
+			return aws.StringValue(r.GatewayId)
+		}
+	}
+	t.Fatalf("route %s not found in table %s", destCidr, rtbID)
+	return ""
+}