package main

import (
	"bufio"
	"bytes"
	"context"
	"crypto/rand"
	"crypto/rsa"
	"crypto/sha256"
	"encoding/hex"
	"fmt"
	"io"
	"log"
	"os"
	"strconv"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/bramvdbogaerde/go-scp"
	"github.com/moby/moby/api/types/container"
	"github.com/pkg/sftp"
	"github.com/shellhub-io/shellhub/pkg/api/requests"
	"github.com/shellhub-io/shellhub/pkg/models"
	"github.com/shellhub-io/shellhub/tests/environment"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
	"github.com/testcontainers/testcontainers-go"
	"golang.org/x/crypto/ssh"
)

const (
	ShellHubAgentUsername = environment.AgentUsername
	ShellHubAgentPassword = environment.AgentPassword
)

const (
	ShellHubUsername      = "test"
	ShellHubPassword      = "password"
	ShellHubNamespaceName = "testspace"
	ShellHubNamespace     = "00000000-0000-4000-0000-000000000000"
	ShellHubEmail         = "test@ossystems.com.br"
)

type NewAgentContainerOption func(envs map[string]string)

func NewAgentContainerWithIdentity(identity string) NewAgentContainerOption {
	return func(envs map[string]string) {
		envs["SHELLHUB_PREFERRED_IDENTITY"] = identity
	}
}

func NewAgentContainerWithConnectionVersion(version int) NewAgentContainerOption {
	return func(envs map[string]string) {
		envs["SHELLHUB_TRANSPORT_VERSION"] = strconv.Itoa(version)
	}
}

func NewAgentContainerWithTenant(tenant string) NewAgentContainerOption {
	return func(envs map[string]string) {
		envs["SHELLHUB_TENANT_ID"] = tenant
	}
}

func NewAgentContainerWithHostname(hostname string) NewAgentContainerOption {
	return func(envs map[string]string) {
		envs["SHELLHUB_PREFERRED_HOSTNAME"] = hostname
	}
}

func NewAgentContainerWithTenantOnly(tenant string) NewAgentContainerOption {
	return func(envs map[string]string) {
		delete(envs, "SHELLHUB_PROVISIONING_KEY")
		envs["SHELLHUB_TENANT_ID"] = tenant
	}
}

// NewAgentContainerWithProvisioningKey drops the tenant id, so the device proves the key alone
// resolved the namespace.
func NewAgentContainerWithProvisioningKey(key string) NewAgentContainerOption {
	return func(envs map[string]string) {
		delete(envs, "SHELLHUB_TENANT_ID")
		envs["SHELLHUB_PROVISIONING_KEY"] = key
	}
}

// NewAgentContainerUnpaired drops the tenant id, so the agent asks for a pairing code and joins the
// namespace of whoever accepts it.
func NewAgentContainerUnpaired() NewAgentContainerOption {
	return func(envs map[string]string) {
		delete(envs, "SHELLHUB_TENANT_ID")
		delete(envs, "SHELLHUB_PROVISIONING_KEY")
	}
}

func envOr(key, fallback string) string {
	if value := os.Getenv(key); value != "" {
		return value
	}

	return fallback
}

func agentBuildLog() io.Writer {
	if os.Getenv("SHELLHUB_E2E_AGENT_CONTEXT") != "" || os.Getenv("SHELLHUB_E2E_AGENT_DOCKERFILE") != "" {
		return os.Stderr
	}

	return nil
}

type agentImageBuild struct {
	done chan struct{}
	tag  string
	err  error
}

var agentImages struct {
	sync.Mutex
	builds map[string]*agentImageBuild
}

func buildAgentImage(ctx context.Context, version string, tags ...string) (string, error) {
	return cachedAgentImage(ctx, strings.Join(append([]string{version}, tags...), "-"), environment.AgentBuild{
		Repository: os.Getenv("SHELLHUB_E2E_AGENT_IMAGE"),
		Context:    envOr("SHELLHUB_E2E_AGENT_CONTEXT", ".."),
		Dockerfile: os.Getenv("SHELLHUB_E2E_AGENT_DOCKERFILE"),
		Version:    version,
		Tags:       tags,
		Output:     agentBuildLog(),
	})
}

func cachedAgentImage(ctx context.Context, key string, build environment.AgentBuild) (string, error) {
	agentImages.Lock()
	inflight, ok := agentImages.builds[key]
	if !ok {
		if agentImages.builds == nil {
			agentImages.builds = make(map[string]*agentImageBuild)
		}

		inflight = &agentImageBuild{done: make(chan struct{})}
		agentImages.builds[key] = inflight
	}
	agentImages.Unlock()

	if ok {
		select {
		case <-inflight.done:
			return inflight.tag, inflight.err
		case <-ctx.Done():
			return "", ctx.Err()
		}
	}

	defer close(inflight.done)

	inflight.tag, inflight.err = environment.BuildAgentImage(ctx, run, build)
	if inflight.err != nil {
		agentImages.Lock()
		delete(agentImages.builds, key)
		agentImages.Unlock()
	}

	return inflight.tag, inflight.err
}

func NewAgentContainer(ctx context.Context, relayID string, opts ...NewAgentContainerOption) (testcontainers.Container, error) {
	return NewAgentContainerAtVersion(ctx, relayID, "", opts...)
}

// NewAgentContainerAtVersion builds an agent that reports version instead of "latest" and returns
// its container, created but not started. Each version is built under its own tag, so the agent
// every other test starts keeps its tag; an empty version builds that default agent. A version
// builds once per run unless its build fails, and a caller asking for a version another caller is
// building waits for that build without blocking callers of other versions. It returns the error
// of the build it waited for, even one that the building caller's context ended, or ctx.Err()
// when ctx ends first, or the error of a container creation that fails, bounded by ctx.
func NewAgentContainerAtVersion(ctx context.Context, relayID, version string, opts ...NewAgentContainerOption) (testcontainers.Container, error) {
	image, err := buildAgentImage(ctx, version)
	if err != nil {
		return nil, err
	}

	return newAgentContainer(ctx, relayID, image, opts...)
}

func newAgentContainer(ctx context.Context, relayID, image string, opts ...NewAgentContainerOption) (testcontainers.Container, error) {
	req, err := agentContainerRequest(relayID, image, opts...)
	if err != nil {
		return nil, err
	}

	return createAgentContainer(ctx, req)
}

func agentContainerRequest(relayID, image string, opts ...NewAgentContainerOption) (testcontainers.ContainerRequest, error) {
	raw := make([]byte, 6)
	if _, err := rand.Read(raw); err != nil {
		return testcontainers.ContainerRequest{}, err
	}

	identity := hex.EncodeToString(raw)

	envs := map[string]string{
		"SHELLHUB_PREFERRED_IDENTITY": identity,
		"SHELLHUB_PREFERRED_HOSTNAME": "agent-" + identity,
		"SHELLHUB_SERVER_ADDRESS":     "http://localhost",
		"SHELLHUB_TENANT_ID":          "00000000-0000-4000-0000-000000000000",
		"SHELLHUB_PRIVATE_KEY":        "/tmp/shellhub.key",
		"SHELLHUB_LOG_FORMAT":         "json",
		"SHELLHUB_KEEPALIVE_INTERVAL": "1",
		"SHELLHUB_LOG_LEVEL":          "trace",
		"SHELLHUB_TRANSPORT_VERSION":  "2",
	}

	for _, opt := range opts {
		opt(envs)
	}

	return testcontainers.ContainerRequest{
		Image:  image,
		Env:    envs,
		Labels: run.Labels(),
		HostConfigModifier: func(hc *container.HostConfig) {
			hc.NetworkMode = container.NetworkMode("container:" + relayID)
		},
	}, nil
}

