From 92e0e5da7e7ef31c27233568adcf50298142e382 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Isma=C3=ABl=20Mej=C3=ADa?= <iemejia@gmail.com>
Date: Sat, 11 Jul 2026 22:27:56 +0200
Subject: [PATCH 01/16] AVRO-4298: [php] Validate available bytes before
 allocating for length-prefixed values and collections

A bytes or string value is a length prefix followed by that many bytes, and an
array or map block is an element count followed by that many items. A malicious
or truncated input can declare a huge length or count with little or no data.

- AvroIOBinaryDecoder::bytesRemaining() reports the bytes still readable (found
  by seeking to the end and restoring the position). read() uses it to reject an
  over-large declared length above a threshold before allocating.
- AvroIODatumReader::readArray/readMap reject a block whose element count could
  not be backed by the bytes remaining, using minBytesPerElement() computed from
  the element schema so a zero-byte element type (e.g. null) is not falsely
  rejected.

Mirrors the Java SDK's checks (AVRO-4241).

Assisted-by: GitHub Copilot:claude-opus-4.8
---
 lang/php/lib/Datum/AvroIOBinaryDecoder.php | 31 +++++++++
 lang/php/lib/Datum/AvroIODatumReader.php   | 79 ++++++++++++++++++++++
 lang/php/test/DatumIOTest.php              | 74 ++++++++++++++++++++
 3 files changed, 184 insertions(+)

diff --git a/lang/php/lib/Datum/AvroIOBinaryDecoder.php b/lang/php/lib/Datum/AvroIOBinaryDecoder.php
index ac5fcf7e929..b1cf96532e1 100644
--- a/lang/php/lib/Datum/AvroIOBinaryDecoder.php
+++ b/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,32 @@ public function readBoolean(): bool
      */
     public function read(int $len): string
     {
+        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);
+
+        return $end - $current;
+    }
+
     public function readInt(): int
     {
         return (int) $this->readLong();
diff --git a/lang/php/lib/Datum/AvroIODatumReader.php b/lang/php/lib/Datum/AvroIODatumReader.php
index 48f193f4c02..f6746b667fc 100644
--- a/lang/php/lib/Datum/AvroIODatumReader.php
+++ b/lang/php/lib/Datum/AvroIODatumReader.php
@@ -290,6 +290,7 @@ public function readArray(
                 $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,80 @@ 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. A type that can encode to zero bytes (null)
+     * returns 0, which disables the collection check for it (so an array of
+     * nulls is not falsely rejected). Types that cannot be resolved cheaply
+     * default to 1, which is safe because the only zero-byte primitive is null.
+     *
+     * @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 0; // break recursion for self-referencing schemas
+                }
+                $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 * $minBytesPerElement > $remaining) {
+            throw new AvroException(
+                "Collection claims {$count} elements with at least {$minBytesPerElement} "
+                ."bytes each, but only {$remaining} bytes are available."
+            );
+        }
+    }
 }
diff --git a/lang/php/test/DatumIOTest.php b/lang/php/test/DatumIOTest.php
index 8614cf61365..5c8eb301914 100644
--- a/lang/php/test/DatumIOTest.php
+++ b/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

From 5e997eba6a7df631cdc7a3d71ec21ba34dc22d21 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Isma=C3=ABl=20Mej=C3=ADa?= <iemejia@gmail.com>
Date: Sun, 12 Jul 2026 00:25:43 +0200
Subject: [PATCH 02/16] AVRO-4298: [php] Reject negative lengths; avoid
 overflow; GMP-safe block loop

Review feedback:
- read(int) now rejects a negative length. AvroStringIO::read() accepts a
  negative value and moves the pointer backwards, leading to incorrect decoding.
- ensureCollectionAvailable compares count against intdiv(remaining, minBytes)
  rather than multiplying, so a large count cannot overflow (to a float or wrap)
  and slip past the guard.
- readArray uses a non-strict comparison (0 != blockCount) to match readMap; on
  32-bit builds with GMP, readLong returns numeric strings, so a strict compare
  against 0 would not terminate the loop.

Assisted-by: GitHub Copilot:claude-opus-4.8
---
 lang/php/lib/Datum/AvroIOBinaryDecoder.php | 5 +++++
 lang/php/lib/Datum/AvroIODatumReader.php   | 4 ++--
 2 files changed, 7 insertions(+), 2 deletions(-)

