Fix memory leak when user filter does not fully consume the input brigade buckets

ext/standard/tests/filters/user_filter_feed_me_keeps_input.phpt

@@ -0,0 +1,32 @@
+--TEST--
+User filter returning PSFS_FEED_ME keeps buckets it put back on the input brigade
+--FILE--
+<?php
+class defer extends php_user_filter {
+    private int $calls = 0;
+    public function filter($in, $out, &$consumed, bool $closing): int {
+        $this->calls++;
+        $buckets = [];
+        while ($b = stream_bucket_make_writeable($in)) { $buckets[] = $b; }
+        if ($this->calls < 3) {
+            foreach ($buckets as $b) { stream_bucket_prepend($in, $b); }
+            return PSFS_FEED_ME;
+        }
+        foreach ($buckets as $b) { stream_bucket_append($out, $b); }
+        return PSFS_PASS_ON;
+    }
+}
+stream_filter_register("defer", "defer");
+$f = __DIR__ . "/user_filter_feed_me_keeps_input.bin";
+file_put_contents($f, str_repeat("a", 8192) . str_repeat("b", 8192) . str_repeat("c", 8192));
+$fp = fopen($f, 'r');
+stream_filter_append($fp, "defer", STREAM_FILTER_READ);
+var_dump(strlen(stream_get_contents($fp)));
+fclose($fp);
+?>
+--CLEAN--
+<?php
+@unlink(__DIR__ . "/user_filter_feed_me_keeps_input.bin");
+?>
+--EXPECT--
+int(24576)

ext/standard/tests/filters/user_filter_no_consume_leak.phpt

@@ -0,0 +1,22 @@
+--TEST--
+User filter that does not consume any input bucket leaks
+--FILE--
+<?php
+
+class noop_filter extends php_user_filter {
+	public function filter($in, $out, &$consumed, bool $closing): int {
+		return PSFS_PASS_ON;
+	}
+}
+
+stream_filter_register("noop_filter", "noop_filter");
+
+stream_filter_append(STDOUT, "noop_filter");
+
+$out = fwrite(STDOUT, "Hello\n");
+var_dump($out);
+
+?>
+--EXPECTF--
+Warning: fwrite(): Unprocessed filter buckets remaining on input brigade in %s on line %d
+int(0)

ext/standard/user_filters.c

