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 "net/quic/quic_http_stream.h"
6
7#include "base/callback_helpers.h"
8#include "base/metrics/histogram.h"
9#include "base/strings/stringprintf.h"
10#include "net/base/io_buffer.h"
11#include "net/base/net_errors.h"
12#include "net/http/http_response_headers.h"
13#include "net/http/http_util.h"
14#include "net/quic/quic_client_session.h"
15#include "net/quic/quic_http_utils.h"
16#include "net/quic/quic_reliable_client_stream.h"
17#include "net/quic/quic_utils.h"
18#include "net/socket/next_proto.h"
19#include "net/spdy/spdy_frame_builder.h"
20#include "net/spdy/spdy_framer.h"
21#include "net/spdy/spdy_http_utils.h"
22#include "net/ssl/ssl_info.h"
23
24namespace net {
25
26static const size_t kHeaderBufInitialSize = 4096;
27
28QuicHttpStream::QuicHttpStream(const base::WeakPtr<QuicClientSession>& session)
29    : next_state_(STATE_NONE),
30      session_(session),
31      session_error_(OK),
32      was_handshake_confirmed_(session->IsCryptoHandshakeConfirmed()),
33      stream_(NULL),
34      request_info_(NULL),
35      request_body_stream_(NULL),
36      priority_(MINIMUM_PRIORITY),
37      response_info_(NULL),
38      response_status_(OK),
39      response_headers_received_(false),
40      read_buf_(new GrowableIOBuffer()),
41      closed_stream_received_bytes_(0),
42      user_buffer_len_(0),
43      weak_factory_(this) {
44  DCHECK(session_);
45  session_->AddObserver(this);
46}
47
48QuicHttpStream::~QuicHttpStream() {
49  Close(false);
50  if (session_)
51    session_->RemoveObserver(this);
52}
53
54int QuicHttpStream::InitializeStream(const HttpRequestInfo* request_info,
55                                     RequestPriority priority,
56                                     const BoundNetLog& stream_net_log,
57                                     const CompletionCallback& callback) {
58  DCHECK(!stream_);
59  if (!session_)
60    return was_handshake_confirmed_ ? ERR_CONNECTION_CLOSED :
61        ERR_QUIC_HANDSHAKE_FAILED;
62
63  if (request_info->url.SchemeIsSecure()) {
64    SSLInfo ssl_info;
65    bool secure_session =
66        session_->GetSSLInfo(&ssl_info) && ssl_info.cert.get();
67    UMA_HISTOGRAM_BOOLEAN("Net.QuicSession.SecureResourceSecureSession",
68                          secure_session);
69    if (!secure_session)
70      return ERR_REQUEST_FOR_SECURE_RESOURCE_OVER_INSECURE_QUIC;
71  }
72
73  stream_net_log_ = stream_net_log;
74  request_info_ = request_info;
75  request_time_ = base::Time::Now();
76  priority_ = priority;
77
78  int rv = stream_request_.StartRequest(
79      session_, &stream_, base::Bind(&QuicHttpStream::OnStreamReady,
80                                     weak_factory_.GetWeakPtr()));
81  if (rv == ERR_IO_PENDING) {
82    callback_ = callback;
83  } else if (rv == OK) {
84    stream_->SetDelegate(this);
85  } else if (!was_handshake_confirmed_) {
86    rv = ERR_QUIC_HANDSHAKE_FAILED;
87  }
88
89  return rv;
90}
91
92void QuicHttpStream::OnStreamReady(int rv) {
93  DCHECK(rv == OK || !stream_);
94  if (rv == OK) {
95    stream_->SetDelegate(this);
96  } else if (!was_handshake_confirmed_) {
97    rv = ERR_QUIC_HANDSHAKE_FAILED;
98  }
99
100  ResetAndReturn(&callback_).Run(rv);
101}
102
103int QuicHttpStream::SendRequest(const HttpRequestHeaders& request_headers,
104                                HttpResponseInfo* response,
105                                const CompletionCallback& callback) {
106  CHECK(!request_body_stream_);
107  CHECK(!response_info_);
108  CHECK(!callback.is_null());
109  CHECK(response);
110
111   if (!stream_) {
112    return ERR_CONNECTION_CLOSED;
113  }
114
115  QuicPriority priority = ConvertRequestPriorityToQuicPriority(priority_);
116  stream_->set_priority(priority);
117  // Store the serialized request headers.
118  CreateSpdyHeadersFromHttpRequest(*request_info_, request_headers,
119                                   SPDY3, /*direct=*/true, &request_headers_);
120
121  // Store the request body.
122  request_body_stream_ = request_info_->upload_data_stream;
123  if (request_body_stream_) {
124    // TODO(rch): Can we be more precise about when to allocate
125    // raw_request_body_buf_. Removed the following check. DoReadRequestBody()
126    // was being called even if we didn't yet allocate raw_request_body_buf_.
127    //   && (request_body_stream_->size() ||
128    //       request_body_stream_->is_chunked()))
129    // Use 10 packets as the body buffer size to give enough space to
130    // help ensure we don't often send out partial packets.
131    raw_request_body_buf_ = new IOBufferWithSize(10 * kMaxPacketSize);
132    // The request body buffer is empty at first.
133    request_body_buf_ = new DrainableIOBuffer(raw_request_body_buf_.get(), 0);
134  }
135
136  // Store the response info.
137  response_info_ = response;
138
139  next_state_ = STATE_SEND_HEADERS;
140  int rv = DoLoop(OK);
141  if (rv == ERR_IO_PENDING)
142    callback_ = callback;
143
144  return rv > 0 ? OK : rv;
145}
146
147UploadProgress QuicHttpStream::GetUploadProgress() const {
148  if (!request_body_stream_)
149    return UploadProgress();
150
151  return UploadProgress(request_body_stream_->position(),
152                        request_body_stream_->size());
153}
154
155int QuicHttpStream::ReadResponseHeaders(const CompletionCallback& callback) {
156  CHECK(!callback.is_null());
157
158  if (stream_ == NULL)
159    return response_status_;
160
161  // Check if we already have the response headers. If so, return synchronously.
162  if (response_headers_received_)
163    return OK;
164
165  // Still waiting for the response, return IO_PENDING.
166  CHECK(callback_.is_null());
167  callback_ = callback;
168  return ERR_IO_PENDING;
169}
170
171int QuicHttpStream::ReadResponseBody(
172    IOBuffer* buf, int buf_len, const CompletionCallback& callback) {
173  CHECK(buf);
174  CHECK(buf_len);
175  CHECK(!callback.is_null());
176
177  // If we have data buffered, complete the IO immediately.
178  if (!response_body_.empty()) {
179    int bytes_read = 0;
180    while (!response_body_.empty() && buf_len > 0) {
181      scoped_refptr<IOBufferWithSize> data = response_body_.front();
182      const int bytes_to_copy = std::min(buf_len, data->size());
183      memcpy(&(buf->data()[bytes_read]), data->data(), bytes_to_copy);
184      buf_len -= bytes_to_copy;
185      if (bytes_to_copy == data->size()) {
186        response_body_.pop_front();
187      } else {
188        const int bytes_remaining = data->size() - bytes_to_copy;
189        IOBufferWithSize* new_buffer = new IOBufferWithSize(bytes_remaining);
190        memcpy(new_buffer->data(), &(data->data()[bytes_to_copy]),
191               bytes_remaining);
192        response_body_.pop_front();
193        response_body_.push_front(make_scoped_refptr(new_buffer));
194      }
195      bytes_read += bytes_to_copy;
196    }
197    return bytes_read;
198  }
199
200  if (!stream_) {
201    // If the stream is already closed, there is no body to read.
202    return response_status_;
203  }
204
205  CHECK(callback_.is_null());
206  CHECK(!user_buffer_.get());
207  CHECK_EQ(0, user_buffer_len_);
208
209  callback_ = callback;
210  user_buffer_ = buf;
211  user_buffer_len_ = buf_len;
212  return ERR_IO_PENDING;
213}
214
215void QuicHttpStream::Close(bool not_reusable) {
216  // Note: the not_reusable flag has no meaning for SPDY streams.
217  if (stream_) {
218    closed_stream_received_bytes_ = stream_->stream_bytes_read();
219    stream_->SetDelegate(NULL);
220    stream_->Reset(QUIC_STREAM_CANCELLED);
221    stream_ = NULL;
222    response_status_ = was_handshake_confirmed_ ?
223        ERR_CONNECTION_CLOSED : ERR_QUIC_HANDSHAKE_FAILED;
224  }
225}
226
227HttpStream* QuicHttpStream::RenewStreamForAuth() {
228  return NULL;
229}
230
231bool QuicHttpStream::IsResponseBodyComplete() const {
232  return next_state_ == STATE_OPEN && !stream_;
233}
234
235bool QuicHttpStream::CanFindEndOfResponse() const {
236  return true;
237}
238
239bool QuicHttpStream::IsConnectionReused() const {
240  // TODO(rch): do something smarter here.
241  return stream_ && stream_->id() > 1;
242}
243
244void QuicHttpStream::SetConnectionReused() {
245  // QUIC doesn't need an indicator here.
246}
247
248bool QuicHttpStream::IsConnectionReusable() const {
249  // QUIC streams aren't considered reusable.
250  return false;
251}
252
253int64 QuicHttpStream::GetTotalReceivedBytes() const {
254  if (stream_) {
255    return stream_->stream_bytes_read();
256  }
257
258  return closed_stream_received_bytes_;
259}
260
261bool QuicHttpStream::GetLoadTimingInfo(LoadTimingInfo* load_timing_info) const {
262  // TODO(mmenke):  Figure out what to do here.
263  return true;
264}
265
266void QuicHttpStream::GetSSLInfo(SSLInfo* ssl_info) {
267  DCHECK(stream_);
268  stream_->GetSSLInfo(ssl_info);
269}
270
271void QuicHttpStream::GetSSLCertRequestInfo(
272    SSLCertRequestInfo* cert_request_info) {
273  DCHECK(stream_);
274  NOTIMPLEMENTED();
275}
276
277bool QuicHttpStream::IsSpdyHttpStream() const {
278  return false;
279}
280
281void QuicHttpStream::Drain(HttpNetworkSession* session) {
282  Close(false);
283  delete this;
284}
285
286void QuicHttpStream::SetPriority(RequestPriority priority) {
287  priority_ = priority;
288}
289
290int QuicHttpStream::OnDataReceived(const char* data, int length) {
291  DCHECK_NE(0, length);
292  // Are we still reading the response headers.
293  if (!response_headers_received_) {
294    // Grow the read buffer if necessary.
295    if (read_buf_->RemainingCapacity() < length) {
296      size_t additional_capacity = length - read_buf_->RemainingCapacity();
297      if (additional_capacity < kHeaderBufInitialSize)
298        additional_capacity = kHeaderBufInitialSize;
299      read_buf_->SetCapacity(read_buf_->capacity() + additional_capacity);
300    }
301    memcpy(read_buf_->data(), data, length);
302    read_buf_->set_offset(read_buf_->offset() + length);
303    int rv = ParseResponseHeaders();
304    if (rv != ERR_IO_PENDING && !callback_.is_null()) {
305      DoCallback(rv);
306    }
307    return OK;
308  }
309
310  if (callback_.is_null()) {
311    BufferResponseBody(data, length);
312    return OK;
313  }
314
315  if (length <= user_buffer_len_) {
316    memcpy(user_buffer_->data(), data, length);
317  } else {
318    memcpy(user_buffer_->data(), data, user_buffer_len_);
319    int delta = length - user_buffer_len_;
320    BufferResponseBody(data + user_buffer_len_, delta);
321    length = user_buffer_len_;
322  }
323
324  user_buffer_ = NULL;
325  user_buffer_len_ = 0;
326  DoCallback(length);
327  return OK;
328}
329
330void QuicHttpStream::OnClose(QuicErrorCode error) {
331  if (error != QUIC_NO_ERROR) {
332    response_status_ = was_handshake_confirmed_ ?
333        ERR_QUIC_PROTOCOL_ERROR : ERR_QUIC_HANDSHAKE_FAILED;
334  } else if (!response_headers_received_) {
335    response_status_ = ERR_ABORTED;
336  }
337
338  closed_stream_received_bytes_ = stream_->stream_bytes_read();
339  stream_ = NULL;
340  if (!callback_.is_null())
341    DoCallback(response_status_);
342}
343
344void QuicHttpStream::OnError(int error) {
345  stream_ = NULL;
346  response_status_ = was_handshake_confirmed_ ?
347      error : ERR_QUIC_HANDSHAKE_FAILED;
348  if (!callback_.is_null())
349    DoCallback(response_status_);
350}
351
352bool QuicHttpStream::HasSendHeadersComplete() {
353  return next_state_ > STATE_SEND_HEADERS_COMPLETE;
354}
355
356void QuicHttpStream::OnCryptoHandshakeConfirmed() {
357  was_handshake_confirmed_ = true;
358}
359
360void QuicHttpStream::OnSessionClosed(int error) {
361  Close(false);
362  session_error_ = error;
363  session_.reset();
364}
365
366void QuicHttpStream::OnIOComplete(int rv) {
367  rv = DoLoop(rv);
368
369  if (rv != ERR_IO_PENDING && !callback_.is_null()) {
370    DoCallback(rv);
371  }
372}
373
374void QuicHttpStream::DoCallback(int rv) {
375  CHECK_NE(rv, ERR_IO_PENDING);
376  CHECK(!callback_.is_null());
377
378  // The client callback can do anything, including destroying this class,
379  // so any pending callback must be issued after everything else is done.
380  base::ResetAndReturn(&callback_).Run(rv);
381}
382
383int QuicHttpStream::DoLoop(int rv) {
384  do {
385    State state = next_state_;
386    next_state_ = STATE_NONE;
387    switch (state) {
388      case STATE_SEND_HEADERS:
389        CHECK_EQ(OK, rv);
390        rv = DoSendHeaders();
391        break;
392      case STATE_SEND_HEADERS_COMPLETE:
393        rv = DoSendHeadersComplete(rv);
394        break;
395      case STATE_READ_REQUEST_BODY:
396        CHECK_EQ(OK, rv);
397        rv = DoReadRequestBody();
398        break;
399      case STATE_READ_REQUEST_BODY_COMPLETE:
400        rv = DoReadRequestBodyComplete(rv);
401        break;
402      case STATE_SEND_BODY:
403        CHECK_EQ(OK, rv);
404        rv = DoSendBody();
405        break;
406      case STATE_SEND_BODY_COMPLETE:
407        rv = DoSendBodyComplete(rv);
408        break;
409      case STATE_OPEN:
410        CHECK_EQ(OK, rv);
411        break;
412      default:
413        NOTREACHED() << "next_state_: " << next_state_;
414        break;
415    }
416  } while (next_state_ != STATE_NONE && next_state_ != STATE_OPEN &&
417           rv != ERR_IO_PENDING);
418
419  return rv;
420}
421
422int QuicHttpStream::DoSendHeaders() {
423  if (!stream_)
424    return ERR_UNEXPECTED;
425
426  // Log the actual request with the URL Request's net log.
427  stream_net_log_.AddEvent(
428      NetLog::TYPE_HTTP_TRANSACTION_QUIC_SEND_REQUEST_HEADERS,
429      base::Bind(&QuicRequestNetLogCallback, stream_->id(), &request_headers_,
430                 priority_));
431  // Also log to the QuicSession's net log.
432  stream_->net_log().AddEvent(
433      NetLog::TYPE_QUIC_HTTP_STREAM_SEND_REQUEST_HEADERS,
434      base::Bind(&QuicRequestNetLogCallback, stream_->id(), &request_headers_,
435                 priority_));
436
437  bool has_upload_data = request_body_stream_ != NULL;
438
439  next_state_ = STATE_SEND_HEADERS_COMPLETE;
440  int rv = stream_->WriteHeaders(request_headers_, !has_upload_data, NULL);
441  request_headers_.clear();
442  return rv;
443}
444
445int QuicHttpStream::DoSendHeadersComplete(int rv) {
446  if (rv < 0)
447    return rv;
448
449  next_state_ = request_body_stream_ ?
450      STATE_READ_REQUEST_BODY : STATE_OPEN;
451
452  return OK;
453}
454
455int QuicHttpStream::DoReadRequestBody() {
456  next_state_ = STATE_READ_REQUEST_BODY_COMPLETE;
457  return request_body_stream_->Read(
458      raw_request_body_buf_.get(),
459      raw_request_body_buf_->size(),
460      base::Bind(&QuicHttpStream::OnIOComplete, weak_factory_.GetWeakPtr()));
461}
462
463int QuicHttpStream::DoReadRequestBodyComplete(int rv) {
464  // |rv| is the result of read from the request body from the last call to
465  // DoSendBody().
466  if (rv < 0)
467    return rv;
468
469  request_body_buf_ = new DrainableIOBuffer(raw_request_body_buf_.get(), rv);
470  if (rv == 0) {  // Reached the end.
471    DCHECK(request_body_stream_->IsEOF());
472  }
473
474  next_state_ = STATE_SEND_BODY;
475  return OK;
476}
477
478int QuicHttpStream::DoSendBody() {
479  if (!stream_)
480    return ERR_UNEXPECTED;
481
482  CHECK(request_body_stream_);
483  CHECK(request_body_buf_.get());
484  const bool eof = request_body_stream_->IsEOF();
485  int len = request_body_buf_->BytesRemaining();
486  if (len > 0 || eof) {
487    next_state_ = STATE_SEND_BODY_COMPLETE;
488    base::StringPiece data(request_body_buf_->data(), len);
489    return stream_->WriteStreamData(
490        data, eof,
491        base::Bind(&QuicHttpStream::OnIOComplete, weak_factory_.GetWeakPtr()));
492  }
493
494  next_state_ = STATE_OPEN;
495  return OK;
496}
497
498int QuicHttpStream::DoSendBodyComplete(int rv) {
499  if (rv < 0)
500    return rv;
501
502  request_body_buf_->DidConsume(request_body_buf_->BytesRemaining());
503
504  if (!request_body_stream_->IsEOF()) {
505    next_state_ = STATE_READ_REQUEST_BODY;
506    return OK;
507  }
508
509  next_state_ = STATE_OPEN;
510  return OK;
511}
512
513int QuicHttpStream::ParseResponseHeaders() {
514  size_t read_buf_len = static_cast<size_t>(read_buf_->offset());
515  SpdyFramer framer(SPDY3);
516  SpdyHeaderBlock headers;
517  char* data = read_buf_->StartOfBuffer();
518  size_t len = framer.ParseHeaderBlockInBuffer(data, read_buf_->offset(),
519                                               &headers);
520
521  if (len == 0) {
522    return ERR_IO_PENDING;
523  }
524
525  // Save the remaining received data.
526  size_t delta = read_buf_len - len;
527  if (delta > 0) {
528    BufferResponseBody(data + len, delta);
529  }
530
531  // The URLRequest logs these headers, so only log to the QuicSession's
532  // net log.
533  stream_->net_log().AddEvent(
534      NetLog::TYPE_QUIC_HTTP_STREAM_READ_RESPONSE_HEADERS,
535      base::Bind(&SpdyHeaderBlockNetLogCallback, &headers));
536
537  if (!SpdyHeadersToHttpResponse(headers, SPDY3, response_info_)) {
538    DLOG(WARNING) << "Invalid headers";
539    return ERR_QUIC_PROTOCOL_ERROR;
540  }
541  // Put the peer's IP address and port into the response.
542  IPEndPoint address = stream_->GetPeerAddress();
543  response_info_->socket_address = HostPortPair::FromIPEndPoint(address);
544  response_info_->connection_info =
545      HttpResponseInfo::CONNECTION_INFO_QUIC1_SPDY3;
546  response_info_->vary_data
547      .Init(*request_info_, *response_info_->headers.get());
548  response_info_->was_npn_negotiated = true;
549  response_info_->npn_negotiated_protocol = "quic/1+spdy/3";
550  response_info_->response_time = base::Time::Now();
551  response_info_->request_time = request_time_;
552  response_headers_received_ = true;
553
554  return OK;
555}
556
557void QuicHttpStream::BufferResponseBody(const char* data, int length) {
558  if (length == 0)
559    return;
560  IOBufferWithSize* io_buffer = new IOBufferWithSize(length);
561  memcpy(io_buffer->data(), data, length);
562  response_body_.push_back(make_scoped_refptr(io_buffer));
563}
564
565}  // namespace net
566