Implement two phase upload protocol handler class.

BUG=169557


Review URL: https://chromiumcodereview.appspot.com/12387058

git-svn-id: svn://svn.chromium.org/chrome/trunk/src@186944 0039d316-1c4b-4281-b951-d872f2087c98

8 files changed, 627 insertions, 0 deletions

diff --git a/chrome/browser/safe_browsing/local_two_phase_testserver.cc b/chrome/browser/safe_browsing/local_two_phase_testserver.cc
new file mode 100644
index 0000000..c27d058
--- /dev/null
+++ b/chrome/browser/safe_browsing/local_two_phase_testserver.cc
@@ -0,0 +1,38 @@
+// Copyright 2013 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 "chrome/browser/safe_browsing/local_two_phase_testserver.h"
+
+#include "base/command_line.h"
+#include "base/path_service.h"
+#include "base/string_number_conversions.h"
+#include "base/values.h"
+#include "net/test/python_utils.h"
+#include "net/test/test_server.h"
+
+LocalTwoPhaseTestServer::LocalTwoPhaseTestServer()
+    : net::LocalTestServer(net::TestServer::TYPE_HTTP,
+                           net::TestServer::kLocalhost,
+                           base::FilePath()) {
+}
+
+LocalTwoPhaseTestServer::~LocalTwoPhaseTestServer() {}
+
+bool LocalTwoPhaseTestServer::GetTestServerPath(
+    base::FilePath* testserver_path) const {
+  base::FilePath testserver_dir;
+  if (!PathService::Get(base::DIR_SOURCE_ROOT, &testserver_dir)) {
+    LOG(ERROR) << "Failed to get DIR_SOURCE_ROOT";
+    return false;
+  }
+
+  testserver_dir = testserver_dir
+      .Append(FILE_PATH_LITERAL("chrome"))
+      .Append(FILE_PATH_LITERAL("browser"))
+      .Append(FILE_PATH_LITERAL("safe_browsing"));
+
+  *testserver_path = testserver_dir.Append(FILE_PATH_LITERAL(
+      "two_phase_testserver.py"));
+  return true;
+}
diff --git a/chrome/browser/safe_browsing/local_two_phase_testserver.h b/chrome/browser/safe_browsing/local_two_phase_testserver.h
new file mode 100644
index 0000000..b6b49e0
--- /dev/null
+++ b/chrome/browser/safe_browsing/local_two_phase_testserver.h
@@ -0,0 +1,31 @@
+// Copyright 2013 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.
+
+#ifndef CHROME_BROWSER_SAFE_BROWSING_LOCAL_TWO_PHASE_TESTSERVER_H_
+#define CHROME_BROWSER_SAFE_BROWSING_LOCAL_TWO_PHASE_TESTSERVER_H_
+
+#include "base/basictypes.h"
+#include "base/compiler_specific.h"
+#include "base/files/file_path.h"
+#include "net/test/local_test_server.h"
+
+// Runs a Python-based two phase upload test server on the same machine in which
+// the LocalTwoPhaseTestServer runs.
+class LocalTwoPhaseTestServer : public net::LocalTestServer {
+ public:
+  // Initialize a two phase protocol test server.
+  LocalTwoPhaseTestServer();
+
+  virtual ~LocalTwoPhaseTestServer();
+
+  // Returns the path to two_phase_testserver.py.
+  virtual bool GetTestServerPath(
+      base::FilePath* testserver_path) const OVERRIDE;
+
+ private:
+  DISALLOW_COPY_AND_ASSIGN(LocalTwoPhaseTestServer);
+};
+
+#endif  // CHROME_BROWSER_SAFE_BROWSING_LOCAL_TWO_PHASE_TESTSERVER_H_
+
diff --git a/chrome/browser/safe_browsing/two_phase_testserver.py b/chrome/browser/safe_browsing/two_phase_testserver.py
new file mode 100755
index 0000000..c054047
--- /dev/null
+++ b/chrome/browser/safe_browsing/two_phase_testserver.py
@@ -0,0 +1,110 @@
+#!/usr/bin/env python
+# Copyright 2013 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.
+
+"""Testserver for the two phase upload protocol."""
+
+import base64
+import BaseHTTPServer
+import hashlib
+import os
+import sys
+import urlparse
+
+BASE_DIR = os.path.dirname(os.path.abspath(__file__))
+
+sys.path.append(os.path.join(BASE_DIR, '..', '..', '..', 'net',
+                             'tools', 'testserver'))
+import testserver_base
+
+
+class RequestHandler(BaseHTTPServer.BaseHTTPRequestHandler):
+  def ReadRequestBody(self):
+    """This function reads the body of the current HTTP request, handling
+    both plain and chunked transfer encoded requests."""
+
+    if self.headers.getheader('transfer-encoding') == 'chunked':
+      return ''
+
+    length = int(self.headers.getheader('content-length'))
+    return self.rfile.read(length)
+
+  def do_GET(self):
+    print 'GET', self.path
+    self.send_error(400, 'GET not supported')
+
+  def do_POST(self):
+    request_body = self.ReadRequestBody()
+    print 'POST', repr(self.path), repr(request_body)
+
+    kStartHeader = 'x-goog-resumable'
+    if kStartHeader not in self.headers:
+      self.send_error(400, 'Missing header: ' + kStartHeader)
+      return
+    if self.headers.get(kStartHeader) != 'start':
+      self.send_error(400, 'Invalid %s header value: %s' % (
+          kStartHeader, self.headers.get(kStartHeader)))
+      return
+
+    metadata_hash = hashlib.sha1(request_body).hexdigest()
+    _, _, url_path, _, query, _ = urlparse.urlparse(self.path)
+    query_args = urlparse.parse_qs(query)
+
+    if query_args.get('p1close'):
+      self.close_connection = 1
+      return
+
+    put_url = 'http://%s:%d/put?%s,%s,%s' % (self.server.server_address[0],
+                                          self.server.server_port,
+                                          url_path,
+                                          metadata_hash,
+                                          base64.urlsafe_b64encode(query))
+    self.send_response(int(query_args.get('p1code', [201])[0]))
+    self.send_header('Location', put_url)
+    self.end_headers()
+
+  def do_PUT(self):
+    _, _, url_path, _, query, _ = urlparse.urlparse(self.path)
+    if url_path != '/put':
+      self.send_error(400, 'invalid path on 2nd phase: ' + url_path)
+      return
+
+    initial_path, metadata_hash, config_query_b64 = query.split(',', 2)
+    config_query = urlparse.parse_qs(base64.urlsafe_b64decode(config_query_b64))
+
+    request_body = self.ReadRequestBody()
+    print 'PUT', repr(self.path), len(request_body), 'bytes'
+
+    if config_query.get('p2close'):
+      self.close_connection = 1
+      return
+
+    self.send_response(int(config_query.get('p2code', [200])[0]))
+    self.end_headers()
+    self.wfile.write('%s\n%s\n%s\n' % (
+        initial_path,
+        metadata_hash,
+        hashlib.sha1(request_body).hexdigest()))
+
+
+class ServerRunner(testserver_base.TestServerRunner):
+  """TestServerRunner for safebrowsing_test_server.py."""
+
+  def create_server(self, server_data):
+    server = BaseHTTPServer.HTTPServer((self.options.host, self.options.port),
+                                       RequestHandler)
+    print 'server started on port %d...' % server.server_port
+    server_data['port'] = server.server_port
+
+    return server
+
+  def add_options(self):
+    testserver_base.TestServerRunner.add_options(self)
+    self.option_parser.add_option('--data-file', dest='data_file',
+                                  help='File containing safebrowsing test '
+                                  'data and expectations')
+
+
+if __name__ == '__main__':
+  sys.exit(ServerRunner().main())
diff --git a/chrome/browser/safe_browsing/two_phase_uploader.cc b/chrome/browser/safe_browsing/two_phase_uploader.cc
new file mode 100644
index 0000000..9b6165e
--- /dev/null
+++ b/chrome/browser/safe_browsing/two_phase_uploader.cc
@@ -0,0 +1,146 @@
+// Copyright 2013 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 "chrome/browser/safe_browsing/two_phase_uploader.h"
+
+#include "base/bind.h"
+#include "base/task_runner.h"
+#include "net/base/net_errors.h"
+#include "net/http/http_response_headers.h"
+#include "net/url_request/url_fetcher.h"
+#include "net/url_request/url_request_status.h"
+
+namespace {
+
+// Header sent on initial request to start the two phase upload process.
+const char* kStartHeader = "x-goog-resumable: start";
+
+// Header returned on initial response with URL to use for the second phase.
+const char* kLocationHeader = "Location";
+
+const char* kUploadContentType = "application/octet-stream";
+
+}  // namespace
+
+TwoPhaseUploader::TwoPhaseUploader(
+    net::URLRequestContextGetter* url_request_context_getter,
+    base::TaskRunner* file_task_runner,
+    const GURL& base_url,
+    const std::string& metadata,
+    const base::FilePath& file_path,
+    const ProgressCallback& progress_callback,
+    const FinishCallback& finish_callback)
+    : state_(STATE_NONE),
+      url_request_context_getter_(url_request_context_getter),
+      file_task_runner_(file_task_runner),
+      base_url_(base_url),
+      metadata_(metadata),
+      file_path_(file_path),
+      progress_callback_(progress_callback),
+      finish_callback_(finish_callback) {
+}
+
+TwoPhaseUploader::~TwoPhaseUploader() {
+  DCHECK(CalledOnValidThread());
+}
+
+void TwoPhaseUploader::Start() {
+  DCHECK(CalledOnValidThread());
+  DCHECK_EQ(STATE_NONE, state_);
+
+  UploadMetadata();
+}
+
+void TwoPhaseUploader::OnURLFetchComplete(const net::URLFetcher* source) {
+  DCHECK(CalledOnValidThread());
+  net::URLRequestStatus status = source->GetStatus();
+  int response_code = source->GetResponseCode();
+
+  DVLOG(1) << __FUNCTION__ << " " << source->GetURL().spec()
+           << " " << status.status() << " " << response_code;
+
+  if (!status.is_success()) {
+    LOG(ERROR) << "URLFetcher failed, status=" << status.status()
+               << " err=" << status.error();
+    Finish(status.error(), response_code, "");
+    return;
+  }
+
+  std::string response;
+  source->GetResponseAsString(&response);
+
+  switch (state_) {
+    case UPLOAD_METADATA:
+      {
+        if (response_code != 201) {
+          LOG(ERROR) << "Invalid response to initial request: "
+                     << response_code;
+          Finish(net::OK, response_code, response);
+          return;
+        }
+        std::string location;
+        if (!source->GetResponseHeaders()->EnumerateHeader(
+              NULL, kLocationHeader, &location)) {
+          LOG(ERROR) << "no location header";
+          Finish(net::OK, response_code, "");
+          return;
+        }
+        DVLOG(1) << "upload location: " << location;
+        upload_url_ = GURL(location);
+        UploadFile();
+        break;
+      }
+    case UPLOAD_FILE:
+      if (response_code != 200) {
+          LOG(ERROR) << "Invalid response to upload request: "
+                     << response_code;
+      } else {
+        state_ = STATE_SUCCESS;
+      }
+      Finish(net::OK, response_code, response);
+      return;
+    default:
+      NOTREACHED();
+  };
+}
+
+void TwoPhaseUploader::OnURLFetchUploadProgress(const net::URLFetcher* source,
+                                                int64 current, int64 total) {
+  DCHECK(CalledOnValidThread());
+  DVLOG(3) << __FUNCTION__ << " " << source->GetURL().spec()
+           << " " << current << "/" << total;
+  if (state_ == UPLOAD_FILE)
+    progress_callback_.Run(current, total);
+}
+
+void TwoPhaseUploader::UploadMetadata() {
+  DCHECK(CalledOnValidThread());
+  state_ = UPLOAD_METADATA;
+  url_fetcher_.reset(net::URLFetcher::Create(base_url_, net::URLFetcher::POST,
+                                             this));
+  url_fetcher_->SetRequestContext(url_request_context_getter_);
+  url_fetcher_->SetExtraRequestHeaders(kStartHeader);
+  url_fetcher_->SetUploadData(kUploadContentType, metadata_);
+  url_fetcher_->Start();
+}
+
+void TwoPhaseUploader::UploadFile() {
+  DCHECK(CalledOnValidThread());
+  state_ = UPLOAD_FILE;
+
+  url_fetcher_.reset(net::URLFetcher::Create(upload_url_, net::URLFetcher::PUT,
+                                             this));
+  url_fetcher_->SetRequestContext(url_request_context_getter_);
+  url_fetcher_->SetUploadFilePath(kUploadContentType,
+                                  file_path_,
+                                  file_task_runner_);
+  url_fetcher_->Start();
+}
+
+void TwoPhaseUploader::Finish(int net_error,
+                              int response_code,
+                              const std::string& response) {
+  DCHECK(CalledOnValidThread());
+  finish_callback_.Run(state_, net_error, response_code, response);
+}
diff --git a/chrome/browser/safe_browsing/two_phase_uploader.h b/chrome/browser/safe_browsing/two_phase_uploader.h
new file mode 100644
index 0000000..460f1b3
--- /dev/null
+++ b/chrome/browser/safe_browsing/two_phase_uploader.h
@@ -0,0 +1,102 @@
+// Copyright 2013 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.
+
+#ifndef CHROME_BROWSER_SAFE_BROWSING_TWO_PHASE_UPLOADER_H_
+#define CHROME_BROWSER_SAFE_BROWSING_TWO_PHASE_UPLOADER_H_
+
+#include <string>
+
+#include "base/basictypes.h"
+#include "base/callback.h"
+#include "base/files/file_path.h"
+#include "base/memory/scoped_ptr.h"
+#include "base/platform_file.h"
+#include "base/threading/non_thread_safe.h"
+#include "googleurl/src/gurl.h"
+#include "net/url_request/url_fetcher_delegate.h"
+#include "net/url_request/url_request_context_getter.h"
+
+namespace base {
+class TaskRunner;
+}
+namespace net {
+class URLFetcher;
+class URLRequestContextGetter;
+}
+
+// Implements the Google two-phase resumable upload protocol.
+// Protocol documentation:
+// https://developers.google.com/storage/docs/developer-guide#resumable
+// Note: This doc is for the Cloud Storage API which specifies the POST body
+// must be empty, however the safebrowsing use of the two-phase protocol
+// supports sending metadata in the POST request body. We also do not need the
+// api-version and authorization headers.
+// TODO(mattm): support retry / resume.
+class TwoPhaseUploader : public net::URLFetcherDelegate,
+                         public base::NonThreadSafe {
+ public:
+  enum State {
+    STATE_NONE,
+    UPLOAD_METADATA,
+    UPLOAD_FILE,
+    STATE_SUCCESS,
+  };
+  typedef base::Callback<void(int64 sent, int64 total)> ProgressCallback;
+  typedef base::Callback<void(State state,
+                              int net_error,
+                              int response_code,
+                              const std::string& response_data)> FinishCallback;
+
+  // Create the uploader.  The Start method must be called to begin the upload.
+  // Network processing will use |url_request_context_getter|.
+  // The uploaded |file_path| will be read on |file_task_runner|.
+  // The first phase request will be sent to |base_url|, with |metadata|
+  // included.
+  // |progress_callback| will be called periodically as the second phase
+  // progresses.
+  // On success |finish_callback| will be called with state = STATE_SUCCESS and
+  // the server response in response_data. On failure, state will specify
+  // which step the failure occurred in, and net_error, response_code, and
+  // response_data will specify information about the error. |finish_callback|
+  // will not be called if the upload is cancelled by destructing the
+  // TwoPhaseUploader object before completion.
+  TwoPhaseUploader(
+      net::URLRequestContextGetter* url_request_context_getter,
+      base::TaskRunner* file_task_runner,
+      const GURL& base_url,
+      const std::string& metadata,
+      const base::FilePath& file_path,
+      const ProgressCallback& progress_callback,
+      const FinishCallback& finish_callback);
+  virtual ~TwoPhaseUploader();
+
+  // Begins the upload process.
+  void Start();
+
+  // net::URLFetcherDelegate implementation:
+  virtual void OnURLFetchComplete(const net::URLFetcher* source) OVERRIDE;
+  virtual void OnURLFetchUploadProgress(const net::URLFetcher* source,
+                                        int64 current, int64 total) OVERRIDE;
+
+ private:
+  void UploadMetadata();
+  void UploadFile();
+  void Finish(int net_error, int response_code, const std::string& response);
+
+  State state_;
+  scoped_refptr<net::URLRequestContextGetter> url_request_context_getter_;
+  scoped_refptr<base::TaskRunner> file_task_runner_;
+  GURL base_url_;
+  GURL upload_url_;
+  std::string metadata_;
+  const base::FilePath file_path_;
+  ProgressCallback progress_callback_;
+  FinishCallback finish_callback_;
+
+  scoped_ptr<net::URLFetcher> url_fetcher_;
+
+  DISALLOW_COPY_AND_ASSIGN(TwoPhaseUploader);
+};
+
+#endif  // CHROME_BROWSER_SAFE_BROWSING_TWO_PHASE_UPLOADER_H_
diff --git a/chrome/browser/safe_browsing/two_phase_uploader_unittest.cc b/chrome/browser/safe_browsing/two_phase_uploader_unittest.cc
new file mode 100644
index 0000000..0201999
--- /dev/null
+++ b/chrome/browser/safe_browsing/two_phase_uploader_unittest.cc
@@ -0,0 +1,196 @@
+// Copyright 2013 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 "chrome/browser/safe_browsing/two_phase_uploader.h"
+
+#include "base/files/file_path.h"
+#include "base/message_loop.h"
+#include "chrome/browser/safe_browsing/local_two_phase_testserver.h"
+#include "content/public/test/test_browser_thread.h"
+#include "content/public/test/test_utils.h"
+#include "net/base/net_errors.h"
+#include "net/url_request/url_fetcher.h"
+#include "net/url_request/url_request_test_util.h"
+#include "testing/gtest/include/gtest/gtest.h"
+
+using content::BrowserThread;
+using content::MessageLoopRunner;
+
+namespace {
+
+class Delegate {
+ public:
+  Delegate() : state_(TwoPhaseUploader::STATE_NONE) {
+  }
+
+  void ProgressCallback(int64 current, int64 total) {}
+
+  void FinishCallback(scoped_refptr<MessageLoopRunner> runner,
+                      TwoPhaseUploader::State state,
+                      int net_error,
+                      int response_code,
+                      const std::string& response) {
+    state_ = state;
+    net_error_ = net_error;
+    response_code_ = response_code;
+    response_ = response;
+    runner->Quit();
+  }
+
+  TwoPhaseUploader::State state_;
+  int net_error_;
+  int response_code_;
+  std::string response_;
+};
+
+base::FilePath GetTestFilePath() {
+  base::FilePath file_path;
+  PathService::Get(base::DIR_SOURCE_ROOT, &file_path);
+  file_path = file_path.Append(FILE_PATH_LITERAL("net"));
+  file_path = file_path.Append(FILE_PATH_LITERAL("data"));
+  file_path = file_path.Append(FILE_PATH_LITERAL("url_request_unittest"));
+  file_path = file_path.Append(FILE_PATH_LITERAL("BullRunSpeech.txt"));
+  return file_path;
+}
+
+}  // namespace
+
+class TwoPhaseUploaderTest : public testing::Test {
+ public:
+  TwoPhaseUploaderTest()
+      : db_thread_(BrowserThread::DB),
+        io_thread_(BrowserThread::IO, &message_loop_),
+        url_request_context_getter_(new net::TestURLRequestContextGetter(
+            BrowserThread::GetMessageLoopProxyForThread(BrowserThread::IO))) {
+  }
+
+  virtual void SetUp() {
+    db_thread_.Start();
+  }
+
+ protected:
+  MessageLoopForIO message_loop_;
+  content::TestBrowserThread db_thread_;
+  content::TestBrowserThread io_thread_;
+
+  scoped_refptr<net::TestURLRequestContextGetter> url_request_context_getter_;
+};
+
+TEST_F(TwoPhaseUploaderTest, UploadFile) {
+  scoped_refptr<MessageLoopRunner> runner = new MessageLoopRunner;
+  LocalTwoPhaseTestServer test_server;
+  ASSERT_TRUE(test_server.Start());
+  Delegate delegate;
+  TwoPhaseUploader uploader(
+      url_request_context_getter_,
+      BrowserThread::GetMessageLoopProxyForThread(BrowserThread::DB),
+      test_server.GetURL("start"),
+      "metadata",
+      GetTestFilePath(),
+      base::Bind(&Delegate::ProgressCallback, base::Unretained(&delegate)),
+      base::Bind(&Delegate::FinishCallback, base::Unretained(&delegate),
+                 runner));
+  uploader.Start();
+  runner->Run();
+  EXPECT_EQ(TwoPhaseUploader::STATE_SUCCESS, delegate.state_);
+  EXPECT_EQ(net::OK, delegate.net_error_);
+  EXPECT_EQ(200, delegate.response_code_);
+  EXPECT_EQ(
+      "/start\n"                                     // path of start request
+      "4c24b2612e94e2ae622e54397663f2b7bf0a2e17\n"   // sha1sum of "metadata"
+      "944857cc626f2cafe232521986b4c6d3f9993c97\n",  // sha1sum of test file
+      delegate.response_);
+}
+
+TEST_F(TwoPhaseUploaderTest, BadPhaseOneResponse) {
+  scoped_refptr<MessageLoopRunner> runner = new MessageLoopRunner;
+  LocalTwoPhaseTestServer test_server;
+  ASSERT_TRUE(test_server.Start());
+  Delegate delegate;
+  TwoPhaseUploader uploader(
+      url_request_context_getter_,
+      BrowserThread::GetMessageLoopProxyForThread(BrowserThread::DB),
+      test_server.GetURL("start?p1code=500"),
+      "metadata",
+      GetTestFilePath(),
+      base::Bind(&Delegate::ProgressCallback, base::Unretained(&delegate)),
+      base::Bind(&Delegate::FinishCallback, base::Unretained(&delegate),
+                 runner));
+  uploader.Start();
+  runner->Run();
+  EXPECT_EQ(TwoPhaseUploader::UPLOAD_METADATA, delegate.state_);
+  EXPECT_EQ(net::OK, delegate.net_error_);
+  EXPECT_EQ(500, delegate.response_code_);
+  EXPECT_EQ("", delegate.response_);
+}
+
+TEST_F(TwoPhaseUploaderTest, BadPhaseTwoResponse) {
+  scoped_refptr<MessageLoopRunner> runner = new MessageLoopRunner;
+  LocalTwoPhaseTestServer test_server;
+  ASSERT_TRUE(test_server.Start());
+  Delegate delegate;
+  TwoPhaseUploader uploader(
+      url_request_context_getter_,
+      BrowserThread::GetMessageLoopProxyForThread(BrowserThread::DB),
+      test_server.GetURL("start?p2code=500"),
+      "metadata",
+      GetTestFilePath(),
+      base::Bind(&Delegate::ProgressCallback, base::Unretained(&delegate)),
+      base::Bind(&Delegate::FinishCallback, base::Unretained(&delegate),
+                 runner));
+  uploader.Start();
+  runner->Run();
+  EXPECT_EQ(TwoPhaseUploader::UPLOAD_FILE, delegate.state_);
+  EXPECT_EQ(net::OK, delegate.net_error_);
+  EXPECT_EQ(500, delegate.response_code_);
+  EXPECT_EQ(
+      "/start\n"                                     // path of start request
+      "4c24b2612e94e2ae622e54397663f2b7bf0a2e17\n"   // sha1sum of "metadata"
+      "944857cc626f2cafe232521986b4c6d3f9993c97\n",  // sha1sum of test file
+      delegate.response_);
+}
+
+TEST_F(TwoPhaseUploaderTest, PhaseOneConnectionClosed) {
+  scoped_refptr<MessageLoopRunner> runner = new MessageLoopRunner;
+  LocalTwoPhaseTestServer test_server;
+  ASSERT_TRUE(test_server.Start());
+  Delegate delegate;
+  TwoPhaseUploader uploader(
+      url_request_context_getter_,
+      BrowserThread::GetMessageLoopProxyForThread(BrowserThread::DB),
+      test_server.GetURL("start?p1close=1"),
+      "metadata",
+      GetTestFilePath(),
+      base::Bind(&Delegate::ProgressCallback, base::Unretained(&delegate)),
+      base::Bind(&Delegate::FinishCallback, base::Unretained(&delegate),
+                 runner));
+  uploader.Start();
+  runner->Run();
+  EXPECT_EQ(TwoPhaseUploader::UPLOAD_METADATA, delegate.state_);
+  EXPECT_EQ(net::ERR_EMPTY_RESPONSE, delegate.net_error_);
+  EXPECT_EQ(net::URLFetcher::RESPONSE_CODE_INVALID, delegate.response_code_);
+  EXPECT_EQ("", delegate.response_);
+}
+
+TEST_F(TwoPhaseUploaderTest, PhaseTwoConnectionClosed) {
+  scoped_refptr<MessageLoopRunner> runner = new MessageLoopRunner;
+  LocalTwoPhaseTestServer test_server;
+  ASSERT_TRUE(test_server.Start());
+  Delegate delegate;
+  TwoPhaseUploader uploader(
+      url_request_context_getter_,
+      BrowserThread::GetMessageLoopProxyForThread(BrowserThread::DB),
+      test_server.GetURL("start?p2close=1"),
+      "metadata",
+      GetTestFilePath(),
+      base::Bind(&Delegate::ProgressCallback, base::Unretained(&delegate)),
+      base::Bind(&Delegate::FinishCallback, base::Unretained(&delegate),
+                 runner));
+  uploader.Start();
+  runner->Run();
+  EXPECT_EQ(TwoPhaseUploader::UPLOAD_FILE, delegate.state_);
+  EXPECT_EQ(net::ERR_EMPTY_RESPONSE, delegate.net_error_);
+  EXPECT_EQ(net::URLFetcher::RESPONSE_CODE_INVALID, delegate.response_code_);
+  EXPECT_EQ("", delegate.response_);
+}
diff --git a/chrome/chrome_browser.gypi b/chrome/chrome_browser.gypi
index 93ca645..d0f2d0d 100644
--- a/chrome/chrome_browser.gypi
+++ b/chrome/chrome_browser.gypi
@@ -1764,6 +1764,8 @@
         'browser/safe_browsing/signature_util_posix.cc',
         'browser/safe_browsing/signature_util_win.cc',
         'browser/safe_browsing/signature_util.h',
