worker/internal/installer/sourceinstall_ssh_test.go

package installer

import (
	"crypto/ed25519"
	"crypto/rand"
	"fmt"
	"io"
	"net"
	"strings"
	"sync"
	"testing"
	"time"

	"golang.org/x/crypto/ssh"

	"rocketgit.ru/rsmon/worker/internal/sshinstall"
)

const (
	fakeSSHPassword = "fake-ssh-password"
	fakeCommitHex   = "deadbeefdeadbeefdeadbeefdeadbeefdeadbeef"
	fakeOsRelease   = "ID=ubuntu\nNAME=\"Ubuntu\"\nVERSION_ID=24.04\n"
)

// fakeSSHServer is a minimal in-process SSH server that simulates a
// remote Linux host for the SourceInstall orchestration tests. It uses
// real golang.org/x/crypto/ssh transport (no mocked SSH library), so the
// executor's dial, session, exec, and stdout/stderr plumbing is
// exercised end to end, and it records every command it ran.
type fakeSSHServer struct {
	addr     string
	onExec   func(command string) (stdout, stderr string, code int)
	mu       sync.Mutex
	commands []string
}

func startFakeSSHServer(t *testing.T, onExec func(command string) (string, string, int)) *fakeSSHServer {
	t.Helper()
	_, priv, err := ed25519.GenerateKey(rand.Reader)
	if err != nil {
		t.Fatal(err)
	}
	signer, err := ssh.NewSignerFromKey(priv)
	if err != nil {
		t.Fatal(err)
	}
	config := &ssh.ServerConfig{
		PasswordCallback: func(_ ssh.ConnMetadata, pass []byte) (*ssh.Permissions, error) {
			if string(pass) == fakeSSHPassword {
				return nil, nil
			}
			return nil, fmt.Errorf("password rejected")
		},
	}
	config.AddHostKey(signer)

	ln, err := net.Listen("tcp", "127.0.0.1:0")
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = ln.Close() })

	srv := &fakeSSHServer{addr: ln.Addr().String(), onExec: onExec}
	go func() {
		for {
			conn, err := ln.Accept()
			if err != nil {
				return
			}
			go srv.handleConn(conn, config)
		}
	}()
	return srv
}

func (s *fakeSSHServer) Port() int {
	_, port, err := net.SplitHostPort(s.addr)
	if err != nil {
		return 0
	}
	p := 0
	fmt.Sscanf(port, "%d", &p)
	return p
}

func (s *fakeSSHServer) Commands() []string {
	s.mu.Lock()
	defer s.mu.Unlock()
	return append([]string(nil), s.commands...)
}

func (s *fakeSSHServer) handleConn(conn net.Conn, config *ssh.ServerConfig) {
	sconn, chans, reqs, err := ssh.NewServerConn(conn, config)
	if err != nil {
		return
	}
	defer sconn.Close() //nolint:errcheck
	go ssh.DiscardRequests(reqs)
	for newChannel := range chans {
		if newChannel.ChannelType() != "session" {
			_ = newChannel.Reject(ssh.UnknownChannelType, "unknown channel type")
			continue
		}
		channel, requests, err := newChannel.Accept()
		if err != nil {
			continue
		}
		go func() {
			defer channel.Close()
			// Drain the client's stdin so uploads (which stream base64
			// over the session stdin) never block the channel window.
			// The channel is not closed until the client's write side is
			// exhausted, mirroring the real server behavior.
			var drained sync.WaitGroup
			drained.Add(1)
			go func() {
				defer drained.Done()
				_, _ = io.Copy(io.Discard, channel)
			}()
			s.handleSession(channel, requests)
			drained.Wait()
		}()
	}
}

func (s *fakeSSHServer) handleSession(channel ssh.Channel, requests <-chan *ssh.Request) {
	for req := range requests {
		if req.Type != "exec" {
			if req.WantReply {
				_ = req.Reply(false, nil)
			}
			continue
		}
		var payload struct{ Command string }
		if err := ssh.Unmarshal(req.Payload, &payload); err != nil {
			if req.WantReply {
				_ = req.Reply(false, nil)
			}
			continue
		}
		if req.WantReply {
			_ = req.Reply(true, nil)
		}
		s.mu.Lock()
		s.commands = append(s.commands, payload.Command)
		s.mu.Unlock()
		s.execCommand(channel, payload.Command)
		return
	}
}

