feat: add support for agentsock on Windows (#22171)

relates to #21335

Adds support for the agentsock and thus `coder exp sync` commands on Windows. This support was initially missing.

agent/agentsocket/server_test.go

@@ -1,37 +1,22 @@
 package agentsocket_test
 
 import (
-	"context"
-	"path/filepath"
-	"runtime"
 	"testing"
 
-	"github.com/google/uuid"
-	"github.com/spf13/afero"
 	"github.com/stretchr/testify/require"
 
 	"cdr.dev/slog/v3"
-	"github.com/coder/coder/v2/agent"
 	"github.com/coder/coder/v2/agent/agentsocket"
-	"github.com/coder/coder/v2/agent/agenttest"
-	agentproto "github.com/coder/coder/v2/agent/proto"
-	"github.com/coder/coder/v2/codersdk/agentsdk"
-	"github.com/coder/coder/v2/tailnet"
-	"github.com/coder/coder/v2/tailnet/tailnettest"
 	"github.com/coder/coder/v2/testutil"
 )
 
 func TestServer(t *testing.T) {
 	t.Parallel()
 
-	if runtime.GOOS == "windows" {
-		t.Skip("agentsocket is not supported on Windows")
-	}
-
 	t.Run("StartStop", func(t *testing.T) {
 		t.Parallel()
 
-		socketPath := filepath.Join(t.TempDir(), "test.sock")
+		socketPath := testutil.AgentSocketPath(t)
 		logger := slog.Make().Leveled(slog.LevelDebug)
 		server, err := agentsocket.NewServer(logger, agentsocket.WithPath(socketPath))
 		require.NoError(t, err)
@@ -41,7 +26,7 @@ func TestServer(t *testing.T) {
 	t.Run("AlreadyStarted", func(t *testing.T) {
 		t.Parallel()
 
-		socketPath := filepath.Join(t.TempDir(), "test.sock")
+		socketPath := testutil.AgentSocketPath(t)
 		logger := slog.Make().Leveled(slog.LevelDebug)
 		server1, err := agentsocket.NewServer(logger, agentsocket.WithPath(socketPath))
 		require.NoError(t, err)
@@ -49,90 +34,4 @@ func TestServer(t *testing.T) {
 		_, err = agentsocket.NewServer(logger, agentsocket.WithPath(socketPath))
 		require.ErrorContains(t, err, "create socket")
 	})
-
-	t.Run("AutoSocketPath", func(t *testing.T) {
-		t.Parallel()
-
-		socketPath := filepath.Join(t.TempDir(), "test.sock")
-		logger := slog.Make().Leveled(slog.LevelDebug)
-		server, err := agentsocket.NewServer(logger, agentsocket.WithPath(socketPath))
-		require.NoError(t, err)
-		require.NoError(t, server.Close())
-	})
-}
-
-func TestServerWindowsNotSupported(t *testing.T) {
-	t.Parallel()
-
-	if runtime.GOOS != "windows" {
-		t.Skip("this test only runs on Windows")
-	}
-
-	t.Run("NewServer", func(t *testing.T) {
-		t.Parallel()
-
-		socketPath := filepath.Join(t.TempDir(), "test.sock")
-		logger := slog.Make().Leveled(slog.LevelDebug)
-		_, err := agentsocket.NewServer(logger, agentsocket.WithPath(socketPath))
-		require.ErrorContains(t, err, "agentsocket is not supported on Windows")
-	})
-
-	t.Run("NewClient", func(t *testing.T) {
-		t.Parallel()
-
-		_, err := agentsocket.NewClient(context.Background(), agentsocket.WithPath("test.sock"))
-		require.ErrorContains(t, err, "agentsocket is not supported on Windows")
-	})
-}
-
-func TestAgentInitializesOnWindowsWithoutSocketServer(t *testing.T) {
-	t.Parallel()
-
-	if runtime.GOOS != "windows" {
-		t.Skip("this test only runs on Windows")
-	}
-
-	ctx := testutil.Context(t, testutil.WaitShort)
-	logger := testutil.Logger(t).Named("agent")
-
-	derpMap, _ := tailnettest.RunDERPAndSTUN(t)
-
-	coordinator := tailnet.NewCoordinator(logger)
-	t.Cleanup(func() {
-		_ = coordinator.Close()
-	})
-
-	statsCh := make(chan *agentproto.Stats, 50)
-	agentID := uuid.New()
-	manifest := agentsdk.Manifest{
-		AgentID:       agentID,
-		AgentName:     "test-agent",
-		WorkspaceName: "test-workspace",
-		OwnerName:     "test-user",
-		WorkspaceID:   uuid.New(),
-		DERPMap:       derpMap,
-	}
-
-	client := agenttest.NewClient(t, logger.Named("agenttest"), agentID, manifest, statsCh, coordinator)
-	t.Cleanup(client.Close)
-
-	options := agent.Options{
-		Client:                 client,
-		Filesystem:             afero.NewMemMapFs(),
-		Logger:                 logger.Named("agent"),
-		ReconnectingPTYTimeout: testutil.WaitShort,
-		EnvironmentVariables:   map[string]string{},
-		SocketPath:             "",
-	}
-
-	agnt := agent.New(options)
-	t.Cleanup(func() {
-		_ = agnt.Close()
-	})
-
-	startup := testutil.TryReceive(ctx, t, client.GetStartup())
-	require.NotNil(t, startup, "agent should send startup message")
-
-	err := agnt.Close()
-	require.NoError(t, err, "agent should close cleanly")
 }

agent/agentsocket/service_test.go

@@ -2,8 +2,6 @@ package agentsocket_test
 
 import (
 	"context"
-	"path/filepath"
-	"runtime"
 	"testing"
 
 	"github.com/stretchr/testify/require"
@@ -30,14 +28,10 @@ func newSocketClient(ctx context.Context, t *testing.T, socketPath string) *agen
 func TestDRPCAgentSocketService(t *testing.T) {
 	t.Parallel()
 
-	if runtime.GOOS == "windows" {
-		t.Skip("agentsocket is not supported on Windows")
-	}
-
 	t.Run("Ping", func(t *testing.T) {
 		t.Parallel()
 
-		socketPath := filepath.Join(testutil.TempDirUnixSocket(t), "test.sock")
+		socketPath := testutil.AgentSocketPath(t)
 		ctx := testutil.Context(t, testutil.WaitShort)
 		server, err := agentsocket.NewServer(
 			slog.Make().Leveled(slog.LevelDebug),
@@ -57,7 +51,7 @@ func TestDRPCAgentSocketService(t *testing.T) {
 
 		t.Run("NewUnit", func(t *testing.T) {
 			t.Parallel()
-			socketPath := filepath.Join(testutil.TempDirUnixSocket(t), "test.sock")
+			socketPath := testutil.AgentSocketPath(t)
 			ctx := testutil.Context(t, testutil.WaitShort)
 			server, err := agentsocket.NewServer(
 				slog.Make().Leveled(slog.LevelDebug),
@@ -79,7 +73,7 @@ func TestDRPCAgentSocketService(t *testing.T) {
 		t.Run("UnitAlreadyStarted", func(t *testing.T) {
 			t.Parallel()
 
-			socketPath := filepath.Join(testutil.TempDirUnixSocket(t), "test.sock")
+			socketPath := testutil.AgentSocketPath(t)
 			ctx := testutil.Context(t, testutil.WaitShort)
 			server, err := agentsocket.NewServer(
 				slog.Make().Leveled(slog.LevelDebug),
@@ -109,7 +103,7 @@ func TestDRPCAgentSocketService(t *testing.T) {
 		t.Run("UnitAlreadyCompleted", func(t *testing.T) {
 			t.Parallel()
 
-			socketPath := filepath.Join(testutil.TempDirUnixSocket(t), "test.sock")
+			socketPath := testutil.AgentSocketPath(t)
 			ctx := testutil.Context(t, testutil.WaitShort)
 			server, err := agentsocket.NewServer(
 				slog.Make().Leveled(slog.LevelDebug),
@@ -148,7 +142,7 @@ func TestDRPCAgentSocketService(t *testing.T) {
 		t.Run("UnitNotReady", func(t *testing.T) {
 			t.Parallel()
 
-			socketPath := filepath.Join(testutil.TempDirUnixSocket(t), "test.sock")
+			socketPath := testutil.AgentSocketPath(t)
 			ctx := testutil.Context(t, testutil.WaitShort)
 			server, err := agentsocket.NewServer(
 				slog.Make().Leveled(slog.LevelDebug),
@@ -178,7 +172,7 @@ func TestDRPCAgentSocketService(t *testing.T) {
 		t.Run("NewUnits", func(t *testing.T) {
 			t.Parallel()
 
-			socketPath := filepath.Join(testutil.TempDirUnixSocket(t), "test.sock")
+			socketPath := testutil.AgentSocketPath(t)
 			ctx := testutil.Context(t, testutil.WaitShort)
 			server, err := agentsocket.NewServer(
 				slog.Make().Leveled(slog.LevelDebug),
@@ -203,7 +197,7 @@ func TestDRPCAgentSocketService(t *testing.T) {
 		t.Run("DependencyAlreadyRegistered", func(t *testing.T) {
 			t.Parallel()
 
-			socketPath := filepath.Join(testutil.TempDirUnixSocket(t), "test.sock")
+			socketPath := testutil.AgentSocketPath(t)
 			ctx := testutil.Context(t, testutil.WaitShort)
 			server, err := agentsocket.NewServer(
 				slog.Make().Leveled(slog.LevelDebug),
@@ -238,7 +232,7 @@ func TestDRPCAgentSocketService(t *testing.T) {
 		t.Run("DependencyAddedAfterDependentStarted", func(t *testing.T) {
 			t.Parallel()
 
-			socketPath := filepath.Join(testutil.TempDirUnixSocket(t), "test.sock")
+			socketPath := testutil.AgentSocketPath(t)
 			ctx := testutil.Context(t, testutil.WaitShort)
 			server, err := agentsocket.NewServer(
 				slog.Make().Leveled(slog.LevelDebug),
@@ -280,7 +274,7 @@ func TestDRPCAgentSocketService(t *testing.T) {
 		t.Run("UnregisteredUnit", func(t *testing.T) {
 			t.Parallel()
 
-			socketPath := filepath.Join(testutil.TempDirUnixSocket(t), "test.sock")
+			socketPath := testutil.AgentSocketPath(t)
 			ctx := testutil.Context(t, testutil.WaitShort)
 			server, err := agentsocket.NewServer(
 				slog.Make().Leveled(slog.LevelDebug),
@@ -299,7 +293,7 @@ func TestDRPCAgentSocketService(t *testing.T) {
 		t.Run("UnitNotReady", func(t *testing.T) {
 			t.Parallel()
 
-			socketPath := filepath.Join(testutil.TempDirUnixSocket(t), "test.sock")
+			socketPath := testutil.AgentSocketPath(t)
 			ctx := testutil.Context(t, testutil.WaitShort)
 			server, err := agentsocket.NewServer(
 				slog.Make().Leveled(slog.LevelDebug),
@@ -323,7 +317,7 @@ func TestDRPCAgentSocketService(t *testing.T) {
 		t.Run("UnitReady", func(t *testing.T) {
 			t.Parallel()
 
-			socketPath := filepath.Join(testutil.TempDirUnixSocket(t), "test.sock")
+			socketPath := testutil.AgentSocketPath(t)
 			ctx := testutil.Context(t, testutil.WaitShort)
 			server, err := agentsocket.NewServer(
 				slog.Make().Leveled(slog.LevelDebug),

agent/agentsocket/socket_windows.go

@@ -4,19 +4,60 @@ package agentsocket
 
 import (
 	"context"
+	"fmt"
 	"net"
+	"os"
+	"os/user"
+	"strings"
 
+	"github.com/Microsoft/go-winio"
 	"golang.org/x/xerrors"
 )
 
-func createSocket(_ string) (net.Listener, error) {
-	return nil, xerrors.New("agentsocket is not supported on Windows")
+const defaultSocketPath = `\\.\pipe\com.coder.agentsocket`
+
+func createSocket(path string) (net.Listener, error) {
+	if path == "" {
+		path = defaultSocketPath
+	}
+	if !strings.HasPrefix(path, `\\.\pipe\`) {
+		return nil, xerrors.Errorf("%q is not a valid local socket path", path)
+	}
+
+	user, err := user.Current()
+	if err != nil {
+		return nil, fmt.Errorf("unable to look up current user: %w", err)
+	}
+	sid := user.Uid
+
+	// SecurityDescriptor is in SDDL format. c.f.
+	// https://learn.microsoft.com/en-us/windows/win32/secauthz/security-descriptor-string-format for full details.
+	// D: indicates this is a Discretionary Access Control List (DACL), which is Windows-speak for ACLs that allow or
+	// deny access (as opposed to SACL which controls audit logging).
+	// P indicates that this DACL is "protected" from being modified thru inheritance
+	// () delimit access control entries (ACEs), here we only have one, which, allows (A) generic all (GA) access to our
+	// specific user's security ID (SID).
+	//
+	// Note that although Microsoft docs at https://learn.microsoft.com/en-us/windows/win32/ipc/named-pipes warns that
+	// named pipes are accessible from remote machines in the general case, the `winio` package sets the flag
+	// windows.FILE_PIPE_REJECT_REMOTE_CLIENTS when creating pipes, so connections from remote machines are always
+	// denied. This is important because we sort of expect customers to run the Coder agent under a generic user
+	// account unless they are very sophisticated. We don't want this socket to cross the boundary of the local machine.
+	configuration := &winio.PipeConfig{
+		SecurityDescriptor: fmt.Sprintf("D:P(A;;GA;;;%s)", sid),
+	}
+
+	listener, err := winio.ListenPipe(path, configuration)
+	if err != nil {
+		return nil, xerrors.Errorf("failed to open named pipe: %w", err)
+	}
+	return listener, nil
 }
 
-func cleanupSocket(_ string) error {
-	return nil
+func cleanupSocket(path string) error {
+	return os.Remove(path)
 }
 
-func dialSocket(_ context.Context, _ string) (net.Conn, error) {
-	return nil, xerrors.New("agentsocket is not supported on Windows")
+func dialSocket(ctx context.Context, path string) (net.Conn, error) {
+	return winio.DialPipeContext(ctx, path)
 }