+        'browser/safe_browsing/two_phase_uploader.cc',
+        'browser/safe_browsing/two_phase_uploader.h',
         'browser/safe_browsing/ui_manager.cc',
         'browser/safe_browsing/ui_manager.h',
         'browser/screensaver_window_finder_gtk.cc',
diff --git a/chrome/chrome_tests_unit.gypi b/chrome/chrome_tests_unit.gypi
index f5f99c5..838493c 100644
--- a/chrome/chrome_tests_unit.gypi
+++ b/chrome/chrome_tests_unit.gypi
@@ -1058,6 +1058,7 @@
         'browser/safe_browsing/client_side_detection_host_unittest.cc',
         'browser/safe_browsing/client_side_detection_service_unittest.cc',
         'browser/safe_browsing/download_protection_service_unittest.cc',
+        'browser/safe_browsing/local_two_phase_testserver.cc',
         'browser/safe_browsing/malware_details_unittest.cc',
         'browser/safe_browsing/ping_manager_unittest.cc',
         'browser/safe_browsing/prefix_set_unittest.cc',
@@ -1070,6 +1071,7 @@
         'browser/safe_browsing/safe_browsing_store_unittest_helper.cc',
         'browser/safe_browsing/safe_browsing_util_unittest.cc',
         'browser/safe_browsing/signature_util_win_unittest.cc',
+        'browser/safe_browsing/two_phase_uploader_unittest.cc',
         'browser/search_engines/search_host_to_urls_map_unittest.cc',
         'browser/search_engines/search_provider_install_data_unittest.cc',
         'browser/search_engines/template_url_fetcher_unittest.cc',