Fix leak of stream buckets left behind by misbehaving user filters

NEWS

@@ -141,6 +141,9 @@ PHP                                                                        NEWS
 - Streams:
   . Fixed bug GH-24508 (HTTP wrapper leaks the dechunk filter when header
     parsing fails after Transfer-Encoding). (bercedev)
+  . Fixed memory leak of stream buckets left behind by user filters that do
+    not consume their input brigade or return PSFS_FEED_ME with output.
+    (zeynallow)
 
 - Tidy:
   . Fixed tidy_repair_string(), tidy_repair_file(), tidy::repairString() and

ext/standard/tests/filters/user_filter_unconsumed_buckets_leak.phpt

@@ -0,0 +1,55 @@
+--TEST--
+User filters leaving buckets behind on the input or output brigade must not leak them
+--FILE--
+<?php
+
+class pass_on_without_consuming extends php_user_filter {
+	public function filter($in, $out, &$consumed, bool $closing): int {
+		return PSFS_PASS_ON;
+	}
+}
+
+class feed_me_with_output extends php_user_filter {
+	public function filter($in, $out, &$consumed, bool $closing): int {
+		while ($bucket = stream_bucket_make_writeable($in)) {
+			stream_bucket_append($out, $bucket);
+		}
+		return PSFS_FEED_ME;
+	}
+}
+
+stream_filter_register("pass_on_without_consuming", "pass_on_without_consuming");
+stream_filter_register("feed_me_with_output", "feed_me_with_output");
+
+echo "Read filter, PSFS_PASS_ON without consuming input:\n";
+$fp = fopen("php://memory", "w+");
+fwrite($fp, "Hello");
+rewind($fp);
+stream_filter_append($fp, "pass_on_without_consuming", STREAM_FILTER_READ);
+var_dump(fread($fp, 10));
+fclose($fp);
+
+echo "Write filter, PSFS_PASS_ON without consuming input:\n";
+$fp = fopen("php://memory", "w+");
+stream_filter_append($fp, "pass_on_without_consuming", STREAM_FILTER_WRITE);
+var_dump(fwrite($fp, "Hello"));
+fclose($fp);
+
+echo "Write filter, PSFS_FEED_ME with buckets on output:\n";
+$fp = fopen("php://memory", "w+");
+stream_filter_append($fp, "feed_me_with_output", STREAM_FILTER_WRITE);
+var_dump(fwrite($fp, "Hello"));
+fclose($fp);
+
+?>
+--EXPECTF--
+Read filter, PSFS_PASS_ON without consuming input:
+
+Warning: fread(): Unprocessed filter buckets remaining on input brigade in %s on line %d
+string(0) ""
+Write filter, PSFS_PASS_ON without consuming input:
+
+Warning: fwrite(): Unprocessed filter buckets remaining on input brigade in %s on line %d
+int(0)
+Write filter, PSFS_FEED_ME with buckets on output:
+int(0)

main/streams/streams.c

@@ -557,6 +557,16 @@ fprintf(stderr, "stream_free: %s:%p[%s] preserve_handle=%d release_cast=%d remov
 
 /* {{{ generic stream operations */
 
+static void php_stream_bucket_brigade_discard(php_stream_bucket_brigade *brigade)
+{
+	php_stream_bucket *bucket;
+
+	while ((bucket = brigade->head)) {
+		php_stream_bucket_unlink(bucket);
+		php_stream_bucket_delref(bucket);
+	}
+}
+
 PHPAPI zend_result _php_stream_fill_read_buffer(php_stream *stream, size_t size)
 {
 	/* allocate/fill the buffer */
@@ -609,8 +619,10 @@ PHPAPI zend_result _php_stream_fill_read_buffer(php_stream *stream, size_t size)
 				}
 
 				/* brig_out becomes brig_in.
-				 * brig_in will always be empty here, as the filter MUST attach any un-consumed buckets
-				 * to its own brigade */
+				 * brig_in should be empty here, as the filter MUST attach any un-consumed buckets
+				 * to its own brigade. A misbehaving filter may leave buckets behind; discard them
+				 * so they are not leaked when the brigade is reset below. */
+				php_stream_bucket_brigade_discard(brig_inp);
 				brig_swap = brig_inp;
 				brig_inp = brig_outp;
 				brig_outp = brig_swap;
@@ -1289,8 +1301,10 @@ static ssize_t _php_stream_write_filtered(php_stream *stream, const char *buf, s
 			break;
 		}
 		/* brig_out becomes brig_in.
-		 * brig_in will always be empty here, as the filter MUST attach any un-consumed buckets
-		 * to its own brigade */
+		 * brig_in should be empty here, as the filter MUST attach any un-consumed buckets
+		 * to its own brigade. A misbehaving filter may leave buckets behind; discard them
+		 * so they are not leaked when the brigade is reset below. */
+		php_stream_bucket_brigade_discard(brig_inp);
 		brig_swap = brig_inp;
 		brig_inp = brig_outp;
 		brig_outp = brig_swap;
@@ -1325,12 +1339,10 @@ static ssize_t _php_stream_write_filtered(php_stream *stream, const char *buf, s
 
 		case PSFS_FEED_ME:
 			/* need more data before we can push data through to the stream */
-			/* Filter could have added buckets anyway, but signalled that it did not return any. Discard them. */
-			while (brig_inp->head) {
-				bucket = brig_inp->head;
-				php_stream_bucket_unlink(bucket);
-				php_stream_bucket_delref(bucket);
-			}
+			/* Filter could have added buckets anyway, but signalled that it did not return any. Discard them,
+			 * together with any input buckets it did not consume. */
+			php_stream_bucket_brigade_discard(brig_inp);
+			php_stream_bucket_brigade_discard(brig_outp);
 			break;
 	}