func createAgentContainer(ctx context.Context, req testcontainers.ContainerRequest) (testcontainers.Container, error) {
	return testcontainers.GenericContainer(ctx, testcontainers.GenericContainerRequest{
		ContainerRequest: req,
		Logger:           log.New(io.Discard, "", log.LstdFlags),
	})
}

func TestSSHNewNamespace(t *testing.T) {
	t.Parallel()

	ctx := context.Background()

	compose := newSSHEnvironment(t, ctx, newNamespace)
	_, device := startAcceptedAgent(t, ctx, compose)

	sshid := deviceSSHID(device)
	addr := compose.SSHAddress()

	t.Run("password authentication is not offered", func(t *testing.T) {
		config := &ssh.ClientConfig{
			User:            sshid,
			Auth:            []ssh.AuthMethod{ssh.Password(ShellHubAgentPassword)},
			HostKeyCallback: ssh.InsecureIgnoreHostKey(), //nolint:gosec
		}

		_, err := ssh.Dial("tcp", addr, config)
		require.Error(t, err)
		require.Contains(t, err.Error(), "attempted methods []")
	})

	t.Run("authenticate with an enrolled identity", func(t *testing.T) {
		signer, data := newSigner(t)

		compose.EnrollIdentity(t, "integration", data)

		config := &ssh.ClientConfig{
			User:            sshid,
			Auth:            []ssh.AuthMethod{ssh.PublicKeys(signer)},
			HostKeyCallback: ssh.InsecureIgnoreHostKey(), //nolint:gosec
		}

		var conn *ssh.Client

		require.EventuallyWithT(t, func(tt *assert.CollectT) {
			var err error

			conn, err = ssh.Dial("tcp", addr, config)
			assert.NoError(tt, err)
		}, 30*time.Second, 1*time.Second)

		_ = conn.Close()
	})
}

func TestSSHConnectionV1(t *testing.T) {
	t.Parallel()

	testSSHWithVersion(t, 1)
}

func TestSSHConnectionV2(t *testing.T) {
	t.Parallel()

	testSSHWithVersion(t, 2)
}