func (s *fakeSSHServer) execCommand(channel ssh.Channel, command string) {
	handler := s.onExec
	if handler == nil {
		handler = defaultFakeExec
	}
	stdout, stderr, code := handler(command)
	_, _ = channel.Write([]byte(stdout))
	_, _ = channel.Stderr().Write([]byte(stderr))
	_, _ = channel.SendRequest("exit-status", false, ssh.Marshal(struct{ Status uint32 }{uint32(code)}))
	_ = channel.CloseWrite()
}

// defaultFakeExec simulates a minimal Linux host: it answers os-release,
// uname, the init-marker file probe, origin URL, branch resolution, and
// rev-parse, accepts every install step, and simulates the work-package-4
// activation (running-init probe, env/unit uploads, and the activate
// script succeeding). A pinned-branch existence check (show-ref) fails by
// default so the missing-branch path is exercised without extra setup.
func defaultFakeExec(command string) (string, string, int) {
	switch {
	case strings.Contains(command, "cat /etc/os-release"):
		return fakeOsRelease, "", 0
	case strings.Contains(command, "uname -m"):
		return "x86_64\n", "", 0
	case strings.Contains(command, "[ -e"):
		return "/usr/lib/systemd/system\n", "", 0
	case strings.Contains(command, "remote.origin.url"):
		return sshinstall.DefaultRepo + "\n", "", 0
	case strings.Contains(command, "symbolic-ref"):
		return "origin/master\n", "", 0
	case strings.Contains(command, "rev-parse HEAD"):
		return fakeCommitHex + "\n", "", 0
	case strings.Contains(command, "show-ref"):
		return "", "branch not found", 1
	case strings.Contains(command, "softlevel"):
		// running-init probe: no systemd/openrc is actually booted.
		return "none\n", "", 0
	case strings.Contains(command, "mktemp -d /tmp/rsmon-worker-act"):
		// server-side secure 0700 upload dir.
		return "/tmp/rsmon-worker-act-fake\n", "", 0
	case strings.Contains(command, "base64 -d"):
		// env/unit uploads succeed.
		return "", "", 0
	case strings.Contains(command, "rsmon-worker activated"):
		// the activate script completed successfully.
		return "", "", 0
	default:
		return "", "", 0
	}
}

func testSSHOptions(port int) SourceInstallOptions {
	return SourceInstallOptions{
		SSHOptions: SSHOptions{
			Host:            "127.0.0.1",
			Port:            port,
			User:            "root",
			Password:        fakeSSHPassword,
			InsecureHostKey: true,
		},
		// Staging-only by default so the work-package-3 orchestration
		// tests keep their historical shape; activation tests opt in.
		Activation: ActivationOptions{Activate: false},
	}
}

// testActivationOptions wraps testSSHOptions with the credentials and
// activation flag needed to run the work-package-4 flow against the fake
// server.
func testActivationOptions(port int) SourceInstallOptions {
	opts := testSSHOptions(port)
	opts.Activation = ActivationOptions{
		Activate: true,
		URL:      "https://rsmon.ru",
		Token:    fakeSSHPassword + "-token",
	}
	return opts
}

