DockMV/test/ssh_e2e_test.go

//go:build e2e

package e2e

import (
	"context"
	"os"
	"path/filepath"
	"strings"
	"testing"
	"time"

	"github.com/arescom/docker-migrate/internal/dkr"
	"github.com/arescom/docker-migrate/internal/job"
	"github.com/arescom/docker-migrate/internal/migrate"
	"github.com/arescom/docker-migrate/internal/spec"
	"github.com/arescom/docker-migrate/internal/sshx"
	"github.com/docker/docker/api/types/container"
	"github.com/docker/docker/api/types/mount"
	"github.com/docker/docker/api/types/volume"
)

// TestSSHMigration drives the host-to-host engine over a real SSH connection.
//
// Source and target are the same daemon, reached over SSH, so the whole
// transport is exercised — ssh, gzip streaming, the target's docker CLI, the
// staging container for read-only mounts and the verify step — without needing
// a second machine. The container and its volume are renamed on the way in so
// the copy is genuinely verified rather than finding the data already there.
//
// Configure it with:
//
//	DM_SSH_HOST=10.0.0.5 DM_SSH_USER=root DM_SSH_KEY=/root/.ssh/id_ed25519 \
//	  go test -tags e2e ./test/... -run TestSSHMigration -v
const (
	sshSrcName   = "dmssh-src"
	sshSrcVolume = "dmssh-vol"
	sshDstName   = "dmssh-src-viassh"
	sshDstVolume = "dmssh-vol-viassh"
)

func TestSSHMigration(t *testing.T) {
	host := os.Getenv("DM_SSH_HOST")
	user := os.Getenv("DM_SSH_USER")
	key := os.Getenv("DM_SSH_KEY")
	if host == "" || user == "" || key == "" {
		t.Skip("set DM_SSH_HOST, DM_SSH_USER and DM_SSH_KEY to run the SSH migration test")
	}
	keyPEM, err := os.ReadFile(key)
	if err != nil {
		t.Fatalf("read %s: %v", key, err)
	}

	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Minute)
	defer cancel()

	c, err := dkr.New("")
	if err != nil {
		t.Skipf("no docker daemon available: %v", err)
	}
	defer c.Close()

	bindDir := t.TempDir()
	if err := os.WriteFile(filepath.Join(bindDir, "site.conf"), []byte("listen=8443\n"), 0o644); err != nil {
		t.Fatal(err)
	}

	sshCleanup(ctx, c)
	t.Cleanup(func() {
		cctx, ccancel := context.WithTimeout(context.Background(), 2*time.Minute)
		defer ccancel()
		sshCleanup(cctx, c)
	})

	pullImage(ctx, t, c, testImage)
	createSSHSource(ctx, t, c, bindDir)

	// --- connect ---------------------------------------------------------
	hkPath := filepath.Join(t.TempDir(), "known_hosts")
	hk, err := sshx.NewKnownHosts(hkPath)
	if err != nil {
		t.Fatal(err)
	}
	cfg := sshx.Config{
		Name: "loopback", Host: host, Port: 22, User: user,
		Auth: sshx.AuthKey, PrivateKey: string(keyPEM),
	}

	// An unknown host key must be refused before it is trusted; that is the
	// whole point of the trust store.
	if _, err := sshx.Dial(ctx, cfg, hk); err == nil {
		t.Fatal("dialling an untrusted host key must fail")
	} else {
		var hke *sshx.HostKeyError
		if !asHostKeyError(err, &hke) {
			t.Fatalf("expected a host key error, got %v", err)
		}
		t.Logf("host key presented: %s %s", hke.KeyType, hke.Fingerprint)
	}
	if err := sshx.TrustFromProbe(ctx, cfg, hk, ""); err != nil {
		t.Fatalf("trust host key: %v", err)
	}

	client, err := sshx.Dial(ctx, cfg, hk)
	if err != nil {
		t.Fatalf("dial after trusting: %v", err)
	}
	defer client.Close()

	rd := sshx.NewRemoteDocker(client)
	pre, err := rd.Preflight(ctx)
	if err != nil {
		t.Fatal(err)
	}
	t.Logf("target preflight: docker %s %s/%s gzip=%v free=%d problems=%v",
		pre.ServerVersion, pre.OS, pre.Arch, pre.HasGzip, pre.DiskFreeBytes, pre.Problems)
	if pre.ServerVersion == "" {
		t.Fatalf("target docker unusable: %v", pre.Problems)
	}

	// --- plan ------------------------------------------------------------
	inv, err := c.Inventory(ctx)
	if err != nil {
		t.Fatal(err)
	}
	var src *spec.Container
	for i := range inv.Containers {
		if inv.Containers[i].Name == sshSrcName {
			src = &inv.Containers[i]
		}
	}
	if src == nil {
		t.Fatalf("source container %s not found", sshSrcName)
	}

	sel := spec.DefaultSelection(src)
	sel.Include = true
	sel.NameOverride = sshDstName
	sel.ImageMode = spec.ImageSkip // same daemon; the image is already there
	sel.StopSourceAfter = false    // exercise the restart path
	sel.Mounts["/data"] = spec.MountSelection{Action: spec.MountActionCopy, TargetName: sshDstVolume}
	sel.MigratePorts = false // the source publishes nothing, but be explicit

	opts := spec.DefaultOptions()
	opts.Compress = true
	opts.VerifyAfter = true

	runner := &migrate.SSHRunner{
		Src: c, Dst: rd,
		Containers: inv.Containers, Volumes: inv.Volumes, Networks: inv.Networks,
		Plan: spec.Plan{Items: []spec.ItemSelection{sel}, Options: opts},
	}

	jobs := job.NewManager()
	j := jobs.Run(ctx, job.KindSSH, "ssh e2e", false, runner.Run)
	<-j.Done()

	snap := j.Snapshot()
	for _, l := range snap.Log {
		t.Logf("[%-5s] %s", l.Level, l.Message)
	}
	for _, it := range snap.Items {
		for _, st := range it.Steps {
			t.Logf("  step %-12s %-9s %8d bytes  %s", st.ID, st.State, st.BytesDone, st.Label)
		}
	}
	if snap.State != job.StateSucceeded {
		t.Fatalf("migration %s: %s", snap.State, snap.Error)
	}
	if snap.BytesDone == 0 {
		t.Error("no bytes were transferred; the copy did nothing")
	}

	// --- verify ----------------------------------------------------------
	checks := []struct{ path, want string }{
		{"/data/payload.txt", "streamed over ssh"},
		{"/data/sub/inner.txt", "inner payload"},
		{"/conf/site.conf", "listen=8443"},
		{"/anon/scratch.txt", "anonymous over ssh"},
	}
	for _, chk := range checks {
		got, err := readFileFromContainer(ctx, c, sshDstName, chk.path)
		if err != nil {
			t.Errorf("reading %s: %v", chk.path, err)
			continue
		}
		if !strings.Contains(got, chk.want) {
			t.Errorf("%s = %q, want it to contain %q", chk.path, got, chk.want)
		} else {
			t.Logf("verified %s", chk.path)
		}
	}

	// The renamed volume must be a genuinely new one, holding the copied data.
	if _, err := c.API().VolumeInspect(ctx, sshDstVolume); err != nil {
		t.Errorf("renamed volume %s was not created: %v", sshDstVolume, err)
	}

	// The source was asked to keep running.
	if st, err := c.State(ctx, sshSrcName); err != nil {
		t.Errorf("source state: %v", err)
	} else if st != "running" {
		t.Errorf("source container should have been restarted, state is %q", st)
	}

	// The staging container used for the read-only mount must be gone.
	out, err := rd.Run(ctx, "ps", "-a", "--format", "{{.Names}}")
	if err != nil {
		t.Fatal(err)
	}
	if strings.Contains(out, "dm-stage-") {
		t.Errorf("a staging container was left behind:\n%s", out)
	}
}

