GITHUB#16479: verify a CRC32C of each stored-fields and term-vectors chunk before decompressing it

lucene/CHANGES.txt

@@ -131,6 +131,20 @@ New Features
 
 Improvements
 ---------------------
+* GITHUB#16479: Reject an invalid exception offset in LowercaseAsciiCompression#decompress. Corrupt
+  input previously threw ArrayIndexOutOfBoundsException instead of a checked IOException. Same class of
+  issue as the LZ4 match-offset check in GITHUB#16478. (Serhiy Bzhezytskyy)
+
+* GITHUB#16479: Term vectors now record a CRC32C of each chunk's compressed bytes and verify it before
+  decompressing, as stored fields do. Term vectors format version 1; version 0 is read unchanged, and a
+  segment acquires checksums when it is next merged. (Serhiy Bzhezytskyy)
+
+* GITHUB#16479: Stored fields now record a CRC32C of each chunk's compressed bytes and verify it before
+  decompressing, so a corrupt chunk is reported as a CorruptIndexException naming the affected
+  documents rather than surfacing from inside the decompressor or returning a wrong document. Stored
+  fields format version 2; version 1 is read unchanged, and a segment acquires checksums when it is
+  next merged. (Serhiy Bzhezytskyy)
+
 * GITHUB#15704: Replace LinkedList with more efficient data structure. (Renato Haeberli)
 
 * GITHUB#15682: Use ArrayDeque instead of LinkedList in CompoundWordTokenFilterBase.java. (Renato Haeberli)

lucene/core/src/java/org/apache/lucene/codecs/lucene104/package-info.java

@@ -421,6 +421,9 @@
  *   <li>In version 10.3, the index of block tree changed to be specialized trie instead of FST.
  *   <li>In version 10.4, the block size was increased from 128 to 256. There are now 256 and 8,192
  *       postings between skip pointers instead of 128 and 4,096.
+ *   <li>In version 11.0, each chunk of stored fields and of term vectors is followed by a CRC32C of
+ *       its compressed bytes, which is verified before the chunk is decompressed. Chunks written by
+ *       an earlier version have no checksum and are read as before.
  * </ul>
  *
  * <a id="Limitations"></a>

lucene/core/src/java/org/apache/lucene/codecs/lucene90/compressing/Lucene90CompressingStoredFieldsReader.java

@@ -35,12 +35,14 @@
 import static org.apache.lucene.codecs.lucene90.compressing.Lucene90CompressingStoredFieldsWriter.STRING;
 import static org.apache.lucene.codecs.lucene90.compressing.Lucene90CompressingStoredFieldsWriter.TYPE_BITS;
 import static org.apache.lucene.codecs.lucene90.compressing.Lucene90CompressingStoredFieldsWriter.TYPE_MASK;
+import static org.apache.lucene.codecs.lucene90.compressing.Lucene90CompressingStoredFieldsWriter.VERSION_CHUNK_CHECKSUM;
 import static org.apache.lucene.codecs.lucene90.compressing.Lucene90CompressingStoredFieldsWriter.VERSION_CURRENT;
 import static org.apache.lucene.codecs.lucene90.compressing.Lucene90CompressingStoredFieldsWriter.VERSION_START;
 
 import java.io.EOFException;
 import java.io.IOException;
 import java.util.Arrays;
+import java.util.zip.CRC32C;
 import org.apache.lucene.codecs.CodecUtil;
 import org.apache.lucene.codecs.StoredFieldsReader;
 import org.apache.lucene.codecs.compressing.CompressionMode;
