blob: 0dfad0e1e7b4b9c61ac449ae79010bdac74c3202 [file] [log] [blame]
/*
* Copyright (C) 2007 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 TRACE_TAG TRANSPORT
#include "sysdeps.h"
#include "transport.h"
#include <ctype.h>
#include <errno.h>
#include <inttypes.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <algorithm>
#include <list>
#include <memory>
#include <mutex>
#include <set>
#include <thread>
#include <adb/crypto/rsa_2048_key.h>
#include <adb/crypto/x509_generator.h>
#include <adb/tls/tls_connection.h>
#include <android-base/logging.h>
#include <android-base/no_destructor.h>
#include <android-base/parsenetaddress.h>
#include <android-base/stringprintf.h>
#include <android-base/strings.h>
#include <android-base/thread_annotations.h>
#include <diagnose_usb.h>
#include "adb.h"
#include "adb_auth.h"
#include "adb_io.h"
#include "adb_trace.h"
#include "adb_utils.h"
#include "fdevent/fdevent.h"
#include "sysdeps/chrono.h"
#if ADB_HOST
#include "client/usb.h"
#endif
using namespace adb::crypto;
using namespace adb::tls;
using android::base::ScopedLockAssertion;
using TlsError = TlsConnection::TlsError;
static void remove_transport(atransport* transport);
static void transport_destroy(atransport* transport);
// TODO: unordered_map<TransportId, atransport*>
static auto& transport_list = *new std::list<atransport*>();
static auto& pending_list = *new std::list<atransport*>();
static auto& transport_lock = *new std::recursive_mutex();
const char* const kFeatureShell2 = "shell_v2";
const char* const kFeatureCmd = "cmd";
const char* const kFeatureStat2 = "stat_v2";
const char* const kFeatureLs2 = "ls_v2";
const char* const kFeatureLibusb = "libusb";
const char* const kFeaturePushSync = "push_sync";
const char* const kFeatureApex = "apex";
const char* const kFeatureFixedPushMkdir = "fixed_push_mkdir";
const char* const kFeatureAbb = "abb";
const char* const kFeatureFixedPushSymlinkTimestamp = "fixed_push_symlink_timestamp";
const char* const kFeatureAbbExec = "abb_exec";
const char* const kFeatureRemountShell = "remount_shell";
const char* const kFeatureTrackApp = "track_app";
const char* const kFeatureSendRecv2 = "sendrecv_v2";
const char* const kFeatureSendRecv2Brotli = "sendrecv_v2_brotli";
const char* const kFeatureSendRecv2LZ4 = "sendrecv_v2_lz4";
const char* const kFeatureSendRecv2Zstd = "sendrecv_v2_zstd";
const char* const kFeatureSendRecv2DryRunSend = "sendrecv_v2_dry_run_send";
// TODO(joshuaduong): Bump to v2 when openscreen discovery is enabled by default
const char* const kFeatureOpenscreenMdns = "openscreen_mdns";
namespace {
#if ADB_HOST
// Tracks and handles atransport*s that are attempting reconnection.
class ReconnectHandler {
public:
ReconnectHandler() = default;
~ReconnectHandler() = default;
// Starts the ReconnectHandler thread.
void Start();
// Requests the ReconnectHandler thread to stop.
void Stop();
// Adds the atransport* to the queue of reconnect attempts.
void TrackTransport(atransport* transport);
// Wake up the ReconnectHandler thread to have it check for kicked transports.
void CheckForKicked();
private:
// The main thread loop.
void Run();
// Tracks a reconnection attempt.
struct ReconnectAttempt {
atransport* transport;
std::chrono::steady_clock::time_point reconnect_time;
size_t attempts_left;
bool operator<(const ReconnectAttempt& rhs) const {
if (reconnect_time == rhs.reconnect_time) {
return reinterpret_cast<uintptr_t>(transport) <
reinterpret_cast<uintptr_t>(rhs.transport);
}
return reconnect_time < rhs.reconnect_time;
}
};
// Only retry for up to one minute.
static constexpr const std::chrono::seconds kDefaultTimeout = 3s;
static constexpr const size_t kMaxAttempts = 20;
// Protects all members.
std::mutex reconnect_mutex_;
bool running_ GUARDED_BY(reconnect_mutex_) = true;
std::thread handler_thread_;
std::condition_variable reconnect_cv_;
std::set<ReconnectAttempt> reconnect_queue_ GUARDED_BY(reconnect_mutex_);
DISALLOW_COPY_AND_ASSIGN(ReconnectHandler);
};
void ReconnectHandler::Start() {
check_main_thread();
handler_thread_ = std::thread(&ReconnectHandler::Run, this);
}
void ReconnectHandler::Stop() {
check_main_thread();
{
std::lock_guard<std::mutex> lock(reconnect_mutex_);
running_ = false;
}
reconnect_cv_.notify_one();
handler_thread_.join();
// Drain the queue to free all resources.
std::lock_guard<std::mutex> lock(reconnect_mutex_);
while (!reconnect_queue_.empty()) {
ReconnectAttempt attempt = *reconnect_queue_.begin();
reconnect_queue_.erase(reconnect_queue_.begin());
remove_transport(attempt.transport);
}
}
void ReconnectHandler::TrackTransport(atransport* transport) {
check_main_thread();
{
std::lock_guard<std::mutex> lock(reconnect_mutex_);
if (!running_) return;
// Arbitrary sleep to give adbd time to get ready, if we disconnected because it exited.
auto reconnect_time = std::chrono::steady_clock::now() + 250ms;
reconnect_queue_.emplace(
ReconnectAttempt{transport, reconnect_time, ReconnectHandler::kMaxAttempts});
}
reconnect_cv_.notify_one();
}
void ReconnectHandler::CheckForKicked() {
reconnect_cv_.notify_one();
}
void ReconnectHandler::Run() {
while (true) {
ReconnectAttempt attempt;
{
std::unique_lock<std::mutex> lock(reconnect_mutex_);
ScopedLockAssertion assume_lock(reconnect_mutex_);
if (!reconnect_queue_.empty()) {
// FIXME: libstdc++ (used on Windows) implements condition_variable with
// system_clock as its clock, so we're probably hosed if the clock changes,
// even if we use steady_clock throughout. This problem goes away once we
// switch to libc++.
reconnect_cv_.wait_until(lock, reconnect_queue_.begin()->reconnect_time);
} else {
reconnect_cv_.wait(lock);
}
if (!running_) return;
// Scan the whole list for kicked transports, so that we immediately handle an explicit
// disconnect request.
for (auto it = reconnect_queue_.begin(); it != reconnect_queue_.end();) {
if (it->transport->kicked()) {
D("transport %s was kicked. giving up on it.", it->transport->serial.c_str());
remove_transport(it->transport);
it = reconnect_queue_.erase(it);
} else {
++it;
}
}
if (reconnect_queue_.empty()) continue;
// Go back to sleep if we either woke up spuriously, or we were woken up to remove
// a kicked transport, and the first transport isn't ready for reconnection yet.
auto now = std::chrono::steady_clock::now();
if (reconnect_queue_.begin()->reconnect_time > now) {
continue;
}
attempt = *reconnect_queue_.begin();
reconnect_queue_.erase(reconnect_queue_.begin());
}
D("attempting to reconnect %s", attempt.transport->serial.c_str());
switch (attempt.transport->Reconnect()) {
case ReconnectResult::Retry: {
D("attempting to reconnect %s failed.", attempt.transport->serial.c_str());
if (attempt.attempts_left == 0) {
D("transport %s exceeded the number of retry attempts. giving up on it.",
attempt.transport->serial.c_str());
remove_transport(attempt.transport);
continue;
}
std::lock_guard<std::mutex> lock(reconnect_mutex_);
reconnect_queue_.emplace(ReconnectAttempt{
attempt.transport,
std::chrono::steady_clock::now() + ReconnectHandler::kDefaultTimeout,
attempt.attempts_left - 1});
continue;
}
case ReconnectResult::Success:
D("reconnection to %s succeeded.", attempt.transport->serial.c_str());
register_transport(attempt.transport);
continue;
case ReconnectResult::Abort:
D("cancelling reconnection attempt to %s.", attempt.transport->serial.c_str());
remove_transport(attempt.transport);
continue;
}
}
}
static auto& reconnect_handler = *new ReconnectHandler();
#endif
} // namespace
TransportId NextTransportId() {
static std::atomic<TransportId> next(1);
return next++;
}
void Connection::Reset() {
LOG(INFO) << "Connection::Reset(): stopping";
Stop();
}
std::string Connection::Serial() const {
return transport_ ? transport_->serial_name() : "<unknown>";
}
BlockingConnectionAdapter::BlockingConnectionAdapter(std::unique_ptr<BlockingConnection> connection)
: underlying_(std::move(connection)) {}
BlockingConnectionAdapter::~BlockingConnectionAdapter() {
LOG(INFO) << "BlockingConnectionAdapter(" << Serial() << "): destructing";
Stop();
}
void BlockingConnectionAdapter::Start() {
std::lock_guard<std::mutex> lock(mutex_);
if (started_) {
LOG(FATAL) << "BlockingConnectionAdapter(" << Serial() << "): started multiple times";
}
StartReadThread();
write_thread_ = std::thread([this]() {
LOG(INFO) << Serial() << ": write thread spawning";
while (true) {
std::unique_lock<std::mutex> lock(mutex_);
ScopedLockAssertion assume_locked(mutex_);
cv_.wait(lock, [this]() REQUIRES(mutex_) {
return this->stopped_ || !this->write_queue_.empty();
});
if (this->stopped_) {
return;
}
std::unique_ptr<apacket> packet = std::move(this->write_queue_.front());
this->write_queue_.pop_front();
lock.unlock();
if (!this->underlying_->Write(packet.get())) {
break;
}
}
std::call_once(this->error_flag_, [this]() { transport_->HandleError("write failed"); });
});
started_ = true;
}
void BlockingConnectionAdapter::StartReadThread() {
read_thread_ = std::thread([this]() {
LOG(INFO) << Serial() << ": read thread spawning";
while (true) {
auto packet = std::make_unique<apacket>();
if (!underlying_->Read(packet.get())) {
PLOG(INFO) << Serial() << ": read failed";
break;
}
bool got_stls_cmd = false;
if (packet->msg.command == A_STLS) {
got_stls_cmd = true;
}
transport_->HandleRead(std::move(packet));
// If we received the STLS packet, we are about to perform the TLS
// handshake. So this read thread must stop and resume after the
// handshake completes otherwise this will interfere in the process.
if (got_stls_cmd) {
LOG(INFO) << Serial() << ": Received STLS packet. Stopping read thread.";
return;
}
}
std::call_once(this->error_flag_, [this]() { transport_->HandleError("read failed"); });
});
}
bool BlockingConnectionAdapter::DoTlsHandshake(RSA* key, std::string* auth_key) {
std::lock_guard<std::mutex> lock(mutex_);
if (read_thread_.joinable()) {
read_thread_.join();
}
bool success = this->underlying_->DoTlsHandshake(key, auth_key);
StartReadThread();
return success;
}
void BlockingConnectionAdapter::Reset() {
{
std::lock_guard<std::mutex> lock(mutex_);
if (!started_) {
LOG(INFO) << "BlockingConnectionAdapter(" << Serial() << "): not started";
return;
}
if (stopped_) {
LOG(INFO) << "BlockingConnectionAdapter(" << Serial() << "): already stopped";
return;
}
}
LOG(INFO) << "BlockingConnectionAdapter(" << Serial() << "): resetting";
this->underlying_->Reset();
Stop();
}
void BlockingConnectionAdapter::Stop() {
{
std::lock_guard<std::mutex> lock(mutex_);
if (!started_) {
LOG(INFO) << "BlockingConnectionAdapter(" << Serial() << "): not started";
return;
}
if (stopped_) {
LOG(INFO) << "BlockingConnectionAdapter(" << Serial() << "): already stopped";
return;
}
stopped_ = true;
}
LOG(INFO) << "BlockingConnectionAdapter(" << Serial() << "): stopping";
this->underlying_->Close();
this->cv_.notify_one();
// Move the threads out into locals with the lock taken, and then unlock to let them exit.
std::thread read_thread;
std::thread write_thread;
{
std::lock_guard<std::mutex> lock(mutex_);
read_thread = std::move(read_thread_);
write_thread = std::move(write_thread_);
}
read_thread.join();
write_thread.join();
LOG(INFO) << "BlockingConnectionAdapter(" << Serial() << "): stopped";
std::call_once(this->error_flag_, [this]() { transport_->HandleError("requested stop"); });
}
bool BlockingConnectionAdapter::Write(std::unique_ptr<apacket> packet) {
{
std::lock_guard<std::mutex> lock(this->mutex_);
write_queue_.emplace_back(std::move(packet));
}
cv_.notify_one();
return true;
}
FdConnection::FdConnection(unique_fd fd) : fd_(std::move(fd)) {}
FdConnection::~FdConnection() {}
bool FdConnection::DispatchRead(void* buf, size_t len) {
if (tls_ != nullptr) {
// The TlsConnection doesn't allow 0 byte reads
if (len == 0) {
return true;
}
return tls_->ReadFully(buf, len);
}
return ReadFdExactly(fd_.get(), buf, len);
}
bool FdConnection::DispatchWrite(void* buf, size_t len) {
if (tls_ != nullptr) {
// The TlsConnection doesn't allow 0 byte writes
if (len == 0) {
return true;
}
return tls_->WriteFully(std::string_view(reinterpret_cast<const char*>(buf), len));
}
return WriteFdExactly(fd_.get(), buf, len);
}
bool FdConnection::Read(apacket* packet) {
if (!DispatchRead(&packet->msg, sizeof(amessage))) {
D("remote local: read terminated (message)");
return false;
}
if (packet->msg.data_length > MAX_PAYLOAD) {
D("remote local: read overflow (data length = %" PRIu32 ")", packet->msg.data_length);
return false;
}
packet->payload.resize(packet->msg.data_length);
if (!DispatchRead(&packet->payload[0], packet->payload.size())) {
D("remote local: terminated (data)");
return false;
}
return true;
}
bool FdConnection::Write(apacket* packet) {
if (!DispatchWrite(&packet->msg, sizeof(packet->msg))) {
D("remote local: write terminated");
return false;
}
if (packet->msg.data_length) {
if (!DispatchWrite(&packet->payload[0], packet->msg.data_length)) {
D("remote local: write terminated");
return false;
}
}
return true;
}
bool FdConnection::DoTlsHandshake(RSA* key, std::string* auth_key) {
bssl::UniquePtr<EVP_PKEY> evp_pkey(EVP_PKEY_new());
if (!EVP_PKEY_set1_RSA(evp_pkey.get(), key)) {
LOG(ERROR) << "EVP_PKEY_set1_RSA failed";
return false;
}
auto x509 = GenerateX509Certificate(evp_pkey.get());
auto x509_str = X509ToPEMString(x509.get());
auto evp_str = Key::ToPEMString(evp_pkey.get());
int osh = cast_handle_to_int(adb_get_os_handle(fd_));
#if ADB_HOST
tls_ = TlsConnection::Create(TlsConnection::Role::Client, x509_str, evp_str, osh);
#else
tls_ = TlsConnection::Create(TlsConnection::Role::Server, x509_str, evp_str, osh);
#endif
CHECK(tls_);
#if ADB_HOST
// TLS 1.3 gives the client no message if the server rejected the
// certificate. This will enable a check in the tls connection to check
// whether the client certificate got rejected. Note that this assumes
// that, on handshake success, the server speaks first.
tls_->EnableClientPostHandshakeCheck(true);
// Add callback to set the certificate when server issues the
// CertificateRequest.
tls_->SetCertificateCallback(adb_tls_set_certificate);
// Allow any server certificate
tls_->SetCertVerifyCallback([](X509_STORE_CTX*) { return 1; });
#else
// Add callback to check certificate against a list of known public keys
tls_->SetCertVerifyCallback(
[auth_key](X509_STORE_CTX* ctx) { return adbd_tls_verify_cert(ctx, auth_key); });
// Add the list of allowed client CA issuers
auto ca_list = adbd_tls_client_ca_list();
tls_->SetClientCAList(ca_list.get());
#endif
auto err = tls_->DoHandshake();
if (err == TlsError::Success) {
return true;
}
tls_.reset();
return false;
}
void FdConnection::Close() {
adb_shutdown(fd_.get());
fd_.reset();
}
void send_packet(apacket* p, atransport* t) {
p->msg.magic = p->msg.command ^ 0xffffffff;
// compute a checksum for connection/auth packets for compatibility reasons
if (t->get_protocol_version() >= A_VERSION_SKIP_CHECKSUM) {
p->msg.data_check = 0;
} else {
p->msg.data_check = calculate_apacket_checksum(p);
}
VLOG(TRANSPORT) << dump_packet(t->serial.c_str(), "to remote", p);
if (t == nullptr) {
LOG(FATAL) << "Transport is null";
}
if (t->Write(p) != 0) {
D("%s: failed to enqueue packet, closing transport", t->serial.c_str());
t->Kick();
}
}
void kick_transport(atransport* t, bool reset) {
std::lock_guard<std::recursive_mutex> lock(transport_lock);
// As kick_transport() can be called from threads without guarantee that t is valid,
// check if the transport is in transport_list first.
//
// TODO(jmgao): WTF? Is this actually true?
if (std::find(transport_list.begin(), transport_list.end(), t) != transport_list.end()) {
if (reset) {
t->Reset();
} else {
t->Kick();
}
}
#if ADB_HOST
reconnect_handler.CheckForKicked();
#endif
}
static int transport_registration_send = -1;
static int transport_registration_recv = -1;
static fdevent* transport_registration_fde;
#if ADB_HOST
/* this adds support required by the 'track-devices' service.
* this is used to send the content of "list_transport" to any
* number of client connections that want it through a single
* live TCP connection
*/
struct device_tracker {
asocket socket;
bool update_needed = false;
bool long_output = false;
device_tracker* next = nullptr;
};
/* linked list of all device trackers */
static device_tracker* device_tracker_list;
static void device_tracker_remove(device_tracker* tracker) {
device_tracker** pnode = &device_tracker_list;
device_tracker* node = *pnode;
std::lock_guard<std::recursive_mutex> lock(transport_lock);
while (node) {
if (node == tracker) {
*pnode = node->next;
break;
}
pnode = &node->next;
node = *pnode;
}
}
static void device_tracker_close(asocket* socket) {
device_tracker* tracker = (device_tracker*)socket;
asocket* peer = socket->peer;
D("device tracker %p removed", tracker);
if (peer) {
peer->peer = nullptr;
peer->close(peer);
}
device_tracker_remove(tracker);
delete tracker;
}
static int device_tracker_enqueue(asocket* socket, apacket::payload_type) {
/* you can't read from a device tracker, close immediately */
device_tracker_close(socket);
return -1;
}
static int device_tracker_send(device_tracker* tracker, const std::string& string) {
asocket* peer = tracker->socket.peer;
apacket::payload_type data;
data.resize(4 + string.size());
char buf[5];
snprintf(buf, sizeof(buf), "%04x", static_cast<int>(string.size()));
memcpy(&data[0], buf, 4);
memcpy(&data[4], string.data(), string.size());
return peer->enqueue(peer, std::move(data));
}
static void device_tracker_ready(asocket* socket) {
device_tracker* tracker = reinterpret_cast<device_tracker*>(socket);
// We want to send the device list when the tracker connects
// for the first time, even if no update occurred.
if (tracker->update_needed) {
tracker->update_needed = false;
device_tracker_send(tracker, list_transports(tracker->long_output));
}
}
asocket* create_device_tracker(bool long_output) {
device_tracker* tracker = new device_tracker();
if (tracker == nullptr) LOG(FATAL) << "cannot allocate device tracker";
D("device tracker %p created", tracker);
tracker->socket.enqueue = device_tracker_enqueue;
tracker->socket.ready = device_tracker_ready;
tracker->socket.close = device_tracker_close;
tracker->update_needed = true;
tracker->long_output = long_output;
tracker->next = device_tracker_list;
device_tracker_list = tracker;
return &tracker->socket;
}
// Check if all of the USB transports are connected.
bool iterate_transports(std::function<bool(const atransport*)> fn) {
std::lock_guard<std::recursive_mutex> lock(transport_lock);
for (const auto& t : transport_list) {
if (!fn(t)) {
return false;
}
}
for (const auto& t : pending_list) {
if (!fn(t)) {
return false;
}
}
return true;
}
// Call this function each time the transport list has changed.
void update_transports() {
update_transport_status();
// Notify `adb track-devices` clients.
device_tracker* tracker = device_tracker_list;
while (tracker != nullptr) {
device_tracker* next = tracker->next;
// This may destroy the tracker if the connection is closed.
device_tracker_send(tracker, list_transports(tracker->long_output));
tracker = next;
}
}
#else
void update_transports() {
// Nothing to do on the device side.
}
#endif // ADB_HOST
struct tmsg {
atransport* transport;
int action;
};
static int transport_read_action(int fd, struct tmsg* m) {
char* p = (char*)m;
int len = sizeof(*m);
int r;
while (len > 0) {
r = adb_read(fd, p, len);
if (r > 0) {
len -= r;
p += r;
} else {
D("transport_read_action: on fd %d: %s", fd, strerror(errno));
return -1;
}
}
return 0;
}
static int transport_write_action(int fd, struct tmsg* m) {
char* p = (char*)m;
int len = sizeof(*m);
int r;
while (len > 0) {
r = adb_write(fd, p, len);
if (r > 0) {
len -= r;
p += r;
} else {
D("transport_write_action: on fd %d: %s", fd, strerror(errno));
return -1;
}
}
return 0;
}
static bool usb_devices_start_detached() {
#if ADB_HOST
static const char* env = getenv("ADB_LIBUSB_START_DETACHED");
static bool result = env && strcmp("1", env) == 0;
return should_use_libusb() && result;
#else
return false;
#endif
}
static void transport_registration_func(int _fd, unsigned ev, void*) {
tmsg m;
atransport* t;
if (!(ev & FDE_READ)) {
return;
}
if (transport_read_action(_fd, &m)) {
PLOG(FATAL) << "cannot read transport registration socket";
}
t = m.transport;
if (m.action == 0) {
D("transport: %s deleting", t->serial.c_str());
{
std::lock_guard<std::recursive_mutex> lock(transport_lock);
transport_list.remove(t);
}
delete t;
update_transports();
return;
}
/* don't create transport threads for inaccessible devices */
if (t->GetConnectionState() != kCsNoPerm) {
t->connection()->SetTransport(t);
if (t->type == kTransportUsb && usb_devices_start_detached()) {
t->SetConnectionState(kCsDetached);
} else {
t->connection()->Start();
#if ADB_HOST
send_connect(t);
#endif
}
}
{
std::lock_guard<std::recursive_mutex> lock(transport_lock);
auto it = std::find(pending_list.begin(), pending_list.end(), t);
if (it != pending_list.end()) {
pending_list.remove(t);
transport_list.push_front(t);
}
}
update_transports();
}
#if ADB_HOST
void init_reconnect_handler(void) {
reconnect_handler.Start();
}
#endif
void init_transport_registration(void) {
int s[2];
if (adb_socketpair(s)) {
PLOG(FATAL) << "cannot open transport registration socketpair";
}
D("socketpair: (%d,%d)", s[0], s[1]);
transport_registration_send = s[0];
transport_registration_recv = s[1];
transport_registration_fde =
fdevent_create(transport_registration_recv, transport_registration_func, nullptr);
fdevent_set(transport_registration_fde, FDE_READ);
}
void kick_all_transports() {
#if ADB_HOST
reconnect_handler.Stop();
#endif
// To avoid only writing part of a packet to a transport after exit, kick all transports.
std::lock_guard<std::recursive_mutex> lock(transport_lock);
for (auto t : transport_list) {
t->Kick();
}
}
void kick_all_tcp_tls_transports() {
std::lock_guard<std::recursive_mutex> lock(transport_lock);
for (auto t : transport_list) {
if (t->IsTcpDevice() && t->use_tls) {
t->Kick();
}
}
}
#if !ADB_HOST
void kick_all_transports_by_auth_key(std::string_view auth_key) {
std::lock_guard<std::recursive_mutex> lock(transport_lock);
for (auto t : transport_list) {
if (auth_key == t->auth_key) {
t->Kick();
}
}
}
#endif
/* the fdevent select pump is single threaded */
void register_transport(atransport* transport) {
tmsg m;
m.transport = transport;
m.action = 1;
D("transport: %s registered", transport->serial.c_str());
if (transport_write_action(transport_registration_send, &m)) {
PLOG(FATAL) << "cannot write transport registration socket";
}
}
static void remove_transport(atransport* transport) {
tmsg m;
m.transport = transport;
m.action = 0;
D("transport: %s removed", transport->serial.c_str());
if (transport_write_action(transport_registration_send, &m)) {
PLOG(FATAL) << "cannot write transport registration socket";
}
}
static void transport_destroy(atransport* t) {
check_main_thread();
CHECK(t != nullptr);
std::lock_guard<std::recursive_mutex> lock(transport_lock);
LOG(INFO) << "destroying transport " << t->serial_name();
t->connection()->Stop();
#if ADB_HOST
if (t->IsTcpDevice() && !t->kicked()) {
D("transport: %s destroy (attempting reconnection)", t->serial.c_str());
// We need to clear the transport's keys, so that on the next connection, it tries
// again from the beginning.
t->ResetKeys();
reconnect_handler.TrackTransport(t);
return;
}
#endif
D("transport: %s destroy (kicking and closing)", t->serial.c_str());
remove_transport(t);
}
#if ADB_HOST
static int qual_match(const std::string& to_test, const char* prefix, const std::string& qual,
bool sanitize_qual) {
if (to_test.empty()) /* Return true if both the qual and to_test are empty strings. */
return qual.empty();
if (qual.empty()) return 0;
const char* ptr = to_test.c_str();
if (prefix) {
while (*prefix) {
if (*prefix++ != *ptr++) return 0;
}
}
for (char ch : qual) {
if (sanitize_qual && !isalnum(ch)) ch = '_';
if (ch != *ptr++) return 0;
}
/* Everything matched so far. Return true if *ptr is a NUL. */
return !*ptr;
}
// Contains either a device serial string or a USB device address like "usb:2-6"
const char* __transport_server_one_device = nullptr;
void transport_set_one_device(const char* adb_one_device) {
__transport_server_one_device = adb_one_device;
}
const char* transport_get_one_device() {
return __transport_server_one_device;
}
bool transport_server_owns_device(std::string_view serial) {
if (!__transport_server_one_device) {
// If the server doesn't own one device, server owns all devices.
return true;
}
return serial.compare(__transport_server_one_device) == 0;
}
bool transport_server_owns_device(std::string_view dev_path, std::string_view serial) {
if (!__transport_server_one_device) {
// If the server doesn't own one device, server owns all devices.
return true;
}
return serial.compare(__transport_server_one_device) == 0 ||
dev_path.compare(__transport_server_one_device) == 0;
}
atransport* acquire_one_transport(TransportType type, const char* serial, TransportId transport_id,
bool* is_ambiguous, std::string* error_out,
bool accept_any_state) {
atransport* result = nullptr;
if (transport_id != 0) {
*error_out = android::base::StringPrintf("no device with transport id '%" PRIu64 "'",
transport_id);
} else if (serial) {
*error_out = android::base::StringPrintf("device '%s' not found", serial);
} else if (type == kTransportLocal) {
*error_out = "no emulators found";
} else if (type == kTransportAny) {
*error_out = "no devices/emulators found";
} else {
*error_out = "no devices found";
}
std::unique_lock<std::recursive_mutex> lock(transport_lock);
for (const auto& t : transport_list) {
if (t->GetConnectionState() == kCsNoPerm) {
*error_out = UsbNoPermissionsLongHelpText();
continue;
}
if (transport_id) {
if (t->id == transport_id) {
result = t;
break;
}
} else if (serial) {
if (t->MatchesTarget(serial)) {
if (result) {
*error_out = "more than one device";
if (is_ambiguous) *is_ambiguous = true;
result = nullptr;
break;
}
result = t;
}
} else {
if (type == kTransportUsb && t->type == kTransportUsb) {
if (result) {
*error_out = "more than one device";
if (is_ambiguous) *is_ambiguous = true;
result = nullptr;
break;
}
result = t;
} else if (type == kTransportLocal && t->type == kTransportLocal) {
if (result) {
*error_out = "more than one emulator";
if (is_ambiguous) *is_ambiguous = true;
result = nullptr;
break;
}
result = t;
} else if (type == kTransportAny) {
if (result) {
*error_out = "more than one device/emulator";
if (is_ambiguous) *is_ambiguous = true;
result = nullptr;
break;
}
result = t;
}
}
}
lock.unlock();
if (result && !accept_any_state) {
// The caller requires an active transport.
// Make sure that we're actually connected.
ConnectionState state = result->GetConnectionState();
switch (state) {
case kCsConnecting:
*error_out = "device still connecting";
result = nullptr;
break;
case kCsAuthorizing:
*error_out = "device still authorizing";
result = nullptr;
break;
case kCsUnauthorized: {
*error_out = "device unauthorized.\n";
char* ADB_VENDOR_KEYS = getenv("ADB_VENDOR_KEYS");
*error_out += "This adb server's $ADB_VENDOR_KEYS is ";
*error_out += ADB_VENDOR_KEYS ? ADB_VENDOR_KEYS : "not set";
*error_out += "\n";
*error_out += "Try 'adb kill-server' if that seems wrong.\n";
*error_out += "Otherwise check for a confirmation dialog on your device.";
result = nullptr;
break;
}
case kCsOffline:
*error_out = "device offline";
result = nullptr;
break;
default:
break;
}
}
if (result) {
*error_out = "success";
}
return result;
}
bool ConnectionWaitable::WaitForConnection(std::chrono::milliseconds timeout) {
std::unique_lock<std::mutex> lock(mutex_);
ScopedLockAssertion assume_locked(mutex_);
return cv_.wait_for(lock, timeout, [&]() REQUIRES(mutex_) {
return connection_established_ready_;
}) && connection_established_;
}
void ConnectionWaitable::SetConnectionEstablished(bool success) {
{
std::lock_guard<std::mutex> lock(mutex_);
if (connection_established_ready_) return;
connection_established_ready_ = true;
connection_established_ = success;
D("connection established with %d", success);
}
cv_.notify_one();
}
#endif
atransport::~atransport() {
#if ADB_HOST
// If the connection callback had not been run before, run it now.
SetConnectionEstablished(false);
#endif
}
int atransport::Write(apacket* p) {
return this->connection()->Write(std::unique_ptr<apacket>(p)) ? 0 : -1;
}
void atransport::Reset() {
if (!kicked_.exchange(true)) {
LOG(INFO) << "resetting transport " << this << " " << this->serial;
this->connection()->Reset();
}
}
void atransport::Kick() {
if (!kicked_.exchange(true)) {
LOG(INFO) << "kicking transport " << this << " " << this->serial;
this->connection()->Stop();
}
}
ConnectionState atransport::GetConnectionState() const {
return connection_state_;
}
void atransport::SetConnectionState(ConnectionState state) {
check_main_thread();
connection_state_ = state;
update_transports();
}
#if ADB_HOST
bool atransport::Attach(std::string* error) {
D("%s: attach", serial.c_str());
check_main_thread();
if (!should_use_libusb()) {
*error = "attach/detach only implemented for libusb backend";
return false;
}
if (GetConnectionState() != ConnectionState::kCsDetached) {
*error = android::base::StringPrintf("transport %s is not detached", serial.c_str());
return false;
}
ResetKeys();
{
std::lock_guard<std::mutex> lock(mutex_);
if (!connection_->Attach(error)) {
return false;
}
}
send_connect(this);
return true;
}
bool atransport::Detach(std::string* error) {
D("%s: detach", serial.c_str());
check_main_thread();
if (!should_use_libusb()) {
*error = "attach/detach only implemented for libusb backend";
return false;
}
if (GetConnectionState() == ConnectionState::kCsDetached) {
*error = android::base::StringPrintf("transport %s is already detached", serial.c_str());
return false;
}
handle_offline(this);
{
std::lock_guard<std::mutex> lock(mutex_);
if (!connection_->Detach(error)) {
return false;
}
}
this->SetConnectionState(kCsDetached);
return true;
}
#endif
void atransport::SetConnection(std::shared_ptr<Connection> connection) {
std::lock_guard<std::mutex> lock(mutex_);
connection_ = std::shared_ptr<Connection>(std::move(connection));
}
bool atransport::HandleRead(std::unique_ptr<apacket> p) {
if (!check_header(p.get(), this)) {
D("%s: remote read: bad header", serial.c_str());
return false;
}
VLOG(TRANSPORT) << dump_packet(serial.c_str(), "from remote", p.get());
apacket* packet = p.release();
// TODO: Does this need to run on the main thread?
fdevent_run_on_main_thread([packet, this]() { handle_packet(packet, this); });
return true;
}
void atransport::HandleError(const std::string& error) {
LOG(INFO) << serial_name() << ": connection terminated: " << error;
fdevent_run_on_main_thread([this]() {
handle_offline(this);
transport_destroy(this);
});
}
void atransport::update_version(int version, size_t payload) {
protocol_version = std::min(version, A_VERSION);
max_payload = std::min(payload, MAX_PAYLOAD);
}
int atransport::get_protocol_version() const {
return protocol_version;
}
int atransport::get_tls_version() const {
return tls_version;
}
size_t atransport::get_max_payload() const {
return max_payload;
}
const FeatureSet& supported_features() {
static const android::base::NoDestructor<FeatureSet> features([] {
return FeatureSet{
kFeatureShell2,
kFeatureCmd,
kFeatureStat2,
kFeatureLs2,
kFeatureFixedPushMkdir,
kFeatureApex,
kFeatureAbb,
kFeatureFixedPushSymlinkTimestamp,
kFeatureAbbExec,
kFeatureRemountShell,
kFeatureTrackApp,
kFeatureSendRecv2,
kFeatureSendRecv2Brotli,
kFeatureSendRecv2LZ4,
kFeatureSendRecv2Zstd,
kFeatureSendRecv2DryRunSend,
kFeatureOpenscreenMdns,
// Increment ADB_SERVER_VERSION when adding a feature that adbd needs
// to know about. Otherwise, the client can be stuck running an old
// version of the server even after upgrading their copy of adb.
// (http://b/24370690)
};
}());
return *features;
}
std::string FeatureSetToString(const FeatureSet& features) {
return android::base::Join(features, ',');
}
FeatureSet StringToFeatureSet(const std::string& features_string) {
if (features_string.empty()) {
return FeatureSet();
}
return android::base::Split(features_string, ",");
}
template <class Range, class Value>
static bool contains(const Range& r, const Value& v) {
return std::find(std::begin(r), std::end(r), v) != std::end(r);
}
bool CanUseFeature(const FeatureSet& feature_set, const std::string& feature) {
return contains(feature_set, feature) && contains(supported_features(), feature);
}
bool atransport::has_feature(const std::string& feature) const {
return contains(features_, feature);
}
void atransport::SetFeatures(const std::string& features_string) {
features_ = StringToFeatureSet(features_string);
}
void atransport::AddDisconnect(adisconnect* disconnect) {
disconnects_.push_back(disconnect);
}
void atransport::RemoveDisconnect(adisconnect* disconnect) {
disconnects_.remove(disconnect);
}
void atransport::RunDisconnects() {
for (const auto& disconnect : disconnects_) {
disconnect->func(disconnect->opaque, this);
}
disconnects_.clear();
}
#if ADB_HOST
bool atransport::MatchesTarget(const std::string& target) const {
if (!serial.empty()) {
if (target == serial) {
return true;
} else if (type == kTransportLocal) {
// Local transports can match [tcp:|udp:]<hostname>[:port].
const char* local_target_ptr = target.c_str();
// For fastboot compatibility, ignore protocol prefixes.
if (android::base::StartsWith(target, "tcp:") ||
android::base::StartsWith(target, "udp:")) {
local_target_ptr += 4;
}
// Parse our |serial| and the given |target| to check if the hostnames and ports match.
std::string serial_host, error;
int serial_port = -1;
if (android::base::ParseNetAddress(serial, &serial_host, &serial_port, nullptr,
&error)) {
// |target| may omit the port to default to ours.
std::string target_host;
int target_port = serial_port;
if (android::base::ParseNetAddress(local_target_ptr, &target_host, &target_port,
nullptr, &error) &&
serial_host == target_host && serial_port == target_port) {
return true;
}
}
}
}
return (target == devpath) || qual_match(target, "product:", product, false) ||
qual_match(target, "model:", model, true) ||
qual_match(target, "device:", device, false);
}
void atransport::SetConnectionEstablished(bool success) {
connection_waitable_->SetConnectionEstablished(success);
}
ReconnectResult atransport::Reconnect() {
return reconnect_(this);
}
// We use newline as our delimiter, make sure to never output it.
static std::string sanitize(std::string str, bool alphanumeric) {
auto pred = alphanumeric ? [](const char c) { return !isalnum(c); }
: [](const char c) { return c == '\n'; };
std::replace_if(str.begin(), str.end(), pred, '_');
return str;
}
static void append_transport_info(std::string* result, const char* key, const std::string& value,
bool alphanumeric) {
if (value.empty()) {
return;
}
*result += ' ';
*result += key;
*result += sanitize(value, alphanumeric);
}
static void append_transport(const atransport* t, std::string* result, bool long_listing) {
std::string serial = t->serial;
if (serial.empty()) {
serial = "(no serial number)";
}
if (!long_listing) {
*result += serial;
*result += '\t';
*result += to_string(t->GetConnectionState());
} else {
android::base::StringAppendF(result, "%-22s %s", serial.c_str(),
to_string(t->GetConnectionState()).c_str());
append_transport_info(result, "", t->devpath, false);
append_transport_info(result, "product:", t->product, false);
append_transport_info(result, "model:", t->model, true);
append_transport_info(result, "device:", t->device, false);
// Put id at the end, so that anyone parsing the output here can always find it by scanning
// backwards from newlines, even with hypothetical devices named 'transport_id:1'.
*result += " transport_id:";
*result += std::to_string(t->id);
}
*result += '\n';
}
std::string list_transports(bool long_listing) {
std::lock_guard<std::recursive_mutex> lock(transport_lock);
auto sorted_transport_list = transport_list;
sorted_transport_list.sort([](atransport*& x, atransport*& y) {
if (x->type != y->type) {
return x->type < y->type;
}
return x->serial < y->serial;
});
std::string result;
for (const auto& t : sorted_transport_list) {
append_transport(t, &result, long_listing);
}
return result;
}
void close_usb_devices(std::function<bool(const atransport*)> predicate, bool reset) {
std::lock_guard<std::recursive_mutex> lock(transport_lock);
for (auto& t : transport_list) {
if (predicate(t)) {
if (reset) {
t->Reset();
} else {
t->Kick();
}
}
}
}
/* hack for osx */
void close_usb_devices(bool reset) {
close_usb_devices([](const atransport*) { return true; }, reset);
}
#endif
bool register_socket_transport(unique_fd s, std::string serial, int port, int local,
atransport::ReconnectCallback reconnect, bool use_tls, int* error) {
atransport* t = new atransport(std::move(reconnect), kCsOffline);
t->use_tls = use_tls;
D("transport: %s init'ing for socket %d, on port %d", serial.c_str(), s.get(), port);
if (init_socket_transport(t, std::move(s), port, local) < 0) {
delete t;
if (error) *error = errno;
return false;
}
std::unique_lock<std::recursive_mutex> lock(transport_lock);
for (const auto& transport : pending_list) {
if (serial == transport->serial) {
VLOG(TRANSPORT) << "socket transport " << transport->serial
<< " is already in pending_list and fails to register";
delete t;
if (error) *error = EALREADY;
return false;
}
}
for (const auto& transport : transport_list) {
if (serial == transport->serial) {
VLOG(TRANSPORT) << "socket transport " << transport->serial
<< " is already in transport_list and fails to register";
delete t;
if (error) *error = EALREADY;
return false;
}
}
t->serial = std::move(serial);
pending_list.push_front(t);
lock.unlock();
#if ADB_HOST
auto waitable = t->connection_waitable();
#endif
register_transport(t);
if (local == 1) {
// Do not wait for emulator transports.
return true;
}
#if ADB_HOST
if (!waitable->WaitForConnection(std::chrono::seconds(10))) {
if (error) *error = ETIMEDOUT;
return false;
}
if (t->GetConnectionState() == kCsUnauthorized) {
if (error) *error = EPERM;
return false;
}
#endif
return true;
}
#if ADB_HOST
atransport* find_transport(const char* serial) {
atransport* result = nullptr;
std::lock_guard<std::recursive_mutex> lock(transport_lock);
for (auto& t : transport_list) {
if (strcmp(serial, t->serial.c_str()) == 0) {
result = t;
break;
}
}
return result;
}
void kick_all_tcp_devices() {
std::lock_guard<std::recursive_mutex> lock(transport_lock);
for (auto& t : transport_list) {
if (t->IsTcpDevice()) {
// Kicking breaks the read_transport thread of this transport out of any read, then
// the read_transport thread will notify the main thread to make this transport
// offline. Then the main thread will notify the write_transport thread to exit.
// Finally, this transport will be closed and freed in the main thread.
t->Kick();
}
}
reconnect_handler.CheckForKicked();
}
void register_usb_transport(std::shared_ptr<Connection> connection, const char* serial,
const char* devpath, unsigned writeable) {
atransport* t = new atransport(writeable ? kCsOffline : kCsNoPerm);
if (serial) {
t->serial = serial;
}
if (devpath) {
t->devpath = devpath;
}
t->SetConnection(std::move(connection));
t->type = kTransportUsb;
{
std::lock_guard<std::recursive_mutex> lock(transport_lock);
pending_list.push_front(t);
}
register_transport(t);
}
void register_usb_transport(usb_handle* usb, const char* serial, const char* devpath,
unsigned writeable) {
atransport* t = new atransport(writeable ? kCsOffline : kCsNoPerm);
D("transport: %p init'ing for usb_handle %p (sn='%s')", t, usb, serial ? serial : "");
init_usb_transport(t, usb);
if (serial) {
t->serial = serial;
}
if (devpath) {
t->devpath = devpath;
}
{
std::lock_guard<std::recursive_mutex> lock(transport_lock);
pending_list.push_front(t);
}
register_transport(t);
}
// This should only be used for transports with connection_state == kCsNoPerm.
void unregister_usb_transport(usb_handle* usb) {
std::lock_guard<std::recursive_mutex> lock(transport_lock);
transport_list.remove_if([usb](atransport* t) {
return t->GetUsbHandle() == usb && t->GetConnectionState() == kCsNoPerm;
});
}
#endif
bool check_header(apacket* p, atransport* t) {
if (p->msg.magic != (p->msg.command ^ 0xffffffff)) {
VLOG(RWX) << "check_header(): invalid magic command = " << std::hex << p->msg.command
<< ", magic = " << p->msg.magic;
return false;
}
if (p->msg.data_length > t->get_max_payload()) {
VLOG(RWX) << "check_header(): " << p->msg.data_length
<< " atransport::max_payload = " << t->get_max_payload();
return false;
}
return true;
}
#if ADB_HOST
std::shared_ptr<RSA> atransport::Key() {
if (keys_.empty()) {
return nullptr;
}
std::shared_ptr<RSA> result = keys_[0];
return result;
}
std::shared_ptr<RSA> atransport::NextKey() {
if (keys_.empty()) {
LOG(INFO) << "fetching keys for transport " << this->serial_name();
keys_ = adb_auth_get_private_keys();
// We should have gotten at least one key: the one that's automatically generated.
CHECK(!keys_.empty());
} else {
keys_.pop_front();
}
return Key();
}
void atransport::ResetKeys() {
keys_.clear();
}
#endif