// Copyright (c) 2012 The Chromium Authors. All rights reserved. // Use of this source code is governed by a BSD-style license that can be // found in the LICENSE file. #include "net/quic/quic_http_stream.h" #include "base/callback_helpers.h" #include "base/metrics/histogram_macros.h" #include "base/strings/string_split.h" #include "base/strings/stringprintf.h" #include "net/base/io_buffer.h" #include "net/base/load_flags.h" #include "net/base/net_errors.h" #include "net/http/http_response_headers.h" #include "net/http/http_util.h" #include "net/quic/quic_chromium_client_session.h" #include "net/quic/quic_chromium_client_stream.h" #include "net/quic/quic_client_promised_info.h" #include "net/quic/quic_http_utils.h" #include "net/quic/quic_utils.h" #include "net/quic/spdy_utils.h" #include "net/socket/next_proto.h" #include "net/spdy/spdy_frame_builder.h" #include "net/spdy/spdy_framer.h" #include "net/spdy/spdy_http_utils.h" #include "net/ssl/ssl_info.h" namespace net { namespace { scoped_ptr NetLogQuicPushStreamCallback( QuicStreamId stream_id, const GURL* url, NetLogCaptureMode capture_mode) { scoped_ptr dict(new base::DictionaryValue()); dict->SetInteger("stream_id", stream_id); dict->SetString("url", url->spec()); return std::move(dict); } } // namespace QuicHttpStream::QuicHttpStream( const base::WeakPtr& session) : next_state_(STATE_NONE), session_(session), session_error_(OK), was_handshake_confirmed_(session->IsCryptoHandshakeConfirmed()), stream_(nullptr), request_info_(nullptr), request_body_stream_(nullptr), priority_(MINIMUM_PRIORITY), response_info_(nullptr), response_status_(OK), response_headers_received_(false), headers_bytes_received_(0), headers_bytes_sent_(0), closed_stream_received_bytes_(0), closed_stream_sent_bytes_(0), user_buffer_len_(0), quic_connection_error_(QUIC_NO_ERROR), port_migration_detected_(false), found_promise_(false), push_handle_(nullptr), weak_factory_(this) { DCHECK(session_); session_->AddObserver(this); } QuicHttpStream::~QuicHttpStream() { Close(false); if (session_) session_->RemoveObserver(this); } bool QuicHttpStream::CheckVary(const SpdyHeaderBlock& client_request, const SpdyHeaderBlock& promise_request, const SpdyHeaderBlock& promise_response) { HttpResponseInfo promise_response_info; HttpRequestInfo promise_request_info; ConvertHeaderBlockToHttpRequestHeaders(promise_request, &promise_request_info.extra_headers); HttpRequestInfo client_request_info; ConvertHeaderBlockToHttpRequestHeaders(client_request, &client_request_info.extra_headers); if (!SpdyHeadersToHttpResponse(promise_response, HTTP2, &promise_response_info)) { DLOG(WARNING) << "Invalid headers"; return false; } HttpVaryData vary_data; if (!vary_data.Init(promise_request_info, *promise_response_info.headers.get())) { // Promise didn't contain valid vary info, so URL match was sufficient. return true; } // Now compare the client request for matching. return vary_data.MatchesRequest(client_request_info, *promise_response_info.headers.get()); } void QuicHttpStream::OnRendezvousResult(QuicSpdyStream* stream) { push_handle_ = nullptr; if (stream) { stream_ = static_cast(stream); stream_->SetDelegate(this); } // callback_ should be non-null in the case of asynchronous // rendezvous; i.e. |Try()| returned QUIC_PENDING. if (!callback_.is_null()) { if (stream) { next_state_ = STATE_OPEN; stream_net_log_.AddEvent( NetLog::TYPE_QUIC_HTTP_STREAM_ADOPTED_PUSH_STREAM, base::Bind(&NetLogQuicPushStreamCallback, stream_->id(), &request_info_->url)); session_->net_log().AddEvent( NetLog::TYPE_QUIC_HTTP_STREAM_ADOPTED_PUSH_STREAM, base::Bind(&NetLogQuicPushStreamCallback, stream_->id(), &request_info_->url)); DoCallback(OK); return; } // rendezvous has failed so proceed as with a non-push request. next_state_ = STATE_REQUEST_STREAM; OnIOComplete(OK); } } int QuicHttpStream::InitializeStream(const HttpRequestInfo* request_info, RequestPriority priority, const BoundNetLog& stream_net_log, const CompletionCallback& callback) { DCHECK(!stream_); if (!session_) return was_handshake_confirmed_ ? ERR_CONNECTION_CLOSED : ERR_QUIC_HANDSHAKE_FAILED; stream_net_log.AddEvent( NetLog::TYPE_HTTP_STREAM_REQUEST_BOUND_TO_QUIC_SESSION, session_->net_log().source().ToEventParametersCallback()); stream_net_log_ = stream_net_log; request_info_ = request_info; request_time_ = base::Time::Now(); priority_ = priority; bool success = session_->GetSSLInfo(&ssl_info_); DCHECK(success); DCHECK(ssl_info_.cert.get()); string url(request_info->url.spec()); QuicClientPromisedInfo* promised = session_->push_promise_index()->GetPromised(url); if (promised) { found_promise_ = true; stream_net_log_.AddEvent( NetLog::TYPE_QUIC_HTTP_STREAM_PUSH_PROMISE_RENDEZVOUS, base::Bind(&NetLogQuicPushStreamCallback, promised->id(), &request_info_->url)); session_->net_log().AddEvent( NetLog::TYPE_QUIC_HTTP_STREAM_PUSH_PROMISE_RENDEZVOUS, base::Bind(&NetLogQuicPushStreamCallback, promised->id(), &request_info_->url)); return OK; } next_state_ = STATE_REQUEST_STREAM; int rv = DoLoop(OK); if (rv == ERR_IO_PENDING) callback_ = callback; return rv; } int QuicHttpStream::DoStreamRequest() { int rv = stream_request_.StartRequest( session_, &stream_, base::Bind(&QuicHttpStream::OnStreamReady, weak_factory_.GetWeakPtr())); if (rv == OK) { stream_->SetDelegate(this); if (request_info_->load_flags & LOAD_DISABLE_CONNECTION_MIGRATION) { stream_->DisableConnectionMigration(); } if (response_info_) { next_state_ = STATE_SET_REQUEST_PRIORITY; } } else if (rv != ERR_IO_PENDING && !was_handshake_confirmed_) { rv = ERR_QUIC_HANDSHAKE_FAILED; } return rv; } int QuicHttpStream::DoSetRequestPriority() { // Set priority according to request and, and advance to // STATE_SEND_HEADERS. DCHECK(stream_); DCHECK(response_info_); SpdyPriority priority = ConvertRequestPriorityToQuicPriority(priority_); stream_->SetPriority(priority); next_state_ = STATE_SEND_HEADERS; return OK; } void QuicHttpStream::OnStreamReady(int rv) { DCHECK(rv == OK || !stream_); if (rv == OK) { stream_->SetDelegate(this); if (request_info_->load_flags & LOAD_DISABLE_CONNECTION_MIGRATION) { stream_->DisableConnectionMigration(); } if (response_info_) { // This happens in the case of a asynchronous push rendezvous // that ultimately fails (e.g. vary failure). |response_info_| // non-null implies that |DoStreamRequest()| was called via // |SendRequest()|. next_state_ = STATE_SET_REQUEST_PRIORITY; rv = DoLoop(OK); } } else if (!was_handshake_confirmed_) { rv = ERR_QUIC_HANDSHAKE_FAILED; } if (rv != ERR_IO_PENDING) { DoCallback(rv); } } bool QuicHttpStream::CancelPromiseIfHasBody() { if (!request_body_stream_) return false; // Method type or request with body ineligble for push. this->push_handle_->Cancel(); this->push_handle_ = nullptr; next_state_ = STATE_REQUEST_STREAM; return true; } int QuicHttpStream::HandlePromise() { QuicAsyncStatus push_status = session_->push_promise_index()->Try( request_headers_, this, &this->push_handle_); switch (push_status) { case QUIC_FAILURE: // Push rendezvous failed. next_state_ = STATE_REQUEST_STREAM; break; case QUIC_SUCCESS: next_state_ = STATE_OPEN; if (!CancelPromiseIfHasBody()) { stream_net_log_.AddEvent( NetLog::TYPE_QUIC_HTTP_STREAM_ADOPTED_PUSH_STREAM, base::Bind(&NetLogQuicPushStreamCallback, stream_->id(), &request_info_->url)); session_->net_log().AddEvent( NetLog::TYPE_QUIC_HTTP_STREAM_ADOPTED_PUSH_STREAM, base::Bind(&NetLogQuicPushStreamCallback, stream_->id(), &request_info_->url)); // Avoid the call to |DoLoop()| below, which would reset // next_state_ to STATE_NONE. return OK; } break; case QUIC_PENDING: if (!CancelPromiseIfHasBody()) { // Have a promise but the promised stream doesn't exist yet. // Still have to do validation before accepting the promised // stream for sure. return ERR_IO_PENDING; } break; } return DoLoop(OK); } int QuicHttpStream::SendRequest(const HttpRequestHeaders& request_headers, HttpResponseInfo* response, const CompletionCallback& callback) { CHECK(!request_body_stream_); CHECK(!response_info_); CHECK(!callback.is_null()); CHECK(response); // TODO(rch): remove this once we figure out why channel ID is not being // sent when it should be. HostPortPair origin = HostPortPair::FromURL(request_info_->url); if (origin.Equals(HostPortPair("accounts.google.com", 443)) && request_headers.HasHeader(HttpRequestHeaders::kCookie)) { UMA_HISTOGRAM_BOOLEAN("Net.QuicSession.CookieSentToAccountsOverChannelId", ssl_info_.channel_id_sent); } if ((!found_promise_ && !stream_) || !session_) { return was_handshake_confirmed_ ? ERR_CONNECTION_CLOSED : ERR_QUIC_HANDSHAKE_FAILED; } // Store the serialized request headers. CreateSpdyHeadersFromHttpRequest(*request_info_, request_headers, HTTP2, /*direct=*/true, &request_headers_); // Store the request body. request_body_stream_ = request_info_->upload_data_stream; if (request_body_stream_) { // TODO(rch): Can we be more precise about when to allocate // raw_request_body_buf_. Removed the following check. DoReadRequestBody() // was being called even if we didn't yet allocate raw_request_body_buf_. // && (request_body_stream_->size() || // request_body_stream_->is_chunked())) // Use 10 packets as the body buffer size to give enough space to // help ensure we don't often send out partial packets. raw_request_body_buf_ = new IOBufferWithSize(static_cast(10 * kMaxPacketSize)); // The request body buffer is empty at first. request_body_buf_ = new DrainableIOBuffer(raw_request_body_buf_.get(), 0); } // Store the response info. response_info_ = response; int rv; if (found_promise_) { rv = HandlePromise(); } else { next_state_ = STATE_SET_REQUEST_PRIORITY; rv = DoLoop(OK); } if (rv == ERR_IO_PENDING) callback_ = callback; return rv > 0 ? OK : rv; } UploadProgress QuicHttpStream::GetUploadProgress() const { if (!request_body_stream_) return UploadProgress(); return UploadProgress(request_body_stream_->position(), request_body_stream_->size()); } int QuicHttpStream::ReadResponseHeaders(const CompletionCallback& callback) { CHECK(!callback.is_null()); if (stream_ == nullptr) return response_status_; // Check if we already have the response headers. If so, return synchronously. if (response_headers_received_) return OK; // Still waiting for the response, return IO_PENDING. CHECK(callback_.is_null()); callback_ = callback; return ERR_IO_PENDING; } int QuicHttpStream::ReadResponseBody(IOBuffer* buf, int buf_len, const CompletionCallback& callback) { if (!stream_) { // If the stream is already closed, there is no body to read. return response_status_; } int rv = ReadAvailableData(buf, buf_len); if (rv != ERR_IO_PENDING) return rv; CHECK(callback_.is_null()); CHECK(!user_buffer_.get()); CHECK_EQ(0, user_buffer_len_); callback_ = callback; user_buffer_ = buf; user_buffer_len_ = buf_len; return ERR_IO_PENDING; } void QuicHttpStream::Close(bool not_reusable) { // Note: the not_reusable flag has no meaning for SPDY streams. if (stream_) { stream_->SetDelegate(nullptr); stream_->Reset(QUIC_STREAM_CANCELLED); ResetStream(); response_status_ = was_handshake_confirmed_ ? ERR_CONNECTION_CLOSED : ERR_QUIC_HANDSHAKE_FAILED; } } HttpStream* QuicHttpStream::RenewStreamForAuth() { return nullptr; } bool QuicHttpStream::IsResponseBodyComplete() const { return next_state_ == STATE_OPEN && !stream_; } bool QuicHttpStream::IsConnectionReused() const { // TODO(rch): do something smarter here. return stream_ && stream_->id() > 1; } void QuicHttpStream::SetConnectionReused() { // QUIC doesn't need an indicator here. } bool QuicHttpStream::CanReuseConnection() const { // QUIC streams aren't considered reusable. return false; } int64_t QuicHttpStream::GetTotalReceivedBytes() const { // TODO(sclittle): Currently, this only includes headers and response body // bytes. Change this to include QUIC overhead as well. int64_t total_received_bytes = headers_bytes_received_; if (stream_) { total_received_bytes += stream_->stream_bytes_read(); } else { total_received_bytes += closed_stream_received_bytes_; } return total_received_bytes; } int64_t QuicHttpStream::GetTotalSentBytes() const { // TODO(sclittle): Currently, this only includes request headers and body // bytes. Change this to include QUIC overhead as well. int64_t total_sent_bytes = headers_bytes_sent_; if (stream_) { total_sent_bytes += stream_->stream_bytes_written(); } else { total_sent_bytes += closed_stream_sent_bytes_; } return total_sent_bytes; } bool QuicHttpStream::GetLoadTimingInfo(LoadTimingInfo* load_timing_info) const { // TODO(mmenke): Figure out what to do here. return true; } void QuicHttpStream::GetSSLInfo(SSLInfo* ssl_info) { *ssl_info = ssl_info_; } void QuicHttpStream::GetSSLCertRequestInfo( SSLCertRequestInfo* cert_request_info) { DCHECK(stream_); NOTIMPLEMENTED(); } bool QuicHttpStream::GetRemoteEndpoint(IPEndPoint* endpoint) { if (!session_) return false; *endpoint = session_->peer_address(); return true; } Error QuicHttpStream::GetSignedEKMForTokenBinding(crypto::ECPrivateKey* key, std::vector* out) { return session_->GetTokenBindingSignature(key, out); } void QuicHttpStream::Drain(HttpNetworkSession* session) { NOTREACHED(); Close(false); delete this; } void QuicHttpStream::PopulateNetErrorDetails(NetErrorDetails* details) { details->connection_info = HttpResponseInfo::CONNECTION_INFO_QUIC1_SPDY3; if (was_handshake_confirmed_) details->quic_connection_error = quic_connection_error_; if (session_) { session_->PopulateNetErrorDetails(details); } else { details->quic_port_migration_detected = port_migration_detected_; } } void QuicHttpStream::SetPriority(RequestPriority priority) { priority_ = priority; } void QuicHttpStream::OnHeadersAvailable(const SpdyHeaderBlock& headers, size_t frame_len) { headers_bytes_received_ += frame_len; // QuicHttpStream ignores trailers. if (response_headers_received_) { if (stream_->IsDoneReading()) { // Close the read side. If the write side has been closed, this will // invoke QuicHttpStream::OnClose to reset the stream. stream_->OnFinRead(); } return; } int rv = ProcessResponseHeaders(headers); if (rv != ERR_IO_PENDING && !callback_.is_null()) { DoCallback(rv); } } void QuicHttpStream::OnDataAvailable() { if (callback_.is_null()) { // Data is available, but can't be delivered return; } CHECK(user_buffer_.get()); CHECK_NE(0, user_buffer_len_); int rv = ReadAvailableData(user_buffer_.get(), user_buffer_len_); if (rv == ERR_IO_PENDING) { // This was a spurrious notification. Wait for the next one. return; } CHECK(!callback_.is_null()); user_buffer_ = nullptr; user_buffer_len_ = 0; DoCallback(rv); } void QuicHttpStream::OnClose(QuicErrorCode error) { if (error != QUIC_NO_ERROR) { response_status_ = was_handshake_confirmed_ ? ERR_QUIC_PROTOCOL_ERROR : ERR_QUIC_HANDSHAKE_FAILED; } else if (!response_headers_received_) { response_status_ = ERR_ABORTED; } ResetStream(); if (!callback_.is_null()) { quic_connection_error_ = error; DoCallback(response_status_); } } void QuicHttpStream::OnError(int error) { ResetStream(); response_status_ = was_handshake_confirmed_ ? error : ERR_QUIC_HANDSHAKE_FAILED; if (!callback_.is_null()) DoCallback(response_status_); } bool QuicHttpStream::HasSendHeadersComplete() { return next_state_ > STATE_SEND_HEADERS_COMPLETE; } void QuicHttpStream::OnCryptoHandshakeConfirmed() { was_handshake_confirmed_ = true; } void QuicHttpStream::OnSessionClosed(int error, bool port_migration_detected) { Close(false); session_error_ = error; port_migration_detected_ = port_migration_detected; session_.reset(); } void QuicHttpStream::OnIOComplete(int rv) { rv = DoLoop(rv); if (rv != ERR_IO_PENDING && !callback_.is_null()) { DoCallback(rv); } } void QuicHttpStream::DoCallback(int rv) { CHECK_NE(rv, ERR_IO_PENDING); CHECK(!callback_.is_null()); // The client callback can do anything, including destroying this class, // so any pending callback must be issued after everything else is done. base::ResetAndReturn(&callback_).Run(rv); } int QuicHttpStream::DoLoop(int rv) { do { State state = next_state_; next_state_ = STATE_NONE; switch (state) { case STATE_REQUEST_STREAM: rv = DoStreamRequest(); break; case STATE_SET_REQUEST_PRIORITY: rv = DoSetRequestPriority(); break; case STATE_SEND_HEADERS: CHECK_EQ(OK, rv); rv = DoSendHeaders(); break; case STATE_SEND_HEADERS_COMPLETE: rv = DoSendHeadersComplete(rv); break; case STATE_READ_REQUEST_BODY: CHECK_EQ(OK, rv); rv = DoReadRequestBody(); break; case STATE_READ_REQUEST_BODY_COMPLETE: rv = DoReadRequestBodyComplete(rv); break; case STATE_SEND_BODY: CHECK_EQ(OK, rv); rv = DoSendBody(); break; case STATE_SEND_BODY_COMPLETE: rv = DoSendBodyComplete(rv); break; case STATE_OPEN: CHECK_EQ(OK, rv); break; default: NOTREACHED() << "next_state_: " << next_state_; break; } } while (next_state_ != STATE_NONE && next_state_ != STATE_OPEN && rv != ERR_IO_PENDING); return rv; } int QuicHttpStream::DoSendHeaders() { if (!stream_) return ERR_UNEXPECTED; // Log the actual request with the URL Request's net log. stream_net_log_.AddEvent( NetLog::TYPE_HTTP_TRANSACTION_QUIC_SEND_REQUEST_HEADERS, base::Bind(&QuicRequestNetLogCallback, stream_->id(), &request_headers_, priority_)); bool has_upload_data = request_body_stream_ != nullptr; next_state_ = STATE_SEND_HEADERS_COMPLETE; size_t frame_len = stream_->WriteHeaders(request_headers_, !has_upload_data, nullptr); headers_bytes_sent_ += frame_len; request_headers_.clear(); return static_cast(frame_len); } int QuicHttpStream::DoSendHeadersComplete(int rv) { if (rv < 0) return rv; // If the stream is already closed, don't read the request the body. if (!stream_) return response_status_; next_state_ = request_body_stream_ ? STATE_READ_REQUEST_BODY : STATE_OPEN; return OK; } int QuicHttpStream::DoReadRequestBody() { next_state_ = STATE_READ_REQUEST_BODY_COMPLETE; return request_body_stream_->Read( raw_request_body_buf_.get(), raw_request_body_buf_->size(), base::Bind(&QuicHttpStream::OnIOComplete, weak_factory_.GetWeakPtr())); } int QuicHttpStream::DoReadRequestBodyComplete(int rv) { // |rv| is the result of read from the request body from the last call to // DoSendBody(). if (rv < 0) return rv; // If the stream is already closed, don't continue. if (!stream_) return response_status_; request_body_buf_ = new DrainableIOBuffer(raw_request_body_buf_.get(), rv); if (rv == 0) { // Reached the end. DCHECK(request_body_stream_->IsEOF()); } next_state_ = STATE_SEND_BODY; return OK; } int QuicHttpStream::DoSendBody() { if (!stream_) return ERR_UNEXPECTED; CHECK(request_body_stream_); CHECK(request_body_buf_.get()); const bool eof = request_body_stream_->IsEOF(); int len = request_body_buf_->BytesRemaining(); if (len > 0 || eof) { next_state_ = STATE_SEND_BODY_COMPLETE; base::StringPiece data(request_body_buf_->data(), len); return stream_->WriteStreamData( data, eof, base::Bind(&QuicHttpStream::OnIOComplete, weak_factory_.GetWeakPtr())); } next_state_ = STATE_OPEN; return OK; } int QuicHttpStream::DoSendBodyComplete(int rv) { if (rv < 0) return rv; // If the stream is already closed, don't continue. if (!stream_) return response_status_; request_body_buf_->DidConsume(request_body_buf_->BytesRemaining()); if (!request_body_stream_->IsEOF()) { next_state_ = STATE_READ_REQUEST_BODY; return OK; } next_state_ = STATE_OPEN; return OK; } int QuicHttpStream::ProcessResponseHeaders(const SpdyHeaderBlock& headers) { if (!SpdyHeadersToHttpResponse(headers, HTTP2, response_info_)) { DLOG(WARNING) << "Invalid headers"; return ERR_QUIC_PROTOCOL_ERROR; } // Put the peer's IP address and port into the response. IPEndPoint address = session_->peer_address(); response_info_->socket_address = HostPortPair::FromIPEndPoint(address); response_info_->connection_info = HttpResponseInfo::CONNECTION_INFO_QUIC1_SPDY3; response_info_->vary_data.Init(*request_info_, *response_info_->headers.get()); response_info_->was_npn_negotiated = true; response_info_->npn_negotiated_protocol = "quic/1+spdy/3"; response_info_->response_time = base::Time::Now(); response_info_->request_time = request_time_; response_headers_received_ = true; return OK; } int QuicHttpStream::ReadAvailableData(IOBuffer* buf, int buf_len) { int rv = stream_->Read(buf, buf_len); // TODO(rtenneti): Temporary fix for crbug.com/585591. Added a check for null // |stream_| to fix crash bug. Delete |stream_| check and histogram after fix // is merged. bool null_stream = stream_ == nullptr; UMA_HISTOGRAM_BOOLEAN("Net.QuicReadAvailableData.NullStream", null_stream); if (null_stream) return rv; if (stream_->IsDoneReading()) { stream_->SetDelegate(nullptr); stream_->OnFinRead(); ResetStream(); } return rv; } void QuicHttpStream::ResetStream() { if (push_handle_) { push_handle_->Cancel(); push_handle_ = nullptr; } if (!stream_) return; closed_stream_received_bytes_ = stream_->stream_bytes_read(); closed_stream_sent_bytes_ = stream_->stream_bytes_written(); stream_ = nullptr; // If |request_body_stream_| is non-NULL, Reset it, to abort any in progress // read. if (request_body_stream_) request_body_stream_->Reset(); } } // namespace net