#include "dbus/object_proxy.h"
#include <stddef.h>
#include <utility>
#include "base/check.h"
#include "base/containers/contains.h"
#include "base/debug/alias.h"
#include "base/debug/crash_logging.h"
#include "base/debug/leak_annotations.h"
#include "base/functional/bind.h"
#include "base/functional/callback_helpers.h"
#include "base/logging.h"
#include "base/strings/stringprintf.h"
#include "base/task/sequenced_task_runner.h"
#include "base/threading/scoped_blocking_call.h"
#include "base/threading/thread.h"
#include "base/threading/thread_restrictions.h"
#include "build/build_config.h"
#include "dbus/bus.h"
#include "dbus/dbus_statistics.h"
#include "dbus/error.h"
#include "dbus/message.h"
#include "dbus/object_path.h"
#include "dbus/scoped_dbus_error.h"
#include "dbus/util.h"
namespace dbus {
namespace {
constexpr char kErrorServiceUnknown[] =
"org.freedesktop.DBus.Error.ServiceUnknown";
constexpr char kErrorObjectUnknown[] =
"org.freedesktop.DBus.Error.UnknownObject";
constexpr char kDBusSystemObjectPath[] = "/org/freedesktop/DBus";
constexpr char kDBusSystemObjectInterface[] = "org.freedesktop.DBus";
constexpr char kDBusSystemObjectAddress[] = "org.freedesktop.DBus";
constexpr char kNameOwnerChangedMember[] = "NameOwnerChanged";
}
ObjectProxy::ReplyCallbackHolder::ReplyCallbackHolder(
scoped_refptr<base::SequencedTaskRunner> origin_task_runner,
ResponseOrErrorCallback callback)
: origin_task_runner_(std::move(origin_task_runner)),
callback_(std::move(callback)) {
DCHECK(origin_task_runner_.get());
DCHECK(!callback_.is_null());
}
ObjectProxy::ReplyCallbackHolder::ReplyCallbackHolder(
ReplyCallbackHolder&& other) = default;
ObjectProxy::ReplyCallbackHolder::~ReplyCallbackHolder() {
if (callback_.is_null()) {
return;
}
DCHECK(origin_task_runner_.get());
if (origin_task_runner_->RunsTasksInCurrentSequence()) {
return;
}
auto* callback_to_be_deleted =
new ResponseOrErrorCallback(std::move(callback_));
ANNOTATE_LEAKING_OBJECT_PTR(callback_to_be_deleted);
origin_task_runner_->PostTask(
FROM_HERE, base::BindOnce(&base::DeletePointer<ResponseOrErrorCallback>,
callback_to_be_deleted));
}
ObjectProxy::ResponseOrErrorCallback
ObjectProxy::ReplyCallbackHolder::ReleaseCallback() {
DCHECK(origin_task_runner_->RunsTasksInCurrentSequence());
return std::move(callback_);
}
ObjectProxy::ObjectProxy(Bus* bus,
std::string_view service_name,
const ObjectPath& object_path,
int options)
: bus_(bus),
service_name_(service_name),
object_path_(object_path),
ignore_service_unknown_errors_(options & IGNORE_SERVICE_UNKNOWN_ERRORS) {
LOG_IF(FATAL, !object_path_.IsValid()) << object_path_.value();
}
ObjectProxy::~ObjectProxy() {
DCHECK(pending_calls_.empty());
}
base::expected<std::unique_ptr<Response>, Error>
ObjectProxy::CallMethodAndBlock(MethodCall* method_call, int timeout_ms) {
bus_->AssertOnDBusThread();
#if BUILDFLAG(IS_CHROMEOS)
CHECK_LE(timeout_ms, TIMEOUT_MAX);
#endif
if (!bus_->Connect() || !method_call->SetDestination(service_name_) ||
!method_call->SetPath(object_path_)) {
return base::unexpected(Error());
}
auto result =
bus_->SendWithReplyAndBlock(method_call->raw_message(), timeout_ms);
statistics::AddBlockingSentMethodCall(
service_name_, method_call->GetInterface(), method_call->GetMember());
if (!result.has_value()) {
LogMethodCallFailure(method_call->GetInterface(), method_call->GetMember(),
result.error().name(), result.error().message());
}
return result;
}
void ObjectProxy::CallMethod(MethodCall* method_call,
int timeout_ms,
ResponseCallback callback) {
auto internal_callback = base::BindOnce(
&ObjectProxy::OnCallMethod, this, method_call->GetInterface(),
method_call->GetMember(), std::move(callback));
CallMethodWithErrorResponse(method_call, timeout_ms,
std::move(internal_callback));
}
void ObjectProxy::CallMethodWithErrorResponse(
MethodCall* method_call,
int timeout_ms,
ResponseOrErrorCallback callback) {
bus_->AssertOnOriginThread();
#if BUILDFLAG(IS_CHROMEOS)
CHECK_LE(timeout_ms, TIMEOUT_MAX);
#endif
ReplyCallbackHolder callback_holder(bus_->GetOriginTaskRunner(),
std::move(callback));
if (!method_call->SetDestination(service_name_) ||
!method_call->SetPath(object_path_)) {
base::OnceClosure task =
base::BindOnce(&ObjectProxy::RunResponseOrErrorCallback, this,
std::move(callback_holder), nullptr ,
nullptr );
bus_->GetOriginTaskRunner()->PostTask(FROM_HERE, std::move(task));
return;
}
DBusMessage* request_message = method_call->raw_message();
dbus_message_ref(request_message);
statistics::AddSentMethodCall(service_name_, method_call->GetInterface(),
method_call->GetMember());
base::OnceClosure task =
base::BindOnce(&ObjectProxy::StartAsyncMethodCall, this, timeout_ms,
request_message, std::move(callback_holder));
bus_->GetDBusTaskRunner()->PostTask(FROM_HERE, std::move(task));
}
void ObjectProxy::ConnectToSignal(const std::string& interface_name,
const std::string& signal_name,
SignalCallback signal_callback,
OnConnectedCallback on_connected_callback) {
bus_->AssertOnOriginThread();
if (bus_->HasDBusThread()) {
bus_->GetDBusTaskRunner()->PostTaskAndReplyWithResult(
FROM_HERE,
base::BindOnce(&ObjectProxy::ConnectToSignalAndBlock, this,
interface_name, signal_name, signal_callback),
base::BindOnce(std::move(on_connected_callback), interface_name,
signal_name));
} else {
const bool success =
ConnectToSignalAndBlock(interface_name, signal_name, signal_callback);
std::move(on_connected_callback).Run(interface_name, signal_name, success);
}
}
bool ObjectProxy::ConnectToSignalAndBlock(const std::string& interface_name,
const std::string& signal_name,
SignalCallback signal_callback) {
bus_->AssertOnDBusThread();
if (!ConnectToNameOwnerChangedSignal())
return false;
const std::string absolute_signal_name =
GetAbsoluteMemberName(interface_name, signal_name);
const std::string match_rule = base::StringPrintf(
"type='signal', sender='%s', interface='%s', path='%s'",
service_name_.c_str(), interface_name.c_str(),
object_path_.value().c_str());
return AddMatchRuleWithCallback(match_rule, absolute_signal_name,
signal_callback);
}
void ObjectProxy::SetNameOwnerChangedCallback(
NameOwnerChangedCallback callback) {
bus_->AssertOnOriginThread();
name_owner_changed_callback_ = callback;
bus_->GetDBusTaskRunner()->PostTask(
FROM_HERE,
base::BindOnce(&ObjectProxy::TryConnectToNameOwnerChangedSignal, this));
}
void ObjectProxy::WaitForServiceToBeAvailable(
WaitForServiceToBeAvailableCallback callback) {
bus_->AssertOnOriginThread();
wait_for_service_to_be_available_callbacks_.push_back(std::move(callback));
bus_->GetDBusTaskRunner()->PostTask(
FROM_HERE,
base::BindOnce(&ObjectProxy::WaitForServiceToBeAvailableInternal, this));
}
void ObjectProxy::Detach() {
bus_->AssertOnDBusThread();
if (bus_->IsConnected())
bus_->RemoveFilterFunction(&ObjectProxy::HandleMessageThunk, this);
for (const auto& match_rule : match_rules_) {
Error error;
bus_->RemoveMatch(match_rule, &error);
if (error.IsValid()) {
LOG(ERROR) << "Failed to remove match rule: " << match_rule;
}
}
match_rules_.clear();
for (DBusPendingCall* pending_call : pending_calls_) {
base::ScopedBlockingCall scoped_blocking_call(
FROM_HERE, base::BlockingType::MAY_BLOCK);
dbus_pending_call_cancel(pending_call);
dbus_pending_call_unref(pending_call);
}
pending_calls_.clear();
}
void ObjectProxy::StartAsyncMethodCall(int timeout_ms,
DBusMessage* request_message,
ReplyCallbackHolder callback_holder) {
bus_->AssertOnDBusThread();
base::ScopedBlockingCall scoped_blocking_call(FROM_HERE,
base::BlockingType::MAY_BLOCK);
if (!bus_->Connect() || !bus_->SetUpAsyncOperations()) {
base::OnceClosure task =
base::BindOnce(&ObjectProxy::RunResponseOrErrorCallback, this,
std::move(callback_holder), nullptr ,
nullptr );
bus_->GetOriginTaskRunner()->PostTask(FROM_HERE, std::move(task));
dbus_message_unref(request_message);
return;
}
DBusPendingCall* dbus_pending_call = nullptr;
bus_->SendWithReply(request_message, &dbus_pending_call, timeout_ms);
using PendingCallback =
base::OnceCallback<void(DBusPendingCall * pending_call)>;
const bool success = dbus_pending_call_set_notify(
dbus_pending_call,
[](DBusPendingCall* pending_call, void* user_data) {
std::move(*static_cast<PendingCallback*>(user_data)).Run(pending_call);
},
new PendingCallback(base::BindOnce(&ObjectProxy::OnPendingCallIsComplete,
this, std::move(callback_holder))),
[](void* user_data) { delete static_cast<PendingCallback*>(user_data); });
CHECK(success) << "Unable to allocate memory";
pending_calls_.insert(dbus_pending_call);
dbus_message_unref(request_message);
}
void ObjectProxy::OnPendingCallIsComplete(ReplyCallbackHolder callback_holder,
DBusPendingCall* pending_call) {
bus_->AssertOnDBusThread();
base::ScopedBlockingCall scoped_blocking_call(FROM_HERE,
base::BlockingType::MAY_BLOCK);
DBusMessage* response_message = dbus_pending_call_steal_reply(pending_call);
std::unique_ptr<Response> response;
std::unique_ptr<ErrorResponse> error_response;
if (dbus_message_get_type(response_message) == DBUS_MESSAGE_TYPE_ERROR) {
error_response = ErrorResponse::FromRawMessage(response_message);
} else {
response = Response::FromRawMessage(response_message);
}
base::OnceClosure task = base::BindOnce(
&ObjectProxy::RunResponseOrErrorCallback, this,
std::move(callback_holder), response.get(), error_response.get());
bus_->GetOriginTaskRunner()->PostTaskAndReply(
FROM_HERE, std::move(task),
base::BindOnce(
[](Response* response, ErrorResponse* error_response) {
},
base::Owned(response.release()),
base::Owned(error_response.release())));
pending_calls_.erase(pending_call);
dbus_pending_call_unref(pending_call);
}
void ObjectProxy::RunResponseOrErrorCallback(
ReplyCallbackHolder callback_holder,
Response* response,
ErrorResponse* error_response) {
bus_->AssertOnOriginThread();
callback_holder.ReleaseCallback().Run(response, error_response);
}
bool ObjectProxy::ConnectToNameOwnerChangedSignal() {
bus_->AssertOnDBusThread();
if (!bus_->Connect() || !bus_->SetUpAsyncOperations())
return false;
bus_->AddFilterFunction(&ObjectProxy::HandleMessageThunk, this);
const std::string name_owner_changed_match_rule = base::StringPrintf(
"type='signal',interface='org.freedesktop.DBus',"
"member='NameOwnerChanged',path='/org/freedesktop/DBus',"
"sender='org.freedesktop.DBus',arg0='%s'",
service_name_.c_str());
const bool success = AddMatchRuleWithoutCallback(
name_owner_changed_match_rule, "org.freedesktop.DBus.NameOwnerChanged");
UpdateNameOwnerAndBlock();
return success;
}
void ObjectProxy::TryConnectToNameOwnerChangedSignal() {
bus_->AssertOnDBusThread();
bool success = ConnectToNameOwnerChangedSignal();
LOG_IF(WARNING, !success)
<< "Failed to connect to NameOwnerChanged signal for object: "
<< object_path_.value();
}
void ObjectProxy::WaitForServiceToBeAvailableInternal() {
bus_->AssertOnDBusThread();
if (!ConnectToNameOwnerChangedSignal()) {
const bool service_is_ready = false;
bus_->GetOriginTaskRunner()->PostTask(
FROM_HERE,
base::BindOnce(&ObjectProxy::RunWaitForServiceToBeAvailableCallbacks,
this, service_is_ready));
return;
}
const bool service_is_available = !service_name_owner_.empty();
if (service_is_available) {
bus_->GetOriginTaskRunner()->PostTask(
FROM_HERE,
base::BindOnce(&ObjectProxy::RunWaitForServiceToBeAvailableCallbacks,
this, service_is_available));
return;
}
}
DBusHandlerResult ObjectProxy::HandleMessage(DBusConnection* connection,
DBusMessage* raw_message) {
bus_->AssertOnDBusThread();
if (dbus_message_get_type(raw_message) != DBUS_MESSAGE_TYPE_SIGNAL)
return DBUS_HANDLER_RESULT_NOT_YET_HANDLED;
dbus_message_ref(raw_message);
std::unique_ptr<Signal> signal(Signal::FromRawMessage(raw_message));
const ObjectPath path = signal->GetPath();
if (path != object_path_) {
if (path.value() == kDBusSystemObjectPath &&
signal->GetMember() == kNameOwnerChangedMember) {
return HandleNameOwnerChanged(std::move(signal));
}
return DBUS_HANDLER_RESULT_NOT_YET_HANDLED;
}
std::string sender = signal->GetSender();
if (service_name_owner_ != sender)
return DBUS_HANDLER_RESULT_NOT_YET_HANDLED;
const std::string interface = signal->GetInterface();
const std::string member = signal->GetMember();
statistics::AddReceivedSignal(service_name_, interface, member);
const std::string absolute_signal_name =
GetAbsoluteMemberName(interface, member);
MethodTable::const_iterator iter = method_table_.find(absolute_signal_name);
if (iter == method_table_.end()) {
return DBUS_HANDLER_RESULT_NOT_YET_HANDLED;
}
VLOG(1) << "Signal received: " << signal->ToString();
if (bus_->HasDBusThread()) {
Signal* released_signal = signal.release();
bus_->GetOriginTaskRunner()->PostTask(
FROM_HERE, base::BindOnce(&ObjectProxy::RunMethod, this, iter->second,
released_signal));
} else {
Signal* released_signal = signal.release();
RunMethod(iter->second, released_signal);
}
return DBUS_HANDLER_RESULT_NOT_YET_HANDLED;
}
void ObjectProxy::RunMethod(std::vector<SignalCallback> signal_callbacks,
Signal* signal) {
bus_->AssertOnOriginThread();
for (auto& signal_callback : signal_callbacks) {
signal_callback.Run(signal);
}
bus_->GetDBusTaskRunner()->PostTask(
FROM_HERE, base::BindOnce(&base::DeletePointer<Signal>, signal));
}
DBusHandlerResult ObjectProxy::HandleMessageThunk(DBusConnection* connection,
DBusMessage* raw_message,
void* user_data) {
ObjectProxy* self = reinterpret_cast<ObjectProxy*>(user_data);
return self->HandleMessage(connection, raw_message);
}
void ObjectProxy::LogMethodCallFailure(
const std::string_view& interface_name,
const std::string_view& method_name,
const std::string_view& error_name,
const std::string_view& error_message) const {
if (ignore_service_unknown_errors_ &&
(error_name == kErrorServiceUnknown || error_name == kErrorObjectUnknown))
return;
std::ostringstream msg;
msg << "Failed to call method: " << interface_name << "." << method_name
<< ": object_path= " << object_path_.value() << ": " << error_name << ": "
<< error_message;
if (error_name == kErrorObjectUnknown)
LOG(WARNING) << msg.str();
else
LOG(ERROR) << msg.str();
}
void ObjectProxy::OnCallMethod(const std::string& interface_name,
const std::string& method_name,
ResponseCallback response_callback,
Response* response,
ErrorResponse* error_response) {
SCOPED_CRASH_KEY_STRING32("ObjectProxy", "interface_name", interface_name);
SCOPED_CRASH_KEY_STRING32("ObjectProxy", "method_name", method_name);
if (response) {
std::move(response_callback).Run(response);
return;
}
std::string error_name;
std::string error_message;
if (error_response) {
error_name = error_response->GetErrorName();
MessageReader reader(error_response);
reader.PopString(&error_message);
} else {
error_name = "unknown error type";
}
LogMethodCallFailure(interface_name, method_name, error_name, error_message);
std::move(response_callback).Run(nullptr);
}
bool ObjectProxy::AddMatchRuleWithCallback(
const std::string& match_rule,
const std::string& absolute_signal_name,
SignalCallback signal_callback) {
DCHECK(!match_rule.empty());
DCHECK(!absolute_signal_name.empty());
bus_->AssertOnDBusThread();
if (!base::Contains(match_rules_, match_rule)) {
dbus::Error error;
bus_->AddMatch(match_rule, &error);
if (error.IsValid()) {
LOG(ERROR) << "Failed to add match rule \"" << match_rule << "\". Got "
<< error.name() << ": " << error.message();
return false;
} else {
match_rules_.insert(match_rule);
method_table_[absolute_signal_name].push_back(signal_callback);
return true;
}
}
method_table_[absolute_signal_name].push_back(signal_callback);
return true;
}
bool ObjectProxy::AddMatchRuleWithoutCallback(
const std::string& match_rule,
const std::string& absolute_signal_name) {
DCHECK(!match_rule.empty());
DCHECK(!absolute_signal_name.empty());
bus_->AssertOnDBusThread();
if (base::Contains(match_rules_, match_rule)) {
return true;
}
Error error;
bus_->AddMatch(match_rule, &error);
if (error.IsValid()) {
LOG(ERROR) << "Failed to add match rule \"" << match_rule << "\". Got "
<< error.name() << ": " << error.message();
return false;
}
match_rules_.insert(match_rule);
return true;
}
void ObjectProxy::UpdateNameOwnerAndBlock() {
bus_->AssertOnDBusThread();
service_name_owner_ =
bus_->GetServiceOwnerAndBlock(service_name_, Bus::SUPPRESS_ERRORS);
}
DBusHandlerResult ObjectProxy::HandleNameOwnerChanged(
std::unique_ptr<Signal> signal) {
DCHECK(signal);
bus_->AssertOnDBusThread();
if (signal->GetMember() == kNameOwnerChangedMember &&
signal->GetInterface() == kDBusSystemObjectInterface &&
signal->GetSender() == kDBusSystemObjectAddress) {
MessageReader reader(signal.get());
std::string name, old_owner, new_owner;
if (reader.PopString(&name) && reader.PopString(&old_owner) &&
reader.PopString(&new_owner) && name == service_name_) {
service_name_owner_ = new_owner;
bus_->GetOriginTaskRunner()->PostTask(
FROM_HERE, base::BindOnce(&ObjectProxy::RunNameOwnerChangedCallback,
this, old_owner, new_owner));
const bool service_is_available = !service_name_owner_.empty();
if (service_is_available) {
bus_->GetOriginTaskRunner()->PostTask(
FROM_HERE,
base::BindOnce(
&ObjectProxy::RunWaitForServiceToBeAvailableCallbacks, this,
service_is_available));
}
}
}
return DBUS_HANDLER_RESULT_NOT_YET_HANDLED;
}
void ObjectProxy::RunNameOwnerChangedCallback(const std::string& old_owner,
const std::string& new_owner) {
bus_->AssertOnOriginThread();
if (!name_owner_changed_callback_.is_null())
name_owner_changed_callback_.Run(old_owner, new_owner);
}
void ObjectProxy::RunWaitForServiceToBeAvailableCallbacks(
bool service_is_available) {
bus_->AssertOnOriginThread();
std::vector<WaitForServiceToBeAvailableCallback> callbacks;
callbacks.swap(wait_for_service_to_be_available_callbacks_);
for (auto& callback : callbacks) {
std::move(callback).Run(service_is_available);
}
}
}