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