Fix leaks and data loss when user filter leaves buckets on input brigade

NEWS

@@ -215,6 +215,10 @@ PHP                                                                        NEWS
     lookup, and an unchecked CreateFileA() failure. (Ilia Alshanetsky)
   . Fix persistent stream context lifetime during shutdown (Levi Morrison)
 
+- Streams:
+  . Fixed leaks and data loss when user filter leaves buckets on input
+    brigade. (ndossche, Matthias Goergens, Jakub Zelenka)
+
 - Tidy:
   . Fixed a use-after-free when a tidyNode is used after its document is
     reparsed. (Ilia Alshanetsky)

ext/standard/tests/filters/user_filter_feed_me_across_reads.phpt

@@ -0,0 +1,39 @@
+--TEST--
+Input kept back by a read filter with PSFS_FEED_ME survives across reads and is released on close
+--SKIPIF--
+<?php
+if (!function_exists('stream_socket_pair') || !defined('STREAM_PF_UNIX') || PHP_OS_FAMILY === 'Windows') die('skip no unix socket pair');
+?>
+--FILE--
+<?php
+class defer extends php_user_filter {
+    public function filter($in, $out, &$consumed, bool $closing): int {
+        $data = '';
+        while ($b = stream_bucket_make_writeable($in)) { $data .= $b->data; }
+        if (!$closing && strlen($data) < 8) {
+            if ($data !== '') {
+                stream_bucket_prepend($in, stream_bucket_new($this->stream, $data));
+            }
+            return PSFS_FEED_ME;
+        }
+        stream_bucket_append($out, stream_bucket_new($this->stream, "[" . $data . "]"));
+        return PSFS_PASS_ON;
+    }
+}
+stream_filter_register("defer", "defer");
+[$r, $w] = stream_socket_pair(STREAM_PF_UNIX, STREAM_SOCK_STREAM, 0);
+stream_set_blocking($r, false);
+stream_filter_append($r, "defer", STREAM_FILTER_READ);
+fwrite($w, "abc");
+var_dump(fread($r, 100));
+fwrite($w, "defgh");
+var_dump(fread($r, 100));
+fwrite($w, "xy");
+var_dump(fread($r, 100));
+fclose($r);
+fclose($w);
+?>
+--EXPECT--
+string(0) ""
+string(10) "[abcdefgh]"
+string(0) ""

ext/standard/tests/filters/user_filter_feed_me_chained.phpt

