diff --git a/CLAUDE.md b/CLAUDE.md index f6f4225b..9bca2399 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -155,7 +155,7 @@ make download-php Tests use github.com/stretchr/testify for assertions. Table-driven tests are preferred with a "cases" slice containing simple test case structs. -Integration tests (`integration-tests/`) call `t.Parallel()`, so they must not share state: no `t.Setenv` or unsynchronized package variables, and use `cmdFactory.useConfig` to change the CLI config file. Write output with `t.Log` (e.g. `mockapi.RequestLogger`), not to stdout or stderr: it is then attributed to its test, and shown only on failure or with `-v`. Tests of environment operations use `setUpProject`, and tests of Git commands (push, clone, fetch) use `setUpGitProject`, which also serves a bare repository over the mock SSH server. Tests of other SSH commands use `serveSSH` with an app from `sshEnv`, and fake remote executables from `remoteEnv`. +Integration tests (`integration-tests/`) call `t.Parallel()`, so they must not share state: no `t.Setenv` or unsynchronized package variables, and use `cmdFactory.useConfig` to change the CLI config file. Write output with `t.Log` (e.g. `mockapi.RequestLogger`), not to stdout or stderr: it is then attributed to its test, and shown only on failure or with `-v`. Tests of environment operations use `setUpProject`, and tests of Git commands (push, clone, fetch) use `setUpGitProject`, which also serves a bare repository over the mock SSH server. Tests of other SSH commands use `serveSSH` with an app from `sshEnv`, and fake remote executables from `remoteEnv`. The mock SSH server can also serve SFTP (`SFTPDir`) and forward connections (`DialTCP`, `ListenUnix`); tests of commands that start background processes (tunnels) must make sure they exit. ### Configuration diff --git a/go.mod b/go.mod index 6f4b528d..0eb8b999 100644 --- a/go.mod +++ b/go.mod @@ -15,6 +15,7 @@ require ( github.com/godbus/dbus/v5 v5.2.3 github.com/gofrs/flock v0.13.1 github.com/oklog/ulid/v2 v2.1.2 + github.com/pkg/sftp v1.13.11 github.com/platformsh/platformify v0.5.0 github.com/spf13/cobra v1.10.2 github.com/spf13/pflag v1.0.10 @@ -91,6 +92,7 @@ require ( github.com/kevinburke/ssh_config v1.6.0 // indirect github.com/klauspost/compress v1.18.7 // indirect github.com/klauspost/pgzip v1.2.6 // indirect + github.com/kr/fs v0.1.0 // indirect github.com/leodido/go-urn v1.5.0 // indirect github.com/lucasb-eyer/go-colorful v1.4.1 // indirect github.com/mailru/easyjson v0.9.2 // indirect diff --git a/go.sum b/go.sum index 698cd3da..e16d23db 100644 --- a/go.sum +++ b/go.sum @@ -173,6 +173,8 @@ github.com/klauspost/compress v1.18.7/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl github.com/klauspost/cpuid v1.2.0/go.mod h1:Pj4uuM528wm8OyEC2QMXAi2YiTZ96dNQPGgoMS4s3ek= github.com/klauspost/pgzip v1.2.6 h1:8RXeL5crjEUFnR2/Sn6GJNWtSQ3Dk8pq4CL3jvdDyjU= github.com/klauspost/pgzip v1.2.6/go.mod h1:Ch1tH69qFZu15pkjo5kYi6mth2Zzwzt50oCQKQE9RUs= +github.com/kr/fs v0.1.0 h1:Jskdu9ieNAYnjxsi0LbQp1ulIKZV1LAFgK1tWhpZgl8= +github.com/kr/fs v0.1.0/go.mod h1:FFnZGqtBN9Gxj7eW1uZ42v5BccTP0vu6NEaFoC2HwRg= github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo= github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= @@ -226,6 +228,8 @@ github.com/pjbgf/sha1cd v0.7.0 h1:ZRNPKHj+gfkLBf0KJv/p1Hmz+7bqCT5o311YX0nq+DA= github.com/pjbgf/sha1cd v0.7.0/go.mod h1:pKR5Li+qTCo+ebqITZfwPlJGoCFCvSO00qxjbNtTUFQ= github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4= github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= +github.com/pkg/sftp v1.13.11 h1:0N92SLTB8JqASJB14ZLHHzFnBV8mG9zw4K7jghEFWuE= +github.com/pkg/sftp v1.13.11/go.mod h1:uNkH9roSXglNJqM+glJJi+TQXQUm0fXFWqCFmT8hsN0= github.com/platformsh/platformify v0.5.0 h1:i9cRbM0K1f14Yr0dJ6dhYq9BcmZb8fsx/mpYSnw8ssA= github.com/platformsh/platformify v0.5.0/go.mod h1:fgmCcfQfHbhe1oXsIdIhpnniyZu8IdIMOlcBAa/ygic= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= diff --git a/integration-tests/environment_scp_test.go b/integration-tests/environment_scp_test.go new file mode 100644 index 00000000..874f47ff --- /dev/null +++ b/integration-tests/environment_scp_test.go @@ -0,0 +1,53 @@ +package tests + +import ( + "os/exec" + "path/filepath" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/upsun/cli/pkg/mockapi" + "github.com/upsun/cli/pkg/mockssh" +) + +func TestEnvironmentScp(t *testing.T) { + t.Parallel() + if _, err := exec.LookPath("scp"); err != nil { + t.Skip("scp is not installed") + } + remoteDir := t.TempDir() + writeFiles(t, remoteDir, map[string]string{"remote.txt": "remote", "logs/a.log": "a", "logs/b.log": "b"}) + + p := setUpProject(t, sshEnv(&mockapi.Deployment{ + WebApps: map[string]mockapi.App{ + "app": {Name: "app", Type: "golang:1.23", Size: "M", Mounts: map[string]mockapi.Mount{}}, + }, + })) + // Older scp clients run "scp -t" or "scp -f" instead of using SFTP. + sshServer := p.serveSSH(t, mockssh.ExecHandler(remoteDir, nil)) + sshServer.SFTPDir = remoteDir + f := p.f + f.dir = t.TempDir() + writeFiles(t, f.dir, map[string]string{"local.txt": "local"}) + args := []string{"scp", "-p", layoutProjectID, "-e", "main"} + + f.Run(append(args, "--", "local.txt", "remote:uploaded.txt")...) + assert.Equal(t, "local", readFile(t, filepath.Join(remoteDir, "uploaded.txt"))) + + f.Run(append(args, "--", "remote:remote.txt", "downloaded.txt")...) + assert.Equal(t, "remote", readFile(t, filepath.Join(f.dir, "downloaded.txt"))) + + f.Run(append(args, "-r", "--", "remote:logs", "logs")...) + assert.Equal(t, "a", readFile(t, filepath.Join(f.dir, "logs", "a.log"))) + assert.Equal(t, "b", readFile(t, filepath.Join(f.dir, "logs", "b.log"))) + + _, stdErr, err := f.RunCombinedOutput(append(args, "--", "remote:missing.txt", ".")...) + require.Error(t, err) + assert.Contains(t, stdErr, "missing.txt") + + _, stdErr, err = f.RunCombinedOutput(append(args, "--", "local.txt", "other.txt")...) + require.Error(t, err) + assert.Contains(t, stdErr, `At least one argument needs to contain the "remote:" prefix`) +} diff --git a/integration-tests/environment_xdebug_test.go b/integration-tests/environment_xdebug_test.go index 2f8178c7..aa5cdfb2 100644 --- a/integration-tests/environment_xdebug_test.go +++ b/integration-tests/environment_xdebug_test.go @@ -1,10 +1,17 @@ package tests import ( + "io" + "net" "net/http/httptest" + "os" + "path/filepath" "strconv" + "strings" "sync" + "sync/atomic" "testing" + "time" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -121,3 +128,91 @@ func TestEnvironmentXdebug(t *testing.T) { mu.Unlock() } } + +func TestEnvironmentXdebugTunnel(t *testing.T) { + t.Parallel() + p := setUpProject(t, sshEnv(&mockapi.Deployment{ + WebApps: map[string]mockapi.App{ + "app": {Name: "app", Type: "php:8.4", Size: "M", Mounts: map[string]mockapi.Mount{}, + Runtime: map[string]any{"xdebug": map[string]any{"idekey": "secret_key"}}}, + }, + })) + sshServer := p.serveSSH(t, func(_ ssh.ConnMetadata, _ string, _ mockssh.CommandIO) int { return 0 }) + + // The server listens at a local socket instead of /run/xdebug-tunnel.sock. It is + // in /tmp, as socket paths are limited to about 100 characters. + socketDir, err := os.MkdirTemp("/tmp", "xdebug") + require.NoError(t, err) + t.Cleanup(func() { _ = os.RemoveAll(socketDir) }) + socket := filepath.Join(socketDir, "xdebug.sock") + var requestedPath atomic.Value + sshServer.ListenUnix = func(path string) (net.Listener, error) { + requestedPath.Store(path) + return net.Listen("unix", socket) + } + + // The IDE listens locally. + ide, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + t.Cleanup(func() { _ = ide.Close() }) + port := tcpPort(t, ide) + received := make(chan string, 1) + go func() { + c, err := ide.Accept() + if err != nil { + return + } + defer c.Close() + b, _ := io.ReadAll(c) + received <- string(b) + }() + + cmd := p.f.buildCommand("xdebug", "-p", layoutProjectID, "-e", "main", "--port", strconv.Itoa(port)) + var stdErr syncBuffer + cmd.Stderr = &stdErr + require.NoError(t, cmd.Start()) + var waitErr error + exited := make(chan struct{}) + go func() { + waitErr = cmd.Wait() + close(exited) + }() + t.Cleanup(func() { + select { + case <-exited: + case <-time.After(10 * time.Second): + _ = cmd.Process.Kill() + t.Error("xdebug did not exit") + } + p.f.logStderr(stdErr.String()) + }) + + // A connection from the app (Xdebug) to the socket reaches the IDE. + var conn net.Conn + require.Eventually(t, func() bool { + conn, err = net.Dial("unix", socket) + return err == nil + }, 10*time.Second, 50*time.Millisecond) + _, err = io.WriteString(conn, "") + require.NoError(t, err) + require.NoError(t, conn.Close()) + select { + case got := <-received: + assert.Equal(t, "", got) + case <-time.After(10 * time.Second): + t.Fatal("the IDE did not receive a connection") + } + assert.Equal(t, "/run/xdebug-tunnel.sock", requestedPath.Load()) + require.Eventually(t, func() bool { + return strings.Contains(stdErr.String(), "Xdebug tunnel opened at: 127.0.0.1:"+strconv.Itoa(port)) + }, 10*time.Second, 50*time.Millisecond) + + // The tunnel closes with the SSH connection. + require.NoError(t, sshServer.Stop()) + select { + case <-exited: + case <-time.After(10 * time.Second): + t.Fatal("xdebug did not exit after the SSH connection closed") + } + assertExitCode(t, 255, waitErr) +} diff --git a/integration-tests/tests.go b/integration-tests/tests.go index 127c7289..26818400 100644 --- a/integration-tests/tests.go +++ b/integration-tests/tests.go @@ -263,3 +263,21 @@ func waitForServer(t *testing.T, client *http.Client, url string) *http.Response }, 10*time.Second, 50*time.Millisecond, "server did not respond: %s", url) return resp } + +// syncBuffer is a buffer that is safe for concurrent use, e.g. to read a running command's output. +type syncBuffer struct { + mu sync.Mutex + buf bytes.Buffer +} + +func (b *syncBuffer) Write(p []byte) (int, error) { + b.mu.Lock() + defer b.mu.Unlock() + return b.buf.Write(p) +} + +func (b *syncBuffer) String() string { + b.mu.Lock() + defer b.mu.Unlock() + return b.buf.String() +} diff --git a/integration-tests/tunnel_test.go b/integration-tests/tunnel_test.go new file mode 100644 index 00000000..583c34d2 --- /dev/null +++ b/integration-tests/tunnel_test.go @@ -0,0 +1,222 @@ +package tests + +import ( + "encoding/base64" + "encoding/json" + "fmt" + "io" + "net" + "os" + "path/filepath" + "strconv" + "strings" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/upsun/cli/pkg/mockapi" + "github.com/upsun/cli/pkg/mockssh" +) + +// setUpTunnelProject serves an app over SSH, with relationships whose hosts the mock +// SSH server forwards to local services. Each service writes its name to a connection. +func setUpTunnelProject(t *testing.T) *testProject { + t.Helper() + p := setUpProject(t, sshEnv(&mockapi.Deployment{ + WebApps: map[string]mockapi.App{ + "app": {Name: "app", Type: "golang:1.23", Size: "M", Mounts: map[string]mockapi.Mount{}}, + }, + })) + services := map[string]string{} // host:port → local address + for _, name := range []string{"database.internal:3306", "redis.internal:6379"} { + services[name] = serveName(t, strings.Split(name, ".")[0]) + } + sshServer := p.serveSSH(t, mockssh.ExecHandler(t.TempDir(), remoteEnv(t, nil, mockRelationships()))) + sshServer.DialTCP = func(host string, port int) (net.Conn, error) { + addr, ok := services[net.JoinHostPort(host, strconv.Itoa(port))] + if !ok { + return nil, fmt.Errorf("unknown service: %s:%d", host, port) + } + return net.Dial("tcp", addr) + } + return p +} + +// serveName starts a TCP server that writes a name to each connection, returning its address. +func serveName(t *testing.T, name string) string { + t.Helper() + l, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + t.Cleanup(func() { _ = l.Close() }) + go func() { + for { + c, err := l.Accept() + if err != nil { + return + } + _, _ = io.WriteString(c, name) + _ = c.Close() + } + }() + return l.Addr().String() +} + +// readTunnel connects to a local port, retrying until the tunnel is ready, and returns what the service wrote. +func readTunnel(t *testing.T, port int) string { + t.Helper() + var out string + require.EventuallyWithT(t, func(c *assert.CollectT) { + conn, err := net.DialTimeout("tcp", net.JoinHostPort("127.0.0.1", strconv.Itoa(port)), time.Second) + require.NoError(c, err) + defer conn.Close() + _ = conn.SetDeadline(time.Now().Add(5 * time.Second)) + b, err := io.ReadAll(conn) + require.NoError(c, err) + require.NotEmpty(c, b) + out = string(b) + }, 10*time.Second, 100*time.Millisecond) + return out +} + +// runWithFiles runs a command with its output written to files instead of pipes, as background +// processes that it starts inherit them, and waiting for a pipe would wait for those processes. +func runWithFiles(f *cmdFactory, args ...string) (stdOut, stdErr string, err error) { + f.t.Helper() + dir := f.t.TempDir() + stdOutFile, err := os.Create(filepath.Join(dir, "stdout")) + require.NoError(f.t, err) + defer stdOutFile.Close() + stdErrFile, err := os.Create(filepath.Join(dir, "stderr")) + require.NoError(f.t, err) + defer stdErrFile.Close() + cmd := f.buildCommand(args...) + cmd.Stdout = stdOutFile + cmd.Stderr = stdErrFile + f.t.Log("Running:", cmd) + err = cmd.Run() + stdOut, stdErr = readFile(f.t, stdOutFile.Name()), readFile(f.t, stdErrFile.Name()) + f.logStderr(stdErr) + return stdOut, stdErr, err +} + +// freePort returns a local port which is not in use. +func freePort(t *testing.T) int { + t.Helper() + l, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + defer l.Close() + return tcpPort(t, l) +} + +// tcpPort returns the port of a TCP listener. +func tcpPort(t *testing.T, l net.Listener) int { + t.Helper() + addr, ok := l.Addr().(*net.TCPAddr) + require.True(t, ok) + return addr.Port +} + +func TestTunnelOpen(t *testing.T) { + t.Parallel() + p := setUpTunnelProject(t) + f := p.f + args := []string{"-p", layoutProjectID, "-e", "main"} + // The tunnels run in the background until they are closed, or the SSH server stops. + t.Cleanup(func() { _, _, _ = f.RunCombinedOutput("tunnel:close", "--all", "--yes") }) + + _, stdErr, err := runWithFiles(f, append([]string{"tunnel:open"}, args...)...) + require.NoError(t, err) + assert.Contains(t, stdErr, "SSH tunnel opened to database at: mysql://user:@127.0.0.1:") + assert.Contains(t, stdErr, "SSH tunnel opened to redis at: redis://:@127.0.0.1:") + + var relationships map[string][]struct { + Host string `json:"host"` + Port int `json:"port"` + } + encoded := strings.TrimSpace(f.Run(append([]string{"tunnel:info", "--encode"}, args...)...)) + decoded, err := base64.StdEncoding.DecodeString(encoded) + require.NoError(t, err) + require.NoError(t, json.Unmarshal(decoded, &relationships)) + require.Len(t, relationships["database"], 1) + require.Len(t, relationships["redis"], 1) + assert.Equal(t, "127.0.0.1", relationships["database"][0].Host) + + dbPort := relationships["database"][0].Port + redisPort := relationships["redis"][0].Port + assert.Equal(t, "database", readTunnel(t, dbPort)) + assert.Equal(t, "redis", readTunnel(t, redisPort)) + + assertTrimmed(t, strconv.Itoa(dbPort), f.Run(append([]string{"tunnel:info", "-P", "database.0.port"}, args...)...)) + + list := f.Run(append([]string{"tunnel:list", "--format", "csv", "--columns", "port,relationship"}, args...)...) + assert.Equal(t, fmt.Sprintf("Port,Relationship\n%d,database\n%d,redis\n", dbPort, redisPort), list) + + // Opening tunnels again reuses them. + _, stdErr, err = runWithFiles(f, append([]string{"tunnel:open"}, args...)...) + require.NoError(t, err) + assert.Contains(t, stdErr, fmt.Sprintf( + "A tunnel is already opened to the relationship database, at: mysql://user:@127.0.0.1:%d", dbPort)) + + _, stdErr, err = f.RunCombinedOutput(append([]string{"tunnel:close", "--yes"}, args...)...) + require.NoError(t, err) + assert.Contains(t, stdErr, "Close tunnel to relationship database on "+layoutProjectID+"-main--app?") + assert.Contains(t, stdErr, "Close tunnel to relationship redis on "+layoutProjectID+"-main--app?") + + // The tunnel processes exit. + assert.Eventually(t, func() bool { + conn, err := net.Dial("tcp", net.JoinHostPort("127.0.0.1", strconv.Itoa(dbPort))) + if err == nil { + _ = conn.Close() + } + return err != nil + }, 10*time.Second, 100*time.Millisecond) + + _, stdErr, err = f.RunCombinedOutput(append([]string{"tunnel:list"}, args...)...) + assertExitCode(t, 1, err) + assert.Contains(t, stdErr, "No tunnels found.") +} + +func TestTunnelSingle(t *testing.T) { + t.Parallel() + p := setUpTunnelProject(t) + f := p.f + port := freePort(t) + + cmd := f.buildCommand("tunnel:single", "-p", layoutProjectID, "-e", "main", "-r", "redis", + "--port", strconv.Itoa(port)) + var stdErr syncBuffer + cmd.Stderr = &stdErr + require.NoError(t, cmd.Start()) + var waitErr error + exited := make(chan struct{}) + go func() { + waitErr = cmd.Wait() + close(exited) + }() + t.Cleanup(func() { + select { + case <-exited: + case <-time.After(10 * time.Second): + _ = cmd.Process.Kill() + t.Error("tunnel:single did not exit") + } + f.logStderr(stdErr.String()) + }) + + assert.Equal(t, "redis", readTunnel(t, port)) + assertTrimmed(t, strconv.Itoa(port), f.Run("tunnel:info", "-p", layoutProjectID, "-e", "main", + "-P", "redis.0.port")) + + // Closing the tunnel ends the command. + f.Run("tunnel:close", "--all", "--yes") + select { + case <-exited: + // The command reports the stopped process, but it exits successfully. + assert.NoError(t, waitErr) + case <-time.After(10 * time.Second): + t.Fatal("tunnel:single did not exit after tunnel:close") + } + assert.Contains(t, stdErr.String(), fmt.Sprintf("SSH tunnel opened to redis at: redis://:@127.0.0.1:%d", port)) +} diff --git a/legacy/src/Service/TunnelManager.php b/legacy/src/Service/TunnelManager.php index 279a4e46..1a143b68 100644 --- a/legacy/src/Service/TunnelManager.php +++ b/legacy/src/Service/TunnelManager.php @@ -142,8 +142,7 @@ private function saveTunnelInfo(): void * Checks whether a tunnel is already open. * * @return false|Tunnel - * If the tunnel is open, a new Tunnel object is returned with its PID - * set. + * The tunnel with the open tunnel's local port and PID, or false if it is not open. */ public function isOpen(Tunnel $tunnel): false|Tunnel { @@ -157,9 +156,8 @@ public function isOpen(Tunnel $tunnel): false|Tunnel $this->close($t); return false; } - $tunnel->pid = $t->pid; - - return $tunnel; + // The stored metadata may be from an older version, so only the port and PID are used. + return new Tunnel($tunnel->id, $t->localPort, $tunnel->remoteHost, $tunnel->remotePort, $tunnel->metadata, $t->pid); } } diff --git a/legacy/tests/Service/TunnelManagerTest.php b/legacy/tests/Service/TunnelManagerTest.php index c2a91783..0c924560 100644 --- a/legacy/tests/Service/TunnelManagerTest.php +++ b/legacy/tests/Service/TunnelManagerTest.php @@ -302,6 +302,64 @@ public function testIsOpenLoadsStateWhenNotAlreadyLoaded(): void $this->assertSame(getmypid(), $result->pid); } + public function testIsOpenReturnsTheOpenTunnel(): void + { + // A new tunnel gets the next free port, but the open tunnel's port is the one in use. + $id = 'proj1--main--app--database--0'; + $metadata = [ + 'projectId' => 'proj1', + 'environmentId' => 'main', + 'appName' => 'app', + 'relationship' => 'database', + 'serviceKey' => 0, + 'service' => ['scheme' => 'mysql', 'host' => 'database.internal', 'port' => 3306], + ]; + $this->writeTunnelInfo($id, $metadata + [ + 'localPort' => 30000, + 'remoteHost' => 'database.internal', + 'remotePort' => 3306, + 'pid' => getmypid(), + ]); + + $manager = $this->createManager($this->tempDir); + + $result = $manager->isOpen(new Tunnel($id, 30001, 'database.internal', 3306, $metadata)); + + $this->assertInstanceOf(Tunnel::class, $result); + $this->assertSame(30000, $result->localPort); + $this->assertSame(getmypid(), $result->pid); + } + + public function testIsOpenKeepsCurrentMetadata(): void + { + // A stored entry may lack the service metadata, e.g. from an older version. + $id = 'proj1--main--app--database--0'; + $metadata = [ + 'projectId' => 'proj1', + 'environmentId' => 'main', + 'appName' => 'app', + 'relationship' => 'database', + 'serviceKey' => 0, + 'service' => ['scheme' => 'mysql', 'host' => 'database.internal', 'port' => 3306], + ]; + $stored = $metadata; + unset($stored['service']); + $this->writeTunnelInfo($id, $stored + [ + 'localPort' => 30000, + 'remoteHost' => 'database.internal', + 'remotePort' => 3306, + 'pid' => getmypid(), + ]); + + $manager = $this->createManager($this->tempDir); + + $result = $manager->isOpen(new Tunnel($id, 30001, 'database.internal', 3306, $metadata)); + + $this->assertInstanceOf(Tunnel::class, $result); + $this->assertSame(30000, $result->localPort); + $this->assertSame($metadata, $result->metadata); + } + public function testCloseRemovesTunnelFromState(): void { // close() has to rewrite the state file itself. Leaving that to the diff --git a/pkg/mockssh/channels.go b/pkg/mockssh/channels.go new file mode 100644 index 00000000..1aaed9e5 --- /dev/null +++ b/pkg/mockssh/channels.go @@ -0,0 +1,218 @@ +package mockssh + +import ( + "errors" + "io" + "net" + "sync" + "time" + + "github.com/pkg/sftp" + "golang.org/x/crypto/ssh" +) + +func (s *Server) handleConn(conn net.Conn, config *ssh.ServerConfig) { + t := s.t + sshConn, chans, reqs, err := ssh.NewServerConn(conn, config) + if err != nil { + t.Logf("Handshake failed: %v", err) + return + } + defer sshConn.Close() + + t.Logf("Handling SSH connection from %s", sshConn.RemoteAddr()) + + // connClosed stops forwarding, as forwarded connections may stay open after the SSH connection closes. + connClosed := make(chan struct{}) + var wg sync.WaitGroup + defer func() { + close(connClosed) + wg.Wait() + }() + + wg.Go(func() { + for req := range reqs { + if req.Type == "streamlocal-forward@openssh.com" { + s.forwardUnix(sshConn, req, &wg, connClosed) + continue + } + if req.WantReply { + _ = req.Reply(false, nil) + } + } + }) + + for newChannel := range chans { + switch newChannel.ChannelType() { + case "session": + wg.Go(func() { s.handleSession(sshConn, newChannel) }) + case "direct-tcpip": + wg.Go(func() { s.handleDirectTCPIP(newChannel, connClosed) }) + default: + _ = newChannel.Reject(ssh.UnknownChannelType, "unknown channel type") + } + } +} + +// handleSession handles a session channel, which runs a command or the SFTP subsystem. +func (s *Server) handleSession(conn ssh.ConnMetadata, newChannel ssh.NewChannel) { + t := s.t + channel, requests, err := newChannel.Accept() + if err != nil { + t.Errorf("Failed to accept channel: %v", err) + return + } + defer channel.Close() + + exitWithStatus := make(chan int, 1) + go func() { + for req := range requests { + var payload struct{ Value string } + switch { + case req.Type == "exec" && ssh.Unmarshal(req.Payload, &payload) == nil: + _ = req.Reply(true, nil) + t.Logf("Handling command: %s", payload.Value) + exitWithStatus <- s.CommandHandler(conn, payload.Value, CommandIO{ + StdIn: channel, + StdOut: channel, + StdErr: channel.Stderr(), + }) + return + case req.Type == "subsystem" && s.SFTPDir != "" && + ssh.Unmarshal(req.Payload, &payload) == nil && payload.Value == "sftp": + _ = req.Reply(true, nil) + t.Logf("Handling SFTP session in %s", s.SFTPDir) + exitWithStatus <- serveSFTP(channel, s.SFTPDir) + return + default: + if req.WantReply { + _ = req.Reply(false, nil) + } + } + } + }() + + select { + case status := <-exitWithStatus: + _, err = channel.SendRequest("exit-status", false, ssh.Marshal(struct{ Status uint32 }{uint32(status)})) + if err != nil { + t.Errorf("Failed to send exit status: %v", err) + } + case <-time.After(10 * time.Second): + t.Error("Timed out") + } +} + +func serveSFTP(channel ssh.Channel, dir string) int { + server, err := sftp.NewServer(channel, sftp.WithServerWorkingDirectory(dir)) + if err != nil { + return 1 + } + if err := server.Serve(); err != nil && !errors.Is(err, io.EOF) { + return 1 + } + return 0 +} + +// handleDirectTCPIP handles local port forwarding, if DialTCP is set. +func (s *Server) handleDirectTCPIP(newChannel ssh.NewChannel, connClosed <-chan struct{}) { + t := s.t + if s.DialTCP == nil { + _ = newChannel.Reject(ssh.Prohibited, "port forwarding is disabled") + return + } + // See: https://datatracker.ietf.org/doc/html/rfc4254#section-7.2 + var payload struct { + Host string + Port uint32 + OriginHost string + OriginPort uint32 + } + if err := ssh.Unmarshal(newChannel.ExtraData(), &payload); err != nil { + _ = newChannel.Reject(ssh.ConnectionFailed, "invalid payload") + return + } + t.Logf("Forwarding a connection to %s:%d", payload.Host, payload.Port) + target, err := s.DialTCP(payload.Host, int(payload.Port)) + if err != nil { + _ = newChannel.Reject(ssh.ConnectionFailed, err.Error()) + return + } + channel, requests, err := newChannel.Accept() + if err != nil { + _ = target.Close() + return + } + go ssh.DiscardRequests(requests) + proxy(channel, target, connClosed) +} + +// forwardUnix handles a request for remote forwarding of a Unix socket, if ListenUnix is set. +func (s *Server) forwardUnix(conn *ssh.ServerConn, req *ssh.Request, wg *sync.WaitGroup, connClosed <-chan struct{}) { + t := s.t + var payload struct{ Path string } + if s.ListenUnix == nil || ssh.Unmarshal(req.Payload, &payload) != nil { + _ = req.Reply(false, nil) + return + } + l, err := s.ListenUnix(payload.Path) + if err != nil { + t.Logf("Failed to listen at %s: %v", payload.Path, err) + _ = req.Reply(false, nil) + return + } + _ = req.Reply(true, nil) + t.Logf("Forwarding connections from the socket %s", payload.Path) + wg.Go(func() { + <-connClosed + _ = l.Close() + }) + wg.Go(func() { + for { + c, err := l.Accept() + if err != nil { + return + } + wg.Go(func() { + // See: https://github.com/openssh/openssh-portable/blob/master/PROTOCOL + channel, requests, err := conn.OpenChannel("forwarded-streamlocal@openssh.com", + ssh.Marshal(struct{ Path, Reserved string }{Path: payload.Path})) + if err != nil { + _ = c.Close() + return + } + go ssh.DiscardRequests(requests) + proxy(channel, c, connClosed) + }) + } + }) +} + +// proxy copies data in both directions until both sides are closed, or the SSH connection is closed. +func proxy(channel ssh.Channel, conn net.Conn, connClosed <-chan struct{}) { + finished := make(chan struct{}) + defer close(finished) + go func() { + select { + case <-connClosed: + case <-finished: + } + _ = channel.Close() + _ = conn.Close() + }() + done := make(chan struct{}, 2) + go func() { + _, _ = io.Copy(channel, conn) + _ = channel.CloseWrite() + done <- struct{}{} + }() + go func() { + _, _ = io.Copy(conn, channel) + if c, ok := conn.(interface{ CloseWrite() error }); ok { + _ = c.CloseWrite() + } + done <- struct{}{} + }() + <-done + <-done +} diff --git a/pkg/mockssh/server.go b/pkg/mockssh/server.go index d82d2732..8884cb85 100644 --- a/pkg/mockssh/server.go +++ b/pkg/mockssh/server.go @@ -15,7 +15,6 @@ import ( "os/exec" "sync" "testing" - "time" "golang.org/x/crypto/ssh" ) @@ -30,14 +29,32 @@ type Server struct { CertAuthorityKeys []ssh.PublicKey CertChecker ssh.CertChecker + // The following fields must be set before clients connect, as connection handlers read them without locking. + // An optional CommandHandler, which responds to commands sent over SSH. // NewServer will give this a default using ExecHandler, which can also // be reused from custom handlers. CommandHandler CommandHandler + // SFTPDir enables the SFTP subsystem, with this working directory. + SFTPDir string + + // DialTCP enables local port forwarding ("direct-tcpip" channels, as with "ssh -L"), + // by connecting to the requested host and port. Otherwise, forwarding is rejected. + DialTCP func(host string, port int) (net.Conn, error) + + // ListenUnix enables remote forwarding of Unix sockets (as with "ssh -R /path:host:port"), + // by listening at the requested path. Otherwise, forwarding is rejected. + ListenUnix func(path string) (net.Listener, error) + // listener and port are set after Start. listener net.Listener port int + + // conns are the open connections, which Stop closes. + connsMu sync.Mutex + conns map[net.Conn]struct{} + wg sync.WaitGroup } type CommandIO struct { @@ -63,7 +80,7 @@ func NewServer(t *testing.T, authorityEndpoint string) (*Server, error) { return nil, fmt.Errorf("failed to fetch authority keys: %v", err) } - s := &Server{t: t, hostKey: hk} + s := &Server{t: t, hostKey: hk, conns: make(map[net.Conn]struct{})} s.CommandHandler = ExecHandler("", nil) s.CertChecker = s.defaultCertChecker() s.CertAuthorityKeys = keys @@ -106,48 +123,61 @@ func (s *Server) start() error { } t.Logf("Test SSH server listening at %s", listener.Addr()) - go func(l net.Listener) { + s.wg.Go(func() { for { - conn, err := l.Accept() + conn, err := listener.Accept() if err != nil { - if errors.Is(err, net.ErrClosed) { - break + if !errors.Is(err, net.ErrClosed) { + t.Errorf("Failed to accept connection: %v", err) } - t.Errorf("Failed to accept connection: %v", err) - continue + return } - - sshConn, chans, reqs, err := ssh.NewServerConn(conn, config) - if err != nil { - t.Logf("Handshake failed: %v", err) + if !s.trackConn(conn) { + _ = conn.Close() return } - - t.Logf("Handling SSH connection from %s", sshConn.RemoteAddr()) - - wg := sync.WaitGroup{} - wg.Add(1) - go func() { - ssh.DiscardRequests(reqs) - wg.Done() - }() - wg.Add(1) - go func() { - s.handleChannels(sshConn, chans) - wg.Done() - }() - wg.Wait() + s.wg.Go(func() { + defer s.untrackConn(conn) + s.handleConn(conn, config) + }) } - }(s.listener) + }) return nil } +// Stop stops the server, closes open connections, and waits for their handlers. func (s *Server) Stop() error { + var err error if s.listener != nil { - return s.listener.Close() + err = s.listener.Close() } - return nil + s.connsMu.Lock() + for c := range s.conns { + _ = c.Close() + } + s.conns = nil + s.connsMu.Unlock() + s.wg.Wait() + return err +} + +// trackConn records an open connection, returning false if the server has stopped. +func (s *Server) trackConn(c net.Conn) bool { + s.connsMu.Lock() + defer s.connsMu.Unlock() + if s.conns == nil { + return false + } + s.conns[c] = struct{}{} + return true +} + +func (s *Server) untrackConn(c net.Conn) { + _ = c.Close() + s.connsMu.Lock() + defer s.connsMu.Unlock() + delete(s.conns, c) } // ExecHandler returns a CommandHandler to execute a command in the given environment. @@ -216,73 +246,6 @@ func (s *Server) serverConfig() *ssh.ServerConfig { return conf } -func (s *Server) handleChannels(conn ssh.ConnMetadata, chans <-chan ssh.NewChannel) { - t := s.t - - for newChannel := range chans { - if newChannel.ChannelType() != "session" { - err := newChannel.Reject(ssh.UnknownChannelType, "unknown channel type") - if err != nil { - t.Errorf("Failed to reject channel: %v", err) - } - continue - } - - channel, requests, err := newChannel.Accept() - if err != nil { - t.Errorf("Failed to accept channel: %v", err) - return - } - - timer := time.NewTimer(time.Second * 10) - - var exitWithStatus = make(chan int, 1) - go func(in <-chan *ssh.Request) { - for req := range in { - if !req.WantReply { - continue - } - switch req.Type { - case "exec": - err := req.Reply(true, nil) - if err != nil { - t.Errorf("Failed to reply to command: %v", err) - } - // Strip the first four bytes of the payload, the uint32 representing the string length. - // See https://datatracker.ietf.org/doc/html/rfc4251#section-5 - cmd := req.Payload[4:] - t.Logf("Handling command: %s", cmd) - exitWithStatus <- s.CommandHandler(conn, string(cmd), CommandIO{ - StdIn: channel, - StdOut: channel, - StdErr: channel.Stderr(), - }) - return - default: - _ = req.Reply(false, nil) - } - } - }(requests) - - for { - select { - case s := <-exitWithStatus: - _, err = channel.SendRequest("exit-status", false, ssh.Marshal(struct{ Status uint32 }{uint32(s)})) - if err != nil { - t.Fatalf("Failed to send exit status: %v", err) - } - goto closeChannel - case <-timer.C: - t.Error("Timed out") - goto closeChannel - } - } - - closeChannel: - _ = channel.Close() - } -} - // generateHostKey returns a host key, generated once, as servers may run in parallel tests. var generateHostKey = sync.OnceValues(func() (ssh.Signer, error) { _, privateKey, err := ed25519.GenerateKey(rand.Reader)