Mojo: Add NotifyBadMessage API

mojo/edk/system/node_channel.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_channel.h"
 
 #include <cstring>
 #include <limits>
 #include <sstream>
 
@@ skipping 26 matching lines @@
   BROKER_CLIENT_ADDED,
   ACCEPT_BROKER_CLIENT,
   PORTS_MESSAGE,
   REQUEST_PORT_MERGE,
   REQUEST_INTRODUCTION,
   INTRODUCE,
 #if defined(OS_WIN) || (defined(OS_MACOSX) && !defined(OS_IOS))
   RELAY_PORTS_MESSAGE,
 #endif
   BROADCAST,
+#if defined(OS_WIN) || (defined(OS_MACOSX) && !defined(OS_IOS))
+  PORTS_MESSAGE_FROM_RELAY,
+#endif
 };
 
 struct Header {
   MessageType type;
   uint32_t padding;
 };
 
 static_assert(sizeof(Header) % kChannelMessageAlignment == 0,
     "Invalid header size.");
 
@@ skipping 48 matching lines @@
 // cannot be introduced.
 struct IntroductionData {
   ports::NodeName name;
 };
 
 #if defined(OS_WIN) || (defined(OS_MACOSX) && !defined(OS_IOS))
 // This struct is followed by the full payload of a message to be relayed.
 struct RelayPortsMessageData {
   ports::NodeName destination;
 };
+
+// This struct is followed by the full payload of a relayed message.
+struct PortsMessageFromRelayData {
+  ports::NodeName source;
+};
 #endif
 
 template <typename DataType>
 Channel::MessagePtr CreateMessage(MessageType type,
                                   size_t payload_size,
                                   size_t num_handles,
                                   DataType** out_data) {
   Channel::MessagePtr message(
       new Channel::Message(sizeof(Header) + payload_size, num_handles));
   Header* header = reinterpret_cast<Header*>(message->mutable_payload());
@@ skipping 14 matching lines @@
       static_cast<const char*>(bytes) + sizeof(Header));
   return true;
 }
 
 }  // namespace
 
 // static
 scoped_refptr<NodeChannel> NodeChannel::Create(
     Delegate* delegate,
     ScopedPlatformHandle platform_handle,
-    scoped_refptr<base::TaskRunner> io_task_runner) {
+    scoped_refptr<base::TaskRunner> io_task_runner,
+    const ProcessErrorCallback& process_error_callback) {
 #if defined(OS_NACL)
   LOG(FATAL) << "Multi-process not yet supported on NaCl";
   return nullptr;
 #else
-  return new NodeChannel(delegate, std::move(platform_handle), io_task_runner);
+  return new NodeChannel(delegate, std::move(platform_handle), io_task_runner,
+                         process_error_callback);
 #endif
 }
 
 // static
 Channel::MessagePtr NodeChannel::CreatePortsMessage(size_t payload_size,
                                                     void** payload,
                                                     size_t num_handles) {
   return CreateMessage(MessageType::PORTS_MESSAGE, payload_size, num_handles,
                        payload);
 }
@@ skipping 26 matching lines @@
     relay->RemoveObserver(this);
 #endif
 
   base::AutoLock lock(channel_lock_);
   if (channel_) {
     channel_->ShutDown();
     channel_ = nullptr;
   }
 }
 
+void NodeChannel::NotifyBadMessage(const std::string& error) {
+  if (!process_error_callback_.is_null())
+    process_error_callback_.Run("Received bad user message: " + error);
+}
+
 void NodeChannel::SetRemoteProcessHandle(base::ProcessHandle process_handle) {
   DCHECK(io_task_runner_->RunsTasksOnCurrentThread());
   base::AutoLock lock(remote_process_handle_lock_);
   CHECK_NE(remote_process_handle_, base::GetCurrentProcessHandle());
   remote_process_handle_ = process_handle;
 }
 
 bool NodeChannel::HasRemoteProcessHandle() {
   base::AutoLock lock(remote_process_handle_lock_);
   return remote_process_handle_ != base::kNullProcessHandle;
@@ skipping 162 matching lines @@
   RelayPortsMessageData* data;
   Channel::MessagePtr relay_message = CreateMessage(
       MessageType::RELAY_PORTS_MESSAGE, num_bytes, handles->size(), &data);
   data->destination = destination;
   memcpy(data + 1, message->data(), message->data_num_bytes());
   relay_message->SetHandles(std::move(handles));
 #endif  // defined(OS_WIN)
 
   WriteChannelMessage(std::move(relay_message));
 }
