1 // Copyright 2013 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/cast_channel/cast_socket.h"
6
7 #include <stdlib.h>
8 #include <string.h>
9
10 #include "base/bind.h"
11 #include "base/callback_helpers.h"
12 #include "base/lazy_instance.h"
13 #include "base/numerics/safe_conversions.h"
14 #include "base/strings/string_number_conversions.h"
15 #include "base/sys_byteorder.h"
16 #include "chrome/browser/extensions/api/cast_channel/cast_auth_util.h"
17 #include "chrome/browser/extensions/api/cast_channel/cast_channel.pb.h"
18 #include "chrome/browser/extensions/api/cast_channel/cast_message_util.h"
19 #include "net/base/address_list.h"
20 #include "net/base/host_port_pair.h"
21 #include "net/base/net_errors.h"
22 #include "net/base/net_util.h"
23 #include "net/cert/cert_verifier.h"
24 #include "net/cert/x509_certificate.h"
25 #include "net/http/transport_security_state.h"
26 #include "net/socket/client_socket_factory.h"
27 #include "net/socket/client_socket_handle.h"
28 #include "net/socket/ssl_client_socket.h"
29 #include "net/socket/stream_socket.h"
30 #include "net/socket/tcp_client_socket.h"
31 #include "net/ssl/ssl_config_service.h"
32 #include "net/ssl/ssl_info.h"
33
34 // Assumes |ip_endpoint_| of type net::IPEndPoint and |channel_auth_| of enum
35 // type ChannelAuthType are available in the current scope.
36 #define VLOG_WITH_CONNECTION(level) VLOG(level) << "[" << \
37 ip_endpoint_.ToString() << ", auth=" << channel_auth_ << "] "
38
39 namespace {
40
41 // The default keepalive delay. On Linux, keepalives probes will be sent after
42 // the socket is idle for this length of time, and the socket will be closed
43 // after 9 failed probes. So the total idle time before close is 10 *
44 // kTcpKeepAliveDelaySecs.
45 const int kTcpKeepAliveDelaySecs = 10;
46
47 } // namespace
48
49 namespace extensions {
50
51 static base::LazyInstance<BrowserContextKeyedAPIFactory<
52 ApiResourceManager<api::cast_channel::CastSocket> > > g_factory =
53 LAZY_INSTANCE_INITIALIZER;
54
55 // static
56 template <>
57 BrowserContextKeyedAPIFactory<
58 ApiResourceManager<api::cast_channel::CastSocket> >*
GetFactoryInstance()59 ApiResourceManager<api::cast_channel::CastSocket>::GetFactoryInstance() {
60 return g_factory.Pointer();
61 }
62
63 namespace api {
64 namespace cast_channel {
65
CastSocket(const std::string & owner_extension_id,const net::IPEndPoint & ip_endpoint,ChannelAuthType channel_auth,CastSocket::Delegate * delegate,net::NetLog * net_log)66 CastSocket::CastSocket(const std::string& owner_extension_id,
67 const net::IPEndPoint& ip_endpoint,
68 ChannelAuthType channel_auth,
69 CastSocket::Delegate* delegate,
70 net::NetLog* net_log) :
71 ApiResource(owner_extension_id),
72 channel_id_(0),
73 ip_endpoint_(ip_endpoint),
74 channel_auth_(channel_auth),
75 delegate_(delegate),
76 current_message_size_(0),
77 current_message_(new CastMessage()),
78 net_log_(net_log),
79 connect_state_(CONN_STATE_NONE),
80 write_state_(WRITE_STATE_NONE),
81 read_state_(READ_STATE_NONE),
82 error_state_(CHANNEL_ERROR_NONE),
83 ready_state_(READY_STATE_NONE) {
84 DCHECK(net_log_);
85 DCHECK(channel_auth_ == CHANNEL_AUTH_TYPE_SSL ||
86 channel_auth_ == CHANNEL_AUTH_TYPE_SSL_VERIFIED);
87 net_log_source_.type = net::NetLog::SOURCE_SOCKET;
88 net_log_source_.id = net_log_->NextID();
89
90 // Reuse these buffers for each message.
91 header_read_buffer_ = new net::GrowableIOBuffer();
92 header_read_buffer_->SetCapacity(MessageHeader::header_size());
93 body_read_buffer_ = new net::GrowableIOBuffer();
94 body_read_buffer_->SetCapacity(MessageHeader::max_message_size());
95 current_read_buffer_ = header_read_buffer_;
96 }
97
~CastSocket()98 CastSocket::~CastSocket() { }
99
ready_state() const100 ReadyState CastSocket::ready_state() const {
101 return ready_state_;
102 }
103
error_state() const104 ChannelError CastSocket::error_state() const {
105 return error_state_;
106 }
107
CreateTcpSocket()108 scoped_ptr<net::TCPClientSocket> CastSocket::CreateTcpSocket() {
109 net::AddressList addresses(ip_endpoint_);
110 return scoped_ptr<net::TCPClientSocket>(
111 new net::TCPClientSocket(addresses, net_log_, net_log_source_));
112 // Options cannot be set on the TCPClientSocket yet, because the
113 // underlying platform socket will not be created until Bind()
114 // or Connect() is called.
115 }
116
CreateSslSocket(scoped_ptr<net::StreamSocket> socket)117 scoped_ptr<net::SSLClientSocket> CastSocket::CreateSslSocket(
118 scoped_ptr<net::StreamSocket> socket) {
119 net::SSLConfig ssl_config;
120 // If a peer cert was extracted in a previous attempt to connect, then
121 // whitelist that cert.
122 if (!peer_cert_.empty()) {
123 net::SSLConfig::CertAndStatus cert_and_status;
124 cert_and_status.cert_status = net::CERT_STATUS_AUTHORITY_INVALID;
125 cert_and_status.der_cert = peer_cert_;
126 ssl_config.allowed_bad_certs.push_back(cert_and_status);
127 }
128
129 cert_verifier_.reset(net::CertVerifier::CreateDefault());
130 transport_security_state_.reset(new net::TransportSecurityState);
131 net::SSLClientSocketContext context;
132 // CertVerifier and TransportSecurityState are owned by us, not the
133 // context object.
134 context.cert_verifier = cert_verifier_.get();
135 context.transport_security_state = transport_security_state_.get();
136
137 scoped_ptr<net::ClientSocketHandle> connection(new net::ClientSocketHandle);
138 connection->SetSocket(socket.Pass());
139 net::HostPortPair host_and_port = net::HostPortPair::FromIPEndPoint(
140 ip_endpoint_);
141
142 return net::ClientSocketFactory::GetDefaultFactory()->CreateSSLClientSocket(
143 connection.Pass(), host_and_port, ssl_config, context);
144 }
145
ExtractPeerCert(std::string * cert)146 bool CastSocket::ExtractPeerCert(std::string* cert) {
147 DCHECK(cert);
148 DCHECK(peer_cert_.empty());
149 net::SSLInfo ssl_info;
150 if (!socket_->GetSSLInfo(&ssl_info) || !ssl_info.cert.get())
151 return false;
152 bool result = net::X509Certificate::GetDEREncoded(
153 ssl_info.cert->os_cert_handle(), cert);
154 if (result)
155 VLOG_WITH_CONNECTION(1) << "Successfully extracted peer certificate: "
156 << *cert;
157 return result;
158 }
159
VerifyChallengeReply()160 bool CastSocket::VerifyChallengeReply() {
161 return AuthenticateChallengeReply(*challenge_reply_, peer_cert_);
162 }
163
Connect(const net::CompletionCallback & callback)164 void CastSocket::Connect(const net::CompletionCallback& callback) {
165 DCHECK(CalledOnValidThread());
166 VLOG_WITH_CONNECTION(1) << "Connect readyState = " << ready_state_;
167 if (ready_state_ != READY_STATE_NONE) {
168 callback.Run(net::ERR_CONNECTION_FAILED);
169 return;
170 }
171 ready_state_ = READY_STATE_CONNECTING;
172 connect_callback_ = callback;
173 connect_state_ = CONN_STATE_TCP_CONNECT;
174 DoConnectLoop(net::OK);
175 }
176
PostTaskToStartConnectLoop(int result)177 void CastSocket::PostTaskToStartConnectLoop(int result) {
178 DCHECK(CalledOnValidThread());
179 base::MessageLoop::current()->PostTask(
180 FROM_HERE,
181 base::Bind(&CastSocket::DoConnectLoop, AsWeakPtr(), result));
182 }
183
184 // This method performs the state machine transitions for connection flow.
185 // There are two entry points to this method:
186 // 1. Connect method: this starts the flow
187 // 2. Callback from network operations that finish asynchronously
DoConnectLoop(int result)188 void CastSocket::DoConnectLoop(int result) {
189 // Network operations can either finish synchronously or asynchronously.
190 // This method executes the state machine transitions in a loop so that
191 // correct state transitions happen even when network operations finish
192 // synchronously.
193 int rv = result;
194 do {
195 ConnectionState state = connect_state_;
196 // Default to CONN_STATE_NONE, which breaks the processing loop if any
197 // handler fails to transition to another state to continue processing.
198 connect_state_ = CONN_STATE_NONE;
199 switch (state) {
200 case CONN_STATE_TCP_CONNECT:
201 rv = DoTcpConnect();
202 break;
203 case CONN_STATE_TCP_CONNECT_COMPLETE:
204 rv = DoTcpConnectComplete(rv);
205 break;
206 case CONN_STATE_SSL_CONNECT:
207 DCHECK_EQ(net::OK, rv);
208 rv = DoSslConnect();
209 break;
210 case CONN_STATE_SSL_CONNECT_COMPLETE:
211 rv = DoSslConnectComplete(rv);
212 break;
213 case CONN_STATE_AUTH_CHALLENGE_SEND:
214 rv = DoAuthChallengeSend();
215 break;
216 case CONN_STATE_AUTH_CHALLENGE_SEND_COMPLETE:
217 rv = DoAuthChallengeSendComplete(rv);
218 break;
219 case CONN_STATE_AUTH_CHALLENGE_REPLY_COMPLETE:
220 rv = DoAuthChallengeReplyComplete(rv);
221 break;
222 default:
223 NOTREACHED() << "BUG in connect flow. Unknown state: " << state;
224 break;
225 }
226 } while (rv != net::ERR_IO_PENDING && connect_state_ != CONN_STATE_NONE);
227 // Get out of the loop either when:
228 // a. A network operation is pending, OR
229 // b. The Do* method called did not change state
230
231 // Connect loop is finished: if there is no pending IO invoke the callback.
232 if (rv != net::ERR_IO_PENDING)
233 DoConnectCallback(rv);
234 }
235
DoTcpConnect()236 int CastSocket::DoTcpConnect() {
237 VLOG_WITH_CONNECTION(1) << "DoTcpConnect";
238 connect_state_ = CONN_STATE_TCP_CONNECT_COMPLETE;
239 tcp_socket_ = CreateTcpSocket();
240 return tcp_socket_->Connect(
241 base::Bind(&CastSocket::DoConnectLoop, AsWeakPtr()));
242 }
243
DoTcpConnectComplete(int result)244 int CastSocket::DoTcpConnectComplete(int result) {
245 VLOG_WITH_CONNECTION(1) << "DoTcpConnectComplete: " << result;
246 if (result == net::OK) {
247 // Enable TCP protocol-level keep-alive.
248 bool result = tcp_socket_->SetKeepAlive(true, kTcpKeepAliveDelaySecs);
249 LOG_IF(WARNING, !result) << "Failed to SetKeepAlive.";
250 connect_state_ = CONN_STATE_SSL_CONNECT;
251 }
252 return result;
253 }
254
DoSslConnect()255 int CastSocket::DoSslConnect() {
256 VLOG_WITH_CONNECTION(1) << "DoSslConnect";
257 connect_state_ = CONN_STATE_SSL_CONNECT_COMPLETE;
258 socket_ = CreateSslSocket(tcp_socket_.PassAs<net::StreamSocket>());
259 return socket_->Connect(
260 base::Bind(&CastSocket::DoConnectLoop, AsWeakPtr()));
261 }
262
DoSslConnectComplete(int result)263 int CastSocket::DoSslConnectComplete(int result) {
264 VLOG_WITH_CONNECTION(1) << "DoSslConnectComplete: " << result;
265 if (result == net::ERR_CERT_AUTHORITY_INVALID &&
266 peer_cert_.empty() && ExtractPeerCert(&peer_cert_)) {
267 connect_state_ = CONN_STATE_TCP_CONNECT;
268 } else if (result == net::OK &&
269 channel_auth_ == CHANNEL_AUTH_TYPE_SSL_VERIFIED) {
270 connect_state_ = CONN_STATE_AUTH_CHALLENGE_SEND;
271 }
272 return result;
273 }
274
DoAuthChallengeSend()275 int CastSocket::DoAuthChallengeSend() {
276 VLOG_WITH_CONNECTION(1) << "DoAuthChallengeSend";
277 connect_state_ = CONN_STATE_AUTH_CHALLENGE_SEND_COMPLETE;
278 CastMessage challenge_message;
279 CreateAuthChallengeMessage(&challenge_message);
280 VLOG_WITH_CONNECTION(1) << "Sending challenge: "
281 << CastMessageToString(challenge_message);
282 // Post a task to send auth challenge so that DoWriteLoop is not nested inside
283 // DoConnectLoop. This is not strictly necessary but keeps the write loop
284 // code decoupled from connect loop code.
285 base::MessageLoop::current()->PostTask(
286 FROM_HERE,
287 base::Bind(&CastSocket::SendCastMessageInternal, AsWeakPtr(),
288 challenge_message,
289 base::Bind(&CastSocket::DoConnectLoop, AsWeakPtr())));
290 // Always return IO_PENDING since the result is always asynchronous.
291 return net::ERR_IO_PENDING;
292 }
293
DoAuthChallengeSendComplete(int result)294 int CastSocket::DoAuthChallengeSendComplete(int result) {
295 VLOG_WITH_CONNECTION(1) << "DoAuthChallengeSendComplete: " << result;
296 if (result < 0)
297 return result;
298 connect_state_ = CONN_STATE_AUTH_CHALLENGE_REPLY_COMPLETE;
299 // Post a task to start read loop so that DoReadLoop is not nested inside
300 // DoConnectLoop. This is not strictly necessary but keeps the read loop
301 // code decoupled from connect loop code.
302 PostTaskToStartReadLoop();
303 // Always return IO_PENDING since the result is always asynchronous.
304 return net::ERR_IO_PENDING;
305 }
306
DoAuthChallengeReplyComplete(int result)307 int CastSocket::DoAuthChallengeReplyComplete(int result) {
308 VLOG_WITH_CONNECTION(1) << "DoAuthChallengeReplyComplete: " << result;
309 if (result < 0)
310 return result;
311 if (!VerifyChallengeReply())
312 return net::ERR_FAILED;
313 VLOG_WITH_CONNECTION(1) << "Auth challenge verification succeeded";
314 return net::OK;
315 }
316
DoConnectCallback(int result)317 void CastSocket::DoConnectCallback(int result) {
318 ready_state_ = (result == net::OK) ? READY_STATE_OPEN : READY_STATE_CLOSED;
319 error_state_ = (result == net::OK) ?
320 CHANNEL_ERROR_NONE : CHANNEL_ERROR_CONNECT_ERROR;
321 if (result == net::OK) // Start the read loop
322 PostTaskToStartReadLoop();
323 base::ResetAndReturn(&connect_callback_).Run(result);
324 }
325
Close(const net::CompletionCallback & callback)326 void CastSocket::Close(const net::CompletionCallback& callback) {
327 DCHECK(CalledOnValidThread());
328 VLOG_WITH_CONNECTION(1) << "Close ReadyState = " << ready_state_;
329 tcp_socket_.reset();
330 socket_.reset();
331 cert_verifier_.reset();
332 transport_security_state_.reset();
333 ready_state_ = READY_STATE_CLOSED;
334 callback.Run(net::OK);
335 // |callback| can delete |this|
336 }
337
SendMessage(const MessageInfo & message,const net::CompletionCallback & callback)338 void CastSocket::SendMessage(const MessageInfo& message,
339 const net::CompletionCallback& callback) {
340 DCHECK(CalledOnValidThread());
341 if (ready_state_ != READY_STATE_OPEN) {
342 callback.Run(net::ERR_FAILED);
343 return;
344 }
345 CastMessage message_proto;
346 if (!MessageInfoToCastMessage(message, &message_proto)) {
347 callback.Run(net::ERR_FAILED);
348 return;
349 }
350
351 SendCastMessageInternal(message_proto, callback);
352 }
353
SendCastMessageInternal(const CastMessage & message,const net::CompletionCallback & callback)354 void CastSocket::SendCastMessageInternal(
355 const CastMessage& message,
356 const net::CompletionCallback& callback) {
357 WriteRequest write_request(callback);
358 if (!write_request.SetContent(message)) {
359 callback.Run(net::ERR_FAILED);
360 return;
361 }
362
363 write_queue_.push(write_request);
364 if (write_state_ == WRITE_STATE_NONE) {
365 write_state_ = WRITE_STATE_WRITE;
366 DoWriteLoop(net::OK);
367 }
368 }
369
DoWriteLoop(int result)370 void CastSocket::DoWriteLoop(int result) {
371 DCHECK(CalledOnValidThread());
372 VLOG_WITH_CONNECTION(1) << "DoWriteLoop queue size: " << write_queue_.size();
373
374 if (write_queue_.empty()) {
375 write_state_ = WRITE_STATE_NONE;
376 return;
377 }
378
379 // Network operations can either finish synchronously or asynchronously.
380 // This method executes the state machine transitions in a loop so that
381 // write state transitions happen even when network operations finish
382 // synchronously.
383 int rv = result;
384 do {
385 WriteState state = write_state_;
386 write_state_ = WRITE_STATE_NONE;
387 switch (state) {
388 case WRITE_STATE_WRITE:
389 rv = DoWrite();
390 break;
391 case WRITE_STATE_WRITE_COMPLETE:
392 rv = DoWriteComplete(rv);
393 break;
394 case WRITE_STATE_DO_CALLBACK:
395 rv = DoWriteCallback();
396 break;
397 case WRITE_STATE_ERROR:
398 rv = DoWriteError(rv);
399 break;
400 default:
401 NOTREACHED() << "BUG in write flow. Unknown state: " << state;
402 break;
403 }
404 } while (!write_queue_.empty() &&
405 rv != net::ERR_IO_PENDING &&
406 write_state_ != WRITE_STATE_NONE);
407
408 // If write loop is done because the queue is empty then set write
409 // state to NONE
410 if (write_queue_.empty())
411 write_state_ = WRITE_STATE_NONE;
412
413 // Write loop is done - if the result is ERR_FAILED then close with error.
414 if (rv == net::ERR_FAILED)
415 CloseWithError(error_state_);
416 }
417
DoWrite()418 int CastSocket::DoWrite() {
419 DCHECK(!write_queue_.empty());
420 WriteRequest& request = write_queue_.front();
421
422 VLOG_WITH_CONNECTION(2) << "WriteData byte_count = "
423 << request.io_buffer->size() << " bytes_written "
424 << request.io_buffer->BytesConsumed();
425
426 write_state_ = WRITE_STATE_WRITE_COMPLETE;
427
428 return socket_->Write(
429 request.io_buffer.get(),
430 request.io_buffer->BytesRemaining(),
431 base::Bind(&CastSocket::DoWriteLoop, AsWeakPtr()));
432 }
433
DoWriteComplete(int result)434 int CastSocket::DoWriteComplete(int result) {
435 DCHECK(!write_queue_.empty());
436 if (result <= 0) { // NOTE that 0 also indicates an error
437 error_state_ = CHANNEL_ERROR_SOCKET_ERROR;
438 write_state_ = WRITE_STATE_ERROR;
439 return result == 0 ? net::ERR_FAILED : result;
440 }
441
442 // Some bytes were successfully written
443 WriteRequest& request = write_queue_.front();
444 scoped_refptr<net::DrainableIOBuffer> io_buffer = request.io_buffer;
445 io_buffer->DidConsume(result);
446 if (io_buffer->BytesRemaining() == 0) // Message fully sent
447 write_state_ = WRITE_STATE_DO_CALLBACK;
448 else
449 write_state_ = WRITE_STATE_WRITE;
450
451 return net::OK;
452 }
453
DoWriteCallback()454 int CastSocket::DoWriteCallback() {
455 DCHECK(!write_queue_.empty());
456 WriteRequest& request = write_queue_.front();
457 int bytes_consumed = request.io_buffer->BytesConsumed();
458
459 // If inside connection flow, then there should be exaclty one item in
460 // the write queue.
461 if (ready_state_ == READY_STATE_CONNECTING) {
462 write_queue_.pop();
463 DCHECK(write_queue_.empty());
464 PostTaskToStartConnectLoop(bytes_consumed);
465 } else {
466 WriteRequest& request = write_queue_.front();
467 request.callback.Run(bytes_consumed);
468 write_queue_.pop();
469 }
470 write_state_ = WRITE_STATE_WRITE;
471 return net::OK;
472 }
473
DoWriteError(int result)474 int CastSocket::DoWriteError(int result) {
475 DCHECK(!write_queue_.empty());
476 DCHECK_LT(result, 0);
477
478 // If inside connection flow, then there should be exactly one item in
479 // the write queue.
480 if (ready_state_ == READY_STATE_CONNECTING) {
481 write_queue_.pop();
482 DCHECK(write_queue_.empty());
483 PostTaskToStartConnectLoop(result);
484 // Connect loop will handle the error. Return net::OK so that write flow
485 // does not try to report error also.
486 return net::OK;
487 }
488
489 while (!write_queue_.empty()) {
490 WriteRequest& request = write_queue_.front();
491 request.callback.Run(result);
492 write_queue_.pop();
493 }
494 return net::ERR_FAILED;
495 }
496
PostTaskToStartReadLoop()497 void CastSocket::PostTaskToStartReadLoop() {
498 DCHECK(CalledOnValidThread());
499 base::MessageLoop::current()->PostTask(
500 FROM_HERE,
501 base::Bind(&CastSocket::StartReadLoop, AsWeakPtr()));
502 }
503
StartReadLoop()504 void CastSocket::StartReadLoop() {
505 // Read loop would have already been started if read state is not NONE
506 if (read_state_ == READ_STATE_NONE) {
507 read_state_ = READ_STATE_READ;
508 DoReadLoop(net::OK);
509 }
510 }
511
DoReadLoop(int result)512 void CastSocket::DoReadLoop(int result) {
513 DCHECK(CalledOnValidThread());
514 // Network operations can either finish synchronously or asynchronously.
515 // This method executes the state machine transitions in a loop so that
516 // write state transitions happen even when network operations finish
517 // synchronously.
518 int rv = result;
519 do {
520 ReadState state = read_state_;
521 read_state_ = READ_STATE_NONE;
522
523 switch (state) {
524 case READ_STATE_READ:
525 rv = DoRead();
526 break;
527 case READ_STATE_READ_COMPLETE:
528 rv = DoReadComplete(rv);
529 break;
530 case READ_STATE_DO_CALLBACK:
531 rv = DoReadCallback();
532 break;
533 case READ_STATE_ERROR:
534 rv = DoReadError(rv);
535 DCHECK_EQ(read_state_, READ_STATE_NONE);
536 break;
537 default:
538 NOTREACHED() << "BUG in read flow. Unknown state: " << state;
539 break;
540 }
541 } while (rv != net::ERR_IO_PENDING && read_state_ != READ_STATE_NONE);
542
543 if (rv == net::ERR_FAILED) {
544 if (ready_state_ == READY_STATE_CONNECTING) {
545 // Read errors during the handshake should notify the caller via
546 // the connect callback, rather than the message event delegate.
547 PostTaskToStartConnectLoop(net::ERR_FAILED);
548 } else {
549 // Connection is already established.
550 // Close and send error status via the message event delegate.
551 CloseWithError(error_state_);
552 }
553 }
554 }
555
DoRead()556 int CastSocket::DoRead() {
557 read_state_ = READ_STATE_READ_COMPLETE;
558 // Figure out whether to read header or body, and the remaining bytes.
559 uint32 num_bytes_to_read = 0;
560 if (header_read_buffer_->RemainingCapacity() > 0) {
561 current_read_buffer_ = header_read_buffer_;
562 num_bytes_to_read = header_read_buffer_->RemainingCapacity();
563 CHECK_LE(num_bytes_to_read, MessageHeader::header_size());
564 } else {
565 DCHECK_GT(current_message_size_, 0U);
566 num_bytes_to_read = current_message_size_ - body_read_buffer_->offset();
567 current_read_buffer_ = body_read_buffer_;
568 CHECK_LE(num_bytes_to_read, MessageHeader::max_message_size());
569 }
570 CHECK_GT(num_bytes_to_read, 0U);
571
572 // Read up to num_bytes_to_read into |current_read_buffer_|.
573 return socket_->Read(
574 current_read_buffer_.get(),
575 num_bytes_to_read,
576 base::Bind(&CastSocket::DoReadLoop, AsWeakPtr()));
577 }
578
DoReadComplete(int result)579 int CastSocket::DoReadComplete(int result) {
580 VLOG_WITH_CONNECTION(2) << "DoReadComplete result = " << result
581 << " header offset = "
582 << header_read_buffer_->offset()
583 << " body offset = " << body_read_buffer_->offset();
584 if (result <= 0) { // 0 means EOF: the peer closed the socket
585 VLOG_WITH_CONNECTION(1) << "Read error, peer closed the socket";
586 error_state_ = CHANNEL_ERROR_SOCKET_ERROR;
587 read_state_ = READ_STATE_ERROR;
588 return result == 0 ? net::ERR_FAILED : result;
589 }
590
591 // Some data was read. Move the offset in the current buffer forward.
592 CHECK_LE(current_read_buffer_->offset() + result,
593 current_read_buffer_->capacity());
594 current_read_buffer_->set_offset(current_read_buffer_->offset() + result);
595 read_state_ = READ_STATE_READ;
596
597 if (current_read_buffer_.get() == header_read_buffer_.get() &&
598 current_read_buffer_->RemainingCapacity() == 0) {
599 // A full header is read, process the contents.
600 if (!ProcessHeader()) {
601 error_state_ = cast_channel::CHANNEL_ERROR_INVALID_MESSAGE;
602 read_state_ = READ_STATE_ERROR;
603 }
604 } else if (current_read_buffer_.get() == body_read_buffer_.get() &&
605 static_cast<uint32>(current_read_buffer_->offset()) ==
606 current_message_size_) {
607 // Full body is read, process the contents.
608 if (ProcessBody()) {
609 read_state_ = READ_STATE_DO_CALLBACK;
610 } else {
611 error_state_ = cast_channel::CHANNEL_ERROR_INVALID_MESSAGE;
612 read_state_ = READ_STATE_ERROR;
613 }
614 }
615
616 return net::OK;
617 }
618
DoReadCallback()619 int CastSocket::DoReadCallback() {
620 read_state_ = READ_STATE_READ;
621 const CastMessage& message = *current_message_;
622 if (ready_state_ == READY_STATE_CONNECTING) {
623 if (IsAuthMessage(message)) {
624 challenge_reply_.reset(new CastMessage(message));
625 PostTaskToStartConnectLoop(net::OK);
626 return net::OK;
627 } else {
628 // Expected an auth message, got something else instead. Handle as error.
629 read_state_ = READ_STATE_ERROR;
630 return net::ERR_INVALID_RESPONSE;
631 }
632 }
633
634 MessageInfo message_info;
635 if (!CastMessageToMessageInfo(message, &message_info)) {
636 current_message_->Clear();
637 read_state_ = READ_STATE_ERROR;
638 return net::ERR_INVALID_RESPONSE;
639 }
640 delegate_->OnMessage(this, message_info);
641 current_message_->Clear();
642 return net::OK;
643 }
644
DoReadError(int result)645 int CastSocket::DoReadError(int result) {
646 DCHECK_LE(result, 0);
647 return net::ERR_FAILED;
648 }
649
ProcessHeader()650 bool CastSocket::ProcessHeader() {
651 CHECK_EQ(static_cast<uint32>(header_read_buffer_->offset()),
652 MessageHeader::header_size());
653 MessageHeader header;
654 MessageHeader::ReadFromIOBuffer(header_read_buffer_.get(), &header);
655 if (header.message_size > MessageHeader::max_message_size())
656 return false;
657
658 VLOG_WITH_CONNECTION(2) << "Parsed header { message_size: "
659 << header.message_size << " }";
660 current_message_size_ = header.message_size;
661 return true;
662 }
663
ProcessBody()664 bool CastSocket::ProcessBody() {
665 CHECK_EQ(static_cast<uint32>(body_read_buffer_->offset()),
666 current_message_size_);
667 if (!current_message_->ParseFromArray(
668 body_read_buffer_->StartOfBuffer(), current_message_size_)) {
669 return false;
670 }
671 current_message_size_ = 0;
672 header_read_buffer_->set_offset(0);
673 body_read_buffer_->set_offset(0);
674 current_read_buffer_ = header_read_buffer_;
675 return true;
676 }
677
678 // static
Serialize(const CastMessage & message_proto,std::string * message_data)679 bool CastSocket::Serialize(const CastMessage& message_proto,
680 std::string* message_data) {
681 DCHECK(message_data);
682 message_proto.SerializeToString(message_data);
683 size_t message_size = message_data->size();
684 if (message_size > MessageHeader::max_message_size()) {
685 message_data->clear();
686 return false;
687 }
688 CastSocket::MessageHeader header;
689 header.SetMessageSize(message_size);
690 header.PrependToString(message_data);
691 return true;
692 };
693
CloseWithError(ChannelError error)694 void CastSocket::CloseWithError(ChannelError error) {
695 DCHECK(CalledOnValidThread());
696 socket_.reset(NULL);
697 ready_state_ = READY_STATE_CLOSED;
698 error_state_ = error;
699 if (delegate_)
700 delegate_->OnError(this, error);
701 }
702
CastUrl() const703 std::string CastSocket::CastUrl() const {
704 return ((channel_auth_ == CHANNEL_AUTH_TYPE_SSL_VERIFIED) ?
705 "casts://" : "cast://") + ip_endpoint_.ToString();
706 }
707
CalledOnValidThread() const708 bool CastSocket::CalledOnValidThread() const {
709 return thread_checker_.CalledOnValidThread();
710 }
711
MessageHeader()712 CastSocket::MessageHeader::MessageHeader() : message_size(0) { }
713
SetMessageSize(size_t size)714 void CastSocket::MessageHeader::SetMessageSize(size_t size) {
715 DCHECK(size < static_cast<size_t>(kuint32max));
716 DCHECK(size > 0);
717 message_size = size;
718 }
719
720 // TODO(mfoltz): Investigate replacing header serialization with base::Pickle,
721 // if bit-for-bit compatible.
PrependToString(std::string * str)722 void CastSocket::MessageHeader::PrependToString(std::string* str) {
723 MessageHeader output = *this;
724 output.message_size = base::HostToNet32(message_size);
725 size_t header_size = base::checked_cast<size_t,uint32>(
726 MessageHeader::header_size());
727 scoped_ptr<char, base::FreeDeleter> char_array(
728 static_cast<char*>(malloc(header_size)));
729 memcpy(char_array.get(), &output, header_size);
730 str->insert(0, char_array.get(), header_size);
731 }
732
733 // TODO(mfoltz): Investigate replacing header deserialization with base::Pickle,
734 // if bit-for-bit compatible.
ReadFromIOBuffer(net::GrowableIOBuffer * buffer,MessageHeader * header)735 void CastSocket::MessageHeader::ReadFromIOBuffer(
736 net::GrowableIOBuffer* buffer, MessageHeader* header) {
737 uint32 message_size;
738 size_t header_size = base::checked_cast<size_t,uint32>(
739 MessageHeader::header_size());
740 memcpy(&message_size, buffer->StartOfBuffer(), header_size);
741 header->message_size = base::NetToHost32(message_size);
742 }
743
ToString()744 std::string CastSocket::MessageHeader::ToString() {
745 return "{message_size: " + base::UintToString(message_size) + "}";
746 }
747
WriteRequest(const net::CompletionCallback & callback)748 CastSocket::WriteRequest::WriteRequest(const net::CompletionCallback& callback)
749 : callback(callback) { }
750
SetContent(const CastMessage & message_proto)751 bool CastSocket::WriteRequest::SetContent(const CastMessage& message_proto) {
752 DCHECK(!io_buffer.get());
753 std::string message_data;
754 if (!Serialize(message_proto, &message_data))
755 return false;
756 io_buffer = new net::DrainableIOBuffer(new net::StringIOBuffer(message_data),
757 message_data.size());
758 return true;
759 }
760
~WriteRequest()761 CastSocket::WriteRequest::~WriteRequest() { }
762
763 } // namespace cast_channel
764 } // namespace api
765 } // namespace extensions
766
767 #undef VLOG_WITH_CONNECTION
768