fix(agentcontainer,credentials): fix credentials-server port race noise, distinguish cross-user collisions

.github/workflows/pr-ci.yml

@@ -275,6 +275,30 @@ jobs:
             install-kind: false
             requires-secret: false
 
+          - label: ssh-agent-forward
+            runner: ubuntu-latest
+            free-disk-space: false
+            install-kind: false
+            requires-secret: false
+
+          - label: ssh-ports-attributes
+            runner: ubuntu-latest
+            free-disk-space: false
+            install-kind: false
+            requires-secret: false
+
+          - label: ssh-tunnel-mode
+            runner: ubuntu-latest
+            free-disk-space: false
+            install-kind: false
+            requires-secret: false
+
+          - label: ssh-credentials-server-race
+            runner: ubuntu-latest
+            free-disk-space: false
+            install-kind: false
+            requires-secret: false
+
           - label: build
             runner: ubuntu-latest
             free-disk-space: false

cmd/internal/agentcontainer/credentials_server.go

@@ -4,10 +4,12 @@ import (
 	"context"
 	"encoding/base64"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"net"
 	"os"
 	"strconv"
+	"syscall"
 
 	"github.com/devsy-org/devsy/cmd/flags"
 	"github.com/devsy-org/devsy/pkg/agent/tunnel"
@@ -91,6 +93,16 @@ func (cmd *CredentialsServerCmd) Run(ctx context.Context, port int) error {
 	runCtx, cancel := context.WithCancel(ctx)
 	defer cancel()
 
+	ln, err := claimPort(port)
+	if err != nil {
+		if errors.Is(err, errPortOwnedByAnotherSession) {
+			cmd.logPortOwnedByAnotherSession(ctx, port)
+			return nil
+		}
+		return err
+	}
+	defer func() { _ = ln.Close() }()
+
 	tunnelClient, err := tunnelserver.NewTunnelClient(os.Stdin, os.Stdout, true, ExitCodeIO)
 	if err != nil {
 		return fmt.Errorf("error creating tunnel client: %w", err)
@@ -100,12 +112,6 @@ func (cmd *CredentialsServerCmd) Run(ctx context.Context, port int) error {
 		return fmt.Errorf("ping client: %w", err)
 	}
 
-	ln, err := claimPort(port)
-	if err != nil {
-		return err
-	}
-	defer func() { _ = ln.Close() }()
-
 	cmd.maybeForwardPorts(runCtx, tunnelClient)
 
 	if err := cmd.configureDockerHelper(port); err != nil {
@@ -126,9 +132,13 @@ func (cmd *CredentialsServerCmd) Run(ctx context.Context, port int) error {
 	cleanupGitSigning := cmd.configureGitSigningKey()
 	defer cleanupGitSigning()
 
-	return credentials.RunCredentialsServerWithListener(runCtx, ln, tunnelClient)
+	return credentials.RunCredentialsServerWithListener(runCtx, ln, tunnelClient, cmd.User)
 }
 
+// errPortOwnedByAnotherSession marks a bind failure as another session
+// already owning the port, not a real error.
+var errPortOwnedByAnotherSession = errors.New("credentials server port owned by another session")
+
 // 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
@@ -137,13 +147,40 @@ 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",
+		if errors.Is(err, syscall.EADDRINUSE) {
+			return nil, fmt.Errorf("%w: %w", errPortOwnedByAnotherSession, err)
+		}
+		return nil, fmt.Errorf("port %d not available: %w", port, err)
+	}
+	return ln, nil
+}
+
+func (cmd *CredentialsServerCmd) logPortOwnedByAnotherSession(ctx context.Context, port int) {
+	owner, err := credentials.FetchOwner(ctx, port)
+	switch {
+	case err != nil:
+		log.Debugf(
+			"skipping credentials server for user %s: port %d is taken and its owner could not be determined: %v",
+			cmd.User,
 			port,
 			err,
 		)
+	case owner == "" || owner == cmd.User:
+		log.Debugf(
+			"skipping credentials server for user %s: another session already provides it on port %d",
+			cmd.User,
+			port,
+		)
+	default:
+		log.Warnf(
+			"credentials server for user %s was not started: port %d is already owned by user %s's session; "+
+				"git/docker/signing credential helpers for %s will not work until that session ends",
+			cmd.User,
+			port,
+			owner,
+			cmd.User,
+		)
 	}
-	return ln, nil
 }
 
 func (cmd *CredentialsServerCmd) maybeForwardPorts(

cmd/internal/agentcontainer/credentials_server_test.go

@@ -1,12 +1,21 @@
 package agentcontainer
 
 import (
+	"context"
+	"errors"
+	"fmt"
 	"net"
+	"strings"
 	"sync"
 	"testing"
+	"time"
 
+	"github.com/devsy-org/devsy/pkg/agent/tunnel"
+	"github.com/devsy-org/devsy/pkg/credentials"
+	"github.com/devsy-org/devsy/pkg/log"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc"
 )
 
 func TestClaimPort_SucceedsWhenPortFree(t *testing.T) {
@@ -23,7 +32,7 @@ func TestClaimPort_ErrorsWhenPortHeld(t *testing.T) {
 
 	_, err = claimPort(port)
 	require.Error(t, err)
-	assert.Contains(t, err.Error(), "not available")
+	assert.ErrorIs(t, err, errPortOwnedByAnotherSession)
 }
 
 func TestClaimPort_BecomesClaimableAfterHolderReleases(t *testing.T) {
@@ -79,3 +88,100 @@ func TestClaimPort_OnlyOneConcurrentCallerWins(t *testing.T) {
 		_ = winner.Close()
 	}
 }
+
+func startFakeCredentialsServer(t *testing.T, owner string) int {
+	t.Helper()
+
+	ln, err := net.Listen("tcp", "localhost:0")
+	require.NoError(t, err)
+	port := ln.Addr().(*net.TCPAddr).Port
+
+	ctx, cancel := context.WithCancel(context.Background())
+	t.Cleanup(cancel)
+	go func() {
+		_ = credentials.RunCredentialsServerWithListener(ctx, ln, &fakeCredentialsClient{}, owner)
+	}()
+
+	require.Eventually(t, func() bool {
+		conn, dialErr := net.Dial("tcp", ln.Addr().String())
+		if dialErr != nil {
+			return false
+		}
+		_ = conn.Close()
+		return true
+	}, time.Second, 5*time.Millisecond, "fake credentials server must become dialable")
+
+	return port
+}
+
+func TestCredentialsServerCmd_Run_SameOwnerCollisionIsSilentNoOp(t *testing.T) {
+	port := startFakeCredentialsServer(t, "alice")
+
+	var sink strings.Builder
+	log.Init(log.Config{Verbosity: 2, Format: "json"})
+	remove := log.AddSink(&sink)
+	defer remove()
+
+	cmd := &CredentialsServerCmd{User: "alice"}
+	err := cmd.Run(context.Background(), port)
+	require.NoError(t, err, "losing to the same owner's session must not be an error")
+	_ = log.Sync()
+
+	assert.NotContains(t, sink.String(), "\"level\":\"warn\"", "same-owner collision must not warn")
+}
+
+func TestCredentialsServerCmd_Run_DifferentOwnerCollisionWarnsButDoesNotError(t *testing.T) {
+	port := startFakeCredentialsServer(t, "alice")
+
+	var sink strings.Builder
+	log.Init(log.Config{Verbosity: 2, Format: "json"})
+	remove := log.AddSink(&sink)
+	defer remove()
+
+	cmd := &CredentialsServerCmd{User: "root"}
+	err := cmd.Run(context.Background(), port)
+	require.NoError(t, err, "losing the race must still not fail the session")
+	_ = log.Sync()
+
+	logged := sink.String()
+	assert.Contains(t, logged, "root", "warning must name the user left without credentials")
+	assert.Contains(t, logged, "alice", "warning must name the owning session")
+}
+
+type fakeCredentialsClient struct{}
+
+func (fakeCredentialsClient) GitCredentials(
+	_ context.Context, _ *tunnel.Message, _ ...grpc.CallOption,
+) (*tunnel.Message, error) {
+	return nil, fmt.Errorf("not implemented")
+}
+
+func (fakeCredentialsClient) DockerCredentials(
+	_ context.Context, _ *tunnel.Message, _ ...grpc.CallOption,
+) (*tunnel.Message, error) {
+	return nil, fmt.Errorf("not implemented")
+}
+
+func (fakeCredentialsClient) GitSSHSignature(
+	_ context.Context, _ *tunnel.Message, _ ...grpc.CallOption,
+) (*tunnel.Message, error) {
+	return nil, fmt.Errorf("not implemented")
+}
+
+func (fakeCredentialsClient) GPGPublicKeys(
+	_ context.Context, _ *tunnel.Message, _ ...grpc.CallOption,
+) (*tunnel.Message, error) {
+	return nil, fmt.Errorf("not implemented")
+}
+
+func (fakeCredentialsClient) DevsyConfig(
+	_ context.Context, _ *tunnel.Message, _ ...grpc.CallOption,
+) (*tunnel.Message, error) {
+	return nil, fmt.Errorf("not implemented")
+}
+
+func TestClaimPort_WrapsNonAddrInUseErrorsWithoutSentinel(t *testing.T) {
+	_, err := claimPort(-1)
+	require.Error(t, err)
+	assert.False(t, errors.Is(err, errPortOwnedByAnotherSession))
+}

e2e/tests/ssh/agent_forward.go

@@ -21,8 +21,7 @@ import (
 // per-connection socket directory must be cleaned up on disconnect.
 var _ = ginkgo.Describe(
 	"devsy ssh agent forwarding",
-	ginkgo.Label("ssh"),
-	ginkgo.Label("agent-forward"),
+	ginkgo.Label("ssh-agent-forward"),
 	ginkgo.Ordered,
 	func() {
 		var (
@@ -253,8 +252,6 @@ var _ = ginkgo.Describe(
 
 		ginkgo.It(
 			"connection without any agent request still cleans up",
-			ginkgo.Label("ssh"),
-			ginkgo.Label("agent-forward"),
 			ginkgo.SpecTimeout(framework.TimeoutModerate()),
 			func(ctx ginkgo.SpecContext) {
 				tmpDir, err := os.MkdirTemp("", "devsy-ssh-cm-clean-")
@@ -325,7 +322,6 @@ var _ = ginkgo.Describe(
 
 		ginkgo.It(
 			"parallel sessions on one connection observe the same socket concurrently",
-			ginkgo.Label("agent-forward"),
 			ginkgo.SpecTimeout(framework.TimeoutModerate()),
 			func(_ ginkgo.SpecContext) {
 				controlPath, closeCM, err := framework.OpenSSHControlMaster(

e2e/tests/ssh/credentials_server_race.go

@@ -0,0 +1,85 @@
+package ssh
+
+import (
+	"context"
+	"os"
+	"sync"
+	"time"
+
+	"github.com/devsy-org/devsy/e2e/framework"
+	"github.com/onsi/ginkgo/v2"
+	"github.com/onsi/gomega"
+)
+
+const (
+	cmdWorkspace = "workspace"
+	cmdSSH       = "ssh"
+)
+
+var _ = ginkgo.Describe(
+	"devsy ssh credentials server race",
+	ginkgo.Label("ssh-credentials-server-race"),
+	ginkgo.Ordered,
+	func() {
+		var initialDir string
+
+		ginkgo.BeforeEach(func() {
+			var err error
+			initialDir, err = os.Getwd()
+			framework.ExpectNoError(err)
+		})
+
+		ginkgo.It(
+			"should not surface credentials-server port errors when two ssh sessions race for the same workspace",
+			ginkgo.SpecTimeout(framework.TimeoutModerate()),
+			func(ctx context.Context) {
+				tempDir, err := framework.CopyToTempDir("tests/ssh/testdata/local-test")
+				framework.ExpectNoError(err)
+
+				f := framework.NewDefaultFramework(initialDir + "/bin")
+				_ = f.DevsyProviderAdd(ctx, "docker")
+				err = f.DevsyProviderUse(ctx, "docker")
+				framework.ExpectNoError(err)
+
+				ginkgo.DeferCleanup(func(cleanupCtx context.Context) {
+					_ = f.DevsyWorkspaceDelete(cleanupCtx, tempDir)
+					framework.CleanupTempDir(initialDir, tempDir)
+				})
+
+				upDeadline := time.Now().Add(5 * time.Minute)
+				upCtx, cancelUp := context.WithDeadline(ctx, upDeadline)
+				defer cancelUp()
+				err = f.DevsyUp(upCtx, tempDir)
+				framework.ExpectNoError(err)
+
+				const sessions = 2
+				var wg sync.WaitGroup
+				stderrs := make([]string, sessions)
+				runErrs := make([]error, sessions)
+				for i := range sessions {
+					wg.Add(1)
+					go func(i int) {
+						defer wg.Done()
+						sshCtx, cancelSSH := context.WithDeadline(
+							ctx,
+							time.Now().Add(30*time.Second),
+						)
+						defer cancelSSH()
+						_, stderr, sshErr := f.ExecCommandCapture(sshCtx, []string{
+							cmdWorkspace, cmdSSH, tempDir, "--command", "sleep 2",
+						})
+						stderrs[i] = stderr
+						runErrs[i] = sshErr
+					}(i)
+				}
+				wg.Wait()
+
+				for i := range sessions {
+					framework.ExpectNoError(runErrs[i], "ssh session %d stderr: %s", i, stderrs[i])
+					gomega.Expect(stderrs[i]).NotTo(gomega.ContainSubstring("not available"))
+					gomega.Expect(stderrs[i]).NotTo(gomega.ContainSubstring("credentials server"))
+				}
+			},
+		)
+	},
+)