ipc/ipc_channel_proxy.cc

   1 // Copyright (c) 2012 The Chromium Authors. All rights reserved.
   2 // Use of this source code is governed by a BSD-style license that can be
   3 // found in the LICENSE file.
   4
   5 #include "base/bind.h"
   6 #include "base/compiler_specific.h"
   7 #include "base/debug/trace_event.h"
   8 #include "base/location.h"
   9 #include "base/memory/ref_counted.h"
  10 #include "base/memory/scoped_ptr.h"
  11 #include "base/single_thread_task_runner.h"
  12 #include "base/thread_task_runner_handle.h"
  13 #include "ipc/ipc_channel_proxy.h"
  14 #include "ipc/ipc_listener.h"
  15 #include "ipc/ipc_logging.h"
  16 #include "ipc/ipc_message_macros.h"
  17 #include "ipc/ipc_message_start.h"
  18 #include "ipc/ipc_message_utils.h"
  19
  20 namespace IPC {
  21
  22 //------------------------------------------------------------------------------
  23
  24 class ChannelProxy::Context::MessageFilterRouter {
  25  public:
  26   typedef std::vector<MessageFilter*> MessageFilters;
  27
  28   MessageFilterRouter() {}
  29   ~MessageFilterRouter() {}
  30
  31   void AddFilter(MessageFilter* filter) {
  32     // Determine if the filter should be applied to all messages, or only
  33     // messages of a certain class.
  34     std::vector<uint32> supported_message_classes;
  35     if (filter->GetSupportedMessageClasses(&supported_message_classes)) {
  36       DCHECK(!supported_message_classes.empty());
  37       for (size_t i = 0; i < supported_message_classes.size(); ++i) {
  38         DCHECK(ValidMessageClass(supported_message_classes[i]));
  39         message_class_filters_[supported_message_classes[i]].push_back(filter);
  40       }
  41     } else {
  42       global_filters_.push_back(filter);
  43     }
  44   }
  45
  46   void RemoveFilter(MessageFilter* filter) {
  47     if (RemoveFilter(global_filters_, filter))
  48       return;
  49
  50     for (size_t i = 0; i < arraysize(message_class_filters_); ++i)
  51       RemoveFilter(message_class_filters_[i], filter);
  52   }
  53
  54   bool TryFilters(const Message& message) {
  55     if (TryFilters(global_filters_, message))
  56       return true;
  57
  58     const int message_class = IPC_MESSAGE_CLASS(message);
  59     if (!ValidMessageClass(message_class))
  60       return false;
  61
  62     return TryFilters(message_class_filters_[message_class], message);
  63   }
  64
  65   void Clear() {
  66     global_filters_.clear();
  67     for (size_t i = 0; i < arraysize(message_class_filters_); ++i)
  68       message_class_filters_[i].clear();
  69   }
  70
  71  private:
  72   static bool TryFilters(MessageFilters& filters, const IPC::Message& message) {
  73     for (size_t i = 0; i < filters.size(); ++i) {
  74       if (filters[i]->OnMessageReceived(message)) {
  75         return true;
  76       }
  77     }
  78     return false;
  79   }
  80
  81   static bool RemoveFilter(MessageFilters& filters, MessageFilter* filter) {
  82     MessageFilters::iterator it =
  83         std::find(filters.begin(), filters.end(), filter);
  84     if (it == filters.end())
  85       return false;
  86
  87     filters.erase(it);
  88     return true;
  89   }
  90
  91   static bool ValidMessageClass(int message_class) {
  92     return message_class >= 0 && message_class < LastIPCMsgStart;
  93   }
  94
  95   // List of global and selective filters; a given filter will exist in either
  96   // |message_global_filters_| OR |message_class_filters_|, but not both.
  97   // Note that |message_global_filters_| will be given first offering of any
  98   // given message.  It's the filter implementer and installer's
  99   // responsibility to ensure that a filter is either global or selective to
 100   // ensure proper message filtering order.
 101   MessageFilters global_filters_;
 102   MessageFilters message_class_filters_[LastIPCMsgStart];
 103 };
 104
 105 //------------------------------------------------------------------------------
 106
 107 ChannelProxy::MessageFilter::MessageFilter() {}
 108
 109 void ChannelProxy::MessageFilter::OnFilterAdded(Channel* channel) {}
 110
 111 void ChannelProxy::MessageFilter::OnFilterRemoved() {}
 112
 113 void ChannelProxy::MessageFilter::OnChannelConnected(int32 peer_pid) {}
 114
 115 void ChannelProxy::MessageFilter::OnChannelError() {}
 116
 117 void ChannelProxy::MessageFilter::OnChannelClosing() {}
 118
 119 bool ChannelProxy::MessageFilter::OnMessageReceived(const Message& message) {
 120   return false;
 121 }
 122
 123 bool ChannelProxy::MessageFilter::GetSupportedMessageClasses(
 124     std::vector<uint32>* /*supported_message_classes*/) const {
 125   return false;
 126 }
 127
 128 ChannelProxy::MessageFilter::~MessageFilter() {}
 129
 130 //------------------------------------------------------------------------------
 131
 132 ChannelProxy::Context::Context(Listener* listener,
 133                                base::SingleThreadTaskRunner* ipc_task_runner)
 134     : listener_task_runner_(base::ThreadTaskRunnerHandle::Get()),
 135       listener_(listener),
 136       ipc_task_runner_(ipc_task_runner),
 137       channel_connected_called_(false),
 138       message_filter_router_(new MessageFilterRouter()),
 139       peer_pid_(base::kNullProcessId) {
 140   DCHECK(ipc_task_runner_.get());
 141 }
 142
 143 ChannelProxy::Context::~Context() {
 144 }
 145
 146 void ChannelProxy::Context::ClearIPCTaskRunner() {
 147   ipc_task_runner_ = NULL;
 148 }
 149
 150 void ChannelProxy::Context::CreateChannel(const IPC::ChannelHandle& handle,
 151                                           const Channel::Mode& mode) {
 152   DCHECK(channel_.get() == NULL);
 153   channel_id_ = handle.name;
 154   channel_.reset(new Channel(handle, mode, this));
 155 }
 156
 157 bool ChannelProxy::Context::TryFilters(const Message& message) {
 158   DCHECK(message_filter_router_);
 159 #ifdef IPC_MESSAGE_LOG_ENABLED
 160   Logging* logger = Logging::GetInstance();
 161   if (logger->Enabled())
 162     logger->OnPreDispatchMessage(message);
 163 #endif
 164
 165   if (message_filter_router_->TryFilters(message)) {
 166 #ifdef IPC_MESSAGE_LOG_ENABLED
 167     if (logger->Enabled())
 168       logger->OnPostDispatchMessage(message, channel_id_);
 169 #endif
 170     return true;
 171   }
 172   return false;
 173 }
 174
 175 // Called on the IPC::Channel thread
 176 bool ChannelProxy::Context::OnMessageReceived(const Message& message) {
 177   // First give a chance to the filters to process this message.
 178   if (!TryFilters(message))
 179     OnMessageReceivedNoFilter(message);
 180   return true;
 181 }
 182
 183 // Called on the IPC::Channel thread
 184 bool ChannelProxy::Context::OnMessageReceivedNoFilter(const Message& message) {
 185   listener_task_runner_->PostTask(
 186       FROM_HERE, base::Bind(&Context::OnDispatchMessage, this, message));
 187   return true;
 188 }
 189
 190 // Called on the IPC::Channel thread
 191 void ChannelProxy::Context::OnChannelConnected(int32 peer_pid) {
 192   // Add any pending filters.  This avoids a race condition where someone
 193   // creates a ChannelProxy, calls AddFilter, and then right after starts the
 194   // peer process.  The IO thread could receive a message before the task to add
 195   // the filter is run on the IO thread.
 196   OnAddFilter();
 197
 198   // We cache off the peer_pid so it can be safely accessed from both threads.
 199   peer_pid_ = channel_->peer_pid();
 200   for (size_t i = 0; i < filters_.size(); ++i)
 201     filters_[i]->OnChannelConnected(peer_pid);
 202
 203   // See above comment about using listener_task_runner_ here.
 204   listener_task_runner_->PostTask(
 205       FROM_HERE, base::Bind(&Context::OnDispatchConnected, this));
 206 }
 207
 208 // Called on the IPC::Channel thread
 209 void ChannelProxy::Context::OnChannelError() {
 210   for (size_t i = 0; i < filters_.size(); ++i)
 211     filters_[i]->OnChannelError();
 212
 213   // See above comment about using listener_task_runner_ here.
 214   listener_task_runner_->PostTask(
 215       FROM_HERE, base::Bind(&Context::OnDispatchError, this));
 216 }
 217
 218 // Called on the IPC::Channel thread
 219 void ChannelProxy::Context::OnChannelOpened() {
 220   DCHECK(channel_ != NULL);
 221
 222   // Assume a reference to ourselves on behalf of this thread.  This reference
 223   // will be released when we are closed.
 224   AddRef();
 225
 226   if (!channel_->Connect()) {
 227     OnChannelError();
 228     return;
 229   }
 230
 231   for (size_t i = 0; i < filters_.size(); ++i)
 232     filters_[i]->OnFilterAdded(channel_.get());
 233 }
 234
 235 // Called on the IPC::Channel thread
 236 void ChannelProxy::Context::OnChannelClosed() {
 237   // It's okay for IPC::ChannelProxy::Close to be called more than once, which
 238   // would result in this branch being taken.
 239   if (!channel_.get())
 240     return;
 241
 242   for (size_t i = 0; i < filters_.size(); ++i) {
 243     filters_[i]->OnChannelClosing();
 244     filters_[i]->OnFilterRemoved();
 245   }
 246
 247   // We don't need the filters anymore.
 248   filters_.clear();
 249   message_filter_router_->Clear();
 250
 251   channel_.reset();
 252
 253   // Balance with the reference taken during startup.  This may result in
 254   // self-destruction.
 255   Release();
 256 }
 257
 258 void ChannelProxy::Context::Clear() {
 259   listener_ = NULL;
 260 }
 261
 262 // Called on the IPC::Channel thread
 263 void ChannelProxy::Context::OnSendMessage(scoped_ptr<Message> message) {
 264   if (!channel_.get()) {
 265     OnChannelClosed();
 266     return;
 267   }
 268   if (!channel_->Send(message.release()))
 269     OnChannelError();
 270 }
 271
 272 // Called on the IPC::Channel thread
 273 void ChannelProxy::Context::OnAddFilter() {
 274   std::vector<scoped_refptr<MessageFilter> > new_filters;
 275   {
 276     base::AutoLock auto_lock(pending_filters_lock_);
 277     new_filters.swap(pending_filters_);
 278   }
 279
 280   for (size_t i = 0; i < new_filters.size(); ++i) {
 281     filters_.push_back(new_filters[i]);
 282
 283     message_filter_router_->AddFilter(new_filters[i].get());
 284
 285     // If the channel has already been created, then we need to send this
 286     // message so that the filter gets access to the Channel.
 287     if (channel_.get())
 288       new_filters[i]->OnFilterAdded(channel_.get());
 289     // Ditto for if the channel has been connected.
 290     if (peer_pid_)
 291       new_filters[i]->OnChannelConnected(peer_pid_);
 292   }
 293 }
 294
 295 // Called on the IPC::Channel thread
 296 void ChannelProxy::Context::OnRemoveFilter(MessageFilter* filter) {
 297   if (!channel_.get())
 298     return;  // The filters have already been deleted.
 299
 300   message_filter_router_->RemoveFilter(filter);
 301
 302   for (size_t i = 0; i < filters_.size(); ++i) {
 303     if (filters_[i].get() == filter) {
 304       filter->OnFilterRemoved();
 305       filters_.erase(filters_.begin() + i);
 306       return;
 307     }
 308   }
 309
 310   NOTREACHED() << "filter to be removed not found";
 311 }
 312
 313 // Called on the listener's thread
 314 void ChannelProxy::Context::AddFilter(MessageFilter* filter) {
 315   base::AutoLock auto_lock(pending_filters_lock_);
 316   pending_filters_.push_back(make_scoped_refptr(filter));
 317   ipc_task_runner_->PostTask(
 318       FROM_HERE, base::Bind(&Context::OnAddFilter, this));
 319 }
 320
 321 // Called on the listener's thread
 322 void ChannelProxy::Context::OnDispatchMessage(const Message& message) {
 323 #ifdef IPC_MESSAGE_LOG_ENABLED
 324   Logging* logger = Logging::GetInstance();
 325   std::string name;
 326   logger->GetMessageText(message.type(), &name, &message, NULL);
 327   TRACE_EVENT1("toplevel", "ChannelProxy::Context::OnDispatchMessage",
 328                "name", name);
 329 #else
 330   TRACE_EVENT2("toplevel", "ChannelProxy::Context::OnDispatchMessage",
 331                "class", IPC_MESSAGE_ID_CLASS(message.type()),
 332                "line", IPC_MESSAGE_ID_LINE(message.type()));
 333 #endif
 334
 335   if (!listener_)
 336     return;
 337
 338   OnDispatchConnected();
 339
 340 #ifdef IPC_MESSAGE_LOG_ENABLED
 341   if (message.type() == IPC_LOGGING_ID) {
 342     logger->OnReceivedLoggingMessage(message);
 343     return;
 344   }
 345
 346   if (logger->Enabled())
 347     logger->OnPreDispatchMessage(message);
 348 #endif
 349
 350   listener_->OnMessageReceived(message);
 351
 352 #ifdef IPC_MESSAGE_LOG_ENABLED
 353   if (logger->Enabled())
 354     logger->OnPostDispatchMessage(message, channel_id_);
 355 #endif
 356 }
 357
 358 // Called on the listener's thread
 359 void ChannelProxy::Context::OnDispatchConnected() {
 360   if (channel_connected_called_)
 361     return;
 362
 363   channel_connected_called_ = true;
 364   if (listener_)
 365     listener_->OnChannelConnected(peer_pid_);
 366 }
 367
 368 // Called on the listener's thread
 369 void ChannelProxy::Context::OnDispatchError() {
 370   if (listener_)
 371     listener_->OnChannelError();
 372 }
 373
 374 //-----------------------------------------------------------------------------
 375
 376 ChannelProxy::ChannelProxy(const IPC::ChannelHandle& channel_handle,
 377                            Channel::Mode mode,
 378                            Listener* listener,
 379                            base::SingleThreadTaskRunner* ipc_task_runner)
 380     : context_(new Context(listener, ipc_task_runner)),
 381       did_init_(false) {
 382   Init(channel_handle, mode, true);
 383 }
 384
 385 ChannelProxy::ChannelProxy(Context* context)
 386     : context_(context),
 387       did_init_(false) {
 388 }
 389
 390 ChannelProxy::~ChannelProxy() {
 391   DCHECK(CalledOnValidThread());
 392
 393   Close();
 394 }
 395
 396 void ChannelProxy::Init(const IPC::ChannelHandle& channel_handle,
 397                         Channel::Mode mode,
 398                         bool create_pipe_now) {
 399   DCHECK(CalledOnValidThread());
 400   DCHECK(!did_init_);
 401 #if defined(OS_POSIX)
 402   // When we are creating a server on POSIX, we need its file descriptor
 403   // to be created immediately so that it can be accessed and passed
 404   // to other processes. Forcing it to be created immediately avoids
 405   // race conditions that may otherwise arise.
 406   if (mode & Channel::MODE_SERVER_FLAG) {
 407     create_pipe_now = true;
 408   }
 409 #endif  // defined(OS_POSIX)
 410
 411   if (create_pipe_now) {
 412     // Create the channel immediately.  This effectively sets up the
 413     // low-level pipe so that the client can connect.  Without creating
 414     // the pipe immediately, it is possible for a listener to attempt
 415     // to connect and get an error since the pipe doesn't exist yet.
 416     context_->CreateChannel(channel_handle, mode);
 417   } else {
 418     context_->ipc_task_runner()->PostTask(
 419         FROM_HERE, base::Bind(&Context::CreateChannel, context_.get(),
 420                               channel_handle, mode));
 421   }
 422
 423   // complete initialization on the background thread
 424   context_->ipc_task_runner()->PostTask(
 425       FROM_HERE, base::Bind(&Context::OnChannelOpened, context_.get()));
 426
 427   did_init_ = true;
 428 }
 429
 430 void ChannelProxy::Close() {
 431   DCHECK(CalledOnValidThread());
 432
 433   // Clear the backpointer to the listener so that any pending calls to
 434   // Context::OnDispatchMessage or OnDispatchError will be ignored.  It is
 435   // possible that the channel could be closed while it is receiving messages!
 436   context_->Clear();
 437
 438   if (context_->ipc_task_runner()) {
 439     context_->ipc_task_runner()->PostTask(
 440         FROM_HERE, base::Bind(&Context::OnChannelClosed, context_.get()));
 441   }
 442 }
 443
 444 bool ChannelProxy::Send(Message* message) {
 445   DCHECK(did_init_);
 446
 447   // TODO(alexeypa): add DCHECK(CalledOnValidThread()) here. Currently there are
 448   // tests that call Send() from a wrong thread. See http://crbug.com/163523.
 449
 450 #ifdef IPC_MESSAGE_LOG_ENABLED
 451   Logging::GetInstance()->OnSendMessage(message, context_->channel_id());
 452 #endif
 453
 454   context_->ipc_task_runner()->PostTask(
 455       FROM_HERE,
 456       base::Bind(&ChannelProxy::Context::OnSendMessage,
 457                  context_, base::Passed(scoped_ptr<Message>(message))));
 458   return true;
 459 }
 460
 461 void ChannelProxy::AddFilter(MessageFilter* filter) {
 462   DCHECK(CalledOnValidThread());
 463
 464   context_->AddFilter(filter);
 465 }
 466
 467 void ChannelProxy::RemoveFilter(MessageFilter* filter) {
 468   DCHECK(CalledOnValidThread());
 469
 470   context_->ipc_task_runner()->PostTask(
 471       FROM_HERE, base::Bind(&Context::OnRemoveFilter, context_.get(),
 472                             make_scoped_refptr(filter)));
 473 }
 474
 475 void ChannelProxy::ClearIPCTaskRunner() {
 476   DCHECK(CalledOnValidThread());
 477
 478   context()->ClearIPCTaskRunner();
 479 }
 480
 481 #if defined(OS_POSIX) && !defined(OS_NACL)
 482 // See the TODO regarding lazy initialization of the channel in
 483 // ChannelProxy::Init().
 484 int ChannelProxy::GetClientFileDescriptor() {
 485   DCHECK(CalledOnValidThread());
 486
 487   Channel* channel = context_.get()->channel_.get();
 488   // Channel must have been created first.
 489   DCHECK(channel) << context_.get()->channel_id_;
 490   return channel->GetClientFileDescriptor();
 491 }
 492
 493 int ChannelProxy::TakeClientFileDescriptor() {
 494   DCHECK(CalledOnValidThread());
 495
 496   Channel* channel = context_.get()->channel_.get();
 497   // Channel must have been created first.
 498   DCHECK(channel) << context_.get()->channel_id_;
 499   return channel->TakeClientFileDescriptor();
 500 }
 501
 502 bool ChannelProxy::GetPeerEuid(uid_t* peer_euid) const {
 503   DCHECK(CalledOnValidThread());
 504
 505   Channel* channel = context_.get()->channel_.get();
 506   // Channel must have been created first.
 507   DCHECK(channel) << context_.get()->channel_id_;
 508   return channel->GetPeerEuid(peer_euid);
 509 }
 510 #endif
 511
 512 //-----------------------------------------------------------------------------
 513
 514 }  // namespace IPC