Fix out-of-bounds read in SstFileWriter::DeleteRange

Summary:
**What**: Fix an out-of-bounds read and a forged end timestamp in the timestamped `SstFileWriter::DeleteRange()` overload.

**How**:
- `Rep::DeleteRange()` now builds each boundary through a new `Rep::AppendTimestamp()` helper, which tests one key's adjacency to `timestamp` and either aliases the two or materializes `key + timestamp` into a caller-owned buffer.
- Drops the fast path's `assert`. It was never a guard: it performed the same out-of-bounds read it was meant to police, and `DeleteRangeImpl()` already asserts the two boundary timestamps match.
- Behavior change for callers whose two keys each carry their own trailing timestamp: they now take one extra copy. Aliasing is still used for whichever key is genuinely adjacent.

**Why**: A single `timestamp` buffer can be byte-adjacent to at most one of two distinct keys, but the old branch tested `end_key` using `begin_key`'s length and then widened both slices whenever either test passed - so an asymmetric caller, one key contiguous with the timestamp and the other bare, read past an allocation and persisted a range tombstone carrying foreign heap bytes as its end timestamp. Nothing forbids that layout: the public contract requires only that the keys carry no timestamp.

Reviewed By: xingbowang

Differential Revision: D119401932

fbshipit-source-id: 4c842d4afbe7bd7c4eabc68c6aee0543cdee2c9c

table/sst_file_reader_test.cc

@@ -7,6 +7,8 @@
 
 #include <atomic>
 #include <cinttypes>
+#include <cstring>
+#include <memory>
 
 #include "db/db_test_util.h"
 #include "db/dbformat.h"
@@ -1078,6 +1080,45 @@ class SstFileReaderTimestampTest : public testing::Test {
     ASSERT_OK(iter->status());
   }
 
+  // Writes a file holding a single timestamped range tombstone. Each key gets
+  // its own exactly sized heap buffer so that reading past one is caught, and
+  // `timestamp` is placed right after whichever key `timestamp_follows_end_key`
+  // selects -- a lone timestamp can only ever be adjacent to one of the two.
+  void CreateFileWithDeleteRange(const std::string& begin_key,
+                                 const std::string& end_key,
+                                 const std::string& timestamp,
+                                 bool timestamp_follows_end_key,
+                                 ExternalSstFileInfo* file_info) {
+    const std::string& adjacent_key =
+        timestamp_follows_end_key ? end_key : begin_key;
+    const std::string& lone_key =
+        timestamp_follows_end_key ? begin_key : end_key;
+
+    std::unique_ptr<char[]> adjacent_buf(
+        new char[adjacent_key.size() + timestamp.size()]);
+    memcpy(adjacent_buf.get(), adjacent_key.data(), adjacent_key.size());
+    memcpy(adjacent_buf.get() + adjacent_key.size(), timestamp.data(),
+           timestamp.size());
+    std::unique_ptr<char[]> lone_buf(new char[lone_key.size()]);
+    memcpy(lone_buf.get(), lone_key.data(), lone_key.size());
+
+    const Slice adjacent_slice(adjacent_buf.get(), adjacent_key.size());
+    const Slice lone_slice(lone_buf.get(), lone_key.size());
+    const Slice timestamp_slice(adjacent_buf.get() + adjacent_key.size(),
+                                timestamp.size());
+
+    SstFileWriter writer(soptions_, options_);
+    ASSERT_OK(writer.Open(sst_name_));
+    if (timestamp_follows_end_key) {
+      ASSERT_OK(
+          writer.DeleteRange(lone_slice, adjacent_slice, timestamp_slice));
+    } else {
+      ASSERT_OK(
+          writer.DeleteRange(adjacent_slice, lone_slice, timestamp_slice));
+    }
+    ASSERT_OK(writer.Finish(file_info));
+  }
+
  protected:
   std::shared_ptr<Env> env_guard_;
   Options options_;
@@ -1156,6 +1197,31 @@ TEST_F(SstFileReaderTimestampTest, Basic) {
   }
 }
 
