AVRO-4286: [csharp] Enforce a maximum decompressed block size

lang/csharp/src/apache/main/File/Codec.cs

@@ -18,6 +18,7 @@
 
 using System;
 using System.Collections.Generic;
+using System.Globalization;
 using System.IO;
 using System.Reflection;
 
@@ -29,6 +30,98 @@ namespace Avro.File
     /// </summary>
     public abstract class Codec
     {
+        /// <summary>
+        /// 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.
+        /// </summary>
+        public static readonly long DefaultMaxDecompressLength = 200L * 1024 * 1024; // 200 MiB
+
+        /// <summary>
+        /// Name of the environment variable used to override the default maximum
+        /// decompressed size of a single block.
+        /// </summary>
+        public const string MaxDecompressLengthEnvVar = "AVRO_MAX_DECOMPRESS_LENGTH";
+
+        /// <summary>
+        /// The maximum number of bytes a single block is allowed to decompress to.
+        /// </summary>
+        /// <returns>The configured limit, honoring the environment override.</returns>
+        public static long GetMaxDecompressLength()
+        {
+            var value = Environment.GetEnvironmentVariable(MaxDecompressLengthEnvVar);
+            if (value != null && long.TryParse(value, NumberStyles.Integer, CultureInfo.InvariantCulture, out var parsed) && parsed > 0)
+            {
+                return parsed;
+            }
+
+            return DefaultMaxDecompressLength;
+        }
+
+        /// <summary>
+        /// Throws if the given decompressed length exceeds the maximum allowed.
+        /// </summary>
+        /// <param name="length">The number of decompressed bytes.</param>
+        /// <param name="maxLength">The maximum number of decompressed bytes allowed.</param>
+        public static void CheckDecompressLength(long length, long maxLength)
+        {
+            if (length > maxLength)
+            {
+                throw new AvroRuntimeException(
+                    $"Decompressed block size {length} exceeds the maximum allowed of {maxLength} bytes. " +
+                    $"For data-file reads, the {MaxDecompressLengthEnvVar} environment variable raises the limit.");
+            }
+        }
+
+        /// <summary>
+        /// Copies a decompression stream to the destination, rejecting the block as
+        /// soon as its decompressed size would exceed <paramref name="maxLength"/> so
+        /// an over-large (or malicious) block is not fully materialized in memory.
+        /// </summary>
+        /// <param name="source">The decompression stream to read from.</param>
+        /// <param name="destination">The stream to write the decompressed data to.</param>
+        /// <param name="maxLength">The maximum number of decompressed bytes allowed.</param>
+        public static void CopyBounded(Stream source, Stream destination, long maxLength)
+        {
+            if (source == null)
+            {
+                throw new ArgumentNullException(nameof(source));
+            }
+
+            if (destination == null)
+            {
+                throw new ArgumentNullException(nameof(destination));
+            }
+
+            if (maxLength < 0)
+            {
+                throw new ArgumentOutOfRangeException(nameof(maxLength), "maxLength must not be negative.");
+            }
+
+            byte[] buffer = new byte[81920];
+            long total = 0;
+            int read;
+            while ((read = source.Read(buffer, 0, buffer.Length)) > 0)
+            {
+                // Pre-add bound check: total is always <= maxLength here and
+                // read > 0, so maxLength - total >= 0 and this cannot overflow.
+                // Rejecting before adding stops total from overflowing and
+                // wrapping past the limit for a very large maxLength.
+                if (read > maxLength - total)
+                {
+                    throw new AvroRuntimeException(
+                        $"Decompressed block size exceeds the maximum allowed of {maxLength} bytes " +
+                        $"(at least {total} bytes already decompressed). " +
+                        $"For data-file reads, the {MaxDecompressLengthEnvVar} environment variable raises the limit.");
+                }
+
+                total += read;
+                destination.Write(buffer, 0, read);
+            }
+        }
+
         /// <summary>
         /// Compress data using implemented codec.
         /// </summary>

lang/csharp/src/apache/main/File/DataFileReader.cs

@@ -335,6 +335,11 @@ public bool HasNext()
                     {
                         _currentBlock = NextRawBlock(_currentBlock);
                         _currentBlock.Data = _codec.Decompress(_currentBlock.Data, (int)_blockSize);
+                        // Guard against a block that decompresses to more than the
+                        // allowed maximum (a decompression bomb). The built-in deflate
+                        // codec is already bounded during decompression; this covers
+                        // any codec that returns a fully decompressed buffer.
+                        Codec.CheckDecompressLength(_currentBlock.Data.Length, Codec.GetMaxDecompressLength());
                         _datumDecoder = new BinaryDecoder(_currentBlock.GetDataAsStream());
                     }
                 }
@@ -343,7 +348,7 @@ public bool HasNext()
             catch (Exception e)
             {
                 throw new AvroRuntimeException(string.Format(CultureInfo.InvariantCulture,
-                    "Error fetching next object from block: {0}", e));
+                    "Error fetching next object from block: {0}", e.Message), e);
             }
         }
 

lang/csharp/src/apache/main/File/DeflateCodec.cs

