PG-2234 Fix pg_rewind with encrypted tables

.github/ISSUE_TEMPLATE/bug.yml

@@ -32,7 +32,7 @@ body:
     attributes:
       label: Version
       description: What version of pg_tde and Percona Server PostgreSQL are you running?
-      placeholder: pg_tde 2.1, Percona Server for PostgreSQL 18.0
+      placeholder: pg_tde 2.2, Percona Server for PostgreSQL 18.0
     validations:
       required: true
   - type: textarea

Makefile

@@ -1,7 +1,11 @@
 PGFILEDESC = "pg_tde access method"
 MODULE_big = pg_tde
 EXTENSION = pg_tde
-DATA = pg_tde--2.0--2.1.sql pg_tde--1.0--2.0.sql pg_tde--1.0.sql
+DATA = \
+	pg_tde--1.0.sql \
+	pg_tde--1.0--2.0.sql \
+	pg_tde--2.0--2.1.sql \
+	pg_tde--2.1--2.2.sql
 
 TAP_TESTS = 1
 
@@ -66,6 +70,7 @@ RWOBJS = \
 	$(FETOOLS)/pg_rewind/local_source.o \
 	$(FETOOLS)/pg_rewind/parsexlog.o \
 	$(FETOOLS)/pg_rewind/pg_rewind.o \
