zenpwnpython

zenpwnpy的使用教程

ZenDuk 2026-06-09

zenpwnpy只是一个我自用的python库，没啥特别的，面向pwn

项目地址：

https://github.com/Zenquiem/zenpwnpy

1. 安装

1.1 从 GitHub 安装

如果只是想使用，可以直接从 GitHub 安装：

python3 -m pip install git+https://github.com/Zenquiem/zenpwnpy.git

如果你希望在本地修改源码，建议克隆后editable安装：

git clone https://github.com/Zenquiem/zenpwnpy.git cd zenpwnpy python3 -m pip install -e .

-e 表示editable install。
这样安装后，你修改本地源码，不需要反复重新安装，Python会直接使用当前目录下的源码。

1.2 验证安装

安装完成后可以执行：

python3 - <<'PY' import zenpwnpy print("zenpwnpy import ok") print(len(zenpwnpy.__all__)) PY

如果没有报错，说明库已经可以正常导入。

1.3 运行测试

如果你是从源码目录使用，可以跑单元测试：

python3 -m unittest discover -s tests -v

当前测试覆盖了 PoW、payload layout、RISC-V 编码、shell upload 等主要功能。

2. 推荐导入方式

最推荐的方式是按需导入：

from zenpwnpy import command_pow, ShellUploader, PayloadLayout

也可以使用：

from zenpwnpy import *

不过需要注意：zenpwnpy 的 import * 只导出推荐的核心 API。
一些兼容旧脚本的别名仍然能直接导入，但不会出现在默认 __all__ 中。

例如下面这些是推荐名字：

from zenpwnpy import nonzero_mask
from zenpwnpy import load_imm32
from zenpwnpy import pack_instruction_words
from zenpwnpy import command_pow

下面这些是兼容旧脚本的别名，能用，但新脚本不建议优先使用：

from zenpwnpy import care_nonzero
from zenpwnpy import li32
from zenpwnpy import pack_words
from zenpwnpy import solve_command_pow

3. PoW 模块

PoW 模块主要解决题目开头常见的 proof-of-work 场景。

对应能力：

  • command_pow
  • hashcash_pow
  • is_hashcash_challenge
  • PowError
  • PowTimeoutError
  • PowCommandError

3.1 命令型 PoW：command_pow

很多pwn题开头会给你一条命令，例如：

run this command:
python3 -c 'print("answer")'

然后服务端要求你把命令的 stdout 发回去。

这类场景可以用 command_pow()：

io.recvuntil(b"run: ")
cmd = io.recvline().decode().strip()

answer = command_pow(cmd, timeout=20)
io.sendline(answer.encode())

command_pow() 的核心含义很简单：

本地执行一条命令 -> 取 stdout -> 返回字符串

默认行为：

  • 命令通过 shell 执行
  • 超时时间为 30 秒
  • 返回 stdout 去掉首尾空白后的结果
  • 命令返回非 0 会抛异常
  • 命令没有输出会抛异常

3.2 常用参数

answer = command_pow(
    cmd,
    timeout=30,
    shell=True,
    strip=True,
    check=True,
    require_output=True,
)

参数说明：

  • timeout：命令执行超时时间，单位秒
  • shell：是否通过 shell 执行
  • cwd：指定命令运行目录
  • env：指定环境变量
  • encoding：stdout/stderr 解码编码，默认 utf-8
  • errors：解码错误处理方式，默认 replace
  • strip：是否对 stdout 做 .strip()
  • check：返回码非 0 是否报错
  • require_output：stdout 为空是否报错

3.3 保留原始输出

如果题目需要保留输出末尾换行，可以设置：

answer = command_pow(cmd, strip=False)

默认 strip=True，适合大多数 PoW 回答，因为服务端通常只要那一段 token。

3.4 不要求输出

如果你只是想跑一条命令，不关心输出，可以设置：

command_pow("true", require_output=False)

不过这种用法不是 PoW 的主场景。

3.5 捕获失败信息

当命令执行失败时，会抛 PowCommandError：

from zenpwnpy import command_pow, PowCommandError

try:
    answer = command_pow("false")
