Mojo: Add NotifyBadMessage API

mojo/edk/system/node_controller.cc

 // Copyright 2016 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 "mojo/edk/system/node_controller.h"
 
 #include <algorithm>
 #include <limits>
 
 #include "base/bind.h"
@@ skipping 121 matching lines @@
 #endif
 
 void NodeController::SetIOTaskRunner(
     scoped_refptr<base::TaskRunner> task_runner) {
   io_task_runner_ = task_runner;
   ThreadDestructionObserver::Create(
       io_task_runner_,
       base::Bind(&NodeController::DropAllPeers, base::Unretained(this)));
 }
 
-void NodeController::ConnectToChild(base::ProcessHandle process_handle,
-                                    ScopedPlatformHandle platform_handle,
-                                    const std::string& child_token) {
+void NodeController::ConnectToChild(
+    base::ProcessHandle process_handle,
+    ScopedPlatformHandle platform_handle,
+    const std::string& child_token,
+    const ProcessErrorCallback& process_error_callback) {
   // Generate the temporary remote node name here so that it can be associated
   // with the embedder's child_token. If an error occurs in the child process
   // after it is launched, but before any reserved ports are connected, this can
   // be used to clean up any dangling ports.
   ports::NodeName node_name;
   GenerateRandomName(&node_name);
 
   {
     base::AutoLock lock(reserved_ports_lock_);
     bool inserted = pending_child_tokens_.insert(
         std::make_pair(node_name, child_token)).second;
     DCHECK(inserted);
   }
 
   io_task_runner_->PostTask(
       FROM_HERE,
       base::Bind(&NodeController::ConnectToChildOnIOThread,
                  base::Unretained(this),
                  process_handle,
                  base::Passed(&platform_handle),
-                 node_name));
+                 node_name,
+                 process_error_callback));
 }
 
 void NodeController::CloseChildPorts(const std::string& child_token) {
   std::vector<ports::PortRef> ports_to_close;
   {
     std::vector<std::string> port_tokens;
     base::AutoLock lock(reserved_ports_lock_);
     for (const auto& port : reserved_ports_) {
       if (port.second.child_token == child_token) {
         DVLOG(1) << "Closing reserved port " << port.second.port.name();
@@ skipping 127 matching lines @@
 void NodeController::RequestShutdown(const base::Closure& callback) {
   {
     base::AutoLock lock(shutdown_lock_);
     shutdown_callback_ = callback;
     shutdown_callback_flag_.Set(true);
   }
 
   AttemptShutdownIfRequested();
 }
 
+void NodeController::NotifyBadMessageFrom(const ports::NodeName& source_node,
+                                          const std::string& error) {
+  scoped_refptr<NodeChannel> peer = GetPeerChannel(source_node);
+  if (peer)
+    peer->NotifyBadMessage(error);
+}
+
 void NodeController::ConnectToChildOnIOThread(
     base::ProcessHandle process_handle,
     ScopedPlatformHandle platform_handle,
-    ports::NodeName token) {
+    ports::NodeName token,
+    const ProcessErrorCallback& process_error_callback) {
   DCHECK(io_task_runner_->RunsTasksOnCurrentThread());
 
 #if defined(OS_POSIX) && !defined(OS_MACOSX) && !defined(OS_NACL)
   PlatformChannelPair node_channel;
   // BrokerHost owns itself.
   BrokerHost* broker_host = new BrokerHost(std::move(platform_handle));
   broker_host->SendChannel(node_channel.PassClientHandle());
   scoped_refptr<NodeChannel> channel = NodeChannel::Create(
-      this, node_channel.PassServerHandle(), io_task_runner_);
+      this, node_channel.PassServerHandle(), io_task_runner_,
+      process_error_callback);
 #else
   scoped_refptr<NodeChannel> channel =
-      NodeChannel::Create(this, std::move(platform_handle), io_task_runner_);
+      NodeChannel::Create(this, std::move(platform_handle), io_task_runner_,
+                          process_error_callback);
 #endif
 
   // We set up the child channel with a temporary name so it can be identified
   // as a pending child if it writes any messages to the channel. We may start
   // receiving messages from it (though we shouldn't) as soon as Start() is
   // called below.
 
   pending_children_.insert(std::make_pair(token, channel));
   RecordPendingChildCount(pending_children_.size());
 
   channel->SetRemoteNodeName(token);
   channel->SetRemoteProcessHandle(process_handle);
   channel->Start();
 
   channel->AcceptChild(name_, token);
 }
 
 void NodeController::ConnectToParentOnIOThread(
     ScopedPlatformHandle platform_handle) {
   DCHECK(io_task_runner_->RunsTasksOnCurrentThread());
 
   {
     base::AutoLock lock(parent_lock_);
     DCHECK(parent_name_ == ports::kInvalidNodeName);
 
     // At this point we don't know the parent's name, so we can't yet insert it
     // into our |peers_| map. That will happen as soon as we receive an
     // AcceptChild message from them.
     bootstrap_parent_channel_ =
-        NodeChannel::Create(this, std::move(platform_handle), io_task_runner_);
+        NodeChannel::Create(this, std::move(platform_handle), io_task_runner_,
+                            ProcessErrorCallback());
   }
   bootstrap_parent_channel_->Start();
 }
 
 scoped_refptr<NodeChannel> NodeController::GetPeerChannel(
     const ports::NodeName& name) {
   base::AutoLock lock(peers_lock_);
   auto it = peers_.find(name);
   if (it == peers_.end())
     return nullptr;
@@ skipping 380 matching lines @@
   }
 
   if (GetPeerChannel(client_name)) {
     DLOG(ERROR) << "Ignoring AddBrokerClient for known client.";
     DropPeer(from_node);
     return;
   }
 
   PlatformChannelPair broker_channel;
   scoped_refptr<NodeChannel> client = NodeChannel::Create(
-      this, broker_channel.PassServerHandle(), io_task_runner_);
+      this, broker_channel.PassServerHandle(), io_task_runner_,
+      ProcessErrorCallback());
 
 #if defined(OS_WIN)
   // The broker must have a working handle to the client process in order to
   // properly copy other handles to and from the client.
   if (!scoped_process_handle.is_valid()) {
     DLOG(ERROR) << "Broker rejecting client with invalid process handle.";
     return;
   }
   client->SetRemoteProcessHandle(scoped_process_handle.release().handle);
 #else
@@ skipping 56 matching lines @@
 
   // It's now possible to add both the broker and the parent as peers.
   // Note that the broker and parent may be the same node.
   scoped_refptr<NodeChannel> broker;
   if (broker_name == parent_name) {
     DCHECK(!broker_channel.is_valid());
     broker = parent;
   } else {
     DCHECK(broker_channel.is_valid());
     broker = NodeChannel::Create(this, std::move(broker_channel),
-                                 io_task_runner_);
+                                 io_task_runner_, ProcessErrorCallback());
     AddPeer(broker_name, broker, true /* start_channel */);
   }
 
   AddPeer(parent_name, parent, false /* start_channel */);
 
   {
     // Complete any port merge requests we have waiting for the parent.
     base::AutoLock lock(pending_port_merges_lock_);
     for (const auto& request : pending_port_merges_)
       parent->RequestPortMerge(request.second.name(), request.first);
@@ skipping 41 matching lines @@
   if (!ports::Message::Parse(data,
                              num_data_bytes,
                              &num_header_bytes,
                              &num_payload_bytes,
                              &num_ports_bytes)) {
     DropPeer(from_node);
     return;
   }
 
   CHECK(channel_message);
-  ports::ScopedMessage message(
+  std::unique_ptr<PortsMessage> ports_message(
       new PortsMessage(num_header_bytes,
                        num_payload_bytes,
                        num_ports_bytes,
                        std::move(channel_message)));
-
-  node_->AcceptMessage(std::move(message));
+  ports_message->set_source_node(from_node);
+  node_->AcceptMessage(ports::ScopedMessage(ports_message.release()));
   AcceptIncomingMessages();
 }
 
 void NodeController::OnRequestPortMerge(
     const ports::NodeName& from_node,
     const ports::PortName& connector_port_name,
     const std::string& token) {
   DCHECK(io_task_runner_->RunsTasksOnCurrentThread());
 
   DVLOG(2) << "Node " << name_ << " received RequestPortMerge for token "
@@ skipping 47 matching lines @@
   if (!channel_handle.is_valid()) {
     node_->LostConnectionToNode(name);
 
     DLOG(ERROR) << "Could not be introduced to peer " << name;
     base::AutoLock lock(peers_lock_);
     pending_peer_messages_.erase(name);
     return;
   }
 
   scoped_refptr<NodeChannel> channel =
-      NodeChannel::Create(this, std::move(channel_handle), io_task_runner_);
+      NodeChannel::Create(this, std::move(channel_handle), io_task_runner_,
+                          ProcessErrorCallback());
 
   DVLOG(1) << "Adding new peer " << name << " via parent introduction.";
   AddPeer(name, channel, true /* start_channel */);
 }
 
 #if defined(OS_WIN) || (defined(OS_MACOSX) && !defined(OS_IOS))
 void NodeController::OnRelayPortsMessage(const ports::NodeName& from_node,
                                          base::ProcessHandle from_process,
                                          const ports::NodeName& destination,
                                          Channel::MessagePtr message) {
@@ skipping 46 matching lines @@
 #endif  // defined(OS_WIN)
 
   if (destination == name_) {
     // Great, we can deliver this message locally.
     OnPortsMessage(from_node, std::move(message));
     return;
   }
 
   scoped_refptr<NodeChannel> peer = GetPeerChannel(destination);
   if (peer)
-    peer->PortsMessage(std::move(message));
+    peer->PortsMessageFromRelay(from_node, std::move(message));
   else
     DLOG(ERROR) << "Dropping relay message for unknown node " << destination;
 }
+
+void NodeController::OnPortsMessageFromRelay(const ports::NodeName& from_node,
+                                             const ports::NodeName& source_node,
+                                             Channel::MessagePtr message) {
+  if (GetPeerChannel(from_node) != GetBrokerChannel()) {
+    LOG(ERROR) << "Refusing relayed message from non-broker node.";
+    DropPeer(from_node);
+    return;
+  }
+
+  OnPortsMessage(source_node, std::move(message));
+}
 #endif
 
 void NodeController::OnChannelError(const ports::NodeName& from_node) {
   if (io_task_runner_->RunsTasksOnCurrentThread()) {
     DropPeer(from_node);
     // DropPeer may have caused local port closures, so be sure to process any
     // pending local messages.
     AcceptIncomingMessages();
   } else {
     io_task_runner_->PostTask(
@@ skipping 40 matching lines @@
     shutdown_callback_flag_.Set(false);
   }
 
   DCHECK(!callback.is_null());
 
   callback.Run();
 }
 
 }  // namespace edk
 }  // namespace mojo