@@ -423,6 +425,7 @@ private class BlockState {
     // the start pointer at which you can read the compressed documents
     private long startPointer;
 
+    private final byte[] chunkCrcScratch = new byte[8192];
     private final BytesRef spare;
     private final BytesRef bytes;
 
@@ -502,6 +505,12 @@ private void doReset(int docID) throws IOException {
 
       startPointer = fieldsStream.getFilePointer();
 
+      if (version >= VERSION_CHUNK_CHECKSUM) {
+        // Verify the compressed bytes before anything decompresses them. Leaves the stream where it
+        // was, so startPointer keeps its meaning.
+        verifyCompressedChunk(docID, startPointer);
+      }
+
       if (merging) {
         final int totalLength = Math.toIntExact(offsets[chunkDocs]);
         // decompress eagerly
@@ -526,6 +535,62 @@ private void doReset(int docID) throws IOException {
       }
     }
 
+    /**
+     * Verifies the CRC32C that follows a chunk's compressed bytes, before anything decompresses
+     * them.
+     *
+     * <p>Without this, a corrupt byte inside a chunk surfaces as whatever the decompressor happens
+     * to do with it: an {@link ArrayIndexOutOfBoundsException} from LZ4, or no error at all and a
+     * different document than the one that was stored. The frame format of LZ4 specifies the same
+     * check over the same bytes, "before decoding", but Lucene implements the block format, which
+     * has no checksum of its own.
+     *
+     * <p>The chunk's length comes from the fields index rather than from the file, so a corrupt
+     * length cannot be used to read out of the chunk. Leaves the stream where it was found.
+     */
+    private void verifyCompressedChunk(int docID, long compressedStart) throws IOException {
+      final long blockID = indexReader.getBlockID(docID);
+      final long blockEnd =
+          indexReader.getBlockStartPointer(blockID) + indexReader.getBlockLength(blockID);
+      final long compressedLength = blockEnd - Integer.BYTES - compressedStart;
+      if (compressedLength < 0) {
+        throw new CorruptIndexException(
+            "chunk shorter than its checksum: docBase="
+                + docBase
+                + ", chunkDocs="
+                + chunkDocs
+                + ", length="
+                + compressedLength,
+            fieldsStream);
+      }
+
+      final CRC32C crc = new CRC32C();
+      long remaining = compressedLength;
+      while (remaining > 0) {
+        final int n = (int) Math.min(chunkCrcScratch.length, remaining);
+        fieldsStream.readBytes(chunkCrcScratch, 0, n);
+        crc.update(chunkCrcScratch, 0, n);
+        remaining -= n;
+      }
+      final int actual = (int) crc.getValue();
+      final int expected = fieldsStream.readInt();
+      if (actual != expected) {
+        throw new CorruptIndexException(
+            "chunk checksum mismatch: docBase="
+                + docBase
+                + ", chunkDocs="
+                + chunkDocs
+                + ", compressedLength="
+                + compressedLength
+                + ", expected="
+                + Integer.toHexString(expected)
+                + ", actual="
+                + Integer.toHexString(actual),
+            fieldsStream);
+      }
+      fieldsStream.seek(compressedStart);
+    }
+
     /**
      * Get the serialized representation of the given docID. This docID has to be contained in the
      * current block.
@@ -611,9 +676,13 @@ public void skipBytes(long numBytes) throws IOException {
             };
       } else {
         fieldsStream.seek(startPointer);
-        decompressor.decompress(fieldsStream, totalLength, offset, length, bytes);
-        assert bytes.length == length;
-        documentInput = new ByteArrayDataInput(bytes.bytes, bytes.offset, bytes.length);
+        // PROTOTYPE: decompress the WHOLE chunk so the per-chunk checksum can be verified, then
+        // hand
+        // out the requested document's slice. This is the cost the 2013 proposal did not discuss:
+        // on-demand reads currently decompress only the bytes they need, and a chunk-wide checksum
+        // forces the whole chunk. See PROTOTYPE-NOTES in the branch.
+        decompressor.decompress(fieldsStream, totalLength, 0, totalLength, bytes);
+        documentInput = new ByteArrayDataInput(bytes.bytes, bytes.offset + offset, length);
       }
 
       return new SerializedDocument(documentInput, length, numStoredFields);

lucene/core/src/java/org/apache/lucene/codecs/lucene90/compressing/Lucene90CompressingStoredFieldsWriter.java

@@ -21,6 +21,7 @@
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.zip.CRC32C;
 import org.apache.lucene.codecs.CodecUtil;
 import org.apache.lucene.codecs.StoredFieldsReader;
 import org.apache.lucene.codecs.StoredFieldsWriter;
@@ -78,7 +79,14 @@ public final class Lucene90CompressingStoredFieldsWriter extends StoredFieldsWri
   static final int TYPE_MASK = (int) PackedInts.maxValue(TYPE_BITS);
 
   static final int VERSION_START = 1;
-  static final int VERSION_CURRENT = VERSION_START;
+
+  /**
+   * Each chunk is followed by a CRC32C of its compressed bytes, so that a corrupt chunk is rejected
+   * before it reaches the decompressor.
+   */
+  static final int VERSION_CHUNK_CHECKSUM = 2;
+
+  static final int VERSION_CURRENT = VERSION_CHUNK_CHECKSUM;
   static final int META_VERSION_START = 0;
 
   private final String segment;
@@ -171,6 +179,36 @@ public void close() throws IOException {
     }
   }
 
+  /**
+   * Passes bytes through to a delegate while computing a CRC32C of them, so that the compressed
+   * bytes of a chunk can be checksummed as they are written rather than buffered and hashed
+   * afterwards.
+   */
+  private static final class ChecksummingDataOutput extends DataOutput {
+    private final DataOutput in;
+    private final CRC32C crc = new CRC32C();
+
+    ChecksummingDataOutput(DataOutput in) {
+      this.in = in;
+    }
+
+    @Override
+    public void writeByte(byte b) throws IOException {
+      crc.update(b);
+      in.writeByte(b);
+    }
+
+    @Override
+    public void writeBytes(byte[] b, int offset, int length) throws IOException {
+      crc.update(b, offset, length);
+      in.writeBytes(b, offset, length);
+    }
+
+    long getChecksum() {
+      return crc.getValue();
+    }
+  }
+
   private int numStoredFieldsInDoc;
 
   @Override
@@ -246,18 +284,30 @@ private void flush(boolean force) throws IOException {
     final boolean dirtyChunk = force;
     writeHeader(docBase, numBufferedDocs, numStoredFields, lengths, sliced, dirtyChunk);
     ByteBuffersDataInput bytebuffers = bufferedDocs.toDataInput();
-    // compress stored fields to fieldsStream.
+
+    // A CRC32C of the compressed bytes, so that a corrupt chunk is rejected before it reaches the
+    // decompressor rather than surfacing as an exception from inside it, or as a wrong document.
+    // This is where the LZ4 frame format puts its optional per-block checksum, "calculated by using
+    // the xxHash-32 algorithm on the raw (compressed) data block [...] The intention is to detect
+    // data corruption immediately, before decoding". CRC32C is used rather than xxHash-32 because
+    // it
+    // is in the JDK and hardware-accelerated on the platforms Lucene targets.
+    //
+    // The checksum is written after the compressed bytes, since its value is not known until they
+    // have been produced, and read from the end of the chunk, whose length the fields index knows.
+    final ChecksummingDataOutput checksummed = new ChecksummingDataOutput(fieldsStream);
     if (sliced) {
       // big chunk, slice it, using ByteBuffersDataInput ignore memory copy
       final int capacity = (int) bytebuffers.length();
       for (int compressed = 0; compressed < capacity; compressed += chunkSize) {
         int l = Math.min(chunkSize, capacity - compressed);
         ByteBuffersDataInput bbdi = bytebuffers.slice(compressed, l);
-        compressor.compress(bbdi, fieldsStream);
+        compressor.compress(bbdi, checksummed);
       }
     } else {
-      compressor.compress(bytebuffers, fieldsStream);
+      compressor.compress(bytebuffers, checksummed);
     }
+    fieldsStream.writeInt((int) checksummed.getChecksum());
 
     // reset
     docBase += numBufferedDocs;

lucene/core/src/java/org/apache/lucene/codecs/lucene90/compressing/Lucene90CompressingTermVectorsReader.java

@@ -26,6 +26,7 @@
 import static org.apache.lucene.codecs.lucene90.compressing.Lucene90CompressingTermVectorsWriter.VECTORS_INDEX_CODEC_NAME;
 import static org.apache.lucene.codecs.lucene90.compressing.Lucene90CompressingTermVectorsWriter.VECTORS_INDEX_EXTENSION;
 import static org.apache.lucene.codecs.lucene90.compressing.Lucene90CompressingTermVectorsWriter.VECTORS_META_EXTENSION;
+import static org.apache.lucene.codecs.lucene90.compressing.Lucene90CompressingTermVectorsWriter.VERSION_CHUNK_CHECKSUM;
 import static org.apache.lucene.codecs.lucene90.compressing.Lucene90CompressingTermVectorsWriter.VERSION_CURRENT;
 import static org.apache.lucene.codecs.lucene90.compressing.Lucene90CompressingTermVectorsWriter.VERSION_START;
 
@@ -35,6 +36,7 @@
 import java.util.Collections;
 import java.util.Iterator;
 import java.util.NoSuchElementException;
+import java.util.zip.CRC32C;
 import org.apache.lucene.codecs.CodecUtil;
 import org.apache.lucene.codecs.TermVectorsReader;
 import org.apache.lucene.codecs.compressing.CompressionMode;
@@ -358,6 +360,57 @@ public void prefetch(int docID) throws IOException {
     prefetchedBlockIDCache[prefetchedBlockIDCacheIndex++ & PREFETCH_CACHE_MASK] = blockID;
   }
 
+  /**
+   * Verifies the CRC32C that follows a chunk's compressed bytes, before anything decompresses them.
+   *
+   * <p>The chunk's length comes from the fields index rather than from the file, so a corrupt
+   * length cannot be used to read outside the chunk. Leaves the stream where it was found.
+   */
+  private void verifyCompressedChunk(int doc, int docBase, int chunkDocs) throws IOException {
+    final long compressedStart = vectorsStream.getFilePointer();
+    final long blockID = indexReader.getBlockID(doc);
+    final long blockEnd =
+        indexReader.getBlockStartPointer(blockID) + indexReader.getBlockLength(blockID);
+    final long compressedLength = blockEnd - Integer.BYTES - compressedStart;
+    if (compressedLength < 0) {
+      throw new CorruptIndexException(
+          "chunk shorter than its checksum: docBase="
+              + docBase
+              + ", chunkDocs="
+              + chunkDocs
+              + ", length="
+              + compressedLength,
+          vectorsStream);
+    }
+
+    final CRC32C crc = new CRC32C();
+    final byte[] scratch = new byte[8192];
+    long remaining = compressedLength;
+    while (remaining > 0) {
+      final int n = (int) Math.min(scratch.length, remaining);
+      vectorsStream.readBytes(scratch, 0, n);
+      crc.update(scratch, 0, n);
+      remaining -= n;
+    }
+    final int actual = (int) crc.getValue();
+    final int expected = vectorsStream.readInt();
+    if (actual != expected) {
+      throw new CorruptIndexException(
+          "chunk checksum mismatch: docBase="
+              + docBase
+              + ", chunkDocs="
+              + chunkDocs
+              + ", compressedLength="
+              + compressedLength
+              + ", expected="
+              + Integer.toHexString(expected)
+              + ", actual="
+              + Integer.toHexString(actual),
+          vectorsStream);
+    }
+    vectorsStream.seek(compressedStart);
+  }
+
   @Override
   public Fields get(int doc) throws IOException {
     ensureOpen();
@@ -701,6 +754,12 @@ public Fields get(int doc) throws IOException {
       assert termIndex == totalTerms : termIndex + " " + totalTerms;
     }
 
+    if (version >= VERSION_CHUNK_CHECKSUM) {
+      // Verify the compressed bytes before the decompressor sees them, so that a corrupt chunk is
+      // reported as such rather than surfacing from inside the decompression routine.
+      verifyCompressedChunk(doc, docBase, chunkDocs);
+    }
+
     // decompress data
     final BytesRef suffixBytes = new BytesRef();
     decompressor.decompress(