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