libbinder: RpcConnection: add ID

In preparation to have the server be able to distinguish clients and
clients to be able to dynamically create threads that are assigned to
them.

Future considerations:
- make ID impossible to guess (right now, one client might be able to
  get ahold of a thread from a server). We may implement something here
  or go for something existing like TLS.
- combine getting max threads and this? will wait until dynamic threads
  are actually implemented and we know we need this ID and we're looking
  at performance. For now this is a placeholder to enable dynamic client
  APIs.

Bug: 185167543
Test: binderRpcTest
Change-Id: If8563c69930c23b9ca91090b4f59ef1f51073f24

6 files changed, 90 insertions, 12 deletions

diff --git a/libs/binder/RpcConnection.cpp b/libs/binder/RpcConnection.cpp
index 930bcbde62..95eba87b68 100644
--- a/libs/binder/RpcConnection.cpp
+++ b/libs/binder/RpcConnection.cpp
@@ -115,6 +115,24 @@ status_t RpcConnection::sendDecStrong(const RpcAddress& address) {
     return state()->sendDecStrong(socket.fd(), address);
 }
 
+status_t RpcConnection::readId() {
+    {
+        std::lock_guard<std::mutex> _l(mSocketMutex);
+        LOG_ALWAYS_FATAL_IF(mForServer != nullptr, "Can only update ID for client.");
+    }
+
+    int32_t id;
+
+    ExclusiveSocket socket(sp<RpcConnection>::fromExisting(this), SocketUse::CLIENT);
+    status_t status =
+            state()->getConnectionId(socket.fd(), sp<RpcConnection>::fromExisting(this), &id);
+    if (status != OK) return status;
+
+    LOG_RPC_DETAIL("RpcConnection %p has id %d", this, id);
+    mId = id;
+    return OK;
+}
+
 void RpcConnection::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)
@@ -134,10 +152,6 @@ void RpcConnection::join(unique_fd client) {
                         "bad state: socket object guaranteed to be in list");
 }
 
-void RpcConnection::setForServer(const wp<RpcServer>& server) {
-    mForServer = server;
-}
-
 wp<RpcServer> RpcConnection::server() {
     return mForServer;
 }
@@ -162,6 +176,12 @@ bool RpcConnection::setupSocketClient(const RpcSocketAddress& addr) {
         return false;
     }
 
