From 9714798a7431d0c871c1972f86fa4f9e3e3512c2 Mon Sep 17 00:00:00 2001 From: Jakub Zelenka Date: Mon, 5 Oct 2026 21:29:49 +0200 Subject: [PATCH] Fix leaks and data loss when user filter leaves buckets on input brigade The callers expect the input brigade to be empty when the filter returns, so buckets left on it leaked. Buckets put back with PSFS_FEED_ME to be seen again with the next data are now kept by the user filter and prepended to its input on the next call, which also fixes them aliasing the read chunk buffer, being re-filtered by preceding filters in a chain and being lost on the write path. Leftover buckets on any other status, or on the closing call, are released with the existing warning. Closes GH-20058 Co-authored-by: Nora Dossche <7771979+ndossche@users.noreply.github.com> Co-authored-by: Matthias Goergens --- NEWS | 4 ++ .../user_filter_feed_me_across_reads.phpt | 39 ++++++++++++++++++ .../filters/user_filter_feed_me_chained.phpt | 34 +++++++++++++++ .../user_filter_feed_me_closing_warning.phpt | 29 +++++++++++++ .../user_filter_feed_me_keeps_input.phpt | 32 +++++++++++++++ .../user_filter_feed_me_read_closing.phpt | 31 ++++++++++++++ .../user_filter_feed_me_untouched_input.phpt | 35 ++++++++++++++++ .../filters/user_filter_feed_me_write.phpt | 39 ++++++++++++++++++ .../filters/user_filter_no_consume_leak.phpt | 22 ++++++++++ .../user_filter_pass_on_leftover_read.phpt | 38 +++++++++++++++++ ..._filter_unprocessed_buckets_reentrant.phpt | 34 +++++++++++++++ ext/standard/user_filters.c | 41 ++++++++++++++++++- 12 files changed, 377 insertions(+), 1 deletion(-) create mode 100644 ext/standard/tests/filters/user_filter_feed_me_across_reads.phpt create mode 100644 ext/standard/tests/filters/user_filter_feed_me_chained.phpt create mode 100644 ext/standard/tests/filters/user_filter_feed_me_closing_warning.phpt create mode 100644 ext/standard/tests/filters/user_filter_feed_me_keeps_input.phpt create mode 100644 ext/standard/tests/filters/user_filter_feed_me_read_closing.phpt create mode 100644 ext/standard/tests/filters/user_filter_feed_me_untouched_input.phpt create mode 100644 ext/standard/tests/filters/user_filter_feed_me_write.phpt create mode 100644 ext/standard/tests/filters/user_filter_no_consume_leak.phpt create mode 100644 ext/standard/tests/filters/user_filter_pass_on_leftover_read.phpt create mode 100644 ext/standard/tests/filters/user_filter_unprocessed_buckets_reentrant.phpt diff --git a/NEWS b/NEWS index 25a261cb17c2..7b08f5a9d728 100644 --- a/NEWS +++ b/NEWS @@ -23,6 +23,10 @@ PHP NEWS or never terminates). (Ilia Alshanetsky) . Fixed memory leak in JIT ASSIGN_DIM when the assignment fails. (Jorg Sowa) +- Streams: + . Fixed leaks and data loss when user filter leaves buckets on input + brigade. (ndossche, Matthias Goergens, Jakub Zelenka) + 19 Nov 2026, PHP 8.4.28 - Core: diff --git a/ext/standard/tests/filters/user_filter_feed_me_across_reads.phpt b/ext/standard/tests/filters/user_filter_feed_me_across_reads.phpt new file mode 100644 index 000000000000..602665a535aa --- /dev/null +++ b/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-- + +--FILE-- +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) "" diff --git a/ext/standard/tests/filters/user_filter_feed_me_chained.phpt b/ext/standard/tests/filters/user_filter_feed_me_chained.phpt new file mode 100644 index 000000000000..959f5d121f29 --- /dev/null +++ b/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-- +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-- + +--EXPECT-- +bool(true) diff --git a/ext/standard/tests/filters/user_filter_feed_me_closing_warning.phpt b/ext/standard/tests/filters/user_filter_feed_me_closing_warning.phpt new file mode 100644 index 000000000000..3e543ae6b0a6 --- /dev/null +++ b/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-- +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-- + +--EXPECTF-- +int(5) + +Warning: fclose(): Unprocessed filter buckets remaining on input brigade in %s on line %d +string(0) "" diff --git a/ext/standard/tests/filters/user_filter_feed_me_keeps_input.phpt b/ext/standard/tests/filters/user_filter_feed_me_keeps_input.phpt new file mode 100644 index 000000000000..a4a1d44d7ebf --- /dev/null +++ b/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-- +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-- + +--EXPECT-- +int(24576) diff --git a/ext/standard/tests/filters/user_filter_feed_me_read_closing.phpt b/ext/standard/tests/filters/user_filter_feed_me_read_closing.phpt new file mode 100644 index 000000000000..3202a91d0c53 --- /dev/null +++ b/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-- +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-- + +--EXPECT-- +string(12) "[abcdefghij]" diff --git a/ext/standard/tests/filters/user_filter_feed_me_untouched_input.phpt b/ext/standard/tests/filters/user_filter_feed_me_untouched_input.phpt new file mode 100644 index 000000000000..7e05131e0726 --- /dev/null +++ b/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-- +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-- + +--EXPECT-- +int(24576) +int(8192) +int(8192) +int(8192) diff --git a/ext/standard/tests/filters/user_filter_feed_me_write.phpt b/ext/standard/tests/filters/user_filter_feed_me_write.phpt new file mode 100644 index 000000000000..2c4afa99e1fd --- /dev/null +++ b/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-- +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-- + +--EXPECT-- +int(3) +int(3) +int(3) +int(2) +string(15) "[abcdefghi][jk]" diff --git a/ext/standard/tests/filters/user_filter_no_consume_leak.phpt b/ext/standard/tests/filters/user_filter_no_consume_leak.phpt new file mode 100644 index 000000000000..c13b69c5892c --- /dev/null +++ b/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-- + +--EXPECTF-- +Warning: fwrite(): Unprocessed filter buckets remaining on input brigade in %s on line %d +int(0) diff --git a/ext/standard/tests/filters/user_filter_pass_on_leftover_read.phpt b/ext/standard/tests/filters/user_filter_pass_on_leftover_read.phpt new file mode 100644 index 000000000000..2261689cae9d --- /dev/null +++ b/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-- +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-- + +--EXPECTF-- +Warning: stream_get_contents(): Unprocessed filter buckets remaining on input brigade in %s on line %d +string(10) "abc +f +ghi +" diff --git a/ext/standard/tests/filters/user_filter_unprocessed_buckets_reentrant.phpt b/ext/standard/tests/filters/user_filter_unprocessed_buckets_reentrant.phpt new file mode 100644 index 000000000000..2845dd18c79a --- /dev/null +++ b/ext/standard/tests/filters/user_filter_unprocessed_buckets_reentrant.phpt @@ -0,0 +1,34 @@ +--TEST-- +Buckets appended to the input brigade by an error handler during the unprocessed buckets warning do not leak +--FILE-- +stream, "refilled"); + } + return PSFS_PASS_ON; + } +} +stream_filter_register("reentrant", "reentrant"); + +set_error_handler(function (int $severity, string $message): bool { + if (str_contains($message, "Unprocessed filter buckets")) { + stream_bucket_append($GLOBALS['brigade'], $GLOBALS['bucket']); + echo "Handled warning\n"; + return true; + } + return false; +}); + +$fp = fopen("php://memory", "w+"); +stream_filter_append($fp, "reentrant", STREAM_FILTER_WRITE); +var_dump(fwrite($fp, "input")); +fclose($fp); +?> +--EXPECT-- +Handled warning +int(0) diff --git a/ext/standard/user_filters.c b/ext/standard/user_filters.c index f5e58041ca80..74be83ee6ee7 100644 --- a/ext/standard/user_filters.c +++ b/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; @@ -167,6 +179,7 @@ static php_stream_filter_status_t userfilter_filter( EG(fake_scope) = old_scope; if (buckets_in->head) { php_error_docref(NULL, E_WARNING, "Unprocessed filter buckets remaining on input brigade"); + userfilter_release_brigade(buckets_in); } zend_string_release(stream_name); stream->flags &= ~PHP_STREAM_FLAG_NO_FCLOSE; @@ -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 { + php_error_docref(NULL, E_WARNING, "Unprocessed filter buckets remaining on input brigade"); + userfilter_release_brigade(buckets_in); + } + } + + /* 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,