@@ -0,0 +1,34 @@
+--TEST--
+Input kept by a PSFS_FEED_ME filter is re-presented to that filter only, not to the filters before it
+--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 (array_reverse($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_chained.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, "string.rot13", STREAM_FILTER_READ);
+stream_filter_append($fp, "defer", STREAM_FILTER_READ);
+$data = stream_get_contents($fp);
+fclose($fp);
+var_dump($data === str_repeat("n", 8192) . str_repeat("o", 8192) . str_repeat("p", 8192));
+?>
+--CLEAN--
+<?php
+@unlink(__DIR__ . "/user_filter_feed_me_chained.bin");
+?>
+--EXPECT--
+bool(true)

ext/standard/tests/filters/user_filter_feed_me_closing_warning.phpt

@@ -0,0 +1,29 @@
+--TEST--
+Input kept back by a PSFS_FEED_ME filter on the closing call is discarded with a warning
+--FILE--
+<?php
+class defer extends php_user_filter {
+    public function filter($in, $out, &$consumed, bool $closing): int {
+        $buckets = [];
+        while ($b = stream_bucket_make_writeable($in)) { $buckets[] = $b; $consumed += $b->datalen; }
+        foreach (array_reverse($buckets) as $b) { stream_bucket_prepend($in, $b); }
+        return PSFS_FEED_ME;
+    }
+}
+stream_filter_register("defer", "defer");
+$f = __DIR__ . "/user_filter_feed_me_closing_warning.bin";
+$fp = fopen($f, 'w');
+stream_filter_append($fp, "defer", STREAM_FILTER_WRITE);
+var_dump(fwrite($fp, "hello"));
+fclose($fp);
+var_dump(file_get_contents($f));
+?>
+--CLEAN--
+<?php
+@unlink(__DIR__ . "/user_filter_feed_me_closing_warning.bin");
+?>
+--EXPECTF--
+int(5)
+
+Warning: fclose(): Unprocessed filter buckets remaining on input brigade in %s on line %d
+string(0) ""

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_feed_me_read_closing.phpt

@@ -0,0 +1,31 @@
+--TEST--
+Input kept back by a read filter with PSFS_FEED_ME is re-presented on the closing call
+--FILE--
+<?php
+class defer extends php_user_filter {
+    public function filter($in, $out, &$consumed, bool $closing): int {
+        $data = '';
+        while ($b = stream_bucket_make_writeable($in)) { $data .= $b->data; }
+        if (!$closing) {
+            stream_bucket_prepend($in, stream_bucket_new($this->stream, $data));
+            return PSFS_FEED_ME;
+        }
+        stream_bucket_append($out, stream_bucket_new($this->stream, "[" . $data . "]"));
+        return PSFS_PASS_ON;
+    }
+}
+stream_filter_register("defer", "defer");
+$f = __DIR__ . "/user_filter_feed_me_read_closing.bin";
+file_put_contents($f, "abcdefghij");
+$fp = fopen($f, 'r');
+stream_set_chunk_size($fp, 4);
+stream_filter_append($fp, "defer", STREAM_FILTER_READ);
+var_dump(stream_get_contents($fp));
+fclose($fp);
+?>
+--CLEAN--
+<?php
+@unlink(__DIR__ . "/user_filter_feed_me_read_closing.bin");
+?>
+--EXPECT--
+string(12) "[abcdefghij]"

ext/standard/tests/filters/user_filter_feed_me_untouched_input.phpt

@@ -0,0 +1,35 @@
+--TEST--
+User filter returning PSFS_FEED_ME without touching the input brigade sees the original data again
+--FILE--
+<?php
+class defer extends php_user_filter {
+    private int $calls = 0;
+    public function filter($in, $out, &$consumed, bool $closing): int {
+        $this->calls++;
+        if ($this->calls < 3) {
+            return PSFS_FEED_ME;
+        }
+        while ($b = stream_bucket_make_writeable($in)) {
+            stream_bucket_append($out, $b);
+        }
+        return PSFS_PASS_ON;
+    }
+}
+stream_filter_register("defer", "defer");
+$f = __DIR__ . "/user_filter_feed_me_untouched_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);
+$data = stream_get_contents($fp);
+fclose($fp);
+var_dump(strlen($data), substr_count($data, "a"), substr_count($data, "b"), substr_count($data, "c"));
+?>
+--CLEAN--
+<?php
+@unlink(__DIR__ . "/user_filter_feed_me_untouched_input.bin");
+?>
+--EXPECT--
+int(24576)
+int(8192)
+int(8192)
+int(8192)

ext/standard/tests/filters/user_filter_feed_me_write.phpt

@@ -0,0 +1,39 @@
+--TEST--
+Write filter returning PSFS_FEED_ME sees the input it kept back again with the next write
+--FILE--
+<?php
+class defer extends php_user_filter {
+    private int $kept = 0;
+    public function filter($in, $out, &$consumed, bool $closing): int {
+        $data = '';
+        while ($b = stream_bucket_make_writeable($in)) { $data .= $b->data; }
+        /* the bytes kept back from the previous call were already reported as consumed */
+        $consumed += strlen($data) - $this->kept;
+        if (!$closing && strlen($data) < 8) {
+            $this->kept = strlen($data);
+            stream_bucket_prepend($in, stream_bucket_new($this->stream, $data));
+            return PSFS_FEED_ME;
+        }
+        $this->kept = 0;
+        stream_bucket_append($out, stream_bucket_new($this->stream, "[" . $data . "]"));
+        return PSFS_PASS_ON;
+    }
+}
+stream_filter_register("defer", "defer");
+$f = __DIR__ . "/user_filter_feed_me_write.bin";
+$fp = fopen($f, 'w');
+stream_filter_append($fp, "defer", STREAM_FILTER_WRITE);
+var_dump(fwrite($fp, "abc"), fwrite($fp, "def"), fwrite($fp, "ghi"), fwrite($fp, "jk"));
+fclose($fp);
+var_dump(file_get_contents($f));
+?>
+--CLEAN--
+<?php
+@unlink(__DIR__ . "/user_filter_feed_me_write.bin");
+?>
+--EXPECT--
+int(3)
+int(3)
+int(3)
+int(2)
+string(15) "[abcdefghi][jk]"

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/tests/filters/user_filter_pass_on_leftover_read.phpt

@@ -0,0 +1,38 @@
+--TEST--
+Input left on the brigade by a read filter returning PSFS_PASS_ON is discarded with a warning
+--FILE--
+<?php
+class lines extends php_user_filter {
+    public function filter($in, $out, &$consumed, bool $closing): int {
+        $data = '';
+        while ($b = stream_bucket_make_writeable($in)) { $data .= $b->data; $consumed += $b->datalen; }
+        $p = strrpos($data, "\n");
+        if ($p === false) {
+            return PSFS_FEED_ME;
+        }
+        stream_bucket_append($out, stream_bucket_new($this->stream, substr($data, 0, $p + 1)));
+        if ($p + 1 < strlen($data)) {
+            stream_bucket_prepend($in, stream_bucket_new($this->stream, substr($data, $p + 1)));
+        }
+        return PSFS_PASS_ON;
+    }
+}
+stream_filter_register("lines", "lines");
+$f = __DIR__ . "/user_filter_pass_on_leftover_read.bin";
+file_put_contents($f, "abc\ndef\nghi\n");
+$fp = fopen($f, 'r');
+stream_set_chunk_size($fp, 6);
+stream_filter_append($fp, "lines", STREAM_FILTER_READ);
+var_dump(stream_get_contents($fp));
+fclose($fp);
+?>
+--CLEAN--
+<?php
+@unlink(__DIR__ . "/user_filter_pass_on_leftover_read.bin");
+?>
+--EXPECTF--
+Warning: stream_get_contents(): Unprocessed filter buckets remaining on input brigade in %s on line %d
+string(10) "abc
+f
+ghi
+"

ext/standard/user_filters.c

@@ -103,11 +103,23 @@ PHP_RSHUTDOWN_FUNCTION(user_filters)
 	return SUCCESS;
 }
 