+    if (status_t status = readId(); status != OK) {
+        ALOGE("Could not get connection id after initial connection to %s; %s",
+              addr.toString().c_str(), statusToString(status).c_str());
+        return false;
+    }
+
     // we've already setup one client
     for (size_t i = 0; i + 1 < numThreadsAvailable; i++) {
         // TODO(b/185167543): avoid race w/ accept4 not being called on server
@@ -202,6 +222,11 @@ void RpcConnection::addClient(unique_fd fd) {
     mClients.push_back(connection);
 }
 
+void RpcConnection::setForServer(const wp<RpcServer>& server, int32_t connectionId) {
+    mId = connectionId;
+    mForServer = server;
+}
+
 sp<RpcConnection::ConnectionSocket> RpcConnection::assignServerToThisThread(unique_fd fd) {
     std::lock_guard<std::mutex> _l(mSocketMutex);
     sp<ConnectionSocket> connection = sp<ConnectionSocket>::make();
diff --git a/libs/binder/RpcServer.cpp b/libs/binder/RpcServer.cpp
index 0753b54763..5f024caf32 100644
--- a/libs/binder/RpcServer.cpp
+++ b/libs/binder/RpcServer.cpp
@@ -128,7 +128,7 @@ void RpcServer::join() {
         LOG_ALWAYS_FATAL_IF(mServer.get() == -1, "RpcServer must be setup to join.");
         // TODO(b/185167543): support more than one client at once
         mConnection = RpcConnection::make();
-        mConnection->setForServer(sp<RpcServer>::fromExisting(this));
+        mConnection->setForServer(sp<RpcServer>::fromExisting(this), 42 /*placeholder id*/);
 
         mStarted = true;
             for (size_t i = 0; i < mMaxThreads; i++) {
diff --git a/libs/binder/RpcState.cpp b/libs/binder/RpcState.cpp
index 6bfcc42469..19dea7e607 100644
--- a/libs/binder/RpcState.cpp
+++ b/libs/binder/RpcState.cpp
@@ -249,7 +249,7 @@ sp<IBinder> RpcState::getRootObject(const base::unique_fd& fd,
 }
 
 status_t RpcState::getMaxThreads(const base::unique_fd& fd, const sp<RpcConnection>& connection,
-                                 size_t* maxThreads) {
+                                 size_t* maxThreadsOut) {
     Parcel data;
     data.markForRpc(connection);
     Parcel reply;
@@ -261,15 +261,36 @@ status_t RpcState::getMaxThreads(const base::unique_fd& fd, const sp<RpcConnecti
         return status;
     }
 
-    int32_t threads;
-    status = reply.readInt32(&threads);
+    int32_t maxThreads;
+    status = reply.readInt32(&maxThreads);
     if (status != OK) return status;
-    if (threads <= 0) {
-        ALOGE("Error invalid max threads: %d", threads);
+    if (maxThreads <= 0) {
+        ALOGE("Error invalid max maxThreads: %d", maxThreads);
         return BAD_VALUE;
     }
 
-    *maxThreads = threads;
+    *maxThreadsOut = maxThreads;
+    return OK;
+}
+
+status_t RpcState::getConnectionId(const base::unique_fd& fd, const sp<RpcConnection>& connection,
+                                   int32_t* connectionIdOut) {
+    Parcel data;
+    data.markForRpc(connection);
+    Parcel reply;
+
+    status_t status = transact(fd, RpcAddress::zero(), RPC_SPECIAL_TRANSACT_GET_CONNECTION_ID, data,
+                               connection, &reply, 0);
+    if (status != OK) {
+        ALOGE("Error getting connection ID: %s", statusToString(status).c_str());
+        return status;
+    }
+
+    int32_t connectionId;
+    status = reply.readInt32(&connectionId);
+    if (status != OK) return status;
+
+    *connectionIdOut = connectionId;
     return OK;
 }
 
@@ -554,6 +575,16 @@ status_t RpcState::processTransactInternal(const base::unique_fd& fd,
                         replyStatus = reply.writeInt32(server->getMaxThreads());
                         break;
                     }
+                    case RPC_SPECIAL_TRANSACT_GET_CONNECTION_ID: {
+                        // only connections w/ services can be the source of a
+                        // connection ID (so still guarded by non-null server)
+                        //
+                        // connections associated with servers must have an ID
+                        // (hence abort)
+                        int32_t id = connection->getPrivateAccessorForId().get().value();
+                        replyStatus = reply.writeInt32(id);
+                        break;
+                    }
                     default: {
                         replyStatus = UNKNOWN_TRANSACTION;
                     }
diff --git a/libs/binder/RpcState.h b/libs/binder/RpcState.h
index 1cfa406791..825fd7c4e5 100644
--- a/libs/binder/RpcState.h
+++ b/libs/binder/RpcState.h
@@ -50,9 +50,12 @@ public:
     RpcState();
     ~RpcState();
 
+    // TODO(b/182940634): combine some special transactions into one "getServerInfo" call?
     sp<IBinder> getRootObject(const base::unique_fd& fd, const sp<RpcConnection>& connection);
     status_t getMaxThreads(const base::unique_fd& fd, const sp<RpcConnection>& connection,
                            size_t* maxThreadsOut);
+    status_t getConnectionId(const base::unique_fd& fd, const sp<RpcConnection>& connection,
+                             int32_t* connectionIdOut);
 
     [[nodiscard]] status_t transact(const base::unique_fd& fd, const RpcAddress& address,
                                     uint32_t code, const Parcel& data,
diff --git a/libs/binder/RpcWireFormat.h b/libs/binder/RpcWireFormat.h
index cc7cacb32d..56af0d3ab5 100644
--- a/libs/binder/RpcWireFormat.h
+++ b/libs/binder/RpcWireFormat.h
@@ -48,6 +48,7 @@ enum : uint32_t {
 enum : uint32_t {
     RPC_SPECIAL_TRANSACT_GET_ROOT = 0,
     RPC_SPECIAL_TRANSACT_GET_MAX_THREADS = 1,
+    RPC_SPECIAL_TRANSACT_GET_CONNECTION_ID = 2,
 };
 
 // serialization is like:
diff --git a/libs/binder/include/binder/RpcConnection.h b/libs/binder/include/binder/RpcConnection.h
index 82bc7df96d..7e31e8a4fc 100644
--- a/libs/binder/include/binder/RpcConnection.h
+++ b/libs/binder/include/binder/RpcConnection.h
@@ -88,17 +88,31 @@ public:
 
     ~RpcConnection();
 
-    void setForServer(const wp<RpcServer>& server);
     wp<RpcServer> server();
 
     // internal only
     const std::unique_ptr<RpcState>& state() { return mState; }
 
+    class PrivateAccessorForId {
+    private:
+        friend class RpcConnection;
+        friend class RpcState;
+        explicit PrivateAccessorForId(const RpcConnection* connection) : mConnection(connection) {}
+
+        const std::optional<int32_t> get() { return mConnection->mId; }
+
+        const RpcConnection* mConnection;
+    };
+    PrivateAccessorForId getPrivateAccessorForId() const { return PrivateAccessorForId(this); }
+
 private:
+    friend PrivateAccessorForId;
     friend sp<RpcConnection>;
     friend RpcServer;
     RpcConnection();
 
+    status_t readId();
+
     void join(base::unique_fd client);
 
     struct ConnectionSocket : public RefBase {
@@ -112,6 +126,7 @@ private:
     bool setupSocketClient(const RpcSocketAddress& address);
     bool setupOneSocketClient(const RpcSocketAddress& address);
     void addClient(base::unique_fd fd);
+    void setForServer(const wp<RpcServer>& server, int32_t connectionId);
     sp<ConnectionSocket> assignServerToThisThread(base::unique_fd fd);
     bool removeServerSocket(const sp<ConnectionSocket>& socket);
 
@@ -158,6 +173,9 @@ private:
 
     wp<RpcServer> mForServer; // maybe null, for client connections
 
+    // TODO(b/183988761): this shouldn't be guessable
+    std::optional<int32_t> mId;
+
     std::unique_ptr<RpcState> mState;
 
     std::mutex mSocketMutex;           // for all below