except PowCommandError as exc:
    print("returncode:", exc.returncode)
    print("stdout:", exc.stdout)
    print("stderr:", exc.stderr)

这比简单地 subprocess.check_output() 更适合做题调试，因为失败时 stdout/stderr 都能拿到。

3.6 超时处理

命令超时时会抛 PowTimeoutError：

from zenpwnpy import command_pow, PowTimeoutError

try:
    answer = command_pow("sleep 999", timeout=1)
except PowTimeoutError:
    print("PoW timeout")

3.7 Hashcash PoW：hashcash_pow

有些题直接输出 hashcash 命令，例如：

hashcash -mb29 abcdef

这类可以用：

from zenpwnpy import hashcash_pow

challenge = io.recvline().decode().strip()
token = hashcash_pow(challenge, timeout=30)
io.sendline(token.encode())

hashcash_pow() 会先检查输入是不是看起来像 hashcash 命令。
如果不是，会抛 PowCommandError。

3.8 判断是否为 hashcash challenge

from zenpwnpy import is_hashcash_challenge

if is_hashcash_challenge(challenge):
    token = hashcash_pow(challenge)

它只做轻量判断：命令的第一个部分名字里包含 hashcash。

3.9 指定本地 hashcash 路径

如果远端给的是：

hashcash -mb29 abcdef

但你本机 hashcash 不在 PATH，或者你想使用自己编译的版本，可以：

token = hashcash_pow(challenge, binary="/usr/bin/hashcash")

这会替换命令中的第一个参数。

4. PayloadLayout：稀疏约束的 payload 重叠布局

它解决的问题是：
做 fake structure 时，我们经常会构造多个逻辑结构，但它们实际可以在内存中重叠，因为很多 padding 字节并没有语义约束。

传统写法可能是：

payload = flat({ 0x00: b"...", 0x88: p64(lock), 0xa0: p64(wide_data), 0xd8: p64(vtable), })

但是如果结构越来越多，并且想让它们自动找一个更短的重叠布局，手写 offset 会变得很难维护。

PayloadLayout 的核心思想是：

只有显式写入的字节才算约束。 没有写入的字节，不自动视为必须是 0。

这和普通 dense payload 很不一样。

4.1 最小例子

from zenpwnpy import PayloadLayout

layout = PayloadLayout(base=0x404000)

a = layout.fragment("a", size=0x20, offset=0)
a.write(0x00, b"AAAA")

b = layout.fragment("b", size=0x20, align=8)
b.write(0x10, b"BBBB")

payload = layout.build(search_end=0x40)
print(payload)

这里：

  • a 是固定片段，放在 offset 0
  • b 是可重定位片段，按 align=8 搜索位置
  • search_end=0x40 表示搜索范围最多到 0x40

如果 b 的约束字节和 a 不冲突，它可能会和 a 重叠。

4.2 创建 layout

layout = PayloadLayout(base=heap_base)

base 表示最终 payload 对应的基址。
它主要影响 write_ptr() 和 absolute_address_of()。

例如某个片段最终 offset 是 0x80，base 是 0x404000，那么它的绝对地址就是：

0x404000 + 0x80 = 0x404080

4.3 创建 fragment

frag = layout.fragment( "fake_file", size=0xe0, offset=0, )

参数说明：

  • name：片段名字，必须唯一
  • size：片段逻辑大小
  • offset：固定偏移；如果不传，表示可重定位
  • align：可重定位时的对齐
  • candidates：可选候选偏移列表

固定片段：

layout.fragment("base", size=0x100, offset=0)

可重定位片段：

layout.fragment("wide_data", size=0xe8, align=8)

只允许某些位置：

layout.fragment("node", size=0x40, candidates=[0x20, 0x80, 0x100])

4.4 写入普通字节

frag.write(0x00, b"/bin/sh\x00")

含义是：
在这个 fragment 的相对 offset 0x00 写入这些字节。

如果同一个 fragment 内部对同一位置写入不同字节，会抛 LayoutConflictError。

4.5 写入整数

写入 64 位整数：

frag.write_u64(0x88, lock_addr)

写入任意大小整数：

