diff --git a/.github/workflows/tests.yaml b/.github/workflows/tests.yaml index 213f134..d596ff6 100644 --- a/.github/workflows/tests.yaml +++ b/.github/workflows/tests.yaml @@ -10,23 +10,61 @@ jobs: runs-on: ubuntu-latest steps: - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7 + with: + persist-credentials: false - name: setup Go uses: actions/setup-go@b7ad1dad31e06c5925ef5d2fc7ad053ef454303e # v7 with: go-version-file: go.mod + - name: install login shells + run: sudo apt-get update && sudo apt-get install -y bash zsh fish openssh-server + - name: unit tests run: go test -race -covermode=atomic . ./cmd/... ./hack/... ./pkg/... env: CGO_ENABLED: "1" - - name: install goreleaser + - name: disk usage before snapshot + run: | + df -h /tmp "$RUNNER_TEMP" "$GITHUB_WORKSPACE" + du -sh "$(go env GOCACHE)" "$(go env GOMODCACHE)" + + - name: build snapshot artifacts uses: goreleaser/goreleaser-action@a4f02ab5355a589de60d917a659ce617ed19b3de # v7 with: version: "~> v2" - args: release --snapshot --clean + args: release --snapshot --clean --parallelism=1 + + - name: disk usage after snapshot + if: always() + run: | + df -h /tmp "$RUNNER_TEMP" "$GITHUB_WORKSPACE" + du -sh "$(go env GOCACHE)" "$(go env GOMODCACHE)" + + - name: provision login shell accounts + run: | + sudo service ssh start + for shell in bash zsh fish; do + sudo useradd --create-home --shell "$(command -v "$shell")" "devsy-$shell" + done - name: integration tests working-directory: ./e2e run: go test -v -ginkgo.v -timeout 3600s --ginkgo.label-filter=integration + env: + DEVSY_E2E_SHELL_MATRIX: "true" + + windows-stdin: + name: Windows redirected SSH stdin + runs-on: windows-latest + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7 + with: + persist-credentials: false + - uses: actions/setup-go@b7ad1dad31e06c5925ef5d2fc7ad053ef454303e # v7 + with: + go-version-file: go.mod + - name: test redirected pipe bytes and cancellation + run: go test ./pkg/ssh -run TestWindows -timeout 60s diff --git a/README.md b/README.md index 233bfa0..ec3047d 100644 --- a/README.md +++ b/README.md @@ -50,3 +50,17 @@ This provider has the following options: # Extra For more detail, see the [Devsy Documentation](https://devsy.sh/docs/managing-providers/what-are-providers). + +### Remote shell requirements + +The SSH server must run Linux with `/bin/sh` and the standard base tools used by +Devsy (`uname`, `id`, `mkdir`, `test`, `cat`, `chmod`, and `mv`). Bash, Zsh, and Fish +are supported as the SSH account login shell. Bootstrap scripts execute through +`/bin/sh`; the login shell does not need to understand POSIX script syntax. +Login startup files must leave stdout clean because it carries the bootstrap +handshake and tunnel bytes. Send diagnostics to stderr. + +The SSH login shell is independent of the shell inside the workspace container. +`DOCKER_PATH` selects one executable path, including paths containing spaces; it +is not a command line with flags. The provider sends the bootstrap as a command +argument and reserves stdin for protocol data. It does not require SCP. diff --git a/cmd/command.go b/cmd/command.go index 3e968fe..0d07ada 100644 --- a/cmd/command.go +++ b/cmd/command.go @@ -9,23 +9,23 @@ import ( "github.com/spf13/cobra" ) -// CommandCmd holds the cmd flags +// CommandCmd holds the cmd flags. type CommandCmd struct{} -// NewCommandCmd defines a command +// NewCommandCmd defines a command. func NewCommandCmd() *cobra.Command { cmd := &CommandCmd{} commandCmd := &cobra.Command{ Use: "command", Short: "Command an instance", - RunE: func(_ *cobra.Command, args []string) error { + RunE: func(cobraCmd *cobra.Command, args []string) error { sshProvider, err := ssh.NewProvider() if err != nil { return err } return cmd.Run( - context.Background(), + cobraCmd.Context(), sshProvider, ) }, @@ -34,7 +34,7 @@ func NewCommandCmd() *cobra.Command { return commandCmd } -// Run runs the command logic +// Run runs the command logic. func (cmd *CommandCmd) Run( ctx context.Context, providerSSH *ssh.SSHProvider, @@ -44,5 +44,10 @@ func (cmd *CommandCmd) Run( return fmt.Errorf("command environment variable is missing") } - return ssh.Command(providerSSH, command) + return ssh.Command( + ctx, + providerSSH, + command, + ssh.Streams{Stdin: os.Stdin, Stdout: os.Stdout, Stderr: os.Stderr}, + ) } diff --git a/cmd/init.go b/cmd/init.go index 0fd7b1f..fc8b770 100644 --- a/cmd/init.go +++ b/cmd/init.go @@ -2,28 +2,29 @@ package cmd import ( "context" + "os" "github.com/devsy-org/devsy-provider-ssh/pkg/ssh" "github.com/spf13/cobra" ) -// InitCmd holds the cmd flags +// InitCmd holds the cmd flags. type InitCmd struct{} -// NewInitCmd defines a init +// NewInitCmd defines a init. func NewInitCmd() *cobra.Command { cmd := &InitCmd{} initCmd := &cobra.Command{ Use: "init", Short: "Init account", - RunE: func(_ *cobra.Command, args []string) error { + RunE: func(cobraCmd *cobra.Command, args []string) error { sshProvider, err := ssh.NewProvider() if err != nil { return err } return cmd.Run( - context.Background(), + cobraCmd.Context(), sshProvider, ) }, @@ -32,10 +33,14 @@ func NewInitCmd() *cobra.Command { return initCmd } -// Run runs the init logic +// Run runs the init logic. func (cmd *InitCmd) Run( ctx context.Context, providerSSH *ssh.SSHProvider, ) error { - return ssh.Init(providerSSH) + return ssh.Init( + ctx, + providerSSH, + ssh.Streams{Stdout: os.Stdout, Stderr: os.Stderr}, + ) } diff --git a/cmd/root.go b/cmd/root.go index e2945c1..2142442 100644 --- a/cmd/root.go +++ b/cmd/root.go @@ -1,15 +1,19 @@ package cmd import ( + "context" + "errors" "os" "os/exec" + "os/signal" + "syscall" "github.com/devsy-org/devsy/pkg/log" "github.com/spf13/cobra" "golang.org/x/crypto/ssh" ) -// NewRootCmd returns a new root command +// NewRootCmd returns a new root command. func NewRootCmd() *cobra.Command { sshCmd := &cobra.Command{ Use: "devsy-provider-ssh", @@ -38,23 +42,28 @@ func Execute() { rootCmd := BuildRoot() // execute command - err := rootCmd.Execute() - if err != nil { - if exitErr, ok := err.(*ssh.ExitError); ok { - os.Exit(exitErr.ExitStatus()) - } - if exitErr, ok := err.(*exec.ExitError); ok { - if len(exitErr.Stderr) > 0 { - log.Error(string(exitErr.Stderr)) - } - os.Exit(exitErr.ExitCode()) - } - + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + err := rootCmd.ExecuteContext(ctx) + stop() + if err == nil { + return + } + if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) { log.Fatal(err) } + if sshExit, ok := errors.AsType[*ssh.ExitError](err); ok { + os.Exit(sshExit.ExitStatus()) + } + if exitErr, ok := errors.AsType[*exec.ExitError](err); ok { + if len(exitErr.Stderr) > 0 { + log.Error(string(exitErr.Stderr)) + } + os.Exit(exitErr.ExitCode()) + } + log.Fatal(err) } -// BuildRoot creates a new root command from the +// BuildRoot creates the SSH provider command tree. func BuildRoot() *cobra.Command { rootCmd := NewRootCmd() diff --git a/cmd/root_unix_test.go b/cmd/root_unix_test.go new file mode 100644 index 0000000..1fc436c --- /dev/null +++ b/cmd/root_unix_test.go @@ -0,0 +1,95 @@ +//go:build !windows + +package cmd + +import ( + "bufio" + "bytes" + "context" + "errors" + "os" + "os/exec" + "path/filepath" + "strings" + "syscall" + "testing" + "time" +) + +func TestExecuteProcess(t *testing.T) { + if os.Getenv("DEVSY_TEST_EXECUTE") != "1" { + return + } + os.Args = []string{"devsy-provider-ssh", "command"} + Execute() +} + +func executeTestProcess(t *testing.T, script string) *exec.Cmd { + t.Helper() + dir := t.TempDir() + // #nosec G306 -- Controlled executable models the SSH child at the CLI boundary. + if err := os.WriteFile( + filepath.Join(dir, "ssh"), + []byte("#!/bin/sh\n"+script), + 0o700, + ); err != nil { + t.Fatal(err) + } + for key, value := range map[string]string{ + "DEVSY_TEST_EXECUTE": "1", "DOCKER_PATH": "docker", "AGENT_PATH": "/tmp/agent", + "HOST": "fixture", "PORT": "22", "USE_BUILTIN_SSH": "false", "COMMAND": "fixture", + "EXTRA_FLAGS": "", "PATH": dir + string(os.PathListSeparator) + os.Getenv("PATH"), + } { + t.Setenv(key, value) + } + executable, err := os.Executable() + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + t.Cleanup(cancel) + // #nosec G204 -- Re-execute only this test binary with a controlled helper selector. + return exec.CommandContext(ctx, executable, "-test.run=^TestExecuteProcess$") +} + +func TestExecuteCancellation(t *testing.T) { + cmd := executeTestProcess(t, "printf 'ready\\n'\nexec sleep 30\n") + var stderr bytes.Buffer + cmd.Stderr = &stderr + stdout, err := cmd.StdoutPipe() + if err != nil { + t.Fatal(err) + } + if err = cmd.Start(); err != nil { + t.Fatal(err) + } + ready, err := bufio.NewReader(stdout).ReadString('\n') + if err != nil || ready != "ready\n" { + t.Fatalf("SSH child did not start: %q, %v", ready, err) + } + if err = cmd.Process.Signal(syscall.SIGTERM); err != nil { + t.Fatal(err) + } + assertProcessExit(t, cmd.Wait(), 1) + if !strings.Contains(stderr.String(), "context canceled") { + t.Fatalf("missing cancellation diagnostic: %q", stderr.String()) + } +} + +func TestExecuteRemoteExit(t *testing.T) { + cmd := executeTestProcess(t, "printf 'remote diagnostic' >&2\nexit 37\n") + var stderr bytes.Buffer + cmd.Stderr = &stderr + assertProcessExit(t, cmd.Run(), 37) + if stderr.String() != "remote diagnostic" { + t.Fatalf("remote diagnostic changed: %q", stderr.String()) + } +} + +func assertProcessExit(t *testing.T, err error, code int) { + t.Helper() + var exit *exec.ExitError + if !errors.As(err, &exit) || exit.ExitCode() != code { + t.Fatalf("expected exit %d, got %v", code, err) + } +} diff --git a/e2e/tests/integration/integration.go b/e2e/tests/integration/integration.go index d0444cb..3e5fa01 100644 --- a/e2e/tests/integration/integration.go +++ b/e2e/tests/integration/integration.go @@ -73,9 +73,19 @@ func setupSSHKeys() { } func setupDevsyCLI() { + if local := os.Getenv("DEVSY_E2E_BINARY"); local != "" { + // #nosec G304,G703 -- Test runner explicitly selects the local core binary. + data, err := os.ReadFile(local) + framework.ExpectNoError(err) + framework.ExpectNoError(os.MkdirAll("bin", 0o750)) + // #nosec G306,G703 -- Local core binary must be executable. + framework.ExpectNoError(os.WriteFile("bin/devsy", data, 0o700)) + return + } + client := &http.Client{Timeout: time.Second * 30} resp, err := client.Get( - "https://github.com/devsy-org/devsy/releases/latest/download/devsy-linux-amd64", + "https://github.com/devsy-org/devsy/releases/download/v1.24.0/devsy-linux-amd64", ) framework.ExpectNoError(err) framework.ExpectEqual(resp.StatusCode, http.StatusOK) @@ -241,12 +251,15 @@ echo line3`, "PORT=22", "USE_BUILTIN_SSH=false", }...) - output, err := cmd.CombinedOutput() + var stdout, stderr bytes.Buffer + cmd.Stdout, cmd.Stderr = &stdout, &stderr + err := cmd.Run() framework.ExpectError(err) - output = bytes.TrimSpace(output) - msg := string(output) - gomega.Expect(msg).To(gomega.ContainSubstring("not-a-command")) - gomega.Expect(msg).To(gomega.ContainSubstring("command not found")) + gomega.Expect(cmd.ProcessState).NotTo(gomega.BeNil()) + gomega.Expect(cmd.ProcessState.ExitCode()).To(gomega.Equal(127)) + gomega.Expect(stdout.Bytes()).To(gomega.BeEmpty()) + gomega.Expect(stderr.Bytes()).NotTo(gomega.BeEmpty()) + gomega.Expect(stderr.String()).To(gomega.ContainSubstring("not-a-command")) }) ginkgo.It("should run devsy up", func() { diff --git a/e2e/tests/integration/login_shells.go b/e2e/tests/integration/login_shells.go new file mode 100644 index 0000000..3d55250 --- /dev/null +++ b/e2e/tests/integration/login_shells.go @@ -0,0 +1,68 @@ +package integration + +import ( + "bytes" + "fmt" + "os" + "os/exec" + "path/filepath" + "strconv" + + "github.com/devsy-org/devsy/e2e/framework" + "github.com/onsi/ginkgo/v2" + "github.com/onsi/gomega" +) + +var _ = ginkgo.Describe("SSH login shells", ginkgo.Label("integration"), func() { + ginkgo.BeforeEach(func() { + if os.Getenv("DEVSY_E2E_SHELL_MATRIX") != "true" { + ginkgo.Skip("requires provisioned Linux login shell accounts") + } + setupSSHKeys() + }) + for _, shell := range []string{"bash", "zsh", "fish"} { + for _, builtin := range []bool{false, true} { + ginkgo.It( + fmt.Sprintf("preserves bootstrap streams with %s builtin=%t", shell, builtin), + func() { + home, err := os.UserHomeDir() + framework.ExpectNoError(err) + user := "devsy-" + shell + sshDir := "/home/" + user + "/.ssh" + for _, args := range [][]string{ + {"install", "-d", "-m", "700", "-o", user, "-g", user, sshDir}, + { + "install", "-m", "600", "-o", user, "-g", user, + filepath.Join(home, ".ssh", "id_rsa.pub"), sshDir + "/authorized_keys", + }, + } { + // #nosec G204 -- Install only the controlled fixture key into provisioned test accounts. + output, err := exec.Command("sudo", args...).CombinedOutput() + framework.ExpectNoError(err, string(output)) + } + payload := bytes.Repeat([]byte{0, 1, 2, 255, '\n', '\'', '\\'}, 512) + cmd := exec.Command(providerPath, "command") + cmd.Env = append( + os.Environ(), + "HOST="+user+"@127.0.0.1", + "PORT=22", + "USE_BUILTIN_SSH="+strconv.FormatBool(builtin), + "EXTRA_FLAGS=-i "+filepath.Join(home, ".ssh", "id_rsa"), + "AGENT_PATH=/tmp/devsy/agent", + "DOCKER_PATH=docker", + "COMMAND=printf 'READY\\n'\ncat\nprintf 'diagnostic' >&2\nexit 37", + ) + cmd.Stdin = bytes.NewReader(payload) + var stdout, stderr bytes.Buffer + cmd.Stdout, cmd.Stderr = &stdout, &stderr + err = cmd.Run() + gomega.Expect(err).To(gomega.HaveOccurred()) + gomega.Expect(cmd.ProcessState.ExitCode()).To(gomega.Equal(37)) + gomega.Expect(stdout.Bytes()). + To(gomega.Equal(append([]byte("READY\n"), payload...))) + gomega.Expect(stderr.String()).To(gomega.ContainSubstring("diagnostic")) + }, + ) + } + } +}) diff --git a/go.mod b/go.mod index 31e70db..8f02ce6 100644 --- a/go.mod +++ b/go.mod @@ -10,6 +10,7 @@ require ( github.com/onsi/gomega v1.44.0 github.com/spf13/cobra v1.10.2 golang.org/x/crypto v0.58.0 + golang.org/x/sys v0.49.0 ) require ( @@ -318,7 +319,6 @@ require ( golang.org/x/net v0.60.0 // indirect golang.org/x/oauth2 v0.36.0 // indirect golang.org/x/sync v0.24.0 // indirect - golang.org/x/sys v0.49.0 // indirect golang.org/x/term v0.47.0 // indirect golang.org/x/text v0.43.0 // indirect golang.org/x/time v0.15.0 // indirect diff --git a/pkg/ssh/builtin_context_test.go b/pkg/ssh/builtin_context_test.go new file mode 100644 index 0000000..d0d3bd9 --- /dev/null +++ b/pkg/ssh/builtin_context_test.go @@ -0,0 +1,197 @@ +package ssh + +import ( + "bufio" + "context" + "crypto/ed25519" + "crypto/rand" + "errors" + "io" + "net" + "os" + "os/exec" + "path/filepath" + "sync" + "testing" + "time" + + "github.com/devsy-org/devsy-provider-ssh/pkg/options" + cryptossh "golang.org/x/crypto/ssh" +) + +type phaseReadyWriter struct { + ready chan struct{} + once sync.Once +} + +func (writer *phaseReadyWriter) Write(data []byte) (int, error) { + writer.once.Do(func() { close(writer.ready) }) + return len(data), nil +} + +func configTestProvider(t *testing.T, address, script string) *SSHProvider { + t.Helper() + _, key, err := ed25519.GenerateKey(rand.Reader) + requireNoError(t, err) + dir, identity := writeTestIdentity(t, key) + host, port, err := net.SplitHostPort(address) + requireNoError(t, err) + if script == "" { + script = "printf '%s' 'hostname " + host + "\nuser fixture\nport " + port + "\nidentityfile " + identity + "\n'\n" + } + // #nosec G306 -- Controlled executable models SSH configuration lookup. + requireNoError(t, os.WriteFile(filepath.Join(dir, "ssh"), []byte("#!/bin/sh\n"+script), 0o700)) + t.Setenv("PATH", dir+string(os.PathListSeparator)+os.Getenv("PATH")) + return &SSHProvider{ + Config: &options.Options{Host: "context-fixture", Port: port, UseBuiltinSSH: true}, + } +} + +func cancelAfterReady(t *testing.T, ready <-chan struct{}, run func(context.Context) error) error { + t.Helper() + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + result := make(chan error, 1) + go func() { result <- run(ctx) }() + select { + case <-ready: + case <-time.After(3 * time.Second): + t.Fatal("SSH phase did not start") + } + cancel() + select { + case err := <-result: + if !errors.Is(err, context.Canceled) { + t.Fatalf("lost cancellation identity: %v", err) + } + return err + case <-time.After(3 * time.Second): + t.Fatal("SSH phase did not stop on cancellation") + return nil + } +} + +func TestBuiltinLookupCancellation(t *testing.T) { + provider := configTestProvider(t, "127.0.0.1:22", "printf ready >&2\nexec sleep 30\n") + output := &phaseReadyWriter{ready: make(chan struct{})} + err := cancelAfterReady(t, output.ready, func(ctx context.Context) error { + return provider.execBuiltin(ctx, "unused", Streams{Stderr: output}) + }) + if _, ok := errors.AsType[*exec.ExitError](err); !ok { + t.Fatalf("lost started lookup process error: %v", err) + } +} + +func TestBuiltinLookupOwnDeadline(t *testing.T) { + provider := configTestProvider(t, "127.0.0.1:22", "exec sleep 30\n") + ctx, cancel := context.WithTimeout(context.Background(), 8*time.Second) + defer cancel() + err := provider.execBuiltin(ctx, "unused", Streams{}) + if !errors.Is(err, context.DeadlineExceeded) || ctx.Err() != nil { + t.Fatalf("lost independent lookup deadline: %v, parent=%v", err, ctx.Err()) + } +} + +func TestBuiltinDialCancellation(t *testing.T) { + provider := configTestProvider(t, "127.0.0.1:22", "") + ready := make(chan struct{}) + dialErr := errors.New("controlled dial failure") + err := cancelAfterReady(t, ready, func(ctx context.Context) error { + return provider.execBuiltinWithDial(ctx, "unused", Streams{}, + func(ctx context.Context, _, _ string) (net.Conn, error) { + close(ready) + <-ctx.Done() + return nil, dialErr + }) + }) + if !errors.Is(err, dialErr) { + t.Fatalf("lost dial cause: %v", err) + } +} + +func pendingBuiltinPhase(t *testing.T, session bool) (*SSHProvider, <-chan struct{}) { + t.Helper() + _, key, err := ed25519.GenerateKey(rand.Reader) + requireNoError(t, err) + signer, err := cryptossh.NewSignerFromKey(key) + requireNoError(t, err) + config := &cryptossh.ServerConfig{NoClientAuth: true} + config.AddHostKey(signer) + listener, err := net.Listen("tcp", "127.0.0.1:0") + requireNoError(t, err) + ready, done := make(chan struct{}), make(chan struct{}) + go func() { + defer close(done) + conn, err := listener.Accept() + if err != nil { + return + } + defer func() { _ = conn.Close() }() + if !session { + reader := bufio.NewReader(conn) + if _, err := reader.ReadString('\n'); err != nil { + return + } + close(ready) + _, _ = io.Copy(io.Discard, reader) + return + } + server, channels, requests, err := cryptossh.NewServerConn(conn, config) + if err != nil { + return + } + defer func() { _ = server.Close() }() + go cryptossh.DiscardRequests(requests) + if _, ok := <-channels; !ok { + return + } + close(ready) + _ = server.Wait() + }() + t.Cleanup(func() { + _ = listener.Close() + select { + case <-done: + case <-time.After(3 * time.Second): + t.Error("pending SSH phase did not release its connection") + } + }) + return configTestProvider(t, listener.Addr().String(), ""), ready +} + +func TestBuiltinHandshakeCancellation(t *testing.T) { + provider, ready := pendingBuiltinPhase(t, false) + err := cancelAfterReady(t, ready, func(ctx context.Context) error { + return provider.execBuiltin(ctx, "unused", Streams{}) + }) + if !errors.Is(err, net.ErrClosed) { + t.Fatalf("lost closed handshake cause: %v", err) + } +} + +func TestBuiltinSessionOpeningCancellation(t *testing.T) { + provider, ready := pendingBuiltinPhase(t, true) + _ = cancelAfterReady(t, ready, func(ctx context.Context) error { + return provider.execBuiltin(ctx, "unused", Streams{}) + }) +} + +func TestBuiltinCancellationPreservesSourceError(t *testing.T) { + provider := builtinTestProvider( + t, + func(channel cryptossh.Channel, requests <-chan *cryptossh.Request) { + defer func() { _ = channel.Close() }() + request := <-requests + _ = request.Reply(true, nil) + _, _ = io.Copy(io.Discard, channel) + }, + ) + input := newLateErrorInput() + defer func() { _ = input.Close() }() + err := cancelAfterReady(t, input.started, func(ctx context.Context) error { + return provider.execBuiltin(ctx, "unused", Streams{Stdin: input}) + }) + if !errors.Is(err, errInputRead) { + t.Fatalf("lost canceled session source error: %v", err) + } +} diff --git a/pkg/ssh/builtin_test.go b/pkg/ssh/builtin_test.go new file mode 100644 index 0000000..f46b877 --- /dev/null +++ b/pkg/ssh/builtin_test.go @@ -0,0 +1,228 @@ +package ssh + +import ( + "bytes" + "context" + "crypto/ed25519" + "crypto/rand" + "crypto/x509" + "encoding/pem" + "errors" + "io" + "net" + "os" + "path/filepath" + "testing" + "time" + + "github.com/devsy-org/devsy-provider-ssh/pkg/options" + cryptossh "golang.org/x/crypto/ssh" +) + +func builtinTestProvider( + t *testing.T, + handle func(cryptossh.Channel, <-chan *cryptossh.Request), +) *SSHProvider { + t.Helper() + _, key, err := ed25519.GenerateKey(rand.Reader) + requireNoError(t, err) + signer, err := cryptossh.NewSignerFromKey(key) + requireNoError(t, err) + config := &cryptossh.ServerConfig{NoClientAuth: true} + config.AddHostKey(signer) + listener, err := net.Listen("tcp", "127.0.0.1:0") + requireNoError(t, err) + t.Cleanup(func() { _ = listener.Close() }) + done := make(chan struct{}) + go func() { + defer close(done) + conn, err := listener.Accept() + if err != nil { + return + } + defer func() { _ = conn.Close() }() + server, channels, requests, err := cryptossh.NewServerConn(conn, config) + if err != nil { + return + } + defer func() { _ = server.Close() }() + go cryptossh.DiscardRequests(requests) + for incoming := range channels { + channel, requests, err := incoming.Accept() + if err != nil { + return + } + handle(channel, requests) + } + }() + t.Cleanup(func() { + _ = listener.Close() + select { + case <-done: + case <-time.After(5 * time.Second): + t.Error("SSH server did not terminate") + } + }) + dir, identity := writeTestIdentity(t, key) + host, port, err := net.SplitHostPort(listener.Addr().String()) + requireNoError(t, err) + output := "hostname " + host + "\nuser fixture\nport " + port + "\nidentityfile " + identity + "\n" + // #nosec G306 -- Controlled SSH configuration stub must be executable. + requireNoError( + t, + os.WriteFile( + filepath.Join(dir, "ssh"), + []byte("#!/bin/sh\nprintf '%s' '"+output+"'\n"), + 0o700, + ), + ) + + t.Setenv("PATH", dir+string(os.PathListSeparator)+os.Getenv("PATH")) + return &SSHProvider{Config: &options.Options{Host: "fixture", Port: port, UseBuiltinSSH: true}} +} + +func TestBuiltinStreamsCommandAndExit(t *testing.T) { + script := "echo 'bootstrap'\ncat" + expected, err := buildPOSIXRemoteExec(script) + requireNoError(t, err) + received := make(chan string, 1) + provider := builtinTestProvider( + t, + func(channel cryptossh.Channel, requests <-chan *cryptossh.Request) { + defer func() { _ = channel.Close() }() + request := <-requests + var command struct{ Command string } + if err := cryptossh.Unmarshal(request.Payload, &command); err != nil { + return + } + received <- command.Command + _ = request.Reply(true, nil) + data := make([]byte, 5) + _, _ = io.ReadFull(channel, data) + _, _ = channel.Write(data) + _, _ = channel.Stderr().Write([]byte("diagnostic")) + _, _ = channel.SendRequest( + "exit-status", + false, + cryptossh.Marshal(struct{ Status uint32 }{37}), + ) + }, + ) + var stdout, stderr bytes.Buffer + payload := []byte{0, 1, 2, 255, '\n'} + err = Command( + context.Background(), + provider, + script, + Streams{Stdin: bytes.NewReader(payload), Stdout: &stdout, Stderr: &stderr}, + ) + var exit *cryptossh.ExitError + if !errors.As(err, &exit) || exit.ExitStatus() != 37 { + t.Fatalf("exit: %v", err) + } + if got := <-received; got != expected { + t.Fatalf("command: %q", got) + } + if !bytes.Equal(stdout.Bytes(), payload) { + t.Fatalf("stdout: %q", stdout.Bytes()) + } + if stderr.String() != "diagnostic" { + t.Fatalf("stderr: %q", stderr.String()) + } +} + +func TestBuiltinCancellation(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + provider := builtinTestProvider( + t, + func(channel cryptossh.Channel, requests <-chan *cryptossh.Request) { + request := <-requests + _ = request.Reply(true, nil) + cancel() + for range requests { + } + _ = channel.Close() + }, + ) + stdin, writer, err := os.Pipe() + requireNoError(t, err) + defer func() { _ = stdin.Close(); _ = writer.Close() }() + err = Command(ctx, provider, "sleep 30", Streams{Stdin: stdin}) + if !errors.Is(err, context.Canceled) { + t.Fatalf("cancellation: %v", err) + } +} + +func requireNoError(t *testing.T, err error) { + t.Helper() + if err != nil { + t.Fatal(err) + } +} + +func writeTestIdentity(t *testing.T, key ed25519.PrivateKey) (string, string) { + t.Helper() + raw, err := x509.MarshalPKCS8PrivateKey(key) + requireNoError(t, err) + dir := t.TempDir() + identity := filepath.Join(dir, "key") + if err := os.WriteFile( + identity, + pem.EncodeToMemory(&pem.Block{Type: "PRIVATE KEY", Bytes: raw}), + 0o600, + ); err != nil { + t.Fatal(err) + } + return dir, identity +} + +func TestBuiltinCancellationWithPipeReader(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + provider := builtinTestProvider( + t, + func(channel cryptossh.Channel, requests <-chan *cryptossh.Request) { + request := <-requests + _ = request.Reply(true, nil) + cancel() + for range requests { + } + _ = channel.Close() + }, + ) + reader, writer := io.Pipe() + defer func() { _ = writer.Close() }() + err := Command(ctx, provider, "sleep 30", Streams{Stdin: reader}) + if !errors.Is(err, context.Canceled) { + t.Fatalf("cancellation: %v", err) + } +} + +var errInputRead = errors.New("fixture input read failure") + +type failingInput struct{} + +func (failingInput) Read([]byte) (int, error) { return 0, errInputRead } +func (failingInput) Close() error { return nil } + +func TestBuiltinPreservesStdinReadError(t *testing.T) { + provider := builtinTestProvider( + t, + func(channel cryptossh.Channel, requests <-chan *cryptossh.Request) { + defer func() { _ = channel.Close() }() + request := <-requests + _ = request.Reply(true, nil) + _, _ = io.Copy(io.Discard, channel) + _, _ = channel.SendRequest( + "exit-status", + false, + cryptossh.Marshal(struct{ Status uint32 }{0}), + ) + }, + ) + err := Command(context.Background(), provider, "cat", Streams{Stdin: failingInput{}}) + if !errors.Is(err, errInputRead) { + t.Fatalf("stdin failure lost: %v", err) + } +} diff --git a/pkg/ssh/init_stdin_test.go b/pkg/ssh/init_stdin_test.go new file mode 100644 index 0000000..f15a459 --- /dev/null +++ b/pkg/ssh/init_stdin_test.go @@ -0,0 +1,84 @@ +package ssh + +import ( + "bytes" + "context" + "io" + "os" + "path/filepath" + "testing" + "time" + + "github.com/devsy-org/devsy-provider-ssh/pkg/options" + cryptossh "golang.org/x/crypto/ssh" +) + +func TestInitLeavesCallerInputUntouched(t *testing.T) { + for _, builtin := range []bool{false, true} { + name := "external" + if builtin { + name = "builtin" + } + t.Run(name, func(t *testing.T) { + provider := initProbeFailureProvider(t, builtin) + payload := bytes.NewReader([]byte("caller protocol input")) + input := &observedInput{ReadCloser: io.NopCloser(payload), started: make(chan struct{})} + var stdout, stderr bytes.Buffer + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + err := Init(ctx, provider, Streams{Stdin: input, Stdout: &stdout, Stderr: &stderr}) + assertInitProbeFailure(t, err, &stdout, &stderr) + select { + case <-input.started: + t.Error("initialization read caller input") + default: + } + if input.closed.Load() != 0 || payload.Len() != len("caller protocol input") { + t.Fatalf( + "initialization consumed or closed caller input: remaining=%d closes=%d", + payload.Len(), + input.closed.Load(), + ) + } + }) + } +} + +func initProbeFailureProvider(t *testing.T, builtin bool) *SSHProvider { + t.Helper() + if builtin { + return builtinTestProvider( + t, + func(channel cryptossh.Channel, requests <-chan *cryptossh.Request) { + defer func() { _ = channel.Close() }() + request := <-requests + _ = request.Reply(true, nil) + _, _ = io.Copy(io.Discard, channel) + _, _ = io.WriteString(channel.Stderr(), "probe diagnostic") + _, _ = channel.SendRequest( + "exit-status", + false, + cryptossh.Marshal(struct{ Status uint32 }{37}), + ) + }, + ) + } + dir := t.TempDir() + // #nosec G306 -- Controlled SSH stub drains input to expose accidental forwarding. + requireNoError(t, os.WriteFile(filepath.Join(dir, "ssh"), + []byte("#!/bin/sh\ncat >/dev/null\nprintf 'probe diagnostic' >&2\nexit 37\n"), 0o700)) + t.Setenv("PATH", dir+string(os.PathListSeparator)+os.Getenv("PATH")) + return &SSHProvider{Config: &options.Options{Host: "unused", Port: "22"}} +} + +func assertInitProbeFailure(t *testing.T, err error, stdout, stderr *bytes.Buffer) { + t.Helper() + if err == nil || stderr.String() != "probe diagnostic" || stdout.Len() != 0 { + t.Fatalf( + "probe error/streams changed: %v, stdout=%q stderr=%q", + err, + stdout.Bytes(), + stderr.Bytes(), + ) + } +} diff --git a/pkg/ssh/input_console_windows.go b/pkg/ssh/input_console_windows.go new file mode 100644 index 0000000..05b1e96 --- /dev/null +++ b/pkg/ssh/input_console_windows.go @@ -0,0 +1,94 @@ +package ssh + +import ( + "errors" + "io" + "os" + "runtime" + "sync" + "time" + + "golang.org/x/sys/windows" +) + +var cancelSynchronousIO = windows.NewLazySystemDLL("kernel32.dll").NewProc("CancelSynchronousIo") + +type consoleInput struct { + file *os.File + reading sync.Mutex + state sync.Mutex + closed bool + thread windows.Handle + done chan struct{} + once sync.Once + closeErr error +} + +func newConsoleInput(file *os.File) (io.ReadCloser, error) { + owned, err := duplicateInputFile(file) + if err != nil { + return nil, err + } + return &consoleInput{file: owned}, nil +} + +func (input *consoleInput) Read(data []byte) (int, error) { + input.reading.Lock() + defer input.reading.Unlock() + runtime.LockOSThread() + defer runtime.UnlockOSThread() + thread, err := windows.OpenThread(windows.THREAD_TERMINATE, false, windows.GetCurrentThreadId()) + if err != nil { + return 0, err + } + defer func() { _ = windows.CloseHandle(thread) }() + input.state.Lock() + if input.closed { + input.state.Unlock() + return 0, os.ErrClosed + } + input.thread = thread + input.done = make(chan struct{}) + input.state.Unlock() + n, err := input.file.Read(data) + input.state.Lock() + if input.closed && errors.Is(err, windows.ERROR_OPERATION_ABORTED) { + err = os.ErrClosed + } + input.thread = 0 + close(input.done) + input.state.Unlock() + return n, err +} + +func (input *consoleInput) Close() error { + input.once.Do(func() { + input.state.Lock() + input.closed = true + done := input.done + input.state.Unlock() + if done != nil { + input.cancelRead(done) + } + input.closeErr = input.file.Close() + }) + return input.closeErr +} + +func (input *consoleInput) cancelRead(done <-chan struct{}) { + // Retry covers cancellation between publishing the thread and entering ReadConsole. + ticker := time.NewTicker(10 * time.Millisecond) + defer ticker.Stop() + for { + input.state.Lock() + if input.thread != 0 { + _, _, _ = cancelSynchronousIO.Call(uintptr(input.thread)) + } + input.state.Unlock() + select { + case <-done: + return + case <-ticker.C: + } + } +} diff --git a/pkg/ssh/input_console_windows_test.go b/pkg/ssh/input_console_windows_test.go new file mode 100644 index 0000000..e2351f3 --- /dev/null +++ b/pkg/ssh/input_console_windows_test.go @@ -0,0 +1,123 @@ +package ssh + +import ( + "context" + "errors" + "io" + "os" + "os/exec" + "syscall" + "testing" + "time" + + "golang.org/x/sys/windows" +) + +type consoleReadResult struct { + count int + err error +} + +func TestWindowsConsoleCancellation(t *testing.T) { + if os.Getenv("DEVSY_TEST_PRIVATE_CONSOLE") == "true" { + runPrivateConsoleChecks(t) + return + } + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + // #nosec G204,G702 -- Re-execute this test binary in an isolated console helper. + cmd := exec.CommandContext(ctx, os.Args[0], "-test.run=^TestWindowsConsoleCancellation$") + cmd.Env = append(os.Environ(), "DEVSY_TEST_PRIVATE_CONSOLE=true") + cmd.SysProcAttr = &syscall.SysProcAttr{CreationFlags: windows.DETACHED_PROCESS} + output, err := cmd.CombinedOutput() + if err != nil { + t.Fatalf("private console helper: %v\n%s", err, output) + } +} + +func runPrivateConsoleChecks(t *testing.T) { + t.Helper() + kernel := windows.NewLazySystemDLL("kernel32.dll") + result, _, err := kernel.NewProc("AllocConsole").Call() + if result == 0 { + t.Fatalf("allocate isolated console: %v", err) + } + defer func() { _, _, _ = kernel.NewProc("FreeConsole").Call() }() + file, err := os.OpenFile("CONIN$", os.O_RDONLY, 0) + requireNoError(t, err) + defer func() { _ = file.Close() }() + for range 32 { + cancelConsoleRead(t, file) + } + reader, err := newConsoleInput(file) + requireNoError(t, err) + requireNoError(t, reader.Close()) + _, err = reader.Read(make([]byte, 1)) + if !errors.Is(err, os.ErrClosed) { + t.Fatalf("read after console close: %v", err) + } + var mode uint32 + requireNoError(t, windows.GetConsoleMode(windows.Handle(file.Fd()), &mode)) +} + +func cancelConsoleRead(t *testing.T, file *os.File) { + t.Helper() + reader, err := newConsoleInput(file) + requireNoError(t, err) + input := reader.(*consoleInput) + result := make(chan consoleReadResult, 1) + go func() { + n, err := input.Read(make([]byte, 8)) + result <- consoleReadResult{count: n, err: err} + }() + deadline := time.Now().Add(3 * time.Second) + for { + input.state.Lock() + started := input.thread != 0 + input.state.Unlock() + if started { + break + } + if time.Now().After(deadline) { + t.Fatal("console read did not start") + } + time.Sleep(time.Millisecond) + } + select { + case completed := <-result: + t.Fatalf( + "console read completed before cancellation: %d, %v", + completed.count, + completed.err, + ) + default: + } + closeConsoleInput(t, input) + select { + case completed := <-result: + assertConsoleReadCanceled(t, completed) + case <-time.After(3 * time.Second): + t.Fatal("console read did not join") + } + requireNoError(t, input.Close()) +} + +func closeConsoleInput(t *testing.T, input *consoleInput) { + t.Helper() + done := make(chan error, 1) + go func() { done <- input.Close() }() + select { + case err := <-done: + requireNoError(t, err) + case <-time.After(3 * time.Second): + t.Fatal("console cancellation did not close and join its reader") + } +} + +func assertConsoleReadCanceled(t *testing.T, result consoleReadResult) { + t.Helper() + if result.count != 0 || + (!errors.Is(result.err, os.ErrClosed) && !errors.Is(result.err, io.EOF)) { + t.Fatalf("console cancellation: bytes=%d, error=%v", result.count, result.err) + } +} diff --git a/pkg/ssh/input_unix.go b/pkg/ssh/input_unix.go new file mode 100644 index 0000000..eeb99c6 --- /dev/null +++ b/pkg/ssh/input_unix.go @@ -0,0 +1,134 @@ +//go:build !windows + +package ssh + +import ( + "errors" + "fmt" + "io" + "math" + "os" + "runtime" + "sync" + "sync/atomic" + "syscall" + + "golang.org/x/sys/unix" +) + +type polledInput struct { + file *os.File + fd int32 + cancelFD int32 + wakeRead, wakeWrite *os.File + reading sync.Mutex + closed atomic.Bool + once sync.Once + closeErr error +} + +func cancellableFile(file *os.File) (io.ReadCloser, error) { + info, err := file.Stat() + if err != nil { + return nil, err + } + if info.Mode().IsRegular() { + return io.NopCloser(file), nil + } + fd, err := pollDescriptor(file) + if err != nil { + return nil, err + } + reader, writer, err := os.Pipe() + if err != nil { + return nil, err + } + cancelFD, err := pollDescriptor(reader) + if err != nil { + _ = reader.Close() + _ = writer.Close() + return nil, err + } + return &polledInput{ + file: file, + fd: fd, + cancelFD: cancelFD, + wakeRead: reader, + wakeWrite: writer, + }, nil +} + +func (input *polledInput) Read(data []byte) (int, error) { + defer runtime.KeepAlive(input.file) + input.reading.Lock() + defer input.reading.Unlock() + if len(data) == 0 { + return 0, nil + } + for !input.closed.Load() { + if err := input.waitReady(); err != nil { + return 0, err + } + n, err := unix.Read(int(input.fd), data) + if errors.Is(err, syscall.EINTR) || errors.Is(err, syscall.EAGAIN) { + continue + } + if err != nil { + return 0, err + } + if n == 0 { + return 0, io.EOF + } + return n, err + } + return 0, os.ErrClosed +} + +func pollDescriptor(file *os.File) (int32, error) { + conn, err := file.SyscallConn() + if err != nil { + return 0, err + } + var fd int32 + var rangeErr error + err = conn.Control(func(value uintptr) { + if value > math.MaxInt32 { + rangeErr = fmt.Errorf("SSH input descriptor exceeds poll range") + return + } + fd = int32(value) + }) + return fd, errors.Join(err, rangeErr) +} + +func (input *polledInput) Close() error { + input.once.Do(func() { + input.closed.Store(true) + _, signalErr := input.wakeWrite.Write([]byte{1}) + input.reading.Lock() + defer input.reading.Unlock() + input.closeErr = errors.Join(signalErr, input.wakeRead.Close(), input.wakeWrite.Close()) + }) + return input.closeErr +} + +func (input *polledInput) waitReady() error { + for { + fds := []unix.PollFd{ + {Fd: input.fd, Events: unix.POLLIN}, + {Fd: input.cancelFD, Events: unix.POLLIN}, + } + if _, err := unix.Poll(fds, -1); err != nil { + if errors.Is(err, syscall.EINTR) { + continue + } + return err + } + if input.closed.Load() || fds[1].Revents != 0 || fds[0].Revents&unix.POLLNVAL != 0 { + return os.ErrClosed + } + return nil + } +} + +func isPipeWriteClosed(err error) bool { return errors.Is(err, syscall.EPIPE) } diff --git a/pkg/ssh/input_unix_test.go b/pkg/ssh/input_unix_test.go new file mode 100644 index 0000000..7119fe1 --- /dev/null +++ b/pkg/ssh/input_unix_test.go @@ -0,0 +1,131 @@ +//go:build !windows + +package ssh + +import ( + "bytes" + "errors" + "io" + "os" + "testing" + "time" + + "golang.org/x/sys/unix" +) + +func fileFlags(t *testing.T, file *os.File) int { + t.Helper() + conn, err := file.SyscallConn() + requireNoError(t, err) + var flags int + requireNoError( + t, + conn.Control(func(fd uintptr) { flags, err = unix.FcntlInt(fd, unix.F_GETFL, 0) }), + ) + requireNoError(t, err) + return flags +} + +func TestUnixFileCancellationPreservesCaller(t *testing.T) { + for range 32 { + reader, writer, err := os.Pipe() + requireNoError(t, err) + t.Cleanup(func() { _ = reader.Close(); _ = writer.Close() }) + flags := fileFlags(t, reader) + input, err := cancellableFile(reader) + requireNoError(t, err) + if fileFlags(t, reader) != flags { + t.Fatal("input adapter changed caller descriptor flags") + } + result := make(chan error, 1) + go func() { _, err := input.Read(make([]byte, 1)); result <- err }() + waitFileRead(t, input.(*polledInput)) + closeFileInput(t, input, writer) + select { + case err := <-result: + if !errors.Is(err, os.ErrClosed) { + t.Fatalf("idle file cancellation: %v", err) + } + case <-time.After(3 * time.Second): + t.Fatal("idle file read did not join") + } + requireNoError(t, input.Close()) + if fileFlags(t, reader) != flags { + t.Fatal("input cleanup changed caller descriptor flags") + } + _, err = writer.Write([]byte{42}) + requireNoError(t, err) + data := make([]byte, 1) + _, err = io.ReadFull(reader, data) + requireNoError(t, err) + if data[0] != 42 { + t.Fatal("caller file no longer usable") + } + requireNoError(t, reader.Close()) + requireNoError(t, writer.Close()) + } +} + +func closeFileInput(t *testing.T, input io.ReadCloser, writer *os.File) { + t.Helper() + done := make(chan error, 1) + go func() { done <- input.Close() }() + select { + case err := <-done: + requireNoError(t, err) + case <-time.After(3 * time.Second): + _, _ = writer.Write([]byte{1}) + select { + case err := <-done: + t.Logf("file cleanup after timeout: %v", err) + case <-time.After(3 * time.Second): + } + t.Fatal("file cancellation did not wake and join its reader") + } +} + +func waitFileRead(t *testing.T, input *polledInput) { + t.Helper() + deadline := time.Now().Add(3 * time.Second) + for input.reading.TryLock() { + input.reading.Unlock() + if time.Now().After(deadline) { + t.Fatal("idle file read did not start") + } + time.Sleep(time.Millisecond) + } +} + +func TestUnixFileCloseBeforeRead(t *testing.T) { + reader, writer, err := os.Pipe() + requireNoError(t, err) + defer func() { _ = reader.Close(); _ = writer.Close() }() + input, err := cancellableFile(reader) + requireNoError(t, err) + requireNoError(t, input.Close()) + _, err = input.Read(make([]byte, 1)) + if !errors.Is(err, os.ErrClosed) { + t.Fatalf("read after cancellation: %v", err) + } +} + +func TestUnixFilePayloadAndEOF(t *testing.T) { + reader, writer, err := os.Pipe() + requireNoError(t, err) + defer func() { _ = reader.Close() }() + input, err := cancellableFile(reader) + requireNoError(t, err) + defer func() { _ = input.Close() }() + payload := make([]byte, 256) + for i := range payload { + payload[i] = byte(i) + } + _, err = writer.Write(payload) + requireNoError(t, err) + requireNoError(t, writer.Close()) + output, err := io.ReadAll(input) + requireNoError(t, err) + if !bytes.Equal(output, payload) { + t.Fatalf("pipe bytes changed: %x", output) + } +} diff --git a/pkg/ssh/input_windows.go b/pkg/ssh/input_windows.go new file mode 100644 index 0000000..343a345 --- /dev/null +++ b/pkg/ssh/input_windows.go @@ -0,0 +1,60 @@ +package ssh + +import ( + "errors" + "fmt" + "io" + "os" + + "golang.org/x/sys/windows" +) + +func cancellableFile(file *os.File) (io.ReadCloser, error) { + kind, err := windows.GetFileType(windows.Handle(file.Fd())) + if err != nil { + return nil, fmt.Errorf("inspect SSH input handle: %w", err) + } + if kind == windows.FILE_TYPE_DISK { + return io.NopCloser(file), nil + } + if kind != windows.FILE_TYPE_PIPE { + var mode uint32 + if windows.GetConsoleMode(windows.Handle(file.Fd()), &mode) != nil { + return io.NopCloser(file), nil + } + return newConsoleInput(file) + } + // Own a duplicate so closing the session interrupts pipe I/O without closing caller stdin. + owned, err := duplicateInputFile(file) + if err != nil { + return nil, err + } + return &closingInput{ReadCloser: owned}, nil +} + +func duplicateInputFile(file *os.File) (*os.File, error) { + var duplicate windows.Handle + process := windows.CurrentProcess() + err := windows.DuplicateHandle( + process, + windows.Handle(file.Fd()), + process, + &duplicate, + 0, + false, + windows.DUPLICATE_SAME_ACCESS, + ) + if err != nil { + return nil, fmt.Errorf("duplicate SSH input handle: %w", err) + } + owned := os.NewFile(uintptr(duplicate), file.Name()) + if owned == nil { + _ = windows.CloseHandle(duplicate) + return nil, fmt.Errorf("open duplicated SSH input handle") + } + return owned, nil +} + +func isPipeWriteClosed(err error) bool { + return errors.Is(err, windows.ERROR_BROKEN_PIPE) || errors.Is(err, windows.ERROR_NO_DATA) +} diff --git a/pkg/ssh/input_windows_test.go b/pkg/ssh/input_windows_test.go new file mode 100644 index 0000000..a2569ff --- /dev/null +++ b/pkg/ssh/input_windows_test.go @@ -0,0 +1,222 @@ +package ssh + +import ( + "bytes" + "context" + "crypto/ed25519" + "crypto/rand" + "io" + "net" + "os" + "os/exec" + "testing" + "time" + + cryptossh "golang.org/x/crypto/ssh" +) + +func TestWindowsRedirectedStdin(t *testing.T) { + if os.Getenv("DEVSY_TEST_REDIRECTED_STDIN") == "true" { + reader, err := cancellableFile(os.Stdin) + requireNoError(t, err) + defer func() { _ = reader.Close() }() + data := make([]byte, 5) + _, err = io.ReadFull(reader, data) + requireNoError(t, err) + _, err = os.Stdout.Write(data) + requireNoError(t, err) + timer := time.AfterFunc(100*time.Millisecond, func() { _ = reader.Close() }) + defer timer.Stop() + _, err = reader.Read(make([]byte, 1)) + if err == nil { + t.Fatal("idle redirected stdin read survived cancellation") + } + return + } + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + // #nosec G204,G702 -- Launch this test executable to exercise redirected process stdin without a console. + cmd := exec.CommandContext(ctx, os.Args[0], "-test.run=^TestWindowsRedirectedStdin$") + cmd.Env = append(os.Environ(), "DEVSY_TEST_REDIRECTED_STDIN=true") + reader, writer, err := os.Pipe() + requireNoError(t, err) + defer func() { _ = reader.Close(); _ = writer.Close() }() + cmd.Stdin = reader + payload := []byte{0, 1, 2, 255, '\n'} + _, err = writer.Write(payload) + requireNoError(t, err) + output, err := cmd.Output() + requireNoError(t, err) + // The Go test process appends its PASS marker after the raw protocol bytes. + if !bytes.HasPrefix(output, payload) { + t.Fatalf("redirected stdin changed: %q", output) + } +} + +func windowsTestSession( + t *testing.T, + handle func(cryptossh.Channel), +) (*cryptossh.Client, *cryptossh.Session) { + t.Helper() + _, key, err := ed25519.GenerateKey(rand.Reader) + requireNoError(t, err) + signer, err := cryptossh.NewSignerFromKey(key) + requireNoError(t, err) + config := &cryptossh.ServerConfig{NoClientAuth: true} + config.AddHostKey(signer) + listener, err := net.Listen("tcp", "127.0.0.1:0") + requireNoError(t, err) + t.Cleanup(func() { _ = listener.Close() }) + done := make(chan struct{}) + go func() { + defer close(done) + conn, err := listener.Accept() + if err != nil { + return + } + defer func() { _ = conn.Close() }() + server, channels, requests, err := cryptossh.NewServerConn(conn, config) + if err != nil { + return + } + defer func() { _ = server.Close() }() + go cryptossh.DiscardRequests(requests) + incoming := <-channels + channel, reqs, err := incoming.Accept() + if err != nil { + return + } + request := <-reqs + _ = request.Reply(true, nil) + handle(channel) + }() + // #nosec G106 -- Isolated loopback SSH test server has an ephemeral host key. + client, err := cryptossh.Dial( + "tcp", + listener.Addr().String(), + &cryptossh.ClientConfig{ + User: "fixture", + HostKeyCallback: cryptossh.InsecureIgnoreHostKey(), + }, + ) + requireNoError(t, err) + session, err := client.NewSession() + requireNoError(t, err) + t.Cleanup(func() { + _ = session.Close() + _ = client.Close() + select { + case <-done: + case <-time.After(5 * time.Second): + t.Error("Windows SSH fixture did not stop") + } + }) + return client, session +} + +func TestWindowsSessionPipeBytes(t *testing.T) { + _, session := windowsTestSession(t, func(channel cryptossh.Channel) { + defer func() { _ = channel.Close() }() + _, _ = io.Copy(channel, channel) + _, _ = channel.SendRequest( + "exit-status", + false, + cryptossh.Marshal(struct{ Status uint32 }{0}), + ) + }) + reader, writer, err := os.Pipe() + requireNoError(t, err) + defer func() { _ = reader.Close(); _ = writer.Close() }() + payload := []byte{0, 1, 2, 255, '\n', '\'', '\\'} + _, err = writer.Write(payload) + requireNoError(t, err) + requireNoError(t, writer.Close()) + var stdout bytes.Buffer + session.Stdout = &stdout + requireNoError(t, runSession(session, "cat", reader)) + if !bytes.Equal(stdout.Bytes(), payload) { + t.Fatalf("SSH pipe bytes changed: %q", stdout.Bytes()) + } + _, err = reader.Stat() + requireNoError(t, err) +} + +func TestWindowsSessionIdlePipeCancellation(t *testing.T) { + client, session := windowsTestSession(t, func(channel cryptossh.Channel) { + _, _ = channel.Write([]byte("ready")) + _, _ = io.Copy(io.Discard, channel) + _ = channel.Close() + }) + reader, writer, err := os.Pipe() + requireNoError(t, err) + defer func() { _ = reader.Close(); _ = writer.Close() }() + ready := make(chan struct{}, 1) + session.Stdout = readyWriter{ready: ready} + done := make(chan struct{}) + defer close(done) + go func() { + select { + case <-ready: + _ = client.Close() + case <-done: + } + }() + if err := runSession(session, "cat", reader); err == nil { + t.Fatal("expected closed SSH session") + } + _, err = reader.Stat() + requireNoError(t, err) +} + +type readyWriter struct{ ready chan struct{} } + +func (writer readyWriter) Write(data []byte) (int, error) { + select { + case writer.ready <- struct{}{}: + default: + } + return len(data), nil +} + +func TestWindowsExternalInputLifecycle(t *testing.T) { + if mode := os.Getenv("DEVSY_TEST_EXTERNAL_INPUT_CHILD"); mode != "" { + if mode == contextCancelScenario { + _, err := io.Copy(io.Discard, os.Stdin) + requireNoError(t, err) + } + return + } + for _, scenario := range []string{"remote-exit", contextCancelScenario, "large-unread"} { + t.Run(scenario, func(t *testing.T) { runWindowsExternalInput(t, scenario) }) + } +} + +func runWindowsExternalInput(t *testing.T, scenario string) { + t.Helper() + reader, writer := io.Pipe() + defer func() { _ = reader.Close(); _ = writer.Close() }() + input := &observedInput{ReadCloser: reader, started: make(chan struct{})} + if scenario == "large-unread" { + input.ReadCloser = io.NopCloser(bytes.NewReader(make([]byte, 1<<20))) + } + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + // #nosec G204,G702 -- Launch the controlled test executable as the external child. + cmd := exec.CommandContext(ctx, os.Args[0], "-test.run=^TestWindowsExternalInputLifecycle$") + cmd.Env = append(os.Environ(), "DEVSY_TEST_EXTERNAL_INPUT_CHILD="+scenario) + result := make(chan error, 1) + go func() { result <- runExternal(cmd, input) }() + select { + case <-input.started: + case <-time.After(3 * time.Second): + t.Fatal("stdin copy did not start") + } + cancelCommand := scenario == contextCancelScenario + if cancelCommand { + cancel() + } + err := waitInputResult(t, result, input) + if (err != nil) != cancelCommand { + t.Fatalf("unexpected external completion: %v", err) + } +} diff --git a/pkg/ssh/remote_command.go b/pkg/ssh/remote_command.go new file mode 100644 index 0000000..db5cc3e --- /dev/null +++ b/pkg/ssh/remote_command.go @@ -0,0 +1,251 @@ +package ssh + +import ( + "bytes" + "context" + "errors" + "fmt" + "io" + "net" + "os" + "os/exec" + "path/filepath" + "strings" + "sync" + "time" + + devsyssh "github.com/devsy-org/devsy/pkg/ssh" + cryptossh "golang.org/x/crypto/ssh" +) + +// Streams keeps protocol bytes separate from diagnostics. Non-file Stdin closers +// transfer ownership to the transport and must unblock concurrent Read on Close. +// Other non-file readers must be finite buffers; caller-owned files remain open. +// The transport must be the only reader of supplied stdin until it returns. +type Streams struct { + Stdin io.Reader + Stdout io.Writer + Stderr io.Writer +} + +func buildPOSIXRemoteExec(script string) (string, error) { + if strings.ContainsRune(script, 0) { + return "", fmt.Errorf("remote script contains NUL") + } + // Fish interprets backslashes inside single quotes differently from POSIX shells. + encoded := strings.NewReplacer("\\", "'\\\\'", "'", "'\\''").Replace(script) + return "/bin/sh -c '" + encoded + "'", nil +} + +func quotePOSIXArgument(value string) (string, error) { + if strings.ContainsRune(value, 0) { + return "", fmt.Errorf("remote argument contains NUL") + } + return "'" + strings.ReplaceAll(value, "'", "'\\''") + "'", nil +} + +func (provider *SSHProvider) execRemoteScript( + ctx context.Context, + script string, + streams Streams, +) error { + command, err := buildPOSIXRemoteExec(script) + if err != nil { + return err + } + if provider.Config.UseBuiltinSSH { + return provider.execBuiltin(ctx, command, streams) + } + args, err := getSSHCommand(provider) + if err != nil { + return err + } + // #nosec G204 -- SSH arguments are explicit provider options; script is one encoded argument. + cmd := exec.CommandContext(ctx, "ssh", append(args, command)...) + cmd.Stdout, cmd.Stderr = streams.Stdout, streams.Stderr + if err := runExternal(cmd, streams.Stdin); err != nil { + return fmt.Errorf("execute remote /bin/sh script: %w", joinContextError(ctx, err)) + } + return nil +} + +func joinContextError(ctx context.Context, err error) error { + if err == nil { + return nil + } + if ctxErr := ctx.Err(); ctxErr != nil && !errors.Is(err, ctxErr) { + return errors.Join(ctxErr, err) + } + return err +} + +func runExternal(cmd *exec.Cmd, input io.Reader) error { + if input == nil { + return cmd.Run() + } + if file, ok := input.(*os.File); ok { + cmd.Stdin = file + return cmd.Run() + } + reader, err := cancellableInput(input) + if err != nil { + return fmt.Errorf("prepare SSH stdin: %w", err) + } + defer func() { _ = reader.Close() }() + stdin, err := cmd.StdinPipe() + if err != nil { + return fmt.Errorf("open SSH stdin: %w", err) + } + defer func() { _ = stdin.Close() }() + if err := cmd.Start(); err != nil { + return err + } + copy := startStdinCopy(reader, stdin) + err = cmd.Wait() + _ = stdin.Close() + return errors.Join(err, copy.finish()) +} + +func (provider *SSHProvider) builtinConfig( + ctx context.Context, + stderr io.Writer, +) (*cryptossh.ClientConfig, string, error) { + lookupCtx, cancel := context.WithTimeout(ctx, 5*time.Second) + defer cancel() + args, err := getSSHCommand(provider) + if err != nil { + return nil, "", err + } + // #nosec G204 -- Resolve explicit SSH options without executing a remote command. + lookup := exec.CommandContext(lookupCtx, "ssh", append([]string{"-G"}, args...)...) + lookup.Stderr = stderr + config, err := lookup.Output() + if err != nil { + return nil, "", fmt.Errorf("read SSH config: %w", joinContextError(lookupCtx, err)) + } + resolved := parseConfig(string(config)) + if resolved.hostname == "" || resolved.user == "" || resolved.port == "" || + resolved.identityfile == "" { + return nil, "", fmt.Errorf("SSH config requires hostname, user, port and identityfile") + } + clientConfig, err := identityConfig(resolved.identityfile) + if err != nil { + return nil, "", err + } + clientConfig.User = resolved.user + return clientConfig, net.JoinHostPort(resolved.hostname, resolved.port), nil +} + +func identityConfig(identityfile string) (*cryptossh.ClientConfig, error) { + if strings.HasPrefix(identityfile, "~/") { + home, err := os.UserHomeDir() + if err != nil { + return nil, fmt.Errorf("resolve identity home: %w", err) + } + identityfile = filepath.Join(home, identityfile[2:]) + } + // #nosec G304 -- Identity file is intentionally selected by the user SSH configuration. + key, err := os.ReadFile(identityfile) + if err != nil { + return nil, fmt.Errorf("read SSH identity file: %w", err) + } + clientConfig, err := devsyssh.ConfigFromKeyBytes(key) + if err != nil { + return nil, fmt.Errorf("parse SSH identity: %w", err) + } + return clientConfig, nil +} + +func (provider *SSHProvider) execBuiltin( + ctx context.Context, + command string, + streams Streams, +) error { + return provider.execBuiltinWithDial(ctx, command, streams, (&net.Dialer{}).DialContext) +} + +func (provider *SSHProvider) execBuiltinWithDial( + ctx context.Context, + command string, + streams Streams, + dial func(context.Context, string, string) (net.Conn, error), +) (err error) { + defer func() { + err = joinContextError(ctx, err) + }() + config, addr, err := provider.builtinConfig(ctx, streams.Stderr) + if err != nil { + return err + } + conn, err := dial(ctx, "tcp", addr) + if err != nil { + return fmt.Errorf("dial SSH: %w", err) + } + defer func() { _ = conn.Close() }() + stop := context.AfterFunc(ctx, func() { _ = conn.Close() }) + defer stop() + sshConn, channels, requests, err := cryptossh.NewClientConn(conn, addr, config) + if err != nil { + return fmt.Errorf("SSH handshake: %w", err) + } + client := cryptossh.NewClient(sshConn, channels, requests) + defer func() { _ = client.Close() }() + session, err := client.NewSession() + if err != nil { + return fmt.Errorf("create SSH session: %w", err) + } + defer func() { _ = session.Close() }() + session.Stdout, session.Stderr = streams.Stdout, streams.Stderr + if err := runSession(session, command, streams.Stdin); err != nil { + return fmt.Errorf("execute remote /bin/sh script: %w", err) + } + return nil +} + +func runSession(session *cryptossh.Session, command string, input io.Reader) error { + if input == nil { + return session.Run(command) + } + reader, err := cancellableInput(input) + if err != nil { + return fmt.Errorf("prepare SSH stdin: %w", err) + } + defer func() { _ = reader.Close() }() + stdin, err := session.StdinPipe() + if err != nil { + return fmt.Errorf("open SSH stdin: %w", err) + } + if err := session.Start(command); err != nil { + return err + } + copy := startStdinCopy(reader, stdin) + err = session.Wait() + _ = stdin.Close() + return errors.Join(err, copy.finish()) +} + +func cancellableInput(input io.Reader) (io.ReadCloser, error) { + switch reader := input.(type) { + case *os.File: + return cancellableFile(reader) + case *bytes.Buffer, *bytes.Reader, *strings.Reader: + return io.NopCloser(input), nil + case io.ReadCloser: + return &closingInput{ReadCloser: reader}, nil + default: + return nil, fmt.Errorf( + "SSH stdin must be a file, finite buffer, or interruptible ReadCloser", + ) + } +} + +type closingInput struct { + io.ReadCloser + once sync.Once + closeErr error +} + +func (input *closingInput) Close() error { + input.once.Do(func() { input.closeErr = input.ReadCloser.Close() }) + return input.closeErr +} diff --git a/pkg/ssh/remote_command_test.go b/pkg/ssh/remote_command_test.go new file mode 100644 index 0000000..1291d33 --- /dev/null +++ b/pkg/ssh/remote_command_test.go @@ -0,0 +1,237 @@ +package ssh + +import ( + "bytes" + "context" + "errors" + "io" + "os" + "os/exec" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/devsy-org/devsy-provider-ssh/pkg/options" +) + +func TestRemoteExecShellRoundTrip(t *testing.T) { + for _, shell := range []string{"bash", "zsh", "fish"} { + t.Run(shell, func(t *testing.T) { + executable, err := exec.LookPath(shell) + if err != nil { + if os.Getenv("CI") != "" { + t.Fatalf("required shell %s unavailable", shell) + } + t.Skipf("%s unavailable", shell) + } + values := []string{ + "", "\n", "a'b", `a\b\\c`, + "$HOME `touch sentinel` (touch sentinel) $(touch sentinel)", + "UTF8: café 世界\ntrailing\n", "'\\'\n", + } + for _, value := range values { + scriptArg, err := quotePOSIXArgument(value) + requireNoError(t, err) + script := "printf '%s' " + scriptArg + command, err := buildPOSIXRemoteExec(script) + requireNoError(t, err) + // #nosec G204 -- Exercise the selected installed shell with controlled test data. + cmd := exec.Command(executable, "-c", command) + cmd.Dir = t.TempDir() + output, err := cmd.Output() + if err != nil { + t.Fatalf("%q: %v", value, err) + } + if string(output) != value { + t.Fatalf("got %q, want %q", output, value) + } + if _, err := os.Stat( + filepath.Join(cmd.Dir, "sentinel"), + ); !errors.Is( + err, + os.ErrNotExist, + ) { + t.Fatalf("substitution executed: %v", err) + } + } + }) + } +} + +func TestRemoteExecRejectsNUL(t *testing.T) { + if _, err := buildPOSIXRemoteExec("hello\x00world"); err == nil { + t.Fatal("accepted NUL") + } + if _, err := quotePOSIXArgument("hello\x00world"); err == nil { + t.Fatal("accepted NUL argument") + } +} + +func fakeExternalSSH(t *testing.T) *SSHProvider { + t.Helper() + dir := t.TempDir() + script := "#!/bin/sh\nwhile [ \"$#\" -gt 1 ]; do shift; done\nexec /bin/sh -c \"$1\"\n" + // #nosec G306 -- Test SSH stub must be executable. + if err := os.WriteFile(filepath.Join(dir, "ssh"), []byte(script), 0o700); err != nil { + t.Fatal(err) + } + t.Setenv("PATH", dir+string(os.PathListSeparator)+os.Getenv("PATH")) + return &SSHProvider{Config: &options.Options{Host: "unused", Port: "22"}} +} + +func TestExternalStreamsAndExit(t *testing.T) { + provider := fakeExternalSSH(t) + payload := []byte{0, 1, 2, 255, '\n'} + var stdout, stderr bytes.Buffer + err := Command( + context.Background(), + provider, + "cat; printf diagnostic >&2; exit 37", + Streams{Stdin: bytes.NewReader(payload), Stdout: &stdout, Stderr: &stderr}, + ) + var exit *exec.ExitError + if !errors.As(err, &exit) || exit.ExitCode() != 37 { + t.Fatalf("exit: %v", err) + } + if !bytes.Equal(stdout.Bytes(), payload) { + t.Fatalf("stdout changed: %q", stdout.Bytes()) + } + if stderr.String() != "diagnostic" { + t.Fatalf("stderr: %q", stderr.String()) + } +} + +func TestExternalCancellation(t *testing.T) { + for _, deadline := range []bool{false, true} { + name := "canceled" + if deadline { + name = "deadline" + } + t.Run(name, func(t *testing.T) { + provider := fakeExternalSSH(t) + ctx, cancel := context.WithCancel(context.Background()) + expected := context.Canceled + if deadline { + cancel() + ctx, cancel = context.WithTimeout(context.Background(), 100*time.Millisecond) + expected = context.DeadlineExceeded + } else { + timer := time.AfterFunc(100*time.Millisecond, cancel) + defer timer.Stop() + } + defer cancel() + start := time.Now() + err := Command(ctx, provider, "exec sleep 30", Streams{}) + if !errors.Is(err, expected) { + t.Fatalf("expected context identity %v: %v", expected, err) + } + if time.Since(start) > 5*time.Second { + t.Fatal("cancellation did not terminate SSH") + } + }) + } +} + +func TestSSHArguments(t *testing.T) { + provider := &SSHProvider{ + Config: &options.Options{ + Host: "alias", + Port: "2222", + ExtraFlags: `-i '/key with spaces' -F config`, + }, + } + args, err := getSSHCommand(provider) + if err != nil { + t.Fatal(err) + } + if strings.Join( + args, + "|", + ) != "-oStrictHostKeyChecking=no|-oBatchMode=yes|-p|2222|-i|/key with spaces|-F|config|alias" { + t.Fatalf("args: %q", args) + } +} + +func TestExactBootstrapArgument(t *testing.T) { + script := "\n'\\$HOME `touch sentinel` (touch sentinel) $(touch sentinel)\nUTF8: 世界\n" + launcher, err := buildPOSIXRemoteExec(script) + requireNoError(t, err) + dir := t.TempDir() + capture := filepath.Join(dir, "capture") + // #nosec G306 -- Argument capture fixture must be executable. + requireNoError(t, os.WriteFile(capture, []byte("#!/bin/sh\nprintf '%s\\000' \"$@\"\n"), 0o700)) + for _, shell := range []string{"bash", "zsh", "fish"} { + t.Run(shell, func(t *testing.T) { + executable, err := exec.LookPath(shell) + if err != nil { + if os.Getenv("CI") != "" { + t.Fatalf("required shell %s unavailable", shell) + } + t.Skipf("%s unavailable", shell) + } + command := capture + strings.TrimPrefix(launcher, "/bin/sh") + // #nosec G204 -- Capture shell argument boundaries with controlled test fixtures. + cmd := exec.Command(executable, "-c", command) + cmd.Dir = dir + out, err := cmd.Output() + requireNoError(t, err) + if string(out) != "-c\x00"+script+"\x00" { + t.Fatalf("argument boundaries changed: %q", out) + } + if _, err := os.Stat(filepath.Join(dir, "sentinel")); !errors.Is(err, os.ErrNotExist) { + t.Fatalf("substitution executed: %v", err) + } + }) + } +} + +func TestInitQuotesRemotePaths(t *testing.T) { + provider := fakeExternalSSH(t) + dir := t.TempDir() + for name, contents := range map[string]string{ + "uname": "#!/bin/sh\nprintf 'Linux\\n'\n", + "id": "#!/bin/sh\nprintf '1000\\n'\n", + } { + // #nosec G306 -- Controlled platform probes must be executable. + requireNoError(t, os.WriteFile(filepath.Join(dir, name), []byte(contents), 0o700)) + } + t.Setenv("PATH", dir+string(os.PathListSeparator)+os.Getenv("PATH")) + provider.Config.AgentPath = filepath.Join(dir, "agent 'quoted' $HOME `false` \\ path", "agent") + provider.Config.DockerPath = filepath.Join(dir, "docker 'quoted' $HOME \\ path") + // #nosec G306 -- Controlled Docker executable tests literal path semantics. + requireNoError( + t, + os.WriteFile(provider.Config.DockerPath, []byte("#!/bin/sh\ntest \"$1\" = ps\n"), 0o700), + ) + var stdout, stderr bytes.Buffer + requireNoError( + t, + Init(context.Background(), provider, Streams{Stdout: &stdout, Stderr: &stderr}), + ) + _, err := os.Stat(filepath.Dir(provider.Config.AgentPath)) + requireNoError(t, err) + if stdout.Len() != 0 { + t.Fatalf("init polluted stdout: %q", stdout.Bytes()) + } +} + +func TestInitRejectsNULPaths(t *testing.T) { + provider := &SSHProvider{ + Config: &options.Options{AgentPath: "/tmp/agent\x00suffix", DockerPath: "docker"}, + } + if err := Init(context.Background(), provider, Streams{}); err == nil { + t.Fatal("accepted NUL path") + } +} + +func TestRemoteUIDRejectsStartupOutput(t *testing.T) { + for _, output := range []string{"1000", "welcome\n1000\n", "1000\nextra\n", "1000\n\n"} { + _, err := remoteUID( + func(_ string, out io.Writer) error { _, err := io.WriteString(out, output); return err }, + ) + if err == nil { + t.Fatalf("accepted polluted UID stdout: %q", output) + } + } +} diff --git a/pkg/ssh/ssh.go b/pkg/ssh/ssh.go index 04ffd81..78c5013 100644 --- a/pkg/ssh/ssh.go +++ b/pkg/ssh/ssh.go @@ -3,19 +3,14 @@ package ssh import ( "bytes" "context" + "errors" "fmt" "io" - "net" - "os" - "os/exec" "path" - "path/filepath" + "strconv" "strings" - "time" "github.com/devsy-org/devsy-provider-ssh/pkg/options" - "github.com/devsy-org/devsy/pkg/log" - "github.com/devsy-org/devsy/pkg/ssh" "github.com/kballard/go-shellquote" ) @@ -38,17 +33,6 @@ func NewProvider() (*SSHProvider, error) { return provider, nil } -func returnSSHError(provider *SSHProvider, command string) error { - sshError := "Please make sure you have configured the correct SSH host\nand the following command can be executed on your system:\n" - - sshcmd, err := getSSHCommand(provider) - if err != nil { - return err - } - - return fmt.Errorf("%s ssh %s %s", sshError, strings.Join(sshcmd, " "), command) -} - func getSSHCommand(provider *SSHProvider) ([]string, error) { result := []string{"-oStrictHostKeyChecking=no", "-oBatchMode=yes"} @@ -59,7 +43,7 @@ func getSSHCommand(provider *SSHProvider) ([]string, error) { if provider.Config.ExtraFlags != "" { flags, err := shellquote.Split(provider.Config.ExtraFlags) if err != nil { - return nil, fmt.Errorf("error managing EXTRA_ARGS, %v", err) + return nil, fmt.Errorf("parse EXTRA_FLAGS: %w", err) } result = append(result, flags...) @@ -69,249 +53,133 @@ func getSSHCommand(provider *SSHProvider) ([]string, error) { return result, nil } -func execSSHCommand(provider *SSHProvider, command string, output io.Writer) error { - if provider.Config.UseBuiltinSSH { - // get ssh config for host - ctx, cancel := context.WithTimeout(context.Background(), time.Second*5) - defer cancel() - sshConfig, err := exec.CommandContext(ctx, "ssh", "-G", provider.Config.Host).Output() - if err != nil { - return fmt.Errorf("read ssh config for host %s: %w", provider.Config.Host, err) - } - hostname, user, port, identityfile := parseConfig(string(sshConfig)) - if hostname == "" || user == "" || port == "" { - return fmt.Errorf( - "resolve ssh config. Hostname='%s', User='%s', Port='%s'", - hostname, - user, - port, - ) - } - - // expand identityfile path - if strings.HasPrefix(identityfile, "~") { - identityfile = strings.Replace(identityfile, "~", "$userprofile", 1) - identityfile = os.ExpandEnv(identityfile) - } - abs, err := filepath.Abs(identityfile) - if err != nil { - return fmt.Errorf("absolute filepath: %w", err) - } - key, err := os.ReadFile(abs) - if err != nil { - return fmt.Errorf("read identifiyfile: %w", err) - } - - if provider.Config.Port != "" { - port = provider.Config.Port - } - // create ssh session - addr := net.JoinHostPort(hostname, port) - client, err := ssh.NewSSHClient(user, addr, key) - if err != nil { - return fmt.Errorf("create ssh client: %w", err) - } - sess, err := client.NewSession() - if err != nil { - return fmt.Errorf("create ssh session: %w", err) - } - sess.Stdin = os.Stdin - sess.Stdout = output - - return sess.Run(command) - } - - commandToRun, err := getSSHCommand(provider) - if err != nil { - return err - } - - commandToRun = append(commandToRun, command) - - var stderrBuf bytes.Buffer - - cmd := exec.Command("ssh", commandToRun...) - cmd.Stdin = os.Stdin - cmd.Stdout = output - cmd.Stderr = io.Writer(&stderrBuf) - - err = cmd.Run() - if err != nil { - log.Error(stderrBuf.String()) - return err +func Init(ctx context.Context, provider *SSHProvider, streams Streams) error { + if strings.ContainsRune(provider.Config.AgentPath, 0) || + strings.ContainsRune(provider.Config.DockerPath, 0) { + return fmt.Errorf("remote executable or agent path contains NUL") } - - // A non-POSIX shell has been detected: falling back to copy and execute scripts - if strings.Contains(stderrBuf.String(), "fish: Unsupported") { - log.Warn( - "A non-POSIX shell has been detected: falling back to copy and execute scripts", + run := func(script string, output io.Writer) error { + return provider.execRemoteScript( + ctx, + script, + Streams{Stdout: output, Stderr: streams.Stderr}, ) - - return copyAndExecSSHCommand(provider, command, output) } - - return err -} - -func copyAndExecSSHCommand(provider *SSHProvider, command string, output io.Writer) error { - commandToRun, err := getSSHCommand(provider) - if err != nil { + if err := checkRemotePlatform(run); err != nil { return err } - - script, err := copyCommandToRemote(provider, command) + uid, err := remoteUID(run) if err != nil { return err } - - commandToRun = append(commandToRun, []string{ - "/bin/sh", script, ";", "rm", "-f", script, - }...) - - cmd := exec.Command("ssh", commandToRun...) - cmd.Stdin = os.Stdin - cmd.Stdout = output - cmd.Stderr = os.Stderr - - return cmd.Run() -} - -func copyCommandToRemote(provider *SSHProvider, command string) (string, error) { - script, err := os.CreateTemp("", "devsy-command-*") - if err != nil { - return "", err - } - defer func() { - _ = script.Close() - _ = os.Remove(script.Name()) - }() - - commandToRun, err := getSCPCommand(provider, script.Name()) - if err != nil { - return "", err - } - - _, err = script.WriteString(command) - if err != nil { - return "", err + if uid == 0 { + return nil } - - return script.Name(), exec.Command("scp", commandToRun...).Run() + return provider.checkAccess(run) } -func getSCPCommand(provider *SSHProvider, sourcefile string) ([]string, error) { - result := []string{"-oStrictHostKeyChecking=no", "-oBatchMode=yes"} - - if provider.Config.Port != "22" { - result = append(result, []string{"-p", provider.Config.Port}...) - } - - if provider.Config.ExtraFlags != "" { - flags, err := shellquote.Split(provider.Config.ExtraFlags) - if err != nil { - return nil, fmt.Errorf("error managing EXTRA_ARGS, %v", err) +func checkRemotePlatform(run func(string, io.Writer) error) error { + for _, check := range []struct{ script, expected, stage string }{ + {"echo Devsy Test", "Devsy Test\n", "check remote /bin/sh and SSH connection"}, + {"uname", "Linux\n", "check Linux operating system"}, + } { + var out bytes.Buffer + if err := run(check.script, &out); err != nil { + return fmt.Errorf("%s: %w", check.stage, err) + } + if out.String() != check.expected { + return fmt.Errorf( + "%s: unexpected SSH stdout; check login shell startup output", + check.stage, + ) } - - result = append(result, flags...) } - - destfile := "/tmp/" + filepath.Base(sourcefile) - - result = append(result, sourcefile) - result = append(result, provider.Config.Host+":"+destfile) - return result, nil + return nil } -func Init(provider *SSHProvider) error { - out := new(bytes.Buffer) - // check that we can do outputs - err := execSSHCommand(provider, "echo Devsy Test", out) - if err != nil { - return returnSSHError(provider, "echo Devsy Test") +func remoteUID(run func(string, io.Writer) error) (uint64, error) { + var out bytes.Buffer + if err := run("id -ru", &out); err != nil { + return 0, fmt.Errorf("detect remote user: %w", err) } - if out.String() != "Devsy Test\n" { - return fmt.Errorf("error: ssh output mismatch") + uid, err := strconv.ParseUint(strings.TrimSuffix(out.String(), "\n"), 10, 32) + if err != nil || !strings.HasSuffix(out.String(), "\n") { + return 0, fmt.Errorf( + "detect remote user: unexpected SSH stdout; check login shell startup output", + ) } + return uid, nil +} - // We only support running on Linux ssh servers - out = new(bytes.Buffer) - err = execSSHCommand(provider, "uname", out) +func (provider *SSHProvider) checkAccess(run func(string, io.Writer) error) error { + agentDir, err := quotePOSIXArgument(path.Dir(provider.Config.AgentPath)) if err != nil { - return returnSSHError(provider, "uname") + return fmt.Errorf("agent path: %w", err) } - if out.String() != "Linux\n" { - fmt.Println(out.String()) - return fmt.Errorf("error: SSH provider only works on Linux servers") - } - - // If we're root, we won't have problems - out = new(bytes.Buffer) - err = execSSHCommand(provider, "id -ru", out) + dockerPath, err := quotePOSIXArgument(provider.Config.DockerPath) if err != nil { - return returnSSHError(provider, "id -ru") + return fmt.Errorf("docker path: %w", err) } - if out.String() == "0\n" { - return nil + if err := checkAgentDirectory(run, agentDir); err != nil { + return err } - - // check that we have access to AGENT_PATH - out = new(bytes.Buffer) - agentDir := path.Dir(provider.Config.AgentPath) - err1 := execSSHCommand(provider, "mkdir -p "+agentDir, out) - err2 := execSSHCommand(provider, "test -w "+agentDir, out) - if err1 != nil || err2 != nil { - err = execSSHCommand(provider, "sudo -nl", out) - if err != nil { + if err := run(dockerPath+" ps", io.Discard); err != nil { + if sudoErr := run("sudo -nl", io.Discard); sudoErr != nil { return fmt.Errorf( - "%s is not writable, passwordless sudo or root user required", - agentDir, + "docker executable inaccessible; passwordless sudo or docker group required: %w", + err, ) } } + return nil +} - // check that we have access to DOCKER_PATH - err = execSSHCommand(provider, provider.Config.DockerPath+" ps", out) - if err != nil { - err = execSSHCommand(provider, "sudo -nl", out) - if err != nil { - return fmt.Errorf( - "%s not found, passwordless sudo or root user required. If using another user please add to the docker group", - provider.Config.DockerPath, - ) - } +func checkAgentDirectory(run func(string, io.Writer) error, agentDir string) error { + mkdirErr := run("mkdir -p -- "+agentDir, io.Discard) + writableErr := run("test -w "+agentDir, io.Discard) + if mkdirErr == nil && writableErr == nil { + return nil + } + if err := run("sudo -nl", io.Discard); err != nil { + return fmt.Errorf( + "agent directory inaccessible; passwordless sudo or root required: %w", + errors.Join(mkdirErr, writableErr, err), + ) } - return nil } -func Command(provider *SSHProvider, command string) error { - return execSSHCommand(provider, command, os.Stdout) +func Command(ctx context.Context, provider *SSHProvider, command string, streams Streams) error { + return provider.execRemoteScript(ctx, command, streams) } -func parseConfig(config string) (hostname string, user string, port string, identityfile string) { - for _, line := range strings.Split(config, "\n") { - fields := strings.Fields(line) - if len(fields) != 2 { +type resolvedConfig struct{ hostname, user, port, identityfile string } + +func parseConfig(config string) resolvedConfig { + var resolved resolvedConfig + for line := range strings.SplitSeq(config, "\n") { + key, value, found := strings.Cut(line, " ") + if !found { continue } - if fields[0] == "hostname" { - hostname = fields[1] + if key == "hostname" { + resolved.hostname = strings.TrimSpace(value) continue } - if fields[0] == "user" { - user = fields[1] + if key == "user" { + resolved.user = strings.TrimSpace(value) continue } - if fields[0] == "port" { - port = fields[1] + if key == "port" { + resolved.port = strings.TrimSpace(value) continue } // just take the first one - if fields[0] == "identityfile" && identityfile == "" { - identityfile = fields[1] + if key == "identityfile" && resolved.identityfile == "" { + resolved.identityfile = strings.TrimSpace(value) continue } } - return hostname, user, port, identityfile + return resolved } diff --git a/pkg/ssh/stdin_copy.go b/pkg/ssh/stdin_copy.go new file mode 100644 index 0000000..30b17ff --- /dev/null +++ b/pkg/ssh/stdin_copy.go @@ -0,0 +1,80 @@ +package ssh + +import ( + "errors" + "io" + "net" + "os" + "sync/atomic" +) + +type stdinCopy struct { + reader io.ReadCloser + stopping atomic.Bool + done chan struct{} + sourceErr error + sourceStopped bool + copyErr error + destinationErr error +} + +func startStdinCopy(reader io.ReadCloser, writer io.WriteCloser) *stdinCopy { + copy := &stdinCopy{reader: reader, done: make(chan struct{})} + go func() { + defer close(copy.done) + _, copy.copyErr = io.Copy( + trackedStdinWriter{Writer: writer, copy: copy}, + trackedStdin{copy: copy}, + ) + _ = writer.Close() + }() + return copy +} + +type trackedStdin struct{ copy *stdinCopy } + +func (input trackedStdin) Read(data []byte) (int, error) { + n, err := input.copy.reader.Read(data) + if err != nil && !errors.Is(err, io.EOF) { + input.copy.sourceErr = err + input.copy.sourceStopped = input.copy.stopping.Load() + } + return n, err +} + +func (copy *stdinCopy) finish() error { + copy.stopping.Store(true) + _ = copy.reader.Close() + <-copy.done + // Source errors remain meaningful when a successful remote command closes its input. + if copy.sourceErr != nil && (!copy.sourceStopped || !isStdinShutdown(copy.sourceErr)) { + return copy.sourceErr + } + if copy.destinationErr != nil && isStdinDestinationClosed(copy.destinationErr) { + return nil + } + if copy.sourceStopped && isStdinShutdown(copy.copyErr) { + return nil + } + return copy.copyErr +} + +func isStdinShutdown(err error) bool { + return errors.Is(err, io.ErrClosedPipe) || + errors.Is(err, os.ErrClosed) || errors.Is(err, net.ErrClosed) || errors.Is(err, io.EOF) +} + +type trackedStdinWriter struct { + io.Writer + copy *stdinCopy +} + +func (writer trackedStdinWriter) Write(data []byte) (int, error) { + n, err := writer.Writer.Write(data) + writer.copy.destinationErr = err + return n, err +} + +func isStdinDestinationClosed(err error) bool { + return isStdinShutdown(err) || isPipeWriteClosed(err) +} diff --git a/pkg/ssh/stdin_lifecycle_test.go b/pkg/ssh/stdin_lifecycle_test.go new file mode 100644 index 0000000..3c9b95e --- /dev/null +++ b/pkg/ssh/stdin_lifecycle_test.go @@ -0,0 +1,270 @@ +package ssh + +import ( + "bytes" + "context" + "errors" + "io" + "os" + "os/exec" + "path/filepath" + "sync" + "sync/atomic" + "syscall" + "testing" + "time" + + cryptossh "golang.org/x/crypto/ssh" +) + +const contextCancelScenario = "context-cancel" + +func TestExternalCancellationPreservesSourceError(t *testing.T) { + provider := fakeExternalSSH(t) + input := newLateErrorInput() + defer func() { _ = input.Close() }() + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + result := make(chan error, 1) + go func() { result <- Command(ctx, provider, "exec cat", Streams{Stdin: input}) }() + select { + case <-input.started: + case <-time.After(3 * time.Second): + t.Fatal("stdin copy did not start") + } + cancel() + err := waitInputResult(t, result, input) + if !errors.Is(err, context.Canceled) || !errors.Is(err, errInputRead) { + t.Fatalf("lost cancellation or source error: %v", err) + } + if _, ok := errors.AsType[*exec.ExitError](err); !ok { + t.Fatalf("lost started SSH process error: %v", err) + } +} + +type lateErrorInput struct { + started chan struct{} + released chan struct{} + start sync.Once + close sync.Once +} + +func newLateErrorInput() *lateErrorInput { + return &lateErrorInput{started: make(chan struct{}), released: make(chan struct{})} +} + +func (input *lateErrorInput) Read([]byte) (int, error) { + input.start.Do(func() { close(input.started) }) + <-input.released + return 0, errInputRead +} + +func (input *lateErrorInput) Close() error { + input.close.Do(func() { close(input.released) }) + return nil +} + +func waitInputResult(t *testing.T, result <-chan error, input io.Closer) error { + t.Helper() + select { + case err := <-result: + return err + case <-time.After(3 * time.Second): + _ = input.Close() + select { + case err := <-result: + t.Logf("stdin cleanup after timeout: %v", err) + case <-time.After(3 * time.Second): + } + t.Fatal("SSH command did not join its idle stdin copier") + return nil + } +} + +type observedInput struct { + io.ReadCloser + started chan struct{} + once sync.Once + closed atomic.Int32 +} + +func (input *observedInput) Read(data []byte) (int, error) { + input.once.Do(func() { close(input.started) }) + return input.ReadCloser.Read(data) +} + +func (input *observedInput) Close() error { + input.closed.Add(1) + return input.ReadCloser.Close() +} + +func TestExternalIdlePipeCompletion(t *testing.T) { + for _, test := range []struct { + name, script string + cancel bool + }{ + {"remote-exit", "exit 0", false}, + {contextCancelScenario, "exec cat", true}, + } { + t.Run(test.name, func(t *testing.T) { + provider := fakeExternalSSH(t) + reader, writer := io.Pipe() + defer func() { _ = reader.Close(); _ = writer.Close() }() + input := &observedInput{ReadCloser: reader, started: make(chan struct{})} + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + result := make(chan error, 1) + go func() { result <- Command(ctx, provider, test.script, Streams{Stdin: input}) }() + select { + case <-input.started: + case <-time.After(3 * time.Second): + t.Fatal("stdin copy did not start") + } + if test.cancel { + cancel() + } + err := waitInputResult(t, result, reader) + if input.closed.Load() != 1 { + t.Fatalf("owned input closed %d times", input.closed.Load()) + } + if (err != nil) != test.cancel { + t.Fatalf("unexpected completion: %v", err) + } + }) + } +} + +func TestBuiltinLateInputError(t *testing.T) { + input := newLateErrorInput() + defer func() { _ = input.Close() }() + provider := builtinTestProvider( + t, + func(channel cryptossh.Channel, requests <-chan *cryptossh.Request) { + request := <-requests + _ = request.Reply(true, nil) + <-input.started + _, _ = channel.SendRequest( + "exit-status", + false, + cryptossh.Marshal(struct{ Status uint32 }{0}), + ) + _ = channel.Close() + }, + ) + result := make(chan error, 1) + go func() { result <- Command(context.Background(), provider, "exit 0", Streams{Stdin: input}) }() + if err := waitInputResult(t, result, input); !errors.Is(err, errInputRead) { + t.Fatalf("late genuine input error lost: %v", err) + } +} + +func TestExternalKeepsCallerFileOpen(t *testing.T) { + provider := fakeExternalSSH(t) + reader, writer, err := os.Pipe() + requireNoError(t, err) + defer func() { _ = reader.Close(); _ = writer.Close() }() + requireNoError(t, Command(context.Background(), provider, "exit 0", Streams{Stdin: reader})) + _, err = reader.Stat() + requireNoError(t, err) +} + +func TestExternalLateInputError(t *testing.T) { + provider := fakeExternalSSH(t) + input := newLateErrorInput() + defer func() { _ = input.Close() }() + result := make(chan error, 1) + go func() { result <- Command(context.Background(), provider, "exit 0", Streams{Stdin: input}) }() + if err := waitInputResult(t, result, input); !errors.Is(err, errInputRead) { + t.Fatalf("late genuine input error lost: %v", err) + } +} + +func TestBuiltinIdlePipeCompletion(t *testing.T) { + reader, writer := io.Pipe() + defer func() { _ = reader.Close(); _ = writer.Close() }() + input := &observedInput{ReadCloser: reader, started: make(chan struct{})} + provider := builtinTestProvider( + t, + func(channel cryptossh.Channel, requests <-chan *cryptossh.Request) { + request := <-requests + _ = request.Reply(true, nil) + <-input.started + _, _ = channel.SendRequest( + "exit-status", + false, + cryptossh.Marshal(struct{ Status uint32 }{0}), + ) + _ = channel.Close() + }, + ) + result := make(chan error, 1) + go func() { result <- Command(context.Background(), provider, "exit 0", Streams{Stdin: input}) }() + requireNoError(t, waitInputResult(t, result, input)) +} + +func TestExternalEarlyExitWithLargeInput(t *testing.T) { + provider := fakeExternalSSH(t) + for range 5 { + requireNoError( + t, + Command( + context.Background(), + provider, + "exit 0", + Streams{Stdin: bytes.NewReader(make([]byte, 1<<20))}, + ), + ) + } +} + +type sourcePipeError struct{} + +func (sourcePipeError) Read([]byte) (int, error) { return 0, syscall.EPIPE } +func (sourcePipeError) Close() error { return nil } + +func TestGenuineSourcePipeError(t *testing.T) { + for _, builtin := range []bool{false, true} { + t.Run(map[bool]string{false: "external", true: "builtin"}[builtin], func(t *testing.T) { + provider := fakeExternalSSH(t) + if builtin { + provider = builtinTestProvider( + t, + func(channel cryptossh.Channel, requests <-chan *cryptossh.Request) { + defer func() { _ = channel.Close() }() + request := <-requests + _ = request.Reply(true, nil) + _, _ = io.Copy(io.Discard, channel) + _, _ = channel.SendRequest( + "exit-status", + false, + cryptossh.Marshal(struct{ Status uint32 }{0}), + ) + }, + ) + } + err := Command(context.Background(), provider, "cat", Streams{Stdin: sourcePipeError{}}) + if !errors.Is(err, syscall.EPIPE) { + t.Fatalf("genuine source EPIPE lost: %v", err) + } + }) + } +} + +func TestExternalStartFailureClosesOwnedInput(t *testing.T) { + input := newLateErrorInput() + // #nosec G204 -- Intentionally nonexistent controlled path exercises start failure. + cmd := exec.CommandContext(context.Background(), filepath.Join(t.TempDir(), "missing")) + if err := runExternal(cmd, input); !errors.Is(err, os.ErrNotExist) { + t.Fatalf("start error: %v", err) + } + select { + case <-input.released: + default: + t.Fatal("owned stdin not closed on start failure") + } + select { + case <-input.started: + t.Fatal("stdin copier started despite process start failure") + default: + } +}