func testSSHWithVersion(t *testing.T, connectionVersion int) {
	t.Helper()

	type Environment struct {
		services *environment.DockerCompose
		agent    testcontainers.Container
		signer   ssh.Signer
	}

	tests := []struct {
		name    string
		options []NewAgentContainerOption
		run     func(*testing.T, *Environment, *models.Device)
	}{
		{
			name:    "reconnect to server",
			options: []NewAgentContainerOption{NewAgentContainerWithIdentity("custom-reconnect")},
			run: func(t *testing.T, environment *Environment, device *models.Device) {
				t.Helper()

				ctx := context.Background()

				err := environment.agent.Stop(ctx, nil)
				require.NoError(t, err)

				environment.services.AwaitDeviceOffline(t, device.UID)

				err = environment.agent.Start(ctx)
				require.NoError(t, err)

				environment.services.AwaitDeviceOnline(t, device.UID)

				conn := dialDevice(t, ctx, environment.services, device, environment.signer)
				defer conn.Close() //nolint:errcheck // the test is over once the new tunnel answered

				assert.Equal(t, "reconnected", runOnDevice(t, conn, "echo -n reconnected"))
			},
		},
		{
			name:    "authenticate with password",
			options: []NewAgentContainerOption{NewAgentContainerWithIdentity("custom-password")},
			run: func(t *testing.T, environment *Environment, device *models.Device) {
				t.Helper()

				skipUnlessAgentAcceptsPasswords(t)

				config := &ssh.ClientConfig{
					User: deviceSSHID(device),
					Auth: []ssh.AuthMethod{
						ssh.Password(ShellHubAgentPassword),
					},
					HostKeyCallback: ssh.InsecureIgnoreHostKey(), //nolint:gosec
				}

				var conn *ssh.Client

				require.EventuallyWithT(t, func(tt *assert.CollectT) {
					var err error

					conn, err = ssh.Dial("tcp", environment.services.SSHAddress(), config)
					assert.NoError(tt, err)
				}, 30*time.Second, 1*time.Second)

				_ = conn.Close()
			},
		},
		{
			name: "fail to authenticate with password",
			run: func(t *testing.T, environment *Environment, device *models.Device) {
				t.Helper()

				skipUnlessAgentAcceptsPasswords(t)

				config := &ssh.ClientConfig{
					User: deviceSSHID(device),
					Auth: []ssh.AuthMethod{
						ssh.Password("wrongpassword"),
					},
					HostKeyCallback: ssh.InsecureIgnoreHostKey(), //nolint:gosec
				}

				err := handshakeWith(t.Context(), environment.services.SSHAddress(), config)
				require.ErrorContains(t, err, "ssh: unable to authenticate, attempted methods [none password]")
			},
		},
		{
			name: "authenticate with public key",
			run: func(t *testing.T, environment *Environment, device *models.Device) {
				t.Helper()

				ctx := context.Background()

				signer := registerDeviceKey(t, ctx, environment.services)

				config := &ssh.ClientConfig{
					User: deviceSSHID(device),
					Auth: []ssh.AuthMethod{
						ssh.PublicKeys(signer),
					},
					HostKeyCallback: ssh.InsecureIgnoreHostKey(), //nolint:gosec
				}

				conn, err := ssh.Dial("tcp", environment.services.SSHAddress(), config)
				require.NoError(t, err)

				_ = conn.Close()
			},
		},
		{
			name: "fail to authenticate with public key",
			run: func(t *testing.T, environment *Environment, device *models.Device) {
				t.Helper()

				signer, _ := newSigner(t)

				config := &ssh.ClientConfig{
					User: deviceSSHID(device),
					Auth: []ssh.AuthMethod{
						ssh.PublicKeys(signer),
					},
					HostKeyCallback: ssh.InsecureIgnoreHostKey(), //nolint:gosec
				}

				err := handshakeWith(t.Context(), environment.services.SSHAddress(), config)
				require.ErrorContains(t, err, "ssh: unable to authenticate, attempted methods [none publickey]")
			},
		},
		{
			name: "connection EXEC and a SHELL on same connection",
			run: func(t *testing.T, environment *Environment, device *models.Device) {
				t.Helper()

				config := &ssh.ClientConfig{
					User: deviceSSHID(device),
					Auth: []ssh.AuthMethod{
						ssh.PublicKeys(environment.signer),
					},
					HostKeyCallback: ssh.InsecureIgnoreHostKey(), //nolint:gosec
				}

				var conn *ssh.Client

				require.EventuallyWithT(t, func(tt *assert.CollectT) {
					var err error

					conn, err = ssh.Dial("tcp", environment.services.SSHAddress(), config)
					assert.NoError(tt, err)
				}, 30*time.Second, 1*time.Second)

				{
					sess, err := conn.NewSession()
					require.NoError(t, err)

					output, err := sess.Output(`echo -n "test"`)
					require.NoError(t, err)

					assert.Equal(t, "test", string(output))

					_ = sess.Close()
				}
				{
					sess, err := conn.NewSession()
					require.NoError(t, err)

					err = sess.RequestPty("xterm", 100, 100, ssh.TerminalModes{
						ssh.ECHO:          1,
						ssh.TTY_OP_ISPEED: 14400,
						ssh.TTY_OP_OSPEED: 14400,
					})
					require.NoError(t, err)

					err = sess.Shell()
					require.NoError(t, err)

					_ = sess.Close()
				}

				_ = conn.Close()
			},
		},
		{
			name:    "connection EXEC",
			options: []NewAgentContainerOption{NewAgentContainerWithIdentity("custom-exec")},
			run: func(t *testing.T, environment *Environment, device *models.Device) {
				t.Helper()

				config := &ssh.ClientConfig{
					User: deviceSSHID(device),
					Auth: []ssh.AuthMethod{
						ssh.PublicKeys(environment.signer),
					},
					HostKeyCallback: ssh.InsecureIgnoreHostKey(), //nolint:gosec
				}

				var conn *ssh.Client

				require.EventuallyWithT(t, func(tt *assert.CollectT) {
					var err error

					conn, err = ssh.Dial("tcp", environment.services.SSHAddress(), config)
					assert.NoError(tt, err)
				}, 30*time.Second, 1*time.Second)

				sess, err := conn.NewSession()
				require.NoError(t, err)

				output, err := sess.Output(`echo -n "test"`)
				require.NoError(t, err)

				assert.Equal(t, "test", string(output))

				_ = sess.Close()
				_ = conn.Close()
			},
		},
		{
			name: "connection EXEC with non zero status code",
			run: func(t *testing.T, environment *Environment, device *models.Device) {
				t.Helper()

				config := &ssh.ClientConfig{
					User: deviceSSHID(device),
					Auth: []ssh.AuthMethod{
						ssh.PublicKeys(environment.signer),
					},
					HostKeyCallback: ssh.InsecureIgnoreHostKey(), //nolint:gosec
				}

				var conn *ssh.Client

				require.EventuallyWithT(t, func(tt *assert.CollectT) {
					var err error

					conn, err = ssh.Dial("tcp", environment.services.SSHAddress(), config)
					assert.NoError(tt, err)
				}, 30*time.Second, 1*time.Second)

				sess, err := conn.NewSession()
				require.NoError(t, err)

				var status *ssh.ExitError

				output, err := sess.CombinedOutput(`echo -n "test" 1>&2; exit 142`)
				require.ErrorAs(t, err, &status)

				assert.Equal(t, 142, status.ExitStatus())
				assert.Equal(t, "test", string(output))

				_ = sess.Close()
				_ = conn.Close()
			},
		},
		{
			name:    "connection SFTP to download file",
			options: []NewAgentContainerOption{},
			run: func(t *testing.T, environment *Environment, device *models.Device) {
				t.Helper()

				config := &ssh.ClientConfig{
					User: deviceSSHID(device),
					Auth: []ssh.AuthMethod{
						ssh.PublicKeys(environment.signer),
					},
					HostKeyCallback: ssh.InsecureIgnoreHostKey(), //nolint:gosec
				}

				var conn *ssh.Client

				require.EventuallyWithT(t, func(tt *assert.CollectT) {
					var err error

					conn, err = ssh.Dial("tcp", environment.services.SSHAddress(), config)
					assert.NoError(tt, err)
				}, 30*time.Second, 1*time.Second)

				sess, err := sftp.NewClient(conn)
				require.NoError(t, err)

				received, err := sess.OpenFile("/etc/os-release", os.O_RDONLY)
				require.NoError(t, err)

				var data string

				_, err = fmt.Fscanf(received, "%s", &data)
				require.NoError(t, err)

				assert.Contains(t, data, "Alpine")

				_ = sess.Close()
				_ = conn.Close()
			},
		},
		{
			name:    "connection SCP to upload file",
			options: []NewAgentContainerOption{},
			run: func(t *testing.T, environment *Environment, device *models.Device) {
				t.Helper()

				config := &ssh.ClientConfig{
					User: deviceSSHID(device),
					Auth: []ssh.AuthMethod{
						ssh.PublicKeys(environment.signer),
					},
					HostKeyCallback: ssh.InsecureIgnoreHostKey(), //nolint:gosec
				}

				var conn *ssh.Client

				require.EventuallyWithT(t, func(tt *assert.CollectT) {
					var err error

					conn, err = ssh.Dial("tcp", environment.services.SSHAddress(), config)
					assert.NoError(tt, err)
				}, 30*time.Second, 1*time.Second)

				sess, err := scp.NewClientBySSH(conn)
				require.NoError(t, err)

				const content = "uploaded through scp\n"

				err = sess.CopyFilePassThru(t.Context(), strings.NewReader(content), "/tmp/scp-uploaded", "0640", nil)
				require.NoError(t, err)

				sess.Close()

				assert.Equal(t, content, runOnDevice(t, conn, "cat /tmp/scp-uploaded"))
				assert.Equal(t, "640\n", runOnDevice(t, conn, "stat -c %a /tmp/scp-uploaded"))

				_ = conn.Close()
			},
		},
		{
			name:    "connection SCP to download file",
			options: []NewAgentContainerOption{},
			run: func(t *testing.T, environment *Environment, device *models.Device) {
				t.Helper()

				config := &ssh.ClientConfig{
					User: deviceSSHID(device),
					Auth: []ssh.AuthMethod{
						ssh.PublicKeys(environment.signer),
					},
					HostKeyCallback: ssh.InsecureIgnoreHostKey(), //nolint:gosec
				}

				var conn *ssh.Client

				require.EventuallyWithT(t, func(tt *assert.CollectT) {
					var err error

					conn, err = ssh.Dial("tcp", environment.services.SSHAddress(), config)
					assert.NoError(tt, err)
				}, 30*time.Second, 1*time.Second)

				const content = "downloaded through scp\n"

				runOnDevice(t, conn, "printf %s '"+content+"' > /tmp/scp-downloaded")

				sess, err := scp.NewClientBySSH(conn)
				require.NoError(t, err)

				var file bytes.Buffer

				err = sess.CopyFromRemotePassThru(t.Context(), &file, "/tmp/scp-downloaded", nil)
				require.NoError(t, err)

				sess.Close()

				assert.Equal(t, content, file.String())

				_ = conn.Close()
			},
		},
		{
			name: "connection EXEC of a shell builtin",
			run: func(t *testing.T, environment *Environment, device *models.Device) {
				t.Helper()

				conn := dialDevice(t, t.Context(), environment.services, device, environment.signer)
				defer conn.Close() //nolint:errcheck // the test is over once the builtin answered

				sess, err := conn.NewSession()
				require.NoError(t, err)
				defer sess.Close() //nolint:errcheck // Output already waited for the exit status

				output, err := sess.Output("cd /var/log && pwd && type -t cd")
				require.NoError(t, err)

				assert.Equal(t, "/var/log\nbuiltin\n", string(output))
			},
		},
		{
			name: "connection SFTP to make a directory",
			run: func(t *testing.T, environment *Environment, device *models.Device) {
				t.Helper()

				conn := dialDevice(t, t.Context(), environment.services, device, environment.signer)
				defer conn.Close() //nolint:errcheck // the test is over once the directory is read back

				client, err := sftp.NewClient(conn)
				require.NoError(t, err)
				defer client.Close() //nolint:errcheck // the test reads its outcome over an exec channel, so a failed close leaves nothing to assert

				require.NoError(t, client.Mkdir("/tmp/sftp-made"))

				assert.Equal(t, "directory\n", runOnDevice(t, conn, "stat -c %F /tmp/sftp-made"))
			},
		},
		{
			name: "connection SFTP to list a directory",
			run: func(t *testing.T, environment *Environment, device *models.Device) {
				t.Helper()

				conn := dialDevice(t, t.Context(), environment.services, device, environment.signer)
				defer conn.Close() //nolint:errcheck // the test is over once the listing is read

				runOnDevice(t, conn, "mkdir /tmp/sftp-listed && printf 12345 > /tmp/sftp-listed/file && mkdir /tmp/sftp-listed/dir")

				client, err := sftp.NewClient(conn)
				require.NoError(t, err)
				defer client.Close() //nolint:errcheck // the listing is already read

				entries, err := client.ReadDir("/tmp/sftp-listed")
				require.NoError(t, err)

				listed := map[string]os.FileInfo{}
				for _, entry := range entries {
					listed[entry.Name()] = entry
				}

				require.Len(t, listed, 2)
				require.Contains(t, listed, "file")
				require.Contains(t, listed, "dir")
				assert.False(t, listed["file"].IsDir())
				assert.Equal(t, int64(5), listed["file"].Size())
				assert.True(t, listed["dir"].IsDir())
			},
		},
		{
			name: "connection SCP to upload a directory recursively",
			run: func(t *testing.T, environment *Environment, device *models.Device) {
				t.Helper()

				conn := dialDevice(t, t.Context(), environment.services, device, environment.signer)
				defer conn.Close() //nolint:errcheck // the test is over once the tree is read back

				sendSCPTree(t, conn, "/tmp", scpDirectory{
					name: "scp-tree",
					mode: "0750",
					files: map[string]string{
						"top.txt": "at the top",
					},
					dirs: []scpDirectory{{
						name: "nested",
						mode: "0700",
						files: map[string]string{
							"deep.txt": "one level down",
						},
					}},
				})

				assert.Equal(t, "at the top|one level down", runOnDevice(t, conn,
					`printf '%s|%s' "$(cat /tmp/scp-tree/top.txt)" "$(cat /tmp/scp-tree/nested/deep.txt)"`))
				assert.Equal(t, "750 700\n", runOnDevice(t, conn, "stat -c %a /tmp/scp-tree /tmp/scp-tree/nested | paste -sd ' '"))
			},
		},
		{
			name:    "direct tcpip port redirect",
			options: []NewAgentContainerOption{},
			run: func(t *testing.T, env *Environment, device *models.Device) {
				t.Helper()

				config := &ssh.ClientConfig{
					User: deviceSSHID(device),
					Auth: []ssh.AuthMethod{
						ssh.PublicKeys(env.signer),
					},
					HostKeyCallback: ssh.InsecureIgnoreHostKey(), //nolint:gosec
				}

				conn, err := ssh.Dial("tcp", env.services.SSHAddress(), config)
				require.NoError(t, err)

				type Data struct {
					DestAddr   string
					DestPort   uint32
					OriginAddr string
					OriginPort uint32
				}

				orig, err := strconv.Atoi(environment.ReservePort(t))
				require.NoError(t, err)

				data := Data{
					DestAddr:   "127.0.0.1",
					DestPort:   80,
					OriginAddr: "127.0.0.1",
					OriginPort: uint32(orig), //nolint:gosec
				}

				ch, _, err := conn.OpenChannel("direct-tcpip", ssh.Marshal(data))
				require.NoError(t, err)

				_, err = ch.Write([]byte("GET / HTTP/1.0\r\n\r\n"))
				require.NoError(t, err)

				status, err := bufio.NewReader(ch).ReadString('\n')
				require.NoError(t, err)
				require.True(t, strings.HasPrefix(status, "HTTP/1."), status)

				_ = ch.Close()
				_ = conn.Close()
			},
		},
		{
			name: "connection EXEC outlives a silent pause",
			run: func(t *testing.T, environment *Environment, device *models.Device) {
				t.Helper()

				config := &ssh.ClientConfig{
					User: deviceSSHID(device),
					Auth: []ssh.AuthMethod{
						ssh.PublicKeys(environment.signer),
					},
					HostKeyCallback: ssh.InsecureIgnoreHostKey(), //nolint:gosec
				}

				conn, err := ssh.Dial("tcp", environment.services.SSHAddress(), config)
				require.NoError(t, err)

				sess, err := conn.NewSession()
				require.NoError(t, err)

				output, err := sess.CombinedOutput("sleep 3 && echo -n 'still alive'")
				require.NoError(t, err)

				assert.Equal(t, "still alive", string(output))

				_ = sess.Close()
				_ = conn.Close()
			},
		},
		{
			name:    "connection SFTP to upload large file",
			options: []NewAgentContainerOption{},
			run: func(t *testing.T, environment *Environment, device *models.Device) {
				t.Helper()

				config := &ssh.ClientConfig{
					User: deviceSSHID(device),
					Auth: []ssh.AuthMethod{
						ssh.PublicKeys(environment.signer),
					},
					HostKeyCallback: ssh.InsecureIgnoreHostKey(), //nolint:gosec
				}

				conn, err := ssh.Dial("tcp", environment.services.SSHAddress(), config)
				require.NoError(t, err)
				defer conn.Close() //nolint:errcheck

				client, err := sftp.NewClient(conn)
				require.NoError(t, err)
				defer client.Close() //nolint:errcheck

				fileSize := 10 * 1024 * 1024 // 10MB
				randomData := make([]byte, fileSize)
				rand.Read(randomData)

				tempFile, err := os.CreateTemp(t.TempDir(), "large-file-test-*.bin")
				require.NoError(t, err)
				defer os.Remove(tempFile.Name()) //nolint:errcheck

				_, err = tempFile.Write(randomData)
				require.NoError(t, err)
				_ = tempFile.Close()

				localFile, err := os.Open(tempFile.Name())
				require.NoError(t, err)
				defer localFile.Close() //nolint:errcheck

				remoteFile, err := client.Create("/tmp/large-file-test.bin")
				require.NoError(t, err)

				written, err := io.Copy(remoteFile, localFile)
				require.NoError(t, err)
				assert.Equal(t, int64(fileSize), written)
				require.NoError(t, remoteFile.Close())

				sum := sha256.Sum256(randomData)
				received := strings.Fields(runOnDevice(t, conn, "sha256sum /tmp/large-file-test.bin"))
				require.NotEmpty(t, received)
				assert.Equal(t, hex.EncodeToString(sum[:]), received[0])
			},
		},
		{
			name: "OpenSSH scp in SFTP mode uploads a file and exits 0",
			run: func(t *testing.T, environment *Environment, device *models.Device) {
				t.Helper()

				execInAgent(t, environment.agent, "ssh-keygen", "-q", "-t", "ed25519", "-N", "", "-f", "/tmp/scp-key")
				registerAuthorizedKey(t, t.Context(), environment.services, "openssh-scp", execInAgent(t, environment.agent, "cat", "/tmp/scp-key.pub"))

				execInAgent(t, environment.agent, "sh", "-c", "head -c 1048576 /dev/urandom > /tmp/scp-source")

				execInAgent(t, environment.agent, "scp", "-s",
					"-i", "/tmp/scp-key",
					"-o", "BatchMode=yes",
					"-o", "StrictHostKeyChecking=no",
					"-o", "UserKnownHostsFile=/dev/null",
					"-o", "User="+deviceSSHID(device),
					"-P", "2222",
					"/tmp/scp-source", "server:/tmp/scp-received",
				)

				sent := strings.Fields(execInAgent(t, environment.agent, "sha256sum", "/tmp/scp-source"))
				received := strings.Fields(execInAgent(t, environment.agent, "sha256sum", "/tmp/scp-received"))
				require.NotEmpty(t, sent)
				require.NotEmpty(t, received)
				assert.Equal(t, sent[0], received[0])
			},
		},
		{
			name: "connection EXEC with large output",
			run: func(t *testing.T, environment *Environment, device *models.Device) {
				t.Helper()

				config := &ssh.ClientConfig{
					User: deviceSSHID(device),
					Auth: []ssh.AuthMethod{
						ssh.PublicKeys(environment.signer),
					},
					HostKeyCallback: ssh.InsecureIgnoreHostKey(), //nolint:gosec
				}

				conn, err := ssh.Dial("tcp", environment.services.SSHAddress(), config)
				require.NoError(t, err)
				defer conn.Close() //nolint:errcheck

				sess, err := conn.NewSession()
				require.NoError(t, err)
				defer sess.Close() //nolint:errcheck

				output, err := sess.Output("yes X | tr -d '\n' | head -c 1048576")
				require.NoError(t, err)

				assert.Len(t, output, 1024*1024)
				for _, b := range output {
					assert.Equal(t, byte('X'), b)
				}
			},
		},
		{
			name: "connection EXEC with separate stdout and stderr",
			run: func(t *testing.T, environment *Environment, device *models.Device) {
				t.Helper()

				config := &ssh.ClientConfig{
					User: deviceSSHID(device),
					Auth: []ssh.AuthMethod{
						ssh.PublicKeys(environment.signer),
					},
					HostKeyCallback: ssh.InsecureIgnoreHostKey(), //nolint:gosec
				}

				conn, err := ssh.Dial("tcp", environment.services.SSHAddress(), config)
				require.NoError(t, err)
				defer conn.Close() //nolint:errcheck

				sess, err := conn.NewSession()
				require.NoError(t, err)
				defer sess.Close() //nolint:errcheck

				var stdout, stderr bytes.Buffer

				sess.Stdout = &stdout
				sess.Stderr = &stderr

				err = sess.Run(`echo -n "to stdout"; echo -n "to stderr" 1>&2; exit 7`)

				var status *ssh.ExitError

				require.ErrorAs(t, err, &status)
				assert.Equal(t, 7, status.ExitStatus())

				assert.Equal(t, "to stdout", stdout.String())
				assert.Equal(t, "to stderr", stderr.String())
			},
		},
		{
			name: "connection EXEC delivers a signal to the command",
			run: func(t *testing.T, environment *Environment, device *models.Device) {
				t.Helper()

				sess, output := startInterruptReporter(t, environment.services, environment.signer, device, false)

				require.NoError(t, sess.Signal(ssh.SIGINT))

				requireInterruptReported(t, sess, output)
			},
		},
		{
			name: "connection EXEC with Pty delivers a signal to the command",
			run: func(t *testing.T, environment *Environment, device *models.Device) {
				t.Helper()

				sess, output := startInterruptReporter(t, environment.services, environment.signer, device, true)

				require.NoError(t, sess.Signal(ssh.SIGINT))

				requireInterruptReported(t, sess, output)
			},
		},
		{
			name: "connection SHELL with Pty delivers a signal to the foreground job",
			run: func(t *testing.T, environment *Environment, device *models.Device) {
				t.Helper()

				sess := openSignalSession(t, environment.services, environment.signer, device, true)

				stdin, err := sess.StdinPipe()
				require.NoError(t, err)

				stdout, err := sess.StdoutPipe()
				require.NoError(t, err)

				require.NoError(t, sess.Shell())

				_, err = io.WriteString(stdin, `signal-reporter INT; echo "STA""TUS=$?"; exit`+"\n")
				require.NoError(t, err)

				output := bufio.NewReader(stdout)

				var seen strings.Builder
				for !strings.Contains(seen.String(), "READY") {
					line, err := output.ReadString('\n')
					seen.WriteString(line)
					require.NoError(t, err, "the job never started: %q", seen.String())
				}

				require.NoError(t, sess.Signal(ssh.SIGINT))

				rest := readRemaining(t, output, "the shell never finished the line after the signal")
				assert.Contains(t, rest, "GOT-INT")
				assert.Contains(t, rest, "STATUS=3")
			},
		},
		{
			name: "connection EXEC answers a signal that asks for a reply",
			run: func(t *testing.T, environment *Environment, device *models.Device) {
				t.Helper()

				sess, output := startInterruptReporter(t, environment.services, environment.signer, device, false)

				replied := make(chan bool, 1)

				go func() {
					ok, err := sess.SendRequest("signal", true, ssh.Marshal(struct{ Signal string }{Signal: string(ssh.SIGINT)}))
					replied <- err == nil && ok
				}()

				select {
				case ok := <-replied:
					assert.True(t, ok, "the signal was not accepted")
				case <-time.After(10 * time.Second):
					t.Fatal("a signal asking for a reply never got one")
				}

				requireInterruptReported(t, sess, output)
			},
		},
		{
			name: "connection EXEC with large stderr",
			run: func(t *testing.T, environment *Environment, device *models.Device) {
				t.Helper()

				config := &ssh.ClientConfig{
					User: deviceSSHID(device),
					Auth: []ssh.AuthMethod{
						ssh.PublicKeys(environment.signer),
					},
					HostKeyCallback: ssh.InsecureIgnoreHostKey(), //nolint:gosec
				}

				conn, err := ssh.Dial("tcp", environment.services.SSHAddress(), config)
				require.NoError(t, err)
				defer conn.Close() //nolint:errcheck

				sess, err := conn.NewSession()
				require.NoError(t, err)
				defer sess.Close() //nolint:errcheck

				var stdout, stderr bytes.Buffer

				sess.Stdout = &stdout
				sess.Stderr = &stderr

				done := make(chan error, 1)

				go func() {
					done <- sess.Run("yes E | tr -d '\n' | head -c 1048576 1>&2; echo -n done")
				}()

				select {
				case err := <-done:
					require.NoError(t, err)
				case <-time.After(60 * time.Second):
					t.Fatal("writing 1MiB to stderr never completed")
				}

				assert.Equal(t, 1024*1024, stderr.Len())
				assert.Equal(t, "done", stdout.String())
			},
		},
		{
			name: "connection SHELL with Pty reports the exit code",
			run: func(t *testing.T, environment *Environment, device *models.Device) {
				t.Helper()

				config := &ssh.ClientConfig{
					User: deviceSSHID(device),
					Auth: []ssh.AuthMethod{
						ssh.PublicKeys(environment.signer),
					},
					HostKeyCallback: ssh.InsecureIgnoreHostKey(), //nolint:gosec
				}

				conn, err := ssh.Dial("tcp", environment.services.SSHAddress(), config)
				require.NoError(t, err)
				defer conn.Close() //nolint:errcheck

				sess, err := conn.NewSession()
				require.NoError(t, err)
				defer sess.Close() //nolint:errcheck

				stdin, err := sess.StdinPipe()
				require.NoError(t, err)

				require.NoError(t, sess.RequestPty("xterm", 24, 80, ssh.TerminalModes{}))
				require.NoError(t, sess.Shell())

				_, err = io.WriteString(stdin, "exit 42\n")
				require.NoError(t, err)

				done := make(chan error, 1)

				go func() { done <- sess.Wait() }()

				select {
				case err := <-done:
					var status *ssh.ExitError

					require.ErrorAs(t, err, &status)
					assert.Equal(t, 42, status.ExitStatus())
				case <-time.After(60 * time.Second):
					t.Fatal("a shell with a pty never reported an exit status")
				}
			},
		},
		{
			name: "connection SHELL with Pty reports a successful exit",
			run: func(t *testing.T, environment *Environment, device *models.Device) {
				t.Helper()

				config := &ssh.ClientConfig{
					User: deviceSSHID(device),
					Auth: []ssh.AuthMethod{
						ssh.PublicKeys(environment.signer),
					},
					HostKeyCallback: ssh.InsecureIgnoreHostKey(), //nolint:gosec
				}

				conn, err := ssh.Dial("tcp", environment.services.SSHAddress(), config)
				require.NoError(t, err)
				defer conn.Close() //nolint:errcheck

				sess, err := conn.NewSession()
				require.NoError(t, err)
				defer sess.Close() //nolint:errcheck

				stdin, err := sess.StdinPipe()
				require.NoError(t, err)

				require.NoError(t, sess.RequestPty("xterm", 24, 80, ssh.TerminalModes{}))
				require.NoError(t, sess.Shell())

				_, err = io.WriteString(stdin, "exit 0\n")
				require.NoError(t, err)

				done := make(chan error, 1)

				go func() { done <- sess.Wait() }()

				select {
				case err := <-done:
					require.NoError(t, err)
				case <-time.After(60 * time.Second):
					t.Fatal("a shell with a pty never reported an exit status")
				}
			},
		},
		{
			name: "connection SHELL without Pty",
			run: func(t *testing.T, environment *Environment, device *models.Device) {
				t.Helper()

				config := &ssh.ClientConfig{
					User: deviceSSHID(device),
					Auth: []ssh.AuthMethod{
						ssh.PublicKeys(environment.signer),
					},
					HostKeyCallback: ssh.InsecureIgnoreHostKey(), //nolint:gosec
				}

				conn, err := ssh.Dial("tcp", environment.services.SSHAddress(), config)
				require.NoError(t, err)
				defer conn.Close() //nolint:errcheck

				sess, err := conn.NewSession()
				require.NoError(t, err)
				defer sess.Close() //nolint:errcheck

				stdin, err := sess.StdinPipe()
				require.NoError(t, err)

				var stdout, stderr bytes.Buffer

				sess.Stdout = &stdout
				sess.Stderr = &stderr

				require.NoError(t, sess.Shell())

				_, err = io.WriteString(stdin, "echo -n heredoc_out\necho -n heredoc_err 1>&2\nexit 5\n")
				require.NoError(t, err)
				require.NoError(t, stdin.Close())

				done := make(chan error, 1)

				go func() { done <- sess.Wait() }()

				select {
				case err := <-done:
					var status *ssh.ExitError

					require.ErrorAs(t, err, &status)
					assert.Equal(t, 5, status.ExitStatus())
				case <-time.After(60 * time.Second):
					t.Fatal("a shell without a pty never produced anything")
				}

				assert.Equal(t, "heredoc_out", stdout.String())
				assert.Equal(t, "heredoc_err", stderr.String())
			},
		},
		{
			name: "connection keepalive reaches the client",
			run: func(t *testing.T, environment *Environment, device *models.Device) {
				t.Helper()

				reqs := openShellWatchingGlobalRequests(t, t.Context(), environment.services.SSHAddress(), &ssh.ClientConfig{
					User: deviceSSHID(device),
					Auth: []ssh.AuthMethod{
						ssh.PublicKeys(environment.signer),
					},
					HostKeyCallback: ssh.InsecureIgnoreHostKey(), //nolint:gosec
				})

				awaitGlobalRequest(t, reqs, "keepalive", 60*time.Second)
			},
		},
		{
			name: "connection EXEC with environment variables",
			run: func(t *testing.T, environment *Environment, device *models.Device) {
				t.Helper()

				config := &ssh.ClientConfig{
					User: deviceSSHID(device),
					Auth: []ssh.AuthMethod{
						ssh.PublicKeys(environment.signer),
					},
					HostKeyCallback: ssh.InsecureIgnoreHostKey(), //nolint:gosec
				}

				conn, err := ssh.Dial("tcp", environment.services.SSHAddress(), config)
				require.NoError(t, err)
				defer conn.Close() //nolint:errcheck

				sess, err := conn.NewSession()
				require.NoError(t, err)
				defer sess.Close() //nolint:errcheck

				err = sess.Setenv("LC_TEST", "allowed")
				require.NoError(t, err)

				err = sess.Setenv("TEST_VAR1", "blocked")
				require.NoError(t, err)

				output, err := sess.Output("echo -n $LC_TEST:$TEST_VAR1")
				require.NoError(t, err)

				assert.Equal(t, "allowed:", string(output))
			},
		},
		{
			name: "terminal window size change",
			run: func(t *testing.T, environment *Environment, device *models.Device) {
				t.Helper()

				config := &ssh.ClientConfig{
					User: deviceSSHID(device),
					Auth: []ssh.AuthMethod{
						ssh.PublicKeys(environment.signer),
					},
					HostKeyCallback: ssh.InsecureIgnoreHostKey(), //nolint:gosec
				}

				conn, err := ssh.Dial("tcp", environment.services.SSHAddress(), config)
				require.NoError(t, err)
				defer conn.Close() //nolint:errcheck

				sess, err := conn.NewSession()
				require.NoError(t, err)
				defer sess.Close() //nolint:errcheck

				err = sess.RequestPty("xterm", 24, 80, ssh.TerminalModes{
					ssh.ECHO: 0,
				})
				require.NoError(t, err)

				require.NoError(t, sess.WindowChange(48, 160))

				resize := startResizeReporter(t, sess)

				for _, size := range [][2]int{{40, 120}, {24, 80}, {50, 200}, {30, 100}} {
					resize(size[0], size[1])
				}
			},
		},
		{
			name: "connection EXEC with invalid command",
			run: func(t *testing.T, environment *Environment, device *models.Device) {
				t.Helper()

				config := &ssh.ClientConfig{
					User: deviceSSHID(device),
					Auth: []ssh.AuthMethod{
						ssh.PublicKeys(environment.signer),
					},
					HostKeyCallback: ssh.InsecureIgnoreHostKey(), //nolint:gosec
				}

				conn, err := ssh.Dial("tcp", environment.services.SSHAddress(), config)
				require.NoError(t, err)
				defer conn.Close() //nolint:errcheck

				sess, err := conn.NewSession()
				require.NoError(t, err)
				defer sess.Close() //nolint:errcheck

				_, err = sess.Output("this-command-does-not-exist")
				require.Error(t, err)

				var exitErr *ssh.ExitError
				require.ErrorAs(t, err, &exitErr)
				assert.NotEqual(t, 0, exitErr.ExitStatus())
			},
		},
		{
			name: "handling special characters in commands and output",
			run: func(t *testing.T, environment *Environment, device *models.Device) {
				t.Helper()

				config := &ssh.ClientConfig{
					User: deviceSSHID(device),
					Auth: []ssh.AuthMethod{
						ssh.PublicKeys(environment.signer),
					},
					HostKeyCallback: ssh.InsecureIgnoreHostKey(), //nolint:gosec
				}

				conn, err := ssh.Dial("tcp", environment.services.SSHAddress(), config)
				require.NoError(t, err)
				defer conn.Close() //nolint:errcheck

				sess, err := conn.NewSession()
				require.NoError(t, err)
				defer sess.Close() //nolint:errcheck

				specialChars := "!@#$%^&*()_+{}[]|;:,.<>/?`~"
				output, err := sess.Output(fmt.Sprintf("echo -n '%s'", specialChars))
				require.NoError(t, err)

				assert.Equal(t, specialChars, string(output))

				unicodeChars := "こんにちは世界 ñáéíóú 你好世界"
				sess2, err := conn.NewSession()
				require.NoError(t, err)
				defer sess2.Close() //nolint:errcheck

				output, err = sess2.Output(fmt.Sprintf("echo -n '%s'", unicodeChars))
				require.NoError(t, err)

				assert.Equal(t, unicodeChars, string(output))
			},
		},
		{
			name: "multiple concurrent SSH sessions",
			run: func(t *testing.T, environment *Environment, device *models.Device) {
				t.Helper()

				config := &ssh.ClientConfig{
					User: deviceSSHID(device),
					Auth: []ssh.AuthMethod{
						ssh.PublicKeys(environment.signer),
					},
					HostKeyCallback: ssh.InsecureIgnoreHostKey(), //nolint:gosec
				}

				const numConnections = 5
				var wg sync.WaitGroup
				errors := make(chan error, numConnections)

				for i := range numConnections {
					wg.Add(1)
					go func(id int) {
						defer wg.Done()

						conn, err := ssh.Dial("tcp", environment.services.SSHAddress(), config)
						if err != nil {
							errors <- fmt.Errorf("connection %d failed: %w", id, err)

							return
						}
						defer conn.Close() //nolint:errcheck

						sess, err := conn.NewSession()
						if err != nil {
							errors <- fmt.Errorf("session %d failed: %w", id, err)

							return
						}
						defer sess.Close() //nolint:errcheck

						expected := fmt.Sprintf("session-%d", id)
						output, err := sess.Output(fmt.Sprintf("echo -n '%s'", expected))
						if err != nil {
							errors <- fmt.Errorf("command %d failed: %w", id, err)

							return
						}

						if string(output) != expected {
							errors <- fmt.Errorf("unexpected output from session %d: got %q, want %q", id, string(output), expected)
						}
					}(i)
				}

				wg.Wait()
				close(errors)

				for err := range errors {
					require.NoError(t, err)
				}
			},
		},
		{
			name: "connection verifies the server's host key",
			run: func(t *testing.T, environment *Environment, device *models.Device) {
				t.Helper()

				hostKey := environment.services.SSHHostKey(t)

				conn := dialClient(t, t.Context(), environment.services.SSHAddress(), &ssh.ClientConfig{
					User:            deviceSSHID(device),
					Auth:            []ssh.AuthMethod{ssh.PublicKeys(environment.signer)},
					HostKeyCallback: ssh.FixedHostKey(hostKey),
				})
				_ = conn.Close()

				impostor, _ := newSigner(t)

				err := handshakeWith(t.Context(), environment.services.SSHAddress(), &ssh.ClientConfig{
					User:            deviceSSHID(device),
					Auth:            []ssh.AuthMethod{ssh.PublicKeys(environment.signer)},
					HostKeyCallback: ssh.FixedHostKey(impostor.PublicKey()),
				})
				require.ErrorContains(t, err, "ssh: host key mismatch")
			},
		},
		{
			name: "connection with subsystem request (sftp)",
			run: func(t *testing.T, environment *Environment, device *models.Device) {
				t.Helper()

				config := &ssh.ClientConfig{
					User: deviceSSHID(device),
					Auth: []ssh.AuthMethod{
						ssh.PublicKeys(environment.signer),
					},
					HostKeyCallback: ssh.InsecureIgnoreHostKey(), //nolint:gosec
				}

				conn, err := ssh.Dial("tcp", environment.services.SSHAddress(), config)
				require.NoError(t, err)
				defer conn.Close() //nolint:errcheck

				sess, err := conn.NewSession()
				require.NoError(t, err)
				defer sess.Close() //nolint:errcheck

				err = sess.RequestSubsystem("sftp")
				require.NoError(t, err)

				stdin, err := sess.StdinPipe()
				require.NoError(t, err)

				stdout, err := sess.StdoutPipe()
				require.NoError(t, err)

				initPacket := []byte{0, 0, 0, 5, 1, 0, 0, 0, 3} // SSH_FXP_INIT with version 3
				_, err = stdin.Write(initPacket)
				require.NoError(t, err)

				response := make([]byte, 9)
				n, err := stdout.Read(response)
				require.NoError(t, err)
				assert.Equal(t, 9, n)
				assert.Equal(t, byte(2), response[4]) // SSH_FXP_VERSION
			},
		},
		{
			name: "connection with pseudo-terminal modes",
			run: func(t *testing.T, environment *Environment, device *models.Device) {
				t.Helper()

				config := &ssh.ClientConfig{
					User: deviceSSHID(device),
					Auth: []ssh.AuthMethod{
						ssh.PublicKeys(environment.signer),
					},
					HostKeyCallback: ssh.InsecureIgnoreHostKey(), //nolint:gosec
				}

				conn, err := ssh.Dial("tcp", environment.services.SSHAddress(), config)
				require.NoError(t, err)
				defer conn.Close() //nolint:errcheck

				sess, err := conn.NewSession()
				require.NoError(t, err)
				defer sess.Close() //nolint:errcheck

				modes := ssh.TerminalModes{
					ssh.ECHO:  0,
					ssh.VINTR: 0x07,
				}

				require.NoError(t, sess.RequestPty("xterm-256color", 24, 80, modes))

				output, err := sess.Output("stty -a")
				require.NoError(t, err, string(output))

				settings := strings.Fields(string(output))
				assert.Contains(t, settings, "-echo", "echo is still on: %s", output)
				assert.Contains(t, string(output), "intr = ^G", "the interrupt character is not the one asked for: %s", output)
			},
		},
	}

	ctx := context.Background()

	compose := newSSHEnvironment(t, ctx, legacyNamespace)
	signer := registerDeviceKey(t, ctx, compose)

	for _, tc := range tests {
		test := tc
		t.Run(test.name, func(tt *testing.T) {
			opts := append([]NewAgentContainerOption{
				NewAgentContainerWithConnectionVersion(connectionVersion),
			}, test.options...)

			agent, device := startAcceptedAgent(tt, ctx, compose, opts...)

			test.run(tt, &Environment{
				services: compose,
				agent:    agent,
				signer:   signer,
			}, device)
		})
	}
}

