Implement hostinet port forward

Implement support sandboxes using hostinet for their network stack. Included
is an implmentation of a connection servering a sandboxed process forwarding
to a socket on a local port.

PiperOrigin-RevId: 502692640

runsc/boot/portforward/BUILD

@@ -7,13 +7,19 @@ go_library(
     srcs = [
         "portforward.go",
         "portforward_fd_rw.go",
+        "portforward_hostinet.go",
     ],
     deps = [
+        "//pkg/cleanup",
         "//pkg/context",
         "//pkg/errors/linuxerr",
+        "//pkg/fd",
+        "//pkg/fdnotifier",
+        "//pkg/log",
         "//pkg/sentry/vfs",
         "//pkg/usermem",
         "//pkg/waiter",
+        "@org_golang_x_sys//unix:go_default_library",
     ],
 )
 
@@ -21,8 +27,13 @@ go_test(
     name = "portforward_test",
     srcs = [
         "portforward_fd_rw_test.go",
+        "portforward_hostinet_test.go",
     ],
     library = ":portforward",
+    tags = [
+        "requires-net:ipv4",
+        "requires-net:loopback",
+    ],
     deps = [
         "//pkg/abi/linux",
         "//pkg/context",
@@ -31,5 +42,6 @@ go_test(
         "//pkg/sentry/vfs",
         "//pkg/usermem",
         "//pkg/waiter",
+        "@org_golang_x_sync//errgroup:go_default_library",
     ],
 )

runsc/boot/portforward/portforward.go

@@ -14,3 +14,18 @@
 
 // Package portforward holds the infrastructure to support the port forward command.
 package portforward
+
+import (
+	"gvisor.dev/gvisor/pkg/context"
+)
+
+// portForwardConn is a port forwarding connection. It is used to manage the
+// lifecycle of the connection and clean it up if necessary.
+type portForwardConn interface {
+	// start starts the connection goroutines and returns.
+	start(ctx context.Context) error
+	// close closes and cleans up the connection.
+	close(ctx context.Context) error
+	// cleanup registers a callback for when the connection closes.
+	cleanup(func())
+}

runsc/boot/portforward/portforward_fd_rw_test.go

@@ -103,15 +103,12 @@ var _ vfs.FileDescriptionImpl = (*readerWriter)(nil)
 // Read implements vfs.FileDescriptionImpl.Read details for the parent mockFileDescription.
 func (rw *readerWriter) Read(ctx context.Context, dst usermem.IOSequence, opts vfs.ReadOptions) (int64, error) {
 	if rw.released {
-		return 0, nil
-	}
-	if rw.buf.Len() == 0 {
 		return 0, io.EOF
 	}
 	buf := make([]byte, dst.NumBytes())
 	_, err := rw.buf.Read(buf)
 	if err != nil {
-		return 0, err
+		return 0, nil
 	}
 	n, err := dst.CopyOut(ctx, buf)
 	return int64(n), err
@@ -135,7 +132,9 @@ func (rw *readerWriter) EventRegister(we *waiter.Entry) error { return fmt.Error
 func (rw *readerWriter) EventUnregister(we *waiter.Entry) { panic("not implemented") }
 
 // Release implements vfs.FileDescriptionImpl.Release details for the parent mockFileDescription.
-func (rw *readerWriter) Release(context.Context) { rw.released = true }
+func (rw *readerWriter) Release(context.Context) {
+	rw.released = true
+}
 
 // waiterRW implements mockFileDescriptionRWImpl. waiterRW works the same way as readerWriter above,
 // but it interleaves blocks in between Read and Write calls.
@@ -168,7 +167,7 @@ func (w *waiterRW) Read(ctx context.Context, dst usermem.IOSequence, opts vfs.Re
 	w.waitMu.Lock()
 	defer w.waitMu.Unlock()
 	if w.closed {
-		return 0, nil
+		return 0, io.EOF
 	}
 	if w.shouldWait {
 		return 0, linuxerr.ErrWouldBlock
@@ -298,10 +297,16 @@ func TestReaderWriter(t *testing.T) {
 			got := []byte{}
 			buf := make([]byte, 4)
 			for {
-				_, err := readerWriter.Read(buf)
+				n, err := readerWriter.Read(buf)
 				if err == io.EOF {
 					break
 				}
+				if err != nil {
+					t.Fatalf("read failed: %v", err)
+				}
+				if n == 0 {
+					break
+				}
 				got = append(got, buf...)
 				buf = buf[0:]
 			}

runsc/boot/portforward/portforward_hostinet.go

@@ -0,0 +1,284 @@
+// Copyright 2023 The gVisor Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//	http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package portforward
+
+import (
+	"fmt"
+	"io"
+	"sync"
+
+	"golang.org/x/sys/unix"
+	"gvisor.dev/gvisor/pkg/cleanup"
+	"gvisor.dev/gvisor/pkg/context"
+	fileDescriptor "gvisor.dev/gvisor/pkg/fd"
+	"gvisor.dev/gvisor/pkg/fdnotifier"
+	"gvisor.dev/gvisor/pkg/log"
+	"gvisor.dev/gvisor/pkg/sentry/vfs"
+	"gvisor.dev/gvisor/pkg/waiter"
+)
+
+var (
+	localHost = [4]byte{127, 0, 0, 1}
+)
+
+// localHostSocket allows reading and writing to a local host socket for hostinet.
+type localHostSocket struct {
+	// wq is the WaitQueue registered with fdnotifier for this fd.
+	wq waiter.Queue
+	// fd is the file descriptor for the socket.
+	fd *fileDescriptor.FD
+}
+
+// newLocalHostSocket creates a hostSocket for an FD and registers the fd for
+// notifications.
+func newLocalHostSocket() (*localHostSocket, error) {
+	// NOTE: Options must match sandbox seccomp filters. See filter/config.go
+	fd, err := unix.Socket(unix.AF_INET, unix.SOCK_STREAM|unix.SOCK_NONBLOCK|unix.SOCK_CLOEXEC, 0)
+	if err != nil {
+		return nil, err
+	}
+	s := localHostSocket{
+		fd: fileDescriptor.New(fd),
+	}
+	if err := fdnotifier.AddFD(int32(s.fd.FD()), &s.wq); err != nil {
+		return nil, err
+	}
+	return &s, nil
+}
+
+// Connect performs a blocking connect on the socket to an ipv4 address.
+func (s *localHostSocket) Connect(port uint16) error {
+	sockAddr := &unix.SockaddrInet4{
+		Addr: localHost,
+		Port: int(port),
+	}
+
+	if err := unix.Connect(s.fd.FD(), sockAddr); err != nil {
+		if err != unix.EINPROGRESS {
+			return err
+		}
+
+		// Connect is in progress. Wait for the socket to be writable.
+		mask := waiter.WritableEvents
+		waitEntry, notifyCh := waiter.NewChannelEntry(mask)
+		s.eventRegister(&waitEntry)
+		defer s.eventUnregister(&waitEntry)
+
+		// Wait for connect to succeed.
+		// Check the current socket state and if not ready, wait for the event.
+		if fdnotifier.NonBlockingPoll(int32(s.fd.FD()), mask)&mask == 0 {
+			<-notifyCh
+		}
+
+		// Call getsockopt to get the connection result.
+		val, err := unix.GetsockoptInt(s.fd.FD(), unix.SOL_SOCKET, unix.SO_ERROR)
+		if err != nil {
+			return nil
+		}
+		if val != 0 {
+			return unix.Errno(val)
+		}
+	}
+
+	return nil
+}
+
+// Read implements io.Reader.Read. It performs a blocking read on the fd.
+func (s *localHostSocket) Read(buf []byte) (int, error) {
+	var ch chan struct{}
+	var e waiter.Entry
+	n, err := s.fd.Read(buf)
+	for err == unix.EWOULDBLOCK {
+		if ch == nil {
+			e, ch = waiter.NewChannelEntry(waiter.ReadableEvents | waiter.WritableEvents | waiter.EventHUp | waiter.EventErr)
+			// Register for when the endpoint is writable or disconnected.
+			s.eventRegister(&e)
+			defer s.eventUnregister(&e)
+		}
+		<-ch
+		n, err = s.fd.Read(buf)
+	}
+	return n, err
+}
+
+// Write implements io.Writer.Write. It performs a blocking write on the fd.
+func (s *localHostSocket) Write(buf []byte) (int, error) {
+	var ch chan struct{}
+	var e waiter.Entry
+	n, err := s.fd.Write(buf)
+	for err == unix.EWOULDBLOCK {
+		if ch == nil {
+			e, ch = waiter.NewChannelEntry(waiter.WritableEvents | waiter.EventHUp | waiter.EventErr)
+			// Register for when the endpoint is writable or disconnected.
+			s.eventRegister(&e)
+			defer s.eventUnregister(&e)
+		}
+		<-ch
+		n, err = s.fd.Write(buf)
+	}
+	return n, err
+}
+
+func (s *localHostSocket) eventRegister(e *waiter.Entry) {
+	s.wq.EventRegister(e)
+	fdnotifier.UpdateFD(int32(s.fd.FD()))
+}
+
+func (s *localHostSocket) eventUnregister(e *waiter.Entry) {
+	s.wq.EventUnregister(e)
+	fdnotifier.UpdateFD(int32(s.fd.FD()))
+}
+
+// Close closes the host socket and removes it from notifications.
+func (s *localHostSocket) Close() {
+	fdnotifier.RemoveFD(int32(s.fd.FD()))
+	s.fd.Close()
+}
+
+// hostinetportForwardConn is a hostinet port forwarding connection.
+type hostinetPortForwardConn struct {
+	// cid is the container id that this connection is connecting to.
+	cid string
+
+	// Socket is the host socket connected to the application.
+	socket *localHostSocket
+	// fd is the FileDescription for the imported host UDS fd.
+	fd *vfs.FileDescription
+
+	// status holds the status of the connection.
+	status struct {
+		sync.Mutex
+		// started indicates if the connection is started or not.
+		started bool
+		// closed indicates if the connection is closed or not.
+		closed bool
+	}
+
+	// toDone is closed when the copy to the application port is finished.
+	toDone chan struct{}
+
+	// fromDone is closed when the copy from the application socket is finished.
+	fromDone chan struct{}
+
+	// cu is called when the connection finishes.
+	cu cleanup.Cleanup
+}
+
+// newHostinetPortForward starts port forwarding to the given port in hostinet
+// mode.
+func newHostinetPortForward(ctx context.Context, cid string, fd *vfs.FileDescription, port uint16) (portForwardConn, error) {
+	log.Debugf("Handling hostinet port forwarding request for %s on port %d", cid, port)
+	appSocket, err := newLocalHostSocket()
+	if err != nil {
+		return nil, fmt.Errorf("hostinet socket: %w", err)
+	}
+
+	cu := cleanup.Make(func() { appSocket.Close() })
+	defer cu.Clean()
+
+	if err := appSocket.Connect(port); err != nil {
+		return nil, fmt.Errorf("hostinet connect: %w", err)
+	}
+
+	pfConn := hostinetPortForwardConn{
+		cid:      cid,
+		socket:   appSocket,
+		fd:       fd,
+		toDone:   make(chan struct{}),
+		fromDone: make(chan struct{}),
+		cu:       cleanup.Cleanup{},
+	}
+
+	cu.Release()
+	return &pfConn, nil
+}
+
+// Start implements portForwardConn.start.
+func (c *hostinetPortForwardConn) start(ctx context.Context) error {
+	c.status.Lock()
+	defer c.status.Unlock()
+
+	if c.status.closed {
+		return fmt.Errorf("already closed")
+	}
+	if c.status.started {
+		return fmt.Errorf("already started")
+	}
+
+	log.Debugf("Start forwarding to/from container %q and localhost", c.cid)
+
+	importedRW := &fileDescriptionReadWriter{
+		file: c.fd,
+	}
+
+	go func() {
+		_, _ = io.Copy(c.socket, importedRW)
+		// Indicate that this goroutine has completed.
+		close(c.toDone)
+		// Make sure to clean up when one half of the copy has finished.
+		c.close(ctx)
+	}()
+	go func() {
+		_, _ = io.Copy(importedRW, c.socket)
+		// Indicate that this goroutine has completed.
+		close(c.fromDone)
+		// Make sure to clean up when one half of the copy has finished.
+		c.close(ctx)
+	}()
+
+	c.status.started = true
+
+	return nil
+}
+
+// close implements portForwardConn.close.
+func (c *hostinetPortForwardConn) close(ctx context.Context) error {
+	c.status.Lock()
+
+	// This should be a no op if the connection is already closed.
+	if c.status.closed {
+		c.status.Unlock()
+		return nil
+	}
+
+	log.Debugf("Stopping forwarding to/from container %q and localhost...", c.cid)
+
+	// Closing the FileDescription and endpoint should make all
+	// goroutines exit.
+	c.fd.DecRef(ctx)
+	c.socket.Close()
+
+	// Wait for one goroutine to finish or for a save event.
+	<-c.toDone
+	log.Debugf("Stopped forwarding one-half of copy for %q", c.cid)
+
+	// Wait on the other goroutine.
+	<-c.fromDone
+	log.Debugf("Stopped forwarding to/from container %q and localhost", c.cid)
+
+	c.status.closed = true
+
+	c.status.Unlock()
+
+	// Call the cleanup object.
+	c.cu.Clean()
+
+	return nil
+}
+
+// cleanup implements portForwardConn.cleanup.
+func (c *hostinetPortForwardConn) cleanup(f func()) {
+	c.cu.Add(f)
+}

runsc/boot/portforward/portforward_hostinet_test.go

@@ -0,0 +1,311 @@
+// Copyright 2023 The gVisor Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//	http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package portforward
+
+import (
+	"bytes"
+	"fmt"
+	"io"
+	"net"
+	"reflect"
+	"sync"
+	"testing"
+
+	"golang.org/x/sync/errgroup"
+	"gvisor.dev/gvisor/pkg/context"
+	"gvisor.dev/gvisor/pkg/errors/linuxerr"
+	"gvisor.dev/gvisor/pkg/sentry/contexttest"
+	"gvisor.dev/gvisor/pkg/sentry/vfs"
+	"gvisor.dev/gvisor/pkg/usermem"
+	"gvisor.dev/gvisor/pkg/waiter"
+)
+
+func TestLocalHostSocket(t *testing.T) {
+
+	clientData := append(
+		[]byte("do what must be done\n"),
+		[]byte("do not hesitate\n")...,
+	)
+
+	serverData := append(
+		[]byte("commander cody...the time has come\n"),
+		[]byte("execute order 66\n")...,
+	)
+
+	l, err := net.Listen("tcp", ":0")
+	if err != nil {
+		t.Fatalf("net.Listen failed: %v", err)
+	}
+	defer l.Close()
+
+	port := l.Addr().(*net.TCPAddr).Port
+
+	var g errgroup.Group
+
+	g.Go(func() error {
+		conn, err := l.Accept()
+		if err != nil {
+			return fmt.Errorf("could not accept connection: %v", err)
+		}
+		defer conn.Close()
+
+		data := make([]byte, 1024)
+		recLen, err := conn.Read(data)
+		if err != nil {
+			return fmt.Errorf("could not read data: %v", err)
+		}
+
+		if !reflect.DeepEqual(data[:recLen], clientData) {
+			return fmt.Errorf("server mismatch data recieved: got: %s want: %s", data[:recLen], clientData)
+		}
+
+		sentLen, err := conn.Write(serverData)
+		if err != nil {
+			return fmt.Errorf("could not write data: %v", err)
+		}
+
+		if sentLen != len(serverData) {
+			return fmt.Errorf("server mismatch data sent: got: %d want: %d", sentLen, len(serverData))
+		}
+
+		return nil
+	})
+
+	g.Go(func() error {
+		sock, err := newLocalHostSocket()
+		if err != nil {
+			return fmt.Errorf("could not create local host socket: %v", err)
+		}
+		defer sock.Close()
+		if err := sock.Connect(uint16(port)); err != nil {
+			return fmt.Errorf("could not connect to local host socket: %v", err)
+		}
+		for i := 0; i < len(clientData); {
+			n, err := sock.Write(clientData[i:])
+			if err != nil {
+				return fmt.Errorf("could not write to local host socket: %v", err)
+			}
+			i += n
+		}
+
+		data := make([]byte, 1024)
+		dataLen := 0
+		for dataLen < len(serverData) {
+			n, err := sock.Read(data[dataLen:])
+			if err != nil {
+				t.Fatalf("could not read from local host socket: %v", err)
+			}
+			dataLen += n
+		}
+
+		if !reflect.DeepEqual(data[:dataLen], serverData) {
+			return fmt.Errorf("server mismatch data received: got: %s want: %s", data[:dataLen], clientData)
+		}
+		return nil
+	})
+
+	if err := g.Wait(); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func newMockSocketPair() (*mockEndpoint, *mockEndpoint) {
+	client := &mockEndpoint{}
+	server := &mockEndpoint{other: client}
+	client.other = server
+	return client, server
+}
+
+type mockEndpoint struct {
+	vfs.FileDescriptionDefaultImpl
+	vfs.NoLockFD
+	vfs.DentryMetadataFileDescriptionImpl
+	other    *mockEndpoint
+	readBuf  bytes.Buffer
+	mu       sync.Mutex
+	released bool
+	queue    waiter.Queue
+}
+
+var _ vfs.FileDescriptionImpl = (*mockEndpoint)(nil)
+
+// Read implements vfs.FileDescriptionImpl.Read details for the parent mockFileDescription.
+func (s *mockEndpoint) Read(ctx context.Context, dst usermem.IOSequence, opts vfs.ReadOptions) (int64, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if s.released {
+		return 0, io.EOF
+	}
+	if s.readBuf.Len() == 0 {
+		return 0, linuxerr.ErrWouldBlock
+	}
+	buf := s.readBuf.Next(s.readBuf.Len())
+	n, err := dst.CopyOut(ctx, buf)
+	s.queue.Notify(waiter.WritableEvents)
+	return int64(n), err
+}
+
+// Write implements vfs.FileDescriptionImpl.Write details for the parent mockFileDescription.
+func (s *mockEndpoint) Write(ctx context.Context, src usermem.IOSequence, opts vfs.WriteOptions) (int64, error) {
+	return s.other.write(ctx, src, opts)
+}
+
+func (s *mockEndpoint) write(ctx context.Context, src usermem.IOSequence, opts vfs.WriteOptions) (int64, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if s.released {
+		return 0, io.EOF
+	}
+	buf := make([]byte, src.NumBytes())
+	n, err := src.CopyIn(ctx, buf)
+	if err != nil {
+		return 0, err
+	}
+	n, err = s.readBuf.Write(buf[:n])
+	s.queue.Notify(waiter.ReadableEvents)
+	return int64(n), err
+}
+
+func (s *mockEndpoint) IsReadable() bool {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if s.released {
+		return false
+	}
+	return s.readBuf.Len() > 0
+}
+
+func (s *mockEndpoint) IsWritable() bool {
+	return s.other.isWritable()
+}
+
+func (s *mockEndpoint) isWritable() bool {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return !s.released
+}
+
+// EventRegister implements vfs.FileDescriptionImpl.EventRegister details for the parent mockFileDescription.
+func (s *mockEndpoint) EventRegister(we *waiter.Entry) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.queue.EventRegister(we)
+	return nil
+}
+
+// EventUnregister implements vfs.FileDescriptionImpl.Unregister details for the parent mockFileDescription.
+func (s *mockEndpoint) EventUnregister(we *waiter.Entry) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.queue.EventUnregister(we)
+}
+
+// Release implements vfs.FileDescriptionImpl.Release details for the parent mockFileDescription.
+func (s *mockEndpoint) Release(context.Context) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.queue.Notify(waiter.ReadableEvents)
+	s.released = true
+}
+
+var responses = map[string]string{
+	"PING": "PONG",
+	"DING": "DONG",
+	"TING": "TONG",
+}
+
+func TestHostinetPortForwardConn(t *testing.T) {
+	ctx := contexttest.Context(t)
+	clientSock, server := newMockSocketPair()
+	defer server.Release(ctx)
+	client, err := newMockFileDescription(ctx, clientSock)
+	if err != nil {
+		t.Fatalf("newMockFileDescription failed: %v", err)
+	}
+	l, err := net.Listen("tcp", ":0")
+	if err != nil {
+		t.Fatalf("net.Listen failed: %v", err)
+	}
+	defer l.Close()
+	port := l.Addr().(*net.TCPAddr).Port
+	portForwardConn, err := newHostinetPortForward(ctx, "", client, uint16(port))
+	if err != nil {
+		t.Fatalf("newHostinetPortForward failed: %v", err)
+	}
+	if err := portForwardConn.start(ctx); err != nil {
+		t.Fatalf("portForwardConn.start failed: %v", err)
+	}
+	conn, err := l.Accept()
+	if err != nil {
+		t.Fatalf("l.Accept failed: %v", err)
+	}
+	defer conn.Close()
+	buf := make([]byte, 4)
+	for req, resp := range responses {
+
+		for {
+			if server.IsWritable() {
+				break
+			}
+		}
+		_, err := server.Write(ctx, usermem.BytesIOSequence([]byte(req)), vfs.WriteOptions{})
+		if err != nil {
+			t.Fatalf("file.Write failed: %v", err)
+		}
+
+		read := 0
+		for {
+			n, err := conn.Read([]byte(buf)[read:])
+			if err != nil && !linuxerr.Equals(linuxerr.ErrWouldBlock, err) {
+				t.Fatalf("conn.Write failed: %v", err)
+			}
+			read += n
+			if read >= len(resp) {
+				break
+			}
+		}
+
+		if string(buf) != req {
+			t.Fatalf("read mismatch: got: %s want: %s", string(buf), req)
+		}
+
+		written := 0
+		for i := 0; i < 4; i++ {
+			n, err := conn.Write([]byte(resp)[written:])
+			if err != nil && !linuxerr.Equals(linuxerr.ErrWouldBlock, err) {
+				t.Fatalf("conn.Write failed: %v", err)
+			}
+			written += n
+			if written >= len(resp) {
+				break
+			}
+		}
+
+		for {
+			if server.IsReadable() {
+				break
+			}
+		}
+
+		_, err = server.Read(ctx, usermem.BytesIOSequence([]byte(buf[:4])), vfs.ReadOptions{})
+		if err != nil && !linuxerr.Equals(linuxerr.ErrWouldBlock, err) {
+			t.Fatalf("file.Read failed: %v", err)
+		}
+
+		if string(buf) != resp {
+			t.Fatalf("write mismatch: got: %s want: %s", string(buf), resp)
+		}
+	}
+}