From 0efb45a1cfcdb3118c01683a48dfe974a4b8c0ee Mon Sep 17 00:00:00 2001 From: Samuel K Date: Sat, 1 Aug 2026 23:49:02 -0500 Subject: [PATCH 01/14] fix(ssh): make GPG-agent forwarding survive terminal disconnects The desktop app's SSH terminal button silently failed when GPG-agent forwarding hit a container whose SSH server/socket wasn't ready yet, and had no way to recover if the terminal holding the forward disconnected while others depended on it. - Make GPG forwarding failures non-fatal to the SSH session, matching the existing pattern for SSH agent forwarding. - Run GPG tunnel setup as a background health-check loop for the life of the session, so a surviving terminal notices and re-establishes a forward whose owner disconnected, instead of requiring a reconnect. - Surface non-fatal forwarding failures to the desktop UI via an OSC 9977 sequence parsed by xterm.js, so the failure isn't silently swallowed. - Make the in-container credentials-server retry on port contention instead of giving up permanently, so GPG public-key/git/docker credential requests recover once the owning session disconnects. - Harden claimForwardedSocket's readiness wait (exponential backoff, socket-type check) and serialize concurrent setup-gpg invocations with a cross-process lock. - Fix a pre-existing forward-ports timeout that was silently discarded due to variable shadowing, and a related channel deadlock when a port forward exited cleanly without reporting a result. - Extract cmd/workspace/ssh.go's GPG-tunnel and port-forwarding logic into dedicated files, replacing ad hoc bools on the flag-binding SSHCmd struct with an owned gpgTunnel type. --- .../agentcontainer/credentials_server.go | 22 +- .../agentcontainer/credentials_server_test.go | 49 +++ cmd/internal/agentworkspace/setup_gpg.go | 49 ++- cmd/internal/agentworkspace/setup_gpg_test.go | 93 +++++ cmd/workspace/gpg_tunnel.go | 245 +++++++++++ cmd/workspace/gpg_tunnel_test.go | 65 +++ cmd/workspace/port_forward.go | 240 +++++++++++ cmd/workspace/ssh.go | 390 ++---------------- .../lib/components/terminal/Terminal.svelte | 21 +- .../src/lib/stores/terminal-instances.ts | 3 + .../src/pages/WorkspaceDetailPage.svelte | 12 +- pkg/gpg/gpg_forwarding.go | 52 ++- pkg/gpg/gpg_forwarding_test.go | 101 +++++ pkg/tunnel/container.go | 19 +- 14 files changed, 980 insertions(+), 381 deletions(-) create mode 100644 cmd/internal/agentcontainer/credentials_server_test.go create mode 100644 cmd/internal/agentworkspace/setup_gpg_test.go create mode 100644 cmd/workspace/gpg_tunnel.go create mode 100644 cmd/workspace/gpg_tunnel_test.go create mode 100644 cmd/workspace/port_forward.go diff --git a/cmd/internal/agentcontainer/credentials_server.go b/cmd/internal/agentcontainer/credentials_server.go index e00b76230..4cc6dc79e 100644 --- a/cmd/internal/agentcontainer/credentials_server.go +++ b/cmd/internal/agentcontainer/credentials_server.go @@ -102,10 +102,8 @@ func (cmd *CredentialsServerCmd) Run(ctx context.Context, port int) error { cmd.maybeForwardPorts(ctx, tunnelClient) - addr := net.JoinHostPort("localhost", strconv.Itoa(port)) - if ok, err := portpkg.IsAvailable(addr); !ok || err != nil { - log.Debugf("Port %d not available, exiting", port) - return nil + if err := checkPortClaimable(port); err != nil { + return err } // configure docker credential helper @@ -135,6 +133,22 @@ func (cmd *CredentialsServerCmd) Run(ctx context.Context, port int) error { return credentials.RunCredentialsServer(ctx, port, tunnelClient) } +// checkPortClaimable reports an error if port is not free to bind. Only one +// session's credentials-server can hold this port at a time. Returning an +// error (not nil) on contention matters: RunServices (pkg/tunnel/services.go) +// wraps this command in retry.OnError, which only retries on a non-nil error. +func checkPortClaimable(port int) error { + addr := net.JoinHostPort("localhost", strconv.Itoa(port)) + ok, err := portpkg.IsAvailable(addr) + if err != nil { + return fmt.Errorf("check port %d availability: %w", port, err) + } + if !ok { + return fmt.Errorf("port %d not available (another session likely owns the credentials server)", port) + } + return nil +} + func (cmd *CredentialsServerCmd) maybeForwardPorts( ctx context.Context, tunnelClient tunnel.TunnelClient, diff --git a/cmd/internal/agentcontainer/credentials_server_test.go b/cmd/internal/agentcontainer/credentials_server_test.go new file mode 100644 index 000000000..34c919ba1 --- /dev/null +++ b/cmd/internal/agentcontainer/credentials_server_test.go @@ -0,0 +1,49 @@ +package agentcontainer + +import ( + "net" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestCheckPortClaimable_SucceedsWhenPortFree(t *testing.T) { + // checkPortClaimable itself binds the port to check it, so pass 0 to let + // the OS assign a free ephemeral port at call time — this avoids the + // close-then-probe race of reserving a port via net.Listen, closing it, + // and hoping nothing else claims it before checkPortClaimable runs. + assert.NoError(t, checkPortClaimable(0)) +} + +func TestCheckPortClaimable_ErrorsWhenPortHeld(t *testing.T) { + ln, err := net.Listen("tcp", "localhost:0") + require.NoError(t, err) + t.Cleanup(func() { _ = ln.Close() }) + port := ln.Addr().(*net.TCPAddr).Port + + err = checkPortClaimable(port) + require.Error(t, err) + assert.Contains(t, err.Error(), "not available") +} + +func TestCheckPortClaimable_BecomesClaimableAfterHolderReleases(t *testing.T) { + ln, err := net.Listen("tcp", "localhost:0") + require.NoError(t, err) + closed := false + t.Cleanup(func() { + if !closed { + _ = ln.Close() + } + }) + port := ln.Addr().(*net.TCPAddr).Port + + require.Error(t, checkPortClaimable(port), + "port must read as unavailable while the listener is held") + + require.NoError(t, ln.Close()) + closed = true + + assert.NoError(t, checkPortClaimable(port), + "port must read as claimable once the prior holder releases it") +} diff --git a/cmd/internal/agentworkspace/setup_gpg.go b/cmd/internal/agentworkspace/setup_gpg.go index a8bb279c7..5e2cf288f 100644 --- a/cmd/internal/agentworkspace/setup_gpg.go +++ b/cmd/internal/agentworkspace/setup_gpg.go @@ -4,6 +4,7 @@ import ( "context" "encoding/base64" "fmt" + "time" "github.com/devsy-org/devsy/cmd/flags" "github.com/devsy-org/devsy/pkg/credentials" @@ -12,9 +13,18 @@ import ( "github.com/devsy-org/devsy/pkg/gitcredentials" "github.com/devsy-org/devsy/pkg/gpg" "github.com/devsy-org/devsy/pkg/log" + "github.com/gofrs/flock" "github.com/spf13/cobra" ) +// gpgSetupLockPath serializes concurrent setup-gpg invocations against the +// same container's gpg-agent/socket. A var so tests can point it at a temp path. +var gpgSetupLockPath = "/tmp/devsy-gpg-setup.lock" + +// gpgSetupLockTimeout bounds how long an invocation waits for a concurrent +// one to finish. A var so tests can shrink it. +var gpgSetupLockTimeout = 30 * time.Second + // SetupGPGCmd holds the setupGPG cmd flags. type SetupGPGCmd struct { *flags.GlobalFlags @@ -63,6 +73,12 @@ func NewSetupGPGCmd(flags *flags.GlobalFlags) *cobra.Command { func (cmd *SetupGPGCmd) Run(ctx context.Context) error { log.Debugf("Initializing gpg-agent forwarding") + unlock, err := acquireGPGSetupLock(ctx) + if err != nil { + return err + } + defer unlock() + publicKey, ownerTrust, err := fetchAndDecodeKeys(cmd.OwnerTrust) if err != nil { return err @@ -75,7 +91,7 @@ func (cmd *SetupGPGCmd) Run(ctx context.Context) error { GitKey: cmd.GitKey, } - if err := configureGPGAgent(&gpgConf); err != nil { + if err := configureGPGAgent(ctx, &gpgConf); err != nil { return err } @@ -89,6 +105,33 @@ func (cmd *SetupGPGCmd) Run(ctx context.Context) error { return nil } +// acquireGPGSetupLock takes the cross-process lock guarding setup-gpg. On +// success it returns a func that releases the lock. +func acquireGPGSetupLock(ctx context.Context) (func(), error) { + lockCtx, cancel := context.WithTimeout(ctx, gpgSetupLockTimeout) + defer cancel() + + lock := flock.New(gpgSetupLockPath) + locked, err := lock.TryLockContext(lockCtx, 200*time.Millisecond) + if err != nil { + if ctx.Err() != nil { + return nil, ctx.Err() + } + if lockCtx.Err() != nil { + return nil, fmt.Errorf("timed out waiting for another gpg setup to finish: %w", err) + } + return nil, fmt.Errorf("acquire gpg setup lock: %w", err) + } + if !locked { + if ctx.Err() != nil { + return nil, ctx.Err() + } + return nil, fmt.Errorf("timed out waiting for another gpg setup to finish") + } + + return func() { _ = lock.Unlock() }, nil +} + func fetchAndDecodeKeys(ownerTrustB64 string) ([]byte, []byte, error) { log.Debugf("Fetching public key") rawPublicKeys, err := getPublicKeys() @@ -111,7 +154,7 @@ func fetchAndDecodeKeys(ownerTrustB64 string) ([]byte, []byte, error) { return publicKey, ownerTrust, nil } -func configureGPGAgent(gpgConf *gpg.GPGConf) error { +func configureGPGAgent(ctx context.Context, gpgConf *gpg.GPGConf) error { log.Debugf("Stopping container gpg-agent") if err := gpg.StopGpgAgent(); err != nil { return fmt.Errorf("stop container gpg-agent: %w", err) @@ -140,7 +183,7 @@ func configureGPGAgent(gpgConf *gpg.GPGConf) error { } log.Debugf("Setup local gnupg socket links") - if err := gpgConf.SetupRemoteSocketLink(); err != nil { + if err := gpgConf.SetupRemoteSocketLink(ctx); err != nil { return fmt.Errorf("setup local gnupg socket links: %w", err) } diff --git a/cmd/internal/agentworkspace/setup_gpg_test.go b/cmd/internal/agentworkspace/setup_gpg_test.go new file mode 100644 index 000000000..879a16a26 --- /dev/null +++ b/cmd/internal/agentworkspace/setup_gpg_test.go @@ -0,0 +1,93 @@ +package agentworkspace + +import ( + "context" + "errors" + "path/filepath" + "testing" + "time" + + "github.com/gofrs/flock" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestAcquireGPGSetupLock_SucceedsWhenFree(t *testing.T) { + origPath, origTimeout := gpgSetupLockPath, gpgSetupLockTimeout + gpgSetupLockPath = filepath.Join(t.TempDir(), "setup-gpg.lock") + gpgSetupLockTimeout = time.Second + defer func() { gpgSetupLockPath, gpgSetupLockTimeout = origPath, origTimeout }() + + unlock, err := acquireGPGSetupLock(context.Background()) + require.NoError(t, err) + unlock() +} + +func TestAcquireGPGSetupLock_WaitsForConcurrentHolderThenSucceeds(t *testing.T) { + origPath, origTimeout := gpgSetupLockPath, gpgSetupLockTimeout + gpgSetupLockPath = filepath.Join(t.TempDir(), "setup-gpg.lock") + gpgSetupLockTimeout = 2 * time.Second + defer func() { gpgSetupLockPath, gpgSetupLockTimeout = origPath, origTimeout }() + + holder := flock.New(gpgSetupLockPath) + locked, err := holder.TryLock() + require.NoError(t, err) + require.True(t, locked) + + go func() { + time.Sleep(200 * time.Millisecond) + _ = holder.Unlock() + }() + + start := time.Now() + unlock, err := acquireGPGSetupLock(context.Background()) + elapsed := time.Since(start) + + require.NoError(t, err) + defer unlock() + assert.GreaterOrEqual(t, elapsed, 150*time.Millisecond, + "second acquirer must wait for the first to release, not run concurrently") +} + +func TestAcquireGPGSetupLock_TimesOutWhenHeldTooLong(t *testing.T) { + origPath, origTimeout := gpgSetupLockPath, gpgSetupLockTimeout + gpgSetupLockPath = filepath.Join(t.TempDir(), "setup-gpg.lock") + gpgSetupLockTimeout = 300 * time.Millisecond + defer func() { gpgSetupLockPath, gpgSetupLockTimeout = origPath, origTimeout }() + + holder := flock.New(gpgSetupLockPath) + locked, err := holder.TryLock() + require.NoError(t, err) + require.True(t, locked) + defer func() { _ = holder.Unlock() }() + + _, err = acquireGPGSetupLock(context.Background()) + require.Error(t, err) + assert.Contains(t, err.Error(), "timed out waiting") +} + +func TestAcquireGPGSetupLock_ReturnsCancellationErrorWhenCallerCancels(t *testing.T) { + origPath, origTimeout := gpgSetupLockPath, gpgSetupLockTimeout + gpgSetupLockPath = filepath.Join(t.TempDir(), "setup-gpg.lock") + // Longer than the caller's cancellation, so a "timed out waiting" + // lock-timeout error is not what triggers here — only ctx cancellation. + gpgSetupLockTimeout = 10 * time.Second + defer func() { gpgSetupLockPath, gpgSetupLockTimeout = origPath, origTimeout }() + + holder := flock.New(gpgSetupLockPath) + locked, err := holder.TryLock() + require.NoError(t, err) + require.True(t, locked) + defer func() { _ = holder.Unlock() }() + + ctx, cancel := context.WithCancel(context.Background()) + go func() { + time.Sleep(100 * time.Millisecond) + cancel() + }() + + _, err = acquireGPGSetupLock(ctx) + require.Error(t, err) + assert.True(t, errors.Is(err, context.Canceled), + "caller cancellation must surface as context.Canceled, not a generic lock-timeout error: got %v", err) +} diff --git a/cmd/workspace/gpg_tunnel.go b/cmd/workspace/gpg_tunnel.go new file mode 100644 index 000000000..46e894f9f --- /dev/null +++ b/cmd/workspace/gpg_tunnel.go @@ -0,0 +1,245 @@ +package workspace + +import ( + "context" + "encoding/base64" + "fmt" + "io" + "os" + "strings" + "time" + + "al.essio.dev/pkg/shellescape" + "github.com/devsy-org/devsy/pkg/config" + "github.com/devsy-org/devsy/pkg/flags/names" + "github.com/devsy-org/devsy/pkg/gpg" + "github.com/devsy-org/devsy/pkg/log" + devssh "github.com/devsy-org/devsy/pkg/ssh" + "golang.org/x/crypto/ssh" +) + +// gpgForwardFailedOSC is a private-use OSC identifier the desktop app's +// terminal (xterm.js) listens for to surface a non-fatal GPG-forwarding +// failure as a toast; see Terminal.svelte's registerOscHandler. +const gpgForwardFailedOSC = 9977 + +// gpgForwardFailedReasonMaxLen bounds the OSC payload, since the desktop +// toast renders reason verbatim and it can originate from a remote error. +const gpgForwardFailedReasonMaxLen = 256 + +func writeGPGForwardFailedOSC(w io.Writer, reason string) { + runes := []rune(reason) + if len(runes) > gpgForwardFailedReasonMaxLen { + runes = runes[:gpgForwardFailedReasonMaxLen] + } + clean := strings.Map(func(r rune) rune { + // Strip C0/C1 controls (including BEL, ESC, ST) and ';', the OSC + // parameter separator, so reason can't corrupt or extend the sequence. + if r < 0x20 || (r >= 0x7f && r <= 0x9f) || r == ';' { + return -1 + } + return r + }, string(runes)) + _, _ = fmt.Fprintf(w, "\x1b]%d;%s\a", gpgForwardFailedOSC, clean) +} + +// gpgTunnelHealthCheckInterval is how often gpgTunnel.run re-checks the GPG +// tunnel once the session is up, so a terminal whose forward died with +// another (owning) terminal's disconnect can re-establish it itself. +const gpgTunnelHealthCheckInterval = 30 * time.Second + +// gpgTunnel owns the lifecycle of GPG-agent forwarding for one SSH session: +// deciding whether it's requested, binding the reverse-listen socket at most +// once, running the remote setup-gpg step, and periodically checking the +// tunnel is still alive for as long as the session runs. +type gpgTunnel struct { + cmd *SSHCmd + enabled bool + + // forwardBound guards against re-binding the reverse-listen socket on a + // health-check retry: the listener stays open for the session, and the + // server rejects a second bind of the same path. + forwardBound bool + + // failureReported prevents a repeated OSC 9977 notification while the + // tunnel stays down across health-check ticks. + failureReported bool +} + +// newGPGTunnel reports whether GPG-agent forwarding was requested via flag +// or context option, and returns a tunnel that no-ops everywhere if not. +func newGPGTunnel(cmd *SSHCmd, devsyConfig *config.Config) *gpgTunnel { + return &gpgTunnel{ + cmd: cmd, + enabled: cmd.GPGAgentForwarding || + devsyConfig.ContextOptionBool(config.ContextOptionGPGAgentForwarding), + } +} + +// run watches the tunnel for the life of ctx, (re-)establishing it whenever +// it's found down. Call this in a goroutine tied to a context that's +// cancelled as soon as the owning SSH session ends (see +// runGPGTunnelInBackground). +func (t *gpgTunnel) run(ctx context.Context, sshClient *ssh.Client) { + if !t.enabled { + return + } + + t.ensure(ctx, sshClient) + + ticker := time.NewTicker(gpgTunnelHealthCheckInterval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + t.ensure(ctx, sshClient) + } + } +} + +// ensure checks whether the GPG tunnel is currently live and, if not, +// (re-)establishes it. A setup failure is only reported if the tunnel is +// still down immediately after: a concurrent terminal may have won the bind +// race in the interim, which isn't a real failure. The OSC failure +// notification fires only on the healthy-to-failed transition (gated by +// failureReported), not on every health-check tick. +func (t *gpgTunnel) ensure(ctx context.Context, sshClient *ssh.Client) { + if gpg.IsGpgTunnelRunning(ctx, t.cmd.User, sshClient) { + log.Debugf("GPG tunnel is running, skipping setup") + t.failureReported = false + return + } + err := t.setup(ctx, sshClient) + if err == nil { + t.failureReported = false + return + } + if gpg.IsGpgTunnelRunning(ctx, t.cmd.User, sshClient) { + log.Debugf("GPG tunnel setup failed but tunnel is live (won by a concurrent terminal): %v", err) + t.failureReported = false + return + } + if ctx.Err() != nil { + // ctx was cancelled (session ending); the failure above is an + // artifact of that, not a real forwarding problem worth reporting. + log.Debugf("GPG tunnel setup aborted by context cancellation: %v", err) + return + } + log.Warnf("GPG agent forwarding failed (continuing without it): %v", err) + if t.failureReported { + return + } + t.failureReported = true + writeGPGForwardFailedOSC(os.Stderr, err.Error()) +} + +// setup forwards the local gpg-agent into the remote container by using +// cmd/internal/agentworkspace/setup_gpg. +func (t *gpgTunnel) setup(ctx context.Context, containerClient *ssh.Client) error { + cmd := t.cmd + + log.Debugf("[GPG] exporting gpg owner trust from host") + ownerTrustExport, err := gpg.GetHostOwnerTrust() + if err != nil { + return fmt.Errorf("export local ownertrust from GPG: %w", err) + } + ownerTrustArgument := base64.StdEncoding.EncodeToString(ownerTrustExport) + + log.Debugf("detecting gpg-agent socket path on host") + // Detect local agent extra socket, this will be forwarded to the remote and + // symlinked in multiple paths + gpgExtraSocketPath, err := gpg.DetectAgentSocketPath() + if err != nil { + return err + } + log.Debugf("[GPG] detected gpg-agent socket path %s", gpgExtraSocketPath) + + gitKey := gpg.SigningKey(ctx) + + // Now we forward the agent socket to the remote, and setup remote gpg to use it + forwardAgent := []string{ + config.ContainerDevsyHelperLocation, + "internal", + "agent", + "workspace", + "setup-gpg", + names.Flag(names.OwnerTrust), + ownerTrustArgument, + names.Flag(names.SocketPath), + gpg.ContainerSocketPath, + } + + if log.DebugEnabled() { + forwardAgent = append(forwardAgent, names.Flag(names.Debug)) + } + + if gitKey != "" { + forwardAgent = append(forwardAgent, names.Flag(names.GitKey)) + forwardAgent = append(forwardAgent, gitKey) + } + + command := shellescape.QuoteCommand(forwardAgent) + if cmd.User != "" && cmd.User != "root" { + command = shellescape.QuoteCommand([]string{"su", "-c", command, cmd.User}) + } + + // Bind the reverse-listen socket at most once per process (see + // forwardBound); the remote setup-gpg step below still re-runs every + // time to repair remote-side agent state (stopped agent, stale keys). + if !t.forwardBound { + log.Debugf( + "[GPG] start reverse forward of gpg-agent socket %s, keeping connection open", + gpgExtraSocketPath, + ) + reverseForwardPorts := append( + []string{gpg.ContainerSocketPath + ":" + gpgExtraSocketPath}, + cmd.ReverseForwardPorts..., + ) + if err := cmd.startReverseForwardsAndWait(ctx, containerClient, reverseForwardPorts); err != nil { + return fmt.Errorf("start gpg-agent reverse forward: %w", err) + } + t.forwardBound = true + } + + writer, writerDone := log.PipeJSONStream() + defer func() { + _ = writer.Close() + <-writerDone + }() + err = devssh.Run(ctx, devssh.RunOptions{ + Client: containerClient, + Command: command, + Stdout: writer, + Stderr: writer, + }) + if err != nil { + return fmt.Errorf("run gpg agent setup command: %w", err) + } + + return nil +} + +// runGPGTunnelInBackground starts t.run in a goroutine tied to a context +// derived from ctx, and returns a wait func that cancels that context and +// blocks until the goroutine exits. Callers defer the wait func immediately +// after starting the tunnel, so a session that returns while the tunnel's +// health-check loop is mid-tick doesn't block on the session's own (often +// much longer-lived) ctx. +func runGPGTunnelInBackground( + ctx context.Context, + t *gpgTunnel, + sshClient *ssh.Client, +) (wait func()) { + tunnelCtx, cancel := context.WithCancel(ctx) + done := make(chan struct{}) + go func() { + defer close(done) + t.run(tunnelCtx, sshClient) + }() + return func() { + cancel() + <-done + } +} diff --git a/cmd/workspace/gpg_tunnel_test.go b/cmd/workspace/gpg_tunnel_test.go new file mode 100644 index 000000000..d1b94f64a --- /dev/null +++ b/cmd/workspace/gpg_tunnel_test.go @@ -0,0 +1,65 @@ +package workspace + +import ( + "bytes" + "fmt" + "strings" + "testing" +) + +func TestWriteGPGForwardFailedOSC_WellFormedSequence(t *testing.T) { + var buf bytes.Buffer + writeGPGForwardFailedOSC(&buf, "socket did not appear") + + want := fmt.Sprintf("\x1b]%d;socket did not appear\a", gpgForwardFailedOSC) + if got := buf.String(); got != want { + t.Fatalf("writeGPGForwardFailedOSC() = %q, want %q", got, want) + } +} + +func TestWriteGPGForwardFailedOSC_StripsControlCharsFromReason(t *testing.T) { + var buf bytes.Buffer + writeGPGForwardFailedOSC(&buf, "line one\nline\ttwo\x1b[31mred\a") + + got := buf.String() + prefix := fmt.Sprintf("\x1b]%d;", gpgForwardFailedOSC) + if len(got) < len(prefix) || got[:len(prefix)] != prefix { + t.Fatalf("missing OSC prefix: got %q", got) + } + if got[len(got)-1] != '\a' { + t.Fatalf("missing BEL terminator: got %q", got) + } + body := got[len(prefix) : len(got)-1] + for _, r := range body { + if r < 0x20 || r == 0x7f { + t.Fatalf("body still contains control char %q: %q", r, got) + } + } +} + +func TestWriteGPGForwardFailedOSC_StripsC1ControlsAndSeparator(t *testing.T) { + var buf bytes.Buffer + // œ is the 8-bit string terminator; ';' is the OSC field separator. + reason := "abc" + string(rune(0x9c)) + "def;ghi" + writeGPGForwardFailedOSC(&buf, reason) + + got := buf.String() + prefix := fmt.Sprintf("\x1b]%d;", gpgForwardFailedOSC) + body := got[len(prefix) : len(got)-1] + if body != "abcdefghi" { + t.Fatalf("body = %q, want %q", body, "abcdefghi") + } +} + +func TestWriteGPGForwardFailedOSC_TruncatesLongReason(t *testing.T) { + var buf bytes.Buffer + reason := strings.Repeat("a", gpgForwardFailedReasonMaxLen+100) + writeGPGForwardFailedOSC(&buf, reason) + + got := buf.String() + prefix := fmt.Sprintf("\x1b]%d;", gpgForwardFailedOSC) + body := got[len(prefix) : len(got)-1] + if len(body) != gpgForwardFailedReasonMaxLen { + t.Fatalf("body length = %d, want %d", len(body), gpgForwardFailedReasonMaxLen) + } +} diff --git a/cmd/workspace/port_forward.go b/cmd/workspace/port_forward.go new file mode 100644 index 000000000..526042642 --- /dev/null +++ b/cmd/workspace/port_forward.go @@ -0,0 +1,240 @@ +package workspace + +import ( + "context" + "errors" + "fmt" + "io" + "net" + "time" + + "github.com/devsy-org/devsy/pkg/log" + "github.com/devsy-org/devsy/pkg/port" + devssh "github.com/devsy-org/devsy/pkg/ssh" + "golang.org/x/crypto/ssh" +) + +func (cmd *SSHCmd) forwardTimeout() (time.Duration, error) { + if cmd.ForwardPortsTimeout == "" { + return 0, nil + } + + timeout, err := time.ParseDuration(cmd.ForwardPortsTimeout) + if err != nil { + return 0, fmt.Errorf("parse forward ports timeout: %w", err) + } + + log.Infof("Using port forwarding timeout of %s", cmd.ForwardPortsTimeout) + return timeout, nil +} + +// forwardPortsIfRequested handles -L/-R forwarding when requested. The returned +// bool reports whether forwarding took over (the caller should return err). +func (cmd *SSHCmd) forwardPortsIfRequested( + ctx context.Context, + sshClient *ssh.Client, +) (bool, error) { + if len(cmd.ForwardPorts) > 0 { + return true, cmd.forwardPorts(ctx, sshClient) + } + if len(cmd.ReverseForwardPorts) > 0 && !cmd.GPGAgentForwarding { + return true, cmd.reverseForwardPorts(ctx, sshClient) + } + return false, nil +} + +func (cmd *SSHCmd) forwardPorts( + ctx context.Context, + containerClient *ssh.Client, +) error { + timeout, err := cmd.forwardTimeout() + if err != nil { + return fmt.Errorf("parse forward ports timeout: %w", err) + } + + errChan := make(chan error, len(cmd.ForwardPorts)) + for _, portMapping := range cmd.ForwardPorts { + mapping, err := port.ParsePortSpec(portMapping) + if err != nil { + return fmt.Errorf("parse port mapping: %w", err) + } + + // start the forwarding + log.Infof( + "Forwarding local %s/%s to remote %s/%s", + mapping.Host.Protocol, + mapping.Host.Address, + mapping.Container.Protocol, + mapping.Container.Address, + ) + go func(portMapping string) { + err := devssh.PortForward( + ctx, + containerClient, + mapping.Host.Protocol, + mapping.Host.Address, + mapping.Container.Protocol, + mapping.Container.Address, + timeout, + ) + if errors.Is(err, devssh.ErrIdleTimeout) { + log.Infof("port-forward %s exited due to idle timeout", portMapping) + errChan <- nil + return + } + if err == nil || errors.Is(err, io.EOF) { + errChan <- nil + return + } + errChan <- fmt.Errorf("error forwarding %s: %w", portMapping, err) + }(portMapping) + } + + select { + case err := <-errChan: + return err + case <-ctx.Done(): + return ctx.Err() + } +} + +func (cmd *SSHCmd) reverseForwardPorts( + ctx context.Context, + containerClient *ssh.Client, +) error { + timeout, err := cmd.forwardTimeout() + if err != nil { + return fmt.Errorf("parse forward ports timeout: %w", err) + } + + errChan := make(chan error, len(cmd.ReverseForwardPorts)) + for _, portMapping := range cmd.ReverseForwardPorts { + mapping, err := port.ParsePortSpec(portMapping) + if err != nil { + return fmt.Errorf("parse port mapping: %w", err) + } + + // start the forwarding + log.Infof( + "Reverse forwarding local %s/%s to remote %s/%s", + mapping.Host.Protocol, + mapping.Host.Address, + mapping.Container.Protocol, + mapping.Container.Address, + ) + go func(portMapping string) { + err := devssh.ReversePortForward( + ctx, + containerClient, + mapping.Host.Protocol, + mapping.Host.Address, + mapping.Container.Protocol, + mapping.Container.Address, + timeout, + ) + if errors.Is(err, devssh.ErrIdleTimeout) { + log.Infof("reverse port-forward %s exited due to idle timeout", portMapping) + errChan <- nil + return + } + if err == nil || errors.Is(err, io.EOF) { + errChan <- nil + return + } + errChan <- fmt.Errorf("error forwarding %s: %w", portMapping, err) + }(portMapping) + } + + select { + case err := <-errChan: + return err + case <-ctx.Done(): + return ctx.Err() + } +} + +type boundReverseForward struct { + portMapping string + mapping port.Mapping + listener net.Listener +} + +// startReverseForwardsAndWait blocks until every forward's listener is bound, +// unlike reverseForwardPorts which blocks for the forward's lifetime. +func (cmd *SSHCmd) startReverseForwardsAndWait( + ctx context.Context, + containerClient *ssh.Client, + portMappings []string, +) error { + timeout, err := cmd.forwardTimeout() + if err != nil { + return err + } + + bound, err := bindReverseForwards(containerClient, portMappings) + if err != nil { + return err + } + + for _, b := range bound { + log.Infof( + "Reverse forwarding local %s/%s to remote %s/%s", + b.mapping.Host.Protocol, + b.mapping.Host.Address, + b.mapping.Container.Protocol, + b.mapping.Container.Address, + ) + go runReverseForwardInBackground(ctx, containerClient, b, timeout) + } + + return nil +} + +func bindReverseForwards( + containerClient *ssh.Client, + portMappings []string, +) ([]boundReverseForward, error) { + var bound []boundReverseForward + closeBound := func() { + for _, b := range bound { + _ = b.listener.Close() + } + } + for _, portMapping := range portMappings { + mapping, err := port.ParsePortSpec(portMapping) + if err != nil { + closeBound() + return nil, fmt.Errorf("parse port mapping: %w", err) + } + + listener, err := devssh.ReverseListen( + containerClient, + mapping.Host.Protocol, + mapping.Host.Address, + ) + if err != nil { + closeBound() + return nil, fmt.Errorf("listen for reverse forward %s: %w", portMapping, err) + } + bound = append(bound, boundReverseForward{portMapping, mapping, listener}) + } + return bound, nil +} + +func runReverseForwardInBackground( + ctx context.Context, + containerClient *ssh.Client, + b boundReverseForward, + timeout time.Duration, +) { + err := devssh.RunReverseForward(ctx, containerClient, devssh.ReverseForwardOpts{ + Listener: b.listener, + RemoteAddr: b.mapping.Host.Address, + LocalNetwork: b.mapping.Container.Protocol, + LocalAddr: b.mapping.Container.Address, + ExitAfterTimeout: timeout, + }) + if err != nil && !errors.Is(err, devssh.ErrIdleTimeout) && !errors.Is(err, io.EOF) { + log.Errorf("error forwarding %s: %v", b.portMapping, err) + } +} diff --git a/cmd/workspace/ssh.go b/cmd/workspace/ssh.go index 918b1493a..3493c4e1d 100644 --- a/cmd/workspace/ssh.go +++ b/cmd/workspace/ssh.go @@ -2,11 +2,8 @@ package workspace import ( "context" - "encoding/base64" - "errors" "fmt" "io" - "net" "os" "path" "strings" @@ -21,9 +18,7 @@ import ( "github.com/devsy-org/devsy/pkg/config" cliflags "github.com/devsy-org/devsy/pkg/flags" "github.com/devsy-org/devsy/pkg/flags/names" - "github.com/devsy-org/devsy/pkg/gpg" "github.com/devsy-org/devsy/pkg/log" - "github.com/devsy-org/devsy/pkg/port" "github.com/devsy-org/devsy/pkg/provider" devssh "github.com/devsy-org/devsy/pkg/ssh" "github.com/devsy-org/devsy/pkg/tunnel" @@ -232,7 +227,7 @@ func (cmd *SSHCmd) jumpContainerTailscale( devsyConfig *config.Config, client client2.DaemonClient, ) error { - log.Debugf("Starting tailscale connection") + log.Debugf("starting tailscale connection") err := client.CheckWorkspaceReachable(ctx) if err != nil { @@ -253,10 +248,8 @@ func (cmd *SSHCmd) jumpContainerTailscale( cmd.startServicesDaemon(ctx, devsyConfig, client, toolSSHClient) - // Handle GPG agent forwarding - if err := cmd.maybeSetupGPGAgent(ctx, devsyConfig, toolSSHClient); err != nil { - return err - } + gpgTunnel := newGPGTunnel(cmd, devsyConfig) + defer runGPGTunnelInBackground(ctx, gpgTunnel, toolSSHClient)() // Handle ssh stdio mode if cmd.Stdio { @@ -283,21 +276,6 @@ func (cmd *SSHCmd) jumpContainerTailscale( ) } -// forwardPortsIfRequested handles -L/-R forwarding when requested. The returned -// bool reports whether forwarding took over (the caller should return err). -func (cmd *SSHCmd) forwardPortsIfRequested( - ctx context.Context, - sshClient *ssh.Client, -) (bool, error) { - if len(cmd.ForwardPorts) > 0 { - return true, cmd.forwardPorts(ctx, sshClient) - } - if len(cmd.ReverseForwardPorts) > 0 && !cmd.GPGAgentForwarding { - return true, cmd.reverseForwardPorts(ctx, sshClient) - } - return false, nil -} - func (cmd *SSHCmd) startServicesDaemon( ctx context.Context, devsyConfig *config.Config, @@ -325,22 +303,6 @@ func (cmd *SSHCmd) startServicesDaemon( }() } -func (cmd *SSHCmd) maybeSetupGPGAgent( - ctx context.Context, - devsyConfig *config.Config, - sshClient *ssh.Client, -) error { - if !cmd.GPGAgentForwarding && - !devsyConfig.ContextOptionBool(config.ContextOptionGPGAgentForwarding) { - return nil - } - if gpg.IsGpgTunnelRunning(ctx, cmd.User, sshClient) { - log.Debugf("[GPG] exporting already running, skipping") - return nil - } - return cmd.setupGPGAgent(ctx, sshClient) -} - func (cmd *SSHCmd) startProxyTunnel( ctx context.Context, devsyConfig *config.Config, @@ -415,222 +377,28 @@ func (cmd *SSHCmd) jumpContainer( }, devsyConfig, envVars) } -func (cmd *SSHCmd) forwardTimeout() (time.Duration, error) { - timeout := time.Duration(0) - if cmd.ForwardPortsTimeout != "" { - timeout, err := time.ParseDuration(cmd.ForwardPortsTimeout) - if err != nil { - return timeout, fmt.Errorf("parse forward ports timeout: %w", err) - } - - log.Infof("Using port forwarding timeout of %s", cmd.ForwardPortsTimeout) - } - - return timeout, nil -} - -func (cmd *SSHCmd) reverseForwardPorts( - ctx context.Context, - containerClient *ssh.Client, -) error { - timeout, err := cmd.forwardTimeout() - if err != nil { - return fmt.Errorf("parse forward ports timeout: %w", err) - } - - errChan := make(chan error, len(cmd.ReverseForwardPorts)) - for _, portMapping := range cmd.ReverseForwardPorts { - mapping, err := port.ParsePortSpec(portMapping) - if err != nil { - return fmt.Errorf("parse port mapping: %w", err) - } - - // start the forwarding - log.Infof( - "Reverse forwarding local %s/%s to remote %s/%s", - mapping.Host.Protocol, - mapping.Host.Address, - mapping.Container.Protocol, - mapping.Container.Address, - ) - go func(portMapping string) { - err := devssh.ReversePortForward( - ctx, - containerClient, - mapping.Host.Protocol, - mapping.Host.Address, - mapping.Container.Protocol, - mapping.Container.Address, - timeout, - ) - if errors.Is(err, devssh.ErrIdleTimeout) { - log.Infof("reverse port-forward %s exited due to idle timeout", portMapping) - errChan <- nil - return - } - if !errors.Is(err, io.EOF) { - errChan <- fmt.Errorf("error forwarding %s: %w", portMapping, err) - } - }(portMapping) - } - - return <-errChan -} - -type boundReverseForward struct { - portMapping string - mapping port.Mapping - listener net.Listener -} - -// startReverseForwardsAndWait blocks until every forward's listener is bound, -// unlike reverseForwardPorts which blocks for the forward's lifetime. -func (cmd *SSHCmd) startReverseForwardsAndWait( - ctx context.Context, - containerClient *ssh.Client, -) error { - timeout, err := cmd.forwardTimeout() - if err != nil { - return err - } - - bound, err := bindReverseForwards(containerClient, cmd.ReverseForwardPorts) - if err != nil { - return err - } - - for _, b := range bound { - log.Infof( - "Reverse forwarding local %s/%s to remote %s/%s", - b.mapping.Host.Protocol, - b.mapping.Host.Address, - b.mapping.Container.Protocol, - b.mapping.Container.Address, - ) - go runReverseForwardInBackground(ctx, containerClient, b, timeout) - } - - return nil -} - -func bindReverseForwards( - containerClient *ssh.Client, - portMappings []string, -) ([]boundReverseForward, error) { - var bound []boundReverseForward - closeBound := func() { - for _, b := range bound { - _ = b.listener.Close() - } - } - for _, portMapping := range portMappings { - mapping, err := port.ParsePortSpec(portMapping) - if err != nil { - closeBound() - return nil, fmt.Errorf("parse port mapping: %w", err) - } - - listener, err := devssh.ReverseListen( - containerClient, - mapping.Host.Protocol, - mapping.Host.Address, - ) - if err != nil { - closeBound() - return nil, fmt.Errorf("listen for reverse forward %s: %w", portMapping, err) - } - bound = append(bound, boundReverseForward{portMapping, mapping, listener}) - } - return bound, nil -} - -func runReverseForwardInBackground( - ctx context.Context, - containerClient *ssh.Client, - b boundReverseForward, - timeout time.Duration, -) { - err := devssh.RunReverseForward(ctx, containerClient, devssh.ReverseForwardOpts{ - Listener: b.listener, - RemoteAddr: b.mapping.Host.Address, - LocalNetwork: b.mapping.Container.Protocol, - LocalAddr: b.mapping.Container.Address, - ExitAfterTimeout: timeout, - }) - if err != nil && !errors.Is(err, devssh.ErrIdleTimeout) && !errors.Is(err, io.EOF) { - log.Errorf("error forwarding %s: %v", b.portMapping, err) - } -} - -func (cmd *SSHCmd) forwardPorts( - ctx context.Context, - containerClient *ssh.Client, -) error { - timeout, err := cmd.forwardTimeout() - if err != nil { - return fmt.Errorf("parse forward ports timeout: %w", err) - } - - errChan := make(chan error, len(cmd.ForwardPorts)) - for _, portMapping := range cmd.ForwardPorts { - mapping, err := port.ParsePortSpec(portMapping) - if err != nil { - return fmt.Errorf("parse port mapping: %w", err) - } - - // start the forwarding - log.Infof( - "Forwarding local %s/%s to remote %s/%s", - mapping.Host.Protocol, - mapping.Host.Address, - mapping.Container.Protocol, - mapping.Container.Address, - ) - go func(portMapping string) { - err := devssh.PortForward( - ctx, - containerClient, - mapping.Host.Protocol, - mapping.Host.Address, - mapping.Container.Protocol, - mapping.Container.Address, - timeout, - ) - if errors.Is(err, devssh.ErrIdleTimeout) { - log.Infof("port-forward %s exited due to idle timeout", portMapping) - errChan <- nil - return - } - if !errors.Is(err, io.EOF) { - errChan <- fmt.Errorf("error forwarding %s: %w", portMapping, err) - } - }(portMapping) - } - - return <-errChan -} - func (cmd *SSHCmd) startTunnel( ctx context.Context, devsyConfig *config.Config, containerClient *ssh.Client, workspaceClient client2.BaseWorkspaceClient, ) error { - // check if we should forward ports if handled, err := cmd.forwardPortsIfRequested(ctx, containerClient); handled { return err } cmd.startTunnelServices(ctx, devsyConfig, containerClient, workspaceClient) + // buildSSHServerCommand runs `devsy internal ssh-server`, which always + // logs structured JSON on stderr; PipeJSONStream re-emits each line at + // its original level instead of double-wrapping it as another log entry. + writer, writerDone := log.PipeJSONStream() + defer func() { + _ = writer.Close() + <-writerDone + }() - // start ssh - writer := log.Writer(log.LevelInfo) - defer func() { _ = writer.Close() }() - - // check if we should do gpg agent forwarding - if err := cmd.maybeSetupGPGAgent(ctx, devsyConfig, containerClient); err != nil { - return err - } + gpgTunnel := newGPGTunnel(cmd, devsyConfig) + defer runGPGTunnelInBackground(ctx, gpgTunnel, containerClient)() workdir := resolveWorkdir(cmd.WorkDir, workspaceClient) @@ -689,26 +457,19 @@ func (cmd *SSHCmd) startTunnelServices( if !cmd.StartServices { return } - configureDockerCredentials := devsyConfig.ContextOption( - config.ContextOptionSSHInjectDockerCredentials, - ) == config.BoolTrue - configureGitCredentials := devsyConfig.ContextOption( - config.ContextOptionSSHInjectGitCredentials, - ) == config.BoolTrue - configureGitSSHSignatureHelper := devsyConfig.ContextOption( - config.ContextOptionGitSSHSignatureForwarding, - ) == config.BoolTrue - - go cmd.startServices( - ctx, - devsyConfig, - containerClient, - workspaceClient.WorkspaceConfig(), - configureDockerCredentials, - configureGitCredentials, - configureGitSSHSignatureHelper, - cmd.GitSSHSigningKey, - ) + + go cmd.startServices(ctx, devsyConfig, containerClient, workspaceClient.WorkspaceConfig(), startServicesOptions{ + ConfigureDockerCredentials: devsyConfig.ContextOption( + config.ContextOptionSSHInjectDockerCredentials, + ) == config.BoolTrue, + ConfigureGitCredentials: devsyConfig.ContextOption( + config.ContextOptionSSHInjectGitCredentials, + ) == config.BoolTrue, + ConfigureGitSSHSignatureHelper: devsyConfig.ContextOption( + config.ContextOptionGitSSHSignatureForwarding, + ) == config.BoolTrue, + GitSSHSigningKey: cmd.GitSSHSigningKey, + }) } func (cmd *SSHCmd) buildSSHServerCommand(workdir string) string { @@ -776,13 +537,20 @@ func resolveMergedWorkspaceFolder( return result.MergedConfig.WorkspaceFolder } +// startServicesOptions groups the credential-helper toggles for startServices. +type startServicesOptions struct { + ConfigureDockerCredentials bool + ConfigureGitCredentials bool + ConfigureGitSSHSignatureHelper bool + GitSSHSigningKey string +} + func (cmd *SSHCmd) startServices( ctx context.Context, devsyConfig *config.Config, containerClient *ssh.Client, workspace *provider.Workspace, - configureDockerCredentials, configureGitCredentials, configureGitSSHSignatureHelper bool, - gitSSHSigningKey string, + opts startServicesOptions, ) { if cmd.User != "" { err := tunnel.RunServices( @@ -795,10 +563,10 @@ func (cmd *SSHCmd) startServices( ExtraPorts: nil, PlatformOptions: nil, Workspace: workspace, - ConfigureDockerCredentials: configureDockerCredentials, - ConfigureGitCredentials: configureGitCredentials, - ConfigureGitSSHSignatureHelper: configureGitSSHSignatureHelper, - GitSSHSigningKey: gitSSHSigningKey, + ConfigureDockerCredentials: opts.ConfigureDockerCredentials, + ConfigureGitCredentials: opts.ConfigureGitCredentials, + ConfigureGitSSHSignatureHelper: opts.ConfigureGitSSHSignatureHelper, + GitSSHSigningKey: opts.GitSSHSigningKey, }, ) if err != nil { @@ -807,86 +575,6 @@ func (cmd *SSHCmd) startServices( } } -// setupGPGAgent will forward a local gpg-agent into the remote container -// this works by using cmd/internal/agentworkspace/setup_gpg. -func (cmd *SSHCmd) setupGPGAgent( - ctx context.Context, - containerClient *ssh.Client, -) error { - log.Debugf("[GPG] exporting gpg owner trust from host") - ownerTrustExport, err := gpg.GetHostOwnerTrust() - if err != nil { - return fmt.Errorf("export local ownertrust from GPG: %w", err) - } - ownerTrustArgument := base64.StdEncoding.EncodeToString(ownerTrustExport) - - log.Debugf("[GPG] detecting gpg-agent socket path on host") - // Detect local agent extra socket, this will be forwarded to the remote and - // symlinked in multiple paths - gpgExtraSocketPath, err := gpg.DetectAgentSocketPath() - if err != nil { - return err - } - log.Debugf("[GPG] detected gpg-agent socket path %s", gpgExtraSocketPath) - - gitKey := gpg.SigningKey(ctx) - - cmd.ReverseForwardPorts = append( - cmd.ReverseForwardPorts, - gpg.ContainerSocketPath+":"+gpgExtraSocketPath, - ) - - // Now we forward the agent socket to the remote, and setup remote gpg to use it - forwardAgent := []string{ - config.ContainerDevsyHelperLocation, - "internal", - "agent", - "workspace", - "setup-gpg", - names.Flag(names.OwnerTrust), - ownerTrustArgument, - names.Flag(names.SocketPath), - gpg.ContainerSocketPath, - } - - if log.DebugEnabled() { - forwardAgent = append(forwardAgent, names.Flag(names.Debug)) - } - - if gitKey != "" { - forwardAgent = append(forwardAgent, names.Flag(names.GitKey)) - forwardAgent = append(forwardAgent, gitKey) - } - - command := shellescape.QuoteCommand(forwardAgent) - if cmd.User != "" && cmd.User != "root" { - command = shellescape.QuoteCommand([]string{"su", "-c", command, cmd.User}) - } - - log.Debugf( - "[GPG] start reverse forward of gpg-agent socket %s, keeping connection open", - gpgExtraSocketPath, - ) - - if err := cmd.startReverseForwardsAndWait(ctx, containerClient); err != nil { - return fmt.Errorf("start gpg-agent reverse forward: %w", err) - } - - writer := log.Writer(log.LevelInfo) - defer func() { _ = writer.Close() }() - err = devssh.Run(ctx, devssh.RunOptions{ - Client: containerClient, - Command: command, - Stdout: writer, - Stderr: writer, - }) - if err != nil { - return fmt.Errorf("run gpg agent setup command: %w", err) - } - - return nil -} - func startSSHKeepAlive( ctx context.Context, client *ssh.Client, diff --git a/desktop/src/renderer/src/lib/components/terminal/Terminal.svelte b/desktop/src/renderer/src/lib/components/terminal/Terminal.svelte index a34c61f8c..c2a50be96 100644 --- a/desktop/src/renderer/src/lib/components/terminal/Terminal.svelte +++ b/desktop/src/renderer/src/lib/components/terminal/Terminal.svelte @@ -17,14 +17,20 @@ import { } from "$lib/stores/terminal-instances.js" import { get } from "svelte/store" +// OSC 9977 is a private-use sequence the devsy CLI emits when GPG-agent +// forwarding fails but the SSH session continues anyway. +const GPG_FORWARD_FAILED_OSC = 9977 + let { sessionId, active = true, onExit, + onGpgForwardFailed, }: { sessionId: string active?: boolean onExit?: (exitCode?: number, signal?: number) => void + onGpgForwardFailed?: (reason: string) => void } = $props() let containerEl: HTMLDivElement | undefined = $state() @@ -103,6 +109,7 @@ onMount(async () => { // Reattach existing terminal to the new container term = existing.term fitAddon = existing.fitAddon + existing.onGpgForwardFailed = onGpgForwardFailed const el = term.element?.parentElement ?? term.element if (el) containerEl.appendChild(el) requestAnimationFrame(() => { @@ -179,13 +186,23 @@ onMount(async () => { } }) - setTerminalInstance(sessionId, { + const instance: TerminalInstance = { term, fitAddon, unlistenOutput, unlistenExit, unsubscribeTheme, - }) + onGpgForwardFailed, + } + const oscHandler = term.parser.registerOscHandler( + GPG_FORWARD_FAILED_OSC, + (data) => { + instance.onGpgForwardFailed?.(data) + return true + }, + ) + instance.disposeOscHandler = () => oscHandler.dispose() + setTerminalInstance(sessionId, instance) } resizeObserver = new ResizeObserver(() => { diff --git a/desktop/src/renderer/src/lib/stores/terminal-instances.ts b/desktop/src/renderer/src/lib/stores/terminal-instances.ts index 2403e51f2..f5c3eade2 100644 --- a/desktop/src/renderer/src/lib/stores/terminal-instances.ts +++ b/desktop/src/renderer/src/lib/stores/terminal-instances.ts @@ -7,6 +7,8 @@ export interface TerminalInstance { unlistenOutput?: () => void unlistenExit?: () => void unsubscribeTheme?: () => void + disposeOscHandler?: () => void + onGpgForwardFailed?: (reason: string) => void } const instances = new Map() @@ -30,6 +32,7 @@ export function destroyTerminalInstance(sessionId: string): void { instance.unlistenOutput?.() instance.unlistenExit?.() instance.unsubscribeTheme?.() + instance.disposeOscHandler?.() instance.term.dispose() instances.delete(sessionId) } diff --git a/desktop/src/renderer/src/pages/WorkspaceDetailPage.svelte b/desktop/src/renderer/src/pages/WorkspaceDetailPage.svelte index 38fee4dd6..801998bf1 100644 --- a/desktop/src/renderer/src/pages/WorkspaceDetailPage.svelte +++ b/desktop/src/renderer/src/pages/WorkspaceDetailPage.svelte @@ -400,6 +400,12 @@ function handleSshExit(exitCode?: number, _signal?: number) { } } +function handleGpgForwardFailed(reason: string) { + toasts.error( + `GPG agent forwarding failed for ${id}, continuing without it: ${reason}`, + ) +} + function isDebug(): boolean { return loadLocalOptions().debugFlag } @@ -989,7 +995,11 @@ async function handleRenameConfirmed() { {:else} - + {/if} {#if sshExited} diff --git a/pkg/gpg/gpg_forwarding.go b/pkg/gpg/gpg_forwarding.go index 94a7fce39..06ce155ad 100644 --- a/pkg/gpg/gpg_forwarding.go +++ b/pkg/gpg/gpg_forwarding.go @@ -1,7 +1,6 @@ package gpg import ( - "bytes" "context" "fmt" "os" @@ -15,6 +14,7 @@ import ( "github.com/devsy-org/devsy/pkg/log" devssh "github.com/devsy-org/devsy/pkg/ssh" "golang.org/x/crypto/ssh" + "k8s.io/apimachinery/pkg/util/wait" ) type GPGConf struct { @@ -24,6 +24,10 @@ type GPGConf struct { GitKey string } +// IsGpgTunnelRunning reports whether a live gpg-agent forward already +// reaches the container by pinging the agent rather than scanning its +// keyring. gpg-connect-agent always exits 0 even when unreachable, so +// liveness is read from stdout: a live agent answers with a trailing "OK". func IsGpgTunnelRunning( ctx context.Context, user string, @@ -32,13 +36,12 @@ func IsGpgTunnelRunning( writer := log.PassthroughWriter() defer func() { _ = writer.Close() }() - command := "gpg -K" + command := `echo "GETINFO version" | timeout 5 gpg-connect-agent --no-autostart` if user != "" && user != "root" { command = shellescape.QuoteCommand([]string{"su", "-c", command, user}) } - // empty output means the forwarded agent exposes no secret keys - var out bytes.Buffer + var out strings.Builder err := devssh.Run(ctx, devssh.RunOptions{ Client: client, Command: command, @@ -46,7 +49,7 @@ func IsGpgTunnelRunning( Stderr: writer, }) - return err == nil && strings.TrimSpace(out.String()) != "" + return err == nil && strings.HasSuffix(strings.TrimSpace(out.String()), "OK") } func GetHostPubKey() ([]byte, error) { @@ -155,7 +158,7 @@ func (g *GPGConf) SetupRemoteSocketDirTree() error { ).Run() } -func (g *GPGConf) SetupRemoteSocketLink() error { +func (g *GPGConf) SetupRemoteSocketLink(ctx context.Context) error { links := []string{ filepath.Join(os.Getenv("HOME"), ".gnupg", "S.gpg-agent"), filepath.Join("/run/user", strconv.Itoa(os.Getuid()), "gnupg", "S.gpg-agent"), @@ -172,22 +175,39 @@ func (g *GPGConf) SetupRemoteSocketLink() error { } } - return g.claimForwardedSocket() + return g.claimForwardedSocket(ctx) } // claimForwardedSocket takes ownership of the socket, which the ssh server -// binds as root; a non-root user needs write access to connect. It is bound -// asynchronously, so wait briefly for it to appear. -func (g *GPGConf) claimForwardedSocket() error { +// binds as root; a non-root user needs write access to connect. The socket +// is bound asynchronously by the ssh server, so this waits for it to appear. +func (g *GPGConf) claimForwardedSocket(ctx context.Context) error { owner := strconv.Itoa(os.Getuid()) + ":" + strconv.Itoa(os.Getgid()) - for range 30 { - if _, err := os.Stat(g.SocketPath); err == nil { - //nolint:gosec // g.SocketPath is the fixed forwarded socket path - return exec.Command("sudo", "chown", owner, g.SocketPath).Run() + + backoff := wait.Backoff{ + Duration: 200 * time.Millisecond, + Factor: 1.5, + Jitter: 0.1, + Steps: 15, + Cap: 2 * time.Second, + } + + err := wait.ExponentialBackoffWithContext(ctx, backoff, func(_ context.Context) (bool, error) { + info, err := os.Stat(g.SocketPath) + if err != nil { + return false, nil // Retry + } + if info.Mode()&os.ModeSocket == 0 { + return false, fmt.Errorf("path %q exists but is not a unix socket", g.SocketPath) } - time.Sleep(100 * time.Millisecond) + return true, nil + }) + if err != nil { + return fmt.Errorf("forwarded gpg socket %q did not appear as expected: %w", g.SocketPath, err) } - return fmt.Errorf("forwarded gpg socket %q did not appear", g.SocketPath) + + //nolint:gosec // g.SocketPath is the fixed forwarded socket path + return exec.Command("sudo", "chown", owner, g.SocketPath).Run() } func gpgConfigPath() string { diff --git a/pkg/gpg/gpg_forwarding_test.go b/pkg/gpg/gpg_forwarding_test.go index f3f485291..a47091f99 100644 --- a/pkg/gpg/gpg_forwarding_test.go +++ b/pkg/gpg/gpg_forwarding_test.go @@ -1,15 +1,43 @@ package gpg import ( + "context" + "errors" + "net" "os" "path/filepath" "strings" "testing" + "time" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) +// listenUnixSocket creates a real Unix domain socket at path, registers it +// for cleanup, and returns it. The caller must run this on the test +// goroutine (or join the calling goroutine) before the test returns, so the +// listener is registered before t.Cleanup fires and a net.Listen failure is +// observed by the test rather than silently dropped in a detached goroutine. +func listenUnixSocket(t *testing.T, path string) net.Listener { + t.Helper() + ln, err := net.Listen("unix", path) + require.NoError(t, err) + t.Cleanup(func() { _ = ln.Close() }) + return ln +} + +// shortSocketDir returns a short-enough temp dir for a unix socket path: +// t.TempDir()'s nested path can exceed the ~104-byte sun_path limit on +// macOS/BSD, so this creates directly under os.TempDir() instead. +func shortSocketDir(t *testing.T) string { + t.Helper() + dir, err := os.MkdirTemp("", "gpgsock") + require.NoError(t, err) + t.Cleanup(func() { _ = os.RemoveAll(dir) }) + return dir +} + func TestSetupGpgConf_WritesRequiredDirectives(t *testing.T) { home := t.TempDir() t.Setenv("HOME", home) @@ -66,6 +94,79 @@ func TestSetupGpgConf_ExistingFileWithoutTrailingNewline(t *testing.T) { } } +func TestClaimForwardedSocket_StopsPollingAsSoonAsSocketAppears(t *testing.T) { + socketPath := filepath.Join(shortSocketDir(t), "S.gpg-agent") + g := &GPGConf{SocketPath: socketPath} + + listenerReady := make(chan struct{}) + go func() { + defer close(listenerReady) + time.Sleep(50 * time.Millisecond) + listenUnixSocket(t, socketPath) + }() + defer func() { <-listenerReady }() + + start := time.Now() + err := g.claimForwardedSocket(context.Background()) + elapsed := time.Since(start) + + assert.Less(t, elapsed, 2*time.Second, "must return shortly after the socket appears") + // claimForwardedSocket's final step (chown to the caller's uid) requires + // sudo, which may not be passwordless in this environment; only the + // socket-appear wait itself is under test here, so a chown-only failure + // is tolerated, but any error about the socket not appearing is not. + if err != nil { + assert.NotContains(t, err.Error(), "did not appear") + assert.NotContains(t, err.Error(), "not a unix socket") + } +} + +func TestClaimForwardedSocket_TimesOutWhenSocketNeverAppears(t *testing.T) { + g := &GPGConf{SocketPath: filepath.Join(t.TempDir(), "S.gpg-agent")} + + // The backoff's own worst case (15 steps, 200ms*1.5^n capped at 2s, plus + // up to 10% jitter per step) is a bit over 20s, so the timeout used to + // wait it out must exceed that, not sit just under it, or this assertion + // itself becomes flaky on the exact path it verifies. + ctx, cancel := context.WithTimeout(context.Background(), 25*time.Second) + defer cancel() + + err := g.claimForwardedSocket(ctx) + + require.Error(t, err) + assert.Contains(t, err.Error(), "did not appear") +} + +func TestClaimForwardedSocket_RejectsNonSocketPath(t *testing.T) { + socketPath := filepath.Join(t.TempDir(), "S.gpg-agent") + require.NoError(t, os.WriteFile(socketPath, []byte("not a socket"), 0o600)) + g := &GPGConf{SocketPath: socketPath} + + start := time.Now() + err := g.claimForwardedSocket(context.Background()) + elapsed := time.Since(start) + + require.Error(t, err) + assert.Contains(t, err.Error(), "not a unix socket") + assert.Less(t, elapsed, time.Second, "must fail immediately, not retry against a non-socket path") +} + +func TestClaimForwardedSocket_RespectsContextCancellation(t *testing.T) { + g := &GPGConf{SocketPath: filepath.Join(t.TempDir(), "S.gpg-agent")} + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + start := time.Now() + err := g.claimForwardedSocket(ctx) + elapsed := time.Since(start) + + require.Error(t, err) + assert.True(t, errors.Is(err, context.Canceled), + "error must wrap context.Canceled, not just any error: got %v", err) + assert.Less(t, elapsed, time.Second, "cancelled context must not wait out the full backoff") +} + func readConf(t *testing.T, path string) string { t.Helper() b, err := os.ReadFile(path) //nolint:gosec // test path is created by the test diff --git a/pkg/tunnel/container.go b/pkg/tunnel/container.go index 5cf6175cf..77786e847 100644 --- a/pkg/tunnel/container.go +++ b/pkg/tunnel/container.go @@ -91,8 +91,13 @@ func (c *ContainerTunnel) runHostTunnel( stdinReader, stdoutWriter *os.File, timeout time.Duration, ) error { - writer := log.Writer(log.LevelInfo) - defer func() { _ = writer.Close() }() + // `devsy internal ...` always logs structured JSON on stderr; PipeJSONStream + // re-emits each line at its original level instead of double-wrapping it. + writer, done := log.PipeJSONStream() + defer func() { + _ = writer.Close() + <-done + }() defer log.Debugf("Tunnel to host closed") command := fmt.Sprintf("'%s' internal ssh-server --stdio", c.client.AgentPath()) @@ -234,8 +239,14 @@ type containerTunnelOpts struct { // stdoutWriter on exit so StdioClient gets EOF when the tunnel dies. // Context-cancelled errors are suppressed (expected during normal shutdown). func (c *ContainerTunnel) runContainerTunnel(ctx context.Context, opts containerTunnelOpts) error { - writer := log.Writer(log.LevelInfo) - defer func() { _ = writer.Close() }() + // See runHostTunnel: this command is also `devsy internal ...`, which + // always logs structured JSON on stderr, so PipeJSONStream (not + // log.Writer) is what avoids double-wrapping each already-JSON line. + writer, done := log.PipeJSONStream() + defer func() { + _ = writer.Close() + <-done + }() defer func() { _ = opts.stdoutWriter.Close() }() log.Debugf("Run container tunnel") From f6549441cef43a7936b3ba21a040e2a834b1a910 Mon Sep 17 00:00:00 2001 From: Samuel K Date: Sun, 2 Aug 2026 00:13:03 -0500 Subject: [PATCH 02/14] fix(lint): resolve golangci-lint findings in ssh/gpg-tunnel changes - Deduplicate forwardPorts/reverseForwardPorts into a shared runPortForwards helper parameterized by forwardDirection (dupl). - Split gpgTunnel.setup into buildSetupCommand and ensureForwardBound to reduce cyclomatic complexity (cyclop). - Bundle runPortForwards' inputs into a portForwardRun struct to stay within the function argument-count limit (revive). - Reformat several long lines/calls (golines). --- .../agentcontainer/credentials_server.go | 5 +- cmd/internal/agentworkspace/setup_gpg_test.go | 8 +- cmd/workspace/gpg_tunnel.go | 111 ++++++++------ cmd/workspace/port_forward.go | 136 +++++++++--------- cmd/workspace/ssh.go | 5 +- pkg/gpg/gpg_forwarding.go | 6 +- pkg/gpg/gpg_forwarding_test.go | 7 +- 7 files changed, 160 insertions(+), 118 deletions(-) diff --git a/cmd/internal/agentcontainer/credentials_server.go b/cmd/internal/agentcontainer/credentials_server.go index 4cc6dc79e..454979ae5 100644 --- a/cmd/internal/agentcontainer/credentials_server.go +++ b/cmd/internal/agentcontainer/credentials_server.go @@ -144,7 +144,10 @@ func checkPortClaimable(port int) error { return fmt.Errorf("check port %d availability: %w", port, err) } if !ok { - return fmt.Errorf("port %d not available (another session likely owns the credentials server)", port) + return fmt.Errorf( + "port %d not available (another session likely owns the credentials server)", + port, + ) } return nil } diff --git a/cmd/internal/agentworkspace/setup_gpg_test.go b/cmd/internal/agentworkspace/setup_gpg_test.go index 879a16a26..5a1376889 100644 --- a/cmd/internal/agentworkspace/setup_gpg_test.go +++ b/cmd/internal/agentworkspace/setup_gpg_test.go @@ -88,6 +88,10 @@ func TestAcquireGPGSetupLock_ReturnsCancellationErrorWhenCallerCancels(t *testin _, err = acquireGPGSetupLock(ctx) require.Error(t, err) - assert.True(t, errors.Is(err, context.Canceled), - "caller cancellation must surface as context.Canceled, not a generic lock-timeout error: got %v", err) + assert.True( + t, + errors.Is(err, context.Canceled), + "caller cancellation must surface as context.Canceled, not a generic lock-timeout error: got %v", + err, + ) } diff --git a/cmd/workspace/gpg_tunnel.go b/cmd/workspace/gpg_tunnel.go index 46e894f9f..07ec97473 100644 --- a/cmd/workspace/gpg_tunnel.go +++ b/cmd/workspace/gpg_tunnel.go @@ -117,7 +117,10 @@ func (t *gpgTunnel) ensure(ctx context.Context, sshClient *ssh.Client) { return } if gpg.IsGpgTunnelRunning(ctx, t.cmd.User, sshClient) { - log.Debugf("GPG tunnel setup failed but tunnel is live (won by a concurrent terminal): %v", err) + log.Debugf( + "GPG tunnel setup failed but tunnel is live (won by a concurrent terminal): %v", + err, + ) t.failureReported = false return } @@ -138,15 +141,6 @@ func (t *gpgTunnel) ensure(ctx context.Context, sshClient *ssh.Client) { // setup forwards the local gpg-agent into the remote container by using // cmd/internal/agentworkspace/setup_gpg. func (t *gpgTunnel) setup(ctx context.Context, containerClient *ssh.Client) error { - cmd := t.cmd - - log.Debugf("[GPG] exporting gpg owner trust from host") - ownerTrustExport, err := gpg.GetHostOwnerTrust() - if err != nil { - return fmt.Errorf("export local ownertrust from GPG: %w", err) - } - ownerTrustArgument := base64.StdEncoding.EncodeToString(ownerTrustExport) - log.Debugf("detecting gpg-agent socket path on host") // Detect local agent extra socket, this will be forwarded to the remote and // symlinked in multiple paths @@ -156,9 +150,46 @@ func (t *gpgTunnel) setup(ctx context.Context, containerClient *ssh.Client) erro } log.Debugf("[GPG] detected gpg-agent socket path %s", gpgExtraSocketPath) + command, err := t.buildSetupCommand(ctx) + if err != nil { + return err + } + + if err := t.ensureForwardBound(ctx, containerClient, gpgExtraSocketPath); err != nil { + return err + } + + writer, writerDone := log.PipeJSONStream() + defer func() { + _ = writer.Close() + <-writerDone + }() + if err := devssh.Run(ctx, devssh.RunOptions{ + Client: containerClient, + Command: command, + Stdout: writer, + Stderr: writer, + }); err != nil { + return fmt.Errorf("run gpg agent setup command: %w", err) + } + + return nil +} + +// buildSetupCommand assembles the remote `setup-gpg` invocation, exporting +// the host's owner trust and signing key into its arguments. +func (t *gpgTunnel) buildSetupCommand(ctx context.Context) (string, error) { + cmd := t.cmd + + log.Debugf("[GPG] exporting gpg owner trust from host") + ownerTrustExport, err := gpg.GetHostOwnerTrust() + if err != nil { + return "", fmt.Errorf("export local ownertrust from GPG: %w", err) + } + ownerTrustArgument := base64.StdEncoding.EncodeToString(ownerTrustExport) + gitKey := gpg.SigningKey(ctx) - // Now we forward the agent socket to the remote, and setup remote gpg to use it forwardAgent := []string{ config.ContainerDevsyHelperLocation, "internal", @@ -170,54 +201,46 @@ func (t *gpgTunnel) setup(ctx context.Context, containerClient *ssh.Client) erro names.Flag(names.SocketPath), gpg.ContainerSocketPath, } - if log.DebugEnabled() { forwardAgent = append(forwardAgent, names.Flag(names.Debug)) } - if gitKey != "" { - forwardAgent = append(forwardAgent, names.Flag(names.GitKey)) - forwardAgent = append(forwardAgent, gitKey) + forwardAgent = append(forwardAgent, names.Flag(names.GitKey), gitKey) } command := shellescape.QuoteCommand(forwardAgent) if cmd.User != "" && cmd.User != "root" { command = shellescape.QuoteCommand([]string{"su", "-c", command, cmd.User}) } + return command, nil +} - // Bind the reverse-listen socket at most once per process (see - // forwardBound); the remote setup-gpg step below still re-runs every - // time to repair remote-side agent state (stopped agent, stale keys). - if !t.forwardBound { - log.Debugf( - "[GPG] start reverse forward of gpg-agent socket %s, keeping connection open", - gpgExtraSocketPath, - ) - reverseForwardPorts := append( - []string{gpg.ContainerSocketPath + ":" + gpgExtraSocketPath}, - cmd.ReverseForwardPorts..., - ) - if err := cmd.startReverseForwardsAndWait(ctx, containerClient, reverseForwardPorts); err != nil { - return fmt.Errorf("start gpg-agent reverse forward: %w", err) - } - t.forwardBound = true +// ensureForwardBound binds the reverse-listen socket at most once per +// process (see forwardBound); setup's remote command still re-runs every +// time to repair remote-side agent state (stopped agent, stale keys), which +// doesn't require touching the reverse-forward at all. +func (t *gpgTunnel) ensureForwardBound( + ctx context.Context, + containerClient *ssh.Client, + gpgExtraSocketPath string, +) error { + if t.forwardBound { + return nil } - writer, writerDone := log.PipeJSONStream() - defer func() { - _ = writer.Close() - <-writerDone - }() - err = devssh.Run(ctx, devssh.RunOptions{ - Client: containerClient, - Command: command, - Stdout: writer, - Stderr: writer, - }) + log.Debugf( + "[GPG] start reverse forward of gpg-agent socket %s, keeping connection open", + gpgExtraSocketPath, + ) + reverseForwardPorts := append( + []string{gpg.ContainerSocketPath + ":" + gpgExtraSocketPath}, + t.cmd.ReverseForwardPorts..., + ) + err := t.cmd.startReverseForwardsAndWait(ctx, containerClient, reverseForwardPorts) if err != nil { - return fmt.Errorf("run gpg agent setup command: %w", err) + return fmt.Errorf("start gpg-agent reverse forward: %w", err) } - + t.forwardBound = true return nil } diff --git a/cmd/workspace/port_forward.go b/cmd/workspace/port_forward.go index 526042642..e38290559 100644 --- a/cmd/workspace/port_forward.go +++ b/cmd/workspace/port_forward.go @@ -43,97 +43,99 @@ func (cmd *SSHCmd) forwardPortsIfRequested( return false, nil } -func (cmd *SSHCmd) forwardPorts( - ctx context.Context, - containerClient *ssh.Client, -) error { - timeout, err := cmd.forwardTimeout() - if err != nil { - return fmt.Errorf("parse forward ports timeout: %w", err) - } - - errChan := make(chan error, len(cmd.ForwardPorts)) - for _, portMapping := range cmd.ForwardPorts { - mapping, err := port.ParsePortSpec(portMapping) - if err != nil { - return fmt.Errorf("parse port mapping: %w", err) - } +// forwardDirection abstracts the one difference between forwardPorts and +// reverseForwardPorts: which devssh function establishes each connection and +// how log messages describe it. +type forwardDirection struct { + logPrefix string // e.g. "Forwarding" or "Reverse forwarding" + logLabel string // e.g. "port-forward" or "reverse port-forward" + forward func(ctx context.Context, client *ssh.Client, mapping port.Mapping, timeout time.Duration) error +} - // start the forwarding - log.Infof( - "Forwarding local %s/%s to remote %s/%s", - mapping.Host.Protocol, - mapping.Host.Address, - mapping.Container.Protocol, - mapping.Container.Address, - ) - go func(portMapping string) { - err := devssh.PortForward( - ctx, - containerClient, - mapping.Host.Protocol, - mapping.Host.Address, - mapping.Container.Protocol, - mapping.Container.Address, +var ( + directionForward = forwardDirection{ + logPrefix: "Forwarding", + logLabel: "port-forward", + forward: func(ctx context.Context, client *ssh.Client, mapping port.Mapping, timeout time.Duration) error { + return devssh.PortForward( + ctx, client, + mapping.Host.Protocol, mapping.Host.Address, + mapping.Container.Protocol, mapping.Container.Address, timeout, ) - if errors.Is(err, devssh.ErrIdleTimeout) { - log.Infof("port-forward %s exited due to idle timeout", portMapping) - errChan <- nil - return - } - if err == nil || errors.Is(err, io.EOF) { - errChan <- nil - return - } - errChan <- fmt.Errorf("error forwarding %s: %w", portMapping, err) - }(portMapping) + }, + } + directionReverse = forwardDirection{ + logPrefix: "Reverse forwarding", + logLabel: "reverse port-forward", + forward: func(ctx context.Context, client *ssh.Client, mapping port.Mapping, timeout time.Duration) error { + return devssh.ReversePortForward( + ctx, client, + mapping.Host.Protocol, mapping.Host.Address, + mapping.Container.Protocol, mapping.Container.Address, + timeout, + ) + }, } +) - select { - case err := <-errChan: - return err - case <-ctx.Done(): - return ctx.Err() +// portForwardRun bundles a single forwardPorts/reverseForwardPorts +// invocation's inputs so runPortForwards stays within the linter's +// argument-count limit. +type portForwardRun struct { + portMappings []string + timeout time.Duration + dir forwardDirection +} + +func (cmd *SSHCmd) forwardPorts(ctx context.Context, containerClient *ssh.Client) error { + timeout, err := cmd.forwardTimeout() + if err != nil { + return fmt.Errorf("parse forward ports timeout: %w", err) } + return runPortForwards(ctx, containerClient, portForwardRun{ + portMappings: cmd.ForwardPorts, + timeout: timeout, + dir: directionForward, + }) } -func (cmd *SSHCmd) reverseForwardPorts( - ctx context.Context, - containerClient *ssh.Client, -) error { +func (cmd *SSHCmd) reverseForwardPorts(ctx context.Context, containerClient *ssh.Client) error { timeout, err := cmd.forwardTimeout() if err != nil { return fmt.Errorf("parse forward ports timeout: %w", err) } + return runPortForwards(ctx, containerClient, portForwardRun{ + portMappings: cmd.ReverseForwardPorts, + timeout: timeout, + dir: directionReverse, + }) +} - errChan := make(chan error, len(cmd.ReverseForwardPorts)) - for _, portMapping := range cmd.ReverseForwardPorts { +// runPortForwards starts one forwarding goroutine per mapping and blocks +// until the first one reports a result (an error, or a clean exit via idle +// timeout / EOF) or ctx is done. +func runPortForwards(ctx context.Context, containerClient *ssh.Client, run portForwardRun) error { + timeout, dir := run.timeout, run.dir + errChan := make(chan error, len(run.portMappings)) + for _, portMapping := range run.portMappings { mapping, err := port.ParsePortSpec(portMapping) if err != nil { return fmt.Errorf("parse port mapping: %w", err) } - // start the forwarding log.Infof( - "Reverse forwarding local %s/%s to remote %s/%s", + "%s local %s/%s to remote %s/%s", + dir.logPrefix, mapping.Host.Protocol, mapping.Host.Address, mapping.Container.Protocol, mapping.Container.Address, ) - go func(portMapping string) { - err := devssh.ReversePortForward( - ctx, - containerClient, - mapping.Host.Protocol, - mapping.Host.Address, - mapping.Container.Protocol, - mapping.Container.Address, - timeout, - ) + go func(portMapping string, mapping port.Mapping) { + err := dir.forward(ctx, containerClient, mapping, timeout) if errors.Is(err, devssh.ErrIdleTimeout) { - log.Infof("reverse port-forward %s exited due to idle timeout", portMapping) + log.Infof("%s %s exited due to idle timeout", dir.logLabel, portMapping) errChan <- nil return } @@ -142,7 +144,7 @@ func (cmd *SSHCmd) reverseForwardPorts( return } errChan <- fmt.Errorf("error forwarding %s: %w", portMapping, err) - }(portMapping) + }(portMapping, mapping) } select { diff --git a/cmd/workspace/ssh.go b/cmd/workspace/ssh.go index 3493c4e1d..9d64e1b3b 100644 --- a/cmd/workspace/ssh.go +++ b/cmd/workspace/ssh.go @@ -458,7 +458,7 @@ func (cmd *SSHCmd) startTunnelServices( return } - go cmd.startServices(ctx, devsyConfig, containerClient, workspaceClient.WorkspaceConfig(), startServicesOptions{ + opts := startServicesOptions{ ConfigureDockerCredentials: devsyConfig.ContextOption( config.ContextOptionSSHInjectDockerCredentials, ) == config.BoolTrue, @@ -469,7 +469,8 @@ func (cmd *SSHCmd) startTunnelServices( config.ContextOptionGitSSHSignatureForwarding, ) == config.BoolTrue, GitSSHSigningKey: cmd.GitSSHSigningKey, - }) + } + go cmd.startServices(ctx, devsyConfig, containerClient, workspaceClient.WorkspaceConfig(), opts) } func (cmd *SSHCmd) buildSSHServerCommand(workdir string) string { diff --git a/pkg/gpg/gpg_forwarding.go b/pkg/gpg/gpg_forwarding.go index 06ce155ad..3b8b87673 100644 --- a/pkg/gpg/gpg_forwarding.go +++ b/pkg/gpg/gpg_forwarding.go @@ -203,7 +203,11 @@ func (g *GPGConf) claimForwardedSocket(ctx context.Context) error { return true, nil }) if err != nil { - return fmt.Errorf("forwarded gpg socket %q did not appear as expected: %w", g.SocketPath, err) + return fmt.Errorf( + "forwarded gpg socket %q did not appear as expected: %w", + g.SocketPath, + err, + ) } //nolint:gosec // g.SocketPath is the fixed forwarded socket path diff --git a/pkg/gpg/gpg_forwarding_test.go b/pkg/gpg/gpg_forwarding_test.go index a47091f99..1a1663652 100644 --- a/pkg/gpg/gpg_forwarding_test.go +++ b/pkg/gpg/gpg_forwarding_test.go @@ -148,7 +148,12 @@ func TestClaimForwardedSocket_RejectsNonSocketPath(t *testing.T) { require.Error(t, err) assert.Contains(t, err.Error(), "not a unix socket") - assert.Less(t, elapsed, time.Second, "must fail immediately, not retry against a non-socket path") + assert.Less( + t, + elapsed, + time.Second, + "must fail immediately, not retry against a non-socket path", + ) } func TestClaimForwardedSocket_RespectsContextCancellation(t *testing.T) { From efd2604ff04d78f4e7078ee564cb4d495c54fd45 Mon Sep 17 00:00:00 2001 From: Samuel K Date: Sun, 2 Aug 2026 01:25:14 -0500 Subject: [PATCH 03/14] fix(review): address CodeRabbit findings on GPG-tunnel reliability PR - Register the desktop terminal's OSC 9977 handler before flushing buffered output, so a GPG-forward-failure notice that arrives during the async xterm import window isn't parsed and discarded early. - Give the desktop toast a fixed reason string instead of surfacing err.Error() verbatim, since it can wrap remote SSH exit output. - Reject combined -L/-R forwarding with an explicit error instead of silently dropping one; stop double-wrapping errors already wrapped by forwardTimeout; wait for all -L/-R mappings to finish instead of returning after the first one. - Claim the credentials-server port before starting the port watcher, avoiding a leaked goroutine on failed port claims. - Add a fallback path to PipeJSONStream so plain-text stderr from the agent-injection script isn't silently dropped alongside the structured JSON logs that follow it on the same stream. - Fix a test helper that called require.NoError/t.Cleanup from a background goroutine, which testing disallows. --- .../agentcontainer/credentials_server.go | 8 +- cmd/workspace/gpg_tunnel.go | 6 +- cmd/workspace/gpg_tunnel_test.go | 15 ++++ cmd/workspace/port_forward.go | 85 +++++++++++++------ .../lib/components/terminal/Terminal.svelte | 39 +++++---- pkg/gpg/gpg_forwarding_test.go | 30 ++++--- pkg/log/jsonstream.go | 42 ++++++++- pkg/tunnel/container.go | 16 ++-- 8 files changed, 175 insertions(+), 66 deletions(-) diff --git a/cmd/internal/agentcontainer/credentials_server.go b/cmd/internal/agentcontainer/credentials_server.go index 454979ae5..d20622738 100644 --- a/cmd/internal/agentcontainer/credentials_server.go +++ b/cmd/internal/agentcontainer/credentials_server.go @@ -100,12 +100,16 @@ func (cmd *CredentialsServerCmd) Run(ctx context.Context, port int) error { return fmt.Errorf("ping client: %w", err) } - cmd.maybeForwardPorts(ctx, tunnelClient) - + // Claim the port before starting anything else: on contention this + // returns an error so RunServices' retry (pkg/tunnel/services.go) tries + // again later, and starting the port watcher beforehand would leak an + // orphaned goroutine on every failed attempt. if err := checkPortClaimable(port); err != nil { return err } + cmd.maybeForwardPorts(ctx, tunnelClient) + // configure docker credential helper if err := cmd.configureDockerHelper(port); err != nil { return err diff --git a/cmd/workspace/gpg_tunnel.go b/cmd/workspace/gpg_tunnel.go index 07ec97473..afdbf32bc 100644 --- a/cmd/workspace/gpg_tunnel.go +++ b/cmd/workspace/gpg_tunnel.go @@ -135,7 +135,11 @@ func (t *gpgTunnel) ensure(ctx context.Context, sshClient *ssh.Client) { return } t.failureReported = true - writeGPGForwardFailedOSC(os.Stderr, err.Error()) + // The desktop toast gets a fixed, concise reason rather than err.Error(): + // the underlying error can wrap remote SSH exit output, which isn't + // something to surface verbatim in the UI. Full detail stays in the log + // line above (visible with --debug). + writeGPGForwardFailedOSC(os.Stderr, "check logs for details") } // setup forwards the local gpg-agent into the remote container by using diff --git a/cmd/workspace/gpg_tunnel_test.go b/cmd/workspace/gpg_tunnel_test.go index d1b94f64a..209fb324d 100644 --- a/cmd/workspace/gpg_tunnel_test.go +++ b/cmd/workspace/gpg_tunnel_test.go @@ -63,3 +63,18 @@ func TestWriteGPGForwardFailedOSC_TruncatesLongReason(t *testing.T) { t.Fatalf("body length = %d, want %d", len(body), gpgForwardFailedReasonMaxLen) } } + +func TestWriteGPGForwardFailedOSC_TruncatesByRunesNotBytes(t *testing.T) { + var buf bytes.Buffer + // "é" is 2 bytes in UTF-8; byte-based truncation would produce a body + // longer than gpgForwardFailedReasonMaxLen bytes or split a rune in two. + reason := strings.Repeat("é", gpgForwardFailedReasonMaxLen+100) + writeGPGForwardFailedOSC(&buf, reason) + + got := buf.String() + prefix := fmt.Sprintf("\x1b]%d;", gpgForwardFailedOSC) + body := got[len(prefix) : len(got)-1] + if n := len([]rune(body)); n != gpgForwardFailedReasonMaxLen { + t.Fatalf("body rune count = %d, want %d", n, gpgForwardFailedReasonMaxLen) + } +} diff --git a/cmd/workspace/port_forward.go b/cmd/workspace/port_forward.go index e38290559..8ed62cd56 100644 --- a/cmd/workspace/port_forward.go +++ b/cmd/workspace/port_forward.go @@ -30,14 +30,22 @@ func (cmd *SSHCmd) forwardTimeout() (time.Duration, error) { // forwardPortsIfRequested handles -L/-R forwarding when requested. The returned // bool reports whether forwarding took over (the caller should return err). +// -L and -R together aren't supported: each takes over the whole session +// (never returning to hand off to the shell), so running both would mean +// silently dropping one — this reports an explicit error instead. func (cmd *SSHCmd) forwardPortsIfRequested( ctx context.Context, sshClient *ssh.Client, ) (bool, error) { - if len(cmd.ForwardPorts) > 0 { + hasForward := len(cmd.ForwardPorts) > 0 + hasReverse := len(cmd.ReverseForwardPorts) > 0 && !cmd.GPGAgentForwarding + if hasForward && hasReverse { + return true, fmt.Errorf("-L and -R cannot be combined in a single ssh invocation") + } + if hasForward { return true, cmd.forwardPorts(ctx, sshClient) } - if len(cmd.ReverseForwardPorts) > 0 && !cmd.GPGAgentForwarding { + if hasReverse { return true, cmd.reverseForwardPorts(ctx, sshClient) } return false, nil @@ -91,7 +99,7 @@ type portForwardRun struct { func (cmd *SSHCmd) forwardPorts(ctx context.Context, containerClient *ssh.Client) error { timeout, err := cmd.forwardTimeout() if err != nil { - return fmt.Errorf("parse forward ports timeout: %w", err) + return err } return runPortForwards(ctx, containerClient, portForwardRun{ portMappings: cmd.ForwardPorts, @@ -103,7 +111,7 @@ func (cmd *SSHCmd) forwardPorts(ctx context.Context, containerClient *ssh.Client func (cmd *SSHCmd) reverseForwardPorts(ctx context.Context, containerClient *ssh.Client) error { timeout, err := cmd.forwardTimeout() if err != nil { - return fmt.Errorf("parse forward ports timeout: %w", err) + return err } return runPortForwards(ctx, containerClient, portForwardRun{ portMappings: cmd.ReverseForwardPorts, @@ -113,10 +121,11 @@ func (cmd *SSHCmd) reverseForwardPorts(ctx context.Context, containerClient *ssh } // runPortForwards starts one forwarding goroutine per mapping and blocks -// until the first one reports a result (an error, or a clean exit via idle -// timeout / EOF) or ctx is done. +// until every mapping has exited cleanly (idle timeout or EOF), one reports +// a real error, or ctx is done. A single mapping's clean exit doesn't tear +// down the others: with multiple -L/-R mappings, an idle timeout on one +// shouldn't end the whole session while its siblings are still useful. func runPortForwards(ctx context.Context, containerClient *ssh.Client, run portForwardRun) error { - timeout, dir := run.timeout, run.dir errChan := make(chan error, len(run.portMappings)) for _, portMapping := range run.portMappings { mapping, err := port.ParsePortSpec(portMapping) @@ -126,33 +135,59 @@ func runPortForwards(ctx context.Context, containerClient *ssh.Client, run portF log.Infof( "%s local %s/%s to remote %s/%s", - dir.logPrefix, + run.dir.logPrefix, mapping.Host.Protocol, mapping.Host.Address, mapping.Container.Protocol, mapping.Container.Address, ) - go func(portMapping string, mapping port.Mapping) { - err := dir.forward(ctx, containerClient, mapping, timeout) - if errors.Is(err, devssh.ErrIdleTimeout) { - log.Infof("%s %s exited due to idle timeout", dir.logLabel, portMapping) - errChan <- nil - return - } - if err == nil || errors.Is(err, io.EOF) { - errChan <- nil - return + go runPortForward(ctx, containerClient, singlePortForward{ + dir: run.dir, + timeout: run.timeout, + portMapping: portMapping, + mapping: mapping, + errChan: errChan, + }) + } + + for range run.portMappings { + select { + case err := <-errChan: + if err != nil { + return err } - errChan <- fmt.Errorf("error forwarding %s: %w", portMapping, err) - }(portMapping, mapping) + case <-ctx.Done(): + return ctx.Err() + } } + return nil +} - select { - case err := <-errChan: - return err - case <-ctx.Done(): - return ctx.Err() +// singlePortForward bundles a single runPortForward invocation's inputs so +// it stays within the linter's argument-count limit. +type singlePortForward struct { + dir forwardDirection + timeout time.Duration + portMapping string + mapping port.Mapping + errChan chan<- error +} + +// runPortForward runs a single mapping's forward and reports its outcome on +// f.errChan: nil for a clean exit (idle timeout or EOF), the wrapped error +// otherwise. +func runPortForward(ctx context.Context, containerClient *ssh.Client, f singlePortForward) { + err := f.dir.forward(ctx, containerClient, f.mapping, f.timeout) + if errors.Is(err, devssh.ErrIdleTimeout) { + log.Infof("%s %s exited due to idle timeout", f.dir.logLabel, f.portMapping) + f.errChan <- nil + return + } + if err == nil || errors.Is(err, io.EOF) { + f.errChan <- nil + return } + f.errChan <- fmt.Errorf("error forwarding %s: %w", f.portMapping, err) } type boundReverseForward struct { diff --git a/desktop/src/renderer/src/lib/components/terminal/Terminal.svelte b/desktop/src/renderer/src/lib/components/terminal/Terminal.svelte index c2a50be96..ac6250a2e 100644 --- a/desktop/src/renderer/src/lib/components/terminal/Terminal.svelte +++ b/desktop/src/renderer/src/lib/components/terminal/Terminal.svelte @@ -170,30 +170,18 @@ onMount(async () => { term.open(containerEl) fitAddon.fit() - // Flush any output that arrived during async imports - for (const data of outputBuffer) { - term.write(data) - } - - term.onData((data) => { - const encoded = new TextEncoder().encode(data) - terminalWrite(sessionId, Array.from(encoded)) - }) - - const unsubscribeTheme = theme.subscribe(() => { - if (term) { - term.options.theme = resolveTheme() - } - }) - const instance: TerminalInstance = { term, fitAddon, unlistenOutput, unlistenExit, - unsubscribeTheme, + unsubscribeTheme: undefined, onGpgForwardFailed, } + // Register before flushing outputBuffer below: term.write drives the OSC + // parser synchronously, so a failure notification buffered during the + // async import window above would otherwise be parsed and discarded + // before this handler exists to catch it. const oscHandler = term.parser.registerOscHandler( GPG_FORWARD_FAILED_OSC, (data) => { @@ -202,6 +190,23 @@ onMount(async () => { }, ) instance.disposeOscHandler = () => oscHandler.dispose() + + // Flush any output that arrived during async imports + for (const data of outputBuffer) { + term.write(data) + } + + term.onData((data) => { + const encoded = new TextEncoder().encode(data) + terminalWrite(sessionId, Array.from(encoded)) + }) + + instance.unsubscribeTheme = theme.subscribe(() => { + if (term) { + term.options.theme = resolveTheme() + } + }) + setTerminalInstance(sessionId, instance) } diff --git a/pkg/gpg/gpg_forwarding_test.go b/pkg/gpg/gpg_forwarding_test.go index 1a1663652..5979a9060 100644 --- a/pkg/gpg/gpg_forwarding_test.go +++ b/pkg/gpg/gpg_forwarding_test.go @@ -14,17 +14,15 @@ import ( "github.com/stretchr/testify/require" ) -// listenUnixSocket creates a real Unix domain socket at path, registers it -// for cleanup, and returns it. The caller must run this on the test -// goroutine (or join the calling goroutine) before the test returns, so the -// listener is registered before t.Cleanup fires and a net.Listen failure is -// observed by the test rather than silently dropped in a detached goroutine. -func listenUnixSocket(t *testing.T, path string) net.Listener { - t.Helper() - ln, err := net.Listen("unix", path) - require.NoError(t, err) - t.Cleanup(func() { _ = ln.Close() }) - return ln +// listenUnixSocket creates a real Unix domain socket at path and returns it +// along with any error. It deliberately avoids require.NoError/t.Cleanup: +// those call t.FailNow, which the testing package requires to run on the +// test's own goroutine, but this helper is called from a background +// goroutine in TestClaimForwardedSocket_StopsPollingAsSoonAsSocketAppears. +// The caller must check the returned error and register cleanup itself, on +// the test goroutine. +func listenUnixSocket(path string) (net.Listener, error) { + return net.Listen("unix", path) } // shortSocketDir returns a short-enough temp dir for a unix socket path: @@ -98,13 +96,19 @@ func TestClaimForwardedSocket_StopsPollingAsSoonAsSocketAppears(t *testing.T) { socketPath := filepath.Join(shortSocketDir(t), "S.gpg-agent") g := &GPGConf{SocketPath: socketPath} + var listener net.Listener + var listenErr error listenerReady := make(chan struct{}) go func() { defer close(listenerReady) time.Sleep(50 * time.Millisecond) - listenUnixSocket(t, socketPath) + listener, listenErr = listenUnixSocket(socketPath) + }() + defer func() { + <-listenerReady + require.NoError(t, listenErr) + _ = listener.Close() }() - defer func() { <-listenerReady }() start := time.Now() err := g.claimForwardedSocket(context.Background()) diff --git a/pkg/log/jsonstream.go b/pkg/log/jsonstream.go index 47a707032..b61d1e59e 100644 --- a/pkg/log/jsonstream.go +++ b/pkg/log/jsonstream.go @@ -19,6 +19,23 @@ func PipeJSONStream() (io.WriteCloser, chan struct{}) { return writer, done } +// PipeJSONStreamWithFallback is like PipeJSONStream, but a line that isn't +// valid JSON is written verbatim (with a trailing newline) to fallback +// instead of being silently dropped. Use this when the writer's lifetime +// spans more than one process/phase and only some of them are known to +// produce structured JSON — e.g. a shell script's plain-text stderr +// followed by a devsy subcommand's JSON logs on the same stream. +func PipeJSONStreamWithFallback(fallback io.Writer) (io.WriteCloser, chan struct{}) { + done := make(chan struct{}) + reader, writer := io.Pipe() + go func() { + readJSONStreamWithFallback(reader, fallback) + close(done) + }() + + return writer, done +} + // jsonLine is a self-contained representation of a JSON log line, // replacing the old github.com/devsy-org/log.Line type. type jsonLine struct { @@ -46,6 +63,14 @@ var levelFuncs = map[string]func(...any){ } func ReadJSONStream(reader io.Reader) { + readJSONStreamWithFallback(reader, nil) +} + +// readJSONStreamWithFallback parses reader line by line as JSON log lines. +// A line that isn't valid JSON, or whose level isn't recognized, is written +// to fallback verbatim if fallback is non-nil; otherwise it's dropped, +// matching ReadJSONStream's existing behavior for genuinely-JSON-only streams. +func readJSONStreamWithFallback(reader io.Reader, fallback io.Writer) { scan := scanner.NewScanner(reader) for scan.Scan() { line := scan.Bytes() @@ -54,14 +79,27 @@ func ReadJSONStream(reader io.Reader) { } obj := &jsonLine{} if err := json.Unmarshal(line, obj); err != nil { + writeFallbackLine(fallback, line) continue } msg := obj.text() if msg == "" { + writeFallbackLine(fallback, line) continue } - if fn, ok := levelFuncs[strings.ToLower(obj.Level)]; ok { - fn(msg) + fn, ok := levelFuncs[strings.ToLower(obj.Level)] + if !ok { + writeFallbackLine(fallback, line) + continue } + fn(msg) + } +} + +func writeFallbackLine(fallback io.Writer, line []byte) { + if fallback == nil { + return } + _, _ = fallback.Write(line) + _, _ = fallback.Write([]byte("\n")) } diff --git a/pkg/tunnel/container.go b/pkg/tunnel/container.go index 77786e847..f76703b4d 100644 --- a/pkg/tunnel/container.go +++ b/pkg/tunnel/container.go @@ -91,9 +91,13 @@ func (c *ContainerTunnel) runHostTunnel( stdinReader, stdoutWriter *os.File, timeout time.Duration, ) error { - // `devsy internal ...` always logs structured JSON on stderr; PipeJSONStream - // re-emits each line at its original level instead of double-wrapping it. - writer, done := log.PipeJSONStream() + // This stderr spans two phases: InjectAgent's own shell script (plain + // text) and, once injection succeeds, the actual `devsy internal + // ssh-server --stdio` invocation (always structured JSON). Use the + // fallback variant so the script's plain-text errors aren't silently + // dropped by the JSON-only parser while JSON lines still get re-emitted + // at their original level instead of double-wrapped. + writer, done := log.PipeJSONStreamWithFallback(log.PassthroughWriter()) defer func() { _ = writer.Close() <-done @@ -239,9 +243,9 @@ type containerTunnelOpts struct { // stdoutWriter on exit so StdioClient gets EOF when the tunnel dies. // Context-cancelled errors are suppressed (expected during normal shutdown). func (c *ContainerTunnel) runContainerTunnel(ctx context.Context, opts containerTunnelOpts) error { - // See runHostTunnel: this command is also `devsy internal ...`, which - // always logs structured JSON on stderr, so PipeJSONStream (not - // log.Writer) is what avoids double-wrapping each already-JSON line. + // Unlike runHostTunnel, this stderr only ever carries the `devsy internal + // agent container-tunnel` command's own JSON logs (no injection-script + // phase), so the plain PipeJSONStream is correct here. writer, done := log.PipeJSONStream() defer func() { _ = writer.Close() From 4e0e61bb854f9621a982a3be3408fcf957f47097 Mon Sep 17 00:00:00 2001 From: Samuel K Date: Sun, 2 Aug 2026 10:48:51 -0500 Subject: [PATCH 04/14] fix(log): close pipe reader when JSON stream scanner stops early An oversized line makes bufio.Scanner stop before io.Pipe is closed, leaving the paired writer blocked forever on the next Write since nothing is left draining the pipe. Close the reader once the goroutine's scan loop returns so the writer fails instead of hanging. --- pkg/log/jsonstream.go | 8 ++++++++ pkg/log/jsonstream_test.go | 39 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 47 insertions(+) create mode 100644 pkg/log/jsonstream_test.go diff --git a/pkg/log/jsonstream.go b/pkg/log/jsonstream.go index b61d1e59e..878bc5c97 100644 --- a/pkg/log/jsonstream.go +++ b/pkg/log/jsonstream.go @@ -13,6 +13,9 @@ func PipeJSONStream() (io.WriteCloser, chan struct{}) { reader, writer := io.Pipe() go func() { ReadJSONStream(reader) + // See PipeJSONStreamWithFallback: closing here unblocks a Write on + // writer if the scanner stopped early instead of at pipe close. + _ = reader.Close() close(done) }() @@ -30,6 +33,11 @@ func PipeJSONStreamWithFallback(fallback io.Writer) (io.WriteCloser, chan struct reader, writer := io.Pipe() go func() { readJSONStreamWithFallback(reader, fallback) + // If the scanner stopped early (e.g. an oversized line), reader + // otherwise stays open and a later Write on writer blocks forever + // with nothing left reading the pipe. Closing it here unblocks the + // paired writer with io.ErrClosedPipe instead of deadlocking. + _ = reader.Close() close(done) }() diff --git a/pkg/log/jsonstream_test.go b/pkg/log/jsonstream_test.go new file mode 100644 index 000000000..6a06ec45f --- /dev/null +++ b/pkg/log/jsonstream_test.go @@ -0,0 +1,39 @@ +package log + +import ( + "strings" + "testing" + "time" +) + +// TestPipeJSONStreamWithFallback_OversizedLineUnblocksWriter verifies that a +// line exceeding the scanner's buffer cap doesn't leave the pipe reader open +// after the goroutine exits: a later Write must fail promptly instead of +// blocking forever with nothing left draining the pipe. +func TestPipeJSONStreamWithFallback_OversizedLineUnblocksWriter(t *testing.T) { + writer, done := PipeJSONStreamWithFallback(PassthroughWriter()) + + oversized := strings.Repeat("a", 2*1024*1024) + "\n" + _, _ = writer.Write([]byte(oversized)) + + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("goroutine did not exit after oversized line") + } + + writeDone := make(chan error, 1) + go func() { + _, err := writer.Write([]byte("more\n")) + writeDone <- err + }() + + select { + case err := <-writeDone: + if err == nil { + t.Fatal("expected write to a closed pipe to fail") + } + case <-time.After(2 * time.Second): + t.Fatal("write blocked: reader was not closed when the scanner stopped early") + } +} From 46024fccb11478505782519d6764791784c25b98 Mon Sep 17 00:00:00 2001 From: Samuel K Date: Sun, 2 Aug 2026 10:59:21 -0500 Subject: [PATCH 05/14] refactor: trim comments that restate code Remove or shorten comments describing what the code already says plainly, keeping only the parts that explain non-obvious rationale. --- cmd/internal/agentcontainer/credentials_server.go | 8 ++++---- cmd/internal/agentcontainer/credentials_server_test.go | 7 +++---- cmd/workspace/gpg_tunnel.go | 6 ++---- cmd/workspace/ssh.go | 1 - 4 files changed, 9 insertions(+), 13 deletions(-) diff --git a/cmd/internal/agentcontainer/credentials_server.go b/cmd/internal/agentcontainer/credentials_server.go index d20622738..7086a71eb 100644 --- a/cmd/internal/agentcontainer/credentials_server.go +++ b/cmd/internal/agentcontainer/credentials_server.go @@ -137,10 +137,10 @@ func (cmd *CredentialsServerCmd) Run(ctx context.Context, port int) error { return credentials.RunCredentialsServer(ctx, port, tunnelClient) } -// checkPortClaimable reports an error if port is not free to bind. Only one -// session's credentials-server can hold this port at a time. Returning an -// error (not nil) on contention matters: RunServices (pkg/tunnel/services.go) -// wraps this command in retry.OnError, which only retries on a non-nil error. +// Only one session's credentials-server can hold this port at a time. +// Returning an error (not nil) on contention matters: RunServices +// (pkg/tunnel/services.go) wraps this command in retry.OnError, which only +// retries on a non-nil error. func checkPortClaimable(port int) error { addr := net.JoinHostPort("localhost", strconv.Itoa(port)) ok, err := portpkg.IsAvailable(addr) diff --git a/cmd/internal/agentcontainer/credentials_server_test.go b/cmd/internal/agentcontainer/credentials_server_test.go index 34c919ba1..5414d9294 100644 --- a/cmd/internal/agentcontainer/credentials_server_test.go +++ b/cmd/internal/agentcontainer/credentials_server_test.go @@ -9,10 +9,9 @@ import ( ) func TestCheckPortClaimable_SucceedsWhenPortFree(t *testing.T) { - // checkPortClaimable itself binds the port to check it, so pass 0 to let - // the OS assign a free ephemeral port at call time — this avoids the - // close-then-probe race of reserving a port via net.Listen, closing it, - // and hoping nothing else claims it before checkPortClaimable runs. + // Pass 0 to let the OS assign a free ephemeral port at call time — this + // avoids the close-then-probe race of reserving a port via net.Listen, + // closing it, and hoping nothing else claims it before this call runs. assert.NoError(t, checkPortClaimable(0)) } diff --git a/cmd/workspace/gpg_tunnel.go b/cmd/workspace/gpg_tunnel.go index afdbf32bc..c52406478 100644 --- a/cmd/workspace/gpg_tunnel.go +++ b/cmd/workspace/gpg_tunnel.go @@ -19,8 +19,7 @@ import ( ) // gpgForwardFailedOSC is a private-use OSC identifier the desktop app's -// terminal (xterm.js) listens for to surface a non-fatal GPG-forwarding -// failure as a toast; see Terminal.svelte's registerOscHandler. +// terminal (xterm.js) listens for; see Terminal.svelte's registerOscHandler. const gpgForwardFailedOSC = 9977 // gpgForwardFailedReasonMaxLen bounds the OSC payload, since the desktop @@ -146,8 +145,7 @@ func (t *gpgTunnel) ensure(ctx context.Context, sshClient *ssh.Client) { // cmd/internal/agentworkspace/setup_gpg. func (t *gpgTunnel) setup(ctx context.Context, containerClient *ssh.Client) error { log.Debugf("detecting gpg-agent socket path on host") - // Detect local agent extra socket, this will be forwarded to the remote and - // symlinked in multiple paths + // this socket gets forwarded to the remote and symlinked in multiple paths gpgExtraSocketPath, err := gpg.DetectAgentSocketPath() if err != nil { return err diff --git a/cmd/workspace/ssh.go b/cmd/workspace/ssh.go index 9d64e1b3b..162374954 100644 --- a/cmd/workspace/ssh.go +++ b/cmd/workspace/ssh.go @@ -538,7 +538,6 @@ func resolveMergedWorkspaceFolder( return result.MergedConfig.WorkspaceFolder } -// startServicesOptions groups the credential-helper toggles for startServices. type startServicesOptions struct { ConfigureDockerCredentials bool ConfigureGitCredentials bool From d5074706b4563cc1885674d48bae6a94ea389026 Mon Sep 17 00:00:00 2001 From: Samuel K Date: Sun, 2 Aug 2026 13:01:06 -0500 Subject: [PATCH 06/14] fix(review): address CodeRabbit findings on credentials-server race and test flakiness - Close the check-then-bind gap in the credentials-server port claim: claimPort now binds the port itself and hands the listener straight to RunCredentialsServerWithListener, so two concurrent sessions can't both pass an availability probe and race to bind. - Stop retrying permanent os.Stat errors (permission, invalid path) while waiting for the forwarded gpg socket to appear; only retry on the socket not existing yet, and don't mislabel those errors as "did not appear". - Deflake TestAcquireGPGSetupLock_WaitsForConcurrentHolderThenSucceeds by synchronizing on the release goroutine's start instead of racing time.Now() against its initial scheduling delay. - Strengthen TestAcquireGPGSetupLock_SucceedsWhenFree to reacquire the lock after unlock(), so a no-op unlock would actually fail the test. --- .../agentcontainer/credentials_server.go | 40 +++++++------ .../agentcontainer/credentials_server_test.go | 58 ++++++++++++++++--- cmd/internal/agentworkspace/setup_gpg_test.go | 14 ++++- pkg/credentials/server.go | 22 +++++-- pkg/gpg/gpg_forwarding.go | 10 +++- pkg/gpg/gpg_forwarding_test.go | 28 +++++++++ 6 files changed, 136 insertions(+), 36 deletions(-) diff --git a/cmd/internal/agentcontainer/credentials_server.go b/cmd/internal/agentcontainer/credentials_server.go index 7086a71eb..c8e5ab7b3 100644 --- a/cmd/internal/agentcontainer/credentials_server.go +++ b/cmd/internal/agentcontainer/credentials_server.go @@ -22,7 +22,6 @@ import ( "github.com/devsy-org/devsy/pkg/gitsshsigning" "github.com/devsy-org/devsy/pkg/log" "github.com/devsy-org/devsy/pkg/netstat" - portpkg "github.com/devsy-org/devsy/pkg/port" "github.com/spf13/cobra" ) @@ -100,13 +99,18 @@ func (cmd *CredentialsServerCmd) Run(ctx context.Context, port int) error { return fmt.Errorf("ping client: %w", err) } - // Claim the port before starting anything else: on contention this - // returns an error so RunServices' retry (pkg/tunnel/services.go) tries - // again later, and starting the port watcher beforehand would leak an + // Claim the port by binding it before starting anything else, and hold + // the listener until it's handed to RunCredentialsServerWithListener + // below: a check-then-bind gap here would let two concurrent sessions + // both pass the check and race to bind. On contention this returns an + // error so RunServices' retry (pkg/tunnel/services.go) tries again + // later, and starting the port watcher beforehand would leak an // orphaned goroutine on every failed attempt. - if err := checkPortClaimable(port); err != nil { + ln, err := claimPort(port) + if err != nil { return err } + defer func() { _ = ln.Close() }() cmd.maybeForwardPorts(ctx, tunnelClient) @@ -134,26 +138,26 @@ func (cmd *CredentialsServerCmd) Run(ctx context.Context, port int) error { cleanupGitSigning := cmd.configureGitSigningKey() defer cleanupGitSigning() - return credentials.RunCredentialsServer(ctx, port, tunnelClient) + return credentials.RunCredentialsServerWithListener(ctx, ln, tunnelClient) } -// Only one session's credentials-server can hold this port at a time. -// Returning an error (not nil) on contention matters: RunServices -// (pkg/tunnel/services.go) wraps this command in retry.OnError, which only -// retries on a non-nil error. -func checkPortClaimable(port int) error { +// claimPort binds port and returns the listener, holding it exclusively so +// no other session can bind the same port until the caller closes it (or +// hands it to RunCredentialsServerWithListener). Only one session's +// credentials-server can hold this port at a time. Returning an error (not +// nil) on contention matters: RunServices (pkg/tunnel/services.go) wraps +// this command in retry.OnError, which only retries on a non-nil error. +func claimPort(port int) (net.Listener, error) { addr := net.JoinHostPort("localhost", strconv.Itoa(port)) - ok, err := portpkg.IsAvailable(addr) + ln, err := net.Listen("tcp", addr) if err != nil { - return fmt.Errorf("check port %d availability: %w", port, err) - } - if !ok { - return fmt.Errorf( - "port %d not available (another session likely owns the credentials server)", + return nil, fmt.Errorf( + "port %d not available (another session likely owns the credentials server): %w", port, + err, ) } - return nil + return ln, nil } func (cmd *CredentialsServerCmd) maybeForwardPorts( diff --git a/cmd/internal/agentcontainer/credentials_server_test.go b/cmd/internal/agentcontainer/credentials_server_test.go index 5414d9294..d5987d3f3 100644 --- a/cmd/internal/agentcontainer/credentials_server_test.go +++ b/cmd/internal/agentcontainer/credentials_server_test.go @@ -2,31 +2,34 @@ package agentcontainer import ( "net" + "sync" "testing" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) -func TestCheckPortClaimable_SucceedsWhenPortFree(t *testing.T) { +func TestClaimPort_SucceedsWhenPortFree(t *testing.T) { // Pass 0 to let the OS assign a free ephemeral port at call time — this // avoids the close-then-probe race of reserving a port via net.Listen, // closing it, and hoping nothing else claims it before this call runs. - assert.NoError(t, checkPortClaimable(0)) + ln, err := claimPort(0) + require.NoError(t, err) + _ = ln.Close() } -func TestCheckPortClaimable_ErrorsWhenPortHeld(t *testing.T) { +func TestClaimPort_ErrorsWhenPortHeld(t *testing.T) { ln, err := net.Listen("tcp", "localhost:0") require.NoError(t, err) t.Cleanup(func() { _ = ln.Close() }) port := ln.Addr().(*net.TCPAddr).Port - err = checkPortClaimable(port) + _, err = claimPort(port) require.Error(t, err) assert.Contains(t, err.Error(), "not available") } -func TestCheckPortClaimable_BecomesClaimableAfterHolderReleases(t *testing.T) { +func TestClaimPort_BecomesClaimableAfterHolderReleases(t *testing.T) { ln, err := net.Listen("tcp", "localhost:0") require.NoError(t, err) closed := false @@ -37,12 +40,49 @@ func TestCheckPortClaimable_BecomesClaimableAfterHolderReleases(t *testing.T) { }) port := ln.Addr().(*net.TCPAddr).Port - require.Error(t, checkPortClaimable(port), - "port must read as unavailable while the listener is held") + _, err = claimPort(port) + require.Error(t, err, "port must read as unavailable while the listener is held") require.NoError(t, ln.Close()) closed = true - assert.NoError(t, checkPortClaimable(port), - "port must read as claimable once the prior holder releases it") + claimed, err := claimPort(port) + require.NoError(t, err, "port must read as claimable once the prior holder releases it") + _ = claimed.Close() +} + +// TestClaimPort_OnlyOneConcurrentCallerWins proves claimPort binds the port +// as the claim itself, rather than merely probing it: with a check-then-bind +// gap, two concurrent callers could both observe the port as free before +// either binds it. +func TestClaimPort_OnlyOneConcurrentCallerWins(t *testing.T) { + ln, err := net.Listen("tcp", "localhost:0") + require.NoError(t, err) + port := ln.Addr().(*net.TCPAddr).Port + require.NoError(t, ln.Close()) + + const callers = 20 + var wg sync.WaitGroup + var successes int + var mu sync.Mutex + var winner net.Listener + + for range callers { + wg.Go(func() { + claimedLn, claimErr := claimPort(port) + if claimErr != nil { + return + } + mu.Lock() + successes++ + winner = claimedLn + mu.Unlock() + }) + } + wg.Wait() + + assert.Equal(t, 1, successes, "exactly one concurrent caller must win the claim") + if winner != nil { + _ = winner.Close() + } } diff --git a/cmd/internal/agentworkspace/setup_gpg_test.go b/cmd/internal/agentworkspace/setup_gpg_test.go index 5a1376889..a1de4b3ce 100644 --- a/cmd/internal/agentworkspace/setup_gpg_test.go +++ b/cmd/internal/agentworkspace/setup_gpg_test.go @@ -21,6 +21,10 @@ func TestAcquireGPGSetupLock_SucceedsWhenFree(t *testing.T) { unlock, err := acquireGPGSetupLock(context.Background()) require.NoError(t, err) unlock() + + reacquire, err := acquireGPGSetupLock(context.Background()) + require.NoError(t, err, "unlock must actually release the lock") + reacquire() } func TestAcquireGPGSetupLock_WaitsForConcurrentHolderThenSucceeds(t *testing.T) { @@ -34,18 +38,22 @@ func TestAcquireGPGSetupLock_WaitsForConcurrentHolderThenSucceeds(t *testing.T) require.NoError(t, err) require.True(t, locked) + const releaseDelay = 200 * time.Millisecond + start := time.Now() + releasing := make(chan struct{}) go func() { - time.Sleep(200 * time.Millisecond) + close(releasing) + time.Sleep(releaseDelay) _ = holder.Unlock() }() + <-releasing - start := time.Now() unlock, err := acquireGPGSetupLock(context.Background()) elapsed := time.Since(start) require.NoError(t, err) defer unlock() - assert.GreaterOrEqual(t, elapsed, 150*time.Millisecond, + assert.GreaterOrEqual(t, elapsed, releaseDelay/2, "second acquirer must wait for the first to release, not run concurrently") } diff --git a/pkg/credentials/server.go b/pkg/credentials/server.go index 5e3c37377..99b7944f8 100644 --- a/pkg/credentials/server.go +++ b/pkg/credentials/server.go @@ -45,19 +45,33 @@ func RunCredentialsServer( port int, client CredentialsClient, ) error { - addr := net.JoinHostPort("localhost", strconv.Itoa(port)) + ln, err := net.Listen("tcp", net.JoinHostPort("localhost", strconv.Itoa(port))) + if err != nil { + return fmt.Errorf("listen on port %d: %w", port, err) + } + return RunCredentialsServerWithListener(ctx, ln, client) +} + +// RunCredentialsServerWithListener is like RunCredentialsServer, but takes an +// already-bound listener. Use this when the caller must hold the port +// exclusively (via net.Listen) from before startup through to serving, so no +// other process can bind the same port in between. +func RunCredentialsServerWithListener( + ctx context.Context, + ln net.Listener, + client CredentialsClient, +) error { srv := &http.Server{ - Addr: addr, Handler: newCredentialsHandler(ctx, client), ReadHeaderTimeout: 10 * time.Second, } errChan := make(chan error, 1) go func() { - log.Debugf("credentials server started: port=%v", port) + log.Debugf("credentials server started: addr=%v", ln.Addr()) // always returns error. ErrServerClosed on graceful close - if err := srv.ListenAndServe(); err != http.ErrServerClosed { + if err := srv.Serve(ln); err != http.ErrServerClosed { errChan <- err } else { errChan <- nil diff --git a/pkg/gpg/gpg_forwarding.go b/pkg/gpg/gpg_forwarding.go index 3b8b87673..9a4e18124 100644 --- a/pkg/gpg/gpg_forwarding.go +++ b/pkg/gpg/gpg_forwarding.go @@ -195,20 +195,26 @@ func (g *GPGConf) claimForwardedSocket(ctx context.Context) error { err := wait.ExponentialBackoffWithContext(ctx, backoff, func(_ context.Context) (bool, error) { info, err := os.Stat(g.SocketPath) if err != nil { - return false, nil // Retry + if os.IsNotExist(err) { + return false, nil // Retry + } + return false, fmt.Errorf("inspect forwarded gpg socket %q: %w", g.SocketPath, err) } if info.Mode()&os.ModeSocket == 0 { return false, fmt.Errorf("path %q exists but is not a unix socket", g.SocketPath) } return true, nil }) - if err != nil { + if wait.Interrupted(err) { return fmt.Errorf( "forwarded gpg socket %q did not appear as expected: %w", g.SocketPath, err, ) } + if err != nil { + return err + } //nolint:gosec // g.SocketPath is the fixed forwarded socket path return exec.Command("sudo", "chown", owner, g.SocketPath).Run() diff --git a/pkg/gpg/gpg_forwarding_test.go b/pkg/gpg/gpg_forwarding_test.go index 5979a9060..1fe415ce9 100644 --- a/pkg/gpg/gpg_forwarding_test.go +++ b/pkg/gpg/gpg_forwarding_test.go @@ -160,6 +160,34 @@ func TestClaimForwardedSocket_RejectsNonSocketPath(t *testing.T) { ) } +func TestClaimForwardedSocket_FailsFastOnPermissionError(t *testing.T) { + if os.Getuid() == 0 { + t.Skip("root bypasses directory permission checks") + } + + parent := t.TempDir() + socketPath := filepath.Join(parent, "S.gpg-agent") + require.NoError(t, os.Chmod(parent, 0o000)) + //nolint:gosec // restore the temp dir's own perms so t.TempDir() cleanup can remove it + t.Cleanup(func() { _ = os.Chmod(parent, 0o700) }) + + g := &GPGConf{SocketPath: socketPath} + + start := time.Now() + err := g.claimForwardedSocket(context.Background()) + elapsed := time.Since(start) + + require.Error(t, err) + assert.NotContains(t, err.Error(), "did not appear", + "a permission error must not be reported as the socket never appearing") + assert.Less( + t, + elapsed, + time.Second, + "must fail immediately, not retry against a permission error", + ) +} + func TestClaimForwardedSocket_RespectsContextCancellation(t *testing.T) { g := &GPGConf{SocketPath: filepath.Join(t.TempDir(), "S.gpg-agent")} From 2ec3818ae02bcaf9503f853234d686438bd78ae2 Mon Sep 17 00:00:00 2001 From: Samuel K Date: Sun, 2 Aug 2026 13:14:17 -0500 Subject: [PATCH 07/14] fix(test): tighten error assertion and timeout margin in gpg socket tests Assert the specific *exec.ExitError/*exec.Error a tolerated sudo chown failure produces instead of excluding by error message, so an unrelated bug can no longer slip through undetected. Also widen the timeout in the never-appears test: the backoff's own jittered worst case is close to 24.4s, leaving too little margin at 25s. --- pkg/gpg/gpg_forwarding_test.go | 25 +++++++++++++++---------- 1 file changed, 15 insertions(+), 10 deletions(-) diff --git a/pkg/gpg/gpg_forwarding_test.go b/pkg/gpg/gpg_forwarding_test.go index 1fe415ce9..fdf73a335 100644 --- a/pkg/gpg/gpg_forwarding_test.go +++ b/pkg/gpg/gpg_forwarding_test.go @@ -5,6 +5,7 @@ import ( "errors" "net" "os" + "os/exec" "path/filepath" "strings" "testing" @@ -115,13 +116,16 @@ func TestClaimForwardedSocket_StopsPollingAsSoonAsSocketAppears(t *testing.T) { elapsed := time.Since(start) assert.Less(t, elapsed, 2*time.Second, "must return shortly after the socket appears") - // claimForwardedSocket's final step (chown to the caller's uid) requires - // sudo, which may not be passwordless in this environment; only the - // socket-appear wait itself is under test here, so a chown-only failure - // is tolerated, but any error about the socket not appearing is not. + // claimForwardedSocket's final step runs `sudo chown`, which may not be + // passwordless in this environment; only the socket-appear wait itself is + // under test here, so that specific failure is tolerated, but nothing + // else is: a sudo failure surfaces as *exec.ExitError (ran, non-zero + // exit) or *exec.Error (sudo binary missing). if err != nil { - assert.NotContains(t, err.Error(), "did not appear") - assert.NotContains(t, err.Error(), "not a unix socket") + var exitErr *exec.ExitError + var execErr *exec.Error + assert.True(t, errors.As(err, &exitErr) || errors.As(err, &execErr), + "only a sudo chown failure is tolerated here, got: %v", err) } } @@ -129,10 +133,11 @@ func TestClaimForwardedSocket_TimesOutWhenSocketNeverAppears(t *testing.T) { g := &GPGConf{SocketPath: filepath.Join(t.TempDir(), "S.gpg-agent")} // The backoff's own worst case (15 steps, 200ms*1.5^n capped at 2s, plus - // up to 10% jitter per step) is a bit over 20s, so the timeout used to - // wait it out must exceed that, not sit just under it, or this assertion - // itself becomes flaky on the exact path it verifies. - ctx, cancel := context.WithTimeout(context.Background(), 25*time.Second) + // up to 10% jitter per step) is close to 24.4s, so the timeout used to + // wait it out needs real margin above that or scheduler delays can make + // ctx expire first, replacing the "did not appear" error this test + // checks for with a context-deadline error instead. + ctx, cancel := context.WithTimeout(context.Background(), 35*time.Second) defer cancel() err := g.claimForwardedSocket(ctx) From d09ad8a6df100ad0801f81364d4c8862d0c4893e Mon Sep 17 00:00:00 2001 From: Samuel K Date: Sun, 2 Aug 2026 13:31:25 -0500 Subject: [PATCH 08/14] docs: trim comments and lowercase log messages Remove comments that restated code, tighten a couple that remained, and lowercase log messages for consistency. Add a build tag to gpg_forwarding_test.go for the unix-only test helpers it uses. --- .../agentcontainer/credentials_server.go | 21 +++-------------- .../agentcontainer/credentials_server_test.go | 7 ------ cmd/internal/agentworkspace/setup_gpg.go | 20 +++++----------- cmd/internal/agentworkspace/setup_gpg_test.go | 2 -- cmd/workspace/gpg_tunnel.go | 12 ++++++---- cmd/workspace/port_forward.go | 2 +- pkg/credentials/server.go | 14 ++++++----- pkg/gpg/gpg_forwarding_test.go | 23 ++----------------- pkg/log/jsonstream.go | 9 +++----- 9 files changed, 30 insertions(+), 80 deletions(-) diff --git a/cmd/internal/agentcontainer/credentials_server.go b/cmd/internal/agentcontainer/credentials_server.go index c8e5ab7b3..6d61ef04c 100644 --- a/cmd/internal/agentcontainer/credentials_server.go +++ b/cmd/internal/agentcontainer/credentials_server.go @@ -88,24 +88,15 @@ func NewCredentialsServerCmd(flags *flags.GlobalFlags) *cobra.Command { // Run runs the command logic. func (cmd *CredentialsServerCmd) Run(ctx context.Context, port int) error { - // create a grpc client tunnelClient, err := tunnelserver.NewTunnelClient(os.Stdin, os.Stdout, true, ExitCodeIO) if err != nil { return fmt.Errorf("error creating tunnel client: %w", err) } - // this message serves as a ping to the client if _, err := tunnelClient.Ping(ctx, &tunnel.Empty{}); err != nil { return fmt.Errorf("ping client: %w", err) } - // Claim the port by binding it before starting anything else, and hold - // the listener until it's handed to RunCredentialsServerWithListener - // below: a check-then-bind gap here would let two concurrent sessions - // both pass the check and race to bind. On contention this returns an - // error so RunServices' retry (pkg/tunnel/services.go) tries again - // later, and starting the port watcher beforehand would leak an - // orphaned goroutine on every failed attempt. ln, err := claimPort(port) if err != nil { return err @@ -114,27 +105,21 @@ func (cmd *CredentialsServerCmd) Run(ctx context.Context, port int) error { cmd.maybeForwardPorts(ctx, tunnelClient) - // configure docker credential helper if err := cmd.configureDockerHelper(port); err != nil { return err } - // configure git user if err := configureGitUserLocally(ctx, cmd.User, tunnelClient); err != nil { - log.Debugf("Error configuring git user: %v", err) + log.Warnf("error configuring git user: %v", err) return err } - // configure git credential helper cleanupGitHelper, err := cmd.configureGitCredentialHelper(ctx, port) if err != nil { return err } defer cleanupGitHelper() - // configure git ssh signature helper -- non-fatal so that a signing - // setup failure does not take down the entire credentials server - // (git/docker credential forwarding, port forwarding, etc.) cleanupGitSigning := cmd.configureGitSigningKey() defer cleanupGitSigning() @@ -152,7 +137,7 @@ func claimPort(port int) (net.Listener, error) { ln, err := net.Listen("tcp", addr) if err != nil { return nil, fmt.Errorf( - "port %d not available (another session likely owns the credentials server): %w", + "port %d not available (another session may own the credentials server): %w", port, err, ) @@ -168,7 +153,7 @@ func (cmd *CredentialsServerCmd) maybeForwardPorts( return } go func() { - log.Debugf("Start watching & forwarding open ports") + log.Debugf("start watching & forwarding open ports") if err := forwardPorts(ctx, tunnelClient); err != nil { log.Errorf("error forwarding ports: %v", err) } diff --git a/cmd/internal/agentcontainer/credentials_server_test.go b/cmd/internal/agentcontainer/credentials_server_test.go index d5987d3f3..a4aa77b3b 100644 --- a/cmd/internal/agentcontainer/credentials_server_test.go +++ b/cmd/internal/agentcontainer/credentials_server_test.go @@ -10,9 +10,6 @@ import ( ) func TestClaimPort_SucceedsWhenPortFree(t *testing.T) { - // Pass 0 to let the OS assign a free ephemeral port at call time — this - // avoids the close-then-probe race of reserving a port via net.Listen, - // closing it, and hoping nothing else claims it before this call runs. ln, err := claimPort(0) require.NoError(t, err) _ = ln.Close() @@ -51,10 +48,6 @@ func TestClaimPort_BecomesClaimableAfterHolderReleases(t *testing.T) { _ = claimed.Close() } -// TestClaimPort_OnlyOneConcurrentCallerWins proves claimPort binds the port -// as the claim itself, rather than merely probing it: with a check-then-bind -// gap, two concurrent callers could both observe the port as free before -// either binds it. func TestClaimPort_OnlyOneConcurrentCallerWins(t *testing.T) { ln, err := net.Listen("tcp", "localhost:0") require.NoError(t, err) diff --git a/cmd/internal/agentworkspace/setup_gpg.go b/cmd/internal/agentworkspace/setup_gpg.go index 5e2cf288f..c0c9d10ba 100644 --- a/cmd/internal/agentworkspace/setup_gpg.go +++ b/cmd/internal/agentworkspace/setup_gpg.go @@ -18,11 +18,11 @@ import ( ) // gpgSetupLockPath serializes concurrent setup-gpg invocations against the -// same container's gpg-agent/socket. A var so tests can point it at a temp path. +// same container's gpg-agent/socket. var gpgSetupLockPath = "/tmp/devsy-gpg-setup.lock" // gpgSetupLockTimeout bounds how long an invocation waits for a concurrent -// one to finish. A var so tests can shrink it. +// one to finish. var gpgSetupLockTimeout = 30 * time.Second // SetupGPGCmd holds the setupGPG cmd flags. @@ -61,17 +61,9 @@ func NewSetupGPGCmd(flags *flags.GlobalFlags) *cobra.Command { return setupGPGCmd } -// will forward a local gpg-agent into the remote container -// this works by -// -// - stopping remote gpg-agent and removing the sockets -// - exporting local public keys and owner trust -// - importing those into the container -// - ensuring the gpg-agent is stopped in the container -// - starting a reverse-tunnel of the local unix socket to remote -// - ensuring paths and permissions are correctly set in the remote. +// Run executes the setup-gpg command. func (cmd *SetupGPGCmd) Run(ctx context.Context) error { - log.Debugf("Initializing gpg-agent forwarding") + log.Debugf("initializing gpg-agent forwarding") unlock, err := acquireGPGSetupLock(ctx) if err != nil { @@ -96,9 +88,9 @@ func (cmd *SetupGPGCmd) Run(ctx context.Context) error { } if gpgConf.GitKey != "" { - log.Debugf("Setup git signing key") + log.Debugf("setup git signing key") if err := gitcredentials.SetupGpgGitKey(ctx, gpgConf.GitKey); err != nil { - log.Warnf("Setup git signing key failed (non-fatal): %v", err) + log.Warnf("setup git signing key failed (non-fatal): %v", err) } } diff --git a/cmd/internal/agentworkspace/setup_gpg_test.go b/cmd/internal/agentworkspace/setup_gpg_test.go index a1de4b3ce..93f3f30cd 100644 --- a/cmd/internal/agentworkspace/setup_gpg_test.go +++ b/cmd/internal/agentworkspace/setup_gpg_test.go @@ -77,8 +77,6 @@ func TestAcquireGPGSetupLock_TimesOutWhenHeldTooLong(t *testing.T) { func TestAcquireGPGSetupLock_ReturnsCancellationErrorWhenCallerCancels(t *testing.T) { origPath, origTimeout := gpgSetupLockPath, gpgSetupLockTimeout gpgSetupLockPath = filepath.Join(t.TempDir(), "setup-gpg.lock") - // Longer than the caller's cancellation, so a "timed out waiting" - // lock-timeout error is not what triggers here — only ctx cancellation. gpgSetupLockTimeout = 10 * time.Second defer func() { gpgSetupLockPath, gpgSetupLockTimeout = origPath, origTimeout }() diff --git a/cmd/workspace/gpg_tunnel.go b/cmd/workspace/gpg_tunnel.go index c52406478..4a76f0bb3 100644 --- a/cmd/workspace/gpg_tunnel.go +++ b/cmd/workspace/gpg_tunnel.go @@ -141,8 +141,10 @@ func (t *gpgTunnel) ensure(ctx context.Context, sshClient *ssh.Client) { writeGPGForwardFailedOSC(os.Stderr, "check logs for details") } -// setup forwards the local gpg-agent into the remote container by using -// cmd/internal/agentworkspace/setup_gpg. +// setup runs the remote setup-gpg command, which imports the host's owner trust +// and signing key into the container's gpg-agent. It also ensures the +// reverse-forward socket is bound at most once per process, so concurrent +// terminals don't race to bind the same path. func (t *gpgTunnel) setup(ctx context.Context, containerClient *ssh.Client) error { log.Debugf("detecting gpg-agent socket path on host") // this socket gets forwarded to the remote and symlinked in multiple paths @@ -150,7 +152,7 @@ func (t *gpgTunnel) setup(ctx context.Context, containerClient *ssh.Client) erro if err != nil { return err } - log.Debugf("[GPG] detected gpg-agent socket path %s", gpgExtraSocketPath) + log.Debugf("detected gpg-agent socket path %s", gpgExtraSocketPath) command, err := t.buildSetupCommand(ctx) if err != nil { @@ -183,7 +185,7 @@ func (t *gpgTunnel) setup(ctx context.Context, containerClient *ssh.Client) erro func (t *gpgTunnel) buildSetupCommand(ctx context.Context) (string, error) { cmd := t.cmd - log.Debugf("[GPG] exporting gpg owner trust from host") + log.Debugf("exporting gpg owner trust from host") ownerTrustExport, err := gpg.GetHostOwnerTrust() if err != nil { return "", fmt.Errorf("export local ownertrust from GPG: %w", err) @@ -231,7 +233,7 @@ func (t *gpgTunnel) ensureForwardBound( } log.Debugf( - "[GPG] start reverse forward of gpg-agent socket %s, keeping connection open", + "start reverse forward of gpg-agent socket %s, keeping connection open", gpgExtraSocketPath, ) reverseForwardPorts := append( diff --git a/cmd/workspace/port_forward.go b/cmd/workspace/port_forward.go index 8ed62cd56..fb36e759b 100644 --- a/cmd/workspace/port_forward.go +++ b/cmd/workspace/port_forward.go @@ -215,7 +215,7 @@ func (cmd *SSHCmd) startReverseForwardsAndWait( for _, b := range bound { log.Infof( - "Reverse forwarding local %s/%s to remote %s/%s", + "reverse forwarding local %s/%s to remote %s/%s", b.mapping.Host.Protocol, b.mapping.Host.Address, b.mapping.Container.Protocol, diff --git a/pkg/credentials/server.go b/pkg/credentials/server.go index 99b7944f8..ab0d450ef 100644 --- a/pkg/credentials/server.go +++ b/pkg/credentials/server.go @@ -69,12 +69,10 @@ func RunCredentialsServerWithListener( errChan := make(chan error, 1) go func() { log.Debugf("credentials server started: addr=%v", ln.Addr()) - - // always returns error. ErrServerClosed on graceful close if err := srv.Serve(ln); err != http.ErrServerClosed { - errChan <- err + errChan <- err // unexpected error, not a graceful shutdown } else { - errChan <- nil + errChan <- nil // graceful shutdown, no error } }() @@ -91,10 +89,14 @@ type credentialsHandlerFunc func( context.Context, http.ResponseWriter, *http.Request, CredentialsClient, ) error +// newCredentialsHandler returns an http.Handler that routes requests to the +// appropriate handler function, which calls the CredentialsClient to get the +// credentials and writes them to the response. +// +// Root is a readiness probe (see waitForServer); it must return 200 so the +// server is detected as up. Unknown paths still 404 below. func newCredentialsHandler(ctx context.Context, client CredentialsClient) http.Handler { routes := map[string]credentialsHandlerFunc{ - // Root is a readiness probe (see waitForServer); it must return 200 so the - // server is detected as up. Unknown paths still 404 below. "/": func(_ context.Context, writer http.ResponseWriter, _ *http.Request, _ CredentialsClient) error { writer.WriteHeader(http.StatusOK) return nil diff --git a/pkg/gpg/gpg_forwarding_test.go b/pkg/gpg/gpg_forwarding_test.go index fdf73a335..3f9bce892 100644 --- a/pkg/gpg/gpg_forwarding_test.go +++ b/pkg/gpg/gpg_forwarding_test.go @@ -1,3 +1,5 @@ +//go:build linux || darwin || unix + package gpg import ( @@ -15,20 +17,10 @@ import ( "github.com/stretchr/testify/require" ) -// listenUnixSocket creates a real Unix domain socket at path and returns it -// along with any error. It deliberately avoids require.NoError/t.Cleanup: -// those call t.FailNow, which the testing package requires to run on the -// test's own goroutine, but this helper is called from a background -// goroutine in TestClaimForwardedSocket_StopsPollingAsSoonAsSocketAppears. -// The caller must check the returned error and register cleanup itself, on -// the test goroutine. func listenUnixSocket(path string) (net.Listener, error) { return net.Listen("unix", path) } -// shortSocketDir returns a short-enough temp dir for a unix socket path: -// t.TempDir()'s nested path can exceed the ~104-byte sun_path limit on -// macOS/BSD, so this creates directly under os.TempDir() instead. func shortSocketDir(t *testing.T) string { t.Helper() dir, err := os.MkdirTemp("", "gpgsock") @@ -116,11 +108,6 @@ func TestClaimForwardedSocket_StopsPollingAsSoonAsSocketAppears(t *testing.T) { elapsed := time.Since(start) assert.Less(t, elapsed, 2*time.Second, "must return shortly after the socket appears") - // claimForwardedSocket's final step runs `sudo chown`, which may not be - // passwordless in this environment; only the socket-appear wait itself is - // under test here, so that specific failure is tolerated, but nothing - // else is: a sudo failure surfaces as *exec.ExitError (ran, non-zero - // exit) or *exec.Error (sudo binary missing). if err != nil { var exitErr *exec.ExitError var execErr *exec.Error @@ -131,12 +118,6 @@ func TestClaimForwardedSocket_StopsPollingAsSoonAsSocketAppears(t *testing.T) { func TestClaimForwardedSocket_TimesOutWhenSocketNeverAppears(t *testing.T) { g := &GPGConf{SocketPath: filepath.Join(t.TempDir(), "S.gpg-agent")} - - // The backoff's own worst case (15 steps, 200ms*1.5^n capped at 2s, plus - // up to 10% jitter per step) is close to 24.4s, so the timeout used to - // wait it out needs real margin above that or scheduler delays can make - // ctx expire first, replacing the "did not appear" error this test - // checks for with a context-deadline error instead. ctx, cancel := context.WithTimeout(context.Background(), 35*time.Second) defer cancel() diff --git a/pkg/log/jsonstream.go b/pkg/log/jsonstream.go index 878bc5c97..910f8021f 100644 --- a/pkg/log/jsonstream.go +++ b/pkg/log/jsonstream.go @@ -26,8 +26,8 @@ func PipeJSONStream() (io.WriteCloser, chan struct{}) { // valid JSON is written verbatim (with a trailing newline) to fallback // instead of being silently dropped. Use this when the writer's lifetime // spans more than one process/phase and only some of them are known to -// produce structured JSON — e.g. a shell script's plain-text stderr -// followed by a devsy subcommand's JSON logs on the same stream. +// produce structured JSON (e.g. a shell script's plain-text stderr +// followed by a devsy subcommand's JSON logs on the same stream). func PipeJSONStreamWithFallback(fallback io.Writer) (io.WriteCloser, chan struct{}) { done := make(chan struct{}) reader, writer := io.Pipe() @@ -44,8 +44,6 @@ func PipeJSONStreamWithFallback(fallback io.Writer) (io.WriteCloser, chan struct return writer, done } -// jsonLine is a self-contained representation of a JSON log line, -// replacing the old github.com/devsy-org/log.Line type. type jsonLine struct { Message string `json:"message,omitempty"` Msg string `json:"msg,omitempty"` @@ -76,8 +74,7 @@ func ReadJSONStream(reader io.Reader) { // readJSONStreamWithFallback parses reader line by line as JSON log lines. // A line that isn't valid JSON, or whose level isn't recognized, is written -// to fallback verbatim if fallback is non-nil; otherwise it's dropped, -// matching ReadJSONStream's existing behavior for genuinely-JSON-only streams. +// to fallback verbatim if fallback is non-nil; otherwise it's dropped. func readJSONStreamWithFallback(reader io.Reader, fallback io.Writer) { scan := scanner.NewScanner(reader) for scan.Scan() { From c23196c69f47618a06950111c826a580291037e2 Mon Sep 17 00:00:00 2001 From: Samuel K Date: Sun, 2 Aug 2026 13:56:14 -0500 Subject: [PATCH 09/14] fix(credentials-server): scope the port-forwarding goroutine to Run's own lifetime Run passed its caller-owned ctx straight to maybeForwardPorts, so an early return from docker/git setup didn't stop the forwarding goroutine it started. Derive a child context, cancelled via defer, and use it for every step in Run so an early return tears down what it started. --- .../agentcontainer/credentials_server.go | 16 +++++++++++----- 1 file changed, 11 insertions(+), 5 deletions(-) diff --git a/cmd/internal/agentcontainer/credentials_server.go b/cmd/internal/agentcontainer/credentials_server.go index 6d61ef04c..cdcb35063 100644 --- a/cmd/internal/agentcontainer/credentials_server.go +++ b/cmd/internal/agentcontainer/credentials_server.go @@ -88,12 +88,18 @@ func NewCredentialsServerCmd(flags *flags.GlobalFlags) *cobra.Command { // Run runs the command logic. func (cmd *CredentialsServerCmd) Run(ctx context.Context, port int) error { + // Own a child of ctx for the lifetime of this call: an early return below + // (docker/git setup failing) must stop the port-forwarding goroutine + // started by maybeForwardPorts, not leave it running past this Run call. + runCtx, cancel := context.WithCancel(ctx) + defer cancel() + tunnelClient, err := tunnelserver.NewTunnelClient(os.Stdin, os.Stdout, true, ExitCodeIO) if err != nil { return fmt.Errorf("error creating tunnel client: %w", err) } - if _, err := tunnelClient.Ping(ctx, &tunnel.Empty{}); err != nil { + if _, err := tunnelClient.Ping(runCtx, &tunnel.Empty{}); err != nil { return fmt.Errorf("ping client: %w", err) } @@ -103,18 +109,18 @@ func (cmd *CredentialsServerCmd) Run(ctx context.Context, port int) error { } defer func() { _ = ln.Close() }() - cmd.maybeForwardPorts(ctx, tunnelClient) + cmd.maybeForwardPorts(runCtx, tunnelClient) if err := cmd.configureDockerHelper(port); err != nil { return err } - if err := configureGitUserLocally(ctx, cmd.User, tunnelClient); err != nil { + if err := configureGitUserLocally(runCtx, cmd.User, tunnelClient); err != nil { log.Warnf("error configuring git user: %v", err) return err } - cleanupGitHelper, err := cmd.configureGitCredentialHelper(ctx, port) + cleanupGitHelper, err := cmd.configureGitCredentialHelper(runCtx, port) if err != nil { return err } @@ -123,7 +129,7 @@ func (cmd *CredentialsServerCmd) Run(ctx context.Context, port int) error { cleanupGitSigning := cmd.configureGitSigningKey() defer cleanupGitSigning() - return credentials.RunCredentialsServerWithListener(ctx, ln, tunnelClient) + return credentials.RunCredentialsServerWithListener(runCtx, ln, tunnelClient) } // claimPort binds port and returns the listener, holding it exclusively so From 66f0ef4afa7a2b04d5ddfe7ef1ab52ad23dd22af Mon Sep 17 00:00:00 2001 From: Samuel K Date: Sun, 2 Aug 2026 14:00:04 -0500 Subject: [PATCH 10/14] style: update comments Signed-off-by: Samuel K --- cmd/internal/agentcontainer/credentials_server.go | 14 +------------- 1 file changed, 1 insertion(+), 13 deletions(-) diff --git a/cmd/internal/agentcontainer/credentials_server.go b/cmd/internal/agentcontainer/credentials_server.go index cdcb35063..64ea47125 100644 --- a/cmd/internal/agentcontainer/credentials_server.go +++ b/cmd/internal/agentcontainer/credentials_server.go @@ -88,9 +88,6 @@ func NewCredentialsServerCmd(flags *flags.GlobalFlags) *cobra.Command { // Run runs the command logic. func (cmd *CredentialsServerCmd) Run(ctx context.Context, port int) error { - // Own a child of ctx for the lifetime of this call: an early return below - // (docker/git setup failing) must stop the port-forwarding goroutine - // started by maybeForwardPorts, not leave it running past this Run call. runCtx, cancel := context.WithCancel(ctx) defer cancel() @@ -135,9 +132,7 @@ func (cmd *CredentialsServerCmd) Run(ctx context.Context, port int) error { // claimPort binds port and returns the listener, holding it exclusively so // no other session can bind the same port until the caller closes it (or // hands it to RunCredentialsServerWithListener). Only one session's -// credentials-server can hold this port at a time. Returning an error (not -// nil) on contention matters: RunServices (pkg/tunnel/services.go) wraps -// this command in retry.OnError, which only retries on a non-nil error. +// credentials-server can hold this port at a time. func claimPort(port int) (net.Listener, error) { addr := net.JoinHostPort("localhost", strconv.Itoa(port)) ln, err := net.Listen("tcp", addr) @@ -190,9 +185,6 @@ func (cmd *CredentialsServerCmd) configureGitCredentialHelper( return noop, fmt.Errorf("configure git helper: %w", err) } - // cleanup when we are done. This defer runs after the server loop - // returns on shutdown, when ctx is already canceled — use an uncanceled - // context so the helper is actually removed instead of aborting early. cleanupCtx := context.WithoutCancel(ctx) userName := cmd.User return func() { @@ -231,7 +223,6 @@ func configureGitUserLocally( userName string, client tunnel.TunnelClient, ) error { - // get local credentials localGitUser, err := gitcredentials.GetUser(ctx, userName, "") if err != nil { return err @@ -240,16 +231,13 @@ func configureGitUserLocally( return nil } - // set user & email if not found gitUser, err := fetchRemoteGitUser(ctx, client) if err != nil { return err } - // don't override what is already there clearKnownGitUserFields(localGitUser, gitUser) - // set git user if err := gitcredentials.SetUser(ctx, userName, gitUser); err != nil { return fmt.Errorf("set git user & email: %w", err) } From 89f140f959a99a495b394f6867a41734af5147c6 Mon Sep 17 00:00:00 2001 From: Samuel K Date: Sun, 2 Aug 2026 14:49:03 -0500 Subject: [PATCH 11/14] fix(ssh): make initial GPG tunnel setup synchronous; fix double-logged JSON in sshtunnel MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CI failure: e2e "expose the host GPG secret key" flaked with "no gpg-agent running in this session". Root cause: the ssh.go refactor changed the initial GPG-agent setup from blocking (as it was on main) to fully async — runGPGTunnelInBackground fired the tunnel's first setup in a goroutine and returned immediately, so the SSH command it started would often run before the reverse-forward and remote setup-gpg had finished. Fix: run the first ensure synchronously before starting the periodic health-check goroutine, so the tunnel is up before the SSH command runs; only the recurring health checks stay async. Also fix: executeSSHServerHelper wrote each incoming child-process stderr line as the message of a new INFO-level log entry, so structured JSON logs from the injected `ssh-server` command showed up wrapped inside another log line. It already has a purpose-built TunnelLogStreamer for this in the same file (JSON-aware, with a plain-text level-prefix fallback) that just wasn't being used here. Verified against Docker with the e2e GPG-forwarding suite (3 consecutive passing runs). --- cmd/workspace/gpg_tunnel.go | 24 +++++++++++++++--------- pkg/devcontainer/sshtunnel/sshtunnel.go | 12 +++++++++--- 2 files changed, 24 insertions(+), 12 deletions(-) diff --git a/cmd/workspace/gpg_tunnel.go b/cmd/workspace/gpg_tunnel.go index 4a76f0bb3..7b7023e2d 100644 --- a/cmd/workspace/gpg_tunnel.go +++ b/cmd/workspace/gpg_tunnel.go @@ -78,14 +78,14 @@ func newGPGTunnel(cmd *SSHCmd, devsyConfig *config.Config) *gpgTunnel { // run watches the tunnel for the life of ctx, (re-)establishing it whenever // it's found down. Call this in a goroutine tied to a context that's // cancelled as soon as the owning SSH session ends (see -// runGPGTunnelInBackground). +// runGPGTunnelInBackground). The first ensure runs synchronously before this +// goroutine starts (see runGPGTunnelInBackground), so this loop only needs +// to handle the periodic re-checks. func (t *gpgTunnel) run(ctx context.Context, sshClient *ssh.Client) { if !t.enabled { return } - t.ensure(ctx, sshClient) - ticker := time.NewTicker(gpgTunnelHealthCheckInterval) defer ticker.Stop() for { @@ -248,17 +248,23 @@ func (t *gpgTunnel) ensureForwardBound( return nil } -// runGPGTunnelInBackground starts t.run in a goroutine tied to a context -// derived from ctx, and returns a wait func that cancels that context and -// blocks until the goroutine exits. Callers defer the wait func immediately -// after starting the tunnel, so a session that returns while the tunnel's -// health-check loop is mid-tick doesn't block on the session's own (often -// much longer-lived) ctx. +// runGPGTunnelInBackground runs the tunnel's first setup synchronously, so +// the SSH command that follows doesn't race a still-forwarding gpg-agent, +// then starts t.run's periodic health-check loop in a goroutine tied to a +// context derived from ctx. It returns a wait func that cancels that context +// and blocks until the goroutine exits. Callers defer the wait func +// immediately after starting the tunnel, so a session that returns while the +// tunnel's health-check loop is mid-tick doesn't block on the session's own +// (often much longer-lived) ctx. func runGPGTunnelInBackground( ctx context.Context, t *gpgTunnel, sshClient *ssh.Client, ) (wait func()) { + if t.enabled { + t.ensure(ctx, sshClient) + } + tunnelCtx, cancel := context.WithCancel(ctx) done := make(chan struct{}) go func() { diff --git a/pkg/devcontainer/sshtunnel/sshtunnel.go b/pkg/devcontainer/sshtunnel/sshtunnel.go index 39ef2b44f..21b0af4c8 100644 --- a/pkg/devcontainer/sshtunnel/sshtunnel.go +++ b/pkg/devcontainer/sshtunnel/sshtunnel.go @@ -98,11 +98,17 @@ func executeSSHServerHelper( ) error { defer log.Debug("done executing SSH server helper command") - writer := log.Writer(log.LevelInfo) - defer func() { _ = writer.Close() }() + // AgentInject's stderr always carries the ssh-server command's structured + // JSON logs, and for some callers (see newAgentInjectFunc) also the + // injection script's plain-text preamble first. TunnelLogStreamer handles + // both: JSON lines are re-emitted at their own level, plain-text lines + // fall back to level-prefix extraction, so neither is silently dropped or + // double-wrapped as a single log line. + streamer := NewTunnelLogStreamer() + defer func() { _ = streamer.Close() }() log.Debugf("injecting and running SSH server command: %q", opts.SSHCommand) - err := opts.AgentInject(ctx, opts.SSHCommand, stdin, stdout, writer) + err := opts.AgentInject(ctx, opts.SSHCommand, stdin, stdout, streamer) if err != nil && !isExpectedError(err) { return fmt.Errorf("executing agent command: %w", err) } From 6d9f3945bbf811ce51708d16d50dcfbc5d6b2909 Mon Sep 17 00:00:00 2001 From: Samuel K Date: Sun, 2 Aug 2026 15:47:02 -0500 Subject: [PATCH 12/14] fix(review): bound credentials-server request lifetime, deflake oversized-write test Add ReadTimeout and IdleTimeout to the credentials server's http.Server alongside the existing ReadHeaderTimeout, so a slow request body or an idle keep-alive connection can't hold the listener indefinitely. In the jsonstream test, wrap the initial oversized write in its own timeout the same way the later write is, so a blocked write (were the fix regressed) fails the test via t.Fatal instead of hanging past the overall test timeout. --- pkg/credentials/server.go | 2 ++ pkg/log/jsonstream_test.go | 12 +++++++++++- 2 files changed, 13 insertions(+), 1 deletion(-) diff --git a/pkg/credentials/server.go b/pkg/credentials/server.go index ab0d450ef..9858e2a86 100644 --- a/pkg/credentials/server.go +++ b/pkg/credentials/server.go @@ -64,6 +64,8 @@ func RunCredentialsServerWithListener( srv := &http.Server{ Handler: newCredentialsHandler(ctx, client), ReadHeaderTimeout: 10 * time.Second, + ReadTimeout: 30 * time.Second, + IdleTimeout: 120 * time.Second, } errChan := make(chan error, 1) diff --git a/pkg/log/jsonstream_test.go b/pkg/log/jsonstream_test.go index 6a06ec45f..2931eb363 100644 --- a/pkg/log/jsonstream_test.go +++ b/pkg/log/jsonstream_test.go @@ -14,7 +14,17 @@ func TestPipeJSONStreamWithFallback_OversizedLineUnblocksWriter(t *testing.T) { writer, done := PipeJSONStreamWithFallback(PassthroughWriter()) oversized := strings.Repeat("a", 2*1024*1024) + "\n" - _, _ = writer.Write([]byte(oversized)) + firstWriteDone := make(chan struct{}) + go func() { + defer close(firstWriteDone) + _, _ = writer.Write([]byte(oversized)) + }() + + select { + case <-firstWriteDone: + case <-time.After(2 * time.Second): + t.Fatal("initial write blocked instead of being drained by the scanner") + } select { case <-done: From 645c0b99f0a55c9f9ad4b22e57249625bd1ca227 Mon Sep 17 00:00:00 2001 From: Samuel K Date: Sun, 2 Aug 2026 16:52:54 -0500 Subject: [PATCH 13/14] style: clean comments --- .../src/lib/components/terminal/Terminal.svelte | 4 ---- pkg/tunnel/container.go | 10 ---------- 2 files changed, 14 deletions(-) diff --git a/desktop/src/renderer/src/lib/components/terminal/Terminal.svelte b/desktop/src/renderer/src/lib/components/terminal/Terminal.svelte index ac6250a2e..2d84501a9 100644 --- a/desktop/src/renderer/src/lib/components/terminal/Terminal.svelte +++ b/desktop/src/renderer/src/lib/components/terminal/Terminal.svelte @@ -178,10 +178,6 @@ onMount(async () => { unsubscribeTheme: undefined, onGpgForwardFailed, } - // Register before flushing outputBuffer below: term.write drives the OSC - // parser synchronously, so a failure notification buffered during the - // async import window above would otherwise be parsed and discarded - // before this handler exists to catch it. const oscHandler = term.parser.registerOscHandler( GPG_FORWARD_FAILED_OSC, (data) => { diff --git a/pkg/tunnel/container.go b/pkg/tunnel/container.go index f76703b4d..37d0a4298 100644 --- a/pkg/tunnel/container.go +++ b/pkg/tunnel/container.go @@ -91,12 +91,6 @@ func (c *ContainerTunnel) runHostTunnel( stdinReader, stdoutWriter *os.File, timeout time.Duration, ) error { - // This stderr spans two phases: InjectAgent's own shell script (plain - // text) and, once injection succeeds, the actual `devsy internal - // ssh-server --stdio` invocation (always structured JSON). Use the - // fallback variant so the script's plain-text errors aren't silently - // dropped by the JSON-only parser while JSON lines still get re-emitted - // at their original level instead of double-wrapped. writer, done := log.PipeJSONStreamWithFallback(log.PassthroughWriter()) defer func() { _ = writer.Close() @@ -241,11 +235,7 @@ type containerTunnelOpts struct { // runContainerTunnel runs the container tunnel SSH command. It closes // stdoutWriter on exit so StdioClient gets EOF when the tunnel dies. -// Context-cancelled errors are suppressed (expected during normal shutdown). func (c *ContainerTunnel) runContainerTunnel(ctx context.Context, opts containerTunnelOpts) error { - // Unlike runHostTunnel, this stderr only ever carries the `devsy internal - // agent container-tunnel` command's own JSON logs (no injection-script - // phase), so the plain PipeJSONStream is correct here. writer, done := log.PipeJSONStream() defer func() { _ = writer.Close() From 4eb3ca6c30d1ef92dbecc7c75763b638c343e37b Mon Sep 17 00:00:00 2001 From: Samuel K Date: Sun, 2 Aug 2026 16:53:05 -0500 Subject: [PATCH 14/14] style: clean comments --- cmd/workspace/gpg_tunnel.go | 16 +++++++--------- cmd/workspace/port_forward.go | 4 +--- pkg/log/jsonstream.go | 6 +++--- pkg/log/jsonstream_test.go | 4 ---- 4 files changed, 11 insertions(+), 19 deletions(-) diff --git a/cmd/workspace/gpg_tunnel.go b/cmd/workspace/gpg_tunnel.go index 7b7023e2d..91f0484f4 100644 --- a/cmd/workspace/gpg_tunnel.go +++ b/cmd/workspace/gpg_tunnel.go @@ -101,9 +101,9 @@ func (t *gpgTunnel) run(ctx context.Context, sshClient *ssh.Client) { // ensure checks whether the GPG tunnel is currently live and, if not, // (re-)establishes it. A setup failure is only reported if the tunnel is // still down immediately after: a concurrent terminal may have won the bind -// race in the interim, which isn't a real failure. The OSC failure -// notification fires only on the healthy-to-failed transition (gated by -// failureReported), not on every health-check tick. +// race in the interim, which is not a real failure. The OSC failure +// notification fires only on the healthy-to-failed transition, not on +// every health-check tick. func (t *gpgTunnel) ensure(ctx context.Context, sshClient *ssh.Client) { if gpg.IsGpgTunnelRunning(ctx, t.cmd.User, sshClient) { log.Debugf("GPG tunnel is running, skipping setup") @@ -220,9 +220,7 @@ func (t *gpgTunnel) buildSetupCommand(ctx context.Context) (string, error) { } // ensureForwardBound binds the reverse-listen socket at most once per -// process (see forwardBound); setup's remote command still re-runs every -// time to repair remote-side agent state (stopped agent, stale keys), which -// doesn't require touching the reverse-forward at all. +// process. func (t *gpgTunnel) ensureForwardBound( ctx context.Context, containerClient *ssh.Client, @@ -249,13 +247,13 @@ func (t *gpgTunnel) ensureForwardBound( } // runGPGTunnelInBackground runs the tunnel's first setup synchronously, so -// the SSH command that follows doesn't race a still-forwarding gpg-agent, +// the SSH command that follows does not race a still-forwarding gpg-agent, // then starts t.run's periodic health-check loop in a goroutine tied to a // context derived from ctx. It returns a wait func that cancels that context // and blocks until the goroutine exits. Callers defer the wait func // immediately after starting the tunnel, so a session that returns while the -// tunnel's health-check loop is mid-tick doesn't block on the session's own -// (often much longer-lived) ctx. +// tunnel's health-check loop is mid-tick does not block on the session's own +// ctx. func runGPGTunnelInBackground( ctx context.Context, t *gpgTunnel, diff --git a/cmd/workspace/port_forward.go b/cmd/workspace/port_forward.go index fb36e759b..04368a4bb 100644 --- a/cmd/workspace/port_forward.go +++ b/cmd/workspace/port_forward.go @@ -30,9 +30,7 @@ func (cmd *SSHCmd) forwardTimeout() (time.Duration, error) { // forwardPortsIfRequested handles -L/-R forwarding when requested. The returned // bool reports whether forwarding took over (the caller should return err). -// -L and -R together aren't supported: each takes over the whole session -// (never returning to hand off to the shell), so running both would mean -// silently dropping one — this reports an explicit error instead. +// -L and -R together are not supported. func (cmd *SSHCmd) forwardPortsIfRequested( ctx context.Context, sshClient *ssh.Client, diff --git a/pkg/log/jsonstream.go b/pkg/log/jsonstream.go index 910f8021f..5d497816b 100644 --- a/pkg/log/jsonstream.go +++ b/pkg/log/jsonstream.go @@ -13,8 +13,8 @@ func PipeJSONStream() (io.WriteCloser, chan struct{}) { reader, writer := io.Pipe() go func() { ReadJSONStream(reader) - // See PipeJSONStreamWithFallback: closing here unblocks a Write on - // writer if the scanner stopped early instead of at pipe close. + // closing here unblocks a Write on writer if the scanner + // stopped early instead of at pipe close. _ = reader.Close() close(done) }() @@ -22,7 +22,7 @@ func PipeJSONStream() (io.WriteCloser, chan struct{}) { return writer, done } -// PipeJSONStreamWithFallback is like PipeJSONStream, but a line that isn't +// PipeJSONStreamWithFallback is like PipeJSONStream, but a line that is not // valid JSON is written verbatim (with a trailing newline) to fallback // instead of being silently dropped. Use this when the writer's lifetime // spans more than one process/phase and only some of them are known to diff --git a/pkg/log/jsonstream_test.go b/pkg/log/jsonstream_test.go index 2931eb363..495270dba 100644 --- a/pkg/log/jsonstream_test.go +++ b/pkg/log/jsonstream_test.go @@ -6,10 +6,6 @@ import ( "time" ) -// TestPipeJSONStreamWithFallback_OversizedLineUnblocksWriter verifies that a -// line exceeding the scanner's buffer cap doesn't leave the pipe reader open -// after the goroutine exits: a later Write must fail promptly instead of -// blocking forever with nothing left draining the pipe. func TestPipeJSONStreamWithFallback_OversizedLineUnblocksWriter(t *testing.T) { writer, done := PipeJSONStreamWithFallback(PassthroughWriter())