// Copyright (c) 2012 The Chromium Authors. All rights reserved. // Use of this source code is governed by a BSD-style license that can be // found in the LICENSE file. #include "ipc/ipc_channel_proxy.h" #include #include #include #include "base/bind.h" #include "base/compiler_specific.h" #include "base/containers/contains.h" #include "base/location.h" #include "base/memory/ptr_util.h" #include "base/memory/ref_counted.h" #include "base/task/single_thread_task_runner.h" #include "base/threading/thread_task_runner_handle.h" #include "build/build_config.h" #include "ipc/ipc_channel_factory.h" #include "ipc/ipc_listener.h" #include "ipc/ipc_logging.h" #include "ipc/ipc_message_macros.h" #include "ipc/message_filter.h" #include "ipc/message_filter_router.h" #include "mojo/public/cpp/bindings/pending_associated_receiver.h" namespace IPC { //------------------------------------------------------------------------------ ChannelProxy::Context::Context( Listener* listener, const scoped_refptr& ipc_task_runner, const scoped_refptr& listener_task_runner) : default_listener_task_runner_(listener_task_runner), listener_(listener), ipc_task_runner_(ipc_task_runner), channel_connected_called_(false), message_filter_router_(new MessageFilterRouter()), peer_pid_(base::kNullProcessId) { DCHECK(ipc_task_runner_.get()); // The Listener thread where Messages are handled must be a separate thread // to avoid oversubscribing the IO thread. If you trigger this error, you // need to either: // 1) Create the ChannelProxy on a different thread, or // 2) Just use Channel // We make an exception for NULL listeners. DCHECK(!listener || (ipc_task_runner_.get() != default_listener_task_runner_.get())); } ChannelProxy::Context::~Context() = default; void ChannelProxy::Context::ClearIPCTaskRunner() { ipc_task_runner_.reset(); } void ChannelProxy::Context::CreateChannel( std::unique_ptr factory) { base::AutoLock channel_lock(channel_lifetime_lock_); DCHECK(!channel_); DCHECK_EQ(factory->GetIPCTaskRunner(), ipc_task_runner_); channel_ = factory->BuildChannel(this); Channel::AssociatedInterfaceSupport* support = channel_->GetAssociatedInterfaceSupport(); if (support) { thread_safe_channel_ = support->CreateThreadSafeChannel(); base::AutoLock filter_lock(pending_filters_lock_); for (auto& entry : pending_io_thread_interfaces_) support->AddGenericAssociatedInterface(entry.first, entry.second); pending_io_thread_interfaces_.clear(); } } bool ChannelProxy::Context::TryFilters(const Message& message) { DCHECK(message_filter_router_); #if BUILDFLAG(IPC_MESSAGE_LOG_ENABLED) Logging* logger = Logging::GetInstance(); if (logger->Enabled()) logger->OnPreDispatchMessage(message); #endif if (message_filter_router_->TryFilters(message)) { if (message.dispatch_error()) { GetTaskRunner(message.routing_id()) ->PostTask(FROM_HERE, base::BindOnce(&Context::OnDispatchBadMessage, this, message)); } #if BUILDFLAG(IPC_MESSAGE_LOG_ENABLED) if (logger->Enabled()) logger->OnPostDispatchMessage(message); #endif return true; } return false; } // Called on the IPC::Channel thread void ChannelProxy::Context::PauseChannel() { DCHECK(channel_); channel_->Pause(); } // Called on the IPC::Channel thread void ChannelProxy::Context::UnpauseChannel(bool flush) { DCHECK(channel_); channel_->Unpause(flush); } // Called on the IPC::Channel thread void ChannelProxy::Context::FlushChannel() { DCHECK(channel_); channel_->Flush(); } // Called on the IPC::Channel thread bool ChannelProxy::Context::OnMessageReceived(const Message& message) { // First give a chance to the filters to process this message. if (!TryFilters(message)) OnMessageReceivedNoFilter(message); return true; } // Called on the IPC::Channel thread bool ChannelProxy::Context::OnMessageReceivedNoFilter(const Message& message) { GetTaskRunner(message.routing_id()) ->PostTask(FROM_HERE, base::BindOnce(&Context::OnDispatchMessage, this, message)); return true; } // Called on the IPC::Channel thread void ChannelProxy::Context::OnChannelConnected(int32_t peer_pid) { // We cache off the peer_pid so it can be safely accessed from both threads. { base::AutoLock l(peer_pid_lock_); peer_pid_ = peer_pid; } // Add any pending filters. This avoids a race condition where someone // creates a ChannelProxy, calls AddFilter, and then right after starts the // peer process. The IO thread could receive a message before the task to add // the filter is run on the IO thread. OnAddFilter(); // See above comment about using default_listener_task_runner_ here. default_listener_task_runner_->PostTask( FROM_HERE, base::BindOnce(&Context::OnDispatchConnected, this)); } // Called on the IPC::Channel thread void ChannelProxy::Context::OnChannelError() { for (size_t i = 0; i < filters_.size(); ++i) filters_[i]->OnChannelError(); // See above comment about using default_listener_task_runner_ here. default_listener_task_runner_->PostTask( FROM_HERE, base::BindOnce(&Context::OnDispatchError, this)); } // Called on the IPC::Channel thread void ChannelProxy::Context::OnAssociatedInterfaceRequest( const std::string& interface_name, mojo::ScopedInterfaceEndpointHandle handle) { default_listener_task_runner_->PostTask( FROM_HERE, base::BindOnce(&Context::OnDispatchAssociatedInterfaceRequest, this, interface_name, std::move(handle))); } // Called on the IPC::Channel thread void ChannelProxy::Context::OnChannelOpened() { DCHECK(channel_); // Assume a reference to ourselves on behalf of this thread. This reference // will be released when we are closed. AddRef(); if (!channel_->Connect()) { OnChannelError(); return; } for (size_t i = 0; i < filters_.size(); ++i) filters_[i]->OnFilterAdded(channel_.get()); } // Called on the IPC::Channel thread void ChannelProxy::Context::OnChannelClosed() { // It's okay for IPC::ChannelProxy::Close to be called more than once, which // would result in this branch being taken. if (!channel_) return; for (auto& filter : pending_filters_) { filter->OnChannelClosing(); filter->OnFilterRemoved(); } for (auto& filter : filters_) { filter->OnChannelClosing(); filter->OnFilterRemoved(); } // We don't need the filters anymore. message_filter_router_->Clear(); filters_.clear(); // We don't need the lock, because at this point, the listener thread can't // access it any more. pending_filters_.clear(); ClearChannel(); // Balance with the reference taken during startup. This may result in // self-destruction. Release(); } void ChannelProxy::Context::Clear() { listener_ = nullptr; } // Called on the IPC::Channel thread void ChannelProxy::Context::OnSendMessage(std::unique_ptr message) { if (quota_checker_) quota_checker_->AfterMessagesDequeued(1); if (!channel_) { OnChannelClosed(); return; } if (!channel_->Send(message.release())) OnChannelError(); } // Called on the IPC::Channel thread void ChannelProxy::Context::OnAddFilter() { // Our OnChannelConnected method has not yet been called, so we can't be // sure that channel_ is valid yet. When OnChannelConnected *is* called, // it invokes OnAddFilter, so any pending filter(s) will be added at that // time. // No lock necessary for |peer_pid_| because it is only modified on this // thread. if (peer_pid_ == base::kNullProcessId) return; std::vector > new_filters; { base::AutoLock auto_lock(pending_filters_lock_); new_filters.swap(pending_filters_); } for (size_t i = 0; i < new_filters.size(); ++i) { filters_.push_back(new_filters[i]); message_filter_router_->AddFilter(new_filters[i].get()); // The channel has already been created and connected, so we need to // inform the filters right now. new_filters[i]->OnFilterAdded(channel_.get()); new_filters[i]->OnChannelConnected(peer_pid_); } } // Called on the IPC::Channel thread void ChannelProxy::Context::OnRemoveFilter(MessageFilter* filter) { // No lock necessary for |peer_pid_| because it is only modified on this // thread. if (peer_pid_ == base::kNullProcessId) { // The channel is not yet connected, so any filters are still pending. base::AutoLock auto_lock(pending_filters_lock_); for (size_t i = 0; i < pending_filters_.size(); ++i) { if (pending_filters_[i].get() == filter) { filter->OnFilterRemoved(); pending_filters_.erase(pending_filters_.begin() + i); return; } } return; } if (!channel_) return; // The filters have already been deleted. message_filter_router_->RemoveFilter(filter); for (size_t i = 0; i < filters_.size(); ++i) { if (filters_[i].get() == filter) { filter->OnFilterRemoved(); filters_.erase(filters_.begin() + i); return; } } NOTREACHED() << "filter to be removed not found"; } // Called on the listener's thread void ChannelProxy::Context::AddFilter(MessageFilter* filter) { base::AutoLock auto_lock(pending_filters_lock_); pending_filters_.push_back(base::WrapRefCounted(filter)); ipc_task_runner_->PostTask(FROM_HERE, base::BindOnce(&Context::OnAddFilter, this)); } // Called on the listener's thread void ChannelProxy::Context::OnDispatchMessage(const Message& message) { if (!listener_) return; OnDispatchConnected(); #if BUILDFLAG(IPC_MESSAGE_LOG_ENABLED) Logging* logger = Logging::GetInstance(); if (message.type() == IPC_LOGGING_ID) { logger->OnReceivedLoggingMessage(message); return; } if (logger->Enabled()) logger->OnPreDispatchMessage(message); #endif listener_->OnMessageReceived(message); if (message.dispatch_error()) listener_->OnBadMessageReceived(message); #if BUILDFLAG(IPC_MESSAGE_LOG_ENABLED) if (logger->Enabled()) logger->OnPostDispatchMessage(message); #endif } // Called on the listener's thread. void ChannelProxy::Context::AddListenerTaskRunner( int32_t routing_id, scoped_refptr task_runner) { DCHECK(default_listener_task_runner_->BelongsToCurrentThread()); DCHECK(task_runner); base::AutoLock lock(listener_thread_task_runners_lock_); if (!base::Contains(listener_thread_task_runners_, routing_id)) listener_thread_task_runners_.insert({routing_id, std::move(task_runner)}); } // Called on the listener's thread. void ChannelProxy::Context::RemoveListenerTaskRunner(int32_t routing_id) { DCHECK(default_listener_task_runner_->BelongsToCurrentThread()); base::AutoLock lock(listener_thread_task_runners_lock_); listener_thread_task_runners_.erase(routing_id); } // Called on the IPC::Channel thread. scoped_refptr ChannelProxy::Context::GetTaskRunner(int32_t routing_id) { DCHECK(ipc_task_runner_->BelongsToCurrentThread()); if (routing_id == MSG_ROUTING_NONE) return default_listener_task_runner_; base::AutoLock lock(listener_thread_task_runners_lock_); auto task_runner = listener_thread_task_runners_.find(routing_id); if (task_runner == listener_thread_task_runners_.end()) return default_listener_task_runner_; DCHECK(task_runner->second); return task_runner->second; } // Called on the listener's thread void ChannelProxy::Context::OnDispatchConnected() { if (channel_connected_called_) return; base::ProcessId peer_pid; { base::AutoLock l(peer_pid_lock_); peer_pid = peer_pid_; } channel_connected_called_ = true; if (listener_) listener_->OnChannelConnected(peer_pid); } // Called on the listener's thread void ChannelProxy::Context::OnDispatchError() { if (listener_) listener_->OnChannelError(); } // Called on the listener's thread void ChannelProxy::Context::OnDispatchBadMessage(const Message& message) { if (listener_) listener_->OnBadMessageReceived(message); } // Called on the listener's thread void ChannelProxy::Context::OnDispatchAssociatedInterfaceRequest( const std::string& interface_name, mojo::ScopedInterfaceEndpointHandle handle) { if (listener_) listener_->OnAssociatedInterfaceRequest(interface_name, std::move(handle)); } void ChannelProxy::Context::ClearChannel() { base::AutoLock l(channel_lifetime_lock_); channel_.reset(); } void ChannelProxy::Context::AddGenericAssociatedInterfaceForIOThread( const std::string& name, const GenericAssociatedInterfaceFactory& factory) { base::AutoLock channel_lock(channel_lifetime_lock_); if (!channel_) { base::AutoLock filter_lock(pending_filters_lock_); pending_io_thread_interfaces_.emplace_back(name, factory); return; } Channel::AssociatedInterfaceSupport* support = channel_->GetAssociatedInterfaceSupport(); if (support) support->AddGenericAssociatedInterface(name, factory); } void ChannelProxy::Context::Send(Message* message) { if (quota_checker_) quota_checker_->BeforeMessagesEnqueued(1); ipc_task_runner()->PostTask( FROM_HERE, base::BindOnce(&ChannelProxy::Context::OnSendMessage, this, base::WrapUnique(message))); } //----------------------------------------------------------------------------- // static std::unique_ptr ChannelProxy::Create( const IPC::ChannelHandle& channel_handle, Channel::Mode mode, Listener* listener, const scoped_refptr& ipc_task_runner, const scoped_refptr& listener_task_runner) { std::unique_ptr channel( new ChannelProxy(listener, ipc_task_runner, listener_task_runner)); channel->Init(channel_handle, mode, true); return channel; } // static std::unique_ptr ChannelProxy::Create( std::unique_ptr factory, Listener* listener, const scoped_refptr& ipc_task_runner, const scoped_refptr& listener_task_runner) { std::unique_ptr channel( new ChannelProxy(listener, ipc_task_runner, listener_task_runner)); channel->Init(std::move(factory), true); return channel; } ChannelProxy::ChannelProxy(Context* context) : context_(context), did_init_(false) { #if defined(ENABLE_IPC_FUZZER) outgoing_message_filter_ = nullptr; #endif } ChannelProxy::ChannelProxy( Listener* listener, const scoped_refptr& ipc_task_runner, const scoped_refptr& listener_task_runner) : context_(new Context(listener, ipc_task_runner, listener_task_runner)), did_init_(false) { #if defined(ENABLE_IPC_FUZZER) outgoing_message_filter_ = nullptr; #endif } ChannelProxy::~ChannelProxy() { DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_); Close(); } void ChannelProxy::Init(const IPC::ChannelHandle& channel_handle, Channel::Mode mode, bool create_pipe_now) { #if BUILDFLAG(IS_POSIX) || BUILDFLAG(IS_FUCHSIA) // When we are creating a server on POSIX, we need its file descriptor // to be created immediately so that it can be accessed and passed // to other processes. Forcing it to be created immediately avoids // race conditions that may otherwise arise. if (mode & Channel::MODE_SERVER_FLAG) { create_pipe_now = true; } #endif // BUILDFLAG(IS_POSIX) || BUILDFLAG(IS_FUCHSIA) Init( ChannelFactory::Create(channel_handle, mode, context_->ipc_task_runner()), create_pipe_now); } void ChannelProxy::Init(std::unique_ptr factory, bool create_pipe_now) { DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_); DCHECK(!did_init_); DCHECK(!context_->quota_checker_); context_->quota_checker_ = factory->GetQuotaChecker(); if (create_pipe_now) { // Create the channel immediately. This effectively sets up the // low-level pipe so that the client can connect. Without creating // the pipe immediately, it is possible for a listener to attempt // to connect and get an error since the pipe doesn't exist yet. context_->CreateChannel(std::move(factory)); } else { context_->ipc_task_runner()->PostTask( FROM_HERE, base::BindOnce(&Context::CreateChannel, context_, std::move(factory))); } // complete initialization on the background thread context_->ipc_task_runner()->PostTask( FROM_HERE, base::BindOnce(&Context::OnChannelOpened, context_)); did_init_ = true; OnChannelInit(); } void ChannelProxy::Pause() { context_->ipc_task_runner()->PostTask( FROM_HERE, base::BindOnce(&Context::PauseChannel, context_)); } void ChannelProxy::Unpause(bool flush) { context_->ipc_task_runner()->PostTask( FROM_HERE, base::BindOnce(&Context::UnpauseChannel, context_, flush)); } void ChannelProxy::Flush() { context_->ipc_task_runner()->PostTask( FROM_HERE, base::BindOnce(&Context::FlushChannel, context_)); } void ChannelProxy::Close() { DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_); // Clear the backpointer to the listener so that any pending calls to // Context::OnDispatchMessage or OnDispatchError will be ignored. It is // possible that the channel could be closed while it is receiving messages! context_->Clear(); if (context_->ipc_task_runner()) { context_->ipc_task_runner()->PostTask( FROM_HERE, base::BindOnce(&Context::OnChannelClosed, context_)); } } bool ChannelProxy::Send(Message* message) { DCHECK(!message->is_sync()) << "Need to use IPC::SyncChannel"; SendInternal(message); return true; } void ChannelProxy::SendInternal(Message* message) { DCHECK(did_init_); // TODO(alexeypa): add DCHECK(CalledOnValidThread()) here. Currently there are // tests that call Send() from a wrong thread. See http://crbug.com/163523. #ifdef ENABLE_IPC_FUZZER // In IPC fuzzing builds, it is possible to define a filter to apply to // outgoing messages. It will either rewrite the message and return a new // one, freeing the original, or return the message unchanged. if (outgoing_message_filter()) message = outgoing_message_filter()->Rewrite(message); #endif #if BUILDFLAG(IPC_MESSAGE_LOG_ENABLED) Logging::GetInstance()->OnSendMessage(message); #endif context_->Send(message); } void ChannelProxy::AddFilter(MessageFilter* filter) { DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_); context_->AddFilter(filter); } void ChannelProxy::RemoveFilter(MessageFilter* filter) { DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_); context_->ipc_task_runner()->PostTask( FROM_HERE, base::BindOnce(&Context::OnRemoveFilter, context_, base::RetainedRef(filter))); } void ChannelProxy::AddGenericAssociatedInterfaceForIOThread( const std::string& name, const GenericAssociatedInterfaceFactory& factory) { context()->AddGenericAssociatedInterfaceForIOThread(name, factory); } void ChannelProxy::GetRemoteAssociatedInterface( mojo::GenericPendingAssociatedReceiver receiver) { DCHECK(did_init_); context()->thread_safe_channel().GetAssociatedInterface(std::move(receiver)); } void ChannelProxy::ClearIPCTaskRunner() { DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_); context()->ClearIPCTaskRunner(); } void ChannelProxy::OnChannelInit() { } //----------------------------------------------------------------------------- } // namespace IPC