+	$(FETOOLS)/pg_rewind/tde_ops.o \
 	$(FETOOLS)/pg_rewind/timeline.o
 
 RMGRDESCSOURCES = $(sort $(wildcard $(FETOOLS)/rmgrdesc/*desc*.c))

documentation/variables.yml

@@ -1,10 +1,10 @@
 #Variables used throughout the docs
 
 latestreleasenotes: 'release-notes-v2.1.2'
-tdeversion: '2.1.2'
-release: '2.1.2'
+tdeversion: '2.2.0'
+release: '2.2.0'
 pgversion17: '18.3'
-tdebranch: release-2.1.2
+tdebranch: release-2.2.0
 
 date:
 

expected/version.out

@@ -1,14 +1,14 @@
 SELECT * FROM pg_get_loaded_modules() WHERE file_name IN ('pg_tde.so', 'pg_tde.dylib');
  module_name | version | file_name 
 -------------+---------+-----------
- pg_tde      | 2.1.2   | pg_tde.so
+ pg_tde      | 2.2.0   | pg_tde.so
 (1 row)
 
 CREATE EXTENSION pg_tde;
 SELECT pg_tde_version();
  pg_tde_version 
 ----------------
- pg_tde 2.1.2
+ pg_tde 2.2.0
 (1 row)
 
 DROP EXTENSION pg_tde;

expected/version_1.out

@@ -7,7 +7,7 @@ CREATE EXTENSION pg_tde;
 SELECT pg_tde_version();
  pg_tde_version 
 ----------------
- pg_tde 2.1.2
+ pg_tde 2.2.0
 (1 row)
 
 DROP EXTENSION pg_tde;

expected/version_2.out

@@ -1,14 +1,14 @@
 SELECT * FROM pg_get_loaded_modules() WHERE file_name IN ('pg_tde.so', 'pg_tde.dylib');
  module_name | version |  file_name   
 -------------+---------+--------------
- pg_tde      | 2.1.2   | pg_tde.dylib
+ pg_tde      | 2.2.0   | pg_tde.dylib
 (1 row)
 
 CREATE EXTENSION pg_tde;
 SELECT pg_tde_version();
  pg_tde_version 
 ----------------
- pg_tde 2.1.2
+ pg_tde 2.2.0
 (1 row)
 
 DROP EXTENSION pg_tde;

fetools/pg17/pg_basebackup/bbstreamer_file.c

@@ -220,6 +220,10 @@ bbstreamer_extractor_content(bbstreamer *streamer, bbstreamer_member *member,
 		case BBSTREAMER_MEMBER_HEADER:
 			Assert(mystreamer->file == NULL);
 
+			if (!path_is_safe_for_extraction(member->pathname))
+				pg_fatal("tar member has unsafe path name: \"%s\"",
+						 member->pathname);
+
 			/* Prepend basepath. */
 			snprintf(mystreamer->filename, sizeof(mystreamer->filename),
 					 "%s/%s", mystreamer->basepath, member->pathname);
@@ -238,6 +242,14 @@ bbstreamer_extractor_content(bbstreamer *streamer, bbstreamer_member *member,
 
 				if (mystreamer->link_map)
 					linktarget = mystreamer->link_map(linktarget);
+
+				if (!is_absolute_path(linktarget) &&
+					!path_is_safe_for_extraction(member->linktarget))
+				{
+					pg_fatal("link target has unsafe path name: \"%s\"",
+							 member->linktarget);
+				}
+
 				extract_link(mystreamer->filename, linktarget);
 			}
 			else
@@ -285,7 +297,9 @@ bbstreamer_extractor_content(bbstreamer *streamer, bbstreamer_member *member,
 		case BBSTREAMER_MEMBER_TRAILER:
 			if (mystreamer->file == NULL)
 				break;
-			fclose(mystreamer->file);
+			if (fclose(mystreamer->file) != 0)
+				pg_fatal("could not close file \"%s\": %m",
+						 mystreamer->filename);
 			mystreamer->file = NULL;
 			break;
 

fetools/pg17/pg_basebackup/bbstreamer_gzip.c

@@ -292,8 +292,9 @@ bbstreamer_gzip_decompressor_content(bbstreamer *streamer,
 		 */
 		res = inflate(zs, Z_NO_FLUSH);
 
-		if (res == Z_STREAM_ERROR)
-			pg_log_error("could not decompress data: %s", zs->msg);
+		if (res != Z_OK && res != Z_STREAM_END && res != Z_BUF_ERROR)
+			pg_fatal("could not decompress data: %s",
+					 zs->msg ? zs->msg : "unknown error");
 
 		mystreamer->bytes_written =
 			mystreamer->base.bbs_buffer.maxlen - zs->avail_out;
@@ -323,10 +324,11 @@ bbstreamer_gzip_decompressor_finalize(bbstreamer *streamer)
 	 * End of the stream, if there is some pending data in output buffers then
 	 * we must forward it to next streamer.
 	 */
-	bbstreamer_content(mystreamer->base.bbs_next, NULL,
-					   mystreamer->base.bbs_buffer.data,
-					   mystreamer->base.bbs_buffer.maxlen,
-					   BBSTREAMER_UNKNOWN);
+	if (mystreamer->bytes_written > 0)
+		bbstreamer_content(mystreamer->base.bbs_next, NULL,
+						   mystreamer->base.bbs_buffer.data,
+						   mystreamer->bytes_written,
+						   BBSTREAMER_UNKNOWN);
 
 	bbstreamer_finalize(mystreamer->base.bbs_next);
 }
@@ -337,7 +339,12 @@ bbstreamer_gzip_decompressor_finalize(bbstreamer *streamer)
 static void
 bbstreamer_gzip_decompressor_free(bbstreamer *streamer)
 {
+	bbstreamer_gzip_decompressor *mystreamer;
+
+	mystreamer = (bbstreamer_gzip_decompressor *) streamer;
+
 	bbstreamer_free(streamer->bbs_next);
+	inflateEnd(&mystreamer->zstream);
 	pfree(streamer->bbs_buffer.data);
 	pfree(streamer);
 }

fetools/pg17/pg_basebackup/bbstreamer_lz4.c

@@ -92,8 +92,8 @@ bbstreamer_lz4_compressor_new(bbstreamer *next, pg_compress_specification *compr
 
 	ctxError = LZ4F_createCompressionContext(&streamer->cctx, LZ4F_VERSION);
 	if (LZ4F_isError(ctxError))
- pg_log_error("could not create lz4 compression context: %s",
-   LZ4F_getErrorName(ctxError));
+ pg_fatal("could not create lz4 compression context: %s",
+				 LZ4F_getErrorName(ctxError));
 
 	return &streamer->base;
 #else
@@ -137,8 +137,8 @@ bbstreamer_lz4_compressor_content(bbstreamer *streamer,
 											 &mystreamer->prefs);
 
 		if (LZ4F_isError(compressed_size))
- pg_log_error("could not write lz4 header: %s",
-   LZ4F_getErrorName(compressed_size));
+ pg_fatal("could not write lz4 header: %s",
+					 LZ4F_getErrorName(compressed_size));
 
 		mystreamer->bytes_written += compressed_size;
 		mystreamer->header_written = true;
@@ -186,8 +186,8 @@ bbstreamer_lz4_compressor_content(bbstreamer *streamer,
 										  next_in, len, NULL);
 
 	if (LZ4F_isError(compressed_size))
- pg_log_error("could not compress data: %s",
-   LZ4F_getErrorName(compressed_size));
+ pg_fatal("could not compress data: %s",
+				 LZ4F_getErrorName(compressed_size));
 
 	mystreamer->bytes_written += compressed_size;
 }
@@ -238,8 +238,8 @@ bbstreamer_lz4_compressor_finalize(bbstreamer *streamer)
 									   next_out, avail_out, NULL);
 
 	if (LZ4F_isError(compressed_size))
- pg_log_error("could not end lz4 compression: %s",
-   LZ4F_getErrorName(compressed_size));
+ pg_fatal("could not end lz4 compression: %s",
+				 LZ4F_getErrorName(compressed_size));
 
 	mystreamer->bytes_written += compressed_size;
 
@@ -351,18 +351,21 @@ bbstreamer_lz4_decompressor_content(bbstreamer *streamer,
 							  next_in, &read_size, NULL);
 
 		if (LZ4F_isError(ret))
- pg_log_error("could not decompress data: %s",
-   LZ4F_getErrorName(ret));
+ pg_fatal("could not decompress data: %s",
+					 LZ4F_getErrorName(ret));
 
 		/* Update input buffer based on number of bytes consumed */
 		avail_in -= read_size;
 		next_in += read_size;
 
+		/* Update output buffer based on number of bytes produced */
+		avail_out -= out_size;
+		next_out += out_size;
 		mystreamer->bytes_written += out_size;
 
 		/*
 		 * If output buffer is full then forward the content to next streamer
-		 * and update the output buffer.
+		 * and reset the output buffer.
 		 */
 		if (mystreamer->bytes_written >= mystreamer->base.bbs_buffer.maxlen)
 		{
@@ -372,13 +375,8 @@ bbstreamer_lz4_decompressor_content(bbstreamer *streamer,
 							   context);
 
 			avail_out = mystreamer->base.bbs_buffer.maxlen;
-			mystreamer->bytes_written = 0;
 			next_out = (uint8 *) mystreamer->base.bbs_buffer.data;
-		}
-		else
-		{
-			avail_out = mystreamer->base.bbs_buffer.maxlen - mystreamer->bytes_written;
-			next_out += mystreamer->bytes_written;
+			mystreamer->bytes_written = 0;
 		}
 	}
 }
