fuse: add no_create fallback and offer ATOMIC_O_TRUNC/BIG_WRITES in INIT

pkg/sentry/fsimpl/fuse/connection.go

@@ -261,6 +261,10 @@ type connection struct {
 	// noOpen if FUSE server doesn't support open operation.
 	// This flag only influences performance, not correctness of the program.
 	noOpen bool
+
+	// noCreate if FUSE server doesn't support the create operation. Files are
+	// then created with FUSE_MKNOD followed by FUSE_OPEN, as Linux does.
+	noCreate bool
 }
 
 func linuxError(err error) error {
@@ -531,7 +535,7 @@ func (conn *connection) read(ctx context.Context, dst usermem.IOSequence) (int64
 	// read buffer. It must have the capacity for the fixed parts of any request
 	// header (Linux uses the request header and the FUSEWriteIn header for this
 	// calculation) + the negotiated MaxWrite room for the data.
-	negotiatedMinBuffSize := linux.SizeOfFUSEHeaderIn + linux.SizeOfFUSEHeaderOut + conn.maxWrite
+	negotiatedMinBuffSize := linux.SizeOfFUSEHeaderIn + linux.SizeOfFUSEWriteIn + conn.maxWrite
 	if minBuffSize < negotiatedMinBuffSize {
 		minBuffSize = negotiatedMinBuffSize
 	}

pkg/sentry/fsimpl/fuse/connection_control.go

@@ -42,7 +42,13 @@ const (
 
 	// The FUSE_INIT_IN flags sent to the daemon.
 	// TODO(gvisor.dev/issue/3199): complete the flags.
-	fuseDefaultInitFlags = linux.FUSE_MAX_PAGES
+	fuseDefaultInitFlags = linux.FUSE_MAX_PAGES | linux.FUSE_ATOMIC_O_TRUNC | linux.FUSE_BIG_WRITES
+
+	// fuseDatagramUnsafeInitFlags are flags that cannot be offered over a
+	// transport that carries each request in a single datagram: a
+	// max_write-sized FUSE_WRITE exceeds the default datagram size limit and
+	// cannot be fragmented.
+	fuseDatagramUnsafeInitFlags = linux.FUSE_BIG_WRITES
 
 	// An INIT response needs to be at least this long.
 	minInitSize = 24

pkg/sentry/fsimpl/fuse/host_connection.go

@@ -212,7 +212,9 @@ func (hc *hostConnection) InitSend(creds *auth.Credentials, pid uint32, hasSysAd
 		Major:        linux.FUSE_KERNEL_VERSION,
 		Minor:        linux.FUSE_KERNEL_MINOR_VERSION,
 		MaxReadahead: fuseDefaultMaxReadahead,
-		Flags:        fuseDefaultInitFlags,
+		// Each request is a single SOCK_SEQPACKET datagram, which
+		// writeRequest cannot fragment.
+		Flags: fuseDefaultInitFlags &^ fuseDatagramUnsafeInitFlags,
 	}
 
 	req := hc.conn.NewRequest(creds, pid, 0, linux.FUSE_INIT, &in)
@@ -244,6 +246,9 @@ func (hc *hostConnection) InitSend(creds *auth.Credentials, pid uint32, hasSysAd
 	if err := hc.conn.InitRecv(res, hasSysAdminCap); err != nil {
 		return err
 	}
+	// Big writes stay off even if the server echoes the flag unsolicited;
+	// the transport cannot carry them.
+	hc.conn.bigWrites = false
 
 	hc.startReader()
 	return nil

pkg/sentry/fsimpl/fuse/host_connection_integration_test.go

@@ -35,6 +35,11 @@ type testFUSEServer struct {
 	backDir   string
 	nextFh    uint64
 	openFiles map[uint64]*os.File
+
+	// gotInitFlags records the flags offered in the FUSE_INIT request.
+	gotInitFlags uint32
+	// initReply, when non-nil, overrides the FUSE_INIT reply.
+	initReply *linux.FUSEInitOut
 }
 
 func newTestFUSEServer(fd int, backDir string) *testFUSEServer {
@@ -78,7 +83,7 @@ func (s *testFUSEServer) serve(t *testing.T, done chan struct{}) {
 func (s *testFUSEServer) handleRequest(hdr *linux.FUSEHeaderIn, payload []byte) []byte {
 	switch hdr.Opcode {
 	case linux.FUSE_INIT:
-		return s.handleInit(hdr)
+		return s.handleInit(hdr, payload)
 	case linux.FUSE_GETATTR:
 		return s.handleGetAttr(hdr)
 	case linux.FUSE_LOOKUP:
@@ -100,12 +105,18 @@ func (s *testFUSEServer) handleRequest(hdr *linux.FUSEHeaderIn, payload []byte)
 	}
 }
 
-func (s *testFUSEServer) handleInit(hdr *linux.FUSEHeaderIn) []byte {
+func (s *testFUSEServer) handleInit(hdr *linux.FUSEHeaderIn, payload []byte) []byte {
+	var in linux.FUSEInitIn
+	in.UnmarshalUnsafe(payload)
+	s.gotInitFlags = in.Flags
 	out := linux.FUSEInitOut{
 		Major:    linux.FUSE_KERNEL_VERSION,
 		Minor:    linux.FUSE_KERNEL_MINOR_VERSION,
 		MaxWrite: 65536,
 	}
+	if s.initReply != nil {
+		out = *s.initReply
+	}
 	return s.marshalReply(hdr, &out)
 }
 
@@ -502,3 +513,78 @@ func TestHostFUSEWriteFile(t *testing.T) {
 		t.Fatalf("backing file: got %q, want %q", string(got), string(writeData))
 	}
 }
+
+// newTestHostFUSEConnectionWithReply is newTestHostFUSEConnection with a
+// caller-provided FUSE_INIT reply, returning the server for inspection.
+func newTestHostFUSEConnectionWithReply(t *testing.T, backDir string, reply *linux.FUSEInitOut) (*hostConnection, *testFUSEServer, func()) {
+	t.Helper()
+
+	fds, err := unix.Socketpair(unix.AF_UNIX, unix.SOCK_SEQPACKET, 0)
+	if err != nil {
+		t.Fatalf("Socketpair: %v", err)
+	}
+
+	server := newTestFUSEServer(fds[1], backDir)
+	server.initReply = reply
+	serverDone := make(chan struct{})
+	go server.serve(t, serverDone)
+
+	fsopts := filesystemOptions{
+		maxActiveRequests: maxActiveRequestsDefault,
+		maxRead:           65536,
+	}
+	conn, err := newFUSEConnectionOpts(&fsopts)
+	if err != nil {
+		unix.Close(fds[0])
+		unix.Close(fds[1])
+		t.Fatalf("newFUSEConnectionOpts: %v", err)
+	}
+	hc := newHostConnection(conn, int32(fds[0]))
+
+	cleanup := func() {
+		unix.Shutdown(fds[1], unix.SHUT_RDWR)
+		unix.Shutdown(fds[0], unix.SHUT_RDWR)
+		<-serverDone
+		unix.Close(fds[0])
+		unix.Close(fds[1])
+	}
+	return hc, server, cleanup
+}
+
+// TestHostFUSEBigWritesNotNegotiated verifies that the host passthrough
+// connection never negotiates FUSE_BIG_WRITES: each request travels as one
+// SOCK_SEQPACKET datagram, which cannot carry a max_write-sized FUSE_WRITE.
+// The server here misbehaves by echoing FUSE_BIG_WRITES without it being
+// offered; big writes must stay off regardless.
+func TestHostFUSEBigWritesNotNegotiated(t *testing.T) {
+	s := setup(t)
+	defer s.Destroy()
+
+	backDir := t.TempDir()
+	hc, server, cleanup := newTestHostFUSEConnectionWithReply(t, backDir, &linux.FUSEInitOut{
+		Major:    linux.FUSE_KERNEL_VERSION,
+		Minor:    linux.FUSE_KERNEL_MINOR_VERSION,
+		MaxWrite: 1 << 20,
+		MaxPages: 256,
+		Flags:    linux.FUSE_MAX_PAGES | linux.FUSE_ATOMIC_O_TRUNC | linux.FUSE_BIG_WRITES,
+	})
+	defer cleanup()
+
+	creds := auth.CredentialsFromContext(s.Ctx)
+	if err := hc.InitSend(creds, 1, true); err != nil {
+		t.Fatalf("InitSend: %v", err)
+	}
+
+	if server.gotInitFlags&linux.FUSE_BIG_WRITES != 0 {
+		t.Errorf("INIT offered FUSE_BIG_WRITES (flags %#x); the host connection must not offer it", server.gotInitFlags)
+	}
+	if server.gotInitFlags&linux.FUSE_ATOMIC_O_TRUNC == 0 {
+		t.Errorf("INIT did not offer FUSE_ATOMIC_O_TRUNC (flags %#x)", server.gotInitFlags)
+	}
+	if hc.conn.bigWrites {
+		t.Error("bigWrites negotiated on a host connection despite not being offered")
+	}
+	if !hc.conn.atomicOTrunc {
+		t.Error("atomicOTrunc not negotiated; only FUSE_BIG_WRITES should be withheld")
+	}
+}

pkg/sentry/fsimpl/fuse/inode.go

@@ -540,10 +540,14 @@ func (i *inode) Open(ctx context.Context, rp *vfs.ResolvingPath, d *kernfs.Dentr
 	// FOPEN_KEEP_CACHE is the default flag for noOpen.
 	fd.OpenFlag = linux.FOPEN_KEEP_CACHE
 
+	// An open request is sent unless the file handle was already returned by
+	// FUSE_CREATE, or the server does not support open for regular files.
+	willSendOpen := !i.fh.new && (!i.fs.conn.noOpen || i.filemode().IsDir())
+
 	truncateRegFile := opts.Flags&linux.O_TRUNC != 0 && i.filemode().FileType() == linux.S_IFREG
-	if truncateRegFile && (i.fh.new || !i.fs.conn.atomicOTrunc) {
+	if truncateRegFile && !(willSendOpen && i.fs.conn.atomicOTrunc) {
 		// If the regular file needs to be truncated, but the connection doesn't
-		// support O_TRUNC or if we are optimizing away the Open RPC, then manually
+		// support O_TRUNC or if no Open RPC will be sent, then manually
 		// truncate the file *before* Open. As per libfuse, "If [atomic O_TRUNC is]
 		// disabled, and an application specifies O_TRUNC, fuse first calls
 		// truncate() and then open() with O_TRUNC filtered out.".
@@ -557,9 +561,7 @@ func (i *inode) Open(ctx context.Context, rp *vfs.ResolvingPath, d *kernfs.Dentr
 		fd.OpenFlag = i.fh.flags
 		fd.Fh = i.fh.handle
 		i.fh.new = false
-		// Only send an open request when the FUSE server supports open or is
-		// opening a directory.
-	} else if !i.fs.conn.noOpen || i.filemode().IsDir() {
+	} else if willSendOpen {
 		in := linux.FUSEOpenIn{Flags: opts.Flags & ^uint32(linux.O_CREAT|linux.O_EXCL|linux.O_NOCTTY)}
 		// Clear O_TRUNC if the server doesn't support it.
 		if !i.fs.conn.atomicOTrunc {
@@ -570,6 +572,13 @@ func (i *inode) Open(ctx context.Context, rp *vfs.ResolvingPath, d *kernfs.Dentr
 		if err := i.call(ctx, opcode, &in, &out); err != nil {
 			if linuxerr.Equals(linuxerr.ENOSYS, err) && !i.filemode().IsDir() {
 				i.fs.conn.noOpen = true
+				// The open that was to carry O_TRUNC was refused; truncate with SETATTR.
+				if truncateRegFile && i.fs.conn.atomicOTrunc {
+					opts := vfs.SetStatOptions{Stat: linux.Statx{Size: 0, Mask: linux.STATX_SIZE}}
+					if err := i.setAttr(ctx, i.fs.VFSFilesystem(), auth.CredentialsFromContext(ctx), opts, fhOptions{useFh: false}); err != nil {
+						return nil, err
+					}
+				}
 			} else {
 				return nil, err
 			}
@@ -670,15 +679,31 @@ func (*inode) IterDirents(ctx context.Context, mnt *vfs.Mount, callback vfs.Iter
 func (i *inode) NewFile(ctx context.Context, name string, opts vfs.OpenOptions) (kernfs.Inode, error) {
 	opts.Flags &= linux.O_ACCMODE | linux.O_CREAT | linux.O_EXCL | linux.O_TRUNC |
 		linux.O_DIRECTORY | linux.O_NOFOLLOW | linux.O_NONBLOCK | linux.O_NOCTTY
-	in := linux.FUSECreateIn{
-		CreateMeta: linux.FUSECreateMeta{
-			Flags: opts.Flags,
+	if !i.fs.conn.noCreate {
+		in := linux.FUSECreateIn{
+			CreateMeta: linux.FUSECreateMeta{
+				Flags: opts.Flags,
+				Mode:  uint32(opts.Mode) | linux.S_IFREG,
+				Umask: umaskFromContext(ctx),
+			},
+			Name: linux.CString(name),
+		}
+		child, err := i.newEntry(ctx, name, linux.S_IFREG, linux.FUSE_CREATE, &in)
+		if err == nil || !linuxerr.Equals(linuxerr.ENOSYS, err) {
+			return child, err
+		}
+		i.fs.conn.noCreate = true
+	}
+	// The MKNOD reply carries no file handle, so Open() will issue a FUSE_OPEN.
+	in := linux.FUSEMknodIn{
+		MknodMeta: linux.FUSEMknodMeta{
 			Mode:  uint32(opts.Mode) | linux.S_IFREG,
+			Rdev:  0,
 			Umask: umaskFromContext(ctx),
 		},
 		Name: linux.CString(name),
 	}
-	return i.newEntry(ctx, name, linux.S_IFREG, linux.FUSE_CREATE, &in)
+	return i.newEntry(ctx, name, linux.S_IFREG, linux.FUSE_MKNOD, &in)
 }
 
 // NewNode implements kernfs.Inode.NewNode.

pkg/sentry/fsimpl/fuse/read_write.go

@@ -134,11 +134,16 @@ func (fs *filesystem) ReadCallback(ctx context.Context, i *inode, off uint64, si
 // Write sends FUSE_WRITE requests and return the bytes written according to the
 // response.
 func (fs *filesystem) Write(ctx context.Context, fd *regularFileFD, offset int64, src usermem.IOSequence) (int64, int64, error) {
-	// One request cannot exceed either maxWrite or maxPages.
+	// One request cannot exceed maxWrite, maxPages, or, unless big writes were
+	// negotiated, one page. Note that the bigWrites flag is obsolete, latest
+	// libfuse always sets it on.
 	maxWrite := uint32(fs.conn.maxPages) << hostarch.PageShift
 	if maxWrite > fs.conn.maxWrite {
 		maxWrite = fs.conn.maxWrite
 	}
+	if !fs.conn.bigWrites && maxWrite > hostarch.PageSize {
+		maxWrite = hostarch.PageSize
+	}
 
 	// Reuse the same struct for unmarshalling to avoid unnecessary memory allocation.
 	in := linux.FUSEWritePayloadIn{
@@ -158,28 +163,28 @@ func (fs *filesystem) Write(ctx context.Context, fd *regularFileFD, offset int64
 	// Unless a small value for max_write is explicitly used, this loop
 	// is expected to execute only once for the majority of the writes.
 	n := int64(0)
-	end := offset + src.NumBytes()
-	for n < end {
-		writeSize := uint32(end - n)
-
-		// Limit the write size to one page.
-		// Note that the bigWrites flag is obsolete,
-		// latest libfuse always sets it on.
-		if !fs.conn.bigWrites && writeSize > hostarch.PageSize {
-			writeSize = hostarch.PageSize
-		}
-		// Limit the write size to maxWrite.
-		if writeSize > maxWrite {
-			writeSize = maxWrite
-		}
-
-		// TODO(gvisor.dev/issue/3237): Add cache support:
-		// buffer cache. Ideally we write from src to our buffer cache first.
-		// The slice passed to fs.Write() should be a slice from buffer cache.
-		data := make([]byte, writeSize)
-		cp, err := src.CopyIn(ctx, data)
-		if err != nil {
-			return n, offset, err
+	toWrite := src.NumBytes()
+
+	// TODO(gvisor.dev/issue/3237): Add cache support:
+	// buffer cache. Ideally we write from src to our buffer cache first.
+	// The slice passed to fs.Write() should be a slice from buffer cache.
+	//
+	// The buffer is reused for every request: the payload is copied into the
+	// request when it is marshalled, and call() is synchronous.
+	buf := make([]byte, min(int64(maxWrite), toWrite))
+
+	for n < toWrite {
+		writeSize := uint32(min(toWrite-n, int64(maxWrite)))
+		data := buf[:writeSize]
+		// CopyIn returns the bytes it copied along with any fault. Write
+		// those bytes: a write that moved some bytes reports the count, not
+		// the fault.
+		cp, cpErr := src.CopyIn(ctx, data)
+		if cp == 0 {
+			if n == 0 {
+				return n, offset, cpErr
+			}
+			break
 		}
 		data = data[:cp]
 
@@ -193,16 +198,17 @@ func (fs *filesystem) Write(ctx context.Context, fd *regularFileFD, offset int64
 			return n, offset, err
 		}
 		// Write more than requested? EIO.
-		if out.Size > writeSize {
+		if out.Size > uint32(cp) {
 			return n, offset, linuxerr.EIO
 		}
 
 		n += int64(out.Size)
 		offset += int64(out.Size)
 		src = src.DropFirst64(int64(out.Size))
 
-		// Break if short write. Not necessarily an error.
-		if out.Size != writeSize {
+		// Break if the source faulted, or on a short write. Neither is
+		// necessarily an error.
+		if cpErr != nil || out.Size != writeSize {
 			break
 		}
 	}