func openSignalSession(t *testing.T, compose *environment.DockerCompose, signer ssh.Signer, device *models.Device, pty bool) *ssh.Session {
	t.Helper()

	conn := dialDevice(t, t.Context(), compose, device, signer)
	t.Cleanup(func() { _ = conn.Close() })

	sess, err := conn.NewSession()
	require.NoError(t, err)
	t.Cleanup(func() { _ = sess.Close() })

	if pty {
		require.NoError(t, sess.RequestPty("xterm", 24, 80, ssh.TerminalModes{}))
	}

	return sess
}

func startInterruptReporter(t *testing.T, compose *environment.DockerCompose, signer ssh.Signer, device *models.Device, pty bool) (*ssh.Session, *bufio.Reader) {
	t.Helper()

	sess := openSignalSession(t, compose, signer, device, pty)

	stdout, err := sess.StdoutPipe()
	require.NoError(t, err)

	require.NoError(t, sess.Start("exec signal-reporter INT"))

	output := bufio.NewReader(stdout)

	line, err := output.ReadString('\n')
	require.NoError(t, err)
	require.Equal(t, "READY", strings.TrimSpace(line))

	return sess, output
}

func requireInterruptReported(t *testing.T, sess *ssh.Session, output *bufio.Reader) {
	t.Helper()

	rest := readRemaining(t, output, "the command never reported the interrupt")

	var status *ssh.ExitError

	require.ErrorAs(t, sess.Wait(), &status, "the command finished without reporting the interrupt: %q", rest)
	assert.Equal(t, 3, status.ExitStatus())
	assert.Equal(t, "GOT-INT", strings.TrimSpace(rest))
}

