[format] Validate ORC stripe metadata before reading

paimon-format/src/main/java/org/apache/orc/impl/RecordReaderImpl.java

@@ -38,6 +38,7 @@
 import org.apache.orc.DateColumnStatistics;
 import org.apache.orc.DecimalColumnStatistics;
 import org.apache.orc.DoubleColumnStatistics;
+import org.apache.orc.FileFormatException;
 import org.apache.orc.IntegerColumnStatistics;
 import org.apache.orc.OrcConf;
 import org.apache.orc.OrcFile;
@@ -96,6 +97,7 @@ public class RecordReaderImpl implements RecordReader {
                     .setBytesOnDisk(0)
                     .build();
     protected final Path path;
+    private final long fileLength;
     private final long firstRow;
     private final List<StripeInformation> stripes = new ArrayList<>();
     private OrcProto.StripeFooter stripeFooter;
@@ -255,6 +257,7 @@ public RecordReaderImpl(
         LOG.debug("noSelectedVector={}", this.noSelectedVector);
         this.schema = evolution.getReaderSchema();
         this.path = fileReader.path;
+        this.fileLength = fileReader.getFileTail().getFileLength();
         this.rowIndexStride = fileReader.rowIndexStride;
         boolean ignoreNonUtf8BloomFilter =
                 OrcConf.IGNORE_NON_UTF8_BLOOM_FILTERS.getBoolean(fileReader.conf);
@@ -1481,8 +1484,49 @@ private void readStripe() throws IOException {
         }
     }
 
+    static void validateStripeInformation(StripeInformation stripe, long fileLength, Path path)
+            throws FileFormatException {
+        long offset = stripe.getOffset();
+        long indexLength = stripe.getIndexLength();
+        long dataLength = stripe.getDataLength();
+        long footerLength = stripe.getFooterLength();
+        boolean malformed =
+                offset < 0
+                        || indexLength < 0
+                        || dataLength < 0
+                        || footerLength < 0
+                        || footerLength > Integer.MAX_VALUE;
+        if (!malformed) {
+            try {
+                long total =
+                        Math.addExact(
+                                Math.addExact(Math.addExact(offset, indexLength), dataLength),
+                                footerLength);
+                malformed = total >= fileLength;
+            } catch (ArithmeticException e) {
+                malformed = true;
+            }
+        }
+        if (malformed) {
+            throw new FileFormatException(
+                    "Malformed ORC file "
+                            + path
+                            + ". Invalid stripe offset/length. fileLength="
+                            + fileLength
+                            + ", offset="
+                            + offset
+                            + ", indexLength="
+                            + indexLength
+                            + ", dataLength="
+                            + dataLength
+                            + ", footerLength="
+                            + footerLength);
+        }
+    }
+
     private StripeInformation beginReadStripe() throws IOException {
         StripeInformation stripe = stripes.get(currentStripe);
+        validateStripeInformation(stripe, fileLength, path);
         stripeFooter = readStripeFooter(stripe);
         clearStreams();
         // setup the position in the stripe

paimon-format/src/test/java/org/apache/orc/impl/RecordReaderImplTest.java

@@ -18,15 +18,21 @@
 
 package org.apache.orc.impl;
 
+import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.hive.ql.io.sarg.PredicateLeaf;
 import org.apache.hadoop.hive.ql.io.sarg.SearchArgument;
+import org.apache.orc.FileFormatException;
+import org.apache.orc.OrcProto;
+import org.apache.orc.StripeInformation;
 import org.apache.orc.TypeDescription;
 import org.junit.jupiter.api.Test;
 
 import java.util.Collections;
 import java.util.List;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 /** Tests for {@link RecordReaderImpl}. */
 public class RecordReaderImplTest {
@@ -44,6 +50,56 @@ stats, nullSafeEqualsPredicate("value"), null))
                 .isEqualTo(SearchArgument.TruthValue.YES);
     }
 
+    @Test
+    public void testMalformedStripeLength() {
+        long fileLength = 1000;
+        Path path = new Path("file.orc");
+
+        assertThatCode(
+                        () ->
+                                RecordReaderImpl.validateStripeInformation(
+                                        stripe(3, 10, 20, 30), fileLength, path))
+                .doesNotThrowAnyException();
+
+        long[][] cases =
+                new long[][] {
+                    {0, 0, 0, 1L << 31},
+                    {0, 0, 0, 1L << 32},
+                    {0, 0, 0, -1L},
+                    {-1L, 0, 0, 0},
+                    {0, -1L, 0, 0},
+                    {0, 0, -1L, 0},
+                    {1L << 62, 0, 1L << 62, 0},
+                    {900, 0, 0, 200}
+                };
+        for (long[] values : cases) {
+            assertThatThrownBy(
+                            () ->
+                                    RecordReaderImpl.validateStripeInformation(
+                                            stripe(values[0], values[1], values[2], values[3]),
+                                            fileLength,
+                                            path))
+                    .isInstanceOf(FileFormatException.class)
+                    .hasMessageContaining("Malformed ORC file")
+                    .hasMessageContaining("Invalid stripe offset/length");
+        }
+    }
+
+    private static StripeInformation stripe(
+            long offset, long indexLength, long dataLength, long footerLength) {
+        return new ReaderImpl.StripeInformationImpl(
+                OrcProto.StripeInformation.newBuilder()
+                        .setOffset(offset)
+                        .setIndexLength(indexLength)
+                        .setDataLength(dataLength)
+                        .setFooterLength(footerLength)
+                        .setNumberOfRows(0)
+                        .build(),
+                0,
+                0,
+                null);
+    }
+
     private static PredicateLeaf nullSafeEqualsPredicate(Object literal) {
         return new PredicateLeaf() {
             @Override