@@ -63,7 +63,10 @@ public override byte[] Decompress(byte[] compressedData, int length)
             {
                 using (DeflateStream decompress = new DeflateStream(inStream, CompressionMode.Decompress))
                 {
-                    decompress.CopyTo(outStream);
+                    // Bound the decompressed size to guard against a block with a very
+                    // high compression ratio expanding to far more memory than its
+                    // compressed size.
+                    CopyBounded(decompress, outStream, GetMaxDecompressLength());
                 }
                 return outStream.ToArray();
             }

lang/csharp/src/apache/test/File/FileTests.cs

@@ -425,6 +425,112 @@ public void OpenAppendWriter_IncorrectOutStream_Throws()
             Assert.Throws(typeof(AvroRuntimeException), action);
         }
 
+        /// <summary>
+        /// A block with a very high compression ratio can expand to far more memory
+        /// than its compressed size; decompressing such a block must be rejected once
+        /// its decompressed size would exceed the configured maximum.
+        /// </summary>
+        [Test]
+        public void TestDeflateDecompressionLimit()
+        {
+            var codec = new DeflateCodec();
+            byte[] big = new byte[128 * 1024]; // 128 KiB of zeros, compresses tiny
+            byte[] compressed = codec.Compress(big);
+
+            var previous = Environment.GetEnvironmentVariable(Codec.MaxDecompressLengthEnvVar);
+            Environment.SetEnvironmentVariable(Codec.MaxDecompressLengthEnvVar, "65536"); // 64 KiB
+            try
+            {
+                Assert.Throws<AvroRuntimeException>(
+                    () => codec.Decompress(compressed, compressed.Length));
+            }
+            finally
+            {
+                Environment.SetEnvironmentVariable(Codec.MaxDecompressLengthEnvVar, previous);
+            }
+        }
+
+        [Test]
+        public void TestDeflateWithinDecompressionLimit()
+        {
+            var codec = new DeflateCodec();
+            byte[] payload = System.Text.Encoding.UTF8.GetBytes("hello world");
+            byte[] compressed = codec.Compress(payload);
+
+            var previous = Environment.GetEnvironmentVariable(Codec.MaxDecompressLengthEnvVar);
+            Environment.SetEnvironmentVariable(Codec.MaxDecompressLengthEnvVar, "65536"); // 64 KiB
+            try
+            {
+                byte[] result = codec.Decompress(compressed, compressed.Length);
+                CollectionAssert.AreEqual(payload, result);
+            }
+            finally
+            {
+                Environment.SetEnvironmentVariable(Codec.MaxDecompressLengthEnvVar, previous);
+            }
+        }
+
+        [Test]
+        public void TestCopyBoundedValidatesArguments()
+        {
+            using (var stream = new MemoryStream())
+            {
+                Assert.Throws<ArgumentNullException>(() => Codec.CopyBounded(null, stream, 10));
+                Assert.Throws<ArgumentNullException>(() => Codec.CopyBounded(stream, null, 10));
+                Assert.Throws<ArgumentOutOfRangeException>(() => Codec.CopyBounded(stream, stream, -1));
+            }
+        }
+
+        /// <summary>
+        /// The DataFileReader itself must reject a block whose decompressed size
+        /// exceeds the configured maximum. This covers the safeguard applied to
+        /// every codec that returns a fully materialized buffer (here the Null
+        /// codec, which performs no internally bounded decompression), not just a
+        /// direct call to a codec's Decompress method.
+        /// </summary>
+        [Test]
+        public void TestReaderRejectsOversizedBlock()
+        {
+            const string schemaStr =
+                "{\"type\":\"record\",\"name\":\"n\",\"fields\":[{\"name\":\"f1\",\"type\":\"string\"}]}";
+            Schema schema = Schema.Parse(schemaStr);
+            var recordSchema = schema as RecordSchema;
+
+            // A single record whose string field is larger than the limit below.
+            string big = new string('a', 128 * 1024); // 128 KiB
+
+            MemoryStream outStream = new MemoryStream();
+            using (var writer = DataFileWriter<GenericRecord>.OpenWriter(
+                new GenericWriter<GenericRecord>(schema), outStream, Codec.CreateCodec(Codec.Type.Null)))
+            {
+                writer.Append(mkRecord(new object[] { "f1", big }, recordSchema));
+            }
+
+            MemoryStream inStream = new MemoryStream(outStream.ToArray());
+
+            var previous = Environment.GetEnvironmentVariable(Codec.MaxDecompressLengthEnvVar);
+            Environment.SetEnvironmentVariable(Codec.MaxDecompressLengthEnvVar, "65536"); // 64 KiB
+            try
+            {
+                Assert.Throws<AvroRuntimeException>(() =>
+                {
+                    using (var reader = DataFileReader<GenericRecord>.OpenReader(inStream, schema))
+                    {
+                        // Enumerating forces the block to be read and
+                        // decompressed, which is where the limit is enforced.
+                        foreach (var rec in reader.NextEntries)
+                        {
+                            Assert.NotNull(rec);
+                        }
+                    }
+                });
+            }
+            finally
+            {
+                Environment.SetEnvironmentVariable(Codec.MaxDecompressLengthEnvVar, previous);
+            }
+        }
+
         /// <summary>
         /// This test is a single test case of
         /// <see cref="TestGenericData(string, object[], Codec.Type)"/> but introduces a