@@ -166,6 +166,12 @@ static php_stream_filter_status_t userfilter_filter(
 		if (EG(exception)) {
 			EG(fake_scope) = old_scope;
 			if (buckets_in->head) {
+				php_stream_bucket *bucket;
+				do {
+					bucket = buckets_in->head;
+					php_stream_bucket_unlink(bucket);
+					php_stream_bucket_delref(bucket);
+				} while (buckets_in->head);
 				php_error_docref(NULL, E_WARNING, "Unprocessed filter buckets remaining on input brigade");
 			}
 			zend_string_release(stream_name);
@@ -211,10 +217,30 @@ static php_stream_filter_status_t userfilter_filter(
 		*bytes_consumed = zval_get_long(&args[2]);
 	}
 
-	if (buckets_in->head) {
+	/* A filter that returns PSFS_FEED_ME may put buckets back on the input
+	 * brigade to see them again with the next chunk; the read path keeps
+	 * that brigade across calls. For any other status the input brigade
+	 * is discarded by the caller, so release what the filter left behind. */
+	if (ret != PSFS_FEED_ME && buckets_in->head) {
+		php_stream_bucket *bucket;
+		do {
+			bucket = buckets_in->head;
+			php_stream_bucket_unlink(bucket);
+			php_stream_bucket_delref(bucket);
+		} while (buckets_in->head);
 		php_error_docref(NULL, E_WARNING, "Unprocessed filter buckets remaining on input brigade");
 	}
 
+	/* Filter could've broken contract and added buckets anyway. */
+	if (ret != PSFS_PASS_ON && buckets_out->head) {
+		php_stream_bucket *bucket;
+		do {
+			bucket = buckets_out->head;
+			php_stream_bucket_unlink(bucket);
+			php_stream_bucket_delref(bucket);
+		} while (buckets_out->head);
+	}
+
 	/* filter resources are cleaned up by the stream destructor,
 	 * keeping a reference to the stream resource here would prevent it
 	 * from being destroyed properly.

main/streams/filter.c

@@ -355,10 +355,11 @@ PHPAPI int php_stream_filter_append_ex(php_stream_filter_chain *chain, php_strea
 				   Reset stream's internal read buffer since the filter is "holding" it. */
 				stream->readpos = 0;
 				stream->writepos = 0;
-
-				/* Filter could have added buckets anyway, but signalled that it did not return any. Discard them. */
-				while (brig_out.head) {
-					bucket = brig_out.head;
+				while ((bucket = brig_in.head)) {
+					php_stream_bucket_unlink(bucket);
+					php_stream_bucket_delref(bucket);
+				}
+				while ((bucket = brig_out.head)) {
 					php_stream_bucket_unlink(bucket);
 					php_stream_bucket_delref(bucket);
 				}
@@ -427,6 +428,14 @@ PHPAPI int _php_stream_filter_flush(php_stream_filter *filter, int finish)
 		status = current->fops->filter(stream, current, inp, outp, NULL, flags);
 		if (status == PSFS_FEED_ME) {
 			/* We've flushed the data far enough */
+			while ((bucket = (*inp).head)) {
+				php_stream_bucket_unlink(bucket);
+				php_stream_bucket_delref(bucket);
+			}
+			while ((bucket = (*outp).head)) {
+				php_stream_bucket_unlink(bucket);
+				php_stream_bucket_delref(bucket);
+			}
 			return SUCCESS;
 		}
 		if (status == PSFS_ERR_FATAL) {

main/streams/streams.c

@@ -586,6 +586,14 @@ PHPAPI zend_result _php_stream_fill_read_buffer(php_stream *stream, size_t size)
 			/* read a chunk into a bucket */
 			justread = stream->ops->read(stream, chunk_buf, stream->chunk_size);
 			if (justread < 0 && stream->writepos == stream->readpos) {
+				while ((bucket = brig_in.head)) {
+					php_stream_bucket_unlink(bucket);
+					php_stream_bucket_delref(bucket);
+				}
+				while ((bucket = brig_out.head)) {
+					php_stream_bucket_unlink(bucket);
+					php_stream_bucket_delref(bucket);
+				}
 				efree(chunk_buf);
 				retval = FAILURE;
 				goto out_check_eof;
@@ -652,12 +660,6 @@ PHPAPI zend_result _php_stream_fill_read_buffer(php_stream *stream, size_t size)
 					/* when a filter needs feeding, there is no brig_out to deal with.
 					 * we simply continue the loop; if the caller needs more data,
 					 * we will read again, otherwise out job is done here */
-
-					/* Filter could have added buckets anyway, but signalled that it did not return any. Discard them. */
-					while ((bucket = brig_outp->head)) {
-						php_stream_bucket_unlink(bucket);
-						php_stream_bucket_delref(bucket);
-					}
 					break;
 
 				case PSFS_ERR_FATAL:
@@ -686,6 +688,16 @@ PHPAPI zend_result _php_stream_fill_read_buffer(php_stream *stream, size_t size)
 			}
 		}
 
+		php_stream_bucket *bucket;
+		while ((bucket = brig_in.head)) {
+			php_stream_bucket_unlink(bucket);
+			php_stream_bucket_delref(bucket);
+		}
+		while ((bucket = brig_out.head)) {
+			php_stream_bucket_unlink(bucket);
+			php_stream_bucket_delref(bucket);
+		}
+
 		efree(chunk_buf);
 		return SUCCESS;
 	} else {
@@ -1296,13 +1308,15 @@ static ssize_t _php_stream_write_filtered(php_stream *stream, const char *buf, s
 			/* some fatal error.  Theoretically, the stream is borked, so all
 			 * further writes should fail. */
 			consumed = (ssize_t) -1;
-			ZEND_FALLTHROUGH;
+			break;
 
 		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;
+			while ((bucket = brig_in.head)) {
+				php_stream_bucket_unlink(bucket);
+				php_stream_bucket_delref(bucket);
+			}
+			while ((bucket = brig_out.head)) {
 				php_stream_bucket_unlink(bucket);
 				php_stream_bucket_delref(bucket);
 			}