Handle generation and digests.

Whenever certain record is written generation and its digest are stored
in metadata file, and then all previous metadata file(s) are
asynchronously deleted.
When reopening the StorageQueue, metadata file matching the last
sequencing number is used to restore generation and digest for future
writes.
The code does not handle errors yet - missing or corrupt metadata file,
incomplete data file, etc. In the future we will add code to reset
generation id and eliminate last digest.

MockClient in tests makes sure generation_id is the same for all records.
It also verify records own digest; last record digest is not verified
yet, because the test expects some records to be duplicated.

Bug: b:153364303
Bug: b:153659559
Change-Id: I80d8e7f03097bcc4a510f21e0a546876b3104960
Reviewed-on: https://chromium-review.googlesource.com/c/chromium/src/+/2388716
Commit-Queue: Leonid Baraz <lbaraz@chromium.org>
Reviewed-by: Zach Trudo <zatrudo@google.com>
Cr-Commit-Position: refs/heads/master@{#804486}

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

@@ -101,10 +101,4 @@ Record ReportQueue::AugmentRecord(base::StringPiece record_data) {
   return 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";
-}
-
 }  // namespace reporting

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

@@ -79,7 +79,6 @@ class ReportQueue {
   void SendRecordToStorage(base::StringPiece record, EnqueueCallback callback);
 
   reporting::Record AugmentRecord(base::StringPiece record_data);
-  StatusOr<std::string> GetLastRecordDigest();
 
   std::unique_ptr<ReportQueueConfiguration> config_;
   scoped_refptr<StorageModule> storage_;

chrome/browser/policy/messaging_layer/storage/storage_queue.h

@@ -139,7 +139,8 @@ class StorageQueue : public base::RefCountedThreadSafe<StorageQueue> {
   // caller can "fire and forget" it (|completion_cb| allows to verify that
   // record has been successfully enqueued). If file is going to become too
   // large, it is closed and new file is created.
-  // Helper methods: AssignLastFile, WriteHeaderAndBlock.
+  // Helper methods: AssignLastFile, WriteHeaderAndBlock, OpenNewWriteableFile,
+  // WriteMetadata, DeleteOutdatedMetadata.
   void Write(Record record, base::OnceCallback<void(Status)> completion_cb);
 
   // Confirms acceptance of the records up to |seq_number| (inclusively).
@@ -247,9 +248,14 @@ class StorageQueue : public base::RefCountedThreadSafe<StorageQueue> {
   // Must be called once and only once after construction.
   // Returns OK or error status, if anything failed to initialize.
   // Called once, during initialization.
-  // Helper methods: EnumerateDataFiles, ScanLastFile.
+  // Helper methods: EnumerateDataFiles, ScanLastFile, RestoreMetadata.
   Status Init();
 
+  // Attaches last record digest to the given record (does not exist at a
+  // generation start). Calculates the given record digest and stores it
+  // as the last one for the next record.
+  void UpdateRecordDigest(WrappedRecord* wrapped_record);
+
   // Helper method for Init(): enumerates all data files in the directory.
   // Valid file names are <prefix>.<seq_number>, any other names are ignored.
   Status EnumerateDataFiles();
@@ -270,6 +276,23 @@ class StorageQueue : public base::RefCountedThreadSafe<StorageQueue> {
   // writeable file, adding it to |files_|.
   StatusOr<scoped_refptr<SingleFile>> OpenNewWriteableFile();
 
+  // Helper method for Write(): stores a file with metadata to match the
+  // incoming new record. Synchronously composes metadata to record, then
+  // asynchronously writes it into a file with next sequencing number and then
+  // notifies the Write operation that it can now complete. After that it
+  // asynchronously deletes all other files with lower sequencing number
+  // (multiple Writes can see the same files and attempt to delete them, and
+  // that is not an error).
+  Status WriteMetadata();
+
+  // Helper method for Init(): locates file with metadata that matches the
+  // last sequencing number and loads metadat from it.
+  Status RestoreMetadata();
+
+  // Helper method for Write(): deletes meta files up to, but not including
+  // |seq_number_to_keep|. Any errors are ignored.
+  void DeleteOutdatedMetadata(uint64_t seq_number_to_keep);
+
   // Helper method for Write(): composes record header and writes it to the
   // file, followed by data.
   Status WriteHeaderAndBlock(base::StringPiece data,
@@ -296,6 +319,16 @@ class StorageQueue : public base::RefCountedThreadSafe<StorageQueue> {
   // Immutable options, stored at the time of creation.
   const Options options_;
 
+  // Current generation id, unique per device and queue.
+  // Set up once during initialization by reading from the 'gen_id.NNNN' file
+  // matching the last seq number, or generated anew as a random number if no
+  // such file found (files do not match the id).
+  uint64_t generation_id_ = 0;
+
+  // Digest of the last written record (loaded at queue initialization, absent
+  // if the new generation has just started, and no records where stored yet).
+  base::Optional<std::string> last_record_digest_;
+
   // Next sequencing number to store (not assigned yet).
   uint64_t next_seq_number_ = 0;
 

chrome/browser/policy/messaging_layer/storage/storage_queue_unittest.cc

@@ -9,12 +9,16 @@
 #include <vector>
 
 #include "base/files/scoped_temp_dir.h"
+#include "base/optional.h"
+#include "base/strings/strcat.h"
+#include "base/strings/string_number_conversions.h"
 #include "base/synchronization/waitable_event.h"
 #include "base/test/task_environment.h"
 #include "chrome/browser/policy/messaging_layer/encryption/test_encryption_module.h"
 #include "chrome/browser/policy/messaging_layer/util/status.h"
 #include "chrome/browser/policy/messaging_layer/util/statusor.h"
 #include "components/policy/proto/record.pb.h"
+#include "crypto/sha2.h"
 #include "testing/gmock/include/gmock/gmock.h"
 #include "testing/gtest/include/gtest/gtest.h"
 
@@ -84,6 +88,44 @@ class MockUploadClient : public StorageQueue::UploaderInterface {
     WrappedRecord wrapped_record;
     ASSERT_TRUE(wrapped_record.ParseFromString(
         encrypted_record.ValueOrDie().encrypted_wrapped_record()));
+    // Verify generation match.
+    if (generation_id_.has_value() &&
+        generation_id_.value() != encrypted_record.ValueOrDie()
+                                      .sequencing_information()
+                                      .generation_id()) {
+      std::move(processed_cb)
+          .Run(UploadRecordFailure(Status(
+              error::DATA_LOSS,
+              base::StrCat({"Generation id mismatch, expected=",
+                            base::NumberToString(generation_id_.value()),
+                            " actual=",
+                            base::NumberToString(encrypted_record.ValueOrDie()
+                                                     .sequencing_information()
+                                                     .generation_id())}))));
+      return;
+    }
+    if (!generation_id_.has_value()) {
+      generation_id_ = encrypted_record.ValueOrDie()
+                           .sequencing_information()
+                           .generation_id();
+    }
+
+    // Verify digest and its match.
+    // Last record digest is not verified yet, since duplicate records are
+    // accepted in this test.
+    {
+      std::string serialized_record;
+      wrapped_record.record().SerializeToString(&serialized_record);
+      const auto record_digest = crypto::SHA256HashString(serialized_record);
+      DCHECK_EQ(record_digest.size(), crypto::kSHA256Length);
+      if (record_digest != wrapped_record.record_digest()) {
+        std::move(processed_cb)
+            .Run(UploadRecordFailure(
+                Status(error::DATA_LOSS, "Record digest mismatch")));
+        return;
+      }
+    }
+
     std::move(processed_cb)
         .Run(UploadRecord(encrypted_record.ValueOrDie()
                               .sequencing_information()
@@ -133,6 +175,7 @@ class MockUploadClient : public StorageQueue::UploaderInterface {
   };
 
  private:
+  base::Optional<uint64_t> generation_id_;
   Sequence test_upload_sequence_;
 };
 

chrome/browser/policy/messaging_layer/storage/storage_unittest.cc

@@ -9,6 +9,9 @@
 #include <vector>
 
 #include "base/files/scoped_temp_dir.h"
+#include "base/optional.h"
+#include "base/strings/strcat.h"
+#include "base/strings/string_number_conversions.h"
 #include "base/synchronization/waitable_event.h"
 #include "base/test/task_environment.h"
 #include "chrome/browser/policy/messaging_layer/encryption/test_encryption_module.h"
@@ -16,7 +19,7 @@
 #include "chrome/browser/policy/messaging_layer/util/statusor.h"
 #include "components/policy/proto/record.pb.h"
 #include "components/policy/proto/record_constants.pb.h"
-
+#include "crypto/sha2.h"
 #include "testing/gmock/include/gmock/gmock.h"
 #include "testing/gtest/include/gtest/gtest.h"
 
@@ -87,6 +90,44 @@ class MockUploadClient : public Storage::UploaderInterface {
     WrappedRecord wrapped_record;
     ASSERT_TRUE(wrapped_record.ParseFromString(
         encrypted_record.ValueOrDie().encrypted_wrapped_record()));
+    // Verify generation match.
+    if (generation_id_.has_value() &&
+        generation_id_.value() != encrypted_record.ValueOrDie()
+                                      .sequencing_information()
+                                      .generation_id()) {
+      std::move(processed_cb)
+          .Run(UploadRecordFailure(Status(
+              error::DATA_LOSS,
+              base::StrCat({"Generation id mismatch, expected=",
+                            base::NumberToString(generation_id_.value()),
+                            " actual=",
+                            base::NumberToString(encrypted_record.ValueOrDie()
+                                                     .sequencing_information()
+                                                     .generation_id())}))));
+      return;
+    }
+    if (!generation_id_.has_value()) {
+      generation_id_ = encrypted_record.ValueOrDie()
+                           .sequencing_information()
+                           .generation_id();
+    }
+
+    // Verify digest and its match.
+    // Last record digest is not verified yet, since duplicate records are
+    // accepted in this test.
+    {
+      std::string serialized_record;
+      wrapped_record.record().SerializeToString(&serialized_record);
+      const auto record_digest = crypto::SHA256HashString(serialized_record);
+      DCHECK_EQ(record_digest.size(), crypto::kSHA256Length);
+      if (record_digest != wrapped_record.record_digest()) {
+        std::move(processed_cb)
+            .Run(UploadRecordFailure(
+                Status(error::DATA_LOSS, "Record digest mismatch")));
+        return;
+      }
+    }
+
     std::move(processed_cb)
         .Run(UploadRecord(
             encrypted_record.ValueOrDie().sequencing_information().priority(),
@@ -163,6 +204,7 @@ class MockUploadClient : public Storage::UploaderInterface {
   };
 
  private:
+  base::Optional<uint64_t> generation_id_;
   Sequence test_upload_sequence_;
 };
 

components/policy/proto/record.proto

@@ -75,13 +75,11 @@ message SequencingInformation {
   // (what to do with that is a decision that the caller needs to make).
   optional uint64 sequencing_id = 1;
 
-  // Generation ID (required)
-  // UUID of the last boot that did not find the public key cached - can
-  // happen after powerwash.
+  // Generation ID (required). Unique per device and priority. Generated anew
+  // when previous record digest is not found at startup (e.g. after powerwash).
   optional uint64 generation_id = 2;
 
-  // Priority (required)
-  // Generation IDs are per Priority.
+  // Priority (required).
   optional Priority priority = 3;
 }