added support for stream type npipe (Windows named pipe)

Signed-off-by: Sam Alba <samalba@users.noreply.github.com>

environment/pipeline.go

@@ -574,20 +574,35 @@ func (p *Pipeline) mount(ctx context.Context, dest string, mnt *compiler.Value)
 			return nil, fmt.Errorf("invalid stream %q: not a stream", stream.Path().String())
 		}
 
+		// Unix socket
 		unixValue := stream.Lookup("unix")
-		if !unixValue.Exists() {
-			return nil, fmt.Errorf("invalid stream %q: not a unix socket", stream.Path().String())
+		if unixValue.Exists() {
+			unix, err := unixValue.String()
+			if err != nil {
+				return nil, fmt.Errorf("invalid unix path id: %w", err)
+			}
+
+			return llb.AddSSHSocket(
+				llb.SSHID(fmt.Sprintf("unix=%s", unix)),
+				llb.SSHSocketTarget(dest),
+			), nil
 		}
 
-		unix, err := unixValue.String()
-		if err != nil {
-			return nil, fmt.Errorf("invalid unix path id: %w", err)
+		// Windows named pipe
+		npipeValue := stream.Lookup("npipe")
+		if npipeValue.Exists() {
+			npipe, err := npipeValue.String()
+			if err != nil {
+				return nil, fmt.Errorf("invalid npipe path id: %w", err)
+			}
+
+			return llb.AddSSHSocket(
+				llb.SSHID(fmt.Sprintf("npipe=%s", npipe)),
+				llb.SSHSocketTarget(dest),
+			), nil
 		}
 
-		return llb.AddSSHSocket(
-			llb.SSHID(fmt.Sprintf("unix=%s", unix)),
-			llb.SSHSocketTarget(dest),
-		), nil
+		return nil, fmt.Errorf("invalid stream %q: not a valid stream", stream.Path().String())
 	}
 
 	// eg. mount: "/foo": { from: www.source }

solver/socketprovider.go

@@ -7,6 +7,7 @@ import (
 	"strings"
 	"time"
 
+	"github.com/Microsoft/go-winio"
 	"github.com/moby/buildkit/session"
 	"github.com/moby/buildkit/session/sshforward"
 	"google.golang.org/grpc"
@@ -14,7 +15,8 @@ import (
 )
 
 const (
-	unixPrefix = "unix="
+	unixPrefix  = "unix="
+	npipePrefix = "npipe="
 )
 
 type SocketProvider struct {
@@ -33,12 +35,26 @@ func (sp *SocketProvider) CheckAgent(ctx context.Context, req *sshforward.CheckA
 	if req.ID != "" {
 		id = req.ID
 	}
-	if !strings.HasPrefix(id, unixPrefix) {
+	if !strings.HasPrefix(id, unixPrefix) && !strings.HasPrefix(id, npipePrefix) {
 		return &sshforward.CheckAgentResponse{}, fmt.Errorf("invalid socket forward key %s", id)
 	}
 	return &sshforward.CheckAgentResponse{}, nil
 }
 
+func dialStream(id string) (net.Conn, error) {
+	switch {
+	case strings.HasPrefix(id, unixPrefix):
+		id = strings.TrimPrefix(id, unixPrefix)
+		return net.DialTimeout("unix", id, time.Second)
+	case strings.HasPrefix(id, npipePrefix):
+		id = strings.TrimPrefix(id, npipePrefix)
+		dur := time.Second
+		return winio.DialPipe(id, &dur)
+	default:
+		return nil, fmt.Errorf("invalid socket forward key %s", id)
+	}
+}
+
 func (sp *SocketProvider) ForwardAgent(stream sshforward.SSH_ForwardAgentServer) error {
 	id := sshforward.DefaultID
 
@@ -48,13 +64,7 @@ func (sp *SocketProvider) ForwardAgent(stream sshforward.SSH_ForwardAgentServer)
 		id = v[0]
 	}
 
-	if !strings.HasPrefix(id, unixPrefix) {
-		return fmt.Errorf("invalid socket forward key %s", id)
-	}
-
-	id = strings.TrimPrefix(id, unixPrefix)
-
-	conn, err := net.DialTimeout("unix", id, time.Second)
+	conn, err := dialStream(id)
 	if err != nil {
 		return fmt.Errorf("failed to connect to %s: %w", id, err)
 	}

state/input.go

@@ -280,7 +280,7 @@ func FileInput(data string) Input {
 }
 
 type fileInput struct {
-	Path string `json:"data,omitempty"`
+	Path string `yaml:"path,omitempty"`
 }
 
 func (i fileInput) Compile(_ string, _ *State) (*compiler.Value, error) {
@@ -296,20 +296,31 @@ func (i fileInput) Compile(_ string, _ *State) (*compiler.Value, error) {
 }
 
 // A socket input value
-func SocketInput(data string) Input {
-	i := socketInput{
-		Unix: data,
+func SocketInput(data, socketType string) Input {
+	i := socketInput{}
+
+	switch socketType {
+	case "npipe":
+		i.Npipe = data
+	case "unix":
+		i.Unix = data
 	}
+
 	return Input{
 		Socket: &i,
 	}
 }
 
 type socketInput struct {
-	Unix string `json:"unix,omitempty"`
+	Unix  string `json:"unix,omitempty" yaml:"unix,omitempty"`
+	Npipe string `json:"npipe,omitempty" yaml:"npipe,omitempty"`
 }
 
 func (i socketInput) Compile(_ string, _ *State) (*compiler.Value, error) {
-	socketValue := fmt.Sprintf(`{unix: %q}`, i.Unix)
-	return compiler.Compile("", socketValue)
+	socketValue, err := json.Marshal(i)
+	if err != nil {
+		return nil, err
+	}
+
+	return compiler.Compile("", string(socketValue))
 }

stdlib/dagger/dagger.cue

@@ -18,7 +18,13 @@ import (
 #Stream: {
 	@dagger(stream)
 
-	unix: string
+	{
+		// Unix Socket
+		unix: string
+	} | {
+		// Windows Named Pipe
+		npipe: string
+	}
 }
 
 // Secret value