Support SOCK_STREAM for bridge between system and app

Previously AppFuse use SOCK_SEQPACKET for sockets communicating system
and app. However SOCK_SEQPACKET requires the buffer of message size in
the kernel and sometimes failed to write with ENOBUF.

The CL updates libappfuse so that it can use SOCK_STREAM instead of
SOCK_SEQPACKET.

Bug: 33279206
Test: libappfuse_test
Change-Id: I622ada9ac1d71d0c57b6cfff0904c7829cea7995

3 files changed, 64 insertions, 31 deletions

diff --git a/libappfuse/FuseBuffer.cc b/libappfuse/FuseBuffer.cc
index 882d54552..8fb2dbcc5 100644
--- a/libappfuse/FuseBuffer.cc
+++ b/libappfuse/FuseBuffer.cc
@@ -23,6 +23,7 @@
 #include <algorithm>
 #include <type_traits>
 
+#include <android-base/file.h>
 #include <android-base/logging.h>
 #include <android-base/macros.h>
 
@@ -34,57 +35,65 @@ static_assert(
     "FuseBuffer must be standard layout union.");
 
 template <typename T>
-bool FuseMessage<T>::CheckPacketSize(size_t size, const char* name) const {
+bool FuseMessage<T>::CheckHeaderLength(const char* name) const {
   const auto& header = static_cast<const T*>(this)->header;
-  if (size >= sizeof(header) && size <= sizeof(T)) {
+  if (header.len >= sizeof(header) && header.len <= sizeof(T)) {
     return true;
   } else {
-    LOG(ERROR) << name << " is invalid=" << size;
+    LOG(ERROR) << "Invalid header length is found in " << name << ": " <<
+        header.len;
     return false;
   }
 }
 
 template <typename T>
-bool FuseMessage<T>::CheckResult(int result, const char* operation_name) const {
-  if (result == 0) {
-    // Expected close of other endpoints.
-    return false;
-  }
+bool FuseMessage<T>::Read(int fd) {
+  char* const buf = reinterpret_cast<char*>(this);
+  const ssize_t result = TEMP_FAILURE_RETRY(::read(fd, buf, sizeof(T)));
   if (result < 0) {
-    PLOG(ERROR) << "Failed to " << operation_name << " a packet";
+    PLOG(ERROR) << "Failed to read a FUSE message";
     return false;
   }
-  return true;
-}
 
-template <typename T>
-bool FuseMessage<T>::CheckHeaderLength(int result, const char* operation_name) const {
   const auto& header = static_cast<const T*>(this)->header;
-  if (static_cast<uint32_t>(result) == header.len) {
-    return true;
-  } else {
-    LOG(ERROR) << "Invalid header length: operation_name=" << operation_name
-               << " result=" << result
-               << " header.len=" << header.len;
+  if (result < static_cast<ssize_t>(sizeof(header))) {
+    LOG(ERROR) << "Read bytes " << result << " are shorter than header size " <<
+        sizeof(header);
     return false;
   }
-}
 
-template <typename T>
-bool FuseMessage<T>::Read(int fd) {
-  const ssize_t result = TEMP_FAILURE_RETRY(::read(fd, this, sizeof(T)));
-  return CheckResult(result, "read") && CheckPacketSize(result, "read count") &&
-      CheckHeaderLength(result, "read");
+  if (!CheckHeaderLength("Read")) {
+    return false;
+  }
+
+  if (static_cast<uint32_t>(result) > header.len) {
+    LOG(ERROR) << "Read bytes " << result << " are longer than header.len " <<
+        header.len;
+    return false;
+  }
+
+  if (!base::ReadFully(fd, buf + result, header.len - result)) {
+    PLOG(ERROR) << "ReadFully failed";
+    return false;
+  }
+
+  return true;
 }
 
 template <typename T>
 bool FuseMessage<T>::Write(int fd) const {
+  if (!CheckHeaderLength("Write")) {
+    return false;
+  }
+
+  const char* const buf = reinterpret_cast<const char*>(this);
   const auto& header = static_cast<const T*>(this)->header;
-  if (!CheckPacketSize(header.len, "header.len")) {
+  if (!base::WriteFully(fd, buf, header.len)) {
+    PLOG(ERROR) << "WriteFully failed";
     return false;
   }
-  const ssize_t result = TEMP_FAILURE_RETRY(::write(fd, this, header.len));
-  return CheckResult(result, "write") && CheckHeaderLength(result, "write");
+
+  return true;
 }
 
 template class FuseMessage<FuseRequest>;
diff --git a/libappfuse/include/libappfuse/FuseBuffer.h b/libappfuse/include/libappfuse/FuseBuffer.h
index 276db9020..7abd2fa40 100644
--- a/libappfuse/include/libappfuse/FuseBuffer.h
+++ b/libappfuse/include/libappfuse/FuseBuffer.h
@@ -34,9 +34,7 @@ class FuseMessage {
   bool Read(int fd);
   bool Write(int fd) const;
  private:
-  bool CheckPacketSize(size_t size, const char* name) const;
-  bool CheckResult(int result, const char* operation_name) const;
-  bool CheckHeaderLength(int result, const char* operation_name) const;
+  bool CheckHeaderLength(const char* name) const;
 };
 
 // FuseRequest represents file operation requests from /dev/fuse. It starts
diff --git a/libappfuse/tests/FuseBufferTest.cc b/libappfuse/tests/FuseBufferTest.cc
index c82213513..db35d330d 100644
--- a/libappfuse/tests/FuseBufferTest.cc
+++ b/libappfuse/tests/FuseBufferTest.cc
@@ -20,6 +20,8 @@
 #include <string.h>
 #include <sys/socket.h>
 
+#include <thread>
+
 #include <android-base/unique_fd.h>
 #include <gtest/gtest.h>
 
@@ -110,6 +112,30 @@ TEST(FuseMessageTest, Write_TooShort) {
   TestWriteInvalidLength(sizeof(fuse_in_header) - 1);
 }
 
+TEST(FuseMessageTest, ShortWriteAndRead) {
+  int raw_fds[2];
+  ASSERT_EQ(0, socketpair(AF_UNIX, SOCK_STREAM, 0, raw_fds));
+
+  android::base::unique_fd fds[2];
+  fds[0].reset(raw_fds[0]);
+  fds[1].reset(raw_fds[1]);
+
+  const int send_buffer_size = 1024;
+  ASSERT_EQ(0, setsockopt(fds[0], SOL_SOCKET, SO_SNDBUF, &send_buffer_size,
+                          sizeof(int)));
+
+  bool succeed = false;
+  const int sender_fd = fds[0].get();
+  std::thread thread([sender_fd, &succeed] {
+    FuseRequest request;
+    request.header.len = 1024 * 4;
+    succeed = request.Write(sender_fd);
+  });
+  thread.detach();
+  FuseRequest request;
+  ASSERT_TRUE(request.Read(fds[1]));
+}
+
 TEST(FuseResponseTest, Reset) {
   FuseResponse response;
   // Write 1 to the first ten bytes.