From c0c159aaed22ea789515380406e655b945006f29 Mon Sep 17 00:00:00 2001 From: Samuel K Date: Tue, 20 Oct 2026 13:35:08 -0600 Subject: [PATCH 01/10] fix: execute SSH bootstrap through a shell-safe POSIX launcher --- .github/workflows/tests.yaml | 23 ++ README.md | 14 ++ cmd/command.go | 17 +- cmd/init.go | 17 +- cmd/root.go | 18 +- docs/verification/1621-ssh-bootstrap.md | 117 +++++++++ e2e/tests/integration/integration.go | 12 +- e2e/tests/integration/login_shells.go | 68 ++++++ go.mod | 4 +- pkg/ssh/builtin_test.go | 228 ++++++++++++++++++ pkg/ssh/input_unix.go | 22 ++ pkg/ssh/input_windows.go | 48 ++++ pkg/ssh/input_windows_test.go | 179 ++++++++++++++ pkg/ssh/remote_command.go | 211 +++++++++++++++++ pkg/ssh/remote_command_test.go | 219 +++++++++++++++++ pkg/ssh/ssh.go | 302 +++++++----------------- 16 files changed, 1261 insertions(+), 238 deletions(-) create mode 100644 docs/verification/1621-ssh-bootstrap.md create mode 100644 e2e/tests/integration/login_shells.go create mode 100644 pkg/ssh/builtin_test.go create mode 100644 pkg/ssh/input_unix.go create mode 100644 pkg/ssh/input_windows.go create mode 100644 pkg/ssh/input_windows_test.go create mode 100644 pkg/ssh/remote_command.go create mode 100644 pkg/ssh/remote_command_test.go diff --git a/.github/workflows/tests.yaml b/.github/workflows/tests.yaml index 213f134..8bbfc6e 100644 --- a/.github/workflows/tests.yaml +++ b/.github/workflows/tests.yaml @@ -16,6 +16,9 @@ jobs: 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: @@ -27,6 +30,26 @@ jobs: version: "~> v2" args: release --snapshot --clean + - 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 + - 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..5a14b35 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{Stdin: os.Stdin, Stdout: os.Stdout, Stderr: os.Stderr}, + ) } diff --git a/cmd/root.go b/cmd/root.go index e2945c1..df3795f 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,12 +42,14 @@ func Execute() { rootCmd := BuildRoot() // execute command - err := rootCmd.Execute() + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + err := rootCmd.ExecuteContext(ctx) + stop() if err != nil { - if exitErr, ok := err.(*ssh.ExitError); ok { - os.Exit(exitErr.ExitStatus()) + if sshExit, ok := errors.AsType[*ssh.ExitError](err); ok { + os.Exit(sshExit.ExitStatus()) } - if exitErr, ok := err.(*exec.ExitError); ok { + if exitErr, ok := errors.AsType[*exec.ExitError](err); ok { if len(exitErr.Stderr) > 0 { log.Error(string(exitErr.Stderr)) } @@ -54,7 +60,7 @@ func Execute() { } } -// BuildRoot creates a new root command from the +// BuildRoot creates the SSH provider command tree. func BuildRoot() *cobra.Command { rootCmd := NewRootCmd() diff --git a/docs/verification/1621-ssh-bootstrap.md b/docs/verification/1621-ssh-bootstrap.md new file mode 100644 index 0000000..b40e546 --- /dev/null +++ b/docs/verification/1621-ssh-bootstrap.md @@ -0,0 +1,117 @@ +# SSH bootstrap regression verification + +The provider builds one login-shell-safe `/bin/sh -c` launcher before transport +selection. Its original multiline script is one argument; stdin remains reserved +for the bootstrap handshake, binary injection, and the live tunnel. No SCP or +stderr-matching fallback remains. + +## Revisions and environment + +- Reported provider v1.1.44: `e37a6c68a695d9d1d3806b2d42a755750816f60e`. +- Initial provider main: `9c5c3ff673c364818d31627d45f365053100fa4b`. +- Final base refreshed to `839369fc438c4616bace5ee15994ab42b0fd0773`; upstream + changed only the shfmt hook version. The isolated branch was rebased cleanly. +- Original core inspected by the coordinated core change: + `1f47cd185` (v1.24.0). +- Linux qualification: Ubuntu aarch64 container with actual OpenSSH and separate + non-root passwd accounts using Bash, Zsh, Fish, Dash, and BusyBox Ash. +- Local unit tests execute installed Bash, Zsh, and Fish on macOS arm64. These + shell tests are distinct from the Linux SSH qualification. +- The provider's core module remains pinned to v1.23.2. No API upgrade was needed. + The existing full-workspace E2E download is pinned to core v1.24.0, with + `DEVSY_E2E_BINARY` selecting an explicitly supplied local core executable. + +## Reproduced baseline + +The original core bootstrap and original provider passed with Bash through both +transports. External Zsh failed with `command not found: sh -c`; external Fish +failed with unsupported assignment syntax. The built-in transport failed with +exit 127 and lost the useful remote stderr diagnostics. + +## Implemented regression checks + +- Exact launcher argument boundaries and script byte round trips through real + Bash, Zsh, and Fish, including empty input, newlines, apostrophes, backslashes, + dollar signs, backticks, parentheses, UTF-8, and substitution sentinels. +- NUL rejection before remote execution. +- External and built-in transport byte-preserving stdin/stdout, separate stderr, + wrapped exit-status errors, and cancellation. +- Built-in cancellation with an idle file pipe and an idle `io.PipeReader`, with + input-copy completion joined before returning. Non-file closable input belongs + to the session; unsupported potentially unbounded readers are rejected. +- Local stdin read failures remain errors even when the remote consumer exits + successfully after EOF. Independent review identified and verified this fix. +- Windows uses the actual redirected pipe handle, duplicated for session-owned + cancellation rather than opening a console. Native Windows CI covers redirected + process stdin plus an in-process SSH session's pipe bytes and idle cancellation. + Native Windows execution is NOT RUN locally; the final Windows suite + cross-compiled successfully and Windows-target lint reported zero issues. +- Literal agent directory and Docker executable paths containing shell syntax. +- Linux CI provisions Bash/Zsh/Fish passwd accounts and exercises all three with + both transports using binary streams and an exit-37 command. Missing required + shells fail unit tests in CI. + +## Coordinated Linux observations + +The initial corrected provider and corrected core bootstrap passed all 30 cells: +five passwd shells, both transports, and cold/warm/forced-mismatch injection. +Each cell preserved a 4 KiB all-byte payload and a follow-on exchange sent only +after receiving output. Warm runs requested no replacement binary; cold and +forced-mismatch runs requested exactly one. This uses a synthetic agent payload +and must not be confused with the real-agent check below. + +The subsequent `final5` build passed all 30 randomized/hostile-path synthetic +cells and all 10 real-agent shell/transport combinations. The latter started the +actual 207 MB Linux Devsy agent, received its `SSH-2.0-Go` banner, and completed +two sequential live commands per combination. SIGTERM ended both provider +transports in under half a second, left no local SSH child, and left no remote +Devsy agent. Generic non-PTY remote `sleep` commands survive connection closure; +arbitrary remote process reaping is outside this SSH transport contract. + +The final5 binary SHA256 was +`7e18087de9fbc9137a7a561c6fce200b7fe50d523fd99db08b6e8ce7b27015c7`. +Later stdin-source-error and Windows fixes were independently reviewed. The +reviewed final-source binary SHA256 is +`39468e46550ddf4a9ea3a1033cae10ef02d0feabcce42ad7b89a7f88f40ba989`. +Its production diff fingerprint (commands, SSH implementation, module files) is +`fde9ee3eceaba52ab03246f0d59151c7070c967168edbce6b63a04ec1ddef1a0`. +The final rerun passed: + +- Fixed core + reviewed provider: all 30 synthetic cold/warm/mismatch cells. +- Original core v1.24.0 + reviewed provider: Bash/Zsh/Fish through both transports, + all six cold injection and live follow-on cells. +- Actual Linux Devsy agent: all ten shell/transport combinations, two live + commands per session. +- SIGTERM after the actual agent banner: external 0.459 s, built-in 0.405 s; + no local SSH child and no remote Devsy agent remained. +- Initialization through all five shells and both transports: ten passes. +- Exact stdout, separate stderr, and exit 37 through all five shells and both + transports: ten passes. + +The artifact was built before the final base's hook-only update; production +source is unchanged. All observations are Linux aarch64 SSH qualification, +not Windows or GUI acceptance. + +## Local validation + +- CI-parity unit suite: `go test -race -covermode=atomic . ./cmd/... ./hack/... ./pkg/...` PASS. +- SSH package statement coverage: 78.1%. +- E2E package compilation without host setup: PASS. +- Diff-scoped golangci-lint with macOS and Windows build contexts: zero issues. +- `prek run --all-files`: all applicable hooks PASS on the refreshed base. +- Local GoReleaser snapshot with publishing disabled: all five configured target + builds PASS. This is cross-compilation, not native Windows execution. +- Provider YAML generation with genuine snapshot checksums: PASS. The existing + `${SSH_PROVIDER} init` and `${SSH_PROVIDER} command` interface is unchanged. + No generated descriptor was added to the source tree. + +## Review gates + +Local CodeRabbit authentication was verified, but automatic approval review +rejected the review invocation because it uploads uncommitted and untracked +private source. That gate is BLOCKED pending explicit user authorization for the +concrete diff. No workaround or indirect upload was attempted. + +CI and final-SHA Greptile/CodeRabbit reviews are pending draft PR publication and +must not be reported as passed. No merge, tag, release, or customer reply is part +of this change. diff --git a/e2e/tests/integration/integration.go b/e2e/tests/integration/integration.go index d0444cb..d715875 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) 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..95b5e2b 100644 --- a/go.mod +++ b/go.mod @@ -6,10 +6,12 @@ require ( github.com/devsy-org/devsy v1.23.2 github.com/goccy/go-yaml v1.19.2 github.com/kballard/go-shellquote v0.0.0-20180428030007-95032a82bc51 + github.com/muesli/cancelreader v0.2.2 github.com/onsi/ginkgo/v2 v2.33.1 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 ( @@ -240,7 +242,6 @@ require ( github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect github.com/modern-go/reflect2 v1.0.3-0.20250322232337-35a7c28c31ee // indirect github.com/morikuni/aec v1.1.0 // indirect - github.com/muesli/cancelreader v0.2.2 // indirect github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect github.com/oklog/run v1.1.0 // indirect github.com/opencontainers/go-digest v1.0.0 // indirect @@ -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_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/input_unix.go b/pkg/ssh/input_unix.go new file mode 100644 index 0000000..842b564 --- /dev/null +++ b/pkg/ssh/input_unix.go @@ -0,0 +1,22 @@ +//go:build !windows + +package ssh + +import ( + "io" + "os" + + "github.com/muesli/cancelreader" +) + +func cancellableFile(reader *os.File) (cancelreader.CancelReader, error) { + info, err := reader.Stat() + if err != nil { + return nil, err + } + nullInfo, _ := os.Stat(os.DevNull) + if info.Mode().IsRegular() || (nullInfo != nil && os.SameFile(info, nullInfo)) { + return cancelreader.NewReader(struct{ io.Reader }{reader}) + } + return cancelreader.NewReader(reader) +} diff --git a/pkg/ssh/input_windows.go b/pkg/ssh/input_windows.go new file mode 100644 index 0000000..f6635b5 --- /dev/null +++ b/pkg/ssh/input_windows.go @@ -0,0 +1,48 @@ +package ssh + +import ( + "fmt" + "io" + "os" + + "github.com/muesli/cancelreader" + "golang.org/x/sys/windows" +) + +func cancellableFile(file *os.File) (cancelreader.CancelReader, 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 cancelreader.NewReader(struct{ io.Reader }{file}) + } + if kind != windows.FILE_TYPE_PIPE { + var mode uint32 + if windows.GetConsoleMode(windows.Handle(file.Fd()), &mode) != nil { + return cancelreader.NewReader(struct{ io.Reader }{file}) + } + return cancelreader.NewReader(file) + } + // Own a duplicate so closing the session interrupts pipe I/O without closing caller stdin. + 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 pipe: %w", err) + } + owned := os.NewFile(uintptr(duplicate), file.Name()) + if owned == nil { + _ = windows.CloseHandle(duplicate) + return nil, fmt.Errorf("open duplicated SSH input pipe") + } + return &closingInput{ReadCloser: owned}, nil +} diff --git a/pkg/ssh/input_windows_test.go b/pkg/ssh/input_windows_test.go new file mode 100644 index 0000000..0255b8e --- /dev/null +++ b/pkg/ssh/input_windows_test.go @@ -0,0 +1,179 @@ +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.Cancel() }) + 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 +} diff --git a/pkg/ssh/remote_command.go b/pkg/ssh/remote_command.go new file mode 100644 index 0000000..355d972 --- /dev/null +++ b/pkg/ssh/remote_command.go @@ -0,0 +1,211 @@ +package ssh + +import ( + "bytes" + "context" + "errors" + "fmt" + "io" + "net" + "os" + "os/exec" + "path/filepath" + "strings" + "time" + + devsyssh "github.com/devsy-org/devsy/pkg/ssh" + "github.com/muesli/cancelreader" + cryptossh "golang.org/x/crypto/ssh" +) + +// Streams keeps protocol bytes separate from diagnostics. Non-file Stdin closers +// are closed when the built-in session ends; other readers must be finite buffers. +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.Stdin, cmd.Stdout, cmd.Stderr = streams.Stdin, streams.Stdout, streams.Stderr + if err := cmd.Run(); err != nil { + return fmt.Errorf("execute remote /bin/sh script: %w", err) + } + return nil +} + +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", 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 { + config, addr, err := provider.builtinConfig(ctx, streams.Stderr) + if err != nil { + return err + } + conn, err := (&net.Dialer{}).DialContext(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 { + if ctx.Err() != nil { + return fmt.Errorf("execute remote /bin/sh script: %w", ctx.Err()) + } + 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 + } + copied := make(chan struct{}) + copyResult := make(chan error, 1) + go func() { + defer close(copied) + _, copyErr := io.Copy(stdin, reader) + copyResult <- copyErr + _ = stdin.Close() + }() + err = session.Wait() + var copyErr error + select { + case copyErr = <-copyResult: + default: + } + reader.Cancel() + <-copied + return errors.Join(err, copyErr) +} + +func cancellableInput(input io.Reader) (cancelreader.CancelReader, error) { + switch reader := input.(type) { + case *os.File: + return cancellableFile(reader) + case *bytes.Buffer, *bytes.Reader, *strings.Reader: + return cancelreader.NewReader(input) + 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 } + +func (input *closingInput) Cancel() bool { return input.ReadCloser.Close() == nil } + +func (input *closingInput) Close() error { return input.ReadCloser.Close() } diff --git a/pkg/ssh/remote_command_test.go b/pkg/ssh/remote_command_test.go new file mode 100644 index 0000000..795c8bb --- /dev/null +++ b/pkg/ssh/remote_command_test.go @@ -0,0 +1,219 @@ +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) { + provider := fakeExternalSSH(t) + ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) + defer cancel() + start := time.Now() + if err := Command(ctx, provider, "exec sleep 30", Streams{}); err == nil { + t.Fatal("expected cancellation") + } + 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..00df6ad 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{Stdin: streams.Stdin, 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 } From 4722dd2c489ed2b440dfcacb6585ea6e2e97a1c1 Mon Sep 17 00:00:00 2001 From: Samuel K Date: Tue, 20 Oct 2026 14:01:00 -0600 Subject: [PATCH 02/10] test: verify remote command failures without shell-specific wording --- docs/verification/1621-ssh-bootstrap.md | 15 +++++++++++++++ e2e/tests/integration/integration.go | 13 ++++++++----- 2 files changed, 23 insertions(+), 5 deletions(-) diff --git a/docs/verification/1621-ssh-bootstrap.md b/docs/verification/1621-ssh-bootstrap.md index b40e546..bb1b5c0 100644 --- a/docs/verification/1621-ssh-bootstrap.md +++ b/docs/verification/1621-ssh-bootstrap.md @@ -105,6 +105,21 @@ not Windows or GUI acceptance. `${SSH_PROVIDER} init` and `${SSH_PROVIDER} command` interface is unchanged. No generated descriptor was added to the source tree. +## First CI run and test correction + +On provider head `c0c159aaed22ea789515380406e655b945006f29`, native Windows pipe +checks, lint, pre-commit, unit tests, snapshot builds, and Linux shell-account +provisioning passed. Linux E2E reported 14 passes, one failure, and three skips: +the existing failing-command assertion required Bash's English phrase +`command not found`, while Dash correctly emitted +`/bin/sh: 1: not-a-command: not found`. The follow-up changes only that assertion +to require exit 127, empty stdout, and separate nonempty stderr containing the +command name. The three subsequent ordered whole-workspace CLI checks were +skipped on that first run and must not be reported as acceptance passes. + +The production source fingerprint and reviewed binary above remain unchanged; +the final-head CI and reviewer gates must run again after this test-only fix. + ## Review gates Local CodeRabbit authentication was verified, but automatic approval review diff --git a/e2e/tests/integration/integration.go b/e2e/tests/integration/integration.go index d715875..3e5fa01 100644 --- a/e2e/tests/integration/integration.go +++ b/e2e/tests/integration/integration.go @@ -251,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() { From e662d5359c37ee574d644efaa71bc1bfa007a935 Mon Sep 17 00:00:00 2001 From: Samuel K Date: Tue, 20 Oct 2026 14:42:27 -0600 Subject: [PATCH 03/10] fix: join SSH stdin copies and preserve source failures --- docs/verification/1621-ssh-bootstrap.md | 34 +++- pkg/ssh/input_unix.go | 4 + pkg/ssh/input_windows.go | 5 + pkg/ssh/input_windows_test.go | 43 +++++ pkg/ssh/remote_command.go | 66 ++++--- pkg/ssh/stdin_copy.go | 82 ++++++++ pkg/ssh/stdin_lifecycle_test.go | 247 ++++++++++++++++++++++++ 7 files changed, 456 insertions(+), 25 deletions(-) create mode 100644 pkg/ssh/stdin_copy.go create mode 100644 pkg/ssh/stdin_lifecycle_test.go diff --git a/docs/verification/1621-ssh-bootstrap.md b/docs/verification/1621-ssh-bootstrap.md index bb1b5c0..63930ea 100644 --- a/docs/verification/1621-ssh-bootstrap.md +++ b/docs/verification/1621-ssh-bootstrap.md @@ -95,7 +95,7 @@ not Windows or GUI acceptance. ## Local validation - CI-parity unit suite: `go test -race -covermode=atomic . ./cmd/... ./hack/... ./pkg/...` PASS. -- SSH package statement coverage: 78.1%. +- SSH package statement coverage after the stdin review fixes: 81.4%. - E2E package compilation without host setup: PASS. - Diff-scoped golangci-lint with macOS and Windows build contexts: zero issues. - `prek run --all-files`: all applicable hooks PASS on the refreshed base. @@ -117,8 +117,36 @@ to require exit 127, empty stdout, and separate nonempty stderr containing the command name. The three subsequent ordered whole-workspace CLI checks were skipped on that first run and must not be reported as acceptance passes. -The production source fingerprint and reviewed binary above remain unchanged; -the final-head CI and reviewer gates must run again after this test-only fix. +The production source fingerprint and reviewed binary remained unchanged for +that test-only correction. The subsequent stdin review fixes below change +production code and require new final-head CI and reviewer coverage. + +## Greptile stdin lifecycle follow-up + +Greptile reviewed `4722dd2c489ed2b440dfcacb6585ea6e2e97a1c1` at confidence 4/5 +and identified two valid issues. External SSH could hang after cancellation or +remote exit with an idle non-file stdin reader. Built-in SSH could discard a +genuine source error that arrived after remote completion. + +Both were reproduced before the fix: a bounded idle-input test timed out and a +late custom read error incorrectly returned success. Both transports now join +owned non-file stdin copies before reading their final results. External file +stdin remains borrowed directly. Source read errors are recorded separately +from destination writes, preserving genuine source failures (including EPIPE) +while allowing a successful remote command to close unread input. Owned +non-file closers are closed once, including start failure. + +Regressions cover idle-input exit zero, cancellation, late custom source errors, +source EPIPE, unread 1 MiB input with remote success, and ownership cleanup. +Native Windows CI also exercises the external-child variants without a Unix +shell stub. Independent bounded review passed the fixes, including five runs +of the unread-large-input reproducer. Local race tests pass; new final-head +CI and fresh Greptile/CodeRabbit reviews remain required. + +The earlier runtime artifact fingerprints and matrices are explicitly evidence +for the earlier production revision. The launcher encoding is unchanged by +this stdin follow-up; they are not claimed as runtime execution of the new +stdin implementation. ## Review gates diff --git a/pkg/ssh/input_unix.go b/pkg/ssh/input_unix.go index 842b564..875df5d 100644 --- a/pkg/ssh/input_unix.go +++ b/pkg/ssh/input_unix.go @@ -3,8 +3,10 @@ package ssh import ( + "errors" "io" "os" + "syscall" "github.com/muesli/cancelreader" ) @@ -20,3 +22,5 @@ func cancellableFile(reader *os.File) (cancelreader.CancelReader, error) { } return cancelreader.NewReader(reader) } + +func isPipeWriteClosed(err error) bool { return errors.Is(err, syscall.EPIPE) } diff --git a/pkg/ssh/input_windows.go b/pkg/ssh/input_windows.go index f6635b5..857a790 100644 --- a/pkg/ssh/input_windows.go +++ b/pkg/ssh/input_windows.go @@ -1,6 +1,7 @@ package ssh import ( + "errors" "fmt" "io" "os" @@ -46,3 +47,7 @@ func cancellableFile(file *os.File) (cancelreader.CancelReader, error) { } return &closingInput{ReadCloser: 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 index 0255b8e..e182973 100644 --- a/pkg/ssh/input_windows_test.go +++ b/pkg/ssh/input_windows_test.go @@ -177,3 +177,46 @@ func (writer readyWriter) Write(data []byte) (int, error) { } 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 index 355d972..a642f20 100644 --- a/pkg/ssh/remote_command.go +++ b/pkg/ssh/remote_command.go @@ -11,6 +11,7 @@ import ( "os/exec" "path/filepath" "strings" + "sync" "time" devsyssh "github.com/devsy-org/devsy/pkg/ssh" @@ -19,7 +20,7 @@ import ( ) // Streams keeps protocol bytes separate from diagnostics. Non-file Stdin closers -// are closed when the built-in session ends; other readers must be finite buffers. +// are closed when either transport ends; other readers must be finite buffers. type Streams struct { Stdin io.Reader Stdout io.Writer @@ -60,13 +61,40 @@ func (provider *SSHProvider) execRemoteScript( } // #nosec G204 -- SSH arguments are explicit provider options; script is one encoded argument. cmd := exec.CommandContext(ctx, "ssh", append(args, command)...) - cmd.Stdin, cmd.Stdout, cmd.Stderr = streams.Stdin, streams.Stdout, streams.Stderr - if err := cmd.Run(); err != nil { + 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", err) } return nil } +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, @@ -170,23 +198,10 @@ func runSession(session *cryptossh.Session, command string, input io.Reader) err if err := session.Start(command); err != nil { return err } - copied := make(chan struct{}) - copyResult := make(chan error, 1) - go func() { - defer close(copied) - _, copyErr := io.Copy(stdin, reader) - copyResult <- copyErr - _ = stdin.Close() - }() + copy := startStdinCopy(reader, stdin) err = session.Wait() - var copyErr error - select { - case copyErr = <-copyResult: - default: - } - reader.Cancel() - <-copied - return errors.Join(err, copyErr) + _ = stdin.Close() + return errors.Join(err, copy.finish()) } func cancellableInput(input io.Reader) (cancelreader.CancelReader, error) { @@ -204,8 +219,15 @@ func cancellableInput(input io.Reader) (cancelreader.CancelReader, error) { } } -type closingInput struct{ io.ReadCloser } +type closingInput struct { + io.ReadCloser + once sync.Once + closeErr error +} -func (input *closingInput) Cancel() bool { return input.ReadCloser.Close() == nil } +func (input *closingInput) Cancel() bool { return input.Close() == nil } -func (input *closingInput) Close() error { return input.ReadCloser.Close() } +func (input *closingInput) Close() error { + input.once.Do(func() { input.closeErr = input.ReadCloser.Close() }) + return input.closeErr +} diff --git a/pkg/ssh/stdin_copy.go b/pkg/ssh/stdin_copy.go new file mode 100644 index 0000000..db32b04 --- /dev/null +++ b/pkg/ssh/stdin_copy.go @@ -0,0 +1,82 @@ +package ssh + +import ( + "errors" + "io" + "net" + "os" + "sync/atomic" + + "github.com/muesli/cancelreader" +) + +type stdinCopy struct { + reader cancelreader.CancelReader + stopping atomic.Bool + done chan struct{} + sourceErr error + sourceStopped bool + copyErr error + destinationErr error +} + +func startStdinCopy(reader cancelreader.CancelReader, 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.Cancel() + <-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, cancelreader.ErrCanceled) || 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..6ea008d --- /dev/null +++ b/pkg/ssh/stdin_lifecycle_test.go @@ -0,0 +1,247 @@ +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" + +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: + } +} From cf24b4cca842bbae5c0fbbec94dbca01a0f1b084 Mon Sep 17 00:00:00 2001 From: Samuel K Date: Tue, 20 Oct 2026 15:19:33 -0600 Subject: [PATCH 04/10] ci: serialize provider snapshot builds to bound disk usage --- .github/workflows/tests.yaml | 15 +++++++++++++-- 1 file changed, 13 insertions(+), 2 deletions(-) diff --git a/.github/workflows/tests.yaml b/.github/workflows/tests.yaml index 8bbfc6e..838b39f 100644 --- a/.github/workflows/tests.yaml +++ b/.github/workflows/tests.yaml @@ -24,11 +24,22 @@ jobs: 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: | From e9c4c75e81853bcb11cf707d4aff9a49bb739411 Mon Sep 17 00:00:00 2001 From: Samuel K Date: Tue, 20 Oct 2026 16:09:24 -0600 Subject: [PATCH 05/10] fix: leave caller stdin untouched during SSH initialization --- cmd/init.go | 2 +- pkg/ssh/init_stdin_test.go | 84 ++++++++++++++++++++++++++++++++++++++ pkg/ssh/remote_command.go | 3 +- pkg/ssh/ssh.go | 2 +- 4 files changed, 88 insertions(+), 3 deletions(-) create mode 100644 pkg/ssh/init_stdin_test.go diff --git a/cmd/init.go b/cmd/init.go index 5a14b35..fc8b770 100644 --- a/cmd/init.go +++ b/cmd/init.go @@ -41,6 +41,6 @@ func (cmd *InitCmd) Run( return ssh.Init( ctx, providerSSH, - ssh.Streams{Stdin: os.Stdin, Stdout: os.Stdout, Stderr: os.Stderr}, + ssh.Streams{Stdout: os.Stdout, Stderr: os.Stderr}, ) } 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/remote_command.go b/pkg/ssh/remote_command.go index a642f20..c7527fe 100644 --- a/pkg/ssh/remote_command.go +++ b/pkg/ssh/remote_command.go @@ -20,7 +20,8 @@ import ( ) // Streams keeps protocol bytes separate from diagnostics. Non-file Stdin closers -// are closed when either transport ends; other readers must be finite buffers. +// 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. type Streams struct { Stdin io.Reader Stdout io.Writer diff --git a/pkg/ssh/ssh.go b/pkg/ssh/ssh.go index 00df6ad..78c5013 100644 --- a/pkg/ssh/ssh.go +++ b/pkg/ssh/ssh.go @@ -62,7 +62,7 @@ func Init(ctx context.Context, provider *SSHProvider, streams Streams) error { return provider.execRemoteScript( ctx, script, - Streams{Stdin: streams.Stdin, Stdout: output, Stderr: streams.Stderr}, + Streams{Stdout: output, Stderr: streams.Stderr}, ) } if err := checkRemotePlatform(run); err != nil { From b718c873f55c586e265842dd29c380fd6c0e5166 Mon Sep 17 00:00:00 2001 From: Samuel K Date: Tue, 20 Oct 2026 17:06:42 -0600 Subject: [PATCH 06/10] fix: preserve SSH cancellation identity and CLI classification --- cmd/root.go | 25 +++++---- cmd/root_unix_test.go | 95 +++++++++++++++++++++++++++++++++ pkg/ssh/remote_command.go | 2 +- pkg/ssh/remote_command_test.go | 36 +++++++++---- pkg/ssh/stdin_lifecycle_test.go | 23 ++++++++ 5 files changed, 160 insertions(+), 21 deletions(-) create mode 100644 cmd/root_unix_test.go diff --git a/cmd/root.go b/cmd/root.go index df3795f..2142442 100644 --- a/cmd/root.go +++ b/cmd/root.go @@ -45,19 +45,22 @@ func Execute() { ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) err := rootCmd.ExecuteContext(ctx) stop() - if err != nil { - 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()) - } - + 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 the SSH provider command tree. 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/pkg/ssh/remote_command.go b/pkg/ssh/remote_command.go index c7527fe..548fc45 100644 --- a/pkg/ssh/remote_command.go +++ b/pkg/ssh/remote_command.go @@ -64,7 +64,7 @@ func (provider *SSHProvider) execRemoteScript( 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", err) + return fmt.Errorf("execute remote /bin/sh script: %w", errors.Join(ctx.Err(), err)) } return nil } diff --git a/pkg/ssh/remote_command_test.go b/pkg/ssh/remote_command_test.go index 795c8bb..1291d33 100644 --- a/pkg/ssh/remote_command_test.go +++ b/pkg/ssh/remote_command_test.go @@ -103,15 +103,33 @@ func TestExternalStreamsAndExit(t *testing.T) { } func TestExternalCancellation(t *testing.T) { - provider := fakeExternalSSH(t) - ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) - defer cancel() - start := time.Now() - if err := Command(ctx, provider, "exec sleep 30", Streams{}); err == nil { - t.Fatal("expected cancellation") - } - if time.Since(start) > 5*time.Second { - t.Fatal("cancellation did not terminate SSH") + 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") + } + }) } } diff --git a/pkg/ssh/stdin_lifecycle_test.go b/pkg/ssh/stdin_lifecycle_test.go index 6ea008d..3c9b95e 100644 --- a/pkg/ssh/stdin_lifecycle_test.go +++ b/pkg/ssh/stdin_lifecycle_test.go @@ -19,6 +19,29 @@ import ( 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{} From 64378148397498132ffce0befc73dbfbd1c55050 Mon Sep 17 00:00:00 2001 From: Samuel K Date: Tue, 20 Oct 2026 18:10:16 -0600 Subject: [PATCH 07/10] fix: preserve cancellation through all built-in SSH phases --- docs/verification/1621-ssh-bootstrap.md | 13 +- pkg/ssh/builtin_context_test.go | 192 ++++++++++++++++++++++++ pkg/ssh/remote_command.go | 31 +++- 3 files changed, 227 insertions(+), 9 deletions(-) create mode 100644 pkg/ssh/builtin_context_test.go diff --git a/docs/verification/1621-ssh-bootstrap.md b/docs/verification/1621-ssh-bootstrap.md index 63930ea..9190508 100644 --- a/docs/verification/1621-ssh-bootstrap.md +++ b/docs/verification/1621-ssh-bootstrap.md @@ -155,6 +155,13 @@ rejected the review invocation because it uploads uncommitted and untracked private source. That gate is BLOCKED pending explicit user authorization for the concrete diff. No workaround or indirect upload was attempted. -CI and final-SHA Greptile/CodeRabbit reviews are pending draft PR publication and -must not be reported as passed. No merge, tag, release, or customer reply is part -of this change. +[Provider draft PR #152](https://github.com/devsy-org/devsy-provider-ssh/pull/152) +is published alongside [core draft PR #1627](https://github.com/devsy-org/devsy/pull/1627). +The qualification results above describe their recorded source/artifact revisions, +not every later follow-up. CI and Greptile passed on provider revision `b718c873`; +its full CodeRabbit review identified a pre-session cancellation error-identity +gap. The follow-up preserves context and original causes across configuration, +dial, handshake, session opening and execution, including the lookup's own timeout. +Fresh final-head CI and ordered Greptile/CodeRabbit coverage remain required after +that source change. Consult the live PR head, check runs, and review reports for +current gate evidence. No merge, tag, release, or customer reply is authorized. diff --git a/pkg/ssh/builtin_context_test.go b/pkg/ssh/builtin_context_test.go new file mode 100644 index 0000000..635fad7 --- /dev/null +++ b/pkg/ssh/builtin_context_test.go @@ -0,0 +1,192 @@ +package ssh + +import ( + "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 { + close(ready) + _, _ = io.Copy(io.Discard, conn) + 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/remote_command.go b/pkg/ssh/remote_command.go index 548fc45..859ca47 100644 --- a/pkg/ssh/remote_command.go +++ b/pkg/ssh/remote_command.go @@ -64,11 +64,21 @@ func (provider *SSHProvider) execRemoteScript( 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", errors.Join(ctx.Err(), err)) + 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() @@ -111,7 +121,7 @@ func (provider *SSHProvider) builtinConfig( lookup.Stderr = stderr config, err := lookup.Output() if err != nil { - return nil, "", fmt.Errorf("read SSH config: %w", err) + return nil, "", fmt.Errorf("read SSH config: %w", joinContextError(lookupCtx, err)) } resolved := parseConfig(string(config)) if resolved.hostname == "" || resolved.user == "" || resolved.port == "" || @@ -151,11 +161,23 @@ func (provider *SSHProvider) execBuiltin( 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 := (&net.Dialer{}).DialContext(ctx, "tcp", addr) + conn, err := dial(ctx, "tcp", addr) if err != nil { return fmt.Errorf("dial SSH: %w", err) } @@ -175,9 +197,6 @@ func (provider *SSHProvider) execBuiltin( defer func() { _ = session.Close() }() session.Stdout, session.Stderr = streams.Stdout, streams.Stderr if err := runSession(session, command, streams.Stdin); err != nil { - if ctx.Err() != nil { - return fmt.Errorf("execute remote /bin/sh script: %w", ctx.Err()) - } return fmt.Errorf("execute remote /bin/sh script: %w", err) } return nil From df18ed496f3acd1b3e9dfaf4fdfb42eb43c15f8a Mon Sep 17 00:00:00 2001 From: Samuel K Date: Tue, 20 Oct 2026 21:02:57 -0600 Subject: [PATCH 08/10] fix: remove cancelreader use and verification document --- docs/verification/1621-ssh-bootstrap.md | 167 ------------------------ go.mod | 2 +- pkg/ssh/builtin_context_test.go | 7 +- pkg/ssh/input_console_windows.go | 94 +++++++++++++ pkg/ssh/input_console_windows_test.go | 87 ++++++++++++ pkg/ssh/input_unix.go | 122 ++++++++++++++++- pkg/ssh/input_unix_test.go | 131 +++++++++++++++++++ pkg/ssh/input_windows.go | 25 ++-- pkg/ssh/input_windows_test.go | 2 +- pkg/ssh/remote_command.go | 8 +- pkg/ssh/stdin_copy.go | 10 +- 11 files changed, 458 insertions(+), 197 deletions(-) delete mode 100644 docs/verification/1621-ssh-bootstrap.md create mode 100644 pkg/ssh/input_console_windows.go create mode 100644 pkg/ssh/input_console_windows_test.go create mode 100644 pkg/ssh/input_unix_test.go diff --git a/docs/verification/1621-ssh-bootstrap.md b/docs/verification/1621-ssh-bootstrap.md deleted file mode 100644 index 9190508..0000000 --- a/docs/verification/1621-ssh-bootstrap.md +++ /dev/null @@ -1,167 +0,0 @@ -# SSH bootstrap regression verification - -The provider builds one login-shell-safe `/bin/sh -c` launcher before transport -selection. Its original multiline script is one argument; stdin remains reserved -for the bootstrap handshake, binary injection, and the live tunnel. No SCP or -stderr-matching fallback remains. - -## Revisions and environment - -- Reported provider v1.1.44: `e37a6c68a695d9d1d3806b2d42a755750816f60e`. -- Initial provider main: `9c5c3ff673c364818d31627d45f365053100fa4b`. -- Final base refreshed to `839369fc438c4616bace5ee15994ab42b0fd0773`; upstream - changed only the shfmt hook version. The isolated branch was rebased cleanly. -- Original core inspected by the coordinated core change: - `1f47cd185` (v1.24.0). -- Linux qualification: Ubuntu aarch64 container with actual OpenSSH and separate - non-root passwd accounts using Bash, Zsh, Fish, Dash, and BusyBox Ash. -- Local unit tests execute installed Bash, Zsh, and Fish on macOS arm64. These - shell tests are distinct from the Linux SSH qualification. -- The provider's core module remains pinned to v1.23.2. No API upgrade was needed. - The existing full-workspace E2E download is pinned to core v1.24.0, with - `DEVSY_E2E_BINARY` selecting an explicitly supplied local core executable. - -## Reproduced baseline - -The original core bootstrap and original provider passed with Bash through both -transports. External Zsh failed with `command not found: sh -c`; external Fish -failed with unsupported assignment syntax. The built-in transport failed with -exit 127 and lost the useful remote stderr diagnostics. - -## Implemented regression checks - -- Exact launcher argument boundaries and script byte round trips through real - Bash, Zsh, and Fish, including empty input, newlines, apostrophes, backslashes, - dollar signs, backticks, parentheses, UTF-8, and substitution sentinels. -- NUL rejection before remote execution. -- External and built-in transport byte-preserving stdin/stdout, separate stderr, - wrapped exit-status errors, and cancellation. -- Built-in cancellation with an idle file pipe and an idle `io.PipeReader`, with - input-copy completion joined before returning. Non-file closable input belongs - to the session; unsupported potentially unbounded readers are rejected. -- Local stdin read failures remain errors even when the remote consumer exits - successfully after EOF. Independent review identified and verified this fix. -- Windows uses the actual redirected pipe handle, duplicated for session-owned - cancellation rather than opening a console. Native Windows CI covers redirected - process stdin plus an in-process SSH session's pipe bytes and idle cancellation. - Native Windows execution is NOT RUN locally; the final Windows suite - cross-compiled successfully and Windows-target lint reported zero issues. -- Literal agent directory and Docker executable paths containing shell syntax. -- Linux CI provisions Bash/Zsh/Fish passwd accounts and exercises all three with - both transports using binary streams and an exit-37 command. Missing required - shells fail unit tests in CI. - -## Coordinated Linux observations - -The initial corrected provider and corrected core bootstrap passed all 30 cells: -five passwd shells, both transports, and cold/warm/forced-mismatch injection. -Each cell preserved a 4 KiB all-byte payload and a follow-on exchange sent only -after receiving output. Warm runs requested no replacement binary; cold and -forced-mismatch runs requested exactly one. This uses a synthetic agent payload -and must not be confused with the real-agent check below. - -The subsequent `final5` build passed all 30 randomized/hostile-path synthetic -cells and all 10 real-agent shell/transport combinations. The latter started the -actual 207 MB Linux Devsy agent, received its `SSH-2.0-Go` banner, and completed -two sequential live commands per combination. SIGTERM ended both provider -transports in under half a second, left no local SSH child, and left no remote -Devsy agent. Generic non-PTY remote `sleep` commands survive connection closure; -arbitrary remote process reaping is outside this SSH transport contract. - -The final5 binary SHA256 was -`7e18087de9fbc9137a7a561c6fce200b7fe50d523fd99db08b6e8ce7b27015c7`. -Later stdin-source-error and Windows fixes were independently reviewed. The -reviewed final-source binary SHA256 is -`39468e46550ddf4a9ea3a1033cae10ef02d0feabcce42ad7b89a7f88f40ba989`. -Its production diff fingerprint (commands, SSH implementation, module files) is -`fde9ee3eceaba52ab03246f0d59151c7070c967168edbce6b63a04ec1ddef1a0`. -The final rerun passed: - -- Fixed core + reviewed provider: all 30 synthetic cold/warm/mismatch cells. -- Original core v1.24.0 + reviewed provider: Bash/Zsh/Fish through both transports, - all six cold injection and live follow-on cells. -- Actual Linux Devsy agent: all ten shell/transport combinations, two live - commands per session. -- SIGTERM after the actual agent banner: external 0.459 s, built-in 0.405 s; - no local SSH child and no remote Devsy agent remained. -- Initialization through all five shells and both transports: ten passes. -- Exact stdout, separate stderr, and exit 37 through all five shells and both - transports: ten passes. - -The artifact was built before the final base's hook-only update; production -source is unchanged. All observations are Linux aarch64 SSH qualification, -not Windows or GUI acceptance. - -## Local validation - -- CI-parity unit suite: `go test -race -covermode=atomic . ./cmd/... ./hack/... ./pkg/...` PASS. -- SSH package statement coverage after the stdin review fixes: 81.4%. -- E2E package compilation without host setup: PASS. -- Diff-scoped golangci-lint with macOS and Windows build contexts: zero issues. -- `prek run --all-files`: all applicable hooks PASS on the refreshed base. -- Local GoReleaser snapshot with publishing disabled: all five configured target - builds PASS. This is cross-compilation, not native Windows execution. -- Provider YAML generation with genuine snapshot checksums: PASS. The existing - `${SSH_PROVIDER} init` and `${SSH_PROVIDER} command` interface is unchanged. - No generated descriptor was added to the source tree. - -## First CI run and test correction - -On provider head `c0c159aaed22ea789515380406e655b945006f29`, native Windows pipe -checks, lint, pre-commit, unit tests, snapshot builds, and Linux shell-account -provisioning passed. Linux E2E reported 14 passes, one failure, and three skips: -the existing failing-command assertion required Bash's English phrase -`command not found`, while Dash correctly emitted -`/bin/sh: 1: not-a-command: not found`. The follow-up changes only that assertion -to require exit 127, empty stdout, and separate nonempty stderr containing the -command name. The three subsequent ordered whole-workspace CLI checks were -skipped on that first run and must not be reported as acceptance passes. - -The production source fingerprint and reviewed binary remained unchanged for -that test-only correction. The subsequent stdin review fixes below change -production code and require new final-head CI and reviewer coverage. - -## Greptile stdin lifecycle follow-up - -Greptile reviewed `4722dd2c489ed2b440dfcacb6585ea6e2e97a1c1` at confidence 4/5 -and identified two valid issues. External SSH could hang after cancellation or -remote exit with an idle non-file stdin reader. Built-in SSH could discard a -genuine source error that arrived after remote completion. - -Both were reproduced before the fix: a bounded idle-input test timed out and a -late custom read error incorrectly returned success. Both transports now join -owned non-file stdin copies before reading their final results. External file -stdin remains borrowed directly. Source read errors are recorded separately -from destination writes, preserving genuine source failures (including EPIPE) -while allowing a successful remote command to close unread input. Owned -non-file closers are closed once, including start failure. - -Regressions cover idle-input exit zero, cancellation, late custom source errors, -source EPIPE, unread 1 MiB input with remote success, and ownership cleanup. -Native Windows CI also exercises the external-child variants without a Unix -shell stub. Independent bounded review passed the fixes, including five runs -of the unread-large-input reproducer. Local race tests pass; new final-head -CI and fresh Greptile/CodeRabbit reviews remain required. - -The earlier runtime artifact fingerprints and matrices are explicitly evidence -for the earlier production revision. The launcher encoding is unchanged by -this stdin follow-up; they are not claimed as runtime execution of the new -stdin implementation. - -## Review gates - -Local CodeRabbit authentication was verified, but automatic approval review -rejected the review invocation because it uploads uncommitted and untracked -private source. That gate is BLOCKED pending explicit user authorization for the -concrete diff. No workaround or indirect upload was attempted. - -[Provider draft PR #152](https://github.com/devsy-org/devsy-provider-ssh/pull/152) -is published alongside [core draft PR #1627](https://github.com/devsy-org/devsy/pull/1627). -The qualification results above describe their recorded source/artifact revisions, -not every later follow-up. CI and Greptile passed on provider revision `b718c873`; -its full CodeRabbit review identified a pre-session cancellation error-identity -gap. The follow-up preserves context and original causes across configuration, -dial, handshake, session opening and execution, including the lookup's own timeout. -Fresh final-head CI and ordered Greptile/CodeRabbit coverage remain required after -that source change. Consult the live PR head, check runs, and review reports for -current gate evidence. No merge, tag, release, or customer reply is authorized. diff --git a/go.mod b/go.mod index 95b5e2b..8f02ce6 100644 --- a/go.mod +++ b/go.mod @@ -6,7 +6,6 @@ require ( github.com/devsy-org/devsy v1.23.2 github.com/goccy/go-yaml v1.19.2 github.com/kballard/go-shellquote v0.0.0-20180428030007-95032a82bc51 - github.com/muesli/cancelreader v0.2.2 github.com/onsi/ginkgo/v2 v2.33.1 github.com/onsi/gomega v1.44.0 github.com/spf13/cobra v1.10.2 @@ -242,6 +241,7 @@ require ( github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect github.com/modern-go/reflect2 v1.0.3-0.20250322232337-35a7c28c31ee // indirect github.com/morikuni/aec v1.1.0 // indirect + github.com/muesli/cancelreader v0.2.2 // indirect github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect github.com/oklog/run v1.1.0 // indirect github.com/opencontainers/go-digest v1.0.0 // indirect diff --git a/pkg/ssh/builtin_context_test.go b/pkg/ssh/builtin_context_test.go index 635fad7..d0d3bd9 100644 --- a/pkg/ssh/builtin_context_test.go +++ b/pkg/ssh/builtin_context_test.go @@ -1,6 +1,7 @@ package ssh import ( + "bufio" "context" "crypto/ed25519" "crypto/rand" @@ -127,8 +128,12 @@ func pendingBuiltinPhase(t *testing.T, session bool) (*SSHProvider, <-chan struc } defer func() { _ = conn.Close() }() if !session { + reader := bufio.NewReader(conn) + if _, err := reader.ReadString('\n'); err != nil { + return + } close(ready) - _, _ = io.Copy(io.Discard, conn) + _, _ = io.Copy(io.Discard, reader) return } server, channels, requests, err := cryptossh.NewServerConn(conn, config) 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..4ee4115 --- /dev/null +++ b/pkg/ssh/input_console_windows_test.go @@ -0,0 +1,87 @@ +package ssh + +import ( + "context" + "errors" + "os" + "os/exec" + "syscall" + "testing" + "time" + + "golang.org/x/sys/windows" +) + +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 error, 1) + go func() { _, err := input.Read(make([]byte, 8)); result <- 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) + } + requireNoError(t, input.Close()) + select { + case err := <-result: + if !errors.Is(err, os.ErrClosed) { + t.Fatalf("console cancellation: %v", err) + } + case <-time.After(3 * time.Second): + t.Fatal("console read did not join") + } + requireNoError(t, input.Close()) +} diff --git a/pkg/ssh/input_unix.go b/pkg/ssh/input_unix.go index 875df5d..eeb99c6 100644 --- a/pkg/ssh/input_unix.go +++ b/pkg/ssh/input_unix.go @@ -4,23 +4,131 @@ package ssh import ( "errors" + "fmt" "io" + "math" "os" + "runtime" + "sync" + "sync/atomic" "syscall" - "github.com/muesli/cancelreader" + "golang.org/x/sys/unix" ) -func cancellableFile(reader *os.File) (cancelreader.CancelReader, error) { - info, err := reader.Stat() +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 } - nullInfo, _ := os.Stat(os.DevNull) - if info.Mode().IsRegular() || (nullInfo != nil && os.SameFile(info, nullInfo)) { - return cancelreader.NewReader(struct{ io.Reader }{reader}) + 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 } - return cancelreader.NewReader(reader) } 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 index 857a790..343a345 100644 --- a/pkg/ssh/input_windows.go +++ b/pkg/ssh/input_windows.go @@ -6,29 +6,36 @@ import ( "io" "os" - "github.com/muesli/cancelreader" "golang.org/x/sys/windows" ) -func cancellableFile(file *os.File) (cancelreader.CancelReader, error) { +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 cancelreader.NewReader(struct{ io.Reader }{file}) + return io.NopCloser(file), nil } if kind != windows.FILE_TYPE_PIPE { var mode uint32 if windows.GetConsoleMode(windows.Handle(file.Fd()), &mode) != nil { - return cancelreader.NewReader(struct{ io.Reader }{file}) + return io.NopCloser(file), nil } - return cancelreader.NewReader(file) + 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( + err := windows.DuplicateHandle( process, windows.Handle(file.Fd()), process, @@ -38,14 +45,14 @@ func cancellableFile(file *os.File) (cancelreader.CancelReader, error) { windows.DUPLICATE_SAME_ACCESS, ) if err != nil { - return nil, fmt.Errorf("duplicate SSH input pipe: %w", err) + 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 pipe") + return nil, fmt.Errorf("open duplicated SSH input handle") } - return &closingInput{ReadCloser: owned}, nil + return owned, nil } func isPipeWriteClosed(err error) bool { diff --git a/pkg/ssh/input_windows_test.go b/pkg/ssh/input_windows_test.go index e182973..a2569ff 100644 --- a/pkg/ssh/input_windows_test.go +++ b/pkg/ssh/input_windows_test.go @@ -25,7 +25,7 @@ func TestWindowsRedirectedStdin(t *testing.T) { requireNoError(t, err) _, err = os.Stdout.Write(data) requireNoError(t, err) - timer := time.AfterFunc(100*time.Millisecond, func() { reader.Cancel() }) + timer := time.AfterFunc(100*time.Millisecond, func() { _ = reader.Close() }) defer timer.Stop() _, err = reader.Read(make([]byte, 1)) if err == nil { diff --git a/pkg/ssh/remote_command.go b/pkg/ssh/remote_command.go index 859ca47..db5cc3e 100644 --- a/pkg/ssh/remote_command.go +++ b/pkg/ssh/remote_command.go @@ -15,13 +15,13 @@ import ( "time" devsyssh "github.com/devsy-org/devsy/pkg/ssh" - "github.com/muesli/cancelreader" 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 @@ -224,12 +224,12 @@ func runSession(session *cryptossh.Session, command string, input io.Reader) err return errors.Join(err, copy.finish()) } -func cancellableInput(input io.Reader) (cancelreader.CancelReader, error) { +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 cancelreader.NewReader(input) + return io.NopCloser(input), nil case io.ReadCloser: return &closingInput{ReadCloser: reader}, nil default: @@ -245,8 +245,6 @@ type closingInput struct { closeErr error } -func (input *closingInput) Cancel() bool { return input.Close() == nil } - func (input *closingInput) Close() error { input.once.Do(func() { input.closeErr = input.ReadCloser.Close() }) return input.closeErr diff --git a/pkg/ssh/stdin_copy.go b/pkg/ssh/stdin_copy.go index db32b04..30b17ff 100644 --- a/pkg/ssh/stdin_copy.go +++ b/pkg/ssh/stdin_copy.go @@ -6,12 +6,10 @@ import ( "net" "os" "sync/atomic" - - "github.com/muesli/cancelreader" ) type stdinCopy struct { - reader cancelreader.CancelReader + reader io.ReadCloser stopping atomic.Bool done chan struct{} sourceErr error @@ -20,7 +18,7 @@ type stdinCopy struct { destinationErr error } -func startStdinCopy(reader cancelreader.CancelReader, writer io.WriteCloser) *stdinCopy { +func startStdinCopy(reader io.ReadCloser, writer io.WriteCloser) *stdinCopy { copy := &stdinCopy{reader: reader, done: make(chan struct{})} go func() { defer close(copy.done) @@ -46,7 +44,7 @@ func (input trackedStdin) Read(data []byte) (int, error) { func (copy *stdinCopy) finish() error { copy.stopping.Store(true) - copy.reader.Cancel() + _ = 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)) { @@ -62,7 +60,7 @@ func (copy *stdinCopy) finish() error { } func isStdinShutdown(err error) bool { - return errors.Is(err, cancelreader.ErrCanceled) || errors.Is(err, io.ErrClosedPipe) || + return errors.Is(err, io.ErrClosedPipe) || errors.Is(err, os.ErrClosed) || errors.Is(err, net.ErrClosed) || errors.Is(err, io.EOF) } From c9f0c669db947ba57bf972c231b35b87b84e9c5a Mon Sep 17 00:00:00 2001 From: Samuel K Date: Tue, 20 Oct 2026 21:15:18 -0600 Subject: [PATCH 09/10] test: accept EOF from canceled Windows console reads --- pkg/ssh/input_console_windows_test.go | 50 +++++++++++++++++++++++---- 1 file changed, 43 insertions(+), 7 deletions(-) diff --git a/pkg/ssh/input_console_windows_test.go b/pkg/ssh/input_console_windows_test.go index 4ee4115..e2351f3 100644 --- a/pkg/ssh/input_console_windows_test.go +++ b/pkg/ssh/input_console_windows_test.go @@ -3,6 +3,7 @@ package ssh import ( "context" "errors" + "io" "os" "os/exec" "syscall" @@ -12,6 +13,11 @@ import ( "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) @@ -59,8 +65,11 @@ func cancelConsoleRead(t *testing.T, file *os.File) { reader, err := newConsoleInput(file) requireNoError(t, err) input := reader.(*consoleInput) - result := make(chan error, 1) - go func() { _, err := input.Read(make([]byte, 8)); result <- err }() + 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() @@ -74,14 +83,41 @@ func cancelConsoleRead(t *testing.T, file *os.File) { } time.Sleep(time.Millisecond) } - requireNoError(t, input.Close()) select { - case err := <-result: - if !errors.Is(err, os.ErrClosed) { - t.Fatalf("console cancellation: %v", err) - } + 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) + } +} From a14daf32a27e4efa253a913bc067f35261fd9a3c Mon Sep 17 00:00:00 2001 From: Samuel K Date: Tue, 20 Oct 2026 21:56:31 -0600 Subject: [PATCH 10/10] ci: avoid persisting checkout credentials during tests --- .github/workflows/tests.yaml | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/.github/workflows/tests.yaml b/.github/workflows/tests.yaml index 838b39f..d596ff6 100644 --- a/.github/workflows/tests.yaml +++ b/.github/workflows/tests.yaml @@ -10,6 +10,8 @@ jobs: runs-on: ubuntu-latest steps: - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7 + with: + persist-credentials: false - name: setup Go uses: actions/setup-go@b7ad1dad31e06c5925ef5d2fc7ad053ef454303e # v7 @@ -59,6 +61,8 @@ jobs: 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