+static void userfilter_release_brigade(php_stream_bucket_brigade *brigade)
+{
+	php_stream_bucket *bucket;
+
+	while ((bucket = brigade->head)) {
+		php_stream_bucket_unlink(bucket);
+		php_stream_bucket_delref(bucket);
+	}
+}
+
 static void userfilter_dtor(php_stream_filter *thisfilter)
 {
 	zval *obj = &thisfilter->abstract;
 	zval retval;
 
+	userfilter_release_brigade(&thisfilter->buffer);
+
 	if (Z_ISUNDEF_P(obj)) {
 		/* If there's no object associated then there's nothing to dispose of */
 		return;
@@ -166,6 +178,7 @@ static php_stream_filter_status_t userfilter_filter(
 		if (EG(exception)) {
 			EG(fake_scope) = old_scope;
 			if (buckets_in->head) {
+				userfilter_release_brigade(buckets_in);
 				php_error_docref(NULL, E_WARNING, "Unprocessed filter buckets remaining on input brigade");
 			}
 			zend_string_release(stream_name);
@@ -177,6 +190,17 @@ static php_stream_filter_status_t userfilter_filter(
 
 	EG(fake_scope) = old_scope;
 
+	/* Re-present input kept back by a previous PSFS_FEED_ME */
+	if (thisfilter->buffer.head) {
+		php_stream_bucket *bucket = thisfilter->buffer.tail;
+		while (bucket) {
+			php_stream_bucket *prev = bucket->prev;
+			php_stream_bucket_unlink(bucket);
+			php_stream_bucket_prepend(buckets_in, bucket);
+			bucket = prev;
+		}
+	}
+
 	ZVAL_STRINGL(&func_name, "filter", sizeof("filter")-1);
 
 	/* Setup calling arguments */
@@ -212,7 +236,22 @@ static php_stream_filter_status_t userfilter_filter(
 	}
 
 	if (buckets_in->head) {
-		php_error_docref(NULL, E_WARNING, "Unprocessed filter buckets remaining on input brigade");
+		if (ret == PSFS_FEED_ME && !(flags & PSFS_FLAG_FLUSH_CLOSE)) {
+			/* Keep the input for the next call and make sure the buckets own their data */
+			php_stream_bucket *bucket;
+			while ((bucket = buckets_in->head)) {
+				bucket = php_stream_bucket_make_writeable(bucket);
+				php_stream_bucket_append(&thisfilter->buffer, bucket);
+			}
+		} else {
+			userfilter_release_brigade(buckets_in);
+			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) {
+		userfilter_release_brigade(buckets_out);
 	}
 
 	/* filter resources are cleaned up by the stream destructor,