Create a broker interface for the new Mojo EDK so that the browser can create and duplicate messa...

mojo/edk/system/message_pipe_dispatcher.cc

 // Copyright 2015 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/message_pipe_dispatcher.h"
 
 #include "base/bind.h"
 #include "base/logging.h"
 #include "base/message_loop/message_loop.h"
 #include "mojo/edk/embedder/embedder_internal.h"
 #include "mojo/edk/embedder/platform_handle_utils.h"
 #include "mojo/edk/embedder/platform_shared_buffer.h"
 #include "mojo/edk/embedder/platform_support.h"
 #include "mojo/edk/system/configuration.h"
 #include "mojo/edk/system/message_in_transit.h"
 #include "mojo/edk/system/options_validation.h"
 #include "mojo/edk/system/transport_data.h"
 
+#if defined(OS_WIN)
+#include "mojo/edk/system/token_serializer_win.h"
+#endif
+
 namespace mojo {
 namespace edk {
 
 // TODO(jam): do more tests on using channel on same thread if it supports it (
 // i.e. with USE_CHROME_EDK and Windows). Also see ipc_channel_mojo.cc
 bool g_use_channel_on_io_thread_only = true;
 
 namespace {
 
 const size_t kInvalidMessagePipeHandleIndex = static_cast<size_t>(-1);
 
 struct MOJO_ALIGNAS(8) SerializedMessagePipeHandleDispatcher {
   // Could be |kInvalidMessagePipeHandleIndex| if the other endpoint of the MP
   // was closed.
   size_t platform_handle_index;
   bool write_error;
 
   size_t shared_memory_handle_index;  // (Or |kInvalidMessagePipeHandleIndex|.)
   uint32_t shared_memory_size;
 
   size_t serialized_read_buffer_size;
   size_t serialized_write_buffer_size;
-  size_t serialized_messagage_queue_size;
+  size_t serialized_message_queue_size;
 
   // These are the FDs required as part of serializing channel_ and
   // message_queue_. This is only used on POSIX.
   size_t serialized_fds_index;  // (Or |kInvalidMessagePipeHandleIndex|.)
   size_t serialized_read_fds_length;
   size_t serialized_write_fds_length;
   size_t serialized_message_fds_length;
 };
 
 char* SerializeBuffer(char* start, std::vector<char>* buffer) {
@@ skipping 102 matching lines @@
   if (channel_) {
     channel_->Shutdown();
     channel_ = nullptr;
   }
 }
 
 Dispatcher::Type MessagePipeDispatcher::GetType() const {
   return Type::MESSAGE_PIPE;
 }
 
+#if defined(OS_WIN)
 // TODO(jam): this is copied from RawChannelWin till I figure out what's the
-// best way we want to share this. Need to also consider posix which does
-// require access to the RawChannel.
+// best way we want to share this.
 // Since this is used for serialization of messages read/written to a MP that
 // aren't consumed by Mojo primitives yet, there could be an unbounded number of
 // them when a MP is being sent. As a result, even for POSIX we will probably
 // want to send the handles to the shell process and exchange them for tokens
 // (since we can be sure that the shell will respond to our IPCs, compared to
 // the other end where we're sending the MP to, which may not be reading...).
 ScopedPlatformHandleVectorPtr GetReadPlatformHandles(
     size_t num_platform_handles,
     const void* platform_handle_table) {
-  // TODO(jam): this code will have to be updated once it's used in a sandbox
-  // and the receiving process doesn't have duplicate permission for the
-  // receiver. Once there's a broker and we have a connection to it (possibly
-  // through ConnectionManager), then we can make a sync IPC to it here to get a
-  // token for this handle, and it will duplicate the handle to is process. Then
-  // we pass the token to the receiver, which will then make a sync call to the
-  // broker to get a duplicated handle. This will also allow us to avoid leaks
-  // of the handle if the receiver dies, since the broker can notice that.
-  DCHECK_GT(num_platform_handles, 0u);
   ScopedPlatformHandleVectorPtr rv(new PlatformHandleVector());
+  rv->resize(num_platform_handles);
 
-#if defined(OS_WIN)
-  const char* serialization_data =
-      static_cast<const char*>(platform_handle_table);
-  for (size_t i = 0; i < num_platform_handles; i++) {
-    DWORD pid = *reinterpret_cast<const DWORD*>(serialization_data);
-    serialization_data += sizeof(DWORD);
-    HANDLE source_handle = *reinterpret_cast<const HANDLE*>(serialization_data);
-    serialization_data += sizeof(HANDLE);
-    base::Process sender =
-        base::Process::OpenWithAccess(pid, PROCESS_DUP_HANDLE);
-    DCHECK(sender.IsValid());
-    HANDLE target_handle = NULL;
-    BOOL dup_result =
-        DuplicateHandle(sender.Handle(), source_handle,
-                        base::GetCurrentProcessHandle(), &target_handle, 0,
-                        FALSE, DUPLICATE_SAME_ACCESS | DUPLICATE_CLOSE_SOURCE);
-    DCHECK(dup_result);
-    rv->push_back(PlatformHandle(target_handle));
-  }
-#else
-  NOTREACHED() << "TODO(jam): implement";
-#endif
+  const uint64_t* tokens =
+      static_cast<const uint64_t*>(platform_handle_table);
+  internal::g_token_serializer->TokenToHandle(
+      tokens, num_platform_handles, &rv->at(0));
   return rv.Pass();
 }
+#endif
 
 scoped_refptr<MessagePipeDispatcher> MessagePipeDispatcher::Deserialize(
     const void* source,
     size_t size,
     PlatformHandleVector* platform_handles) {
   if (size != sizeof(SerializedMessagePipeHandleDispatcher)) {
     LOG(ERROR) << "Invalid serialized message pipe dispatcher (bad size)";
     return nullptr;
   }
 
   const SerializedMessagePipeHandleDispatcher* serialization =
       static_cast<const SerializedMessagePipeHandleDispatcher*>(source);
   if (serialization->shared_memory_size !=
       (serialization->serialized_read_buffer_size +
        serialization->serialized_write_buffer_size +
-       serialization->serialized_messagage_queue_size)) {
+       serialization->serialized_message_queue_size)) {
     LOG(ERROR) << "Invalid serialized message pipe dispatcher (bad struct)";
     return nullptr;
   }
 
   ScopedPlatformHandle platform_handle, shared_memory_handle;
   if (!GetHandle(serialization->platform_handle_index,
                  platform_handles, &platform_handle) ||
       !GetHandle(serialization->shared_memory_handle_index,
                  platform_handles, &shared_memory_handle)) {
     return nullptr;
@@ skipping 16 matching lines @@
       serialized_read_buffer = buffer;
       serialized_read_buffer_size = serialization->serialized_read_buffer_size;
       buffer += serialized_read_buffer_size;
     }
     if (serialization->serialized_write_buffer_size) {
       serialized_write_buffer = buffer;
       serialized_write_buffer_size =
           serialization->serialized_write_buffer_size;
       buffer += serialized_write_buffer_size;
     }
-    if (serialization->serialized_messagage_queue_size) {
+    if (serialization->serialized_message_queue_size) {
       message_queue_data = buffer;
-      message_queue_size = serialization->serialized_messagage_queue_size;
+      message_queue_size = serialization->serialized_message_queue_size;
       buffer += message_queue_size;
     }
   }
 
   scoped_refptr<MessagePipeDispatcher> rv(
       Create(MessagePipeDispatcher::kDefaultCreateOptions));
   rv->write_error_ = serialization->write_error;
 
   std::vector<int> serialized_read_fds;
   std::vector<int> serialized_write_fds;
@@ skipping 176 matching lines @@
         static_cast<const char*>(message->main_buffer()),
         static_cast<const char*>(message->main_buffer()) + main_buffer_size);
 
     // cont'd
     if (transport_data_buffer_size != 0) {
       // TODO(jam): copied from RawChannelWin::WriteNoLock(
       PlatformHandleVector* all_platform_handles =
           message->transport_data()->platform_handles();
       if (all_platform_handles) {
 #if defined(OS_WIN)
-        char* serialization_data =
+        uint64_t* tokens = reinterpret_cast<uint64_t*>(
             static_cast<char*>(message->transport_data()->buffer()) +
-             message->transport_data()->platform_handle_table_offset();
-        DWORD current_process_id = base::GetCurrentProcId();
-        for (size_t i = 0; i < all_platform_handles->size(); i++) {
-          *reinterpret_cast<DWORD*>(serialization_data) = current_process_id;
-          serialization_data += sizeof(DWORD);
-          *reinterpret_cast<HANDLE*>(serialization_data) =
-              all_platform_handles->at(i).handle;
-          serialization_data += sizeof(HANDLE);
+            message->transport_data()->platform_handle_table_offset());
+        internal::g_token_serializer->HandleToToken(
+            &all_platform_handles->at(0), all_platform_handles->size(), tokens);
+        for (size_t i = 0; i < all_platform_handles->size(); i++)
           all_platform_handles->at(i) = PlatformHandle();
-        }
 #else
         for (size_t i = 0; i < all_platform_handles->size(); i++) {
           serialized_fds_.push_back(all_platform_handles->at(i).fd);
           serialized_message_fds_length_++;
           all_platform_handles->at(i) = PlatformHandle();
         }
 #endif
 
       serialized_message_queue_.insert(
           serialized_message_queue_.end(),
@@ skipping 209 matching lines @@
   if (serialized_platform_handle_.is_valid()) {
     serialization->platform_handle_index = platform_handles->size();
     platform_handles->push_back(serialized_platform_handle_.release());
   } else {
     serialization->platform_handle_index = kInvalidMessagePipeHandleIndex;
   }
 
   serialization->write_error = write_error_;
   serialization->serialized_read_buffer_size = serialized_read_buffer_.size();
   serialization->serialized_write_buffer_size = serialized_write_buffer_.size();
-  serialization->serialized_messagage_queue_size =
+  serialization->serialized_message_queue_size =
       serialized_message_queue_.size();
 
   serialization->shared_memory_size = static_cast<uint32_t>(
       serialization->serialized_read_buffer_size +
       serialization->serialized_write_buffer_size +
-      serialization->serialized_messagage_queue_size);
+      serialization->serialized_message_queue_size);
   if (serialization->shared_memory_size) {
     scoped_refptr<PlatformSharedBuffer> shared_buffer(
         internal::g_platform_support->CreateSharedBuffer(
             serialization->shared_memory_size));
     scoped_ptr<PlatformSharedBufferMapping> mapping(
         shared_buffer->Map(0, serialization->shared_memory_size));
     char* start = static_cast<char*>(mapping->GetBase());
     start = SerializeBuffer(start, &serialized_read_buffer_);
     start = SerializeBuffer(start, &serialized_write_buffer_);
     start = SerializeBuffer(start, &serialized_message_queue_);
@@ skipping 161 matching lines @@
       LOG(WARNING) << "Enqueueing null dispatcher";
       dispatchers->push_back(nullptr);
     }
   }
   message->SetDispatchers(dispatchers.Pass());
   return MOJO_RESULT_OK;
 }
 
 }  // namespace edk
 }  // namespace mojo