diff --git a/cmd/internal/agentcontainer/credentials_server.go b/cmd/internal/agentcontainer/credentials_server.go index e00b76230..64ea47125 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" ) @@ -89,50 +88,62 @@ 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 + 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) } - // this message serves as a ping to the client - 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) } - 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 + ln, err := claimPort(port) + if err != nil { + return err } + defer func() { _ = ln.Close() }() + + cmd.maybeForwardPorts(runCtx, 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) + if err := configureGitUserLocally(runCtx, cmd.User, tunnelClient); err != nil { + log.Warnf("error configuring git user: %v", err) return err } - // configure git credential helper - cleanupGitHelper, err := cmd.configureGitCredentialHelper(ctx, port) + cleanupGitHelper, err := cmd.configureGitCredentialHelper(runCtx, 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() - return credentials.RunCredentialsServer(ctx, port, tunnelClient) + return credentials.RunCredentialsServerWithListener(runCtx, ln, tunnelClient) +} + +// 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. +func claimPort(port int) (net.Listener, error) { + addr := net.JoinHostPort("localhost", strconv.Itoa(port)) + ln, err := net.Listen("tcp", addr) + if err != nil { + return nil, fmt.Errorf( + "port %d not available (another session may own the credentials server): %w", + port, + err, + ) + } + return ln, nil } func (cmd *CredentialsServerCmd) maybeForwardPorts( @@ -143,7 +154,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) } @@ -174,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() { @@ -215,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 @@ -224,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) } diff --git a/cmd/internal/agentcontainer/credentials_server_test.go b/cmd/internal/agentcontainer/credentials_server_test.go new file mode 100644 index 000000000..a4aa77b3b --- /dev/null +++ b/cmd/internal/agentcontainer/credentials_server_test.go @@ -0,0 +1,81 @@ +package agentcontainer + +import ( + "net" + "sync" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestClaimPort_SucceedsWhenPortFree(t *testing.T) { + ln, err := claimPort(0) + require.NoError(t, err) + _ = ln.Close() +} + +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 = claimPort(port) + require.Error(t, err) + assert.Contains(t, err.Error(), "not available") +} + +func TestClaimPort_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 + + _, err = claimPort(port) + require.Error(t, err, "port must read as unavailable while the listener is held") + + require.NoError(t, ln.Close()) + closed = true + + claimed, err := claimPort(port) + require.NoError(t, err, "port must read as claimable once the prior holder releases it") + _ = claimed.Close() +} + +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.go b/cmd/internal/agentworkspace/setup_gpg.go index a8bb279c7..c0c9d10ba 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. +var gpgSetupLockPath = "/tmp/devsy-gpg-setup.lock" + +// gpgSetupLockTimeout bounds how long an invocation waits for a concurrent +// one to finish. +var gpgSetupLockTimeout = 30 * time.Second + // SetupGPGCmd holds the setupGPG cmd flags. type SetupGPGCmd struct { *flags.GlobalFlags @@ -51,17 +61,15 @@ 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 { + return err + } + defer unlock() publicKey, ownerTrust, err := fetchAndDecodeKeys(cmd.OwnerTrust) if err != nil { @@ -75,20 +83,47 @@ 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 } 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) } } 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 +146,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 +175,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..93f3f30cd --- /dev/null +++ b/cmd/internal/agentworkspace/setup_gpg_test.go @@ -0,0 +1,103 @@ +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() + + reacquire, err := acquireGPGSetupLock(context.Background()) + require.NoError(t, err, "unlock must actually release the lock") + reacquire() +} + +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) + + const releaseDelay = 200 * time.Millisecond + start := time.Now() + releasing := make(chan struct{}) + go func() { + close(releasing) + time.Sleep(releaseDelay) + _ = holder.Unlock() + }() + <-releasing + + unlock, err := acquireGPGSetupLock(context.Background()) + elapsed := time.Since(start) + + require.NoError(t, err) + defer unlock() + assert.GreaterOrEqual(t, elapsed, releaseDelay/2, + "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") + 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..91f0484f4 --- /dev/null +++ b/cmd/workspace/gpg_tunnel.go @@ -0,0 +1,276 @@ +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; 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). 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 + } + + 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 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") + 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 + // 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 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 + gpgExtraSocketPath, err := gpg.DetectAgentSocketPath() + if err != nil { + return err + } + log.Debugf("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("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) + + 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), gitKey) + } + + command := shellescape.QuoteCommand(forwardAgent) + if cmd.User != "" && cmd.User != "root" { + command = shellescape.QuoteCommand([]string{"su", "-c", command, cmd.User}) + } + return command, nil +} + +// ensureForwardBound binds the reverse-listen socket at most once per +// process. +func (t *gpgTunnel) ensureForwardBound( + ctx context.Context, + containerClient *ssh.Client, + gpgExtraSocketPath string, +) error { + if t.forwardBound { + return nil + } + + log.Debugf( + "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("start gpg-agent reverse forward: %w", err) + } + t.forwardBound = true + return nil +} + +// runGPGTunnelInBackground runs the tunnel's first setup synchronously, so +// 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 does not block on the session's own +// 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() { + 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..209fb324d --- /dev/null +++ b/cmd/workspace/gpg_tunnel_test.go @@ -0,0 +1,80 @@ +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) + } +} + +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 new file mode 100644 index 000000000..04368a4bb --- /dev/null +++ b/cmd/workspace/port_forward.go @@ -0,0 +1,275 @@ +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). +// -L and -R together are not supported. +func (cmd *SSHCmd) forwardPortsIfRequested( + ctx context.Context, + sshClient *ssh.Client, +) (bool, error) { + 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 hasReverse { + return true, cmd.reverseForwardPorts(ctx, sshClient) + } + return false, nil +} + +// 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 +} + +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, + ) + }, + } + 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, + ) + }, + } +) + +// 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 err + } + return runPortForwards(ctx, containerClient, portForwardRun{ + portMappings: cmd.ForwardPorts, + timeout: timeout, + dir: directionForward, + }) +} + +func (cmd *SSHCmd) reverseForwardPorts(ctx context.Context, containerClient *ssh.Client) error { + timeout, err := cmd.forwardTimeout() + if err != nil { + return err + } + return runPortForwards(ctx, containerClient, portForwardRun{ + portMappings: cmd.ReverseForwardPorts, + timeout: timeout, + dir: directionReverse, + }) +} + +// runPortForwards starts one forwarding goroutine per mapping and blocks +// 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 { + 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) + } + + log.Infof( + "%s local %s/%s to remote %s/%s", + run.dir.logPrefix, + mapping.Host.Protocol, + mapping.Host.Address, + mapping.Container.Protocol, + mapping.Container.Address, + ) + 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 + } + case <-ctx.Done(): + return ctx.Err() + } + } + return nil +} + +// 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 { + 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..162374954 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,20 @@ 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, - ) + + opts := 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, + } + go cmd.startServices(ctx, devsyConfig, containerClient, workspaceClient.WorkspaceConfig(), opts) } func (cmd *SSHCmd) buildSSHServerCommand(workdir string) string { @@ -776,13 +538,19 @@ func resolveMergedWorkspaceFolder( return result.MergedConfig.WorkspaceFolder } +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..2d84501a9 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(() => { @@ -163,6 +170,23 @@ onMount(async () => { term.open(containerEl) fitAddon.fit() + const instance: TerminalInstance = { + term, + fitAddon, + unlistenOutput, + unlistenExit, + unsubscribeTheme: undefined, + onGpgForwardFailed, + } + const oscHandler = term.parser.registerOscHandler( + GPG_FORWARD_FAILED_OSC, + (data) => { + instance.onGpgForwardFailed?.(data) + return true + }, + ) + instance.disposeOscHandler = () => oscHandler.dispose() + // Flush any output that arrived during async imports for (const data of outputBuffer) { term.write(data) @@ -173,19 +197,13 @@ onMount(async () => { terminalWrite(sessionId, Array.from(encoded)) }) - const unsubscribeTheme = theme.subscribe(() => { + instance.unsubscribeTheme = theme.subscribe(() => { if (term) { term.options.theme = resolveTheme() } }) - setTerminalInstance(sessionId, { - term, - fitAddon, - unlistenOutput, - unlistenExit, - unsubscribeTheme, - }) + 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/credentials/server.go b/pkg/credentials/server.go index 5e3c37377..9858e2a86 100644 --- a/pkg/credentials/server.go +++ b/pkg/credentials/server.go @@ -45,22 +45,36 @@ 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, + ReadTimeout: 30 * time.Second, + IdleTimeout: 120 * time.Second, } errChan := make(chan error, 1) go func() { - log.Debugf("credentials server started: port=%v", port) - - // always returns error. ErrServerClosed on graceful close - if err := srv.ListenAndServe(); err != http.ErrServerClosed { - errChan <- err + log.Debugf("credentials server started: addr=%v", ln.Addr()) + if err := srv.Serve(ln); err != http.ErrServerClosed { + errChan <- err // unexpected error, not a graceful shutdown } else { - errChan <- nil + errChan <- nil // graceful shutdown, no error } }() @@ -77,10 +91,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/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) } diff --git a/pkg/gpg/gpg_forwarding.go b/pkg/gpg/gpg_forwarding.go index 94a7fce39..9a4e18124 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,49 @@ 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 { + if os.IsNotExist(err) { + return false, nil // Retry + } + return false, fmt.Errorf("inspect forwarded gpg socket %q: %w", g.SocketPath, err) } - time.Sleep(100 * time.Millisecond) + 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 wait.Interrupted(err) { + 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) + if err != nil { + return err + } + + //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..3f9bce892 100644 --- a/pkg/gpg/gpg_forwarding_test.go +++ b/pkg/gpg/gpg_forwarding_test.go @@ -1,15 +1,34 @@ +//go:build linux || darwin || unix + package gpg import ( + "context" + "errors" + "net" "os" + "os/exec" "path/filepath" "strings" "testing" + "time" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) +func listenUnixSocket(path string) (net.Listener, error) { + return net.Listen("unix", path) +} + +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 +85,111 @@ func TestSetupGpgConf_ExistingFileWithoutTrailingNewline(t *testing.T) { } } +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) + listener, listenErr = listenUnixSocket(socketPath) + }() + defer func() { + <-listenerReady + require.NoError(t, listenErr) + _ = listener.Close() + }() + + 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") + if err != nil { + 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) + } +} + +func TestClaimForwardedSocket_TimesOutWhenSocketNeverAppears(t *testing.T) { + g := &GPGConf{SocketPath: filepath.Join(t.TempDir(), "S.gpg-agent")} + ctx, cancel := context.WithTimeout(context.Background(), 35*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_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")} + + 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/log/jsonstream.go b/pkg/log/jsonstream.go index 47a707032..5d497816b 100644 --- a/pkg/log/jsonstream.go +++ b/pkg/log/jsonstream.go @@ -13,14 +13,37 @@ func PipeJSONStream() (io.WriteCloser, chan struct{}) { reader, writer := io.Pipe() go func() { ReadJSONStream(reader) + // closing here unblocks a Write on writer if the scanner + // stopped early instead of at pipe close. + _ = reader.Close() + close(done) + }() + + return writer, done +} + +// 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 +// 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) + // 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) }() 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"` @@ -46,6 +69,13 @@ 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. +func readJSONStreamWithFallback(reader io.Reader, fallback io.Writer) { scan := scanner.NewScanner(reader) for scan.Scan() { line := scan.Bytes() @@ -54,14 +84,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/log/jsonstream_test.go b/pkg/log/jsonstream_test.go new file mode 100644 index 000000000..495270dba --- /dev/null +++ b/pkg/log/jsonstream_test.go @@ -0,0 +1,45 @@ +package log + +import ( + "strings" + "testing" + "time" +) + +func TestPipeJSONStreamWithFallback_OversizedLineUnblocksWriter(t *testing.T) { + writer, done := PipeJSONStreamWithFallback(PassthroughWriter()) + + oversized := strings.Repeat("a", 2*1024*1024) + "\n" + 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: + 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") + } +} diff --git a/pkg/tunnel/container.go b/pkg/tunnel/container.go index 5cf6175cf..37d0a4298 100644 --- a/pkg/tunnel/container.go +++ b/pkg/tunnel/container.go @@ -91,8 +91,11 @@ func (c *ContainerTunnel) runHostTunnel( stdinReader, stdoutWriter *os.File, timeout time.Duration, ) error { - writer := log.Writer(log.LevelInfo) - defer func() { _ = writer.Close() }() + writer, done := log.PipeJSONStreamWithFallback(log.PassthroughWriter()) + defer func() { + _ = writer.Close() + <-done + }() defer log.Debugf("Tunnel to host closed") command := fmt.Sprintf("'%s' internal ssh-server --stdio", c.client.AgentPath()) @@ -232,10 +235,12 @@ 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 { - writer := log.Writer(log.LevelInfo) - defer func() { _ = writer.Close() }() + writer, done := log.PipeJSONStream() + defer func() { + _ = writer.Close() + <-done + }() defer func() { _ = opts.stdoutWriter.Close() }() log.Debugf("Run container tunnel")