func readRemaining(t *testing.T, output io.Reader, never string) string {
	t.Helper()

	type read struct {
		rest []byte
		err  error
	}

	finished := make(chan read, 1)

	go func() {
		rest, err := io.ReadAll(output)
		finished <- read{rest, err}
	}()

	select {
	case got := <-finished:
		require.NoError(t, got.err)

		return string(got.rest)
	case <-time.After(30 * time.Second):
		t.Fatal(never)

		return ""
	}
}

func newSigner(t *testing.T) (ssh.Signer, string) {
	t.Helper()

	key, err := rsa.GenerateKey(rand.Reader, 2048)
	require.NoError(t, err)

	pub, err := ssh.NewPublicKey(&key.PublicKey)
	require.NoError(t, err)

	signer, err := ssh.NewSignerFromKey(key)
	require.NoError(t, err)

	return signer, string(ssh.MarshalAuthorizedKey(pub))
}

func registerDeviceKey(t *testing.T, ctx context.Context, compose *environment.DockerCompose) ssh.Signer {
	t.Helper()

	signer, data := newSigner(t)

	registerAuthorizedKey(t, ctx, compose, "e2e", data)

	return signer
}

func registerAuthorizedKey(t *testing.T, ctx context.Context, compose *environment.DockerCompose, name, data string) {
	t.Helper()

	resp, err := compose.R(ctx).
		SetBody(&requests.PublicKeyCreate{
			Name:     name,
			Username: ".*",
			Data:     []byte(data),
			Filter: requests.PublicKeyFilter{
				Hostname: ".*",
			},
		}).
		Post("/api/sshkeys/public-keys")
	require.NoError(t, err)
	require.Equal(t, 200, resp.StatusCode(), resp.String())
}

