AVRO-4298: [php] Bound allocation when decoding length-prefixed values and collections

lang/php/lib/Datum/AvroIOBinaryDecoder.php

@@ -38,6 +38,14 @@
  */
 class AvroIOBinaryDecoder
 {
+    /**
+     * Reads with a declared length above this many bytes are validated against
+     * the number of bytes actually remaining before allocating, to guard
+     * against an out-of-memory attack from a malicious or truncated input.
+     * Smaller reads skip the check to avoid per-value overhead.
+     */
+    private const MAX_UNCHECKED_READ = 1048576; // 1 MiB
+
     /**
      * @param AvroIO $io object from which to read.
      */
@@ -65,9 +73,39 @@ public function readBoolean(): bool
      */
     public function read(int $len): string
     {
+        if ($len < 0) {
+            // AvroStringIO::read() accepts a negative length and moves the
+            // pointer backwards; reject it before delegating.
+            throw new AvroException("Cannot read a negative number of bytes: {$len}");
+        }
+        if ($len > self::MAX_UNCHECKED_READ) {
+            $remaining = $this->bytesRemaining();
+            if ($len > $remaining) {
+                throw new AvroException("Cannot read {$len} bytes, only {$remaining} remaining.");
+            }
+        }
+
         return $this->io->read($len);
     }
 
+    /**
+     * Number of bytes still available to read, determined by seeking to the end
+     * and restoring the position (which both the string and file IO
+     * implementations support). Used to reject a declared length or collection
+     * block count that exceeds the data actually available before allocating.
+     */
+    public function bytesRemaining(): int
+    {
+        $current = $this->io->tell();
+        $this->io->seek(0, AvroIO::SEEK_END);
+        $end = $this->io->tell();
+        $this->io->seek($current, AvroIO::SEEK_SET);
+
+        // Clamp to 0: AvroStringIO::seek() allows seeking past EOF, which would
+        // otherwise yield a confusing negative "remaining" count.
+        return max(0, $end - $current);
+    }
+
     public function readInt(): int
     {
         return (int) $this->readLong();

lang/php/lib/Datum/AvroIODatumReader.php

@@ -285,11 +285,12 @@ public function readArray(
     ): array {
         $items = [];
         $blockCount = $decoder->readLong();
-        while (0 !== $blockCount) {
+        while (0 != $blockCount) {
             if ($blockCount < 0) {
                 $blockCount = -$blockCount;
                 $decoder->readLong(); // Read (and ignore) block size
             }
+            self::ensureCollectionAvailable($decoder, $blockCount, self::minBytesPerElement($writersSchema->items()));
             for ($i = 0; $i < $blockCount; $i++) {
                 $items[] = $this->readData(
                     $writersSchema->items(),
@@ -320,6 +321,8 @@ public function readMap(
                 $decoder->readLong();
             }
 
+            // Map keys are strings (>= 1 byte length prefix) plus the value.
+            self::ensureCollectionAvailable($decoder, $pair_count, 1 + self::minBytesPerElement($writersSchema->values()));
             for ($i = 0; $i < $pair_count; $i++) {
                 $key = $decoder->readString();
                 $items[$key] = $this->readData(
@@ -538,4 +541,81 @@ private function readDecimal(string $bytes, int $scale): string
 
         return (string) ($scale > 0 ? ($int / (10 ** $scale)) : $int);
     }
+
+    /**
+     * Minimum number of bytes a single value of the given schema can occupy on
+     * the wire. Used to reject an array/map block count that could not be backed
+     * by the bytes remaining. It returns 0 for any schema that can encode to
+     * zero bytes: the null primitive, but also a record with no fields or whose
+     * fields all encode to zero bytes. A zero return disables the collection
+     * check for that element type (so, e.g., an array of nulls is not falsely
+     * rejected). Types that cannot be resolved cheaply default to 1.
+     *
+     * @param array<int, bool> $visited
+     */
+    private static function minBytesPerElement(mixed $schema, array $visited = []): int
+    {
+        $type = $schema instanceof AvroSchema ? $schema->type() : $schema;
+        // Named/complex field types may nest a schema object; unwrap one level.
+        if ($type instanceof AvroSchema) {
+            return self::minBytesPerElement($type, $visited);
+        }
+        if (!is_string($type)) {
+            return 1;
+        }
+        switch ($type) {
+            case AvroSchema::NULL_TYPE:
+                return 0;
+            case AvroSchema::FLOAT_TYPE:
+                return 4;
+            case AvroSchema::DOUBLE_TYPE:
+                return 8;
+            case AvroSchema::FIXED_SCHEMA:
+                return $schema instanceof AvroFixedSchema ? $schema->size() : 1;
+            case AvroSchema::RECORD_SCHEMA:
+            case AvroSchema::ERROR_SCHEMA:
+                if (!$schema instanceof AvroRecordSchema) {
+                    return 1;
+                }
+                $id = spl_object_id($schema);
+                if (isset($visited[$id])) {
+                    return 1; // self-referencing schema: safe lower bound of 1 byte
+                }
+                $visited[$id] = true;
+                $total = 0;
+                foreach ($schema->fields() as $field) {
+                    $total += self::minBytesPerElement($field, $visited);
+                }
+
+                return $total;
+            default:
+                // boolean, int, long, bytes, string, enum, union, array, map:
+                // all encode to at least one byte.
+                return 1;
+        }
+    }
+
+    /**
+     * Rejects a collection (array or map) block whose declared element count
+     * could not be backed by the bytes actually remaining, before iterating.
+     * Skipped when the per-element minimum is zero (e.g. an array of nulls).
+     *
+     * @throws AvroException if the declared count exceeds the bytes remaining
+     */
+    private static function ensureCollectionAvailable(
+        AvroIOBinaryDecoder $decoder,
+        int $count,
+        int $minBytesPerElement
+    ): void {
+        if ($count <= 0 || $minBytesPerElement <= 0) {
+            return;
+        }
+        $remaining = $decoder->bytesRemaining();
+        if ($count > intdiv($remaining, $minBytesPerElement)) {
+            throw new AvroException(
+                "Collection claims {$count} elements with at least {$minBytesPerElement} "
+                ."bytes each, but only {$remaining} bytes are available."
+            );
+        }
+    }
 }

lang/php/test/DatumIOTest.php

@@ -261,6 +261,66 @@ public function test_invalid_bytes_logical_type_out_of_range(): void
         $writer->write("10000", $encoder);
     }
 
+    // A bytes/string value declares a length prefix; a malicious or truncated
+    // input can declare far more bytes than actually exist. On a reader that
+    // can report its size, that is rejected before allocating for it.
+    public function test_read_bytes_rejects_length_beyond_stream(): void
+    {
+        $io = new AvroStringIO();
+        (new AvroIOBinaryEncoder($io))->writeLong(100 * 1024 * 1024);
+        $io->seek(0);
+        $decoder = new AvroIOBinaryDecoder($io);
+
+        $this->expectException(AvroException::class);
+        $decoder->readBytes();
+    }
+
+    public function test_read_bytes_within_stream_still_reads(): void
+    {
+        $payload = str_repeat('x', 2 * 1024 * 1024);
+        $io = new AvroStringIO();
+        (new AvroIOBinaryEncoder($io))->writeBytes($payload);
+        $io->seek(0);
+        $decoder = new AvroIOBinaryDecoder($io);
+
+        $this->assertEquals($payload, $decoder->readBytes());
+    }
+
+    public function test_read_array_rejects_count_beyond_stream(): void
+    {
+        $io = new AvroStringIO();
+        (new AvroIOBinaryEncoder($io))->writeLong(1000000); // 1,000,000 longs, no data
+        $this->expectException(AvroException::class);
+        $this->decodeWith('{"type":"array","items":"long"}', $io);
+    }
+
+    public function test_read_map_rejects_count_beyond_stream(): void
+    {
+        $io = new AvroStringIO();
+        (new AvroIOBinaryEncoder($io))->writeLong(1000000);
+        $this->expectException(AvroException::class);
+        $this->decodeWith('{"type":"map","values":"long"}', $io);
+    }
+
+    public function test_read_array_of_null_not_falsely_rejected(): void
+    {
+        $count = 100000;
+        $io = new AvroStringIO();
+        $encoder = new AvroIOBinaryEncoder($io);
+        $encoder->writeLong($count); // one block of `count` nulls (zero bytes each)
+        $encoder->writeLong(0);      // end-of-array marker
+        $result = $this->decodeWith('{"type":"array","items":"null"}', $io);
+        $this->assertCount($count, $result);
+    }
+
+    public function test_read_array_within_stream_still_reads(): void
+    {
+        $schema = AvroSchema::parse('{"type":"array","items":"long"}');
+        $io = new AvroStringIO();
+        (new AvroIODatumWriter($schema))->write([1, 2, 3], new AvroIOBinaryEncoder($io));
+        $this->assertEquals([1, 2, 3], $this->decodeWith('{"type":"array","items":"long"}', $io));
+    }
+
     public static function validDurationLogicalTypes(): array
     {
         return [
@@ -420,6 +480,20 @@ public function test_field_default_value(
         }
     }
 
+    // An array/map block declares an element count; a malicious or truncated
+    // input can declare far more elements than the remaining bytes could hold.
+    // The count is validated against the bytes remaining before iterating, using
+    // the minimum on-wire size of the element schema (so 0-byte elements like
+    // null are not falsely rejected).
+    private function decodeWith(string $schemaJson, AvroStringIO $io): mixed
+    {
+        $schema = AvroSchema::parse($schemaJson);
+        $io->seek(0);
+        $reader = new AvroIODatumReader($schema);
+
+        return $reader->read(new AvroIOBinaryDecoder($io));
+    }
+
     /**
      * @param string $datum
      * @param string $expected