frag.write_int(0x10, 0x1234, size=2)
frag.write_int(0x20, 0x12345678, size=4)
frag.write_int(0x30, 0x1234567890abcdef, size=8)

默认小端：

frag.write_int(0x10, 0x1234, size=2, endian="little")

如果需要大端：

frag.write_int(0x10, 0x1234, size=2, endian="big")

4.6 写入指向其他 fragment 的指针

这是 PayloadLayout 很实用的地方。

fake_file.write_ptr(0xa0, "wide_data")

含义是：
等 layout 求解完成后，把 wide_data 的最终地址写到 fake_file + 0xa0。

最终写入的值是：

layout.base + offset_of("wide_data") + addend

可以加偏移：

fake_file.write_ptr(0xa0, "wide_data", addend=0x20)

也可以指定指针大小：

fake_file.write_ptr(0x20, "target", size=4)

4.7 求解布局

placements = layout.solve(search_end=0x300)

返回一个字典：

{ "fake_file": 0, "wide_data": 0, "wide_vtable": 0, }

如果有可重定位 fragment，并且没有提供 candidates，就需要传 search_end。
否则库不知道搜索范围。

4.8 构造最终 payload

payload = layout.build(search_end=0x300)

默认未约束的字节填 0。
如果你希望填充为其他字节：

payload = layout.build(fill=0x41, search_end=0x300)

4.9 查询片段位置

查询相对偏移：

off = layout.offset_of("wide_data", search_end=0x300)

查询绝对地址：

addr = layout.absolute_address_of("wide_data", search_end=0x300)

查询所有 placement：

for placement in layout.placements(search_end=0x300):
    print(placement.name, hex(placement.offset), hex(placement.end), placement.fixed)

查询最终 payload 长度：

length = layout.end_offset(search_end=0x300)

4.10 使用 nonzero_mask

有时候我们已经有一段旧的 dense payload：

old_payload = b"\x00" * 0x18 + p64(0xdeadbeef)

其中很多 \x00 只是 padding，不想让它们参与冲突判断。
可以用：

from zenpwnpy import PayloadLayout, nonzero_mask

layout = PayloadLayout(base=0x404000)

overlay = layout.fragment("overlay", size=len(old_payload), align=8)
overlay.write(0, old_payload, mask=nonzero_mask(old_payload))

nonzero_mask() 会返回一个同长度 mask：

字节非 0 -> 1，表示参与约束 字节为 0 -> 0，表示忽略这个位置

例子：

nonzero_mask(b"\x00A\x00B")

结果：

b"\x00\x01\x00\x01"

注意：
如果某个字段语义上必须为 0，不要依赖 nonzero_mask()，应该显式写入：

frag.write_u64(0x30, 0)

4.11 一个 fake structure 风格例子

下面是一个接近真实做题风格的例子：

from zenpwnpy import PayloadLayout

heap_base = 0x555555800000
fake_io = heap_base + 0x2e0
system_addr = 0x7ffff7e12345
lock_addr = heap_base + 0x100
wfile_jumps = 0x7ffff7fb0000

layout = PayloadLayout(base=fake_io)

fake_file = layout.fragment("fake_file", size=0xe0, offset=0)
fake_file.write(0x00, b"  sh;".ljust(8, b"\x00"))
fake_file.write(0x88, p64(lock_addr))
fake_file.write_ptr(0xa0, "wide_data")
fake_file.write(0xd8, p64(wfile_jumps))

wide_data = layout.fragment("wide_data", size=0xe8, align=8)
wide_data.write_ptr(0xe0, "wide_vtable")

wide_vtable = layout.fragment("wide_vtable", size=0x70, align=8)
wide_vtable.write(0x68, p64(system_addr))

payload = layout.build(search_end=0x300)

print("wide_data offset:", hex(layout.offset_of("wide_data")))
print("wide_vtable addr:", hex(layout.absolute_address_of("wide_vtable")))
print("payload length:", hex(len(payload)))

这个例子的重点不是具体 glibc 结构，而是写法：

  • 每个逻辑结构用一个 fragment 表示
  • 指针关系用 write_ptr() 表示
  • 最终由 layout 自动决定可重定位片段放在哪里
  • 只对真正写入的字节做冲突检查