func skipUnlessAgentAcceptsPasswords(t *testing.T) {
	t.Helper()

	if os.Getenv("SHELLHUB_E2E_AGENT_PASSWORD") == "false" {
		t.Skip("SHELLHUB_E2E_AGENT_PASSWORD=false: the agent under test does not accept passwords")
	}
}

func deviceSSHID(device *models.Device) string {
	return ShellHubAgentUsername + "@" + ShellHubNamespaceName + "." + device.Name
}

type namespaceModel bool

const (
	legacyNamespace namespaceModel = true
	newNamespace    namespaceModel = false
)

func newSSHEnvironment(t *testing.T, ctx context.Context, model namespaceModel) *environment.DockerCompose {
	t.Helper()

	return newConfiguredSSHEnvironment(t, ctx, environment.New(t, run), model)
}

func newConfiguredSSHEnvironment(t *testing.T, ctx context.Context, configurator *environment.DockerComposeConfigurator, model namespaceModel) *environment.DockerCompose {
	t.Helper()

	compose := configurator.Up(ctx)
	t.Cleanup(compose.Down)

	compose.NewUser(t, ShellHubUsername, ShellHubEmail, ShellHubPassword)
	ownNamespace(t, ctx, compose, model)

	return compose
}

var harnessProvisioningKeys sync.Map

