blob: d9e926a9d5c4a641903d2f2428f217b88de85487 [file] [log] [blame]
/*
* Copyright (C) 2020 The Android Open Source Project
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#define LOG_TAG "RpcServer"
#include <inttypes.h>
#include <netinet/tcp.h>
#include <poll.h>
#include <sys/socket.h>
#include <sys/un.h>
#include <thread>
#include <vector>
#include <binder/Functional.h>
#include <binder/Parcel.h>
#include <binder/RpcServer.h>
#include <binder/RpcTransportRaw.h>
#include <log/log.h>
#include "BuildFlags.h"
#include "FdTrigger.h"
#include "OS.h"
#include "RpcSocketAddress.h"
#include "RpcState.h"
#include "RpcTransportUtils.h"
#include "RpcWireFormat.h"
#include "Utils.h"
namespace android {
constexpr size_t kSessionIdBytes = 32;
using namespace android::binder::impl;
using android::binder::borrowed_fd;
using android::binder::unique_fd;
RpcServer::RpcServer(std::unique_ptr<RpcTransportCtx> ctx) : mCtx(std::move(ctx)) {}
RpcServer::~RpcServer() {
RpcMutexUniqueLock _l(mLock);
LOG_ALWAYS_FATAL_IF(mShutdownTrigger != nullptr, "Must call shutdown() before destructor");
}
sp<RpcServer> RpcServer::make(std::unique_ptr<RpcTransportCtxFactory> rpcTransportCtxFactory) {
// Default is without TLS.
if (rpcTransportCtxFactory == nullptr)
rpcTransportCtxFactory = binder::os::makeDefaultRpcTransportCtxFactory();
auto ctx = rpcTransportCtxFactory->newServerCtx();
if (ctx == nullptr) return nullptr;
return sp<RpcServer>::make(std::move(ctx));
}
status_t RpcServer::setupUnixDomainSocketBootstrapServer(unique_fd bootstrapFd) {
return setupExternalServer(std::move(bootstrapFd), &RpcServer::recvmsgSocketConnection);
}
status_t RpcServer::setupUnixDomainServer(const char* path) {
return setupSocketServer(UnixSocketAddress(path));
}
status_t RpcServer::setupVsockServer(unsigned int bindCid, unsigned int port) {
return setupSocketServer(VsockSocketAddress(bindCid, port));
}
status_t RpcServer::setupInetServer(const char* address, unsigned int port,
unsigned int* assignedPort) {
if (assignedPort != nullptr) *assignedPort = 0;
auto aiStart = InetSocketAddress::getAddrInfo(address, port);
if (aiStart == nullptr) return UNKNOWN_ERROR;
for (auto ai = aiStart.get(); ai != nullptr; ai = ai->ai_next) {
if (ai->ai_addr == nullptr) continue;
InetSocketAddress socketAddress(ai->ai_addr, ai->ai_addrlen, address, port);
if (status_t status = setupSocketServer(socketAddress); status != OK) {
continue;
}
LOG_ALWAYS_FATAL_IF(socketAddress.addr()->sa_family != AF_INET, "expecting inet");
sockaddr_in addr{};
socklen_t len = sizeof(addr);
if (0 != getsockname(mServer.fd.get(), reinterpret_cast<sockaddr*>(&addr), &len)) {
int savedErrno = errno;
ALOGE("Could not getsockname at %s: %s", socketAddress.toString().c_str(),
strerror(savedErrno));
return -savedErrno;
}
LOG_ALWAYS_FATAL_IF(len != sizeof(addr), "Wrong socket type: len %zu vs len %zu",
static_cast<size_t>(len), sizeof(addr));
unsigned int realPort = ntohs(addr.sin_port);
LOG_ALWAYS_FATAL_IF(port != 0 && realPort != port,
"Requesting inet server on %s but it is set up on %u.",
socketAddress.toString().c_str(), realPort);
if (assignedPort != nullptr) {
*assignedPort = realPort;
}
return OK;
}
ALOGE("None of the socket address resolved for %s:%u can be set up as inet server.", address,
port);
return UNKNOWN_ERROR;
}
void RpcServer::setMaxThreads(size_t threads) {
LOG_ALWAYS_FATAL_IF(threads <= 0, "RpcServer is useless without threads");
LOG_ALWAYS_FATAL_IF(mJoinThreadRunning, "Cannot set max threads while running");
mMaxThreads = threads;
}
size_t RpcServer::getMaxThreads() {
return mMaxThreads;
}
bool RpcServer::setProtocolVersion(uint32_t version) {
if (!RpcState::validateProtocolVersion(version)) {
return false;
}
mProtocolVersion = version;
return true;
}
void RpcServer::setSupportedFileDescriptorTransportModes(
const std::vector<RpcSession::FileDescriptorTransportMode>& modes) {
mSupportedFileDescriptorTransportModes.reset();
for (RpcSession::FileDescriptorTransportMode mode : modes) {
mSupportedFileDescriptorTransportModes.set(static_cast<size_t>(mode));
}
}
void RpcServer::setRootObject(const sp<IBinder>& binder) {
RpcMutexLockGuard _l(mLock);
mRootObjectFactory = nullptr;
mRootObjectWeak = mRootObject = binder;
}
void RpcServer::setRootObjectWeak(const wp<IBinder>& binder) {
RpcMutexLockGuard _l(mLock);
mRootObject.clear();
mRootObjectFactory = nullptr;
mRootObjectWeak = binder;
}
void RpcServer::setPerSessionRootObject(
std::function<sp<IBinder>(wp<RpcSession> session, const void*, size_t)>&& makeObject) {
RpcMutexLockGuard _l(mLock);
mRootObject.clear();
mRootObjectWeak.clear();
mRootObjectFactory = std::move(makeObject);
}
void RpcServer::setConnectionFilter(std::function<bool(const void*, size_t)>&& filter) {
RpcMutexLockGuard _l(mLock);
LOG_ALWAYS_FATAL_IF(mShutdownTrigger != nullptr, "Already joined");
mConnectionFilter = std::move(filter);
}
void RpcServer::setServerSocketModifier(std::function<void(borrowed_fd)>&& modifier) {
RpcMutexLockGuard _l(mLock);
LOG_ALWAYS_FATAL_IF(mServer.fd.ok(), "Already started");
mServerSocketModifier = std::move(modifier);
}
sp<IBinder> RpcServer::getRootObject() {
RpcMutexLockGuard _l(mLock);
bool hasWeak = mRootObjectWeak.unsafe_get();
sp<IBinder> ret = mRootObjectWeak.promote();
ALOGW_IF(hasWeak && ret == nullptr, "RpcServer root object is freed, returning nullptr");
return ret;
}
std::vector<uint8_t> RpcServer::getCertificate(RpcCertificateFormat format) {
RpcMutexLockGuard _l(mLock);
return mCtx->getCertificate(format);
}
static void joinRpcServer(sp<RpcServer>&& thiz) {
thiz->join();
}
void RpcServer::start() {
RpcMutexLockGuard _l(mLock);
LOG_ALWAYS_FATAL_IF(mJoinThread.get(), "Already started!");
mJoinThread =
std::make_unique<RpcMaybeThread>(&joinRpcServer, sp<RpcServer>::fromExisting(this));
rpcJoinIfSingleThreaded(*mJoinThread);
}
status_t RpcServer::acceptSocketConnection(const RpcServer& server, RpcTransportFd* out) {
RpcTransportFd clientSocket(unique_fd(TEMP_FAILURE_RETRY(
accept4(server.mServer.fd.get(), nullptr, nullptr, SOCK_CLOEXEC | SOCK_NONBLOCK))));
if (!clientSocket.fd.ok()) {
int savedErrno = errno;
ALOGE("Could not accept4 socket: %s", strerror(savedErrno));
return -savedErrno;
}
*out = std::move(clientSocket);
return OK;
}
status_t RpcServer::recvmsgSocketConnection(const RpcServer& server, RpcTransportFd* out) {
int zero = 0;
iovec iov{&zero, sizeof(zero)};
std::vector<std::variant<unique_fd, borrowed_fd>> fds;
ssize_t num_bytes = binder::os::receiveMessageFromSocket(server.mServer, &iov, 1, &fds);
if (num_bytes < 0) {
int savedErrno = errno;
ALOGE("Failed recvmsg: %s", strerror(savedErrno));
return -savedErrno;
}
if (num_bytes == 0) {
return DEAD_OBJECT;
}
if (fds.size() != 1) {
ALOGE("Expected exactly one fd from recvmsg, got %zu", fds.size());
return -EINVAL;
}
unique_fd fd(std::move(std::get<unique_fd>(fds.back())));
if (status_t res = binder::os::setNonBlocking(fd); res != OK) return res;
*out = RpcTransportFd(std::move(fd));
return OK;
}
void RpcServer::join() {
{
RpcMutexLockGuard _l(mLock);
LOG_ALWAYS_FATAL_IF(!mServer.fd.ok(), "RpcServer must be setup to join.");
LOG_ALWAYS_FATAL_IF(mAcceptFn == nullptr, "RpcServer must have an accept() function");
LOG_ALWAYS_FATAL_IF(mShutdownTrigger != nullptr, "Already joined");
mJoinThreadRunning = true;
mShutdownTrigger = FdTrigger::make();
LOG_ALWAYS_FATAL_IF(mShutdownTrigger == nullptr, "Cannot create join signaler");
}
status_t status;
while ((status = mShutdownTrigger->triggerablePoll(mServer, POLLIN)) == OK) {
std::array<uint8_t, kRpcAddressSize> addr;
static_assert(addr.size() >= sizeof(sockaddr_storage), "kRpcAddressSize is too small");
socklen_t addrLen = addr.size();
RpcTransportFd clientSocket;
if ((status = mAcceptFn(*this, &clientSocket)) != OK) {
if (status == DEAD_OBJECT)
break;
else
continue;
}
LOG_RPC_DETAIL("accept on fd %d yields fd %d", mServer.fd.get(), clientSocket.fd.get());
if (getpeername(clientSocket.fd.get(), reinterpret_cast<sockaddr*>(addr.data()),
&addrLen)) {
ALOGE("Could not getpeername socket: %s", strerror(errno));
continue;
}
if (mConnectionFilter != nullptr && !mConnectionFilter(addr.data(), addrLen)) {
ALOGE("Dropped client connection fd %d", clientSocket.fd.get());
continue;
}
{
RpcMutexLockGuard _l(mLock);
RpcMaybeThread thread =
RpcMaybeThread(&RpcServer::establishConnection,
sp<RpcServer>::fromExisting(this), std::move(clientSocket), addr,
addrLen, RpcSession::join);
auto& threadRef = mConnectingThreads[thread.get_id()];
threadRef = std::move(thread);
rpcJoinIfSingleThreaded(threadRef);
}
}
LOG_RPC_DETAIL("RpcServer::join exiting with %s", statusToString(status).c_str());
if constexpr (kEnableRpcThreads) {
RpcMutexLockGuard _l(mLock);
mJoinThreadRunning = false;
} else {
// Multi-threaded builds clear this in shutdown(), but we need it valid
// so the loop above exits cleanly
mShutdownTrigger = nullptr;
}
mShutdownCv.notify_all();
}
bool RpcServer::shutdown() {
RpcMutexUniqueLock _l(mLock);
if (mShutdownTrigger == nullptr) {
LOG_RPC_DETAIL("Cannot shutdown. No shutdown trigger installed (already shutdown, or not "
"joined yet?)");
return false;
}
mShutdownTrigger->trigger();
for (auto& [id, session] : mSessions) {
(void)id;
// server lock is a more general lock
RpcMutexLockGuard _lSession(session->mMutex);
session->mShutdownTrigger->trigger();
}
if constexpr (!kEnableRpcThreads) {
// In single-threaded mode we're done here, everything else that
// needs to happen should be at the end of RpcServer::join()
return true;
}
while (mJoinThreadRunning || !mConnectingThreads.empty() || !mSessions.empty()) {
if (std::cv_status::timeout == mShutdownCv.wait_for(_l, std::chrono::seconds(1))) {
ALOGE("Waiting for RpcServer to shut down (1s w/o progress). Join thread running: %d, "
"Connecting threads: "
"%zu, Sessions: %zu. Is your server deadlocked?",
mJoinThreadRunning, mConnectingThreads.size(), mSessions.size());
}
}
// At this point, we know join() is about to exit, but the thread that calls
// join() may not have exited yet.
// If RpcServer owns the join thread (aka start() is called), make sure the thread exits;
// otherwise ~thread() may call std::terminate(), which may crash the process.
// If RpcServer does not own the join thread (aka join() is called directly),
// then the owner of RpcServer is responsible for cleaning up that thread.
if (mJoinThread.get()) {
mJoinThread->join();
mJoinThread.reset();
}
mServer = RpcTransportFd();
LOG_RPC_DETAIL("Finished waiting on shutdown.");
mShutdownTrigger = nullptr;
return true;
}
std::vector<sp<RpcSession>> RpcServer::listSessions() {
RpcMutexLockGuard _l(mLock);
std::vector<sp<RpcSession>> sessions;
for (auto& [id, session] : mSessions) {
(void)id;
sessions.push_back(session);
}
return sessions;
}
size_t RpcServer::numUninitializedSessions() {
RpcMutexLockGuard _l(mLock);
return mConnectingThreads.size();
}
void RpcServer::establishConnection(
sp<RpcServer>&& server, RpcTransportFd clientFd, std::array<uint8_t, kRpcAddressSize> addr,
size_t addrLen,
std::function<void(sp<RpcSession>&&, RpcSession::PreJoinSetupResult&&)>&& joinFn) {
// mShutdownTrigger can only be cleared once connection threads have joined.
// It must be set before this thread is started
LOG_ALWAYS_FATAL_IF(server->mShutdownTrigger == nullptr);
LOG_ALWAYS_FATAL_IF(server->mCtx == nullptr);
status_t status = OK;
int clientFdForLog = clientFd.fd.get();
auto client = server->mCtx->newTransport(std::move(clientFd), server->mShutdownTrigger.get());
if (client == nullptr) {
ALOGE("Dropping accept4()-ed socket because sslAccept fails");
status = DEAD_OBJECT;
// still need to cleanup before we can return
} else {
LOG_RPC_DETAIL("Created RpcTransport %p for client fd %d", client.get(), clientFdForLog);
}
RpcConnectionHeader header;
if (status == OK) {
iovec iov{&header, sizeof(header)};
status = client->interruptableReadFully(server->mShutdownTrigger.get(), &iov, 1,
std::nullopt, /*ancillaryFds=*/nullptr);
if (status != OK) {
ALOGE("Failed to read ID for client connecting to RPC server: %s",
statusToString(status).c_str());
// still need to cleanup before we can return
}
}
std::vector<uint8_t> sessionId;
if (status == OK) {
if (header.sessionIdSize > 0) {
if (header.sessionIdSize == kSessionIdBytes) {
sessionId.resize(header.sessionIdSize);
iovec iov{sessionId.data(), sessionId.size()};
status = client->interruptableReadFully(server->mShutdownTrigger.get(), &iov, 1,
std::nullopt, /*ancillaryFds=*/nullptr);
if (status != OK) {
ALOGE("Failed to read session ID for client connecting to RPC server: %s",
statusToString(status).c_str());
// still need to cleanup before we can return
}
} else {
ALOGE("Malformed session ID. Expecting session ID of size %zu but got %" PRIu16,
kSessionIdBytes, header.sessionIdSize);
status = BAD_VALUE;
}
}
}
bool incoming = false;
uint32_t protocolVersion = 0;
bool requestingNewSession = false;
if (status == OK) {
incoming = header.options & RPC_CONNECTION_OPTION_INCOMING;
protocolVersion = std::min(header.version,
server->mProtocolVersion.value_or(RPC_WIRE_PROTOCOL_VERSION));
requestingNewSession = sessionId.empty();
if (requestingNewSession) {
RpcNewSessionResponse response{
.version = protocolVersion,
};
iovec iov{&response, sizeof(response)};
status = client->interruptableWriteFully(server->mShutdownTrigger.get(), &iov, 1,
std::nullopt, nullptr);
if (status != OK) {
ALOGE("Failed to send new session response: %s", statusToString(status).c_str());
// still need to cleanup before we can return
}
}
}
RpcMaybeThread thisThread;
sp<RpcSession> session;
{
RpcMutexUniqueLock _l(server->mLock);
auto threadId = server->mConnectingThreads.find(rpc_this_thread::get_id());
LOG_ALWAYS_FATAL_IF(threadId == server->mConnectingThreads.end(),
"Must establish connection on owned thread");
thisThread = std::move(threadId->second);
auto detachGuardLambda = [&]() {
thisThread.detach();
_l.unlock();
server->mShutdownCv.notify_all();
};
auto detachGuard = make_scope_guard(std::ref(detachGuardLambda));
server->mConnectingThreads.erase(threadId);
if (status != OK || server->mShutdownTrigger->isTriggered()) {
return;
}
if (requestingNewSession) {
if (incoming) {
ALOGE("Cannot create a new session with an incoming connection, would leak");
return;
}
// Uniquely identify session at the application layer. Even if a
// client/server use the same certificates, if they create multiple
// sessions, we still want to distinguish between them.
sessionId.resize(kSessionIdBytes);
size_t tries = 0;
do {
// don't block if there is some entropy issue
if (tries++ > 5) {
ALOGE("Cannot find new address: %s",
HexString(sessionId.data(), sessionId.size()).c_str());
return;
}
auto status = binder::os::getRandomBytes(sessionId.data(), sessionId.size());
if (status != OK) {
ALOGE("Failed to read random session ID: %s", strerror(-status));
return;
}
} while (server->mSessions.end() != server->mSessions.find(sessionId));
session = sp<RpcSession>::make(nullptr);
session->setMaxIncomingThreads(server->mMaxThreads);
if (!session->setProtocolVersion(protocolVersion)) return;
if (header.fileDescriptorTransportMode <
server->mSupportedFileDescriptorTransportModes.size() &&
server->mSupportedFileDescriptorTransportModes.test(
header.fileDescriptorTransportMode)) {
session->setFileDescriptorTransportMode(
static_cast<RpcSession::FileDescriptorTransportMode>(
header.fileDescriptorTransportMode));
} else {
ALOGE("Rejecting connection: FileDescriptorTransportMode is not supported: %hhu",
header.fileDescriptorTransportMode);
return;
}
// if null, falls back to server root
sp<IBinder> sessionSpecificRoot;
if (server->mRootObjectFactory != nullptr) {
sessionSpecificRoot =
server->mRootObjectFactory(wp<RpcSession>(session), addr.data(), addrLen);
if (sessionSpecificRoot == nullptr) {
ALOGE("Warning: server returned null from root object factory");
}
}
if (!session->setForServer(server,
sp<RpcServer::EventListener>::fromExisting(
static_cast<RpcServer::EventListener*>(
server.get())),
sessionId, sessionSpecificRoot)) {
ALOGE("Failed to attach server to session");
return;
}
server->mSessions[sessionId] = session;
} else {
auto it = server->mSessions.find(sessionId);
if (it == server->mSessions.end()) {
ALOGE("Cannot add thread, no record of session with ID %s",
HexString(sessionId.data(), sessionId.size()).c_str());
return;
}
session = it->second;
}
if (incoming) {
LOG_ALWAYS_FATAL_IF(OK != session->addOutgoingConnection(std::move(client), true),
"server state must already be initialized");
return;
}
detachGuard.release();
session->preJoinThreadOwnership(std::move(thisThread));
}
auto setupResult = session->preJoinSetup(std::move(client));
// avoid strong cycle
server = nullptr;
joinFn(std::move(session), std::move(setupResult));
}
status_t RpcServer::setupSocketServer(const RpcSocketAddress& addr) {
LOG_RPC_DETAIL("Setting up socket server %s", addr.toString().c_str());
LOG_ALWAYS_FATAL_IF(hasServer(), "Each RpcServer can only have one server.");
unique_fd socket_fd(TEMP_FAILURE_RETRY(
socket(addr.addr()->sa_family, SOCK_STREAM | SOCK_CLOEXEC | SOCK_NONBLOCK, 0)));
if (!socket_fd.ok()) {
int savedErrno = errno;
ALOGE("Could not create socket at %s: %s", addr.toString().c_str(), strerror(savedErrno));
return -savedErrno;
}
if (addr.addr()->sa_family == AF_INET || addr.addr()->sa_family == AF_INET6) {
int noDelay = 1;
int result =
setsockopt(socket_fd.get(), IPPROTO_TCP, TCP_NODELAY, &noDelay, sizeof(noDelay));
if (result < 0) {
int savedErrno = errno;
ALOGE("Could not set TCP_NODELAY on %s", strerror(savedErrno));
return -savedErrno;
}
}
{
RpcMutexLockGuard _l(mLock);
if (mServerSocketModifier != nullptr) {
mServerSocketModifier(socket_fd);
}
}
if (0 != TEMP_FAILURE_RETRY(bind(socket_fd.get(), addr.addr(), addr.addrSize()))) {
int savedErrno = errno;
ALOGE("Could not bind socket at %s: %s", addr.toString().c_str(), strerror(savedErrno));
return -savedErrno;
}
return setupRawSocketServer(std::move(socket_fd));
}
status_t RpcServer::setupRawSocketServer(unique_fd socket_fd) {
LOG_ALWAYS_FATAL_IF(!socket_fd.ok(), "Socket must be setup to listen.");
// Right now, we create all threads at once, making accept4 slow. To avoid hanging the client,
// the backlog is increased to a large number.
// TODO(b/189955605): Once we create threads dynamically & lazily, the backlog can be reduced
// to 1.
if (0 != TEMP_FAILURE_RETRY(listen(socket_fd.get(), 50 /*backlog*/))) {
int savedErrno = errno;
ALOGE("Could not listen initialized Unix socket: %s", strerror(savedErrno));
return -savedErrno;
}
if (status_t status = setupExternalServer(std::move(socket_fd)); status != OK) {
ALOGE("Another thread has set up server while calling setupSocketServer. Race?");
return status;
}
return OK;
}
void RpcServer::onSessionAllIncomingThreadsEnded(const sp<RpcSession>& session) {
const std::vector<uint8_t>& id = session->mId;
LOG_ALWAYS_FATAL_IF(id.empty(), "Server sessions must be initialized with ID");
LOG_RPC_DETAIL("Dropping session with address %s", HexString(id.data(), id.size()).c_str());
RpcMutexLockGuard _l(mLock);
auto it = mSessions.find(id);
LOG_ALWAYS_FATAL_IF(it == mSessions.end(), "Bad state, unknown session id %s",
HexString(id.data(), id.size()).c_str());
LOG_ALWAYS_FATAL_IF(it->second != session, "Bad state, session has id mismatch %s",
HexString(id.data(), id.size()).c_str());
(void)mSessions.erase(it);
}
void RpcServer::onSessionIncomingThreadEnded() {
mShutdownCv.notify_all();
}
bool RpcServer::hasServer() {
RpcMutexLockGuard _l(mLock);
return mServer.fd.ok();
}
unique_fd RpcServer::releaseServer() {
RpcMutexLockGuard _l(mLock);
return std::move(mServer.fd);
}
status_t RpcServer::setupExternalServer(
unique_fd serverFd, std::function<status_t(const RpcServer&, RpcTransportFd*)>&& acceptFn) {
RpcMutexLockGuard _l(mLock);
if (mServer.fd.ok()) {
ALOGE("Each RpcServer can only have one server.");
return INVALID_OPERATION;
}
mServer = std::move(serverFd);
mAcceptFn = std::move(acceptFn);
return OK;
}
status_t RpcServer::setupExternalServer(unique_fd serverFd) {
return setupExternalServer(std::move(serverFd), &RpcServer::acceptSocketConnection);
}
bool RpcServer::hasActiveRequests() {
RpcMutexLockGuard _l(mLock);
for (const auto& [_, session] : mSessions) {
if (session->hasActiveRequests()) {
return true;
}
}
return !mServer.isInPollingState();
}
} // namespace android