4.12 常见错误

如果可重定位 fragment 没有 candidates，且调用时没有传 search_end：

layout.build()

可能会抛：

ValueError: fragment 'xxx' is relocatable; solve() needs search_end or candidates

解决：

layout.build(search_end=0x300)

如果两个约束字节冲突，会抛 LayoutConflictError：

from zenpwnpy import LayoutConflictError

try:
    payload = layout.build(search_end=0x300)
except LayoutConflictError as exc:
    print("relative offset:", exc.relative_offset)
    print("absolute offset:", exc.absolute_offset)
    print("fragments:", exc.fragments)

如果找不到兼容布局，会抛 LayoutSolveError。

5. ShellUploader：只有 shell 时上传文件

这是另一个很实用的模块。

它适合这种场景：

  • 已经拿到远程交互式 shell
  • 远程没有 scp
  • 远程没有 wget
  • 远程没有 curl
  • 远程没有 Python
  • 但远程有基础 shell 命令，例如 cat、base64、chmod、gzip

这时可以通过 shell 把本地文件传上去。

5.1 基本用法

from zenpwnpy import ShellUploader

io = remote("example.com", 31337)

io.recvuntil(b"$ ")

uploader = ShellUploader(io)
result = uploader.upload("./exp", "/home/pwn/exp", chmod_x=True)

io.sendline(b"/home/pwn/exp")
io.interactive()

5.2 推荐参数

远程 TTY shell 比较怪时，可以显式设置：

uploader = ShellUploader(
    io,
    chunk_size=0x100,
    commands_per_sync=16,
    max_batch_bytes=0xc00,
    command_timeout=30,
    decode_timeout=120,
    upload_method="cat_eof",
    suppress_echo=True,
    cat_eof_settle_time=0.2,
)

这些参数的含义：

  • chunk_size：base64 文本分块大小
  • commands_per_sync：append 模式下每轮同步的命令数量
  • max_batch_bytes：单行命令最大估算长度
  • command_timeout：普通远程命令等待超时
  • decode_timeout：base64/gzip 解码等待超时
  • upload_method：上传方法
  • suppress_echo：是否尝试 stty -echo 抑制回显
  • cat_eof_settle_time：Ctrl-D 后等待 shell 返回 prompt 的时间

5.3 上传模式

当前支持三种上传模式：

cat_eof
heredoc
append

默认是：

upload_method="cat_eof"

cat_eof

通过：

cat > remote_file

然后发送 base64 文本，最后发送 Ctrl-D 结束。

优点：

  • 对很多交互式 shell 更稳
  • 不依赖复杂 shell 语法
  • 适合回显很烦的 TTY 场景

缺点：

  • 需要目标 shell 能正确处理 Ctrl-D

heredoc

通过 here-doc 一次写入：

cat > remote_file <<'EOF'
...
EOF

优点：

  • 思路简单
  • 非交互环境里通常不错

缺点：

  • 某些题目 shell 会回显、插 prompt、处理 here-doc 不稳定

append

通过多条命令追加：

printf %s 'chunk' >> remote_file

优点：

  • 不依赖 Ctrl-D

缺点：

  • 往返次数更多
  • 命令行长度和特殊字符处理更敏感

5.4 压缩上传

如果本地文件较大，可以：

uploader.upload("./exp", "/home/pwn/exp", chmod_x=True, compress=True)

这会在本地先 gzip 压缩，再 base64 传输，远程执行：

gzip -dc temp.gz > final_file

要求远程有 gzip。

5.5 自动 chmod

uploader.upload("./exp", "/home/pwn/exp", chmod_x=True)

上传后会执行：

chmod +x /home/pwn/exp

要求远程有 chmod。

5.6 大小校验

默认 verify=True。
上传完成后会用：

wc -c < remote_file

检查远程文件大小是否和本地一致。

如果目标环境没有 wc，可以关掉：

uploader.upload("./exp", "/home/pwn/exp", verify=False)

5.7 清理临时文件

默认 cleanup=True。
成功后会删除传输过程中产生的 .b64、.gz 临时文件。

如果你想调试远端中间文件：

