libbinder: RPC read connection info on 2nd thread

Don't allow the main server thread to be blocked when reading a new ID.
Gotta be available for reading new clients.

Future consideration will be timeouts for reads/writes or trying to
detect DOS here in general, but this CL is what is needed for the
fuzzer.

Bug: 185167543
Bug: 182938024
Test: binderRpcTest
Change-Id: I8ba4a6ce6d1d07f3c36ac554ccb6cc403a15db2b

4 files changed, 73 insertions, 49 deletions

diff --git a/libs/binder/RpcServer.cpp b/libs/binder/RpcServer.cpp
index c0cdcd6bee..3a98cadb94 100644
--- a/libs/binder/RpcServer.cpp
+++ b/libs/binder/RpcServer.cpp
@@ -132,38 +132,12 @@ void RpcServer::join() {
         }
         LOG_RPC_DETAIL("accept4 on fd %d yields fd %d", mServer.get(), clientFd.get());
 
-        // TODO(b/183988761): cannot trust this simple ID, should not block this
-        // thread
-        LOG_ALWAYS_FATAL_IF(!mAgreedExperimental, "no!");
-        int32_t id;
-        if (sizeof(id) != read(clientFd.get(), &id, sizeof(id))) {
-            ALOGE("Could not read ID from fd %d", clientFd.get());
-            continue;
-        }
-
         {
             std::lock_guard<std::mutex> _l(mLock);
-
-            sp<RpcSession> session;
-            if (id == RPC_SESSION_ID_NEW) {
-                // new client!
-                LOG_ALWAYS_FATAL_IF(mSessionIdCounter >= INT32_MAX, "Out of session IDs");
-                mSessionIdCounter++;
-
-                session = RpcSession::make();
-                session->setForServer(wp<RpcServer>::fromExisting(this), mSessionIdCounter);
-
-                mSessions[mSessionIdCounter] = session;
-            } else {
-                auto it = mSessions.find(id);
-                if (it == mSessions.end()) {
-                    ALOGE("Cannot add thread, no record of session with ID %d", id);
-                    continue;
-                }
-                session = it->second;
-            }
-
-            session->startThread(std::move(clientFd));
+            std::thread thread =
+                    std::thread(&RpcServer::establishConnection, this,
+                                std::move(sp<RpcServer>::fromExisting(this)), std::move(clientFd));
+            mConnectingThreads[thread.get_id()] = std::move(thread);
         }
     }
 }
@@ -178,6 +152,56 @@ std::vector<sp<RpcSession>> RpcServer::listSessions() {
     return sessions;
 }
 
