AVRO-4285: [c++] Enforce a maximum decompressed block size

lang/c++/CMakeLists.txt

@@ -223,6 +223,7 @@ if (AVRO_BUILD_TESTS)
     unittest (StreamTests)
     unittest (SpecificTests)
     unittest (DataFileTests)
+    unittest (DecompressionLimitTests)
     unittest (JsonTests)
     unittest (AvrogencppTests)
     unittest (CompilerTests)

lang/c++/impl/DataFile.cc

@@ -18,7 +18,10 @@
 
 #include "DataFile.hh"
 #include "Compiler.hh"
+#include <cerrno>
+#include <cstdlib>
 #include <cstring>
+#include <limits>
 #include "Exception.hh"
 
 #include <random>
@@ -55,6 +58,42 @@ const size_t maxSyncInterval = 1u << 30;
 // Recommended by https://www.zlib.net/zlib_how.html
 const size_t zlibBufGrowSize = 128 * 1024;
 
+// Default upper bound, in bytes, on the size a single data-file block may
+// decompress to. A block with a very high compression ratio (or a malformed
+// block) can otherwise expand to far more memory than its compressed size.
+// Mirrors the Java SDK's decompression limit (AVRO-4247). Overridable with the
+// AVRO_MAX_DECOMPRESS_LENGTH environment variable.
+const size_t defaultMaxDecompressLength = static_cast<size_t>(200) * 1024 * 1024; // 200 MiB
+
+size_t maxDecompressLength() {
+    const char *env = std::getenv("AVRO_MAX_DECOMPRESS_LENGTH");
+    if (env != nullptr && *env != '\0') {
+        // Skip leading whitespace so a signed value such as "  -5" is still
+        // detected below; strtoull silently accepts leading whitespace and
+        // wraps a negative into a huge unsigned value otherwise.
+        const char *p = env;
+        while (*p == ' ' || *p == '\t' || *p == '\n' || *p == '\r' ||
+               *p == '\f' || *p == '\v') {
+            ++p;
+        }
+        // Reject an explicit sign before parsing (strtoull would wrap '-').
+        if (*p != '-' && *p != '+') {
+            errno = 0;
+            char *end = nullptr;
+            unsigned long long value = std::strtoull(p, &end, 10);
+            // Clamp to what size_t can represent so a huge value does not
+            // truncate on 32-bit (or smaller size_t) builds.
+            if (errno == 0 && end != nullptr && *end == '\0' && value > 0) {
+                if (value > std::numeric_limits<size_t>::max()) {
+                    return std::numeric_limits<size_t>::max();
+                }
+                return static_cast<size_t>(value);
+            }
+        }
+    }
+    return defaultMaxDecompressLength;
+}
+
 template<Codec codec>
 struct codec_trait {
     static std::string name() {
@@ -568,6 +607,19 @@ void DataFileReaderBase::readDataBlock() {
         int b4 = compressed_[len - 1] & 0xFF;
 
         checksum = (b1 << 24) + (b2 << 16) + (b3 << 8) + (b4);
+        {
+            // Reject an over-large block before allocating for it, based on the
+            // uncompressed length declared in the Snappy block header.
+            size_t declared = 0;
+            const size_t maxLength = maxDecompressLength();
+            if (snappy::GetUncompressedLength(reinterpret_cast<const char *>(compressed_.data()),
+                                              len - 4, &declared) &&
+                declared > maxLength) {
+                throw Exception(
+                    "Decompressed block size {} exceeds the maximum allowed of {} bytes",
+                    declared, maxLength);
+            }
+        }
         if (!snappy::Uncompress(reinterpret_cast<const char *>(compressed_.data()),
                                 len - 4, &uncompressed)) {
             throw Exception(
@@ -599,7 +651,7 @@ void DataFileReaderBase::readDataBlock() {
         }
 
         ZstdDecompressWrapper zstdDecompressWrapper;
-        uncompressed = zstdDecompressWrapper.decompress(compressed_);
+        uncompressed = zstdDecompressWrapper.decompress(compressed_, maxDecompressLength());
 
         std::unique_ptr<InputStream> in = memoryInputStream(
             reinterpret_cast<const uint8_t *>(uncompressed.data()),
@@ -627,12 +679,45 @@ void DataFileReaderBase::readDataBlock() {
 
             const uint8_t *data;
             size_t len;
+            size_t maxLength = maxDecompressLength();
+            // zlib tracks output in a uLong (z_stream::total_out), which is only
+            // 32-bit on some platforms (e.g. Windows). Cap the working limit to
+            // uLong's range so the buffer-size arithmetic below (which uses
+            // total_out) cannot wrap when a larger AVRO_MAX_DECOMPRESS_LENGTH is
+            // configured.
+            const uLong zlibMax = std::numeric_limits<uLong>::max();
+            if (maxLength > zlibMax) {
+                maxLength = zlibMax;
+            }
             while (ret != Z_STREAM_END && st->next(&data, &len)) {
                 zs.avail_in = static_cast<uInt>(len);
                 zs.next_in = const_cast<Bytef *>(data);
                 do {
                     if (zs.total_out == uncompressed.size()) {
-                        uncompressed.resize(uncompressed.size() + zlibBufGrowSize);
+                        // Reject a block that would decompress to more than the
+                        // allowed maximum, before growing the buffer further.
+                        if (uncompressed.size() >= maxLength) {
+                            (void) inflateEnd(&zs);
+                            // At the trigger uncompressed.size() == maxLength and
+                            // inflate still has output, so the block is at least
+                            // maxLength + 1 bytes. Report that (like the snappy/zstd
+                            // errors) so the message is accurate and consistent,
+                            // saturating the +1 so it cannot wrap to 0 when
+                            // uncompressed.size() is at size_t's maximum.
+                            const size_t reported =
+                                uncompressed.size() < std::numeric_limits<size_t>::max()
+                                    ? uncompressed.size() + 1
+                                    : uncompressed.size();
+                            throw Exception(
+                                "Decompressed block size {} exceeds the maximum allowed of {} bytes",
+                                reported, maxLength);
+                        }
+                        // Grow by the remaining capacity (capped at the grow
+                        // step) rather than size + step, which could overflow
+                        // size_t when maxLength is near SIZE_MAX.
+                        size_t remaining = maxLength - uncompressed.size();
+                        size_t grow = remaining < zlibBufGrowSize ? remaining : zlibBufGrowSize;
+                        uncompressed.resize(uncompressed.size() + grow);
                     }
                     zs.avail_out = static_cast<uInt>(uncompressed.size() - zs.total_out);
                     zs.next_out = reinterpret_cast<Bytef *>(uncompressed.data() + zs.total_out);
@@ -641,6 +726,7 @@ void DataFileReaderBase::readDataBlock() {
                         break;
                     }
                     if (ret != Z_OK) {
+                        (void) inflateEnd(&zs);
                         throw Exception("Failed to inflate, error: {}", ret);
                     }
                 } while (zs.avail_out == 0);

lang/c++/impl/ZstdDecompressWrapper.cc

@@ -21,11 +21,12 @@
 #include "ZstdDecompressWrapper.hh"
 #include "Exception.hh"
 
+#include <limits>
 #include <zstd.h>
 
 namespace avro {
 
-std::string ZstdDecompressWrapper::decompress(const std::vector<char> &compressed) {
+std::string ZstdDecompressWrapper::decompress(const std::vector<char> &compressed, size_t maxLength) {
     std::string uncompressed;
     // Get the decompressed size
     size_t decompressed_size = ZSTD_getFrameContentSize(compressed.data(), compressed.size());
@@ -43,9 +44,28 @@ std::string ZstdDecompressWrapper::decompress(const std::vector<char> &compresse
             if (ZSTD_isError(ret)) {
                 throw Exception("ZSTD decompression error: {}", ZSTD_getErrorName(ret));
             }
+            // Reject before appending so the buffer never grows past the limit.
+            if (out.pos > maxLength - uncompressed.size()) {
+                // Saturate the reported size so the addition cannot wrap when
+                // uncompressed.size() is near size_t's maximum (possible when
+                // AVRO_MAX_DECOMPRESS_LENGTH is configured close to SIZE_MAX).
+                const size_t reported =
+                    out.pos > std::numeric_limits<size_t>::max() - uncompressed.size()
+                        ? std::numeric_limits<size_t>::max()
+                        : uncompressed.size() + out.pos;
+                throw Exception(
+                    "Decompressed block size {} exceeds the maximum allowed of {} bytes",
+                    reported, maxLength);
+            }
             uncompressed.append(tmp.data(), out.pos);
         } while (ret != 0);
     } else {
+        // The frame declares its decompressed size; reject it before allocating.
+        if (decompressed_size > maxLength) {
+            throw Exception(
+                "Decompressed block size {} exceeds the maximum allowed of {} bytes",
+                decompressed_size, maxLength);
+        }
         // Batch decompress the data
         uncompressed.resize(decompressed_size);
         size_t result = ZSTD_decompress(

lang/c++/impl/ZstdDecompressWrapper.hh

@@ -33,7 +33,9 @@ public:
     ZstdDecompressWrapper();
     ~ZstdDecompressWrapper();
 
-    std::string decompress(const std::vector<char> &compressed);
+    // Decompress `compressed`, rejecting a frame that would decompress to more
+    // than `maxLength` bytes (guards against a high-ratio decompression bomb).
+    std::string decompress(const std::vector<char> &compressed, size_t maxLength);
 
 private:
     ZSTD_DCtx *dctx_ = nullptr;

lang/c++/test/DecompressionLimitTests.cc

@@ -0,0 +1,196 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+// AVRO-4285: a data-file block is decompressed according to the file's codec. A
+// block with a very high compression ratio (or a malformed block) can expand to
+// far more memory than its compressed size. Reading such a block must be
+// rejected once its decompressed size would exceed the configured maximum, which
+// these tests set to a small value via AVRO_MAX_DECOMPRESS_LENGTH.
+
+#include <cstdlib>
+#include <filesystem>
+#include <optional>
+#include <sstream>
+#include <string>
+
+#ifdef _WIN32
+#include <process.h>
+#define AVRO_TEST_GETPID _getpid
+#else
+#include <unistd.h>
+#define AVRO_TEST_GETPID getpid
+#endif
+
+#include <boost/test/included/unit_test.hpp>
+
+#include "Compiler.hh"
+#include "DataFile.hh"
+#include "Exception.hh"
+#include "ValidSchema.hh"
+
+namespace avro {
+
+// Return the underlying status (0 on success) rather than asserting inside, so
+// these can be called from DecompressLimitGuard's destructor with a
+// non-throwing check (a throwing BOOST_REQUIRE in a noexcept destructor would
+// terminate the process instead of reporting a clean failure).
+static int setDecompressLimit(const char *value) {
+#ifdef _WIN32
+    return _putenv_s("AVRO_MAX_DECOMPRESS_LENGTH", value);
+#else
+    return setenv("AVRO_MAX_DECOMPRESS_LENGTH", value, 1);
+#endif
+}
+
+static int unsetDecompressLimit() {
+#ifdef _WIN32
+    return _putenv_s("AVRO_MAX_DECOMPRESS_LENGTH", "");
+#else
+    return unsetenv("AVRO_MAX_DECOMPRESS_LENGTH");
+#endif
+}
+
+// Saves AVRO_MAX_DECOMPRESS_LENGTH on construction and restores it on
+// destruction, so a test's override does not leak into other test cases that
+// share the process.
+struct DecompressLimitGuard {
+    std::optional<std::string> previous;
+    DecompressLimitGuard() {
+        const char *env = std::getenv("AVRO_MAX_DECOMPRESS_LENGTH");
+        if (env != nullptr) {
+            previous = std::string(env);
+        }
+    }
+    ~DecompressLimitGuard() {
+        // Use a non-throwing check: this runs in a (noexcept) destructor, where a
+        // throwing BOOST_REQUIRE would terminate the process.
+        if (previous) {
+            BOOST_CHECK_EQUAL(setDecompressLimit(previous->c_str()), 0);
+        } else {
+            BOOST_CHECK_EQUAL(unsetDecompressLimit(), 0);
+        }
+    }
+};
+
+static ValidSchema stringSchema() {
+    std::istringstream iss("\"string\"");
+    ValidSchema vs;
+    compileJsonSchema(iss, vs);
+    return vs;
+}
+
+static std::string tempFile(const char *name) {
+    // Include the process id so concurrent test runs (or a stale file left by a
+    // previously crashed run) do not contend for the same path in the shared
+    // system temp directory.
+    std::ostringstream unique;
+    unique << "avro_" << AVRO_TEST_GETPID() << "_" << name;
+    return (std::filesystem::temp_directory_path() / unique.str()).string();
+}
+
+// Write a single, highly compressible value with the given codec, then read it
+// back with a small decompression limit and confirm the read is rejected.
+static void checkCodecRejectsOversized(Codec codec, const char *name) {
+    DecompressLimitGuard guard;
+    ValidSchema schema = stringSchema();
+    std::string path = tempFile(name);
+    std::string big(4 * 1024 * 1024, 'a'); // 4 MiB, compresses tiny
+
+    // Let any writer exception propagate: a failure here is a real problem
+    // (e.g. permissions or I/O), not a reason to silently skip. Codecs that are
+    // not compiled in are excluded via the #ifdef guards on the callers below.
+    {
+        DataFileWriter<std::string> writer(path.c_str(), schema, 64 * 1024 * 1024, codec);
+        writer.write(big);
+        writer.close();
+    }
+
+    BOOST_REQUIRE_EQUAL(setDecompressLimit("1048576"), 0); // 1 MiB, smaller than the 4 MiB block
+
+    bool rejected = false;
+    try {
+        DataFileReader<std::string> reader(path.c_str(), schema);
+        std::string out;
+        reader.read(out); // triggers block decompression
+    } catch (const Exception &e) {
+        // Assert it failed specifically because of the decompression limit, not
+        // an unrelated I/O or corruption error that would give a false positive.
+        rejected = std::string(e.what()).find("exceeds the maximum allowed") != std::string::npos;
+        BOOST_CHECK_MESSAGE(rejected,
+                            std::string("unexpected exception for ") + name + ": " + e.what());
+    }
+    BOOST_CHECK(std::filesystem::remove(path));
+    BOOST_CHECK_MESSAGE(rejected, std::string("codec not bounded: ") + name);
+}
+
+static void testDeflateDecompressionLimit() {
+    checkCodecRejectsOversized(DEFLATE_CODEC, "avro_decompress_limit_deflate.avro");
+}
+
+static void testSnappyDecompressionLimit() {
+#ifdef SNAPPY_CODEC_AVAILABLE
+    checkCodecRejectsOversized(SNAPPY_CODEC, "avro_decompress_limit_snappy.avro");
+#else
+    BOOST_TEST_MESSAGE("Snappy codec not available; skipping");
+#endif
+}
+
+static void testZstdDecompressionLimit() {
+#ifdef ZSTD_CODEC_AVAILABLE
+    checkCodecRejectsOversized(ZSTD_CODEC, "avro_decompress_limit_zstd.avro");
+#else
+    BOOST_TEST_MESSAGE("Zstandard codec not available; skipping");
+#endif
+}
+
+static void testWithinLimitStillReads() {
+    DecompressLimitGuard guard;
+    ValidSchema schema = stringSchema();
+    std::string path = tempFile("avro_decompress_within_limit.avro");
+    std::string payload = "hello world";
+
+    {
+        DataFileWriter<std::string> writer(path.c_str(), schema, 64 * 1024 * 1024, DEFLATE_CODEC);
+        writer.write(payload);
+        writer.close();
+    }
+
+    BOOST_REQUIRE_EQUAL(setDecompressLimit("1048576"), 0);
+
+    std::string out;
+    {
+        DataFileReader<std::string> reader(path.c_str(), schema);
+        BOOST_CHECK(reader.read(out));
+    }
+    BOOST_CHECK(std::filesystem::remove(path));
+    BOOST_CHECK_EQUAL(out, payload);
+}
+
+} // namespace avro
+
+boost::unit_test::test_suite *
+init_unit_test_suite(int, char *[]) {
+    using namespace boost::unit_test;
+
+    auto *ts = BOOST_TEST_SUITE("Avro C++ decompression limit tests");
+    ts->add(BOOST_TEST_CASE(&avro::testDeflateDecompressionLimit));
+    ts->add(BOOST_TEST_CASE(&avro::testSnappyDecompressionLimit));
+    ts->add(BOOST_TEST_CASE(&avro::testZstdDecompressionLimit));
+    ts->add(BOOST_TEST_CASE(&avro::testWithinLimitStillReads));
+    return ts;
+}