uploader.upload("./exp", "/home/pwn/exp", cleanup=False)

5.8 等待初始 prompt

有些服务连接后会先输出启动日志，shell prompt 不是立刻出现。
可以让 ShellUploader 等 prompt：

uploader = ShellUploader(
    io,
    initial_prompt=b"~ $ ",
    initial_prompt_timeout=20,
    initial_prompt_settle_time=0.2,
    auto_probe=True,
)

参数说明：

  • initial_prompt：要等待的 prompt
  • initial_prompt_timeout：等待 prompt 的超时时间
  • initial_prompt_settle_time：看到 prompt 后再稍等一会，吃掉残留输出
  • auto_probe：初始化时自动执行一次轻量探活

5.9 手动执行远程命令

ShellUploader 内部有远程命令同步机制，也可以拿来执行简单命令：

output = uploader.run("pwd; id; ls -la", check=False)
print(output.decode(errors="ignore"))

如果 check=True，远程命令返回非 0 会抛 ShellCommandError。

from zenpwnpy import ShellCommandError

try:
    uploader.run("false")
except ShellCommandError as exc:
    print("status:", exc.status)
    print("output:", exc.output.decode(errors="ignore"))

5.10 批量执行远程命令

uploader.run_batch([
    "cd /home/pwn",
    "chmod +x exp",
    "./exp",
])

run_batch() 会在远端按顺序执行多条命令，并用一个完成标记同步。
如果中间某条失败，异常里会包含失败的序号和命令。

5.11 常见坑：/tmp 不一定可写

很多人习惯上传到：

/tmp/exp

但 CTF 题目环境里 /tmp 不一定是 1777 权限。
有些内核题、qemu 题、busybox rootfs 里，/tmp 可能是 root 拥有且普通用户不可写。

如果看到类似：

Permission denied

应该先检查：

print(uploader.run("pwd; id; ls -ld . /tmp /dev/shm /home/pwn 2>&1", check=False).decode(errors="ignore"))

优先选择当前用户可写目录，例如：

uploader.upload("./exp", "./exp", chmod_x=True)

或者：

uploader.upload("./exp", "/home/pwn/exp", chmod_x=True)

5.12 一个完整上传片段

from zenpwnpy import ShellUploader, ShellCommandError

io.recvuntil(b"$ ")

uploader = ShellUploader(
    io,
    chunk_size=0x100,
    commands_per_sync=16,
    max_batch_bytes=0xc00,
    command_timeout=30,
    decode_timeout=120,
    upload_method="cat_eof",
    suppress_echo=True,
    cat_eof_settle_time=0.2,
)

try:
    result = uploader.upload(
        "./exp",
        "/home/pwn/exp",
        chmod_x=True,
        compress=True,
    )
    log.success(f"uploaded {result.remote_path}, sha256={result.sha256}")
except ShellCommandError as exc:
    log.failure(str(exc))
    log.failure(exc.output.decode(errors="ignore"))
    raise

io.sendline(b"/home/pwn/exp")
io.interactive()

6. RISC-V 指令编码

zenpwnpy.riscv 提供一组轻量 RISC-V 指令编码 helper。

它不是完整汇编器。
它适合你在题目里需要直接构造 32-bit 指令字，但又不想引入完整 assembler 的场景。

6.1 基础导入

from zenpwnpy import (
    r_type,
    i_type,
    s_type,
    b_type,
    u_type,
    j_type,
    addi,
    lw,
    sw,
    jal,
    jalr,
    load_imm32,
    pack_instruction_words,
)

6.2 底层格式编码器

RISC-V 基础 32-bit 指令格式包括：

  • R-type
  • I-type
  • S-type
  • B-type
  • U-type
  • J-type

对应函数：

r_type(opcode, rd, funct3, rs1, rs2, funct7=0)
i_type(opcode, rd, funct3, rs1, imm)
s_type(opcode, funct3, rs1, rs2, imm)
b_type(opcode, funct3, rs1, rs2, imm)
u_type(opcode, rd, imm)
j_type(opcode, rd, imm)

例子：

word = r_type(0x33, 5, 0x0, 6, 7) print(hex(word))

6.3 常用指令 helper

当前提供：