+void RpcServer::establishConnection(sp<RpcServer>&& server, base::unique_fd clientFd) {
+    LOG_ALWAYS_FATAL_IF(this != server.get(), "Must pass same ownership object");
+
+    // TODO(b/183988761): cannot trust this simple ID
+    LOG_ALWAYS_FATAL_IF(!mAgreedExperimental, "no!");
+    int32_t id;
+    if (sizeof(id) != read(clientFd.get(), &id, sizeof(id))) {
+        ALOGE("Could not read ID from fd %d", clientFd.get());
+        return;
+    }
+
+    std::thread thisThread;
+    sp<RpcSession> session;
+    {
+        std::lock_guard<std::mutex> _l(mLock);
+
+        auto threadId = mConnectingThreads.find(std::this_thread::get_id());
+        LOG_ALWAYS_FATAL_IF(threadId == mConnectingThreads.end(),
+                            "Must establish connection on owned thread");
+        thisThread = std::move(threadId->second);
+        mConnectingThreads.erase(threadId);
+
+        if (id == RPC_SESSION_ID_NEW) {
+            LOG_ALWAYS_FATAL_IF(mSessionIdCounter >= INT32_MAX, "Out of session IDs");
+            mSessionIdCounter++;
+
+            session = RpcSession::make();
+            session->setForServer(wp<RpcServer>::fromExisting(this), mSessionIdCounter);
+
+            mSessions[mSessionIdCounter] = session;
+        } else {
+            auto it = mSessions.find(id);
+            if (it == mSessions.end()) {
+                ALOGE("Cannot add thread, no record of session with ID %d", id);
+                return;
+            }
+            session = it->second;
+        }
+    }
+
+    // avoid strong cycle
+    server = nullptr;
+    //
+    //
+    // DO NOT ACCESS MEMBER VARIABLES BELOW
+    //
+
+    session->join(std::move(thisThread), std::move(clientFd));
+}
+
 bool RpcServer::setupSocketServer(const RpcSocketAddress& addr) {
     LOG_RPC_DETAIL("Setting up socket server %s", addr.toString().c_str());
 
diff --git a/libs/binder/RpcSession.cpp b/libs/binder/RpcSession.cpp
index f38135b2dc..f32aa7a72d 100644
--- a/libs/binder/RpcSession.cpp
+++ b/libs/binder/RpcSession.cpp
@@ -131,24 +131,14 @@ status_t RpcSession::readId() {
     return OK;
 }
 
-void RpcSession::startThread(unique_fd client) {
-    std::lock_guard<std::mutex> _l(mMutex);
-    sp<RpcSession> holdThis = sp<RpcSession>::fromExisting(this);
-    int fd = client.release();
-    auto thread = std::thread([=] {
-        holdThis->join(unique_fd(fd));
-        {
-            std::lock_guard<std::mutex> _l(holdThis->mMutex);
-            auto it = mThreads.find(std::this_thread::get_id());
-            LOG_ALWAYS_FATAL_IF(it == mThreads.end());
-            it->second.detach();
-            mThreads.erase(it);
-        }
-    });
-    mThreads[thread.get_id()] = std::move(thread);
-}
+void RpcSession::join(std::thread thread, unique_fd client) {
+    LOG_ALWAYS_FATAL_IF(thread.get_id() != std::this_thread::get_id(), "Must own this thread");
+
+    {
+        std::lock_guard<std::mutex> _l(mMutex);
+        mThreads[thread.get_id()] = std::move(thread);
+    }
 
-void RpcSession::join(unique_fd client) {
     // must be registered to allow arbitrary client code executing commands to
     // be able to do nested calls (we can't only read from it)
     sp<RpcConnection> connection = assignServerToThisThread(std::move(client));
@@ -165,6 +155,14 @@ void RpcSession::join(unique_fd client) {
 
     LOG_ALWAYS_FATAL_IF(!removeServerConnection(connection),
                         "bad state: connection object guaranteed to be in list");
+
+    {
+        std::lock_guard<std::mutex> _l(mMutex);
+        auto it = mThreads.find(std::this_thread::get_id());
+        LOG_ALWAYS_FATAL_IF(it == mThreads.end());
+        it->second.detach();
+        mThreads.erase(it);
+    }
 }
 
 void RpcSession::terminateLocked() {
diff --git a/libs/binder/include/binder/RpcServer.h b/libs/binder/include/binder/RpcServer.h
index 3534d5196e..b1e5519cff 100644
--- a/libs/binder/include/binder/RpcServer.h
+++ b/libs/binder/include/binder/RpcServer.h
@@ -22,6 +22,7 @@
 #include <utils/RefBase.h>
 
 #include <mutex>
+#include <thread>
 
 // WARNING: This is a feature which is still in development, and it is subject
 // to radical change. Any production use of this may subject your code to any
@@ -115,6 +116,7 @@ private:
     friend sp<RpcServer>;
     RpcServer();
 
+    void establishConnection(sp<RpcServer>&& session, base::unique_fd clientFd);
     bool setupSocketServer(const RpcSocketAddress& address);
 
     bool mAgreedExperimental = false;
@@ -123,6 +125,7 @@ private:
     base::unique_fd mServer; // socket we are accepting sessions on
 
     std::mutex mLock; // for below
+    std::map<std::thread::id, std::thread> mConnectingThreads;
     sp<IBinder> mRootObject;
     std::map<int32_t, sp<RpcSession>> mSessions;
     int32_t mSessionIdCounter = 0;
diff --git a/libs/binder/include/binder/RpcSession.h b/libs/binder/include/binder/RpcSession.h
index c8ab9e4ba1..92ee100dc5 100644
--- a/libs/binder/include/binder/RpcSession.h
+++ b/libs/binder/include/binder/RpcSession.h
@@ -114,8 +114,7 @@ private:
 
     status_t readId();
 
-    void startThread(base::unique_fd client);
-    void join(base::unique_fd client);
+    void join(std::thread thread, base::unique_fd client);
     void terminateLocked();
 
     struct RpcConnection : public RefBase {