fix(ssh): make GPG-agent forwarding survive terminal disconnects

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)
 	}

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()
+	}
+}

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)
 	}