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_.get() == NULL);
 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   // Add any pending filters.  This avoids a race condition where someone
 200   // creates a ChannelProxy, calls AddFilter, and then right after starts the
 201   // peer process.  The IO thread could receive a message before the task to add
 202   // the filter is run on the IO thread.
 203   OnAddFilter();
 204
 205   // We cache off the peer_pid so it can be safely accessed from both threads.
 206   peer_pid_ = channel_->peer_pid();
 207   for (size_t i = 0; i < filters_.size(); ++i)
 208     filters_[i]->OnChannelConnected(peer_pid);
 209
 210   // See above comment about using listener_task_runner_ here.
 211   listener_task_runner_->PostTask(
 212       FROM_HERE, base::Bind(&Context::OnDispatchConnected, this));
 213 }
 214
 215 // Called on the IPC::Channel thread
 216 void ChannelProxy::Context::OnChannelError() {
 217   for (size_t i = 0; i < filters_.size(); ++i)
 218     filters_[i]->OnChannelError();
 219
 220   // See above comment about using listener_task_runner_ here.
 221   listener_task_runner_->PostTask(
 222       FROM_HERE, base::Bind(&Context::OnDispatchError, this));
 223 }
 224
 225 // Called on the IPC::Channel thread
 226 void ChannelProxy::Context::OnChannelOpened() {
 227   DCHECK(channel_ != NULL);
 228
 229   // Assume a reference to ourselves on behalf of this thread.  This reference
 230   // will be released when we are closed.
 231   AddRef();
 232
 233   if (!channel_->Connect()) {
 234     OnChannelError();
 235     return;
 236   }
 237
 238   for (size_t i = 0; i < filters_.size(); ++i)
 239     filters_[i]->OnFilterAdded(channel_.get());
 240 }
 241
 242 // Called on the IPC::Channel thread
 243 void ChannelProxy::Context::OnChannelClosed() {
 244   // It's okay for IPC::ChannelProxy::Close to be called more than once, which
 245   // would result in this branch being taken.
 246   if (!channel_.get())
 247     return;
 248
 249   for (size_t i = 0; i < filters_.size(); ++i) {
 250     filters_[i]->OnChannelClosing();
 251     filters_[i]->OnFilterRemoved();
 252   }
 253
 254   // We don't need the filters anymore.
 255   message_filter_router_->Clear();
 256   filters_.clear();
 257
 258   channel_.reset();
 259
 260   // Balance with the reference taken during startup.  This may result in
 261   // self-destruction.
 262   Release();
 263 }
 264
 265 void ChannelProxy::Context::Clear() {
 266   listener_ = NULL;
 267 }
 268
 269 // Called on the IPC::Channel thread
 270 void ChannelProxy::Context::OnSendMessage(scoped_ptr<Message> message) {
 271   if (!channel_.get()) {
 272     OnChannelClosed();
 273     return;
 274   }
 275   if (!channel_->Send(message.release()))
 276     OnChannelError();
 277 }
 278
 279 // Called on the IPC::Channel thread
 280 void ChannelProxy::Context::OnAddFilter() {
 281   std::vector<scoped_refptr<MessageFilter> > new_filters;
 282   {
 283     base::AutoLock auto_lock(pending_filters_lock_);
 284     new_filters.swap(pending_filters_);
 285   }
 286
 287   for (size_t i = 0; i < new_filters.size(); ++i) {
 288     filters_.push_back(new_filters[i]);
 289
 290     message_filter_router_->AddFilter(new_filters[i].get());
 291
 292     // If the channel has already been created, then we need to send this
 293     // message so that the filter gets access to the Channel.
 294     if (channel_.get())
 295       new_filters[i]->OnFilterAdded(channel_.get());
 296     // Ditto for if the channel has been connected.
 297     if (peer_pid_)
 298       new_filters[i]->OnChannelConnected(peer_pid_);
 299   }
 300 }
 301
 302 // Called on the IPC::Channel thread
 303 void ChannelProxy::Context::OnRemoveFilter(MessageFilter* filter) {
 304   if (!channel_.get())
 305     return;  // The filters have already been deleted.
 306
 307   message_filter_router_->RemoveFilter(filter);
 308
 309   for (size_t i = 0; i < filters_.size(); ++i) {
 310     if (filters_[i].get() == filter) {
 311       filter->OnFilterRemoved();
 312       filters_.erase(filters_.begin() + i);
 313       return;
 314     }
 315   }
 316
 317   NOTREACHED() << "filter to be removed not found";
 318 }
 319
 320 // Called on the listener's thread
 321 void ChannelProxy::Context::AddFilter(MessageFilter* filter) {
 322   base::AutoLock auto_lock(pending_filters_lock_);
 323   pending_filters_.push_back(make_scoped_refptr(filter));
 324   ipc_task_runner_->PostTask(
 325       FROM_HERE, base::Bind(&Context::OnAddFilter, this));
 326 }
 327
 328 // Called on the listener's thread
 329 void ChannelProxy::Context::OnDispatchMessage(const Message& message) {
 330 #ifdef IPC_MESSAGE_LOG_ENABLED
 331   Logging* logger = Logging::GetInstance();
 332   std::string name;
 333   logger->GetMessageText(message.type(), &name, &message, NULL);
 334   TRACE_EVENT1("toplevel", "ChannelProxy::Context::OnDispatchMessage",
 335                "name", name);
 336 #else
 337   TRACE_EVENT2("toplevel", "ChannelProxy::Context::OnDispatchMessage",
 338                "class", IPC_MESSAGE_ID_CLASS(message.type()),
 339                "line", IPC_MESSAGE_ID_LINE(message.type()));
 340 #endif
 341
 342   if (!listener_)
 343     return;
 344
 345   OnDispatchConnected();
 346
 347 #ifdef IPC_MESSAGE_LOG_ENABLED
 348   if (message.type() == IPC_LOGGING_ID) {
 349     logger->OnReceivedLoggingMessage(message);
 350     return;
 351   }
 352
 353   if (logger->Enabled())
 354     logger->OnPreDispatchMessage(message);
 355 #endif
 356
 357   listener_->OnMessageReceived(message);
 358
 359 #ifdef IPC_MESSAGE_LOG_ENABLED
 360   if (logger->Enabled())
 361     logger->OnPostDispatchMessage(message, channel_id_);
 362 #endif
 363 }
 364
 365 // Called on the listener's thread
 366 void ChannelProxy::Context::OnDispatchConnected() {
 367   if (channel_connected_called_)
 368     return;
 369
 370   channel_connected_called_ = true;
 371   if (listener_)
 372     listener_->OnChannelConnected(peer_pid_);
 373 }
 374
 375 // Called on the listener's thread
 376 void ChannelProxy::Context::OnDispatchError() {
 377   if (listener_)
 378     listener_->OnChannelError();
 379 }
 380
 381 //-----------------------------------------------------------------------------
 382
 383 ChannelProxy::ChannelProxy(const IPC::ChannelHandle& channel_handle,
 384                            Channel::Mode mode,
 385                            Listener* listener,
 386                            base::SingleThreadTaskRunner* ipc_task_runner)
 387     : context_(new Context(listener, ipc_task_runner)),
 388       did_init_(false) {
 389   Init(channel_handle, mode, true);
 390 }
 391
 392 ChannelProxy::ChannelProxy(Context* context)
 393     : context_(context),
 394       did_init_(false) {
 395 }
 396
 397 ChannelProxy::~ChannelProxy() {
 398   DCHECK(CalledOnValidThread());
 399
 400   Close();
 401 }
 402
 403 void ChannelProxy::Init(const IPC::ChannelHandle& channel_handle,
 404                         Channel::Mode mode,
 405                         bool create_pipe_now) {
 406   DCHECK(CalledOnValidThread());
 407   DCHECK(!did_init_);
 408 #if defined(OS_POSIX)
 409   // When we are creating a server on POSIX, we need its file descriptor
 410   // to be created immediately so that it can be accessed and passed
 411   // to other processes. Forcing it to be created immediately avoids
 412   // race conditions that may otherwise arise.
 413   if (mode & Channel::MODE_SERVER_FLAG) {
 414     create_pipe_now = true;
 415   }
 416 #endif  // defined(OS_POSIX)
 417
 418   if (create_pipe_now) {
 419     // Create the channel immediately.  This effectively sets up the
 420     // low-level pipe so that the client can connect.  Without creating
 421     // the pipe immediately, it is possible for a listener to attempt
 422     // to connect and get an error since the pipe doesn't exist yet.
 423     context_->CreateChannel(channel_handle, mode);
 424   } else {
 425     context_->ipc_task_runner()->PostTask(
 426         FROM_HERE, base::Bind(&Context::CreateChannel, context_.get(),
 427                               channel_handle, mode));
 428   }
 429
 430   // complete initialization on the background thread
 431   context_->ipc_task_runner()->PostTask(
 432       FROM_HERE, base::Bind(&Context::OnChannelOpened, context_.get()));
 433
 434   did_init_ = true;
 435 }
 436
 437 void ChannelProxy::Close() {
 438   DCHECK(CalledOnValidThread());
 439
 440   // Clear the backpointer to the listener so that any pending calls to
 441   // Context::OnDispatchMessage or OnDispatchError will be ignored.  It is
 442   // possible that the channel could be closed while it is receiving messages!
 443   context_->Clear();
 444
 445   if (context_->ipc_task_runner()) {
 446     context_->ipc_task_runner()->PostTask(
 447         FROM_HERE, base::Bind(&Context::OnChannelClosed, context_.get()));
 448   }
 449 }
 450
 451 bool ChannelProxy::Send(Message* message) {
 452   DCHECK(did_init_);
 453
 454   // TODO(alexeypa): add DCHECK(CalledOnValidThread()) here. Currently there are
 455   // tests that call Send() from a wrong thread. See http://crbug.com/163523.
 456
 457 #ifdef IPC_MESSAGE_LOG_ENABLED
 458   Logging::GetInstance()->OnSendMessage(message, context_->channel_id());
 459 #endif
 460
 461   context_->ipc_task_runner()->PostTask(
 462       FROM_HERE,
 463       base::Bind(&ChannelProxy::Context::OnSendMessage,
 464                  context_, base::Passed(scoped_ptr<Message>(message))));
 465   return true;
 466 }
 467
 468 void ChannelProxy::AddFilter(MessageFilter* filter) {
 469   DCHECK(CalledOnValidThread());
 470
 471   context_->AddFilter(filter);
 472 }
 473
 474 void ChannelProxy::RemoveFilter(MessageFilter* filter) {
 475   DCHECK(CalledOnValidThread());
 476
 477   context_->ipc_task_runner()->PostTask(
 478       FROM_HERE, base::Bind(&Context::OnRemoveFilter, context_.get(),
 479                             make_scoped_refptr(filter)));
 480 }
 481
 482 void ChannelProxy::ClearIPCTaskRunner() {
 483   DCHECK(CalledOnValidThread());
 484
 485   context()->ClearIPCTaskRunner();
 486 }
 487
 488 #if defined(OS_POSIX) && !defined(OS_NACL)
 489 // See the TODO regarding lazy initialization of the channel in
 490 // ChannelProxy::Init().
 491 int ChannelProxy::GetClientFileDescriptor() {
 492   DCHECK(CalledOnValidThread());
 493
 494   Channel* channel = context_.get()->channel_.get();
 495   // Channel must have been created first.
 496   DCHECK(channel) << context_.get()->channel_id_;
 497   return channel->GetClientFileDescriptor();
 498 }
 499
 500 int ChannelProxy::TakeClientFileDescriptor() {
 501   DCHECK(CalledOnValidThread());
 502
 503   Channel* channel = context_.get()->channel_.get();
 504   // Channel must have been created first.
 505   DCHECK(channel) << context_.get()->channel_id_;
 506   return channel->TakeClientFileDescriptor();
 507 }
 508
 509 bool ChannelProxy::GetPeerEuid(uid_t* peer_euid) const {
 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->GetPeerEuid(peer_euid);
 516 }
 517 #endif
 518
 519 //-----------------------------------------------------------------------------
 520
 521 }  // namespace IPC