@@ -397,10 +395,11 @@ bbstreamer_lz4_decompressor_finalize(bbstreamer *streamer)
 	 * End of the stream, if there is some pending data in output buffers then
 	 * we must forward it to next streamer.
 	 */
-	bbstreamer_content(mystreamer->base.bbs_next, NULL,
-					   mystreamer->base.bbs_buffer.data,
-					   mystreamer->base.bbs_buffer.maxlen,
-					   BBSTREAMER_UNKNOWN);
+	if (mystreamer->bytes_written > 0)
+		bbstreamer_content(mystreamer->base.bbs_next, NULL,
+						   mystreamer->base.bbs_buffer.data,
+						   mystreamer->bytes_written,
+						   BBSTREAMER_UNKNOWN);
 
 	bbstreamer_finalize(mystreamer->base.bbs_next);
 }

fetools/pg17/pg_basebackup/bbstreamer_tar.c

@@ -224,7 +224,8 @@ bbstreamer_tar_parser_content(bbstreamer *streamer, bbstreamer_member *member,
 				/* OK, now we can send it. */
 				bbstreamer_content(mystreamer->base.bbs_next,
 								   &mystreamer->member,
-								   data, mystreamer->pad_bytes_expected,
+								   mystreamer->base.bbs_buffer.data,
+								   mystreamer->pad_bytes_expected,
 								   BBSTREAMER_MEMBER_TRAILER);
 
 				/* Expect next file header. */
@@ -290,6 +291,9 @@ bbstreamer_tar_header(bbstreamer_tar_parser *mystreamer)
 	strlcpy(member->pathname, &buffer[TAR_OFFSET_NAME], MAXPGPATH);
 	if (member->pathname[0] == '\0')
 		pg_fatal("tar member has empty name");
+	if (!path_is_safe_for_extraction(member->pathname))
+		pg_fatal("tar member has unsafe path name: \"%s\"",
+				 member->pathname);
 	member->size = read_tar_number(&buffer[TAR_OFFSET_SIZE], 12);
 	member->mode = read_tar_number(&buffer[TAR_OFFSET_MODE], 8);
 	member->uid = read_tar_number(&buffer[TAR_OFFSET_UID], 8);
@@ -342,6 +346,7 @@ bbstreamer_tar_parser_free(bbstreamer *streamer)
 {
 	pfree(streamer->bbs_buffer.data);
 	bbstreamer_free(streamer->bbs_next);
+	pfree(streamer);
 }
 
 /*

fetools/pg17/pg_basebackup/bbstreamer_zstd.c

@@ -112,11 +112,8 @@ bbstreamer_zstd_compressor_new(bbstreamer *next, pg_compress_specification *comp
 									 ZSTD_c_enableLongDistanceMatching,
 									 compress->long_distance);
 		if (ZSTD_isError(ret))
-		{
-			pg_log_error("could not enable long-distance mode: %s",
-						 ZSTD_getErrorName(ret));
-			exit(1);
-		}
+			pg_fatal("could not enable long-distance mode: %s",
+					 ZSTD_getErrorName(ret));
 	}
 
 	/* Initialize the ZSTD output buffer. */
@@ -178,8 +175,8 @@ bbstreamer_zstd_compressor_content(bbstreamer *streamer,
 								 &inBuf, ZSTD_e_continue);
 
 		if (ZSTD_isError(yet_to_flush))
- pg_log_error("could not compress data: %s",
-   ZSTD_getErrorName(yet_to_flush));
+ pg_fatal("could not compress data: %s",
+					 ZSTD_getErrorName(yet_to_flush));
 	}
 }
 
