Fix UAF in new Mojo EDK.

The problem was that MessagePipeDispatcher was calling Broker::CloseMessagePipe asynchronously, and in between the time that the posted task was run the MPD could be deleted and called back by the channel for another error.

Since all these methods are now called on the IO thread only, fix this by allowing reentrancy for CloseMessagePipe.

BUG=561803
TEST= linux_chromeos browser_tests pass with new EDK

Review URL: https://codereview.chromium.org/1537593002

Cr-Commit-Position: refs/heads/master@{#365782}

mojo/edk/system/broker.h

@@ -55,6 +55,8 @@ class MOJO_SYSTEM_IMPL_EXPORT Broker {
 
   // Called by |message_pipe| when it's closing so that its route can be
   // unregistered.
+  // It's ok to call this from a callback (i.e. from OnError or
+  // GotNonTransferableChannel).
   virtual void CloseMessagePipe(uint64_t pipe_id,
                                 MessagePipeDispatcher* message_pipe) = 0;
 };

mojo/edk/system/broker_state.cc

@@ -102,7 +102,6 @@ void BrokerState::ConnectMessagePipe(uint64_t pipe_id,
 void BrokerState::CloseMessagePipe(uint64_t pipe_id,
                                    MessagePipeDispatcher* message_pipe) {
   DCHECK(internal::g_io_thread_task_runner->RunsTasksOnCurrentThread());
-  base::AutoLock auto_lock(lock_);
 
   CHECK(connected_pipes_.find(message_pipe) != connected_pipes_.end());
   connected_pipes_[message_pipe]->RemoveRoute(pipe_id);
@@ -235,7 +234,9 @@ void BrokerState::AttachMessagePipe(MessagePipeDispatcher* message_pipe,
   // then when it's read it returns no messages because it doesn't have the
   // channel yet.
   message_pipe->GotNonTransferableChannel(raw_channel->channel());
-  raw_channel->AddRoute(pipe_id, message_pipe);
+  // The above call could have caused |CloseMessagePipe| to be called.
+  if (connected_pipes_.find(message_pipe) != connected_pipes_.end())
+    raw_channel->AddRoute(pipe_id, message_pipe);
 }
 
 }  // namespace edk

mojo/edk/system/child_broker.cc

@@ -272,7 +272,9 @@ void ChildBroker::AttachMessagePipe(MessagePipeDispatcher* message_pipe,
   // then when it's read it returns no messages because it doesn't have the
   // channel yet.
   message_pipe->GotNonTransferableChannel(raw_channel->channel());
-  raw_channel->AddRoute(pipe_id, message_pipe);
+  // The above call could have caused |CloseMessagePipe| to be called.
+  if (connected_pipes_.find(message_pipe) != connected_pipes_.end())
+    raw_channel->AddRoute(pipe_id, message_pipe);
 }
 
 #if defined(OS_WIN)

mojo/edk/system/message_pipe_dispatcher.cc

@@ -193,12 +193,7 @@ void MessagePipeDispatcher::GotNonTransferableChannel(RawChannel* channel) {
 
   if (non_transferable_state_ == WAITING_FOR_CONNECT_TO_CLOSE) {
     // We kept this object alive until it's connected, we can release it now.
-    // Since we're in a callback from the Broker, call it asynchronously.
-    internal::g_io_thread_task_runner->PostTask(
-        FROM_HERE,
-        base::Bind(&Broker::CloseMessagePipe,
-                   base::Unretained(internal::g_broker), pipe_id_,
-                   base::Unretained(this)));
+    internal::g_broker->CloseMessagePipe(pipe_id_, this);
     non_transferable_state_ = CLOSED;
     channel_ = nullptr;
     base::MessageLoop::current()->ReleaseSoon(FROM_HERE, this);
@@ -924,12 +919,7 @@ void MessagePipeDispatcher::OnError(Error error) {
         channel_->Shutdown();
       } else {
         CHECK_NE(non_transferable_state_, CLOSED);
-        // Since we're in a callback from the Broker, call it asynchronously.
-        internal::g_io_thread_task_runner->PostTask(
-            FROM_HERE,
-            base::Bind(&Broker::CloseMessagePipe,
-                       base::Unretained(internal::g_broker), pipe_id_,
-                       base::Unretained(this)));
+        internal::g_broker->CloseMessagePipe(pipe_id_, this);
         non_transferable_state_ = CLOSED;
       }
       channel_ = nullptr;

mojo/edk/system/routed_raw_channel.cc

@@ -59,7 +59,9 @@ void RoutedRawChannel::AddRoute(uint64_t route_id,
 
 void RoutedRawChannel::RemoveRoute(uint64_t route_id) {
   DCHECK(internal::g_io_thread_task_runner->RunsTasksOnCurrentThread());
-  CHECK(routes_.find(route_id) != routes_.end());
+  // We don't check that routes_ contains route_id because it's possible for it
+  // to not have been added yet (i.e. it's waiting to be added later down the
+  // call stack).
   routes_.erase(route_id);
 
   // Only send a message to the other side to close the route if we hadn't
@@ -133,23 +135,21 @@ void RoutedRawChannel::OnReadMessage(
 void RoutedRawChannel::OnError(Error error) {
   DCHECK(internal::g_io_thread_task_runner->RunsTasksOnCurrentThread());
 
-  // This needs to match non-multiplexed MessagePipeDispatcher's destruction of
-  // the channel only when read errors occur.
-  if (error != ERROR_WRITE || routes_.empty()) {
-    channel_->Shutdown();
-    channel_ = nullptr;
+  // Note: we must ensure we don't call RawChannel::Shutdown until after we've
+  // called OnError on each route's delegate.
+  for (auto it = routes_.begin(); it != routes_.end();) {
+    // The delegate might call RemoveRoute in their OnError implementation which
+    // would invalidate |it|. So increment it first.
+    auto cur_it = it++;
+    cur_it->second->OnError(error);
   }
 
   if (routes_.empty()) {
+    channel_->Shutdown();
+    channel_ = nullptr;
     delete this;
     return;
   }
-
-  for (auto it = routes_.begin(); it != routes_.end();) {
-    // Handle the delegate calling RemoveRoute in this call.
-    auto cur_it = it++;
-    cur_it->second->OnError(error);
-  }
 }
 
 }  // namespace edk