Adding CupsProxyService SocketManager

DDoc: go/cups-plugin

This socket manager handles the actual proxying of the IPP requests to
the CUPS daemon, via the default named UNIX domain socket, /run/cups/cups.sock.

Bug: chromium:945409
Test: Passes added unittests.
Change-Id: I3ed529bf0dc55f8aa1ee9eadbc0accf236f47000
Reviewed-on: https://chromium-review.googlesource.com/c/chromium/src/+/1600132Reviewed-by: Ken Rockot <rockot@google.com>
Reviewed-by: Sean Kau <skau@chromium.org>
Commit-Queue: Luum Habtemariam <luum@chromium.org>
Cr-Commit-Position: refs/heads/master@{#666513}

chrome/services/cups_proxy/BUILD.gn

@@ -14,6 +14,8 @@ source_set("cups_proxy") {
     "cups_proxy_service.h",
     "cups_proxy_service_delegate.cc",
     "cups_proxy_service_delegate.h",
+    "socket_manager.cc",
+    "socket_manager.h",
   ]
 
   deps = [
@@ -55,16 +57,23 @@ static_library("test_support") {
 source_set("unit_tests") {
   testonly = true
 
-  if (use_cups) {
-    sources = [
-      "printer_installer_unittest.cc",
-    ]
+  sources = [
+    "socket_manager_unittest.cc",
+  ]
 
-    deps = [
-      ":cups_proxy",
-      ":test_support",
-      "//base",
-      "//testing/gtest",
-    ]
+  deps = [
+    ":cups_proxy",
+    ":test_support",
+    "//base",
+    "//chrome/services/cups_proxy/public/cpp",
+    "//testing/gtest",
+  ]
+
+  if (use_cups) {
+    sources += [ "printer_installer_unittest.cc" ]
   }
+
+  data = [
+    "//chrome/test/data/cups_proxy",
+  ]
 }

chrome/services/cups_proxy/socket_manager.cc

+// Copyright 2019 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/services/cups_proxy/socket_manager.h"
+
+#include <errno.h>
+#include <sys/poll.h>
+#include <sys/socket.h>
+#include <sys/un.h>
+
+#include <algorithm>
+#include <memory>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "base/logging.h"
+#include "base/macros.h"
+#include "base/memory/weak_ptr.h"
+#include "base/sequence_checker.h"
+#include "base/task/post_task.h"
+#include "base/threading/sequenced_task_runner_handle.h"
+#include "base/threading/thread_checker.h"
+#include "chrome/services/cups_proxy/public/cpp/cups_util.h"
+#include "chrome/services/cups_proxy/public/cpp/type_conversions.h"
+#include "net/base/completion_repeating_callback.h"
+#include "net/base/io_buffer.h"
+#include "net/http/http_response_headers.h"
+#include "net/http/http_util.h"
+#include "net/socket/unix_domain_client_socket_posix.h"
+#include "net/traffic_annotation/network_traffic_annotation_test_helper.h"
+
+namespace cups_proxy {
+namespace {
+
+// CUPS daemon socket path
+const char kCupsSocketPath[] = "/run/cups/cups.sock";
+
+// Returns true if |response_buffer| represents a full HTTP response.
+bool FinishedReadingResponse(const std::vector<uint8_t>& response_buffer) {
+  std::string response = ipp_converter::ConvertToString(response_buffer);
+  size_t end_of_headers =
+      net::HttpUtil::LocateEndOfHeaders(response.data(), response.size());
+  if (end_of_headers < 0) {
+    return false;
+  }
+
+  std::string raw_headers = net::HttpUtil::AssembleRawHeaders(
+      base::StringPiece(response.data(), end_of_headers));
+  auto parsed_headers =
+      base::MakeRefCounted<net::HttpResponseHeaders>(raw_headers);
+
+  // Check that response contains the full body.
+  size_t content_length = parsed_headers->GetContentLength();
+  if (content_length < 0) {
+    return false;
+  }
+
+  if (response.size() < end_of_headers + content_length) {
+    return false;
+  }
+
+  return true;
+}
+
+// POD representation of an in-flight socket request.
+struct SocketRequest {
+  // Explicitly declared/defined defaults since [chromium-style] flagged this as
+  // a complex struct.
+  SocketRequest();
+  SocketRequest(SocketRequest&& other);
+  ~SocketRequest();
+
+  // Used for writing/reading the IPP request/response.
+  std::vector<uint8_t> response;
+  scoped_refptr<net::DrainableIOBuffer> io_buffer;
+  SocketManagerCallback cb;
+};
+
+// All methods accessing |socket_| must be made on the IO thread.
+// TODO(luum): Consider inner IO-thread object, base::SequenceBound refactor.
+class SocketManagerImpl : public SocketManager {
+ public:
+  explicit SocketManagerImpl(
+      std::unique_ptr<net::UnixDomainClientSocket> socket);
+  ~SocketManagerImpl() override;
+
+  void ProxyToCups(std::vector<uint8_t> request,
+                   SocketManagerCallback cb) override;
+
+ private:
+  // These methods support lazily connecting to the CUPS socket.
+  void ConnectIfNeeded();
+  void Connect();
+  void OnConnect(int result);
+
+  void Write();
+  void OnWrite(int result);
+
+  void Read();
+  void OnRead(int result);
+
+  // Methods for ending a request.
+  void Finish(bool success = true);
+  void Fail(const char* error_message);
+
+  // Sequence this manager runs on, |in_flight_->cb_| is posted here.
+  scoped_refptr<base::SequencedTaskRunner> main_runner_;
+
+  // Single thread task runner the thread-affine |socket_| runs on.
+  scoped_refptr<base::SingleThreadTaskRunner> socket_runner_;
+
+  // Created in sequence but accessed and deleted on IO thread.
+  std::unique_ptr<SocketRequest> in_flight_;
+  std::unique_ptr<net::UnixDomainClientSocket> socket_;
+
+  SEQUENCE_CHECKER(sequence_checker_);
+  base::WeakPtrFactory<SocketManagerImpl> weak_factory_;
+};
+
+// Defaults for SocketRequest.
+SocketRequest::SocketRequest() = default;
+SocketRequest::SocketRequest(SocketRequest&& other) = default;
+SocketRequest::~SocketRequest() = default;
+
+SocketManagerImpl::SocketManagerImpl(
+    std::unique_ptr<net::UnixDomainClientSocket> socket)
+    : socket_(std::move(socket)), weak_factory_(this) {
+  socket_runner_ =
+      base::CreateSingleThreadTaskRunnerWithTraits(base::TaskTraits());
+  DETACH_FROM_SEQUENCE(sequence_checker_);
+}
+
+SocketManagerImpl::~SocketManagerImpl() {}
+
+void SocketManagerImpl::ProxyToCups(std::vector<uint8_t> request,
+                                    SocketManagerCallback cb) {
+  DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
+  DCHECK(!in_flight_);  // Only handles one request at a time.
+
+  // Lazily save the main runner, needed to post |cb| to.
+  if (!main_runner_) {
+    main_runner_ = base::SequencedTaskRunnerHandle::Get();
+  }
+
+  // Save request.
+  in_flight_ = std::make_unique<SocketRequest>();
+  in_flight_->cb = std::move(cb);
+
+  // Fill io_buffer with request to write.
+  in_flight_->io_buffer = base::MakeRefCounted<net::DrainableIOBuffer>(
+      base::MakeRefCounted<net::IOBuffer>(request.size()), request.size());
+  std::copy(request.begin(), request.end(), in_flight_->io_buffer->data());
+
+  socket_runner_->PostTask(FROM_HERE,
+                           base::BindOnce(&SocketManagerImpl::ConnectIfNeeded,
+                                          weak_factory_.GetWeakPtr()));
+}
+
+// Separate method since we need to check socket_ on the socket thread.
+void SocketManagerImpl::ConnectIfNeeded() {
+  DCHECK(socket_runner_->BelongsToCurrentThread());
+
+  // If |socket_| isn't connected yet, connect it.
+  if (!socket_->IsConnected()) {
+    return Connect();
+  }
+
+  // Write request to CUPS.
+  return Write();
+}
+
+void SocketManagerImpl::Connect() {
+  DCHECK(socket_runner_->BelongsToCurrentThread());
+
+  int result = socket_->Connect(base::BindOnce(&SocketManagerImpl::OnConnect,
+                                               weak_factory_.GetWeakPtr()));
+  if (result != net::ERR_IO_PENDING) {
+    return OnConnect(result);
+  }
+}
+
+void SocketManagerImpl::OnConnect(int result) {
+  DCHECK(socket_runner_->BelongsToCurrentThread());
+  DCHECK(in_flight_);
+
+  if (result < 0) {
+    return Fail("Failed to connect to the CUPS daemon.");
+  }
+
+  return Write();
+}
+
+void SocketManagerImpl::Write() {
+  DCHECK(socket_runner_->BelongsToCurrentThread());
+
+  int result = socket_->Write(
+      in_flight_->io_buffer.get(), in_flight_->io_buffer->BytesRemaining(),
+      base::BindOnce(&SocketManagerImpl::OnWrite, weak_factory_.GetWeakPtr()),
+      TRAFFIC_ANNOTATION_FOR_TESTS /* Unused NetworkAnnotation */);
+
+  if (result != net::ERR_IO_PENDING) {
+    return OnWrite(result);
+  }
+}
+
+void SocketManagerImpl::OnWrite(int result) {
+  DCHECK(socket_runner_->BelongsToCurrentThread());
+  DCHECK(in_flight_);
+
+  if (result < 0) {
+    return Fail("Failed to write IPP request to the CUPS daemon.");
+  }
+
+  if (result < in_flight_->io_buffer->BytesRemaining()) {
+    in_flight_->io_buffer->DidConsume(result);
+    return Write();
+  }
+
+  // Prime io_buffer for reading.
+  in_flight_->io_buffer = base::MakeRefCounted<net::DrainableIOBuffer>(
+      base::MakeRefCounted<net::IOBuffer>(kHttpMaxBufferSize),
+      kHttpMaxBufferSize);
+
+  // Start reading response from CUPS.
+  return Read();
+}
+
+void SocketManagerImpl::Read() {
+  DCHECK(socket_runner_->BelongsToCurrentThread());
+
+  int result = socket_->Read(
+      in_flight_->io_buffer.get(), kHttpMaxBufferSize,
+      base::BindOnce(&SocketManagerImpl::OnRead, weak_factory_.GetWeakPtr()));
+
+  if (result != net::ERR_IO_PENDING) {
+    return OnRead(result);
+  }
+}
+
+void SocketManagerImpl::OnRead(int num_read) {
+  DCHECK(socket_runner_->BelongsToCurrentThread());
+  DCHECK(in_flight_);
+
+  if (num_read < 0) {
+    return Fail("Failed to read IPP response from the CUPS daemon.");
+  }
+
+  // Save new response data.
+  std::copy(in_flight_->io_buffer->data(),
+            in_flight_->io_buffer->data() + num_read,
+            std::back_inserter(in_flight_->response));
+
+  // If more response left to read, read more.
+  if (!FinishedReadingResponse(in_flight_->response)) {
+    return Read();
+  }
+
+  // Finished, got response.
+  return Finish();
+}
+
+void SocketManagerImpl::Finish(bool success) {
+  DCHECK(socket_runner_->BelongsToCurrentThread());
+
+  base::OnceClosure cb;
+  if (success) {
+    cb = base::BindOnce(std::move(in_flight_->cb),
+                        std::move(in_flight_->response));
+  } else {
+    cb = base::BindOnce(std::move(in_flight_->cb), base::nullopt);
+  }
+
+  // Post callback back to main sequence.
+  main_runner_->PostTask(FROM_HERE, std::move(cb));
+
+  // Discard this request while still on IO thread.
+  in_flight_.reset();
+}
+
+void SocketManagerImpl::Fail(const char* error_message) {
+  DVLOG(1) << "SocketManager Error: " << error_message;
+  return Finish(false /* Fail this request */);
+}
+
+}  // namespace
+
+std::unique_ptr<SocketManager> SocketManager::Create() {
+  return std::make_unique<SocketManagerImpl>(
+      std::make_unique<net::UnixDomainClientSocket>(
+          kCupsSocketPath, false /* not abstract namespace */));
+}
+
+std::unique_ptr<SocketManager> SocketManager::CreateForTesting(
+    std::unique_ptr<net::UnixDomainClientSocket> socket) {
+  return std::make_unique<SocketManagerImpl>(std::move(socket));
+}
+
+}  // namespace cups_proxy

chrome/services/cups_proxy/socket_manager.h

+// Copyright 2019 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_SERVICES_CUPS_PROXY_SOCKET_MANAGER_H_
+#define CHROME_SERVICES_CUPS_PROXY_SOCKET_MANAGER_H_
+
+#include <memory>
+#include <vector>
+
+#include "base/callback.h"
+#include "base/optional.h"
+
+namespace net {
+class UnixDomainClientSocket;
+}  // namespace net
+
+namespace cups_proxy {
+
+using SocketManagerCallback =
+    base::OnceCallback<void(base::Optional<std::vector<uint8_t>>)>;
+
+// This manager proxies IPP requests to the CUPS daemon and asynchronously
+// responds with the IPP response. This class can be created anywhere but must
+// be accessed from a sequenced context.
+class SocketManager {
+ public:
+  // Factory function.
+  static std::unique_ptr<SocketManager> Create();
+
+  // Factory function that allows injected dependencies, for testing.
+  static std::unique_ptr<SocketManager> CreateForTesting(
+      std::unique_ptr<net::UnixDomainClientSocket> socket);
+
+  virtual ~SocketManager() = default;
+
+  // Attempts to send |request| to the CUPS Daemon, and return its response via
+  // |cb|. |cb| will run on the caller's sequence. Note: Can only handle 1
+  // inflight request at a time; attempts to proxy more will DCHECK.
+  virtual void ProxyToCups(std::vector<uint8_t> request,
+                           SocketManagerCallback cb) = 0;
+};
+
+}  // namespace cups_proxy
+
+#endif  // CHROME_SERVICES_CUPS_PROXY_SOCKET_MANAGER_H_

chrome/services/cups_proxy/socket_manager_unittest.cc

+// Copyright 2019 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 <algorithm>
+#include <string>
+#include <utility>
+
+#include "base/files/file_util.h"
+#include "base/path_service.h"
+#include "base/threading/sequenced_task_runner_handle.h"
+#include "base/threading/thread_restrictions.h"
+#include "chrome/common/chrome_paths.h"
+#include "chrome/services/cups_proxy/public/cpp/type_conversions.h"
+#include "chrome/services/cups_proxy/socket_manager.h"
+#include "net/base/io_buffer.h"
+#include "net/socket/unix_domain_client_socket_posix.h"
+#include "net/test/test_with_scoped_task_environment.h"
+#include "testing/gtest/include/gtest/gtest.h"
+
+namespace cups_proxy {
+namespace {
+
+// CupsProxy testing data relative path.
+const base::FilePath::CharType kCupsProxyDataDirectory[] =
+    FILE_PATH_LITERAL("cups_proxy");
+
+// Returns base::nullopt on failure.
+base::Optional<std::string> GetTestFile(std::string test_name) {
+  base::ScopedAllowBlockingForTesting allow_blocking;
+
+  // Build file path.
+  base::FilePath path;
+  if (!base::PathService::Get(chrome::DIR_TEST_DATA, &path)) {
+    return base::nullopt;
+  }
+
+  path = path.Append(kCupsProxyDataDirectory)
+             .Append(FILE_PATH_LITERAL(test_name))
+             .AddExtension(FILE_PATH_LITERAL(".bin"));
+
+  // Read in file contents.
+  std::string contents;
+  if (!base::ReadFileToString(path, &contents)) {
+    return base::nullopt;
+  }
+
+  return contents;
+}
+
+}  // namespace
+
+// Gives full control over the "CUPS daemon" in this test.
+class FakeSocket : public net::UnixDomainClientSocket {
+ public:
+  FakeSocket() : UnixDomainClientSocket("", false) /* Dummy values */ {}
+  ~FakeSocket() override = default;
+
+  // Saves expected request and corresponding response to send back.
+  void set_request(base::StringPiece request) { request_ = request; }
+  void set_response(base::StringPiece response) { response_ = response; }
+
+  // Controls whether each method runs synchronously or asynchronously.
+  void set_connect_async() { connect_async = true; }
+  void set_read_async() { read_async = true; }
+  void set_write_async() { write_async = true; }
+
+  // net::UnixDomainClientSocket overrides.
+  bool IsConnected() const override { return is_connected; }
+
+  int Connect(net::CompletionOnceCallback callback) override {
+    if (is_connected) {
+      // Should've checked IsConnected first.
+      return net::ERR_FAILED;
+    }
+
+    is_connected = true;
+
+    // Sync
+    if (!connect_async) {
+      return net::OK;
+    }
+
+    // Async
+    base::SequencedTaskRunnerHandle::Get()->PostTask(
+        FROM_HERE,
+        base::BindOnce(&FakeSocket::OnAsyncCallback, base::Unretained(this),
+                       std::move(callback), net::OK));
+    return net::ERR_IO_PENDING;
+  }
+
+  int Read(net::IOBuffer* buf,
+           int buf_len,
+           net::CompletionOnceCallback callback) override {
+    if (!is_connected) {
+      return net::ERR_FAILED;
+    }
+
+    size_t num_to_read =
+        std::min(response_.size(), static_cast<size_t>(buf_len));
+    std::copy(response_.begin(), response_.begin() + num_to_read, buf->data());
+    response_.remove_prefix(num_to_read);
+
+    // Sync
+    if (!read_async) {
+      return num_to_read;
+    }
+
+    // Async
+    base::SequencedTaskRunnerHandle::Get()->PostTask(
+        FROM_HERE,
+        base::BindOnce(&FakeSocket::OnAsyncCallback, base::Unretained(this),
+                       std::move(callback), num_to_read));
+    return net::ERR_IO_PENDING;
+  }
+
+  int Write(net::IOBuffer* buf,
+            int buf_len,
+            net::CompletionOnceCallback callback,
+            const net::NetworkTrafficAnnotationTag& unused) override {
+    if (!is_connected) {
+      return net::ERR_FAILED;
+    }
+
+    // Checks that |buf| holds (part of) the expected request.
+    if (!std::equal(buf->data(), buf->data() + buf_len, request_.begin())) {
+      return net::ERR_FAILED;
+    }
+
+    // Arbitrary maximum write buffer size; just forcing partial writes.
+    const size_t kMaxWriteSize = 100;
+    size_t num_to_write = std::min(kMaxWriteSize, static_cast<size_t>(buf_len));
+    request_.remove_prefix(num_to_write);
+
+    // Sync
+    if (!write_async) {
+      return num_to_write;
+    }
+
+    // Async
+    base::SequencedTaskRunnerHandle::Get()->PostTask(
+        FROM_HERE,
+        base::BindOnce(&FakeSocket::OnAsyncCallback, base::Unretained(this),
+                       std::move(callback), num_to_write));
+    return net::ERR_IO_PENDING;
+  }
+
+  // Generic callback used to force called methods to return asynchronously.
+  void OnAsyncCallback(net::CompletionOnceCallback callback, int net_code) {
+    std::move(callback).Run(net_code);
+  }
+
+ private:
+  bool is_connected = false;
+  bool connect_async = false, read_async = false, write_async = false;
+  base::StringPiece request_, response_;
+};
+
+class SocketManagerTest : public testing::Test {
+ public:
+  SocketManagerTest() : weak_factory_(this) {
+    std::unique_ptr<FakeSocket> socket = std::make_unique<FakeSocket>();
+    socket_ = socket.get();
+
+    manager_ = SocketManager::CreateForTesting(std::move(socket));
+  }
+
+  base::Optional<std::vector<uint8_t>> ProxyToCups(std::string request) {
+    std::vector<uint8_t> request_as_bytes =
+        ipp_converter::ConvertToByteBuffer(request);
+    base::Optional<std::vector<uint8_t>> response;
+
+    base::RunLoop run_loop;
+    manager_->ProxyToCups(std::move(request_as_bytes),
+                          base::BindOnce(&SocketManagerTest::OnProxyToCups,
+                                         weak_factory_.GetWeakPtr(),
+                                         run_loop.QuitClosure(), &response));
+    run_loop.Run();
+    return response;
+  }
+
+ protected:
+  // Must be first member.
+  base::test::ScopedTaskEnvironment scoped_task_environment_;
+
+  void OnProxyToCups(base::OnceClosure finish_cb,
+                     base::Optional<std::vector<uint8_t>>* ret,
+                     base::Optional<std::vector<uint8_t>> result) {
+    *ret = std::move(result);
+    std::move(finish_cb).Run();
+  }
+
+  // Not owned.
+  FakeSocket* socket_;
+
+  std::unique_ptr<SocketManager> manager_;
+  base::WeakPtrFactory<SocketManagerTest> weak_factory_;
+};
+
+// "basic_handshake" test file contains a simple HTTP request sent by libCUPS,
+// copied below for convenience:
+//
+// POST / HTTP/1.1
+// Content-Length: 72
+// Content-Type: application/ipp
+// Date: Thu, 04 Oct 2018 20:25:59 GMT
+// Host: localhost:0
+// User-Agent: CUPS/2.3b1 (Linux 4.4.159-15303-g65f4b5a7b3d3; i686) IPP/2.0
+//
+// @Gattributes-charsetutf-8Hattributes-natural-languageen
+
+// All socket accesses are resolved synchronously.
+TEST_F(SocketManagerTest, SyncEverything) {
+  // Read request & response
+  base::Optional<std::string> http_handshake = GetTestFile("basic_handshake");
+  EXPECT_TRUE(http_handshake);
+
+  // Pre-load |socket_| with request/response.
+  // TODO(crbug.com/495409): Test with actual http response.
+  socket_->set_request(*http_handshake);
+  socket_->set_response(*http_handshake);
+
+  auto response = ProxyToCups(*http_handshake);
+  EXPECT_TRUE(response);
+  EXPECT_EQ(*response, ipp_converter::ConvertToByteBuffer(*http_handshake));
+}
+
+TEST_F(SocketManagerTest, AsyncEverything) {
+  auto http_handshake = GetTestFile("basic_handshake");
+  EXPECT_TRUE(http_handshake);
+
+  socket_->set_request(*http_handshake);
+  socket_->set_response(*http_handshake);
+
+  // Set all |socket_| calls to run asynchronously.
+  socket_->set_connect_async();
+  socket_->set_read_async();
+  socket_->set_write_async();
+
+  auto response = ProxyToCups(*http_handshake);
+  EXPECT_TRUE(response);
+  EXPECT_EQ(*response, ipp_converter::ConvertToByteBuffer(*http_handshake));
+}
+
+}  // namespace cups_proxy