func TestSourceInstallSSHFlow(t *testing.T) {
	srv := startFakeSSHServer(t, nil)
	res, err := SourceInstall(testSSHOptions(srv.Port()))
	if err != nil {
		t.Fatal(err)
	}

	if res.Detection.Distro != sshinstall.DistroUbuntu || res.Detection.PackageManager != sshinstall.PkgApt {
		t.Fatalf("detection = %+v", res.Detection)
	}
	if res.Detection.InitSystem != sshinstall.InitSystemd {
		t.Fatalf("init detection = %q, want systemd", res.Detection.InitSystem)
	}
	if res.Plan.Toolchain.Arch != "linux-amd64" || res.Plan.Toolchain.Version != "1.26.0" {
		t.Fatalf("toolchain = %+v", res.Plan.Toolchain)
	}
	if len(res.Plan.Packages) == 0 || res.Plan.Repo == "" {
		t.Fatalf("plan = %+v", res.Plan)
	}
	if res.ResolvedBranch != "master" || res.ResolvedCommit != fakeCommitHex {
		t.Fatalf("resolved = %s @ %s", res.ResolvedBranch, res.ResolvedCommit)
	}
	if res.ToolchainDir != "/usr/local/go" || res.StageBinary != "/opt/rsmon-worker-src/rsmon-worker" ||
		res.RecordFile != "/opt/rsmon-worker-src/rsmon-worker.commit" {
		t.Fatalf("paths = %+v", res)
	}

	commands := srv.Commands()
	var joined strings.Builder
	for _, c := range commands {
		joined.WriteString(c)
		joined.WriteString("\n")
	}
	// Marker substrings that survive the nested `sh -c '<script>'`
	// quoting; exact quoting of each script is asserted by the unit
	// tests (TestPackageScript, TestToolchainScript, ...).
	for _, want := range []string{
		"cat /etc/os-release",
		"uname -m",
		"apt-get update",
		"apt-get install -y --no-install-recommends",
		"sha256sum -c -",
		"git clone",
		"git -C",
		"fetch --prune origin",
		"symbolic-ref --short refs/remotes/origin/HEAD",
		"checkout -q -B",
		"rev-parse HEAD",
		"branch=%s\\ncommit=%s\\n",
		fakeCommitHex,
		"CGO_ENABLED=0",
		"build -trimpath",
		"GOMODCACHE",
		"mv -f",
	} {
		if !strings.Contains(joined.String(), want) {
			t.Fatalf("recorded commands missing %q:\n%s", want, joined.String())
		}
	}

	// Ordering: prerequisites before toolchain before source before build.
	idx := func(sub string) int {
		for i, c := range commands {
			if strings.Contains(c, sub) {
				return i
			}
		}
		t.Fatalf("command %q not found in %v", sub, commands)
		return -1
	}
	if !(idx("apt-get install") < idx("sha256sum") && idx("sha256sum") < idx("git clone") &&
		idx("git clone") < idx("rev-parse") && idx("rev-parse") < idx("-trimpath") &&
		idx("-trimpath") < idx("branch=%s")) {
		t.Fatalf("step order wrong: %v", commands)
	}

	// Adaptive resolution means no pinned-branch existence check ran.
	if strings.Contains(joined.String(), "show-ref") {
		t.Fatalf("show-ref ran despite branch resolution:\n%s", joined.String())
	}
	if strings.Contains(joined.String(), fakeSSHPassword) {
		t.Fatal("SSH password leaked into a remote command")
	}
}

func TestSourceInstallSSHRejectsMissingPinnedBranch(t *testing.T) {
	srv := startFakeSSHServer(t, nil)
	opts := testSSHOptions(srv.Port())
	opts.Branch = "main"
	_, err := SourceInstall(opts)
	if err == nil || !strings.Contains(err.Error(), `branch "main" does not exist`) {
		t.Fatalf("err = %v, want missing-branch error", err)
	}
	if commands := srv.Commands(); !strings.Contains(strings.Join(commands, "\n"), "show-ref") {
		t.Fatalf("pinned branch existence was not verified: %v", commands)
	}
}

func TestSourceInstallSSHBuildFailure(t *testing.T) {
	srv := startFakeSSHServer(t, func(command string) (string, string, int) {
		if strings.Contains(command, "-trimpath") {
			return "", "build exploded", 1
		}
		return defaultFakeExec(command)
	})
	_, err := SourceInstall(testSSHOptions(srv.Port()))
	if err == nil || !strings.Contains(err.Error(), "build worker binary") || !strings.Contains(err.Error(), "build exploded") {
		t.Fatalf("err = %v, want bounded build failure", err)
	}
}

