From 6b8860445fcf6a61a492c7d343596e9db003c009 Mon Sep 17 00:00:00 2001 From: "Chao Li (Evan)" Date: Tue, 13 Oct 2026 14:12:44 +0800 Subject: [PATCH v1 2/2] Fix gzip stream validation in pg_restore. MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit With zlib 1.3.2, pg_restore can reject valid gzip files whose compressed and uncompressed sizes end at buffer boundaries, while accepting truncated files. Relying on gzread() and gzclose() does not reliably detect stream completion. Extract the gzip decompression logic from astreamer_gzip into a shared frontend helper using zlib's inflate API, and use it directly in pg_restore. Validate the stream through EOF, including any unread data when closing the file, and support concatenated gzip members. Extend the gzip regression test to cover incomplete trailers, concatenated members, invalid input, and compressed TOC files, and check the truncation diagnostic. Register the test with Meson and add coverage for reading gzip data through the shared file-reader callbacks. Bug: #19754 Reported-by: Stefan Graßl Author: Chao Li --- src/bin/pg_basebackup/nls.mk | 1 + src/bin/pg_dump/compress_gzip.c | 212 ++++++++++++++++------ src/bin/pg_dump/meson.build | 1 + src/bin/pg_dump/nls.mk | 1 + src/bin/pg_dump/t/006_pg_dump_compress.pl | 21 +++ src/bin/pg_dump/t/007_pg_dump_gzip.pl | 89 ++++++++- src/bin/pg_verifybackup/nls.mk | 3 +- src/bin/pg_waldump/nls.mk | 3 +- src/fe_utils/Makefile | 1 + src/fe_utils/astreamer_gzip.c | 165 ++++++----------- src/fe_utils/gzip.c | 188 +++++++++++++++++++ src/fe_utils/meson.build | 1 + src/include/fe_utils/gzip.h | 57 ++++++ src/tools/pgindent/typedefs.list | 4 + 14 files changed, 579 insertions(+), 168 deletions(-) create mode 100644 src/fe_utils/gzip.c create mode 100644 src/include/fe_utils/gzip.h diff --git a/src/bin/pg_basebackup/nls.mk b/src/bin/pg_basebackup/nls.mk index 6290c5c3649..4cb058fc45e 100644 --- a/src/bin/pg_basebackup/nls.mk +++ b/src/bin/pg_basebackup/nls.mk @@ -19,6 +19,7 @@ GETTEXT_FILES = $(FRONTEND_COMMON_GETTEXT_FILES) \ ../../fe_utils/astreamer_lz4.c \ ../../fe_utils/astreamer_tar.c \ ../../fe_utils/astreamer_zstd.c \ + ../../fe_utils/gzip.c \ ../../fe_utils/option_utils.c \ ../../fe_utils/recovery_gen.c \ ../../fe_utils/string_utils.c \ diff --git a/src/bin/pg_dump/compress_gzip.c b/src/bin/pg_dump/compress_gzip.c index 6781c41ec2f..b75f61ab02b 100644 --- a/src/bin/pg_dump/compress_gzip.c +++ b/src/bin/pg_dump/compress_gzip.c @@ -12,22 +12,17 @@ *------------------------------------------------------------------------- */ #include "postgres_fe.h" +#include #include #include "compress_gzip.h" +#include "fe_utils/gzip.h" #include "pg_backup_utils.h" #ifdef HAVE_LIBZ #include -/* - * We don't use the gzgetc() macro, because zlib's configuration logic is not - * robust enough to guarantee that the macro will have the same ideas about - * struct field layout as the library itself does; see for example - * https://gnats.netbsd.org/cgi-bin/query-pr-single.pl?number=59711 - * Instead, #undef the macro and fall back to the underlying function. - */ -#undef gzgetc +#define GZIP_BUFFER_SIZE 8192 /*---------------------- * Compressor API @@ -261,45 +256,103 @@ InitCompressorGzip(CompressorState *cs, * Compress File API *---------------------- */ +typedef struct GzipFileState +{ + bool reading; + union + { + gzFile gzfp; + struct + { + int fd; + GzipDecompressor decoder; + char *input; + size_t input_len; + size_t input_pos; + char *output; + size_t output_len; + size_t output_pos; + bool eof; + bool done; + } read; + }; +} GzipFileState; +/* + * Use strict gzip parsing: empty physical files, plain input, and trailing + * non-member bytes are errors. Valid empty and concatenated members work. + */ static size_t Gzip_read(void *ptr, size_t size, CompressFileHandle *CFH) { - gzFile gzfp = (gzFile) CFH->private_data; - int gzret; + GzipFileState *state = (GzipFileState *) CFH->private_data; + size_t copied = 0; - /* Reading zero bytes must be a no-op */ - if (size == 0) - return 0; + while (copied < size) + { + size_t bytes; - gzret = gzread(gzfp, ptr, size); + if (state->read.output_pos == state->read.output_len) + { + /* + * No buffered output remains; refill it unless decoding is + * complete. + */ + size_t consumed; + GzipDecompressResult result; - /* - * gzread returns zero on EOF as well as some error conditions, and less - * than zero on other error conditions, so we need to inspect for EOF on - * zero. - */ - if (gzret <= 0) - { - int errnum; - const char *errmsg; + if (state->read.done) + break; - if (gzret == 0 && gzeof(gzfp)) - return 0; + if (state->read.input_pos == state->read.input_len && !state->read.eof) + { + /* The input buffer is empty and EOF has not been reached */ + ssize_t readlen; + + do + { + readlen = read(state->read.fd, state->read.input, GZIP_BUFFER_SIZE); + /* retry if a signal interrupts the read */ + } while (readlen < 0 && errno == EINTR); + + if (readlen < 0) + pg_fatal("could not read from input file: %m"); + state->read.input_pos = 0; + state->read.input_len = readlen; + state->read.eof = readlen == 0; + } - errmsg = gzerror(gzfp, &errnum); + /* Produce more decompressed output */ + state->read.output_pos = 0; + result = gzip_decompressor_read(&state->read.decoder, + state->read.input + state->read.input_pos, + state->read.input_len - state->read.input_pos, + state->read.output, DEFAULT_IO_BUFFER_SIZE, + state->read.eof, &consumed, + &state->read.output_len); + state->read.input_pos += consumed; + if (result == GZIP_DECOMPRESS_ERROR) + pg_fatal("could not decompress data: %s", state->read.decoder.error); + state->read.done = result == GZIP_DECOMPRESS_DONE; + + /* Headers and empty members may produce no output. */ + if (state->read.output_len == 0) + continue; + } - pg_fatal("could not read from input file: %s", - errnum == Z_ERRNO ? strerror(errno) : errmsg); + bytes = Min(size - copied, state->read.output_len - state->read.output_pos); + memcpy((char *) ptr + copied, state->read.output + state->read.output_pos, bytes); + state->read.output_pos += bytes; + copied += bytes; } - - return (size_t) gzret; + return copied; } static void Gzip_write(const void *ptr, size_t size, CompressFileHandle *CFH) { - gzFile gzfp = (gzFile) CFH->private_data; + GzipFileState *state = (GzipFileState *) CFH->private_data; + gzFile gzfp = state->gzfp; int errnum; const char *errmsg; @@ -314,18 +367,10 @@ Gzip_write(const void *ptr, size_t size, CompressFileHandle *CFH) static int Gzip_getc(CompressFileHandle *CFH) { - gzFile gzfp = (gzFile) CFH->private_data; - int ret; + unsigned char ret; - errno = 0; - ret = gzgetc(gzfp); - if (ret == EOF) - { - if (!gzeof(gzfp)) - pg_fatal("could not read from input file: %m"); - else - pg_fatal("could not read from input file: end of file"); - } + if (Gzip_read(&ret, 1, CFH) != 1) + pg_fatal("could not read from input file: end of file"); return ret; } @@ -333,37 +378,71 @@ Gzip_getc(CompressFileHandle *CFH) static char * Gzip_gets(char *ptr, int size, CompressFileHandle *CFH) { - gzFile gzfp = (gzFile) CFH->private_data; + int i; - return gzgets(gzfp, ptr, size); + Assert(size > 0); + for (i = 0; i < size - 1; i++) + { + if (Gzip_read(&ptr[i], 1, CFH) != 1) + break; + if (ptr[i] == '\n') + { + i++; + break; + } + } + ptr[i] = '\0'; + return i > 0 ? ptr : NULL; } static bool Gzip_close(CompressFileHandle *CFH) { - gzFile gzfp = (gzFile) CFH->private_data; + GzipFileState *state = (GzipFileState *) CFH->private_data; + bool ok = true; + if (state->reading) + { + char buf[GZIP_BUFFER_SIZE]; + + /* Consume and validate any input the caller did not read. */ + while (Gzip_read(buf, sizeof(buf), CFH) > 0) + ; + if (!gzip_decompressor_end(&state->read.decoder)) + pg_fatal("could not decompress data: %s", state->read.decoder.error); + pg_free(state->read.input); + pg_free(state->read.output); + if (close(state->read.fd) != 0) + ok = false; + } + else if (gzclose(state->gzfp) != Z_OK) + ok = false; CFH->private_data = NULL; - - return gzclose(gzfp) == Z_OK; + pg_free(state); + return ok; } static bool Gzip_eof(CompressFileHandle *CFH) { - gzFile gzfp = (gzFile) CFH->private_data; + GzipFileState *state = (GzipFileState *) CFH->private_data; - return gzeof(gzfp) == 1; + return state->reading ? + (state->read.done && state->read.output_pos == state->read.output_len) : + gzeof(state->gzfp) == 1; } static const char * Gzip_get_error(CompressFileHandle *CFH) { - gzFile gzfp = (gzFile) CFH->private_data; + GzipFileState *state = (GzipFileState *) CFH->private_data; const char *errmsg; int errnum; - errmsg = gzerror(gzfp, &errnum); + if (state->reading) + return state->read.decoder.error ? state->read.decoder.error : ""; + + errmsg = gzerror(state->gzfp, &errnum); if (errnum == Z_ERRNO) errmsg = strerror(errno); @@ -373,6 +452,7 @@ Gzip_get_error(CompressFileHandle *CFH) static bool Gzip_open(const char *path, int fd, const char *mode, CompressFileHandle *CFH) { + GzipFileState *state; gzFile gzfp; char mode_compression[32]; @@ -387,16 +467,36 @@ Gzip_open(const char *path, int fd, const char *mode, CompressFileHandle *CFH) else strcpy(mode_compression, mode); - if (fd >= 0) + state = pg_malloc0_object(GzipFileState); + state->reading = mode[0] == 'r'; + if (state->reading) + { + state->read.fd = fd >= 0 ? dup(fd) : open(path, O_RDONLY | PG_BINARY); + if (state->read.fd < 0) + { + pg_free(state); + return false; + } + if (!gzip_decompressor_init(&state->read.decoder)) + pg_fatal("could not initialize compression library: %s", + state->read.decoder.error); + state->read.input = pg_malloc(GZIP_BUFFER_SIZE); + state->read.output = pg_malloc(DEFAULT_IO_BUFFER_SIZE); + } + else if (fd >= 0) { int dup_fd = dup(fd); if (dup_fd < 0) + { + pg_free(state); return false; + } gzfp = gzdopen(dup_fd, mode_compression); if (gzfp == NULL) { close(dup_fd); + pg_free(state); return false; } } @@ -404,10 +504,16 @@ Gzip_open(const char *path, int fd, const char *mode, CompressFileHandle *CFH) { gzfp = gzopen(path, mode_compression); if (gzfp == NULL) + { + pg_free(state); return false; + } } - CFH->private_data = gzfp; + if (!state->reading) + state->gzfp = gzfp; + + CFH->private_data = state; return true; } diff --git a/src/bin/pg_dump/meson.build b/src/bin/pg_dump/meson.build index 79bd5036841..4e948b716fb 100644 --- a/src/bin/pg_dump/meson.build +++ b/src/bin/pg_dump/meson.build @@ -103,6 +103,7 @@ tests += { 't/004_pg_dump_parallel.pl', 't/005_pg_dump_filterfile.pl', 't/006_pg_dump_compress.pl', + 't/007_pg_dump_gzip.pl', 't/010_dump_connstr.pl', ], }, diff --git a/src/bin/pg_dump/nls.mk b/src/bin/pg_dump/nls.mk index 9700f7cbe6c..13ecc02288a 100644 --- a/src/bin/pg_dump/nls.mk +++ b/src/bin/pg_dump/nls.mk @@ -29,6 +29,7 @@ GETTEXT_FILES = $(FRONTEND_COMMON_GETTEXT_FILES) \ ../../common/fe_memutils.c \ ../../common/file_utils.c \ ../../common/wait_error.c \ + ../../fe_utils/gzip.c \ ../../fe_utils/option_utils.c \ ../../fe_utils/string_utils.c GETTEXT_TRIGGERS = $(FRONTEND_COMMON_GETTEXT_TRIGGERS) \ diff --git a/src/bin/pg_dump/t/006_pg_dump_compress.pl b/src/bin/pg_dump/t/006_pg_dump_compress.pl index 32475c05c18..6add77fc8ad 100644 --- a/src/bin/pg_dump/t/006_pg_dump_compress.pl +++ b/src/bin/pg_dump/t/006_pg_dump_compress.pl @@ -744,6 +744,27 @@ for my $method (qw(lz4 zstd)) } } +SKIP: +{ + my ($directory_path) = grep { -s $_ } + glob("$tempdir/compression_gzip_dir/*.dat.gz"); + + skip "gzip compression not supported by this build", 2 if !$supports_gzip; + die "could not find compressed data file for gzip" + if !defined($directory_path); + my $directory_size = -s $directory_path; + truncate($directory_path, $directory_size - 1) + or die "could not truncate $directory_path: $!"; + $node->command_fails_like( + [ + 'pg_restore', + '--file' => "$tempdir/gzip_directory_bad.sql", + "$tempdir/compression_gzip_dir" + ], + qr/could not decompress data: compressed stream is incomplete/, + "gzip directory archive with truncated compressed data fails"); +} + SKIP: { # Zstd needs empty-input processing at end-of-input to flush buffered diff --git a/src/bin/pg_dump/t/007_pg_dump_gzip.pl b/src/bin/pg_dump/t/007_pg_dump_gzip.pl index ed29c39d681..ad6d7bc4e48 100644 --- a/src/bin/pg_dump/t/007_pg_dump_gzip.pl +++ b/src/bin/pg_dump/t/007_pg_dump_gzip.pl @@ -30,6 +30,14 @@ $node->safe_psql('postgres', q{ INSERT INTO gzip_boundary_test VALUES (0, ''); }); +# Store a long comment in the table of contents (TOC). The compressed TOC test +# below will read this string across several decompressor output buffers. +$node->safe_psql('postgres', q{ + DO $$ BEGIN + EXECUTE format('COMMENT ON TABLE gzip_boundary_test IS %L', repeat('z', 300000)); + END $$; +}); + # Make an initial dump to find the uncompressed size of the table data. my $dumpdir = "$tempdir/gzip_boundary_dump"; my $restore_file = "$tempdir/gzip_boundary_restore.sql"; @@ -111,17 +119,94 @@ my @rows = $restored =~ /^\d+\t/gm; is(scalar(@rows), 20001, 'all table rows were decompressed'); # Cut the file in half, as in the bug report. Some data can still be read, but -# pg_restore must fail because the gzip stream is incomplete. +# pg_restore must report that the gzip stream is incomplete. my $truncated_size = int(length($gzip_data) / 2); open $fh, '+<:raw', $datafile or die "could not open $datafile: $!"; truncate($fh, $truncated_size) or die "could not truncate $datafile: $!"; close $fh or die "could not close $datafile: $!"; -$node->command_fails( +$node->command_fails_like( [ 'pg_restore', '--file' => "$tempdir/gzip_boundary_truncated.sql", $dumpdir ], + qr/could not decompress data: compressed stream is incomplete/, 'fail to restore the truncated gzip data file'); +# Put back the complete file, then remove its last byte. All table data is +# still present, but the incomplete gzip trailer must cause an error. +open $fh, '>:raw', $datafile or die "could not open $datafile: $!"; +print {$fh} $gzip_data or die "could not write $datafile: $!"; +close $fh or die "could not close $datafile: $!"; +truncate($datafile, length($gzip_data) - 1) + or die "could not truncate $datafile: $!"; + +$node->command_fails_like( + [ + 'pg_restore', '--file' => "$tempdir/gzip_boundary_no_trailer.sql", + $dumpdir + ], + qr/could not decompress data: compressed stream is incomplete/, + 'reject gzip data missing its final trailer byte'); + +# A gzip file can contain several members. Put an empty member before and after +# the table data. The reader must continue after each member to read all rows. +# This hex string is a complete gzip member with no uncompressed data. +my $empty_member = pack('H*', '1f8b08000000000002ff03000000000000000000'); +open $fh, '>:raw', $datafile or die "could not open $datafile: $!"; +print {$fh} $empty_member . $gzip_data . $empty_member + or die "could not write $datafile: $!"; +close $fh or die "could not close $datafile: $!"; +$node->command_ok( + ['pg_restore', '--file' => "$tempdir/gzip_members.sql", $dumpdir], + 'restore concatenated gzip members including empty members'); + +my $members_sql = slurp_file("$tempdir/gzip_members.sql"); +my @member_rows = $members_sql =~ /^\d+\t/gm; +is(scalar(@member_rows), 20001, + 'all table rows were decompressed across gzip member boundaries'); + +# Reject a zero-byte file, plain text, and extra bytes that are not another gzip +# member. A zero-byte file differs from the valid empty member tested above. +for my $case ( + ['empty', '', qr/compressed stream is empty/], + ['plain', "1\tvalue\n\\.\n", qr/incorrect header check/], + ['trailing', $gzip_data . ("\0" x 100), qr/incorrect header check/]) +{ + my ($name, $data, $error) = @$case; + open $fh, '>:raw', $datafile or die "could not open $datafile: $!"; + print {$fh} $data or die "could not write $datafile: $!"; + close $fh or die "could not close $datafile: $!"; + $node->command_fails_like( + ['pg_restore', '--file' => "$tempdir/gzip_${name}.sql", $dumpdir], + $error, + "reject $name gzip input"); +} + +SKIP: +{ + # These TOC tests need an external gzip program to compress toc.dat. + my $gzip = $ENV{'GZIP_PROGRAM'}; + skip 'GZIP_PROGRAM is not configured', 4 if !defined($gzip) || $gzip eq ''; + + # Reading the TOC uses both single-byte reads and buffer reads. The long + # comment requires a buffer read across several decompressor output buffers. + # Add data after the TOC entries that the TOC parser will not read. Closing + # the file must finish reading the gzip stream and check its trailer. + my $tocfile = "$dumpdir/toc.dat"; + open $fh, '>>:raw', $tocfile or die "could not open $tocfile: $!"; + print {$fh} 'x' x (512 * 1024) or die "could not write $tocfile: $!"; + close $fh or die "could not close $tocfile: $!"; + command_ok([$gzip, '-f', $tocfile], 'compress the padded TOC'); + $node->command_ok(['pg_restore', '--list', $dumpdir], + 'read a compressed TOC across output-buffer boundaries'); + # Remove only the last byte of the gzip trailer. The TOC entries can still + # be read, but closing the file must detect the incomplete stream. + truncate("$tocfile.gz", (-s "$tocfile.gz") - 1) + or die "could not truncate $tocfile.gz: $!"; + $node->command_fails_like(['pg_restore', '--list', $dumpdir], + qr/could not decompress data: compressed stream is incomplete/, + 'validate unread compressed input when closing the TOC'); +} + done_testing(); diff --git a/src/bin/pg_verifybackup/nls.mk b/src/bin/pg_verifybackup/nls.mk index 2dfe7eb22c6..0016337a44a 100644 --- a/src/bin/pg_verifybackup/nls.mk +++ b/src/bin/pg_verifybackup/nls.mk @@ -12,7 +12,8 @@ GETTEXT_FILES = $(FRONTEND_COMMON_GETTEXT_FILES) \ ../../fe_utils/astreamer_gzip.c \ ../../fe_utils/astreamer_lz4.c \ ../../fe_utils/astreamer_tar.c \ - ../../fe_utils/astreamer_zstd.c + ../../fe_utils/astreamer_zstd.c \ + ../../fe_utils/gzip.c GETTEXT_TRIGGERS = $(FRONTEND_COMMON_GETTEXT_TRIGGERS) \ json_token_error:2 \ json_manifest_parse_failure:2 \ diff --git a/src/bin/pg_waldump/nls.mk b/src/bin/pg_waldump/nls.mk index 46fb99a2377..e4e02fe08fc 100644 --- a/src/bin/pg_waldump/nls.mk +++ b/src/bin/pg_waldump/nls.mk @@ -6,7 +6,8 @@ GETTEXT_FILES = $(FRONTEND_COMMON_GETTEXT_FILES) \ xlogreader.c \ xlogstats.c \ ../../common/fe_memutils.c \ - ../../common/file_utils.c + ../../common/file_utils.c \ + ../../fe_utils/gzip.c GETTEXT_TRIGGERS = $(FRONTEND_COMMON_GETTEXT_TRIGGERS) \ report_invalid_record:2 GETTEXT_FLAGS = $(FRONTEND_COMMON_GETTEXT_FLAGS) \ diff --git a/src/fe_utils/Makefile b/src/fe_utils/Makefile index cbfbf93ac69..a96af1928a9 100644 --- a/src/fe_utils/Makefile +++ b/src/fe_utils/Makefile @@ -29,6 +29,7 @@ OBJS = \ cancel.o \ conditional.o \ connect_utils.o \ + gzip.o \ mbprint.o \ option_utils.o \ parallel_slot.o \ diff --git a/src/fe_utils/astreamer_gzip.c b/src/fe_utils/astreamer_gzip.c index 2a374cf08a8..130f42db7e1 100644 --- a/src/fe_utils/astreamer_gzip.c +++ b/src/fe_utils/astreamer_gzip.c @@ -34,6 +34,7 @@ #include "common/logging.h" #include "fe_utils/astreamer.h" +#include "fe_utils/gzip.h" #ifdef HAVE_LIBZ typedef struct astreamer_gzip_writer @@ -46,9 +47,8 @@ typedef struct astreamer_gzip_writer typedef struct astreamer_gzip_decompressor { astreamer base; - z_stream zstream; + GzipDecompressor decoder; size_t bytes_written; - astreamer_decompression_state state; } astreamer_gzip_decompressor; static void astreamer_gzip_writer_content(astreamer *streamer, @@ -71,8 +71,6 @@ static void astreamer_gzip_decompressor_content(astreamer *streamer, astreamer_archive_context context); static void astreamer_gzip_decompressor_finalize(astreamer *streamer); static void astreamer_gzip_decompressor_free(astreamer *streamer); -static void *gzip_palloc(void *opaque, unsigned items, unsigned size); -static void gzip_pfree(void *opaque, void *address); static const astreamer_ops astreamer_gzip_decompressor_ops = { .content = astreamer_gzip_decompressor_content, @@ -238,7 +236,6 @@ astreamer_gzip_decompressor_new(astreamer *next) { #ifdef HAVE_LIBZ astreamer_gzip_decompressor *streamer; - z_stream *zs; Assert(next != NULL); @@ -251,27 +248,9 @@ astreamer_gzip_decompressor_new(astreamer *next) /* Use a buffer size comparable to the other decompressors */ enlargeStringInfo(&streamer->base.bbs_buffer, 256 * 1024 - 1); - /* Initialize internal stream state for decompression */ - zs = &streamer->zstream; - zs->zalloc = gzip_palloc; - zs->zfree = gzip_pfree; - zs->next_out = (uint8 *) streamer->base.bbs_buffer.data; - zs->avail_out = streamer->base.bbs_buffer.maxlen; - - /* - * Data compression was initialized using deflateInit2 to request a gzip - * header. Similarly, we are using inflateInit2 to initialize data - * decompression. - * - * Per the documentation for inflateInit2, the second argument is - * "windowBits" and its value must be greater than or equal to the value - * provided while compressing the data, so we are using the maximum - * possible value for safety. - */ - if (inflateInit2(zs, 15 + 16) != Z_OK) - pg_fatal("could not initialize compression library"); - - streamer->state = ASTREAMER_STREAM_NEW; + if (!gzip_decompressor_init(&streamer->decoder)) + pg_fatal("could not initialize compression library: %s", + streamer->decoder.error); return &streamer->base; #else pg_fatal("this build does not support compression with %s", "gzip"); @@ -292,90 +271,72 @@ astreamer_gzip_decompressor_content(astreamer *streamer, astreamer_archive_context context) { astreamer_gzip_decompressor *mystreamer; - z_stream *zs; mystreamer = (astreamer_gzip_decompressor *) streamer; - zs = &mystreamer->zstream; - zs->next_in = (const uint8 *) data; - zs->avail_in = len; - - /* Process the current chunk */ - while (zs->avail_in > 0) + for (;;) { - int res; - - Assert(mystreamer->bytes_written < mystreamer->base.bbs_buffer.maxlen); - - zs->next_out = (uint8 *) - mystreamer->base.bbs_buffer.data + mystreamer->bytes_written; - zs->avail_out = - mystreamer->base.bbs_buffer.maxlen - mystreamer->bytes_written; - - /* - * This call decompresses data starting at zs->next_in and updates - * zs->next_in * and zs->avail_in. It generates output data starting - * at zs->next_out and updates zs->next_out and zs->avail_out - * accordingly. - */ - res = inflate(zs, Z_NO_FLUSH); - - if (res == Z_STREAM_END) - mystreamer->state = ASTREAMER_FRAME_COMPLETE; - else if (res == Z_OK || res == Z_BUF_ERROR) - mystreamer->state = ASTREAMER_FRAME_INCOMPLETE; - else - pg_fatal("could not decompress data: %s", - zs->msg ? zs->msg : "unknown error"); - - mystreamer->bytes_written = - mystreamer->base.bbs_buffer.maxlen - zs->avail_out; - - /* If output buffer is full then pass data to next streamer */ - if (mystreamer->bytes_written >= mystreamer->base.bbs_buffer.maxlen) + size_t consumed; + size_t produced; + GzipDecompressResult result; + + result = gzip_decompressor_read(&mystreamer->decoder, data, len, + streamer->bbs_buffer.data + mystreamer->bytes_written, + streamer->bbs_buffer.maxlen - mystreamer->bytes_written, + false, &consumed, &produced); + if (result == GZIP_DECOMPRESS_ERROR) + pg_fatal("could not decompress data: %s", mystreamer->decoder.error); + + data += consumed; + len -= consumed; + mystreamer->bytes_written += produced; + + if (mystreamer->bytes_written == streamer->bbs_buffer.maxlen) { - astreamer_content(mystreamer->base.bbs_next, member, - mystreamer->base.bbs_buffer.data, - mystreamer->base.bbs_buffer.maxlen, context); + astreamer_content(streamer->bbs_next, member, + streamer->bbs_buffer.data, + streamer->bbs_buffer.maxlen, context); mystreamer->bytes_written = 0; } - - /* - * The input can be several gzip members one after another, which is - * what pg_basebackup writes for pg_wal.tar.gz. Get ready for the - * next one. - */ - if (res == Z_STREAM_END && inflateReset(zs) != Z_OK) - pg_fatal("could not reset compression stream"); + else if (len == 0) + /* Output is not full and no more input, we're done */ + break; } } /* - * End-of-stream processing. + * End-of-stream processing, including output still pending in the decoder. */ static void astreamer_gzip_decompressor_finalize(astreamer *streamer) { astreamer_gzip_decompressor *mystreamer; + GzipDecompressResult result; mystreamer = (astreamer_gzip_decompressor *) streamer; - if (unlikely(mystreamer->state == ASTREAMER_STREAM_NEW)) - pg_fatal("could not decompress data: compressed stream is empty"); - else if (mystreamer->state != ASTREAMER_FRAME_COMPLETE) - pg_fatal("could not decompress data: compressed stream is incomplete"); - - /* - * End of the stream, if there is some pending data in output buffers then - * we must forward it to next streamer. - */ - if (mystreamer->bytes_written > 0) - astreamer_content(mystreamer->base.bbs_next, NULL, - mystreamer->base.bbs_buffer.data, - mystreamer->bytes_written, - ASTREAMER_UNKNOWN); - - astreamer_finalize(mystreamer->base.bbs_next); + do + { + size_t consumed; + size_t produced; + + result = gzip_decompressor_read(&mystreamer->decoder, NULL, 0, + streamer->bbs_buffer.data + mystreamer->bytes_written, + streamer->bbs_buffer.maxlen - mystreamer->bytes_written, + true, &consumed, &produced); + if (result == GZIP_DECOMPRESS_ERROR) + pg_fatal("could not decompress data: %s", mystreamer->decoder.error); + mystreamer->bytes_written += produced; + if (mystreamer->bytes_written > 0) + { + astreamer_content(streamer->bbs_next, NULL, + streamer->bbs_buffer.data, + mystreamer->bytes_written, ASTREAMER_UNKNOWN); + mystreamer->bytes_written = 0; + } + } while (result != GZIP_DECOMPRESS_DONE); + + astreamer_finalize(streamer->bbs_next); } /* @@ -389,28 +350,10 @@ astreamer_gzip_decompressor_free(astreamer *streamer) mystreamer = (astreamer_gzip_decompressor *) streamer; astreamer_free(streamer->bbs_next); - inflateEnd(&mystreamer->zstream); + if (!gzip_decompressor_end(&mystreamer->decoder)) + pg_fatal("could not decompress data: %s", mystreamer->decoder.error); pfree(streamer->bbs_buffer.data); pfree(streamer); } -/* - * Wrapper function to adjust the signature of palloc to match what libz - * expects. - */ -static void * -gzip_palloc(void *opaque, unsigned items, unsigned size) -{ - return palloc(items * size); -} - -/* - * Wrapper function to adjust the signature of pfree to match what libz - * expects. - */ -static void -gzip_pfree(void *opaque, void *address) -{ - pfree(address); -} #endif diff --git a/src/fe_utils/gzip.c b/src/fe_utils/gzip.c new file mode 100644 index 00000000000..f5d9066b2d8 --- /dev/null +++ b/src/fe_utils/gzip.c @@ -0,0 +1,188 @@ +/*------------------------------------------------------------------------- + * + * gzip.c + * + * Shared frontend routines for gzip compression and decompression. + * These routines use zlib's low-level streaming API, without relying on + * its gzip file I/O functions such as gzread() and gzclose(). + * + * Portions Copyright (c) 1996-2026, PostgreSQL Global Development Group + * + * IDENTIFICATION + * src/fe_utils/gzip.c + * + *------------------------------------------------------------------------- + */ +#include "postgres_fe.h" + +#include + +#include "fe_utils/gzip.h" + +#ifdef HAVE_LIBZ +bool +gzip_decompressor_init(GzipDecompressor *decoder) +{ + int res; + + memset(decoder, 0, sizeof(GzipDecompressor)); + decoder->state = GZIP_STREAM_NEW; + + /* + * Data compression was initialized using deflateInit2 to request a gzip + * header. Similarly, we are using inflateInit2 to initialize data + * decompression. + * + * Per the documentation for inflateInit2, the second argument is + * "windowBits" and its value must be greater than or equal to the value + * provided while compressing the data, so we are using the maximum + * possible value for safety. + */ + res = inflateInit2(&decoder->stream, 15 + 16); + if (res != Z_OK) + { + decoder->error = decoder->stream.msg ? + decoder->stream.msg : zError(res); + return false; + } + return true; +} + +/* + * Consume input and produce at most output_size bytes. MORE means that the + * caller must provide more input or output space; it is not an error. DONE + * is returned only when eof is true and input ends at a member boundary. + * Plain input, trailing non-member bytes, and physically empty streams are + * rejected, while valid empty members are accepted. + * + * On return, *consumed is the number of bytes read from input, and *produced + * is the number of decompressed bytes written to output. The caller must + * advance the input pointer by *consumed and reduce input_size accordingly + * before passing the remaining input to the next call. + */ +GzipDecompressResult +gzip_decompressor_read(GzipDecompressor *decoder, + const void *input, size_t input_size, + void *output, size_t output_size, + bool eof, size_t *consumed, size_t *produced) +{ + z_stream *zs = &decoder->stream; + GzipDecompressResult result = GZIP_DECOMPRESS_MORE; + + /* + * zlib's input and output sizes use uInt, which may be narrower than + * size_t. Limit each call to UINT_MAX bytes. + */ + uInt zs_in_size = Min(input_size, UINT_MAX); + uInt zs_out_size = Min(output_size, UINT_MAX); + + Assert(input != NULL || input_size == 0); + Assert(output_size > 0); + *consumed = *produced = 0; + zs->next_in = (const Bytef *) input; + zs->avail_in = zs_in_size; + zs->next_out = output; + zs->avail_out = zs_out_size; + + for (;;) + { + int res; + uInt old_in = zs->avail_in; + uInt old_out = zs->avail_out; + + if (zs->avail_in == 0 && decoder->state != GZIP_MEMBER_INCOMPLETE) + { + /* + * No input remains in the current zlib span. We can only be done + * if the caller indicated EOF and all supplied input was consumed + * at a member boundary. + */ + if (eof && input_size == zs_in_size) + { + /* At EOF, and no more data from the caller's input */ + if (decoder->state == GZIP_STREAM_NEW) + { + decoder->error = _("compressed stream is empty"); + result = GZIP_DECOMPRESS_ERROR; + } + else + result = GZIP_DECOMPRESS_DONE; + } + break; + } + + /* Check if we have no more output space */ + if (zs->avail_out == 0) + break; + + /* + * This call decompresses data starting at zs->next_in and updates + * zs->next_in and zs->avail_in. It generates output data starting at + * zs->next_out and updates zs->next_out and zs->avail_out + * accordingly. + */ + res = inflate(zs, Z_NO_FLUSH); + if (res == Z_STREAM_END) + { + decoder->state = GZIP_MEMBER_COMPLETE; + if (inflateReset(zs) != Z_OK) + { + decoder->error = _("could not reset compression stream"); + result = GZIP_DECOMPRESS_ERROR; + break; + } + } + else if (res == Z_OK || res == Z_BUF_ERROR) + { + decoder->state = GZIP_MEMBER_INCOMPLETE; + if (old_in == zs->avail_in && old_out == zs->avail_out) + { + /* No progress was made */ + if (zs->avail_in > 0) + { + /* Input data is still available. */ + decoder->error = _("compression library made no progress"); + result = GZIP_DECOMPRESS_ERROR; + } + else if (eof && input_size == zs_in_size) + { + /* + * At EOF, inflate made no further progress without + * completing a member. + */ + decoder->error = _("compressed stream is incomplete"); + result = GZIP_DECOMPRESS_ERROR; + } + break; + } + } + else + { + /* An error occurred during decompression */ + decoder->error = zs->msg ? zs->msg : zError(res); + result = GZIP_DECOMPRESS_ERROR; + break; + } + } + + *consumed = zs_in_size - zs->avail_in; + *produced = zs_out_size - zs->avail_out; + zs->next_in = NULL; + zs->next_out = NULL; + return result; +} + +bool +gzip_decompressor_end(GzipDecompressor *decoder) +{ + int res; + + res = inflateEnd(&decoder->stream); + if (res != Z_OK) + { + decoder->error = zError(res); + return false; + } + return true; +} +#endif /* HAVE_LIBZ */ diff --git a/src/fe_utils/meson.build b/src/fe_utils/meson.build index 86befca192e..34362dddf49 100644 --- a/src/fe_utils/meson.build +++ b/src/fe_utils/meson.build @@ -10,6 +10,7 @@ fe_utils_sources = files( 'cancel.c', 'conditional.c', 'connect_utils.c', + 'gzip.c', 'mbprint.c', 'option_utils.c', 'parallel_slot.c', diff --git a/src/include/fe_utils/gzip.h b/src/include/fe_utils/gzip.h new file mode 100644 index 00000000000..682fbcca3b0 --- /dev/null +++ b/src/include/fe_utils/gzip.h @@ -0,0 +1,57 @@ +/*------------------------------------------------------------------------- + * + * gzip.h + * + * Portions Copyright (c) 1996-2026, PostgreSQL Global Development Group + * + * IDENTIFICATION + * src/include/fe_utils/gzip.h + * + *------------------------------------------------------------------------- + */ +#ifndef FE_UTILS_GZIP_H +#define FE_UTILS_GZIP_H + +#ifdef HAVE_LIBZ +#include + +typedef enum +{ + GZIP_DECOMPRESS_MORE, + GZIP_DECOMPRESS_DONE, + GZIP_DECOMPRESS_ERROR +} GzipDecompressResult; + +typedef enum +{ + GZIP_STREAM_NEW, + GZIP_MEMBER_INCOMPLETE, + GZIP_MEMBER_COMPLETE +} GzipDecompressState; + +typedef struct GzipDecompressor +{ + z_stream stream; + GzipDecompressState state; + const char *error; +} GzipDecompressor; + +/* + * Errors are returned to the caller and described in decoder->error. + * After GZIP_DECOMPRESS_ERROR, callers must stop decoding and release the + * decoder with gzip_decompressor_end(). Only a successfully initialized + * decoder may be passed to gzip_decompressor_end(). + * + * gzip_decompressor_end() releases resources without checking completeness. + * To validate the stream, call gzip_decompressor_read() with eof set to true + * until it returns GZIP_DECOMPRESS_DONE (or GZIP_DECOMPRESS_ERROR). + */ +extern bool gzip_decompressor_init(GzipDecompressor *decoder); +extern GzipDecompressResult gzip_decompressor_read(GzipDecompressor *decoder, + const void *input, size_t input_size, + void *output, size_t output_size, + bool eof, size_t *consumed, size_t *produced); +extern bool gzip_decompressor_end(GzipDecompressor *decoder); +#endif + +#endif /* FE_UTILS_GZIP_H */ diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list index a7a8dbdd23f..ffdc973c4b8 100644 --- a/src/tools/pgindent/typedefs.list +++ b/src/tools/pgindent/typedefs.list @@ -1183,6 +1183,10 @@ GucStackState GucStringAssignHook GucStringCheckHook GzipCompressorState +GzipDecompressResult +GzipDecompressState +GzipDecompressor +GzipFileState HANDLE HASHACTION HASHBUCKET -- 2.50.1 (Apple Git-155)