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

1 files changed, 54 insertions, 30 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());