func TestSourceInstallSSHDetectionFailure(t *testing.T) {
	srv := startFakeSSHServer(t, func(command string) (string, string, int) {
		if strings.Contains(command, "os-release") {
			return "", "os-release unreadable", 1
		}
		return defaultFakeExec(command)
	})
	_, err := SourceInstall(testSSHOptions(srv.Port()))
	if err == nil || !strings.Contains(err.Error(), "/etc/os-release") {
		t.Fatalf("err = %v, want detection failure", err)
	}
}

// TestSourceInstallSSHCheckoutFailureNotMasked proves the fail-closed
// contract: a failed checkout (e.g. a dirty working tree) surfaces as an
// error and never reaches the build or commit-record steps, so the
// previous staging binary and record are preserved.
func TestSourceInstallSSHCheckoutFailureNotMasked(t *testing.T) {
	srv := startFakeSSHServer(t, func(command string) (string, string, int) {
		if strings.Contains(command, "checkout -q -B") {
			return "", "your local changes to the following files would be overwritten by checkout", 1
		}
		return defaultFakeExec(command)
	})
	_, err := SourceInstall(testSSHOptions(srv.Port()))
	if err == nil || !strings.Contains(err.Error(), "check out branch") {
		t.Fatalf("err = %v, want checkout failure", err)
	}
	joined := strings.Join(srv.Commands(), "\n")
	if strings.Contains(joined, "-trimpath") || strings.Contains(joined, "branch=%s") {
		t.Fatalf("build or record ran after checkout failed:\n%s", joined)
	}
}

// TestSourceInstallSSHCommandTimeout verifies each remote command is
// bounded by SessionTimeout and the run reports it.
func TestSourceInstallSSHCommandTimeout(t *testing.T) {
	srv := startFakeSSHServer(t, func(command string) (string, string, int) {
		if strings.Contains(command, "uname -m") {
			time.Sleep(5 * time.Second)
			return "x86_64\n", "", 0
		}
		return defaultFakeExec(command)
	})
	opts := testSSHOptions(srv.Port())
	opts.SessionTimeout = 300 * time.Millisecond
	_, err := SourceInstall(opts)
	if err == nil || !strings.Contains(err.Error(), "timed out after") {
		t.Fatalf("err = %v, want command timeout", err)
	}
}

// TestSourceInstallSSHActivationFlow verifies the work-package-4 flow
// over a real SSH session: after the staging build and commit record, the
// installer probes the running init, uploads the env and service
// definition, runs the atomic activation script, and reports the installed
// layout. The secrets-absent contract holds: the worker token never
// reaches a remote command.
func TestSourceInstallSSHActivationFlow(t *testing.T) {
	srv := startFakeSSHServer(t, nil)
	res, err := SourceInstall(testActivationOptions(srv.Port()))
	if err != nil {
		t.Fatal(err)
	}
	if res.Activation == nil {
		t.Fatal("activation result missing")
	}
	if res.Activation.Binary != "/usr/local/bin/rsmon-worker" ||
		res.Activation.EnvFile != "/etc/rsmon-worker/worker.env" ||
		res.Activation.DataDir != "/var/lib/rsmon-worker" ||
		res.Activation.UnitFile != "/etc/systemd/system/rsmon-worker.service" ||
		res.Activation.Supervisor != "none" ||
		!res.Activation.Started {
		t.Fatalf("activation result = %+v", res.Activation)
	}

	commands := srv.Commands()
	var joined strings.Builder
	for _, c := range commands {
		joined.WriteString(c)
		joined.WriteString("\n")
	}
	for _, want := range []string{
		"/run/systemd/system",
		"softlevel",
		"mktemp -d /tmp/rsmon-worker-act",
		"umask 077; base64 -d >",
		"rsmon-worker activated",
		"install -m 0755",
		"install -m 0600",
		"chmod 0750",
		"worker.pid",
		"liveness",
		"rm -rf --",
	} {
		if !strings.Contains(joined.String(), want) {
			t.Fatalf("activation commands missing %q:\n%s", want, joined.String())
		}
	}

	// Ordering: build and commit record, then a server-side secure upload
	// dir is created, the env/unit are uploaded into it, and the activation
	// script runs last.
	idx := func(sub string) int {
		for i, c := range commands {
			if strings.Contains(c, sub) {
				return i
			}
		}
		t.Fatalf("command %q not found in %v", sub, commands)
		return -1
	}
	if idx("branch=%s") >= idx("mktemp -d /tmp/rsmon-worker-act") ||
		idx("mktemp -d /tmp/rsmon-worker-act") >= idx("base64 -d >") ||
		idx("base64 -d >") >= idx("rsmon-worker activated") {
		t.Fatalf("activation order wrong: %v", commands)
	}
	if strings.Contains(joined.String(), fakeSSHPassword+"-token") {
		t.Fatal("worker token leaked into a remote command")
	}
}