diff --git a/lang/php/lib/Datum/AvroIOBinaryDecoder.php b/lang/php/lib/Datum/AvroIOBinaryDecoder.php
index b1cf96532e1..c918c838751 100644
--- a/lang/php/lib/Datum/AvroIOBinaryDecoder.php
+++ b/lang/php/lib/Datum/AvroIOBinaryDecoder.php
@@ -73,6 +73,11 @@ 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) {
diff --git a/lang/php/lib/Datum/AvroIODatumReader.php b/lang/php/lib/Datum/AvroIODatumReader.php
index f6746b667fc..feb4165d75e 100644
--- a/lang/php/lib/Datum/AvroIODatumReader.php
+++ b/lang/php/lib/Datum/AvroIODatumReader.php
@@ -285,7 +285,7 @@ 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
@@ -610,7 +610,7 @@ private static function ensureCollectionAvailable(
             return;
         }
         $remaining = $decoder->bytesRemaining();
-        if ($count * $minBytesPerElement > $remaining) {
+        if ($count > intdiv($remaining, $minBytesPerElement)) {
             throw new AvroException(
                 "Collection claims {$count} elements with at least {$minBytesPerElement} "
                 ."bytes each, but only {$remaining} bytes are available."

From f3a4764a7c62c3d984416599fc2878ebfbca7cd0 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Isma=C3=ABl=20Mej=C3=ADa?= <iemejia@gmail.com>
Date: Sun, 12 Jul 2026 01:04:24 +0200
Subject: [PATCH 03/16] AVRO-4298: [php] Clamp bytesRemaining to 0;
 conservative cycle minimum

Review feedback:
- bytesRemaining() clamps to 0 with max(0, ...), since AvroStringIO::seek()
  allows seeking past EOF, which would otherwise produce a confusing negative
  remaining count.
- minBytesPerElement() returns 1 (not 0) when a self-referencing record cycle is
  detected, keeping the collection-availability guard conservative rather than
  disabling it for arrays/maps of recursive records.

Assisted-by: GitHub Copilot:claude-opus-4.8
---
 lang/php/lib/Datum/AvroIOBinaryDecoder.php | 4 +++-
 lang/php/lib/Datum/AvroIODatumReader.php   | 2 +-
 2 files changed, 4 insertions(+), 2 deletions(-)

diff --git a/lang/php/lib/Datum/AvroIOBinaryDecoder.php b/lang/php/lib/Datum/AvroIOBinaryDecoder.php
index c918c838751..d43aa9263c3 100644
--- a/lang/php/lib/Datum/AvroIOBinaryDecoder.php
+++ b/lang/php/lib/Datum/AvroIOBinaryDecoder.php
@@ -101,7 +101,9 @@ public function bytesRemaining(): int
         $end = $this->io->tell();
         $this->io->seek($current, AvroIO::SEEK_SET);
 
-        return $end - $current;
+        // 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
diff --git a/lang/php/lib/Datum/AvroIODatumReader.php b/lang/php/lib/Datum/AvroIODatumReader.php
index feb4165d75e..3310eb89f2c 100644
--- a/lang/php/lib/Datum/AvroIODatumReader.php
+++ b/lang/php/lib/Datum/AvroIODatumReader.php
@@ -578,7 +578,7 @@ private static function minBytesPerElement(mixed $schema, array $visited = []):
                 }
                 $id = spl_object_id($schema);
                 if (isset($visited[$id])) {
-                    return 0; // break recursion for self-referencing schemas
+                    return 1; // self-referencing schema: safe lower bound of 1 byte
                 }
                 $visited[$id] = true;
                 $total = 0;

From bd6bb27d8cc2040ed15e8166197724787696e9be Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Isma=C3=ABl=20Mej=C3=ADa?= <iemejia@gmail.com>
Date: Sun, 12 Jul 2026 01:27:54 +0200
Subject: [PATCH 04/16] AVRO-4298: [php] Clarify minBytesPerElement doc comment

Review feedback: the doc comment claimed only null can encode to zero bytes, but
minBytesPerElement() also returns 0 for a record with no fields or whose fields
all encode to zero bytes. Update the comment so a 0 return (which disables the
collection guard for that element type) is not misunderstood in future changes.

Assisted-by: GitHub Copilot:claude-opus-4.8
---
 lang/php/lib/Datum/AvroIODatumReader.php | 9 +++++----
 1 file changed, 5 insertions(+), 4 deletions(-)

diff --git a/lang/php/lib/Datum/AvroIODatumReader.php b/lang/php/lib/Datum/AvroIODatumReader.php
index 3310eb89f2c..1fd62817f32 100644
--- a/lang/php/lib/Datum/AvroIODatumReader.php
+++ b/lang/php/lib/Datum/AvroIODatumReader.php
@@ -545,10 +545,11 @@ private function readDecimal(string $bytes, int $scale): string
     /**
      * 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. A type that can encode to zero bytes (null)
-     * returns 0, which disables the collection check for it (so an array of
-     * nulls is not falsely rejected). Types that cannot be resolved cheaply
-     * default to 1, which is safe because the only zero-byte primitive is null.
+     * 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
      */

From 3d7f73665b4b16f80b8bfa01b9301867583e6e46 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Isma=C3=ABl=20Mej=C3=ADa?= <iemejia@gmail.com>
Date: Sun, 12 Jul 2026 13:46:23 +0200
Subject: [PATCH 05/16] AVRO-4298: [php] Cap zero-byte collection element
 allocation

Completes the available-bytes protection for collections and supersedes the
separate collection-limit change. Elements whose schema encodes to zero bytes
(null, a zero-length fixed, or a record with only zero-byte fields) consume no
input, so the bytes-remaining check cannot bound their count. A tiny payload
declaring a huge array block count of such elements (e.g.
{"type":"array","items":"null"} with a count of 200,000,000) therefore drove an
unbounded allocation and exhausted memory.

ensureCollectionAvailable now enforces the bytes-remaining check plus a
zero-byte cap (DEFAULT_MAX_COLLECTION_ITEMS = 10,000,000) and a structural cap on
all collections (DEFAULT_MAX_COLLECTION_STRUCTURAL = Integer.MAX_VALUE - 8).
AVRO_MAX_COLLECTION_ITEMS, when set, caps both. Rejections raise the dedicated
AvroIOCollectionSizeException. The decoder's skipArray/skipMap are also bounded
(element-aware) so a huge block cannot loop unboundedly.

Assisted-by: GitHub Copilot:claude-opus-4.8
---
 lang/php/lib/Datum/AvroIOBinaryDecoder.php    |  25 ++-
 .../Datum/AvroIOCollectionSizeException.php   |  43 +++++
 lang/php/lib/Datum/AvroIODatumReader.php      | 116 ++++++++++--
 lang/php/lib/autoload.php                     |   1 +
 lang/php/test/DatumIOTest.php                 | 179 ++++++++++++++++++
 5 files changed, 344 insertions(+), 20 deletions(-)
 create mode 100644 lang/php/lib/Datum/AvroIOCollectionSizeException.php

diff --git a/lang/php/lib/Datum/AvroIOBinaryDecoder.php b/lang/php/lib/Datum/AvroIOBinaryDecoder.php
index d43aa9263c3..52f7dc7dc9c 100644
--- a/lang/php/lib/Datum/AvroIOBinaryDecoder.php
+++ b/lang/php/lib/Datum/AvroIOBinaryDecoder.php
@@ -273,13 +273,18 @@ public function skipRecord(AvroRecordSchema $writersSchema, AvroIOBinaryDecoder
 
     public function skipArray(AvroArraySchema $writersSchema, AvroIOBinaryDecoder $decoder): void
     {
+        $minBytes = AvroIODatumReader::collectionElementMinBytes($writersSchema->items());
+        $skipped = 0;
         $blockCount = $decoder->readLong();
         while (0 !== $blockCount) {
             if ($blockCount < 0) {
                 $decoder->skip($this->readLong());
-            }
-            for ($i = 0; $i < $blockCount; $i++) {
-                AvroIODatumReader::skipData($writersSchema->items(), $decoder);
+            } else {
+                AvroIODatumReader::checkSkipCollectionCount($skipped, $blockCount, $minBytes);
+                $skipped += $blockCount;
+                for ($i = 0; $i < $blockCount; $i++) {
+                    AvroIODatumReader::skipData($writersSchema->items(), $decoder);
+                }
             }
             $blockCount = $decoder->readLong();
         }
@@ -287,14 +292,20 @@ public function skipArray(AvroArraySchema $writersSchema, AvroIOBinaryDecoder $d
 
     public function skipMap(AvroMapSchema $writersSchema, AvroIOBinaryDecoder $decoder): void
     {
+        // Map entries always carry a >= 1 byte key, so the minimum is positive.
+        $minBytes = 1 + AvroIODatumReader::collectionElementMinBytes($writersSchema->values());
+        $skipped = 0;
         $blockCount = $decoder->readLong();
         while (0 !== $blockCount) {
             if ($blockCount < 0) {
                 $decoder->skip($this->readLong());
-            }
-            for ($i = 0; $i < $blockCount; $i++) {
-                $decoder->skipString();
-                AvroIODatumReader::skipData($writersSchema->values(), $decoder);
+            } else {
+                AvroIODatumReader::checkSkipCollectionCount($skipped, $blockCount, $minBytes);
+                $skipped += $blockCount;
+                for ($i = 0; $i < $blockCount; $i++) {
+                    $decoder->skipString();
+                    AvroIODatumReader::skipData($writersSchema->values(), $decoder);
+                }
             }
             $blockCount = $decoder->readLong();
         }
diff --git a/lang/php/lib/Datum/AvroIOCollectionSizeException.php b/lang/php/lib/Datum/AvroIOCollectionSizeException.php
new file mode 100644
index 00000000000..6aa9b0bce0a
--- /dev/null
+++ b/lang/php/lib/Datum/AvroIOCollectionSizeException.php
@@ -0,0 +1,43 @@
+<?php
+
+/**
+ * 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.
+ */
+
+declare(strict_types=1);
+
+namespace Apache\Avro\Datum;
+
+use Apache\Avro\AvroException;
+
+/**
+ * Raised when an array or map declares more items than the configured maximum.
+ *
+ * The block count of an array or map is read from the (potentially untrusted or
+ * truncated) input and drives allocation of the resulting collection. This
+ * exception guards against unbounded memory allocation from a very large or
+ * malformed block count.
+ */
+class AvroIOCollectionSizeException extends AvroException
+{
+    public function __construct(int $maxItems)
+    {
+        parent::__construct(
+            sprintf('Cannot read collections larger than %d items.', $maxItems)
+        );
+    }
+}
diff --git a/lang/php/lib/Datum/AvroIODatumReader.php b/lang/php/lib/Datum/AvroIODatumReader.php
index 1fd62817f32..38763a4994f 100644
--- a/lang/php/lib/Datum/AvroIODatumReader.php
+++ b/lang/php/lib/Datum/AvroIODatumReader.php
@@ -43,6 +43,29 @@
  */
 class AvroIODatumReader
 {
+    /**
+     * Name of the environment variable overriding the maximum number of
+     * zero-byte-encoded collection elements (e.g. an array of nulls) to
+     * allocate from a single decode.
+     */
+    public const MAX_COLLECTION_ITEMS_ENV = 'AVRO_MAX_COLLECTION_ITEMS';
+
+    /**
+     * Default maximum number of zero-byte-encoded collection elements to
+     * allocate. Such elements consume no input, so the bytes-remaining check
+     * cannot bound their count; without a cap a tiny payload can declare a huge
+     * block count and exhaust memory. Overridable with the
+     * {@see self::MAX_COLLECTION_ITEMS_ENV} environment variable.
+     */
+    public const DEFAULT_MAX_COLLECTION_ITEMS = 10000000;
+
+    /**
+     * Structural cap on the number of elements in any array or map (an overflow
+     * / defense-in-depth guard), matching the historical Integer.MAX_VALUE - 8
+     * limit. Non-zero-byte elements are also bounded by the bytes remaining.
+     */
+    public const DEFAULT_MAX_COLLECTION_STRUCTURAL = 2147483639;
+
     public function __construct(
         private ?AvroSchema $writersSchema = null,
         private ?AvroSchema $readersSchema = null
@@ -284,13 +307,14 @@ public function readArray(
         AvroIOBinaryDecoder $decoder
     ): array {
         $items = [];
+        $minBytes = self::minBytesPerElement($writersSchema->items());
         $blockCount = $decoder->readLong();
         while (0 != $blockCount) {
             if ($blockCount < 0) {
                 $blockCount = -$blockCount;
                 $decoder->readLong(); // Read (and ignore) block size
             }
-            self::ensureCollectionAvailable($decoder, $blockCount, self::minBytesPerElement($writersSchema->items()));
+            self::ensureCollectionAvailable($decoder, count($items), $blockCount, $minBytes);
             for ($i = 0; $i < $blockCount; $i++) {
                 $items[] = $this->readData(
                     $writersSchema->items(),
@@ -313,6 +337,7 @@ public function readMap(
         AvroIOBinaryDecoder $decoder
     ): array {
         $items = [];
+        $minBytes = 1 + self::minBytesPerElement($writersSchema->values());
         $pair_count = $decoder->readLong();
         while (0 != $pair_count) {
             if ($pair_count < 0) {
@@ -322,7 +347,7 @@ public function readMap(
             }
 
             // Map keys are strings (>= 1 byte length prefix) plus the value.
-            self::ensureCollectionAvailable($decoder, $pair_count, 1 + self::minBytesPerElement($writersSchema->values()));
+            self::ensureCollectionAvailable($decoder, count($items), $pair_count, $minBytes);
             for ($i = 0; $i < $pair_count; $i++) {
                 $key = $decoder->readString();
                 $items[$key] = $this->readData(
@@ -533,6 +558,34 @@ public function readDefaultValue(AvroSchema $fieldSchema, mixed $defaultValue):
         }
     }
 
+    /**
+     * Minimum on-wire size of a collection element schema. Exposed for the skip
+     * path, which lives in the decoder.
+     */
+    public static function collectionElementMinBytes(AvroSchema $schema): int
+    {
+        return self::minBytesPerElement($schema);
+    }
+
+    /**
+     * Bounds the cumulative number of elements skipped in an array or map, so a
+     * huge block of zero-byte elements cannot loop unboundedly (a CPU
+     * exhaustion) even though skipping allocates nothing.
+     *
+     * @throws AvroIOCollectionSizeException if the limit is exceeded
+     */
+    public static function checkSkipCollectionCount(int $existing, int $count, int $minBytes): void
+    {
+        if ($count <= 0) {
+            return;
+        }
+        [$zeroByteLimit, $structuralLimit] = self::collectionLimits();
+        $limit = $minBytes > 0 ? $structuralLimit : $zeroByteLimit;
+        if ($count > $limit || $existing > $limit - $count) {
+            throw new AvroIOCollectionSizeException($limit);
+        }
+    }
+
     private function readDecimal(string $bytes, int $scale): string
     {
         $mostSignificantBit = ord($bytes[0]) & 0x80;
@@ -596,26 +649,63 @@ private static function minBytesPerElement(mixed $schema, array $visited = []):
     }
 
     /**
-     * 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).
+     * Returns the configured collection limits as [zeroByteLimit, structuralLimit].
+     * AVRO_MAX_COLLECTION_ITEMS, when a non-negative integer, caps both.
      *
-     * @throws AvroException if the declared count exceeds the bytes remaining
+     * @return array{int, int}
+     */
+    private static function collectionLimits(): array
+    {
+        $value = getenv(self::MAX_COLLECTION_ITEMS_ENV);
+        if (false !== $value && '' !== $value && preg_match('/^-?\d+$/', $value)) {
+            $parsed = (int) $value;
+            if ($parsed >= 0) {
+                return [$parsed, $parsed];
+            }
+        }
+
+        return [self::DEFAULT_MAX_COLLECTION_ITEMS, self::DEFAULT_MAX_COLLECTION_STRUCTURAL];
+    }
+
+    /**
+     * Rejects a collection (array or map) block that could not be backed by the
+     * input, before iterating.
+     *
+     * For elements with a positive minimum on-wire size, the declared count is
+     * checked against the bytes remaining and a structural cap. For zero-byte
+     * elements (e.g. an array of nulls), which consume no input and so cannot be
+     * bounded by the bytes remaining, the cumulative count is checked against the
+     * tighter zero-byte limit.
+     *
+     * @throws AvroException                 if the bytes-remaining check fails
+     * @throws AvroIOCollectionSizeException if a size limit is exceeded
      */
     private static function ensureCollectionAvailable(
         AvroIOBinaryDecoder $decoder,
+        int $existing,
         int $count,
         int $minBytesPerElement
     ): void {
-        if ($count <= 0 || $minBytesPerElement <= 0) {
+        if ($count <= 0) {
+            return;
+        }
+        [$zeroByteLimit, $structuralLimit] = self::collectionLimits();
+        if ($minBytesPerElement > 0) {
+            $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."
+                );
+            }
+            if ($count > $structuralLimit || $existing > $structuralLimit - $count) {
+                throw new AvroIOCollectionSizeException($structuralLimit);
+            }
+
             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."
-            );
+        if ($count > $zeroByteLimit || $existing > $zeroByteLimit - $count) {
+            throw new AvroIOCollectionSizeException($zeroByteLimit);
         }
     }
 }
diff --git a/lang/php/lib/autoload.php b/lang/php/lib/autoload.php
index de1a863513b..12c9475e794 100644
--- a/lang/php/lib/autoload.php
+++ b/lang/php/lib/autoload.php
@@ -33,6 +33,7 @@
 
 include __DIR__.'/Datum/AvroIOBinaryDecoder.php';
 include __DIR__.'/Datum/AvroIOBinaryEncoder.php';
+include __DIR__.'/Datum/AvroIOCollectionSizeException.php';
 include __DIR__.'/Datum/AvroIODatumReader.php';
 include __DIR__.'/Datum/AvroIODatumWriter.php';
 include __DIR__.'/Datum/AvroIOSchemaMatchException.php';
diff --git a/lang/php/test/DatumIOTest.php b/lang/php/test/DatumIOTest.php
index 5c8eb301914..c169259f531 100644
--- a/lang/php/test/DatumIOTest.php
+++ b/lang/php/test/DatumIOTest.php
@@ -24,6 +24,7 @@
 use Apache\Avro\AvroException;
 use Apache\Avro\Datum\AvroIOBinaryDecoder;
 use Apache\Avro\Datum\AvroIOBinaryEncoder;
+use Apache\Avro\Datum\AvroIOCollectionSizeException;
 use Apache\Avro\Datum\AvroIODatumReader;
 use Apache\Avro\Datum\AvroIODatumWriter;
 use Apache\Avro\Datum\Type\AvroDuration;
@@ -321,6 +322,165 @@ public function test_read_array_within_stream_still_reads(): void
         $this->assertEquals([1, 2, 3], $this->decodeWith('{"type":"array","items":"long"}', $io));
     }
 
+    public function test_array_of_null_exceeds_default_limit(): void
+    {
+        // The reported exploit: a ~6 byte payload declaring 200,000,000 nulls.
+        $this->expectException(AvroIOCollectionSizeException::class);
+        $this->decodeWith('{"type":"array","items":"null"}', self::zeroByteBlock(200000000));
+    }
+
+    public function test_array_of_null_within_configured_limit_still_reads(): void
+    {
+        putenv('AVRO_MAX_COLLECTION_ITEMS=1000');
+
+        try {
+            $result = $this->decodeWith('{"type":"array","items":"null"}', self::zeroByteBlock(1000));
+            $this->assertCount(1000, $result);
+        } finally {
+            putenv('AVRO_MAX_COLLECTION_ITEMS');
+        }
+    }
+
+    public function test_array_of_null_exceeds_configured_limit(): void
+    {
+        putenv('AVRO_MAX_COLLECTION_ITEMS=1000');
+
+        try {
+            $this->expectException(AvroException::class);
+            $this->decodeWith('{"type":"array","items":"null"}', self::zeroByteBlock(1001));
+        } finally {
+            putenv('AVRO_MAX_COLLECTION_ITEMS');
+        }
+    }
+
+    public function test_array_of_null_cumulative_across_blocks(): void
+    {
+        $io = new AvroStringIO();
+        $encoder = new AvroIOBinaryEncoder($io);
+        $encoder->writeLong(600);
+        $encoder->writeLong(600);
+        $encoder->writeLong(0);
+        putenv('AVRO_MAX_COLLECTION_ITEMS=1000');
+
+        try {
+            $this->expectException(AvroException::class);
+            $this->decodeWith('{"type":"array","items":"null"}', $io);
+        } finally {
+            putenv('AVRO_MAX_COLLECTION_ITEMS');
+        }
+    }
+
+    public function test_array_of_null_negative_block_count(): void
+    {
+        putenv('AVRO_MAX_COLLECTION_ITEMS=1000');
+
+        try {
+            $this->expectException(AvroException::class);
+            $this->decodeWith('{"type":"array","items":"null"}', self::zeroByteBlock(200000, true));
+        } finally {
+            putenv('AVRO_MAX_COLLECTION_ITEMS');
+        }
+    }
+
+    public function test_array_of_zero_length_fixed_exceeds_limit(): void
+    {
+        putenv('AVRO_MAX_COLLECTION_ITEMS=1000');
+
+        try {
+            $this->expectException(AvroException::class);
+            $this->decodeWith('{"type":"array","items":{"type":"fixed","name":"empty","size":0}}', self::zeroByteBlock(2000));
+        } finally {
+            putenv('AVRO_MAX_COLLECTION_ITEMS');
+        }
+    }
+
+    public function test_array_of_all_null_record_exceeds_limit(): void
+    {
+        putenv('AVRO_MAX_COLLECTION_ITEMS=1000');
+
+        try {
+            $this->expectException(AvroException::class);
+            $this->decodeWith(
+                '{"type":"array","items":{"type":"record","name":"R","fields":[{"name":"n","type":"null"}]}}',
+                self::zeroByteBlock(2000)
+            );
+        } finally {
+            putenv('AVRO_MAX_COLLECTION_ITEMS');
+        }
+    }
+
+    public function test_map_of_null_rejected_by_available_bytes(): void
+    {
+        // Map entries always carry a >= 1 byte key, so a huge map<null> is bounded
+        // by the bytes-remaining check.
+        $this->expectException(AvroException::class);
+        $this->decodeWith('{"type":"map","values":"null"}', self::zeroByteBlock(200000000));
+    }
+
+    public function test_invalid_env_override_falls_back_to_default(): void
+    {
+        // An invalid limit is ignored, so the default (10,000,000) still rejects
+        // the 200,000,000 exploit.
+        putenv('AVRO_MAX_COLLECTION_ITEMS=not-a-number');
+
+        try {
+            $this->expectException(AvroException::class);
+            $this->decodeWith('{"type":"array","items":"null"}', self::zeroByteBlock(200000000));
+        } finally {
+            putenv('AVRO_MAX_COLLECTION_ITEMS');
+        }
+    }
+
+    public function test_env_override_raises_limit(): void
+    {
+        // Raising the limit above the count lets a large null array decode.
+        putenv('AVRO_MAX_COLLECTION_ITEMS=20000');
+
+        try {
+            $result = $this->decodeWith('{"type":"array","items":"null"}', self::zeroByteBlock(15000));
+            $this->assertCount(15000, $result);
+        } finally {
+            putenv('AVRO_MAX_COLLECTION_ITEMS');
+        }
+    }
+
+    public function test_non_zero_collection_bounded_by_structural_cap(): void
+    {
+        // A backed array<long> that passes the bytes check is still rejected once
+        // its count exceeds the (env-lowered) structural cap.
+        $io = new AvroStringIO();
+        $encoder = new AvroIOBinaryEncoder($io);
+        $encoder->writeLong(10);           // block count 10
+        for ($i = 0; $i < 10; $i++) {
+            $encoder->writeLong($i);       // 10 real longs
+        }
+        $encoder->writeLong(0);
+        putenv('AVRO_MAX_COLLECTION_ITEMS=5');
+
+        try {
+            $this->expectException(AvroIOCollectionSizeException::class);
+            $this->decodeWith('{"type":"array","items":"long"}', $io);
+        } finally {
+            putenv('AVRO_MAX_COLLECTION_ITEMS');
+        }
+    }
+
+    public function test_skip_array_of_null_respects_limit(): void
+    {
+        // Skipping a huge zero-byte array (e.g. during projection) is bounded.
+        $schema = AvroSchema::parse('{"type":"array","items":"null"}');
+        $io = self::zeroByteBlock(200000000);
+        $io->seek(0);
+        putenv('AVRO_MAX_COLLECTION_ITEMS=1000');
+
+        try {
+            $this->expectException(AvroIOCollectionSizeException::class);
+            AvroIODatumReader::skipData($schema, new AvroIOBinaryDecoder($io));
+        } finally {
+            putenv('AVRO_MAX_COLLECTION_ITEMS');
+        }
+    }
+
     public static function validDurationLogicalTypes(): array
     {
         return [
@@ -480,6 +640,25 @@ public function test_field_default_value(
         }
     }
 
+    /**
+     * Builds one array/map block of `$count` zero-byte elements plus an end
+     * marker (a negative count is preceded by a block byte-size).
+     */
+    private static function zeroByteBlock(int $count, bool $negative = false): AvroStringIO
+    {
+        $io = new AvroStringIO();
+        $encoder = new AvroIOBinaryEncoder($io);
+        if ($negative) {
+            $encoder->writeLong(-$count);
+            $encoder->writeLong(0); // block byte-size
+        } else {
+            $encoder->writeLong($count);
+        }
+        $encoder->writeLong(0); // end-of-collection marker
+
+        return $io;
+    }
+
     // 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

From 695a4b22fe74c672e9c81bf882483dcd5651f5a2 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Isma=C3=ABl=20Mej=C3=ADa?= <iemejia@gmail.com>
Date: Sun, 12 Jul 2026 18:24:10 +0200
Subject: [PATCH 06/16] AVRO-4298: [php] Reject PHP_INT_MIN array/map block
 count

A negative block count is normalized by negating it, but PHP_INT_MIN cannot be
negated: -PHP_INT_MIN silently promotes to a float. The float count then hits
the int type hint of ensureCollectionAvailable() and raises a raw TypeError
rather than a proper Avro exception. The zig-zag encoding of PHP_INT_MIN is a
valid 10-byte varint, so this is reachable from malformed input.

Reject PHP_INT_MIN explicitly in readArray/readMap before negating, matching the
C, C++ and C# bindings. The skip path already byte-skips negative blocks without
negating, so it is unaffected. Adds a test for a PHP_INT_MIN array block count.

Assisted-by: GitHub Copilot:claude-opus-4.8
---
 lang/php/lib/Datum/AvroIODatumReader.php | 10 ++++++++++
 lang/php/test/DatumIOTest.php            | 12 ++++++++++++
 2 files changed, 22 insertions(+)

diff --git a/lang/php/lib/Datum/AvroIODatumReader.php b/lang/php/lib/Datum/AvroIODatumReader.php
index 38763a4994f..14f89b2543d 100644
--- a/lang/php/lib/Datum/AvroIODatumReader.php
+++ b/lang/php/lib/Datum/AvroIODatumReader.php
@@ -311,6 +311,11 @@ public function readArray(
         $blockCount = $decoder->readLong();
         while (0 != $blockCount) {
             if ($blockCount < 0) {
+                // PHP_INT_MIN cannot be negated: -PHP_INT_MIN promotes to a
+                // float, so reject it rather than propagating a non-int count.
+                if (PHP_INT_MIN === $blockCount) {
+                    throw new AvroException('Invalid array block count');
+                }
                 $blockCount = -$blockCount;
                 $decoder->readLong(); // Read (and ignore) block size
             }
@@ -341,6 +346,11 @@ public function readMap(
         $pair_count = $decoder->readLong();
         while (0 != $pair_count) {
             if ($pair_count < 0) {
+                // PHP_INT_MIN cannot be negated: -PHP_INT_MIN promotes to a
+                // float, so reject it rather than propagating a non-int count.
+                if (PHP_INT_MIN === $pair_count) {
+                    throw new AvroException('Invalid map block count');
+                }
                 $pair_count = -$pair_count;
                 // Note: we're not doing anything with block_size other than skipping it
                 $decoder->readLong();
diff --git a/lang/php/test/DatumIOTest.php b/lang/php/test/DatumIOTest.php
index c169259f531..06a7f876da9 100644
--- a/lang/php/test/DatumIOTest.php
+++ b/lang/php/test/DatumIOTest.php
@@ -329,6 +329,18 @@ public function test_array_of_null_exceeds_default_limit(): void
         $this->decodeWith('{"type":"array","items":"null"}', self::zeroByteBlock(200000000));
     }
 
+    public function test_array_of_null_int64_min_block_count(): void
+    {
+        // PHP_INT_MIN as a block count cannot be negated (-PHP_INT_MIN promotes
+        // to a float), so it must be rejected explicitly. It zig-zag encodes as
+        // the 10-byte varint below, followed by a block byte-size (0).
+        $io = new AvroStringIO(
+            "\xFF\xFF\xFF\xFF\xFF\xFF\xFF\xFF\xFF\x01\x00"
+        );
+        $this->expectException(AvroException::class);
+        $this->decodeWith('{"type":"array","items":"null"}', $io);
+    }
+
     public function test_array_of_null_within_configured_limit_still_reads(): void
     {
         putenv('AVRO_MAX_COLLECTION_ITEMS=1000');

From 23f15e04f4de96c88eb73d088d7d79a752bce9c6 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Isma=C3=ABl=20Mej=C3=ADa?= <iemejia@gmail.com>
Date: Sun, 12 Jul 2026 19:19:49 +0200
Subject: [PATCH 07/16] AVRO-4298: [php] Fix cumulative map cap, GMP skip
 loops, minor cleanups

Addresses review feedback:
 - readMap now bounds against a separate cumulative pairs-read counter instead
   of count($items). Keyed by map key, count($items) shrinks when a stream
   repeats the same key, which could slip past the cumulative/structural cap.
   Adds a regression test.
 - skipArray/skipMap loop conditions use a non-strict comparison (0 != $count)
   like readArray/readMap, so a GMP numeric-string block count of '0' terminates
   the loop on 32-bit builds.
 - minBytesPerElement traverses record fields via $field->type() (the field's
   value schema) explicitly rather than relying on AvroField extending
   AvroSchema.
 - Tighten the collection-limit tests to assert AvroIOCollectionSizeException,
   and reword the collectionLimits() doc comment (the env var overrides both
   limits to the given value rather than only capping them).

Assisted-by: GitHub Copilot:claude-opus-4.8
---
 lang/php/lib/Datum/AvroIOBinaryDecoder.php |  4 +--
 lang/php/lib/Datum/AvroIODatumReader.php   | 12 ++++++---
 lang/php/test/DatumIOTest.php              | 29 +++++++++++++++++++---
 3 files changed, 37 insertions(+), 8 deletions(-)

diff --git a/lang/php/lib/Datum/AvroIOBinaryDecoder.php b/lang/php/lib/Datum/AvroIOBinaryDecoder.php
index 52f7dc7dc9c..14743e8fff3 100644
--- a/lang/php/lib/Datum/AvroIOBinaryDecoder.php
+++ b/lang/php/lib/Datum/AvroIOBinaryDecoder.php
@@ -276,7 +276,7 @@ public function skipArray(AvroArraySchema $writersSchema, AvroIOBinaryDecoder $d
         $minBytes = AvroIODatumReader::collectionElementMinBytes($writersSchema->items());
         $skipped = 0;
         $blockCount = $decoder->readLong();
-        while (0 !== $blockCount) {
+        while (0 != $blockCount) {
             if ($blockCount < 0) {
                 $decoder->skip($this->readLong());
             } else {
@@ -296,7 +296,7 @@ public function skipMap(AvroMapSchema $writersSchema, AvroIOBinaryDecoder $decod
         $minBytes = 1 + AvroIODatumReader::collectionElementMinBytes($writersSchema->values());
         $skipped = 0;
         $blockCount = $decoder->readLong();
-        while (0 !== $blockCount) {
+        while (0 != $blockCount) {
             if ($blockCount < 0) {
                 $decoder->skip($this->readLong());
             } else {
diff --git a/lang/php/lib/Datum/AvroIODatumReader.php b/lang/php/lib/Datum/AvroIODatumReader.php
index 14f89b2543d..def43e8c059 100644
--- a/lang/php/lib/Datum/AvroIODatumReader.php
+++ b/lang/php/lib/Datum/AvroIODatumReader.php
@@ -343,6 +343,7 @@ public function readMap(
     ): array {
         $items = [];
         $minBytes = 1 + self::minBytesPerElement($writersSchema->values());
+        $read = 0; // Cumulative pairs read; count($items) would undercount duplicate keys.
         $pair_count = $decoder->readLong();
         while (0 != $pair_count) {
             if ($pair_count < 0) {
@@ -357,7 +358,11 @@ public function readMap(
             }
 
             // Map keys are strings (>= 1 byte length prefix) plus the value.
-            self::ensureCollectionAvailable($decoder, count($items), $pair_count, $minBytes);
+            // Bound against the cumulative pairs read, not count($items): a
+            // stream repeating the same key would otherwise shrink count($items)
+            // and slip past the cumulative cap.
+            self::ensureCollectionAvailable($decoder, $read, $pair_count, $minBytes);
+            $read += $pair_count;
             for ($i = 0; $i < $pair_count; $i++) {
                 $key = $decoder->readString();
                 $items[$key] = $this->readData(
@@ -647,7 +652,7 @@ private static function minBytesPerElement(mixed $schema, array $visited = []):
                 $visited[$id] = true;
                 $total = 0;
                 foreach ($schema->fields() as $field) {
-                    $total += self::minBytesPerElement($field, $visited);
+                    $total += self::minBytesPerElement($field->type(), $visited);
                 }
 
                 return $total;
@@ -660,7 +665,8 @@ private static function minBytesPerElement(mixed $schema, array $visited = []):
 
     /**
      * Returns the configured collection limits as [zeroByteLimit, structuralLimit].
-     * AVRO_MAX_COLLECTION_ITEMS, when a non-negative integer, caps both.
+     * AVRO_MAX_COLLECTION_ITEMS, when a non-negative integer, overrides both
+     * limits to that value (which may raise or lower them).
      *
      * @return array{int, int}
      */
diff --git a/lang/php/test/DatumIOTest.php b/lang/php/test/DatumIOTest.php
index 06a7f876da9..cd723e02a0a 100644
--- a/lang/php/test/DatumIOTest.php
+++ b/lang/php/test/DatumIOTest.php
@@ -358,7 +358,7 @@ public function test_array_of_null_exceeds_configured_limit(): void
         putenv('AVRO_MAX_COLLECTION_ITEMS=1000');
 
         try {
-            $this->expectException(AvroException::class);
+            $this->expectException(AvroIOCollectionSizeException::class);
             $this->decodeWith('{"type":"array","items":"null"}', self::zeroByteBlock(1001));
         } finally {
             putenv('AVRO_MAX_COLLECTION_ITEMS');
@@ -375,7 +375,7 @@ public function test_array_of_null_cumulative_across_blocks(): void
         putenv('AVRO_MAX_COLLECTION_ITEMS=1000');
 
         try {
-            $this->expectException(AvroException::class);
+            $this->expectException(AvroIOCollectionSizeException::class);
             $this->decodeWith('{"type":"array","items":"null"}', $io);
         } finally {
             putenv('AVRO_MAX_COLLECTION_ITEMS');
@@ -387,13 +387,36 @@ public function test_array_of_null_negative_block_count(): void
         putenv('AVRO_MAX_COLLECTION_ITEMS=1000');
 
         try {
-            $this->expectException(AvroException::class);
+            $this->expectException(AvroIOCollectionSizeException::class);
             $this->decodeWith('{"type":"array","items":"null"}', self::zeroByteBlock(200000, true));
         } finally {
             putenv('AVRO_MAX_COLLECTION_ITEMS');
         }
     }
 
+    public function test_map_cumulative_limit_not_bypassed_by_duplicate_keys(): void
+    {
+        // Two blocks of 600 entries all reusing the same key: count($items) stays
+        // 1, but the cumulative pairs read (1200) must still exceed the 1000 cap.
+        $io = new AvroStringIO();
+        $encoder = new AvroIOBinaryEncoder($io);
+        foreach ([600, 600] as $blockCount) {
+            $encoder->writeLong($blockCount);
+            for ($i = 0; $i < $blockCount; $i++) {
+                $encoder->writeString('k'); // 1-byte key; null value is 0 bytes
+            }
+        }
+        $encoder->writeLong(0);
+        putenv('AVRO_MAX_COLLECTION_ITEMS=1000');
+
+        try {
+            $this->expectException(AvroIOCollectionSizeException::class);
+            $this->decodeWith('{"type":"map","values":"null"}', $io);
+        } finally {
+            putenv('AVRO_MAX_COLLECTION_ITEMS');
+        }
+    }
+
     public function test_array_of_zero_length_fixed_exceeds_limit(): void
     {
         putenv('AVRO_MAX_COLLECTION_ITEMS=1000');

From 063699d58a4719db80320b04f028779ed03666bd Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Isma=C3=ABl=20Mej=C3=ADa?= <iemejia@gmail.com>
Date: Sun, 12 Jul 2026 20:32:34 +0200
Subject: [PATCH 08/16] AVRO-4298: [php] Bound negative-block skips; non-strict
 PHP_INT_MIN compare

Addresses review feedback:
 - skipArray/skipMap applied checkSkipCollectionCount only for non-negative
   blocks, so the negative (byte-sized) block form could skip an unbounded
   collection and bypass the limit; and the block byte-size was read from $this
   rather than the passed $decoder. Both paths now normalize the count, apply the
   skip cap, reject PHP_INT_MIN and a negative block size, and read the block
   size from $decoder. Adds a regression test.
 - readArray/readMap PHP_INT_MIN guards use a non-strict comparison, so on 32-bit
   builds with GMP (where readLong returns numeric strings) the minimum block
   count is still rejected consistently.

Assisted-by: GitHub Copilot:claude-opus-4.8
---
 lang/php/lib/Datum/AvroIOBinaryDecoder.php | 36 ++++++++++++++++++----
 lang/php/lib/Datum/AvroIODatumReader.php   |  4 +--
 lang/php/test/DatumIOTest.php              | 17 ++++++++++
 3 files changed, 49 insertions(+), 8 deletions(-)

diff --git a/lang/php/lib/Datum/AvroIOBinaryDecoder.php b/lang/php/lib/Datum/AvroIOBinaryDecoder.php
index 14743e8fff3..3cf6a8c76b7 100644
--- a/lang/php/lib/Datum/AvroIOBinaryDecoder.php
+++ b/lang/php/lib/Datum/AvroIOBinaryDecoder.php
@@ -277,11 +277,24 @@ public function skipArray(AvroArraySchema $writersSchema, AvroIOBinaryDecoder $d
         $skipped = 0;
         $blockCount = $decoder->readLong();
         while (0 != $blockCount) {
+            $blockSize = null;
             if ($blockCount < 0) {
-                $decoder->skip($this->readLong());
+                if (PHP_INT_MIN == $blockCount) {
+                    throw new AvroException('Invalid array block count');
+                }
+                $blockCount = -$blockCount;
+                $blockSize = $decoder->readLong();
+                if ($blockSize < 0) {
+                    throw new AvroException('Invalid negative array block size');
+                }
+            }
+            // Bound the (normalized) count on both the sized and unsized paths so
+            // a negative block count cannot bypass the skip limit.
+            AvroIODatumReader::checkSkipCollectionCount($skipped, $blockCount, $minBytes);
+            $skipped += $blockCount;
+            if (null !== $blockSize) {
+                $decoder->skip($blockSize);
             } else {
-                AvroIODatumReader::checkSkipCollectionCount($skipped, $blockCount, $minBytes);
-                $skipped += $blockCount;
                 for ($i = 0; $i < $blockCount; $i++) {
                     AvroIODatumReader::skipData($writersSchema->items(), $decoder);
                 }
@@ -297,11 +310,22 @@ public function skipMap(AvroMapSchema $writersSchema, AvroIOBinaryDecoder $decod
         $skipped = 0;
         $blockCount = $decoder->readLong();
         while (0 != $blockCount) {
+            $blockSize = null;
             if ($blockCount < 0) {
-                $decoder->skip($this->readLong());
+                if (PHP_INT_MIN == $blockCount) {
+                    throw new AvroException('Invalid map block count');
+                }
+                $blockCount = -$blockCount;
+                $blockSize = $decoder->readLong();
+                if ($blockSize < 0) {
+                    throw new AvroException('Invalid negative map block size');
+                }
+            }
+            AvroIODatumReader::checkSkipCollectionCount($skipped, $blockCount, $minBytes);
+            $skipped += $blockCount;
+            if (null !== $blockSize) {
+                $decoder->skip($blockSize);
             } else {
-                AvroIODatumReader::checkSkipCollectionCount($skipped, $blockCount, $minBytes);
-                $skipped += $blockCount;
                 for ($i = 0; $i < $blockCount; $i++) {
                     $decoder->skipString();
                     AvroIODatumReader::skipData($writersSchema->values(), $decoder);
diff --git a/lang/php/lib/Datum/AvroIODatumReader.php b/lang/php/lib/Datum/AvroIODatumReader.php
index def43e8c059..fa9492dad85 100644
--- a/lang/php/lib/Datum/AvroIODatumReader.php
+++ b/lang/php/lib/Datum/AvroIODatumReader.php
@@ -313,7 +313,7 @@ public function readArray(
             if ($blockCount < 0) {
                 // PHP_INT_MIN cannot be negated: -PHP_INT_MIN promotes to a
                 // float, so reject it rather than propagating a non-int count.
-                if (PHP_INT_MIN === $blockCount) {
+                if (PHP_INT_MIN == $blockCount) {
                     throw new AvroException('Invalid array block count');
                 }
                 $blockCount = -$blockCount;
@@ -349,7 +349,7 @@ public function readMap(
             if ($pair_count < 0) {
                 // PHP_INT_MIN cannot be negated: -PHP_INT_MIN promotes to a
                 // float, so reject it rather than propagating a non-int count.
-                if (PHP_INT_MIN === $pair_count) {
+                if (PHP_INT_MIN == $pair_count) {
                     throw new AvroException('Invalid map block count');
                 }
                 $pair_count = -$pair_count;
diff --git a/lang/php/test/DatumIOTest.php b/lang/php/test/DatumIOTest.php
index cd723e02a0a..c386cf4cdbd 100644
--- a/lang/php/test/DatumIOTest.php
+++ b/lang/php/test/DatumIOTest.php
@@ -516,6 +516,23 @@ public function test_skip_array_of_null_respects_limit(): void
         }
     }
 
+    public function test_skip_array_of_null_negative_block_respects_limit(): void
+    {
+        // The negative (byte-sized) block form must also be bounded when skipping,
+        // so it cannot bypass the skip limit.
+        $schema = AvroSchema::parse('{"type":"array","items":"null"}');
+        $io = self::zeroByteBlock(200000000, true);
+        $io->seek(0);
+        putenv('AVRO_MAX_COLLECTION_ITEMS=1000');
+
+        try {
+            $this->expectException(AvroIOCollectionSizeException::class);
+            AvroIODatumReader::skipData($schema, new AvroIOBinaryDecoder($io));
+        } finally {
+            putenv('AVRO_MAX_COLLECTION_ITEMS');
+        }
+    }
+
     public static function validDurationLogicalTypes(): array
     {
         return [

From 542afad2034b449e48f2e70ad9abb35640504501 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Isma=C3=ABl=20Mej=C3=ADa?= <iemejia@gmail.com>
Date: Sun, 12 Jul 2026 20:56:54 +0200
Subject: [PATCH 09/16] AVRO-4298: [php] Assert AvroIOCollectionSizeException
 in zero-byte tests

Addresses review feedback: the zero-byte collection-limit tests (zero-length
fixed, all-null record, and the invalid-env-override fallback) asserted only the
base AvroException. Tighten them to AvroIOCollectionSizeException so the dedicated
collection-size guard is what rejects the payload and the tests can't pass due to
an unrelated AvroException.

Assisted-by: GitHub Copilot:claude-opus-4.8
---
 lang/php/test/DatumIOTest.php | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/lang/php/test/DatumIOTest.php b/lang/php/test/DatumIOTest.php
index c386cf4cdbd..f4bd216473c 100644
--- a/lang/php/test/DatumIOTest.php
+++ b/lang/php/test/DatumIOTest.php
@@ -422,7 +422,7 @@ public function test_array_of_zero_length_fixed_exceeds_limit(): void
         putenv('AVRO_MAX_COLLECTION_ITEMS=1000');
 
         try {
-            $this->expectException(AvroException::class);
+            $this->expectException(AvroIOCollectionSizeException::class);
             $this->decodeWith('{"type":"array","items":{"type":"fixed","name":"empty","size":0}}', self::zeroByteBlock(2000));
         } finally {
             putenv('AVRO_MAX_COLLECTION_ITEMS');
@@ -434,7 +434,7 @@ public function test_array_of_all_null_record_exceeds_limit(): void
         putenv('AVRO_MAX_COLLECTION_ITEMS=1000');
 
         try {
-            $this->expectException(AvroException::class);
+            $this->expectException(AvroIOCollectionSizeException::class);
             $this->decodeWith(
                 '{"type":"array","items":{"type":"record","name":"R","fields":[{"name":"n","type":"null"}]}}',
                 self::zeroByteBlock(2000)
@@ -459,7 +459,7 @@ public function test_invalid_env_override_falls_back_to_default(): void
         putenv('AVRO_MAX_COLLECTION_ITEMS=not-a-number');
 
         try {
-            $this->expectException(AvroException::class);
+            $this->expectException(AvroIOCollectionSizeException::class);
             $this->decodeWith('{"type":"array","items":"null"}', self::zeroByteBlock(200000000));
         } finally {
             putenv('AVRO_MAX_COLLECTION_ITEMS');

From 2db3fd1231f0ed0af2db3d0242c79b2c248dc62d Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Isma=C3=ABl=20Mej=C3=ADa?= <iemejia@gmail.com>
Date: Sun, 12 Jul 2026 21:31:21 +0200
Subject: [PATCH 10/16] AVRO-4298: [php] Reject negative block size on negative
 array/map blocks

Addresses review feedback: when decoding a negative array or map block count,
the subsequent block-size long was read and ignored without validation. Avro
binary encoding specifies block size as a byte count, so a negative value is
malformed. Reject it early to match skipArray/skipMap and prevent malformed
input from proceeding further than necessary.

Assisted-by: GitHub Copilot:claude-opus-4.8
---
 lang/php/lib/Datum/AvroIODatumReader.php | 11 ++++++++---
 1 file changed, 8 insertions(+), 3 deletions(-)

diff --git a/lang/php/lib/Datum/AvroIODatumReader.php b/lang/php/lib/Datum/AvroIODatumReader.php
index fa9492dad85..99167c553df 100644
--- a/lang/php/lib/Datum/AvroIODatumReader.php
+++ b/lang/php/lib/Datum/AvroIODatumReader.php
@@ -317,7 +317,10 @@ public function readArray(
                     throw new AvroException('Invalid array block count');
                 }
                 $blockCount = -$blockCount;
-                $decoder->readLong(); // Read (and ignore) block size
+                $blockSize = $decoder->readLong(); // block byte-size
+                if ($blockSize < 0) {
+                    throw new AvroException('Invalid negative array block size');
+                }
             }
             self::ensureCollectionAvailable($decoder, count($items), $blockCount, $minBytes);
             for ($i = 0; $i < $blockCount; $i++) {
@@ -353,8 +356,10 @@ public function readMap(
                     throw new AvroException('Invalid map block count');
                 }
                 $pair_count = -$pair_count;
-                // Note: we're not doing anything with block_size other than skipping it
-                $decoder->readLong();
+                $blockSize = $decoder->readLong(); // block byte-size
+                if ($blockSize < 0) {
+                    throw new AvroException('Invalid negative map block size');
+                }
             }
 
             // Map keys are strings (>= 1 byte length prefix) plus the value.

From 0d02b5b4f4bc6d261be89fcbc69f166f95180379 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Isma=C3=ABl=20Mej=C3=ADa?= <iemejia@gmail.com>
Date: Sun, 12 Jul 2026 21:47:59 +0200
Subject: [PATCH 11/16] AVRO-4298: [php] Restore original
 AVRO_MAX_COLLECTION_ITEMS in tests

Addresses review feedback: the collection-limit tests unconditionally unset
AVRO_MAX_COLLECTION_ITEMS in their finally blocks, breaking isolation when the
variable is already set in the environment (CI or a developer shell). Capture
the pre-existing value in setUp() and restore it via a helper in each finally
block instead of blindly unsetting.

Assisted-by: GitHub Copilot:claude-opus-4.8
---
 lang/php/test/DatumIOTest.php | 46 ++++++++++++++++++++++++++---------
 1 file changed, 34 insertions(+), 12 deletions(-)

diff --git a/lang/php/test/DatumIOTest.php b/lang/php/test/DatumIOTest.php
index f4bd216473c..ec489a0a3f9 100644
--- a/lang/php/test/DatumIOTest.php
+++ b/lang/php/test/DatumIOTest.php
@@ -35,6 +35,19 @@
 
 class DatumIOTest extends TestCase
 {
+    /** @var false|string Value of AVRO_MAX_COLLECTION_ITEMS captured before each test. */
+    private string|false $originalMaxCollectionItems = false;
+
+    protected function setUp(): void
+    {
+        parent::setUp();
+        // Capture any pre-existing value so tests that override
+        // AVRO_MAX_COLLECTION_ITEMS restore it rather than unconditionally
+        // unsetting it, which would break isolation when the variable is already
+        // set in the environment (CI or a developer shell).
+        $this->originalMaxCollectionItems = getenv('AVRO_MAX_COLLECTION_ITEMS');
+    }
+
     #[DataProvider('data_provider')]
     public function test_datum_round_trip(string $schema_json, mixed $datum, string $binary): void
     {
@@ -349,7 +362,7 @@ public function test_array_of_null_within_configured_limit_still_reads(): void
             $result = $this->decodeWith('{"type":"array","items":"null"}', self::zeroByteBlock(1000));
             $this->assertCount(1000, $result);
         } finally {
-            putenv('AVRO_MAX_COLLECTION_ITEMS');
+            $this->restoreMaxCollectionItems();
         }
     }
 
@@ -361,7 +374,7 @@ public function test_array_of_null_exceeds_configured_limit(): void
             $this->expectException(AvroIOCollectionSizeException::class);
             $this->decodeWith('{"type":"array","items":"null"}', self::zeroByteBlock(1001));
         } finally {
-            putenv('AVRO_MAX_COLLECTION_ITEMS');
+            $this->restoreMaxCollectionItems();
         }
     }
 
@@ -378,7 +391,7 @@ public function test_array_of_null_cumulative_across_blocks(): void
             $this->expectException(AvroIOCollectionSizeException::class);
             $this->decodeWith('{"type":"array","items":"null"}', $io);
         } finally {
-            putenv('AVRO_MAX_COLLECTION_ITEMS');
+            $this->restoreMaxCollectionItems();
         }
     }
 
@@ -390,7 +403,7 @@ public function test_array_of_null_negative_block_count(): void
             $this->expectException(AvroIOCollectionSizeException::class);
             $this->decodeWith('{"type":"array","items":"null"}', self::zeroByteBlock(200000, true));
         } finally {
-            putenv('AVRO_MAX_COLLECTION_ITEMS');
+            $this->restoreMaxCollectionItems();
         }
     }
 
@@ -413,7 +426,7 @@ public function test_map_cumulative_limit_not_bypassed_by_duplicate_keys(): void
             $this->expectException(AvroIOCollectionSizeException::class);
             $this->decodeWith('{"type":"map","values":"null"}', $io);
         } finally {
-            putenv('AVRO_MAX_COLLECTION_ITEMS');
+            $this->restoreMaxCollectionItems();
         }
     }
 
@@ -425,7 +438,7 @@ public function test_array_of_zero_length_fixed_exceeds_limit(): void
             $this->expectException(AvroIOCollectionSizeException::class);
             $this->decodeWith('{"type":"array","items":{"type":"fixed","name":"empty","size":0}}', self::zeroByteBlock(2000));
         } finally {
-            putenv('AVRO_MAX_COLLECTION_ITEMS');
+            $this->restoreMaxCollectionItems();
         }
     }
 
@@ -440,7 +453,7 @@ public function test_array_of_all_null_record_exceeds_limit(): void
                 self::zeroByteBlock(2000)
             );
         } finally {
-            putenv('AVRO_MAX_COLLECTION_ITEMS');
+            $this->restoreMaxCollectionItems();
         }
     }
 
@@ -462,7 +475,7 @@ public function test_invalid_env_override_falls_back_to_default(): void
             $this->expectException(AvroIOCollectionSizeException::class);
             $this->decodeWith('{"type":"array","items":"null"}', self::zeroByteBlock(200000000));
         } finally {
-            putenv('AVRO_MAX_COLLECTION_ITEMS');
+            $this->restoreMaxCollectionItems();
         }
     }
 
@@ -475,7 +488,7 @@ public function test_env_override_raises_limit(): void
             $result = $this->decodeWith('{"type":"array","items":"null"}', self::zeroByteBlock(15000));
             $this->assertCount(15000, $result);
         } finally {
-            putenv('AVRO_MAX_COLLECTION_ITEMS');
+            $this->restoreMaxCollectionItems();
         }
     }
 
@@ -496,7 +509,7 @@ public function test_non_zero_collection_bounded_by_structural_cap(): void
             $this->expectException(AvroIOCollectionSizeException::class);
             $this->decodeWith('{"type":"array","items":"long"}', $io);
         } finally {
-            putenv('AVRO_MAX_COLLECTION_ITEMS');
+            $this->restoreMaxCollectionItems();
         }
     }
 
@@ -512,7 +525,7 @@ public function test_skip_array_of_null_respects_limit(): void
             $this->expectException(AvroIOCollectionSizeException::class);
             AvroIODatumReader::skipData($schema, new AvroIOBinaryDecoder($io));
         } finally {
-            putenv('AVRO_MAX_COLLECTION_ITEMS');
+            $this->restoreMaxCollectionItems();
         }
     }
 
@@ -529,7 +542,7 @@ public function test_skip_array_of_null_negative_block_respects_limit(): void
             $this->expectException(AvroIOCollectionSizeException::class);
             AvroIODatumReader::skipData($schema, new AvroIOBinaryDecoder($io));
         } finally {
-            putenv('AVRO_MAX_COLLECTION_ITEMS');
+            $this->restoreMaxCollectionItems();
         }
     }
 
@@ -692,6 +705,15 @@ public function test_field_default_value(
         }
     }
 
+    protected function restoreMaxCollectionItems(): void
+    {
+        if (false === $this->originalMaxCollectionItems) {
+            putenv('AVRO_MAX_COLLECTION_ITEMS');
+        } else {
+            putenv('AVRO_MAX_COLLECTION_ITEMS='.$this->originalMaxCollectionItems);
+        }
+    }
+
     /**
      * Builds one array/map block of `$count` zero-byte elements plus an end
      * marker (a negative count is preceded by a block byte-size).

From 8d2b1cbbe6ce52688cfb022de472062f04beea11 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Isma=C3=ABl=20Mej=C3=ADa?= <iemejia@gmail.com>
Date: Sun, 12 Jul 2026 23:20:30 +0200
Subject: [PATCH 12/16] AVRO-4298: [php] Reject negative union branch index

schemaByIndex validated only the upper bound (count > index), so a negative
index passed the check and then indexed $schemas[-1] => null, violating the
AvroSchema return type with a TypeError instead of a clean AvroException. Add a
lower-bound check (index >= 0) so a negative branch index is rejected with
AvroSchemaParseException. Adds tests for negative and too-large union indices.
(readEnum is already guarded via array_key_exists.)

Assisted-by: GitHub Copilot:claude-opus-4.8
---
 lang/php/lib/Schema/AvroUnionSchema.php |  2 +-
 lang/php/test/DatumIOTest.php           | 18 ++++++++++++++++++
 2 files changed, 19 insertions(+), 1 deletion(-)

diff --git a/lang/php/lib/Schema/AvroUnionSchema.php b/lang/php/lib/Schema/AvroUnionSchema.php
index c6b6cbf279e..dcf80640a85 100644
--- a/lang/php/lib/Schema/AvroUnionSchema.php
+++ b/lang/php/lib/Schema/AvroUnionSchema.php
@@ -98,7 +98,7 @@ public function schemas(): array
      */
     public function schemaByIndex($index): AvroSchema
     {
-        if (count($this->schemas) > $index) {
+        if ($index >= 0 && count($this->schemas) > $index) {
             return $this->schemas[$index];
         }
 
diff --git a/lang/php/test/DatumIOTest.php b/lang/php/test/DatumIOTest.php
index ec489a0a3f9..6398dc77d18 100644
--- a/lang/php/test/DatumIOTest.php
+++ b/lang/php/test/DatumIOTest.php
@@ -316,6 +316,24 @@ public function test_read_map_rejects_count_beyond_stream(): void
         $this->decodeWith('{"type":"map","values":"long"}', $io);
     }
 
+    public function test_read_union_rejects_out_of_range_index(): void
+    {
+        // Index 5 (zig-zag long) is out of range for a 2-branch union.
+        $io = new AvroStringIO();
+        (new AvroIOBinaryEncoder($io))->writeLong(5);
+        $this->expectException(AvroException::class);
+        $this->decodeWith('["null","long"]', $io);
+    }
+
+    public function test_read_union_rejects_negative_index(): void
+    {
+        // A negative index must not wrap to a valid branch.
+        $io = new AvroStringIO();
+        (new AvroIOBinaryEncoder($io))->writeLong(-1);
+        $this->expectException(AvroException::class);
+        $this->decodeWith('["null","long"]', $io);
+    }
+
     public function test_read_array_of_null_not_falsely_rejected(): void
     {
         $count = 100000;

From f6ffd01e5a05f519009c935dbc0f0b1eaf71f6ec Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Isma=C3=ABl=20Mej=C3=ADa?= <iemejia@gmail.com>
Date: Sun, 12 Jul 2026 23:51:20 +0200
Subject: [PATCH 13/16] AVRO-4298: [php] Reject overlong varints in readLong

readLong read the continuation chain into an array with no cap, so an overlong
varint fed an unbounded byte array into the native and GMP decoders and silently
corrupted the value. A 64-bit value uses at most 10 bytes; reject an 11th
continuation byte with AvroException in the read loop (bounding both decode
paths), matching the Java, C and C++ SDKs. Adds a regression test.

Assisted-by: GitHub Copilot:claude-opus-4.8
---
 lang/php/lib/Datum/AvroIOBinaryDecoder.php |  6 ++++++
 lang/php/test/DatumIOTest.php              | 11 +++++++++++
 2 files changed, 17 insertions(+)

diff --git a/lang/php/lib/Datum/AvroIOBinaryDecoder.php b/lang/php/lib/Datum/AvroIOBinaryDecoder.php
index 3cf6a8c76b7..c10f2b909c7 100644
--- a/lang/php/lib/Datum/AvroIOBinaryDecoder.php
+++ b/lang/php/lib/Datum/AvroIOBinaryDecoder.php
@@ -116,6 +116,12 @@ public function readLong(): string|int
         $byte = ord($this->nextByte());
         $bytes = [$byte];
         while (0 != ($byte & 0x80)) {
+            // A 64-bit value uses at most 10 bytes; reject an overlong varint
+            // rather than reading an unbounded continuation chain and silently
+            // corrupting the value. Bounds both the native and GMP decode paths.
+            if (count($bytes) >= 10) {
+                throw new AvroException('Varint is too long');
+            }
             $byte = ord($this->nextByte());
             $bytes[] = $byte;
         }
diff --git a/lang/php/test/DatumIOTest.php b/lang/php/test/DatumIOTest.php
index 6398dc77d18..8a840c00adb 100644
--- a/lang/php/test/DatumIOTest.php
+++ b/lang/php/test/DatumIOTest.php
@@ -334,6 +334,17 @@ public function test_read_union_rejects_negative_index(): void
         $this->decodeWith('["null","long"]', $io);
     }
 
+    public function test_read_long_rejects_overlong_varint(): void
+    {
+        // A 64-bit value uses at most 10 bytes; an 11th continuation byte is a
+        // malformed (overlong) varint and must be rejected.
+        $io = new AvroStringIO(str_repeat("\x80", 10)."\x01");
+        $io->seek(0);
+        $decoder = new AvroIOBinaryDecoder($io);
+        $this->expectException(AvroException::class);
+        $decoder->readLong();
+    }
+
     public function test_read_array_of_null_not_falsely_rejected(): void
     {
         $count = 100000;

From f4481123fe7b757b1e016bd26ee22acf469c3e8c Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Isma=C3=ABl=20Mej=C3=ADa?= <iemejia@gmail.com>
Date: Mon, 13 Jul 2026 02:23:52 +0200
Subject: [PATCH 14/16] AVRO-4298: [php] Validate byte-sized skip blocks
 against bytes remaining

skipArray/skipMap skipped a negative-block's declared byte-size via seek()
without checking it. AvroFile/AvroStringIO seek() can move past EOF, so a
truncated or oversized block was silently skipped, leaving the decoder positioned
beyond the input and hiding truncation during projection. Reject a block size
larger than bytesRemaining() before skipping.

Assisted-by: GitHub Copilot:claude-opus-4.8
---
 lang/php/lib/Datum/AvroIOBinaryDecoder.php | 11 +++++++++++
 1 file changed, 11 insertions(+)

diff --git a/lang/php/lib/Datum/AvroIOBinaryDecoder.php b/lang/php/lib/Datum/AvroIOBinaryDecoder.php
index c10f2b909c7..74c7ec408dc 100644
--- a/lang/php/lib/Datum/AvroIOBinaryDecoder.php
+++ b/lang/php/lib/Datum/AvroIOBinaryDecoder.php
@@ -299,6 +299,12 @@ public function skipArray(AvroArraySchema $writersSchema, AvroIOBinaryDecoder $d
             AvroIODatumReader::checkSkipCollectionCount($skipped, $blockCount, $minBytes);
             $skipped += $blockCount;
             if (null !== $blockSize) {
+                // seek() can move past EOF, so a truncated/oversized block would
+                // otherwise be "skipped" silently, hiding truncation. Reject a
+                // block size larger than the bytes actually remaining.
+                if ($blockSize > $decoder->bytesRemaining()) {
+                    throw new AvroException('Array block size exceeds the remaining input');
+                }
                 $decoder->skip($blockSize);
             } else {
                 for ($i = 0; $i < $blockCount; $i++) {
@@ -330,6 +336,11 @@ public function skipMap(AvroMapSchema $writersSchema, AvroIOBinaryDecoder $decod
             AvroIODatumReader::checkSkipCollectionCount($skipped, $blockCount, $minBytes);
             $skipped += $blockCount;
             if (null !== $blockSize) {
+                // seek() can move past EOF; reject a block size larger than the
+                // bytes remaining so a truncated block isn't silently skipped.
+                if ($blockSize > $decoder->bytesRemaining()) {
+                    throw new AvroException('Map block size exceeds the remaining input');
+                }
                 $decoder->skip($blockSize);
             } else {
                 for ($i = 0; $i < $blockCount; $i++) {

From 33c9885342ca64f145f789e619224b41aa7f2a9a Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Isma=C3=ABl=20Mej=C3=ADa?= <iemejia@gmail.com>
Date: Mon, 13 Jul 2026 03:24:49 +0200
Subject: [PATCH 15/16] AVRO-4298: [php] Clamp fixed-schema minimum size to
 non-negative

minBytesPerElement returned AvroFixedSchema::size() verbatim. Since schema
parsing does not reject a negative "size", a malformed fixed schema could make
this negative, causing ensureCollectionAvailable to treat the element as
zero-byte (minBytesPerElement > 0 false) and skip the bytes-remaining check.
Clamp to max(0, size()).

Assisted-by: GitHub Copilot:claude-opus-4.8
---
 lang/php/lib/Datum/AvroIODatumReader.php | 6 +++++-
 1 file changed, 5 insertions(+), 1 deletion(-)

diff --git a/lang/php/lib/Datum/AvroIODatumReader.php b/lang/php/lib/Datum/AvroIODatumReader.php
index 99167c553df..78d087c1665 100644
--- a/lang/php/lib/Datum/AvroIODatumReader.php
+++ b/lang/php/lib/Datum/AvroIODatumReader.php
@@ -644,7 +644,11 @@ private static function minBytesPerElement(mixed $schema, array $visited = []):
             case AvroSchema::DOUBLE_TYPE:
                 return 8;
             case AvroSchema::FIXED_SCHEMA:
-                return $schema instanceof AvroFixedSchema ? $schema->size() : 1;
+                // Clamp to >= 0: a malformed fixed schema could have a negative
+                // size (AvroSchema::parse does not reject it), which would make
+                // this negative and cause ensureCollectionAvailable to treat the
+                // element as zero-byte.
+                return $schema instanceof AvroFixedSchema ? max(0, $schema->size()) : 1;
             case AvroSchema::RECORD_SCHEMA:
             case AvroSchema::ERROR_SCHEMA:
                 if (!$schema instanceof AvroRecordSchema) {

From c5a8634b13292ef318bb771a5cbafe51fbdcc6b6 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Isma=C3=ABl=20Mej=C3=ADa?= <iemejia@gmail.com>
Date: Mon, 13 Jul 2026 06:27:51 +0200
Subject: [PATCH 16/16] AVRO-4298: [php] Verify sized skip block can hold the
 declared element count

skipArray/skipMap validated the block byte-size against bytesRemaining but not
against the declared element count. Reject a block size too small to contain
blockCount elements at their minimum on-wire size (when minBytes > 0), so a
malformed size can't leave the decoder mid-element and misalign later decoding.

Assisted-by: GitHub Copilot:claude-opus-4.8
---
 lang/php/lib/Datum/AvroIOBinaryDecoder.php | 6 ++++++
 1 file changed, 6 insertions(+)

diff --git a/lang/php/lib/Datum/AvroIOBinaryDecoder.php b/lang/php/lib/Datum/AvroIOBinaryDecoder.php
index 74c7ec408dc..da7888d96ca 100644
--- a/lang/php/lib/Datum/AvroIOBinaryDecoder.php
+++ b/lang/php/lib/Datum/AvroIOBinaryDecoder.php
@@ -305,6 +305,9 @@ public function skipArray(AvroArraySchema $writersSchema, AvroIOBinaryDecoder $d
                 if ($blockSize > $decoder->bytesRemaining()) {
                     throw new AvroException('Array block size exceeds the remaining input');
                 }
+                if ($minBytes > 0 && $blockCount > intdiv($blockSize, $minBytes)) {
+                    throw new AvroException('Array block size too small for the declared element count');
+                }
                 $decoder->skip($blockSize);
             } else {
                 for ($i = 0; $i < $blockCount; $i++) {
@@ -341,6 +344,9 @@ public function skipMap(AvroMapSchema $writersSchema, AvroIOBinaryDecoder $decod
                 if ($blockSize > $decoder->bytesRemaining()) {
                     throw new AvroException('Map block size exceeds the remaining input');
                 }
+                if ($minBytes > 0 && $blockCount > intdiv($blockSize, $minBytes)) {
+                    throw new AvroException('Map block size too small for the declared element count');
+                }
                 $decoder->skip($blockSize);
             } else {
                 for ($i = 0; $i < $blockCount; $i++) {