chrome/browser/extensions/api/messaging/native_message_process_host.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 "chrome/browser/extensions/api/messaging/native_message_process_host.h"
   6
   7 #include "base/bind.h"
   8 #include "base/files/file_path.h"
   9 #include "base/logging.h"
  10 #include "base/process/kill.h"
  11 #include "base/threading/sequenced_worker_pool.h"
  12 #include "chrome/browser/extensions/api/messaging/native_messaging_host_manifest.h"
  13 #include "chrome/browser/extensions/api/messaging/native_process_launcher.h"
  14 #include "chrome/common/chrome_version_info.h"
  15 #include "content/public/browser/browser_thread.h"
  16 #include "extensions/common/constants.h"
  17 #include "extensions/common/features/feature.h"
  18 #include "net/base/file_stream.h"
  19 #include "net/base/io_buffer.h"
  20 #include "net/base/net_errors.h"
  21 #include "net/base/net_util.h"
  22 #include "url/gurl.h"
  23
  24 namespace {
  25
  26 // Maximum message size in bytes for messages received from Native Messaging
  27 // hosts. Message size is limited mainly to prevent Chrome from crashing when
  28 // native application misbehaves (e.g. starts writing garbage to the pipe).
  29 const size_t kMaximumMessageSize = 1024 * 1024;
  30
  31 // Message header contains 4-byte integer size of the message.
  32 const size_t kMessageHeaderSize = 4;
  33
  34 // Size of the buffer to be allocated for each read.
  35 const size_t kReadBufferSize = 4096;
  36
  37 }  // namespace
  38
  39 namespace extensions {
  40
  41 NativeMessageProcessHost::NativeMessageProcessHost(
  42     const std::string& source_extension_id,
  43     const std::string& native_host_name,
  44     scoped_ptr<NativeProcessLauncher> launcher)
  45     : client_(NULL),
  46       source_extension_id_(source_extension_id),
  47       native_host_name_(native_host_name),
  48       launcher_(launcher.Pass()),
  49       closed_(false),
  50 #if defined(OS_POSIX)
  51       read_file_(-1),
  52 #endif
  53       read_pending_(false),
  54       write_pending_(false),
  55       weak_factory_(this) {
  56   DCHECK_CURRENTLY_ON(content::BrowserThread::UI);
  57
  58   task_runner_ = content::BrowserThread::GetMessageLoopProxyForThread(
  59       content::BrowserThread::IO);
  60 }
  61
  62 NativeMessageProcessHost::~NativeMessageProcessHost() {
  63   DCHECK(task_runner_->BelongsToCurrentThread());
  64
  65   if (process_.IsValid()) {
  66     // Kill the host process if necessary to make sure we don't leave zombies.
  67     // On OSX base::EnsureProcessTerminated() may block, so we have to post a
  68     // task on the blocking pool.
  69 #if defined(OS_MACOSX)
  70     content::BrowserThread::PostBlockingPoolTask(
  71         FROM_HERE,
  72         base::Bind(&base::EnsureProcessTerminated, Passed(&process_)));
  73 #else
  74     base::EnsureProcessTerminated(process_.Pass());
  75 #endif
  76   }
  77 }
  78
  79 // static
  80 scoped_ptr<NativeMessageHost> NativeMessageHost::Create(
  81     gfx::NativeView native_view,
  82     const std::string& source_extension_id,
  83     const std::string& native_host_name,
  84     bool allow_user_level,
  85     std::string* error_message) {
  86   return NativeMessageProcessHost::CreateWithLauncher(
  87       source_extension_id,
  88       native_host_name,
  89       NativeProcessLauncher::CreateDefault(allow_user_level, native_view));
  90 }
  91
  92 // static
  93 scoped_ptr<NativeMessageHost> NativeMessageProcessHost::CreateWithLauncher(
  94     const std::string& source_extension_id,
  95     const std::string& native_host_name,
  96     scoped_ptr<NativeProcessLauncher> launcher) {
  97   DCHECK_CURRENTLY_ON(content::BrowserThread::UI);
  98
  99   scoped_ptr<NativeMessageHost> process(
 100       new NativeMessageProcessHost(source_extension_id,
 101                                    native_host_name,
 102                                    launcher.Pass()));
 103
 104   return process.Pass();
 105 }
 106
 107 void NativeMessageProcessHost::LaunchHostProcess() {
 108   DCHECK(task_runner_->BelongsToCurrentThread());
 109
 110   GURL origin(std::string(kExtensionScheme) + "://" + source_extension_id_);
 111   launcher_->Launch(origin, native_host_name_,
 112                     base::Bind(&NativeMessageProcessHost::OnHostProcessLaunched,
 113                                weak_factory_.GetWeakPtr()));
 114 }
 115
 116 void NativeMessageProcessHost::OnHostProcessLaunched(
 117     NativeProcessLauncher::LaunchResult result,
 118     base::Process process,
 119     base::File read_file,
 120     base::File write_file) {
 121   DCHECK(task_runner_->BelongsToCurrentThread());
 122
 123   switch (result) {
 124     case NativeProcessLauncher::RESULT_INVALID_NAME:
 125       Close(kInvalidNameError);
 126       return;
 127     case NativeProcessLauncher::RESULT_NOT_FOUND:
 128       Close(kNotFoundError);
 129       return;
 130     case NativeProcessLauncher::RESULT_FORBIDDEN:
 131       Close(kForbiddenError);
 132       return;
 133     case NativeProcessLauncher::RESULT_FAILED_TO_START:
 134       Close(kFailedToStartError);
 135       return;
 136     case NativeProcessLauncher::RESULT_SUCCESS:
 137       break;
 138   }
 139
 140   process_ = process.Pass();
 141 #if defined(OS_POSIX)
 142   // This object is not the owner of the file so it should not keep an fd.
 143   read_file_ = read_file.GetPlatformFile();
 144 #endif
 145
 146   scoped_refptr<base::TaskRunner> task_runner(
 147       content::BrowserThread::GetBlockingPool()->
 148           GetTaskRunnerWithShutdownBehavior(
 149               base::SequencedWorkerPool::SKIP_ON_SHUTDOWN));
 150
 151   read_stream_.reset(new net::FileStream(read_file.Pass(), task_runner));
 152   write_stream_.reset(new net::FileStream(write_file.Pass(), task_runner));
 153
 154   WaitRead();
 155   DoWrite();
 156 }
 157
 158 void NativeMessageProcessHost::OnMessage(const std::string& json) {
 159   DCHECK(task_runner_->BelongsToCurrentThread());
 160
 161   if (closed_)
 162     return;
 163
 164   // Allocate new buffer for the message.
 165   scoped_refptr<net::IOBufferWithSize> buffer =
 166       new net::IOBufferWithSize(json.size() + kMessageHeaderSize);
 167
 168   // Copy size and content of the message to the buffer.
 169   static_assert(sizeof(uint32) == kMessageHeaderSize,
 170                 "kMessageHeaderSize is incorrect");
 171   *reinterpret_cast<uint32*>(buffer->data()) = json.size();
 172   memcpy(buffer->data() + kMessageHeaderSize, json.data(), json.size());
 173
 174   // Push new message to the write queue.
 175   write_queue_.push(buffer);
 176
 177   // Send() may be called before the host process is started. In that case the
 178   // message will be written when OnHostProcessLaunched() is called. If it's
 179   // already started then write the message now.
 180   if (write_stream_)
 181     DoWrite();
 182 }
 183
 184 void NativeMessageProcessHost::Start(Client* client) {
 185   DCHECK(task_runner_->BelongsToCurrentThread());
 186   DCHECK(!client_);
 187   client_ = client;
 188   // It's safe to use base::Unretained() here because NativeMessagePort always
 189   // deletes us on the IO thread.
 190   task_runner_->PostTask(
 191       FROM_HERE,
 192       base::Bind(&NativeMessageProcessHost::LaunchHostProcess,
 193                  weak_factory_.GetWeakPtr()));
 194 }
 195
 196 scoped_refptr<base::SingleThreadTaskRunner>
 197 NativeMessageProcessHost::task_runner() const {
 198   return task_runner_;
 199 }
 200
 201 #if defined(OS_POSIX)
 202 void NativeMessageProcessHost::OnFileCanReadWithoutBlocking(int fd) {
 203   DCHECK_EQ(fd, read_file_);
 204   DoRead();
 205 }
 206
 207 void NativeMessageProcessHost::OnFileCanWriteWithoutBlocking(int fd) {
 208   NOTREACHED();
 209 }
 210 #endif  // !defined(OS_POSIX)
 211
 212 void NativeMessageProcessHost::ReadNowForTesting() {
 213   DoRead();
 214 }
 215
 216 void NativeMessageProcessHost::WaitRead() {
 217   if (closed_)
 218     return;
 219
 220   DCHECK(!read_pending_);
 221
 222   // On POSIX FileStream::Read() uses blocking thread pool, so it's better to
 223   // wait for the file to become readable before calling DoRead(). Otherwise it
 224   // would always be consuming one thread in the thread pool. On Windows
 225   // FileStream uses overlapped IO, so that optimization isn't necessary there.
 226 #if defined(OS_POSIX)
 227   base::MessageLoopForIO::current()->WatchFileDescriptor(
 228     read_file_, false /* persistent */,
 229     base::MessageLoopForIO::WATCH_READ, &read_watcher_, this);
 230 #else  // defined(OS_POSIX)
 231   DoRead();
 232 #endif  // defined(!OS_POSIX)
 233 }
 234
 235 void NativeMessageProcessHost::DoRead() {
 236   DCHECK(task_runner_->BelongsToCurrentThread());
 237
 238   while (!closed_ && !read_pending_) {
 239     read_buffer_ = new net::IOBuffer(kReadBufferSize);
 240     int result =
 241         read_stream_->Read(read_buffer_.get(), kReadBufferSize,
 242                            base::Bind(&NativeMessageProcessHost::OnRead,
 243                                       weak_factory_.GetWeakPtr()));
 244     HandleReadResult(result);
 245   }
 246 }
 247
 248 void NativeMessageProcessHost::OnRead(int result) {
 249   DCHECK(task_runner_->BelongsToCurrentThread());
 250   DCHECK(read_pending_);
 251   read_pending_ = false;
 252
 253   HandleReadResult(result);
 254   WaitRead();
 255 }
 256
 257 void NativeMessageProcessHost::HandleReadResult(int result) {
 258   DCHECK(task_runner_->BelongsToCurrentThread());
 259
 260   if (closed_)
 261     return;
 262
 263   if (result > 0) {
 264     ProcessIncomingData(read_buffer_->data(), result);
 265   } else if (result == net::ERR_IO_PENDING) {
 266     read_pending_ = true;
 267   } else if (result == 0 || result == net::ERR_CONNECTION_RESET) {
 268     // On Windows we get net::ERR_CONNECTION_RESET for a broken pipe, while on
 269     // Posix read() returns 0 in that case.
 270     Close(kNativeHostExited);
 271   } else {
 272     LOG(ERROR) << "Error when reading from Native Messaging host: " << result;
 273     Close(kHostInputOuputError);
 274   }
 275 }
 276
 277 void NativeMessageProcessHost::ProcessIncomingData(
 278     const char* data, int data_size) {
 279   DCHECK(task_runner_->BelongsToCurrentThread());
 280
 281   incoming_data_.append(data, data_size);
 282
 283   while (true) {
 284     if (incoming_data_.size() < kMessageHeaderSize)
 285       return;
 286
 287     size_t message_size =
 288         *reinterpret_cast<const uint32*>(incoming_data_.data());
 289
 290     if (message_size > kMaximumMessageSize) {
 291       LOG(ERROR) << "Native Messaging host tried sending a message that is "
 292                  << message_size << " bytes long.";
 293       Close(kHostInputOuputError);
 294       return;
 295     }
 296
 297     if (incoming_data_.size() < message_size + kMessageHeaderSize)
 298       return;
 299
 300     client_->PostMessageFromNativeHost(
 301         incoming_data_.substr(kMessageHeaderSize, message_size));
 302
 303     incoming_data_.erase(0, kMessageHeaderSize + message_size);
 304   }
 305 }
 306
 307 void NativeMessageProcessHost::DoWrite() {
 308   DCHECK(task_runner_->BelongsToCurrentThread());
 309
 310   while (!write_pending_ && !closed_) {
 311     if (!current_write_buffer_.get() ||
 312         !current_write_buffer_->BytesRemaining()) {
 313       if (write_queue_.empty())
 314         return;
 315       current_write_buffer_ = new net::DrainableIOBuffer(
 316           write_queue_.front().get(), write_queue_.front()->size());
 317       write_queue_.pop();
 318     }
 319
 320     int result =
 321         write_stream_->Write(current_write_buffer_.get(),
 322                              current_write_buffer_->BytesRemaining(),
 323                              base::Bind(&NativeMessageProcessHost::OnWritten,
 324                                         weak_factory_.GetWeakPtr()));
 325     HandleWriteResult(result);
 326   }
 327 }
 328
 329 void NativeMessageProcessHost::HandleWriteResult(int result) {
 330   DCHECK(task_runner_->BelongsToCurrentThread());
 331
 332   if (result <= 0) {
 333     if (result == net::ERR_IO_PENDING) {
 334       write_pending_ = true;
 335     } else {
 336       LOG(ERROR) << "Error when writing to Native Messaging host: " << result;
 337       Close(kHostInputOuputError);
 338     }
 339     return;
 340   }
 341
 342   current_write_buffer_->DidConsume(result);
 343 }
 344
 345 void NativeMessageProcessHost::OnWritten(int result) {
 346   DCHECK(task_runner_->BelongsToCurrentThread());
 347
 348   DCHECK(write_pending_);
 349   write_pending_ = false;
 350
 351   HandleWriteResult(result);
 352   DoWrite();
 353 }
 354
 355 void NativeMessageProcessHost::Close(const std::string& error_message) {
 356   DCHECK(task_runner_->BelongsToCurrentThread());
 357
 358   if (!closed_) {
 359     closed_ = true;
 360     read_stream_.reset();
 361     write_stream_.reset();
 362     client_->CloseChannel(error_message);
 363   }
 364 }
 365
 366 }  // namespace extensions