diff --git a/runsc/boot/portforward/BUILD b/runsc/boot/portforward/BUILD index af925b918..2d6ed3934 100644 --- a/runsc/boot/portforward/BUILD +++ b/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", ], ) diff --git a/runsc/boot/portforward/portforward.go b/runsc/boot/portforward/portforward.go index b2e31f63e..687c97897 100644 --- a/runsc/boot/portforward/portforward.go +++ b/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()) +} diff --git a/runsc/boot/portforward/portforward_fd_rw_test.go b/runsc/boot/portforward/portforward_fd_rw_test.go index 4e748da30..171393dee 100644 --- a/runsc/boot/portforward/portforward_fd_rw_test.go +++ b/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:] } diff --git a/runsc/boot/portforward/portforward_hostinet.go b/runsc/boot/portforward/portforward_hostinet.go new file mode 100644 index 000000000..023126a0a --- /dev/null +++ b/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) +} diff --git a/runsc/boot/portforward/portforward_hostinet_test.go b/runsc/boot/portforward/portforward_hostinet_test.go new file mode 100644 index 000000000..dbe0ca9db --- /dev/null +++ b/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) + } + } +}