const harnessProvisioningKeyName = "harness"

func useManualProvisioningKey(t *testing.T, compose *environment.DockerCompose) string {
	t.Helper()

	key := compose.CreateProvisioningKey(t, &requests.CreateProvisioningKey{
		Name: harnessProvisioningKeyName,
		Mode: string(models.ProvisioningKeyModeManual),
	})
	harnessProvisioningKeys.Store(compose, key.Key)

	return key.Key
}

func harnessProvisioningKey(compose *environment.DockerCompose) (string, bool) {
	key, ok := harnessProvisioningKeys.Load(compose)
	if !ok {
		return "", false
	}

	return key.(string), true //nolint:forcetypeassert // the map holds nothing but the keys useManualProvisioningKey stores
}

func requireHarnessProvisioningKey(t *testing.T, compose *environment.DockerCompose) string {
	t.Helper()

	key, ok := harnessProvisioningKey(compose)
	require.True(t, ok, "the stack's namespace is legacy and has no provisioning key")

	return key
}

func withHarnessCredential(compose *environment.DockerCompose) NewAgentContainerOption {
	return func(envs map[string]string) {
		if key, ok := harnessProvisioningKey(compose); ok {
			NewAgentContainerWithProvisioningKey(key)(envs)
		}
	}
}

func harnessDeviceAuthRequest(t *testing.T, compose *environment.DockerCompose, hostname, mac string) requests.DeviceAuth {
	t.Helper()

	req := newDeviceAuthRequest(t, hostname, mac)
	if key, ok := harnessProvisioningKey(compose); ok {
		req.TenantID = ""
		req.ProvisioningKey = key
	}

	return req
}

