diff --git a/lang/c++/CMakeLists.txt b/lang/c++/CMakeLists.txt index a88bc77f096..de047345135 100644 --- a/lang/c++/CMakeLists.txt +++ b/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) diff --git a/lang/c++/impl/DataFile.cc b/lang/c++/impl/DataFile.cc index 661ee583d08..5517133d3c5 100644 --- a/lang/c++/impl/DataFile.cc +++ b/lang/c++/impl/DataFile.cc @@ -18,7 +18,10 @@ #include "DataFile.hh" #include "Compiler.hh" +#include +#include #include +#include #include "Exception.hh" #include @@ -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(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::max()) { + return std::numeric_limits::max(); + } + return static_cast(value); + } + } + } + return defaultMaxDecompressLength; +} + template 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(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(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 in = memoryInputStream( reinterpret_cast(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::max(); + if (maxLength > zlibMax) { + maxLength = zlibMax; + } while (ret != Z_STREAM_END && st->next(&data, &len)) { zs.avail_in = static_cast(len); zs.next_in = const_cast(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::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(uncompressed.size() - zs.total_out); zs.next_out = reinterpret_cast(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); diff --git a/lang/c++/impl/ZstdDecompressWrapper.cc b/lang/c++/impl/ZstdDecompressWrapper.cc index 86d996dd315..5037c3c1f00 100644 --- a/lang/c++/impl/ZstdDecompressWrapper.cc +++ b/lang/c++/impl/ZstdDecompressWrapper.cc @@ -21,11 +21,12 @@ #include "ZstdDecompressWrapper.hh" #include "Exception.hh" +#include #include namespace avro { -std::string ZstdDecompressWrapper::decompress(const std::vector &compressed) { +std::string ZstdDecompressWrapper::decompress(const std::vector &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 &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::max() - uncompressed.size() + ? std::numeric_limits::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( diff --git a/lang/c++/impl/ZstdDecompressWrapper.hh b/lang/c++/impl/ZstdDecompressWrapper.hh index b5b97758c48..fea57d59af0 100644 --- a/lang/c++/impl/ZstdDecompressWrapper.hh +++ b/lang/c++/impl/ZstdDecompressWrapper.hh @@ -33,7 +33,9 @@ public: ZstdDecompressWrapper(); ~ZstdDecompressWrapper(); - std::string decompress(const std::vector &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 &compressed, size_t maxLength); private: ZSTD_DCtx *dctx_ = nullptr; diff --git a/lang/c++/test/DecompressionLimitTests.cc b/lang/c++/test/DecompressionLimitTests.cc new file mode 100644 index 00000000000..93e35163071 --- /dev/null +++ b/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 +#include +#include +#include +#include + +#ifdef _WIN32 +#include +#define AVRO_TEST_GETPID _getpid +#else +#include +#define AVRO_TEST_GETPID getpid +#endif + +#include + +#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 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 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 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 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 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; +}