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 "ipc/ipc_channel_nacl.h" 6 7 #include <errno.h> 8 #include <stddef.h> 9 #include <sys/types.h> 10 11 #include <algorithm> 12 13 #include "base/bind.h" 14 #include "base/logging.h" 15 #include "base/message_loop/message_loop_proxy.h" 16 #include "base/synchronization/lock.h" 17 #include "base/task_runner_util.h" 18 #include "base/threading/simple_thread.h" 19 #include "ipc/file_descriptor_set_posix.h" 20 #include "ipc/ipc_logging.h" 21 #include "native_client/src/public/imc_syscalls.h" 22 #include "native_client/src/public/imc_types.h" 23 24 namespace IPC { 25 26 struct MessageContents { 27 std::vector<char> data; 28 std::vector<int> fds; 29 }; 30 31 namespace { 32 33 bool ReadDataOnReaderThread(int pipe, MessageContents* contents) { 34 DCHECK(pipe >= 0); 35 if (pipe < 0) 36 return false; 37 38 contents->data.resize(Channel::kReadBufferSize); 39 contents->fds.resize(FileDescriptorSet::kMaxDescriptorsPerMessage); 40 41 NaClAbiNaClImcMsgIoVec iov = { &contents->data[0], contents->data.size() }; 42 NaClAbiNaClImcMsgHdr msg = { 43 &iov, 1, &contents->fds[0], contents->fds.size() 44 }; 45 46 int bytes_read = imc_recvmsg(pipe, &msg, 0); 47 48 if (bytes_read <= 0) { 49 // NaClIPCAdapter::BlockingReceive returns -1 when the pipe closes (either 50 // due to error or for regular shutdown). 51 contents->data.clear(); 52 contents->fds.clear(); 53 return false; 54 } 55 DCHECK(bytes_read); 56 // Resize the buffers down to the number of bytes and fds we actually read. 57 contents->data.resize(bytes_read); 58 contents->fds.resize(msg.desc_length); 59 return true; 60 } 61 62 } // namespace 63 64 class Channel::ChannelImpl::ReaderThreadRunner 65 : public base::DelegateSimpleThread::Delegate { 66 public: 67 // |pipe|: A file descriptor from which we will read using imc_recvmsg. 68 // |data_read_callback|: A callback we invoke (on the main thread) when we 69 // have read data. 70 // |failure_callback|: A callback we invoke when we have a failure reading 71 // from |pipe|. 72 // |main_message_loop|: A proxy for the main thread, where we will invoke the 73 // above callbacks. 74 ReaderThreadRunner( 75 int pipe, 76 base::Callback<void (scoped_ptr<MessageContents>)> data_read_callback, 77 base::Callback<void ()> failure_callback, 78 scoped_refptr<base::MessageLoopProxy> main_message_loop); 79 80 // DelegateSimpleThread implementation. Reads data from the pipe in a loop 81 // until either we are told to quit or a read fails. 82 virtual void Run() OVERRIDE; 83 84 private: 85 int pipe_; 86 base::Callback<void (scoped_ptr<MessageContents>)> data_read_callback_; 87 base::Callback<void ()> failure_callback_; 88 scoped_refptr<base::MessageLoopProxy> main_message_loop_; 89 90 DISALLOW_COPY_AND_ASSIGN(ReaderThreadRunner); 91 }; 92 93 Channel::ChannelImpl::ReaderThreadRunner::ReaderThreadRunner( 94 int pipe, 95 base::Callback<void (scoped_ptr<MessageContents>)> data_read_callback, 96 base::Callback<void ()> failure_callback, 97 scoped_refptr<base::MessageLoopProxy> main_message_loop) 98 : pipe_(pipe), 99 data_read_callback_(data_read_callback), 100 failure_callback_(failure_callback), 101 main_message_loop_(main_message_loop) { 102 } 103 104 void Channel::ChannelImpl::ReaderThreadRunner::Run() { 105 while (true) { 106 scoped_ptr<MessageContents> msg_contents(new MessageContents); 107 bool success = ReadDataOnReaderThread(pipe_, msg_contents.get()); 108 if (success) { 109 main_message_loop_->PostTask(FROM_HERE, 110 base::Bind(data_read_callback_, base::Passed(&msg_contents))); 111 } else { 112 main_message_loop_->PostTask(FROM_HERE, failure_callback_); 113 // Because the read failed, we know we're going to quit. Don't bother 114 // trying to read again. 115 return; 116 } 117 } 118 } 119 120 Channel::ChannelImpl::ChannelImpl(const IPC::ChannelHandle& channel_handle, 121 Mode mode, 122 Listener* listener) 123 : ChannelReader(listener), 124 mode_(mode), 125 waiting_connect_(true), 126 pipe_(-1), 127 pipe_name_(channel_handle.name), 128 weak_ptr_factory_(this) { 129 if (!CreatePipe(channel_handle)) { 130 // The pipe may have been closed already. 131 const char *modestr = (mode_ & MODE_SERVER_FLAG) ? "server" : "client"; 132 LOG(WARNING) << "Unable to create pipe named \"" << channel_handle.name 133 << "\" in " << modestr << " mode"; 134 } 135 } 136 137 Channel::ChannelImpl::~ChannelImpl() { 138 Close(); 139 } 140 141 bool Channel::ChannelImpl::Connect() { 142 if (pipe_ == -1) { 143 DLOG(INFO) << "Channel creation failed: " << pipe_name_; 144 return false; 145 } 146 147 // Note that Connect is called on the "Channel" thread (i.e., the same thread 148 // where Channel::Send will be called, and the same thread that should receive 149 // messages). The constructor might be invoked on another thread (see 150 // ChannelProxy for an example of that). Therefore, we must wait until Connect 151 // is called to decide which MessageLoopProxy to pass to ReaderThreadRunner. 152 reader_thread_runner_.reset( 153 new ReaderThreadRunner( 154 pipe_, 155 base::Bind(&Channel::ChannelImpl::DidRecvMsg, 156 weak_ptr_factory_.GetWeakPtr()), 157 base::Bind(&Channel::ChannelImpl::ReadDidFail, 158 weak_ptr_factory_.GetWeakPtr()), 159 base::MessageLoopProxy::current())); 160 reader_thread_.reset( 161 new base::DelegateSimpleThread(reader_thread_runner_.get(), 162 "ipc_channel_nacl reader thread")); 163 reader_thread_->Start(); 164 waiting_connect_ = false; 165 // If there were any messages queued before connection, send them. 166 ProcessOutgoingMessages(); 167 return true; 168 } 169 170 void Channel::ChannelImpl::Close() { 171 // For now, we assume that at shutdown, the reader thread will be woken with 172 // a failure (see NaClIPCAdapter::BlockingRead and CloseChannel). Or... we 173 // might simply be killed with no chance to clean up anyway :-). 174 // If untrusted code tries to close the channel prior to shutdown, it's likely 175 // to hang. 176 // TODO(dmichael): Can we do anything smarter here to make sure the reader 177 // thread wakes up and quits? 178 reader_thread_->Join(); 179 close(pipe_); 180 pipe_ = -1; 181 reader_thread_runner_.reset(); 182 reader_thread_.reset(); 183 read_queue_.clear(); 184 output_queue_.clear(); 185 } 186 187 bool Channel::ChannelImpl::Send(Message* message) { 188 DVLOG(2) << "sending message @" << message << " on channel @" << this 189 << " with type " << message->type(); 190 scoped_ptr<Message> message_ptr(message); 191 192 #ifdef IPC_MESSAGE_LOG_ENABLED 193 Logging::GetInstance()->OnSendMessage(message_ptr.get(), ""); 194 #endif // IPC_MESSAGE_LOG_ENABLED 195 196 message->TraceMessageBegin(); 197 output_queue_.push_back(linked_ptr<Message>(message_ptr.release())); 198 if (!waiting_connect_) 199 return ProcessOutgoingMessages(); 200 201 return true; 202 } 203 204 void Channel::ChannelImpl::DidRecvMsg(scoped_ptr<MessageContents> contents) { 205 // Close sets the pipe to -1. It's possible we'll get a buffer sent to us from 206 // the reader thread after Close is called. If so, we ignore it. 207 if (pipe_ == -1) 208 return; 209 210 linked_ptr<std::vector<char> > data(new std::vector<char>); 211 data->swap(contents->data); 212 read_queue_.push_back(data); 213 214 input_fds_.insert(input_fds_.end(), 215 contents->fds.begin(), contents->fds.end()); 216 contents->fds.clear(); 217 218 // In POSIX, we would be told when there are bytes to read by implementing 219 // OnFileCanReadWithoutBlocking in MessageLoopForIO::Watcher. In NaCl, we 220 // instead know at this point because the reader thread posted some data to 221 // us. 222 ProcessIncomingMessages(); 223 } 224 225 void Channel::ChannelImpl::ReadDidFail() { 226 Close(); 227 } 228 229 bool Channel::ChannelImpl::CreatePipe( 230 const IPC::ChannelHandle& channel_handle) { 231 DCHECK(pipe_ == -1); 232 233 // There's one possible case in NaCl: 234 // 1) It's a channel wrapping a pipe that is given to us. 235 // We don't support these: 236 // 2) It's for a named channel. 237 // 3) It's for a client that we implement ourself. 238 // 4) It's the initial IPC channel. 239 240 if (channel_handle.socket.fd == -1) { 241 NOTIMPLEMENTED(); 242 return false; 243 } 244 pipe_ = channel_handle.socket.fd; 245 return true; 246 } 247 248 bool Channel::ChannelImpl::ProcessOutgoingMessages() { 249 DCHECK(!waiting_connect_); // Why are we trying to send messages if there's 250 // no connection? 251 if (output_queue_.empty()) 252 return true; 253 254 if (pipe_ == -1) 255 return false; 256 257 // Write out all the messages. The trusted implementation is guaranteed to not 258 // block. See NaClIPCAdapter::Send for the implementation of imc_sendmsg. 259 while (!output_queue_.empty()) { 260 linked_ptr<Message> msg = output_queue_.front(); 261 output_queue_.pop_front(); 262 263 int fds[FileDescriptorSet::kMaxDescriptorsPerMessage]; 264 const size_t num_fds = msg->file_descriptor_set()->size(); 265 DCHECK(num_fds <= FileDescriptorSet::kMaxDescriptorsPerMessage); 266 msg->file_descriptor_set()->GetDescriptors(fds); 267 268 NaClAbiNaClImcMsgIoVec iov = { 269 const_cast<void*>(msg->data()), msg->size() 270 }; 271 NaClAbiNaClImcMsgHdr msgh = { &iov, 1, fds, num_fds }; 272 ssize_t bytes_written = imc_sendmsg(pipe_, &msgh, 0); 273 274 DCHECK(bytes_written); // The trusted side shouldn't return 0. 275 if (bytes_written < 0) { 276 // The trusted side should only ever give us an error of EPIPE. We 277 // should never be interrupted, nor should we get EAGAIN. 278 DCHECK(errno == EPIPE); 279 Close(); 280 PLOG(ERROR) << "pipe_ error on " 281 << pipe_ 282 << " Currently writing message of size: " 283 << msg->size(); 284 return false; 285 } else { 286 msg->file_descriptor_set()->CommitAll(); 287 } 288 289 // Message sent OK! 290 DVLOG(2) << "sent message @" << msg.get() << " with type " << msg->type() 291 << " on fd " << pipe_; 292 } 293 return true; 294 } 295 296 Channel::ChannelImpl::ReadState Channel::ChannelImpl::ReadData( 297 char* buffer, 298 int buffer_len, 299 int* bytes_read) { 300 *bytes_read = 0; 301 if (pipe_ == -1) 302 return READ_FAILED; 303 if (read_queue_.empty()) 304 return READ_PENDING; 305 while (!read_queue_.empty() && *bytes_read < buffer_len) { 306 linked_ptr<std::vector<char> > vec(read_queue_.front()); 307 size_t bytes_to_read = buffer_len - *bytes_read; 308 if (vec->size() <= bytes_to_read) { 309 // We can read and discard the entire vector. 310 std::copy(vec->begin(), vec->end(), buffer + *bytes_read); 311 *bytes_read += vec->size(); 312 read_queue_.pop_front(); 313 } else { 314 // Read all the bytes we can and discard them from the front of the 315 // vector. (This can be slowish, since erase has to move the back of the 316 // vector to the front, but it's hopefully a temporary hack and it keeps 317 // the code simple). 318 std::copy(vec->begin(), vec->begin() + bytes_to_read, 319 buffer + *bytes_read); 320 vec->erase(vec->begin(), vec->begin() + bytes_to_read); 321 *bytes_read += bytes_to_read; 322 } 323 } 324 return READ_SUCCEEDED; 325 } 326 327 bool Channel::ChannelImpl::WillDispatchInputMessage(Message* msg) { 328 uint16 header_fds = msg->header()->num_fds; 329 CHECK(header_fds == input_fds_.size()); 330 if (header_fds == 0) 331 return true; // Nothing to do. 332 333 // The shenaniganery below with &foo.front() requires input_fds_ to have 334 // contiguous underlying storage (such as a simple array or a std::vector). 335 // This is why the header warns not to make input_fds_ a deque<>. 336 msg->file_descriptor_set()->SetDescriptors(&input_fds_.front(), 337 header_fds); 338 input_fds_.clear(); 339 return true; 340 } 341 342 bool Channel::ChannelImpl::DidEmptyInputBuffers() { 343 // When the input data buffer is empty, the fds should be too. 344 return input_fds_.empty(); 345 } 346 347 void Channel::ChannelImpl::HandleHelloMessage(const Message& msg) { 348 // The trusted side IPC::Channel should handle the "hello" handshake; we 349 // should not receive the "Hello" message. 350 NOTREACHED(); 351 } 352 353 //------------------------------------------------------------------------------ 354 // Channel's methods simply call through to ChannelImpl. 355 356 Channel::Channel(const IPC::ChannelHandle& channel_handle, 357 Mode mode, 358 Listener* listener) 359 : channel_impl_(new ChannelImpl(channel_handle, mode, listener)) { 360 } 361 362 Channel::~Channel() { 363 delete channel_impl_; 364 } 365 366 bool Channel::Connect() { 367 return channel_impl_->Connect(); 368 } 369 370 void Channel::Close() { 371 channel_impl_->Close(); 372 } 373 374 base::ProcessId Channel::peer_pid() const { 375 // This shouldn't actually get used in the untrusted side of the proxy, and we 376 // don't have the real pid anyway. 377 return -1; 378 } 379 380 bool Channel::Send(Message* message) { 381 return channel_impl_->Send(message); 382 } 383 384 } // namespace IPC 385