lui(rd, imm)
auipc(rd, imm)
addi(rd, rs1, imm)
lw(rd, offset, rs1)
sw(rs2, offset, rs1)
jal(rd, offset)
jalr(rd, offset, rs1)

例子：

words = [
    addi(5, 0, 123),
    lw(6, 0, 5),
    sw(6, 4, 5),
    jalr(0, 0, 1),
]

6.4 分支指令

当前提供：

beq(rs1, rs2, offset)
bne(rs1, rs2, offset)
blt(rs1, rs2, offset)
bge(rs1, rs2, offset)
bltu(rs1, rs2, offset)
bgeu(rs1, rs2, offset)

注意：
这里的 offset 是字节偏移，不是你手动右移后的 immediate 字段。

例如：

word = beq(5, 6, 16)

表示：

beq x5, x6, +16

分支和跳转偏移需要 2 字节对齐。
如果传入奇数偏移，会抛 ValueError。

6.5 32 位常量装载：load_imm32

RISC-V 里装载一个 32 位常量通常可以用：

lui
addi

load_imm32() 会帮你生成这两条指令：

from zenpwnpy import load_imm32

words = load_imm32(5, 0x12345678)

含义是把 0x12345678 装进 x5。

内部会做：

upper = ((value + 0x800) >> 12)
lower = value - (upper << 12)

这和常见 lui + addi 拆分方式一致。

也可以单独拆分：

from zenpwnpy import split_imm32

upper, lower = split_imm32(0x12345678)

边界说明：

  • load_imm32() 只承诺处理 32 位常量装载
  • 它不是完整的 li 伪指令展开器
  • 不覆盖 RV64 下所有更复杂的常量装载策略

6.6 打包指令字

RISC-V 指令通常按 little-endian 字节序放进 payload：

from zenpwnpy import pack_instruction_words

payload = pack_instruction_words([
    0x12345678,
    0xaabbccdd,
])

结果是：

b"\x78\x56\x34\x12\xdd\xcc\xbb\xaa"

6.7 一个完整 RISC-V payload 例子

from zenpwnpy import (
    load_imm32,
    lw,
    sw,
    jalr,
    pack_instruction_words,
)

words = []

# x5 = 0x12345678
words.extend(load_imm32(5, 0x12345678))

# x6 = *(uint32_t *)x5
words.append(lw(6, 0, 5))

# *(uint32_t *)(x5 + 4) = x6
words.append(sw(6, 4, 5))

# ret: jalr x0, 0(x1)
words.append(jalr(0, 0, 1))

payload = pack_instruction_words(words)
print(payload.hex())

7. 错误处理风格

zenpwnpy 的异常大多是模块级异常，适合在 exploit 里做精确捕获。

7.1 PoW 异常

from zenpwnpy import PowError, PowTimeoutError, PowCommandError

层级大致是：

PowError
  PowTimeoutError
  PowCommandError

用法：

try:
    answer = command_pow(cmd)
except PowTimeoutError:
    print("timeout")
except PowCommandError as exc:
    print(exc.stderr)

7.2 Layout 异常

from zenpwnpy import PayloadLayoutError, LayoutConflictError, LayoutSolveError

层级：

PayloadLayoutError
  LayoutConflictError
  LayoutSolveError

LayoutConflictError 里可能包含：

  • absolute_offset
  • relative_offset
  • fragments

7.3 Shell 上传异常

from zenpwnpy import (
    ShellUploadError,
    ShellLocalFileError,
    ShellSyncError,
    ShellMissingCommandError,
    ShellVerificationError,
    ShellCommandError,
)

常见含义：

  • ShellLocalFileError：本地文件不存在或读不了
  • ShellSyncError：远端 shell 同步失败
  • ShellMissingCommandError：远端缺少必要命令
  • ShellVerificationError：上传后校验失败
  • ShellCommandError：远端命令返回非 0

捕获远端命令失败：

try:
    uploader.upload("./exp", "/home/pwn/exp")
except ShellCommandError as exc:
    print("command:", exc.command)
    print("status:", exc.status)
    print("context:", exc.context)
    print(exc.output.decode(errors="ignore"))

评论