func createSSHSource(ctx context.Context, t *testing.T, c *dkr.Client, bindDir string) {
	t.Helper()
	api := c.API()
	if _, err := api.VolumeCreate(ctx, volume.CreateOptions{Name: sshSrcVolume}); err != nil {
		t.Fatal(err)
	}
	script := strings.Join([]string{
		"echo 'streamed over ssh' > /data/payload.txt",
		"mkdir -p /data/sub",
		"echo 'inner payload' > /data/sub/inner.txt",
		"echo 'anonymous over ssh' > /anon/scratch.txt",
		"sleep 3600",
	}, " && ")

	resp, err := api.ContainerCreate(ctx,
		&container.Config{
			Image: testImage,
			Cmd:   []string{"sh", "-c", script},
			Env:   []string{"DM_MODE=ssh"},
		},
		&container.HostConfig{
			Mounts: []mount.Mount{
				{Type: mount.TypeVolume, Source: sshSrcVolume, Target: "/data"},
				{Type: mount.TypeBind, Source: bindDir, Target: "/conf", ReadOnly: true},
				{Type: mount.TypeVolume, Target: "/anon"},
			},
		},
		nil, nil, sshSrcName)
	if err != nil {
		t.Fatal(err)
	}
	if err := api.ContainerStart(ctx, resp.ID, container.StartOptions{}); err != nil {
		t.Fatal(err)
	}
	time.Sleep(2 * time.Second)
}

func sshCleanup(ctx context.Context, c *dkr.Client) {
	api := c.API()
	for _, n := range []string{sshSrcName, sshDstName} {
		_ = api.ContainerRemove(ctx, n, container.RemoveOptions{Force: true})
	}
	for _, v := range []string{sshSrcVolume, sshDstVolume} {
		_ = api.VolumeRemove(ctx, v, true)
	}
}