pipe.go

Documentation: github.com/Microsoft/go-winio

     1  //go:build windows
     2  // +build windows
     3  
     4  package winio
     5  
     6  import (
     7  	"context"
     8  	"errors"
     9  	"fmt"
    10  	"io"
    11  	"net"
    12  	"os"
    13  	"runtime"
    14  	"syscall"
    15  	"time"
    16  	"unsafe"
    17  
    18  	"golang.org/x/sys/windows"
    19  
    20  	"github.com/Microsoft/go-winio/internal/fs"
    21  )
    22  
    23  //sys connectNamedPipe(pipe syscall.Handle, o *syscall.Overlapped) (err error) = ConnectNamedPipe
    24  //sys createNamedPipe(name string, flags uint32, pipeMode uint32, maxInstances uint32, outSize uint32, inSize uint32, defaultTimeout uint32, sa *syscall.SecurityAttributes) (handle syscall.Handle, err error)  [failretval==syscall.InvalidHandle] = CreateNamedPipeW
    25  //sys getNamedPipeInfo(pipe syscall.Handle, flags *uint32, outSize *uint32, inSize *uint32, maxInstances *uint32) (err error) = GetNamedPipeInfo
    26  //sys getNamedPipeHandleState(pipe syscall.Handle, state *uint32, curInstances *uint32, maxCollectionCount *uint32, collectDataTimeout *uint32, userName *uint16, maxUserNameSize uint32) (err error) = GetNamedPipeHandleStateW
    27  //sys localAlloc(uFlags uint32, length uint32) (ptr uintptr) = LocalAlloc
    28  //sys ntCreateNamedPipeFile(pipe *syscall.Handle, access uint32, oa *objectAttributes, iosb *ioStatusBlock, share uint32, disposition uint32, options uint32, typ uint32, readMode uint32, completionMode uint32, maxInstances uint32, inboundQuota uint32, outputQuota uint32, timeout *int64) (status ntStatus) = ntdll.NtCreateNamedPipeFile
    29  //sys rtlNtStatusToDosError(status ntStatus) (winerr error) = ntdll.RtlNtStatusToDosErrorNoTeb
    30  //sys rtlDosPathNameToNtPathName(name *uint16, ntName *unicodeString, filePart uintptr, reserved uintptr) (status ntStatus) = ntdll.RtlDosPathNameToNtPathName_U
    31  //sys rtlDefaultNpAcl(dacl *uintptr) (status ntStatus) = ntdll.RtlDefaultNpAcl
    32  
    33  type ioStatusBlock struct {
    34  	Status, Information uintptr
    35  }
    36  
    37  type objectAttributes struct {
    38  	Length             uintptr
    39  	RootDirectory      uintptr
    40  	ObjectName         *unicodeString
    41  	Attributes         uintptr
    42  	SecurityDescriptor *securityDescriptor
    43  	SecurityQoS        uintptr
    44  }
    45  
    46  type unicodeString struct {
    47  	Length        uint16
    48  	MaximumLength uint16
    49  	Buffer        uintptr
    50  }
    51  
    52  type securityDescriptor struct {
    53  	Revision byte
    54  	Sbz1     byte
    55  	Control  uint16
    56  	Owner    uintptr
    57  	Group    uintptr
    58  	Sacl     uintptr //revive:disable-line:var-naming SACL, not Sacl
    59  	Dacl     uintptr //revive:disable-line:var-naming DACL, not Dacl
    60  }
    61  
    62  type ntStatus int32
    63  
    64  func (status ntStatus) Err() error {
    65  	if status >= 0 {
    66  		return nil
    67  	}
    68  	return rtlNtStatusToDosError(status)
    69  }
    70  
    71  var (
    72  	// ErrPipeListenerClosed is returned for pipe operations on listeners that have been closed.
    73  	ErrPipeListenerClosed = net.ErrClosed
    74  
    75  	errPipeWriteClosed = errors.New("pipe has been closed for write")
    76  )
    77  
    78  type win32Pipe struct {
    79  	*win32File
    80  	path string
    81  }
    82  
    83  type win32MessageBytePipe struct {
    84  	win32Pipe
    85  	writeClosed bool
    86  	readEOF     bool
    87  }
    88  
    89  type pipeAddress string
    90  
    91  func (f *win32Pipe) LocalAddr() net.Addr {
    92  	return pipeAddress(f.path)
    93  }
    94  
    95  func (f *win32Pipe) RemoteAddr() net.Addr {
    96  	return pipeAddress(f.path)
    97  }
    98  
    99  func (f *win32Pipe) SetDeadline(t time.Time) error {
   100  	if err := f.SetReadDeadline(t); err != nil {
   101  		return err
   102  	}
   103  	return f.SetWriteDeadline(t)
   104  }
   105  
   106  // CloseWrite closes the write side of a message pipe in byte mode.
   107  func (f *win32MessageBytePipe) CloseWrite() error {
   108  	if f.writeClosed {
   109  		return errPipeWriteClosed
   110  	}
   111  	err := f.win32File.Flush()
   112  	if err != nil {
   113  		return err
   114  	}
   115  	_, err = f.win32File.Write(nil)
   116  	if err != nil {
   117  		return err
   118  	}
   119  	f.writeClosed = true
   120  	return nil
   121  }
   122  
   123  // Write writes bytes to a message pipe in byte mode. Zero-byte writes are ignored, since
   124  // they are used to implement CloseWrite().
   125  func (f *win32MessageBytePipe) Write(b []byte) (int, error) {
   126  	if f.writeClosed {
   127  		return 0, errPipeWriteClosed
   128  	}
   129  	if len(b) == 0 {
   130  		return 0, nil
   131  	}
   132  	return f.win32File.Write(b)
   133  }
   134  
   135  // Read reads bytes from a message pipe in byte mode. A read of a zero-byte message on a message
   136  // mode pipe will return io.EOF, as will all subsequent reads.
   137  func (f *win32MessageBytePipe) Read(b []byte) (int, error) {
   138  	if f.readEOF {
   139  		return 0, io.EOF
   140  	}
   141  	n, err := f.win32File.Read(b)
   142  	if err == io.EOF { //nolint:errorlint
   143  		// If this was the result of a zero-byte read, then
   144  		// it is possible that the read was due to a zero-size
   145  		// message. Since we are simulating CloseWrite with a
   146  		// zero-byte message, ensure that all future Read() calls
   147  		// also return EOF.
   148  		f.readEOF = true
   149  	} else if err == syscall.ERROR_MORE_DATA { //nolint:errorlint // err is Errno
   150  		// ERROR_MORE_DATA indicates that the pipe's read mode is message mode
   151  		// and the message still has more bytes. Treat this as a success, since
   152  		// this package presents all named pipes as byte streams.
   153  		err = nil
   154  	}
   155  	return n, err
   156  }
   157  
   158  func (pipeAddress) Network() string {
   159  	return "pipe"
   160  }
   161  
   162  func (s pipeAddress) String() string {
   163  	return string(s)
   164  }
   165  
   166  // tryDialPipe attempts to dial the pipe at `path` until `ctx` cancellation or timeout.
   167  func tryDialPipe(ctx context.Context, path *string, access fs.AccessMask) (syscall.Handle, error) {
   168  	for {
   169  		select {
   170  		case <-ctx.Done():
   171  			return syscall.Handle(0), ctx.Err()
   172  		default:
   173  			wh, err := fs.CreateFile(*path,
   174  				access,
   175  				0,   // mode
   176  				nil, // security attributes
   177  				fs.OPEN_EXISTING,
   178  				fs.FILE_FLAG_OVERLAPPED|fs.SECURITY_SQOS_PRESENT|fs.SECURITY_ANONYMOUS,
   179  				0, // template file handle
   180  			)
   181  			h := syscall.Handle(wh)
   182  			if err == nil {
   183  				return h, nil
   184  			}
   185  			if err != windows.ERROR_PIPE_BUSY { //nolint:errorlint // err is Errno
   186  				return h, &os.PathError{Err: err, Op: "open", Path: *path}
   187  			}
   188  			// Wait 10 msec and try again. This is a rather simplistic
   189  			// view, as we always try each 10 milliseconds.
   190  			time.Sleep(10 * time.Millisecond)
   191  		}
   192  	}
   193  }
   194  
   195  // DialPipe connects to a named pipe by path, timing out if the connection
   196  // takes longer than the specified duration. If timeout is nil, then we use
   197  // a default timeout of 2 seconds.  (We do not use WaitNamedPipe.)
   198  func DialPipe(path string, timeout *time.Duration) (net.Conn, error) {
   199  	var absTimeout time.Time
   200  	if timeout != nil {
   201  		absTimeout = time.Now().Add(*timeout)
   202  	} else {
   203  		absTimeout = time.Now().Add(2 * time.Second)
   204  	}
   205  	ctx, cancel := context.WithDeadline(context.Background(), absTimeout)
   206  	defer cancel()
   207  	conn, err := DialPipeContext(ctx, path)
   208  	if errors.Is(err, context.DeadlineExceeded) {
   209  		return nil, ErrTimeout
   210  	}
   211  	return conn, err
   212  }
   213  
   214  // DialPipeContext attempts to connect to a named pipe by `path` until `ctx`
   215  // cancellation or timeout.
   216  func DialPipeContext(ctx context.Context, path string) (net.Conn, error) {
   217  	return DialPipeAccess(ctx, path, syscall.GENERIC_READ|syscall.GENERIC_WRITE)
   218  }
   219  
   220  // DialPipeAccess attempts to connect to a named pipe by `path` with `access` until `ctx`
   221  // cancellation or timeout.
   222  func DialPipeAccess(ctx context.Context, path string, access uint32) (net.Conn, error) {
   223  	var err error
   224  	var h syscall.Handle
   225  	h, err = tryDialPipe(ctx, &path, fs.AccessMask(access))
   226  	if err != nil {
   227  		return nil, err
   228  	}
   229  
   230  	var flags uint32
   231  	err = getNamedPipeInfo(h, &flags, nil, nil, nil)
   232  	if err != nil {
   233  		return nil, err
   234  	}
   235  
   236  	f, err := makeWin32File(h)
   237  	if err != nil {
   238  		syscall.Close(h)
   239  		return nil, err
   240  	}
   241  
   242  	// If the pipe is in message mode, return a message byte pipe, which
   243  	// supports CloseWrite().
   244  	if flags&windows.PIPE_TYPE_MESSAGE != 0 {
   245  		return &win32MessageBytePipe{
   246  			win32Pipe: win32Pipe{win32File: f, path: path},
   247  		}, nil
   248  	}
   249  	return &win32Pipe{win32File: f, path: path}, nil
   250  }
   251  
   252  type acceptResponse struct {
   253  	f   *win32File
   254  	err error
   255  }
   256  
   257  type win32PipeListener struct {
   258  	firstHandle syscall.Handle
   259  	path        string
   260  	config      PipeConfig
   261  	acceptCh    chan (chan acceptResponse)
   262  	closeCh     chan int
   263  	doneCh      chan int
   264  }
   265  
   266  func makeServerPipeHandle(path string, sd []byte, c *PipeConfig, first bool) (syscall.Handle, error) {
   267  	path16, err := syscall.UTF16FromString(path)
   268  	if err != nil {
   269  		return 0, &os.PathError{Op: "open", Path: path, Err: err}
   270  	}
   271  
   272  	var oa objectAttributes
   273  	oa.Length = unsafe.Sizeof(oa)
   274  
   275  	var ntPath unicodeString
   276  	if err := rtlDosPathNameToNtPathName(&path16[0],
   277  		&ntPath,
   278  		0,
   279  		0,
   280  	).Err(); err != nil {
   281  		return 0, &os.PathError{Op: "open", Path: path, Err: err}
   282  	}
   283  	defer localFree(ntPath.Buffer)
   284  	oa.ObjectName = &ntPath
   285  	oa.Attributes = windows.OBJ_CASE_INSENSITIVE
   286  
   287  	// The security descriptor is only needed for the first pipe.
   288  	if first {
   289  		if sd != nil {
   290  			l := uint32(len(sd))
   291  			sdb := localAlloc(0, l)
   292  			defer localFree(sdb)
   293  			copy((*[0xffff]byte)(unsafe.Pointer(sdb))[:], sd)
   294  			oa.SecurityDescriptor = (*securityDescriptor)(unsafe.Pointer(sdb))
   295  		} else {
   296  			// Construct the default named pipe security descriptor.
   297  			var dacl uintptr
   298  			if err := rtlDefaultNpAcl(&dacl).Err(); err != nil {
   299  				return 0, fmt.Errorf("getting default named pipe ACL: %w", err)
   300  			}
   301  			defer localFree(dacl)
   302  
   303  			sdb := &securityDescriptor{
   304  				Revision: 1,
   305  				Control:  windows.SE_DACL_PRESENT,
   306  				Dacl:     dacl,
   307  			}
   308  			oa.SecurityDescriptor = sdb
   309  		}
   310  	}
   311  
   312  	typ := uint32(windows.FILE_PIPE_REJECT_REMOTE_CLIENTS)
   313  	if c.MessageMode {
   314  		typ |= windows.FILE_PIPE_MESSAGE_TYPE
   315  	}
   316  
   317  	disposition := uint32(windows.FILE_OPEN)
   318  	access := uint32(syscall.GENERIC_READ | syscall.GENERIC_WRITE | syscall.SYNCHRONIZE)
   319  	if first {
   320  		disposition = windows.FILE_CREATE
   321  		// By not asking for read or write access, the named pipe file system
   322  		// will put this pipe into an initially disconnected state, blocking
   323  		// client connections until the next call with first == false.
   324  		access = syscall.SYNCHRONIZE
   325  	}
   326  
   327  	timeout := int64(-50 * 10000) // 50ms
   328  
   329  	var (
   330  		h    syscall.Handle
   331  		iosb ioStatusBlock
   332  	)
   333  	err = ntCreateNamedPipeFile(&h,
   334  		access,
   335  		&oa,
   336  		&iosb,
   337  		syscall.FILE_SHARE_READ|syscall.FILE_SHARE_WRITE,
   338  		disposition,
   339  		0,
   340  		typ,
   341  		0,
   342  		0,
   343  		0xffffffff,
   344  		uint32(c.InputBufferSize),
   345  		uint32(c.OutputBufferSize),
   346  		&timeout).Err()
   347  	if err != nil {
   348  		return 0, &os.PathError{Op: "open", Path: path, Err: err}
   349  	}
   350  
   351  	runtime.KeepAlive(ntPath)
   352  	return h, nil
   353  }
   354  
   355  func (l *win32PipeListener) makeServerPipe() (*win32File, error) {
   356  	h, err := makeServerPipeHandle(l.path, nil, &l.config, false)
   357  	if err != nil {
   358  		return nil, err
   359  	}
   360  	f, err := makeWin32File(h)
   361  	if err != nil {
   362  		syscall.Close(h)
   363  		return nil, err
   364  	}
   365  	return f, nil
   366  }
   367  
   368  func (l *win32PipeListener) makeConnectedServerPipe() (*win32File, error) {
   369  	p, err := l.makeServerPipe()
   370  	if err != nil {
   371  		return nil, err
   372  	}
   373  
   374  	// Wait for the client to connect.
   375  	ch := make(chan error)
   376  	go func(p *win32File) {
   377  		ch <- connectPipe(p)
   378  	}(p)
   379  
   380  	select {
   381  	case err = <-ch:
   382  		if err != nil {
   383  			p.Close()
   384  			p = nil
   385  		}
   386  	case <-l.closeCh:
   387  		// Abort the connect request by closing the handle.
   388  		p.Close()
   389  		p = nil
   390  		err = <-ch
   391  		if err == nil || err == ErrFileClosed { //nolint:errorlint // err is Errno
   392  			err = ErrPipeListenerClosed
   393  		}
   394  	}
   395  	return p, err
   396  }
   397  
   398  func (l *win32PipeListener) listenerRoutine() {
   399  	closed := false
   400  	for !closed {
   401  		select {
   402  		case <-l.closeCh:
   403  			closed = true
   404  		case responseCh := <-l.acceptCh:
   405  			var (
   406  				p   *win32File
   407  				err error
   408  			)
   409  			for {
   410  				p, err = l.makeConnectedServerPipe()
   411  				// If the connection was immediately closed by the client, try
   412  				// again.
   413  				if err != windows.ERROR_NO_DATA { //nolint:errorlint // err is Errno
   414  					break
   415  				}
   416  			}
   417  			responseCh <- acceptResponse{p, err}
   418  			closed = err == ErrPipeListenerClosed //nolint:errorlint // err is Errno
   419  		}
   420  	}
   421  	syscall.Close(l.firstHandle)
   422  	l.firstHandle = 0
   423  	// Notify Close() and Accept() callers that the handle has been closed.
   424  	close(l.doneCh)
   425  }
   426  
   427  // PipeConfig contain configuration for the pipe listener.
   428  type PipeConfig struct {
   429  	// SecurityDescriptor contains a Windows security descriptor in SDDL format.
   430  	SecurityDescriptor string
   431  
   432  	// MessageMode determines whether the pipe is in byte or message mode. In either
   433  	// case the pipe is read in byte mode by default. The only practical difference in
   434  	// this implementation is that CloseWrite() is only supported for message mode pipes;
   435  	// CloseWrite() is implemented as a zero-byte write, but zero-byte writes are only
   436  	// transferred to the reader (and returned as io.EOF in this implementation)
   437  	// when the pipe is in message mode.
   438  	MessageMode bool
   439  
   440  	// InputBufferSize specifies the size of the input buffer, in bytes.
   441  	InputBufferSize int32
   442  
   443  	// OutputBufferSize specifies the size of the output buffer, in bytes.
   444  	OutputBufferSize int32
   445  }
   446  
   447  // ListenPipe creates a listener on a Windows named pipe path, e.g. \\.\pipe\mypipe.
   448  // The pipe must not already exist.
   449  func ListenPipe(path string, c *PipeConfig) (net.Listener, error) {
   450  	var (
   451  		sd  []byte
   452  		err error
   453  	)
   454  	if c == nil {
   455  		c = &PipeConfig{}
   456  	}
   457  	if c.SecurityDescriptor != "" {
   458  		sd, err = SddlToSecurityDescriptor(c.SecurityDescriptor)
   459  		if err != nil {
   460  			return nil, err
   461  		}
   462  	}
   463  	h, err := makeServerPipeHandle(path, sd, c, true)
   464  	if err != nil {
   465  		return nil, err
   466  	}
   467  	l := &win32PipeListener{
   468  		firstHandle: h,
   469  		path:        path,
   470  		config:      *c,
   471  		acceptCh:    make(chan (chan acceptResponse)),
   472  		closeCh:     make(chan int),
   473  		doneCh:      make(chan int),
   474  	}
   475  	go l.listenerRoutine()
   476  	return l, nil
   477  }
   478  
   479  func connectPipe(p *win32File) error {
   480  	c, err := p.prepareIO()
   481  	if err != nil {
   482  		return err
   483  	}
   484  	defer p.wg.Done()
   485  
   486  	err = connectNamedPipe(p.handle, &c.o)
   487  	_, err = p.asyncIO(c, nil, 0, err)
   488  	if err != nil && err != windows.ERROR_PIPE_CONNECTED { //nolint:errorlint // err is Errno
   489  		return err
   490  	}
   491  	return nil
   492  }
   493  
   494  func (l *win32PipeListener) Accept() (net.Conn, error) {
   495  	ch := make(chan acceptResponse)
   496  	select {
   497  	case l.acceptCh <- ch:
   498  		response := <-ch
   499  		err := response.err
   500  		if err != nil {
   501  			return nil, err
   502  		}
   503  		if l.config.MessageMode {
   504  			return &win32MessageBytePipe{
   505  				win32Pipe: win32Pipe{win32File: response.f, path: l.path},
   506  			}, nil
   507  		}
   508  		return &win32Pipe{win32File: response.f, path: l.path}, nil
   509  	case <-l.doneCh:
   510  		return nil, ErrPipeListenerClosed
   511  	}
   512  }
   513  
   514  func (l *win32PipeListener) Close() error {
   515  	select {
   516  	case l.closeCh <- 1:
   517  		<-l.doneCh
   518  	case <-l.doneCh:
   519  	}
   520  	return nil
   521  }
   522  
   523  func (l *win32PipeListener) Addr() net.Addr {
   524  	return pipeAddress(l.path)
   525  }
   526
View as plain text