@@ -220,8 +217,8 @@ bbstreamer_zstd_compressor_finalize(bbstreamer *streamer)
 											&in, ZSTD_e_end);
 
 		if (ZSTD_isError(yet_to_flush))
- pg_log_error("could not compress data: %s",
-   ZSTD_getErrorName(yet_to_flush));
+ pg_fatal("could not compress data: %s",
+					 ZSTD_getErrorName(yet_to_flush));
 
 	} while (yet_to_flush > 0);
 
@@ -326,8 +323,8 @@ bbstreamer_zstd_decompressor_content(bbstreamer *streamer,
 									&mystreamer->zstd_outBuf, &inBuf);
 
 		if (ZSTD_isError(ret))
- pg_log_error("could not decompress data: %s",
-   ZSTD_getErrorName(ret));
+ pg_fatal("could not decompress data: %s",
+					 ZSTD_getErrorName(ret));
 	}
 }
 
@@ -346,7 +343,7 @@ bbstreamer_zstd_decompressor_finalize(bbstreamer *streamer)
 	if (mystreamer->zstd_outBuf.pos > 0)
 		bbstreamer_content(mystreamer->base.bbs_next, NULL,
 						   mystreamer->base.bbs_buffer.data,
-						   mystreamer->base.bbs_buffer.maxlen,
+						   mystreamer->zstd_outBuf.pos,
 						   BBSTREAMER_UNKNOWN);
 
 	bbstreamer_finalize(mystreamer->base.bbs_next);

fetools/pg17/pg_basebackup/pg_createsubscriber.c

@@ -1063,18 +1063,23 @@ drop_existing_subscriptions(PGconn *conn, const char *subname, const char *dbnam
 {
 	PQExpBuffer query = createPQExpBuffer();
 	PGresult   *res;
+	char	   *subname_esc;
 
 	Assert(conn != NULL);
 
+	subname_esc = PQescapeIdentifier(conn, subname, strlen(subname));
+
 	/*
 	 * Construct a query string. These commands are allowed to be executed
 	 * within a transaction.
 	 */
 	appendPQExpBuffer(query, "ALTER SUBSCRIPTION %s DISABLE;",
- subname);
+ subname_esc);
 	appendPQExpBuffer(query, " ALTER SUBSCRIPTION %s SET (slot_name = NONE);",
-					  subname);
-	appendPQExpBuffer(query, " DROP SUBSCRIPTION %s;", subname);
+					  subname_esc);
+	appendPQExpBuffer(query, " DROP SUBSCRIPTION %s;", subname_esc);
+
+	PQfreemem(subname_esc);
 
 	pg_log_info("dropping subscription \"%s\" in database \"%s\"",
 				subname, dbname);