+
+void NodeChannel::PortsMessageFromRelay(const ports::NodeName& source,
+                                        Channel::MessagePtr message) {
+  size_t num_bytes = sizeof(PortsMessageFromRelayData) +
+      message->payload_size();
+  PortsMessageFromRelayData* data;
+  Channel::MessagePtr relayed_message = CreateMessage(
+      MessageType::PORTS_MESSAGE_FROM_RELAY, num_bytes, message->num_handles(),
+      &data);
+  data->source = source;
+  if (message->payload_size())
+    memcpy(data + 1, message->payload(), message->payload_size());
+  relayed_message->SetHandles(message->TakeHandles());
+  WriteChannelMessage(std::move(relayed_message));
+}
 #endif  // defined(OS_WIN) || (defined(OS_MACOSX) && !defined(OS_IOS))
 
 NodeChannel::NodeChannel(Delegate* delegate,
                          ScopedPlatformHandle platform_handle,
-                         scoped_refptr<base::TaskRunner> io_task_runner)
+                         scoped_refptr<base::TaskRunner> io_task_runner,
+                         const ProcessErrorCallback& process_error_callback)
     : delegate_(delegate),
-      io_task_runner_(io_task_runner)
+      io_task_runner_(io_task_runner),
+      process_error_callback_(process_error_callback)
 #if !defined(OS_NACL)
       , channel_(
           Channel::Create(this, std::move(platform_handle), io_task_runner_))
 #endif
       {
 }
 
 NodeChannel::~NodeChannel() {
   ShutDown();
 }
@@ skipping 236 matching lines @@
     }
 #endif
 
     case MessageType::BROADCAST:
       // This is here as a placeholder for now because nothing sends a BROADCAST
       // message yet. This avoids breakage on version skew (namely for ARC) when
       // we actually begin using the message.
       DVLOG(1) << "Ignoring unhandled BROADCAST message.";
       return;
 
+#if defined(OS_WIN) || (defined(OS_MACOSX) && !defined(OS_IOS))
+    case MessageType::PORTS_MESSAGE_FROM_RELAY:
+      const PortsMessageFromRelayData* data;
+      if (GetMessagePayload(payload, payload_size, &data)) {
+        size_t num_bytes = payload_size - sizeof(*data);
+        if (num_bytes < sizeof(Header))
+          break;
+        num_bytes -= sizeof(Header);
+
+        size_t num_handles = handles ? handles->size() : 0;
+        Channel::MessagePtr message(
+            new Channel::Message(num_bytes, num_handles));
+        message->SetHandles(std::move(handles));
+        if (num_bytes)
+          memcpy(message->mutable_payload(), data + 1, num_bytes);
+        delegate_->OnPortsMessageFromRelay(
+            remote_node_name_, data->source, std::move(message));
+        return;
+      }
+      break;
+
+#endif  // defined(OS_WIN) || (defined(OS_MACOSX) && !defined(OS_IOS))
+
     default:
       break;
   }
 
   DLOG(ERROR) << "Received invalid message. Closing channel.";
   delegate_->OnChannelError(remote_node_name_);
 }
 
 void NodeChannel::OnChannelError() {
   DCHECK(io_task_runner_->RunsTasksOnCurrentThread());
@@ skipping 126 matching lines @@
 
   base::AutoLock lock(channel_lock_);
   if (!channel_)
     DLOG(ERROR) << "Dropping message on closed channel.";
   else
     channel_->Write(std::move(message));
 }
 
 }  // namespace edk
 }  // namespace mojo