+TEST_F(SstFileReaderTimestampTest, DeleteRangeTimestampAdjacentToOneKey) {
+  const std::string timestamp = EncodeAsUint64(1);
+
+  {
+    // Only begin_key is followed in memory by the timestamp.
+    ExternalSstFileInfo file_info;
+    CreateFileWithDeleteRange("begin", "end", timestamp,
+                              /* timestamp_follows_end_key */ false,
+                              &file_info);
+    ASSERT_EQ(file_info.smallest_range_del_key, "begin" + timestamp);
+    ASSERT_EQ(file_info.largest_range_del_key, "end" + timestamp);
+  }
+
+  {
+    // Only end_key is followed in memory by the timestamp, and the two keys
+    // are the same length, so testing end_key's adjacency with begin_key's
+    // size would match here.
+    ExternalSstFileInfo file_info;
+    CreateFileWithDeleteRange("aaa", "bbb", timestamp,
+                              /* timestamp_follows_end_key */ true, &file_info);
+    ASSERT_EQ(file_info.smallest_range_del_key, "aaa" + timestamp);
+    ASSERT_EQ(file_info.largest_range_del_key, "bbb" + timestamp);
+  }
+}
+
 TEST_F(SstFileReaderTimestampTest, TimestampsOutOfOrder) {
   SstFileWriter writer(soptions_, options_);
 

table/sst_file_writer.cc

@@ -265,36 +265,39 @@ struct SstFileWriter::Rep {
     return DeleteRangeImpl(begin_key, end_key);
   }
 
+  // Returns `user_key` with `timestamp` appended. When the two are already
+  // adjacent in memory the result aliases them, otherwise `buf` supplies the
+  // backing storage and must outlive the returned slice. A single timestamp
+  // slice can be adjacent to at most one key, so each key has to be tested on
+  // its own; extending a key that is not ts-adjacent reads out of bounds.
+  static Slice AppendTimestamp(const Slice& user_key, const Slice& timestamp,
+                               std::string* buf) {
+    const size_t user_key_size = user_key.size();
+    const size_t timestamp_size = timestamp.size();
+
+    if (user_key.data() + user_key_size == timestamp.data()) {
+      return Slice(user_key.data(), user_key_size + timestamp_size);
+    }
+
+    buf->reserve(user_key_size + timestamp_size);
+    buf->append(user_key.data(), user_key_size);
+    buf->append(timestamp.data(), timestamp_size);
+    return Slice(*buf);
+  }
+
   // begin_key and end_key should be users keys without timestamp.
   Status DeleteRange(const Slice& begin_key, const Slice& end_key,
                      const Slice& timestamp) {
-    const size_t timestamp_size = timestamp.size();
-
     if (internal_comparator.user_comparator()->timestamp_size() !=
-        timestamp_size) {
+        timestamp.size()) {
       return Status::InvalidArgument("Timestamp size mismatch");
     }
 
-    const size_t begin_key_size = begin_key.size();
-    const size_t end_key_size = end_key.size();
-    if (begin_key.data() + begin_key_size == timestamp.data() ||
-        end_key.data() + begin_key_size == timestamp.data()) {
-      assert(memcmp(begin_key.data() + begin_key_size,
-                    end_key.data() + end_key_size, timestamp_size) == 0);
-      Slice begin_key_with_ts(begin_key.data(),
-                              begin_key_size + timestamp_size);
-      Slice end_key_with_ts(end_key.data(), end_key.size() + timestamp_size);
-      return DeleteRangeImpl(begin_key_with_ts, end_key_with_ts);
-    }
-    std::string begin_key_with_ts;
-    begin_key_with_ts.reserve(begin_key_size + timestamp_size);
-    begin_key_with_ts.append(begin_key.data(), begin_key_size);
-    begin_key_with_ts.append(timestamp.data(), timestamp_size);
-    std::string end_key_with_ts;
-    end_key_with_ts.reserve(end_key_size + timestamp_size);
-    end_key_with_ts.append(end_key.data(), end_key_size);
-    end_key_with_ts.append(timestamp.data(), timestamp_size);
-    return DeleteRangeImpl(begin_key_with_ts, end_key_with_ts);
+    std::string begin_key_buf;
+    std::string end_key_buf;
+    return DeleteRangeImpl(
+        AppendTimestamp(begin_key, timestamp, &begin_key_buf),
+        AppendTimestamp(end_key, timestamp, &end_key_buf));
   }
 
   Status InvalidatePageCache(bool closing) {