Add ReportQueue

This change adds ReportQueue for generating reports.

File Changes:
~ chrome/browser/
| ~ BUILD.gn
| | Add report_queue.h to the buildable unit.
| ~ policy/messaging_layer/
| | ~ proto/record.proto
| | | Change EncryptedRecord::encrypted_wrapped_record from byte to
| | | string. Behind the scenes they are treated the same, but in
| | | practice it is easier to work with std::string().
| | + proto/test.proto
| | | Adds a test proto for testing.
| | + public/report_queue.h/cc
| | | Add ReportQueue, temporary StorageModule and temporary
| | | EncryptionModule.
| | + public/report_queue_unittest.cc
| | | Add tests for ReportQueue
| | ~ public/report_queue_configuration_unittest.cc
| | | Import missing library.
~ chrome/test/BUILD.gn
| Add unit tests and test proto.

Bug: chromium:1078512
Change-Id: If3fff7ac59e7d4a486d47ab87148cb27020c2866
Reviewed-on: https://chromium-review.googlesource.com/c/chromium/src/+/2191201Reviewed-by: Sergey Poromov <poromov@chromium.org>
Commit-Queue: Zach Trudo <zatrudo@google.com>
Cr-Commit-Position: refs/heads/master@{#774354}

chrome/browser/BUILD.gn

@@ -1264,6 +1264,8 @@ static_library("browser") {
     "policy/homepage_location_policy_handler.h",
     "policy/javascript_policy_handler.cc",
     "policy/javascript_policy_handler.h",
+    "policy/messaging_layer/public/report_queue.cc",
+    "policy/messaging_layer/public/report_queue.h",
     "policy/messaging_layer/public/report_queue_configuration.cc",
     "policy/messaging_layer/public/report_queue_configuration.h",
     "policy/messaging_layer/util/status.cc",

chrome/browser/policy/messaging_layer/proto/record.proto

@@ -40,12 +40,12 @@ message WrappedRecord {
   // Record Digest (required)
   // SHA256 hash used to validate that the record has been retrieved without
   // being manipulated while it was on the device or during transfer.
-  optional uint64 record_digest = 2;
+  optional string record_digest = 2;
 
   // Last record digest (required)
   // Created by client and used by server to verify that the sequence of records
   // has not been tampered with.
-  optional uint64 last_record_digest = 3;
+  optional string last_record_digest = 3;
 }
 
 // Information about how the record was encrypted.
@@ -85,7 +85,7 @@ message SequencingInformation {
 message EncryptedRecord {
   // Encrypted Wrapped Record (required)
   // |WrappedRecord| encrypted with the |encryption_key| in |encryption_info|.
-  optional bytes encrypted_wrapped_record = 1;
+  optional string encrypted_wrapped_record = 1;
 
   optional EncryptionInfo encryption_info = 2;
 

chrome/browser/policy/messaging_layer/proto/test.proto

+// Copyright 2020 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.
+
+syntax = "proto2";
+
+option optimize_for = LITE_RUNTIME;
+
+package reporting_messaging_layer.test;
+
+// A simple test proto with a string in it.
+message TestMessage {
+  optional string test = 1;
+}

chrome/browser/policy/messaging_layer/public/report_queue.cc

+// Copyright 2020 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/policy/messaging_layer/public/report_queue.h"
+
+#include <memory>
+
+#include "base/json/json_writer.h"
+#include "base/memory/ptr_util.h"
+#include "base/memory/scoped_refptr.h"
+#include "base/sequence_checker.h"
+#include "base/strings/strcat.h"
+#include "base/task/task_traits.h"
+#include "base/task/thread_pool.h"
+#include "base/values.h"
+#include "chrome/browser/policy/messaging_layer/proto/record.pb.h"
+#include "chrome/browser/policy/messaging_layer/public/report_queue_configuration.h"
+#include "chrome/browser/policy/messaging_layer/util/status.h"
+#include "chrome/browser/policy/messaging_layer/util/status_macros.h"
+#include "chrome/browser/policy/messaging_layer/util/statusor.h"
+#include "components/policy/core/common/cloud/dm_token.h"
+#include "components/policy/proto/record_constants.pb.h"
+#include "crypto/sha2.h"
+
+namespace reporting {
+
+using base::MakeRefCounted;
+using reporting_messaging_layer::EncryptedRecord;
+using reporting_messaging_layer::Record;
+using reporting_messaging_layer::WrappedRecord;
+
+std::unique_ptr<ReportQueue> ReportQueue::Create(
+    std::unique_ptr<ReportQueueConfiguration> config,
+    scoped_refptr<StorageModule> storage,
+    scoped_refptr<EncryptionModule> encryption) {
+  return base::WrapUnique<ReportQueue>(
+      new ReportQueue(std::move(config), storage, encryption));
+}
+
+ReportQueue::~ReportQueue() = default;
+
+ReportQueue::ReportQueue(std::unique_ptr<ReportQueueConfiguration> config,
+                         scoped_refptr<StorageModule> storage,
+                         scoped_refptr<EncryptionModule> encryption)
+    : config_(std::move(config)),
+      storage_(storage),
+      encryption_(encryption),
+      sequenced_task_runner_(
+          base::ThreadPool::CreateSequencedTaskRunner(base::TaskTraits())) {
+  DETACH_FROM_SEQUENCE(sequence_checker_);
+}
+
+Status ReportQueue::Enqueue(base::StringPiece record,
+                            EnqueueCallback callback) {
+  return AddRecord(record, std::move(callback));
+}
+
+Status ReportQueue::Enqueue(const base::Value& record,
+                            EnqueueCallback callback) {
+  std::string json_record;
+  if (!base::JSONWriter::Write(record, &json_record)) {
+    return Status(error::INVALID_ARGUMENT,
+                  "Provided record was not convertable to a std::string");
+  }
+  return AddRecord(json_record, std::move(callback));
+}
+
+Status ReportQueue::Enqueue(google::protobuf::MessageLite* record,
+                            EnqueueCallback callback) {
+  std::string protobuf_record;
+  if (!record->SerializeToString(&protobuf_record)) {
+    return Status(error::INVALID_ARGUMENT,
+                  "Unabled to serialize record to string. Most likely due to "
+                  "unset required fields.");
+  }
+  return AddRecord(protobuf_record, std::move(callback));
+}
+
+Status ReportQueue::AddRecord(base::StringPiece record,
+                              EnqueueCallback callback) {
+  if (!sequenced_task_runner_->PostTask(
+          FROM_HERE, base::BindOnce(&ReportQueue::SendRecordToStorage,
+                                    base::Unretained(this), std::string(record),
+                                    std::move(callback)))) {
+    return Status(error::INTERNAL, "Failed to post the record for processing.");
+  }
+  return Status::StatusOK();
+}
+
+void ReportQueue::SendRecordToStorage(std::string record,
+                                      EnqueueCallback callback) {
+  ASSIGN_OR_ONCE_CALLBACK_AND_RETURN(WrappedRecord wrapped_record, callback,
+                                     WrapRecord(record));
+
+  ASSIGN_OR_ONCE_CALLBACK_AND_RETURN(EncryptedRecord encrypted_record, callback,
+                                     EncryptRecord(wrapped_record));
+
+  storage_->AddRecord(encrypted_record, config_->priority(),
+                      std::move(callback));
+}
+
+StatusOr<WrappedRecord> ReportQueue::WrapRecord(base::StringPiece record_data) {
+  WrappedRecord wrapped_record;
+
+  Record* record = wrapped_record.mutable_record();
+  record->set_data(std::string(record_data));
+  record->set_destination(config_->destination());
+  record->set_dm_token(config_->dm_token().value());
+
+  std::string record_digest;
+  record->SerializeToString(&record_digest);
+  wrapped_record.set_record_digest(crypto::SHA256HashString(record_digest));
+
+  ASSIGN_OR_RETURN(*wrapped_record.mutable_last_record_digest(),
+                   GetLastRecordDigest());
+  return wrapped_record;
+}
+
+StatusOr<std::string> ReportQueue::GetLastRecordDigest() {
+  // TODO(b/153659559) Getting the actual last record digest will come later.
+  // For now we just set to a string.
+  return "LastRecordDigest";
+}
+
+StatusOr<EncryptedRecord> ReportQueue::EncryptRecord(
+    WrappedRecord wrapped_record) {
+  std::string serialized_wrapped_record;
+  wrapped_record.SerializeToString(&serialized_wrapped_record);
+
+  ASSIGN_OR_RETURN(std::string encrypted_string_record,
+                   encryption_->EncryptRecord(serialized_wrapped_record));
+
+  EncryptedRecord encrypted_record;
+  encrypted_record.set_encrypted_wrapped_record(encrypted_string_record);
+  return encrypted_record;
+}
+
+}  // namespace reporting

chrome/browser/policy/messaging_layer/public/report_queue.h

+// Copyright 2020 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_POLICY_MESSAGING_LAYER_PUBLIC_REPORT_QUEUE_H_
+#define CHROME_BROWSER_POLICY_MESSAGING_LAYER_PUBLIC_REPORT_QUEUE_H_
+
+#include <memory>
+
+#include "base/memory/ref_counted.h"
+#include "base/memory/scoped_refptr.h"
+#include "base/sequence_checker.h"
+#include "base/sequenced_task_runner.h"
+#include "base/values.h"
+#include "chrome/browser/policy/messaging_layer/proto/record.pb.h"
+#include "chrome/browser/policy/messaging_layer/public/report_queue_configuration.h"
+#include "chrome/browser/policy/messaging_layer/util/status.h"
+#include "chrome/browser/policy/messaging_layer/util/statusor.h"
+#include "components/policy/proto/record_constants.pb.h"
+#include "third_party/protobuf/src/google/protobuf/message_lite.h"
+
+namespace reporting {
+
+// TODO(b/153659559) Temporary StorageModule until the real one is ready.
+class StorageModule : public base::RefCounted<StorageModule> {
+ public:
+  StorageModule() = default;
+
+  StorageModule(const StorageModule& other) = delete;
+  StorageModule& operator=(const StorageModule& other) = delete;
+
+  // AddRecord will add |record| to the |StorageModule| according to the
+  // provided |priority|. On completion, |callback| will be called.
+  virtual void AddRecord(reporting_messaging_layer::EncryptedRecord record,
+                         reporting_messaging_layer::Priority priority,
+                         base::OnceCallback<void(Status)> callback) = 0;
+
+ protected:
+  virtual ~StorageModule() = default;
+
+ private:
+  friend base::RefCounted<StorageModule>;
+};
+
+// TODO(b/153659559) Temporary EncryptionModule until the real one is ready.
+class EncryptionModule : public base::RefCounted<EncryptionModule> {
+ public:
+  EncryptionModule() = default;
+
+  EncryptionModule(const EncryptionModule& other) = delete;
+  EncryptionModule& operator=(const EncryptionModule& other) = delete;
+
+  // EncryptRecord will attempt to encrypt the provided |record|. On success the
+  // return value will contain the encrypted string.
+  virtual StatusOr<std::string> EncryptRecord(base::StringPiece record) = 0;
+
+ protected:
+  virtual ~EncryptionModule() = default;
+
+ private:
+  friend base::RefCounted<EncryptionModule>;
+};
+
+// A |ReportQueue| is configured with a |ReportQueueConfiguration|.  A
+// |ReportQueue| allows a user to |Enqueue| a message for delivery to a handler
+// specified by the |Destination| held by the provided
+// |ReportQueueConfiguration|. |ReportQueue| handles scheduling encryption,
+// storage, and delivery.
+//
+// TODO(b/156099331): |ReportQueue|s should not be created or managed directly,
+// they should instead be created with the |ReportingClientLibrary| which does
+// not exist at this time.
+//
+// Example Usage:
+// Status SendMessage(google::protobuf::ImportantMessage important_message,
+//                    base::OnceCallback<void(Status)> callback) {
+//   ASSIGN_OR_RETURN(std::unique_ptr<ReportQueueConfiguration> config,
+//                  ReportQueueConfiguration::Create(...));
+//   ASSIGN_OR_RETURN(std::unique_ptr<ReportQueue> report_queue,
+//                  ReportingClientLibrary::CreateReportQueue(config));
+//   return report_queue->Enqueue(important_message, callback);
+// }
+//
+// Enqueue can also be used with a |base::Value| or |std::string|.
+class ReportQueue {
+ public:
+  // An EnqueueCallbacks are called on the completion of any |Enqueue| call.
+  using EnqueueCallback = base::OnceCallback<void(Status)>;
+
+  // Factory
+  static std::unique_ptr<ReportQueue> Create(
+      std::unique_ptr<ReportQueueConfiguration> config,
+      scoped_refptr<StorageModule> storage,
+      scoped_refptr<EncryptionModule> encryption);
+
+  ~ReportQueue();
+  ReportQueue(const ReportQueue& other) = delete;
+  ReportQueue& operator=(const ReportQueue& other) = delete;
+
+  // Enqueue asynchronously encrypts, stores, and delivers a record. Enqueue
+  // will return an OK status if the task is successfully scheduled. The
+  // |callback| will be called on any errors during encryption or storage. If
+  // storage is successful |callback| will be called with an OK status.
+  //
+  // The current destinations have the following data requirements:
+  // (destination : requirement)
+  // UPLOAD_EVENTS : UploadEventsRequest
+  //
+  // |record| will be sent as a string with no conversion.
+  Status Enqueue(base::StringPiece record, EnqueueCallback callback);
+
+  // |record| will be converted to a JSON string with base::JsonWriter::Write.
+  Status Enqueue(const base::Value& record, EnqueueCallback callback);
+
+  // |record| will be converted to a string with SerializeToString(). The
+  // handler is responsible for converting the record back to a proto with a
+  // ParseFromString() call.
+  Status Enqueue(google::protobuf::MessageLite* record,
+                 EnqueueCallback callback);
+
+ private:
+  ReportQueue(std::unique_ptr<ReportQueueConfiguration> config,
+              scoped_refptr<StorageModule> storage,
+              scoped_refptr<EncryptionModule> encryption);
+
+  Status AddRecord(base::StringPiece record, EnqueueCallback callback);
+  void SendRecordToStorage(std::string record, EnqueueCallback callback);
+
+  StatusOr<reporting_messaging_layer::WrappedRecord> WrapRecord(
+      base::StringPiece record_data);
+  StatusOr<std::string> GetLastRecordDigest();
+  StatusOr<reporting_messaging_layer::EncryptedRecord> EncryptRecord(
+      reporting_messaging_layer::WrappedRecord wrapped_record);
+
+  std::unique_ptr<ReportQueueConfiguration> config_;
+  scoped_refptr<StorageModule> storage_;
+  scoped_refptr<EncryptionModule> encryption_;
+  SEQUENCE_CHECKER(sequence_checker_);
+
+  scoped_refptr<base::SequencedTaskRunner> sequenced_task_runner_;
+};
+
+}  // namespace reporting
+
+#endif  // CHROME_BROWSER_POLICY_MESSAGING_LAYER_PUBLIC_REPORT_QUEUE_H_

chrome/browser/policy/messaging_layer/public/report_queue_configuration_unittest.cc

@@ -8,6 +8,7 @@
 
 #include "chrome/browser/policy/messaging_layer/util/statusor.h"
 #include "components/policy/core/common/cloud/dm_token.h"
+#include "components/policy/proto/record_constants.pb.h"
 #include "testing/gtest/include/gtest/gtest.h"
 
 namespace reporting {

chrome/test/BUILD.gn

@@ -35,6 +35,7 @@ import("//ppapi/buildflags/buildflags.gni")
 import("//rlz/buildflags/buildflags.gni")
 import("//testing/libfuzzer/fuzzer_test.gni")
 import("//testing/test.gni")
+import("//third_party/protobuf/proto_library.gni")
 import("//third_party/widevine/cdm/widevine.gni")
 import("//ui/base/ui_features.gni")
 import("//ui/gl/features.gni")
@@ -3297,6 +3298,7 @@ test("unit_tests") {
     "../browser/policy/homepage_location_policy_handler_unittest.cc",
     "../browser/policy/javascript_policy_handler_unittest.cc",
     "../browser/policy/messaging_layer/public/report_queue_configuration_unittest.cc",
+    "../browser/policy/messaging_layer/public/report_queue_unittest.cc",
     "../browser/policy/messaging_layer/util/status_macros_unittest.cc",
     "../browser/policy/messaging_layer/util/status_unittest.cc",
     "../browser/policy/messaging_layer/util/statusor_unittest.cc",
@@ -3681,6 +3683,7 @@ test("unit_tests") {
   defines = []
 
   deps = [
+    ":messaging_test_proto",
     ":test_support",
     ":test_support_unit",
     "//base:base_stack_sampling_profiler_test_util",
@@ -5658,8 +5661,6 @@ if (!is_android) {
     }
   }
 
-  import("//third_party/protobuf/proto_library.gni")
-
   proto_library("test_proto") {
     sources = [ "../common/safe_browsing/ipc_protobuf_message_test.proto" ]
   }
@@ -7027,3 +7028,7 @@ if (!is_android) {
     outputs = [ "$root_out_dir/run_xr_browser_tests.py" ]
   }
 }
+
+proto_library("messaging_test_proto") {
+  sources = [ "//chrome/browser/policy/messaging_layer/proto/test.proto" ]
+}