// TestSourceInstallSSHNoActivationSkipsService proves the staging
// boundary: without Activation the installer never probes the init system,
// uploads an env/unit, or runs the activation script.
func TestSourceInstallSSHNoActivationSkipsService(t *testing.T) {
	srv := startFakeSSHServer(t, nil)
	res, err := SourceInstall(testSSHOptions(srv.Port()))
	if err != nil {
		t.Fatal(err)
	}
	if res.Activation != nil {
		t.Fatalf("activation must not run without the flag: %+v", res.Activation)
	}
	joined := strings.Join(srv.Commands(), "\n")
	for _, forbidden := range []string{"softlevel", "base64 -d >", "rsmon-worker activated", "worker.pid"} {
		if strings.Contains(joined, forbidden) {
			t.Fatalf("staging-only flow ran activation step %q:\n%s", forbidden, joined)
		}
	}
}

// TestSourceInstallSSHActivationRequiresToken verifies the credentials
// gate: activation without a token fails before any remote connection.
func TestSourceInstallSSHActivationRequiresToken(t *testing.T) {
	srv := startFakeSSHServer(t, nil)
	opts := testSSHOptions(srv.Port())
	opts.Activation = ActivationOptions{Activate: true, URL: "https://rsmon.ru"}
	_, err := SourceInstall(opts)
	if err == nil || !strings.Contains(err.Error(), "RSMON_TOKEN") {
		t.Fatalf("err = %v, want token requirement", err)
	}
	if got := len(srv.Commands()); got != 0 {
		t.Fatalf("commands ran before validation failed: %d", got)
	}
}

// TestSourceInstallSSHActivationFailure verifies the bounded failure
// path: a failing activate script surfaces as a labelled error with the
// remote stderr, and the uploaded temp env/unit are cleaned up by the
// controller defer regardless of the outcome.
func TestSourceInstallSSHActivationFailure(t *testing.T) {
	srv := startFakeSSHServer(t, func(command string) (string, string, int) {
		if strings.Contains(command, "rsmon-worker activated") {
			return "", "activation exploded: disk full", 1
		}
		return defaultFakeExec(command)
	})
	_, err := SourceInstall(testActivationOptions(srv.Port()))
	if err == nil || !strings.Contains(err.Error(), "activate worker service") || !strings.Contains(err.Error(), "disk full") {
		t.Fatalf("err = %v, want bounded activation failure", err)
	}
	joined := strings.Join(srv.Commands(), "\n")
	if !strings.Contains(joined, "rm -rf --") {
		t.Fatalf("secure upload dir cleanup not issued after activation failure:\n%s", joined)
	}
}

// TestSourceInstallSSHActivationHealthFailure verifies the health gate
// surfaces as an activation failure and the temp files are still cleaned.
// The fake cannot run the shell script, so the failure is injected at the
// activate-command level; the real rollback semantics are covered by the
// Docker/OpenSSH E2E tests.
func TestSourceInstallSSHActivationHealthFailure(t *testing.T) {
	srv := startFakeSSHServer(t, func(command string) (string, string, int) {
		if strings.Contains(command, "rsmon-worker activated") {
			return "", "health failure: worker did not answer /healthz", 1
		}
		return defaultFakeExec(command)
	})
	_, err := SourceInstall(testActivationOptions(srv.Port()))
	if err == nil || !strings.Contains(err.Error(), "did not answer /healthz") {
		t.Fatalf("err = %v, want /healthz verification failure", err)
	}
}