func ownNamespace(t *testing.T, ctx context.Context, compose *environment.DockerCompose, model namespaceModel) {
	t.Helper()

	if model == legacyNamespace {
		compose.NewLegacyNamespace(t, ShellHubUsername, ShellHubNamespaceName, ShellHubNamespace)
	} else {
		compose.NewNamespace(t, ShellHubUsername, ShellHubNamespaceName, ShellHubNamespace)
	}

	auth := models.UserAuthResponse{}

	require.EventuallyWithT(t, func(tt *assert.CollectT) {
		resp, err := compose.R(ctx).
			SetBody(map[string]string{
				"username": ShellHubUsername,
				"password": ShellHubPassword,
			}).
			SetResult(&auth).
			Post("/api/login")
		assert.Equal(tt, 200, resp.StatusCode())
		assert.NoError(tt, err)
	}, 30*time.Second, 1*time.Second)

	compose.JWT(auth.Token)

	if model == newNamespace {
		useManualProvisioningKey(t, compose)
	}
}

func startAgent(t *testing.T, ctx context.Context, compose *environment.DockerCompose, opts ...NewAgentContainerOption) testcontainers.Container {
	t.Helper()

	opts = append([]NewAgentContainerOption{withHarnessCredential(compose)}, opts...)

	agent, err := NewAgentContainer(ctx, compose.Service(environment.ServiceRelay).GetContainerID(), opts...)
	require.NoError(t, err)

	return runAgent(t, ctx, agent)
}

func startCustomAgent(t *testing.T, ctx context.Context, compose *environment.DockerCompose, customize func(*testcontainers.ContainerRequest), opts ...NewAgentContainerOption) testcontainers.Container {
	t.Helper()

	image, err := buildAgentImage(ctx, "")
	require.NoError(t, err)

	opts = append([]NewAgentContainerOption{withHarnessCredential(compose)}, opts...)

	req, err := agentContainerRequest(compose.Service(environment.ServiceRelay).GetContainerID(), image, opts...)
	require.NoError(t, err)

	customize(&req)

	agent, err := createAgentContainer(ctx, req)
	require.NoError(t, err)

	return runAgent(t, ctx, agent)
}

func runAgent(t *testing.T, ctx context.Context, agent testcontainers.Container) testcontainers.Container {
	t.Helper()

	require.NoError(t, agent.Start(ctx))

	t.Cleanup(func() {
		_ = agent.Stop(context.Background(), nil)
		_ = agent.Terminate(context.Background())
	})

	return agent
}

func startAcceptedAgent(t *testing.T, ctx context.Context, compose *environment.DockerCompose, opts ...NewAgentContainerOption) (testcontainers.Container, *models.Device) {
	t.Helper()

	agent := startAgent(t, ctx, compose, opts...)

	return agent, acceptPendingDevice(t, compose)
}

func acceptPendingDevice(t *testing.T, compose *environment.DockerCompose) *models.Device {
	t.Helper()

	pending := compose.AwaitDeviceWithStatus(t, models.DeviceStatusPending)

	compose.UpdateDeviceStatus(t, pending.UID, environment.DeviceActionAccept)

	device := compose.AwaitDeviceOnline(t, pending.UID)

	return &device
}