x = torch.from_numpy(np.load('x_adv.npy')).unsqueeze(0)
logits = clf(x)
probs = torch.softmax(logits, dim=-1)[0]
for i, name in enumerate(classes):
    print(f'{i}: {name:12s} prob={probs[i]:.4f}, logp={torch.log(probs[i]):.4f}')

sampler = GuidedDDIM(unet, sched, device='cpu', steps=100)
guidance = {'clf': clf, 'target': 3, 'scale': 3.0}
x_T = sampler.invert_fixedpoint(x_adv, guidance=guidance, iters=3)

rng = np.random.default_rng(20261012)
mask = rng.choice([-1.0, 1.0], size=(3, 32, 32))
unmasked = (x_T - G) * mask

██████████████  ████  ██  ████  ██  ██████████████
██          ██  ██    ████  ██  ██  ██          ██
██  ██████  ██  ██      ██  ██  ██  ██  ██████  ██
██  ██████  ██      ██      ██  ██  ██  ██████  ██
██  ██████  ██    ██      ████  ██  ██  ██████  ██
██          ██  ██████████      ██  ██          ██
██████████████  ██  ██  ██  ██  ██  ██████████████
                ██  ██      ██
    ██████  ██  ████        ██    ██████    ██████
████      ██  ██  ██    ████  ████    ██  ██  ██
    ██  ██  ██  ██  ██    ██    ████  ████  ██  ██
  ██    ██      ████  ██  ██  ██      ██████
██  ██  ██████  ██████    ██        ████    ██  ██
████      ██      ██    ████████  ██  ██      ██
██  ██████████████      ████  ████  ██  ████  ████
██    ██  ██    ████  ████  ██      ████        ██
██  ██  ████████  ████████  ██████████████████████
                ████████████    ██      ██  ██
██████████████          ██      ██  ██  ██    ████
██          ██      ████████  ████      ██
██  ██████  ██  ████    ██    ████████████    ██
██  ██████  ██  ██      ████    ██  ██  ██  ██  ██
██  ██████  ██  ████      ██████  ██  ██    ██  ██
██          ██          ██      ██  ██    ██    ██
██████████████    ██████  ██      ████      ██████

dr34m_0f_n01s3

import os
import sys
import numpy as np
import torch
import torch.nn as nn
from PIL import Image
from diffusers import UNet2DModel, DDPMScheduler
import zxingcpp

# ----------------- 1. 定义 ResNet-18 分类器网络 -----------------
def conv3x3(in_planes, out_planes, stride=1):
    return nn.Conv2d(in_planes, out_planes, kernel_size=3, stride=stride, padding=1, bias=False)

def conv1x1(in_planes, out_planes, stride=1):
    return nn.Conv2d(in_planes, out_planes, kernel_size=1, stride=stride, bias=False)

class BasicBlock(nn.Module):
    expansion = 1
    def __init__(self, inplanes, planes, stride=1, downsample=None):
        super().__init__()
        self.conv1 = conv3x3(inplanes, planes, stride)
        self.bn1 = nn.BatchNorm2d(planes)
        self.relu = nn.ReLU(inplace=True)
        self.conv2 = conv3x3(planes, planes)
        self.bn2 = nn.BatchNorm2d(planes)
        self.downsample = downsample
        self.stride = stride

    def forward(self, x):
        identity = x
        out = self.relu(self.bn1(self.conv1(x)))
        out = self.bn2(self.conv2(out))
        if self.downsample isnotNone:
            identity = self.downsample(x)
        return self.relu(out + identity)

class ResNet(nn.Module):
    def __init__(self, block, layers, num_classes=10):
        super().__init__()
        self.inplanes = 64
        self.conv1 = nn.Conv2d(3, 64, kernel_size=3, stride=1, padding=1, bias=False)
        self.bn1 = nn.BatchNorm2d(64)
        self.relu = nn.ReLU(inplace=True)
        self.maxpool = nn.Identity()
        self.layer1 = self._make_layer(block, 64, layers[0])
        self.layer2 = self._make_layer(block, 128, layers[1], stride=2)
        self.layer3 = self._make_layer(block, 256, layers[2], stride=2)
        self.layer4 = self._make_layer(block, 512, layers[3], stride=2)
        self.avgpool = nn.AdaptiveAvgPool2d((1, 1))
        self.fc = nn.Linear(512 * block.expansion, num_classes)

    def _make_layer(self, block, planes, blocks, stride=1):
        downsample = None
        if stride != 1or self.inplanes != planes * block.expansion:
            downsample = nn.Sequential(
                conv1x1(self.inplanes, planes * block.expansion, stride),
                nn.BatchNorm2d(planes * block.expansion),
            )
        layers = [block(self.inplanes, planes, stride, downsample)]
        self.inplanes = planes * block.expansion
        for _ in range(1, blocks):
            layers.append(block(self.inplanes, planes))
        return nn.Sequential(*layers)

    def forward(self, x):
        x = self.maxpool(self.relu(self.bn1(self.conv1(x))))
        x = self.layer4(self.layer3(self.layer2(self.layer1(x))))
        return self.fc(torch.flatten(self.avgpool(x), 1))

class CifarResNet18(nn.Module):
    def __init__(self):
        super().__init__()
        self.net = ResNet(BasicBlock, [2, 2, 2, 2], num_classes=10)
        self.register_buffer(\"mean\", torch.tensor([0.4914, 0.4822, 0.4465]).view(1, 3, 1, 1))
        self.register_buffer(\"std\", torch.tensor([0.2470, 0.2435, 0.2616]).view(1, 3, 1, 1))

    def forward(self, x):
        return self.net(((x + 1.0) / 2.0 - self.mean) / self.std)

# ----------------- 2. DDIM 导引逆向反演采样器 -----------------
class GuidedDDIM:
    def __init__(self, unet, scheduler, device, num_train=1000, steps=100):
        self.unet = unet
        self.device = device
        self.ab = scheduler.alphas_cumprod.to(device=device, dtype=torch.float32)
        self.ts = np.round(np.arange(0, steps) * (num_train / steps)).astype(int)

    def _eps(self, x, t, guidance):
        ab_t = self.ab[t]
        with torch.enable_grad():
            xg = x.detach().requires_grad_(True)
            eps = self.unet(xg, t).sample
            x0 = (xg - (1.0 - ab_t).sqrt() * eps) / ab_t.sqrt()
            logp = torch.log_softmax(guidance[\"clf\"](x0.clamp(-1.0, 1.0)), dim=-1)
            g = torch.autograd.grad(logp[:, guidance[\"target\"]].sum(), xg)[0]
        return (eps - guidance[\"scale\"] * (1.0 - ab_t).sqrt() * g).detach()

    def invert_fixedpoint(self, x0_img, guidance, iters=3):
        with torch.enable_grad():
            x = x0_img
            ab_prev = torch.ones((), device=self.device)
            for i in range(len(self.ts)):
                t = int(self.ts[i])
                ab_t = self.ab[t]
                A = (ab_prev / ab_t).sqrt()
                B = (1.0 - ab_prev).sqrt() - A * (1.0 - ab_t).sqrt()
                xt = x.clone()
                for _ in range(iters):
                    eps = self._eps(xt, t, guidance)
                    xt = (x - B * eps) / A
                x = xt
                ab_prev = ab_t
        return x.detach()

# ----------------- 3. 主解题逻辑 -----------------
def main():
    torch.set_num_threads(12)
    device = \"cpu\"
    challenge_dir = \"C:/Users/34645/Downloads/uae/uae\"
    model_dir = os.path.join(challenge_dir, \"ddpm-cifar10-32\")

    # 1. 载入模型与分类器
    unet = UNet2DModel.from_pretrained(model_dir)
    for p in unet.parameters(): p.requires_grad = False
    unet.eval()

    sched = DDPMScheduler.from_pretrained(model_dir)

    clf = CifarResNet18()
    clf.load_state_dict(torch.load(os.path.join(challenge_dir, \"classifier.pt\"), map_location=device))
    for p in clf.parameters(): p.requires_grad = False
    clf.eval()

    # 2. 读取对抗图像并执行不动点反演
    x_adv = torch.from_numpy(np.load(os.path.join(challenge_dir, \"x_adv.npy\"))).unsqueeze(0)
    guidance = {\"clf\": clf, \"target\": 3, \"scale\": 3.0}

    sampler = GuidedDDIM(unet, sched, device=device, steps=100)
    print(\"[*] Running Guided DDIM fixed-point inversion...\")
    x_T = sampler.invert_fixedpoint(x_adv, guidance=guidance, iters=3).numpy()[0]

    # 3. 构造载波与掩码，完成残差解调
    G = np.random.default_rng(555).standard_normal((3, 32, 32))
    mask = np.random.default_rng(20261012).choice([-1.0, 1.0], size=(3, 32, 32))
    unmasked = (x_T - G) * mask

    # 4. 2x2 空间块聚合并重构 25x25 QR 矩阵
    # 空间块重排: (3, 32, 32) -> (3, 16, 2, 16, 2) -> (768, 4)
    blocks = unmasked.reshape(3, 16, 2, 16, 2).transpose(0, 1, 3, 2, 4).reshape(768, 4)[:625]
    module_values = blocks.mean(axis=1).reshape(25, 25)

    # 符号判决: 负值为深色模块(Dark/0), 正值为浅色模块(Light/255)
    dark_modules = module_values < 0

    # 5. 生成标准二维码并扫码
    border = 4
    qr_img_arr = np.pad(np.where(dark_modules, 0, 255).astype(np.uint8), border, constant_values=255)
    qr_img = Image.fromarray(qr_img_arr).resize(((25 + 2 * border) * 10, (25 + 2 * border) * 10), Image.NEAREST)
    qr_path = \"recovered_qr.png\"
    qr_img.save(qr_path)
    print(f\"[*] Reconstructed QR saved to {qr_path}\")

    barcodes = zxingcpp.read_barcodes(qr_img)
    for b in barcodes:
        print(f\"[+] Decoded Secret ({b.format}): {b.text}\")

if __name__ == \"__main__\":
    main()

题目描述：镜像取证
附件：shadowread.zip

你看到的是影子。

cat 没有骗你,骗你的是 read。真的东西我搬走了,
磁盘上这份 flag.txt 是留给你们的安慰奖——
它确实能解出一面 flag,但那面是假的,别浪费提交次数。

原来的文件?删了,早就删了。
不过这块盘子记性不太好,删掉的东西,它自己还会念叨一阵子。

钥匙也删了——碎成四瓣,跟那几瓣缓存躺在一起。
哪瓣是真的?我的垫片记得,可惜它也跟着失忆了。

对了,那文件我一共写过两版——草稿和正稿我分得很清,希望你也是。

—— ph4nt0m

2026-10-11T03:13:02+08:00 portal-srv sudo:      ctf : TTY=pts/0 ; PWD=/home/ctf ; USER=root ; COMMAND=/usr/bin/insmod /usr/local/bin/kcompat.ko
2026-10-11T03:13:44+08:00 portal-srv sudo:      ctf : TTY=pts/0 ; PWD=/home/ctf ; USER=root ; COMMAND=/bin/cat /home/ctf/flag.txt
2026-10-11T03:14:02+08:00 portal-srv sudo:      ctf : TTY=pts/0 ; PWD=/home/ctf ; USER=root ; COMMAND=/bin/rm /var/.cache/.kmod/.real_cache.bak
2026-10-11T03:14:20+08:00 portal-srv sudo:      ctf : TTY=pts/0 ; PWD=/home/ctf ; USER=root ; COMMAND=/bin/rm /var/.cache/.kmod/.real_cache
2026-10-11T03:14:35+08:00 portal-srv sudo:      ctf : TTY=pts/0 ; PWD=/home/ctf ; USER=root ; COMMAND=/bin/rm -f /var/.cache/.kmod/.s* /var/.cache/.kmod/.k*

Journal starts at block 0, transaction 7

/var/.cache/.kmod/.k
/var/.cache/.kmod/.s3
/var/.cache/.kmod/.s2
/var/.cache/.kmod/.s1
/var/.cache/.kmod/.real_cache
/home/ctf/flag.txt

rsp+0x10  第一份 4 字节
rsp+0x14  第二份 4 字节
rsp+0x18  第三份 4 字节

3f9a2c71 b6567e02 d8c10f99 27e6a450

a65dc30a337a3faee5a58c868fc3e715e3b9cbcd45c9fcec9fd1f00a1f4a2371

mov    ebx, eax
movabs rax, 0x5a5a5a5a5a5a5a5a
xor    rbx, rax

((0xa65dc30a ^ 0x5a5a5a5a) | (0x5a5a5a5a << 32))

state = state * 0x5851f42d4c957f2d + 0x14057b7ef767814f

b = ror8((b - table[ebx % 20]) & 0xff, i)
b ^= table[i % 20]
ebx += 7

flag{r34d_h00k_d3crypt5_3v3ryth1ng}

#!/usr/bin/env python3
\"\"\"Recover the real flag from the shadowread ext4 image.

The module's default dial is high enough to ignore flag.txt and decrypt
/var/.cache/.kmod/.real_cache. Deleted cache inodes are still allocated, so
their bytes can be read without replaying the journal.
\"\"\"

import argparse
import hashlib
import pathlib
import struct

DEFAULT_IMAGE = pathlib.Path(r\"C:\\Users\\34645\\Downloads\\shadowread\\shadowread.img\")

class Ext4:
    def __init__(self, blob: bytes) -> None:
        self.blob = blob
        superblock = blob[1024:2048]
        self.block_size = 1024 << struct.unpack_from(\"<I\", superblock, 0x18)[0]
        self.inodes_per_group = struct.unpack_from(\"<I\", superblock, 0x28)[0]
        self.inode_size = struct.unpack_from(\"<H\", superblock, 0x58)[0]
        incompat = struct.unpack_from(\"<I\", superblock, 0x60)[0]
        self.desc_size = struct.unpack_from(\"<H\", superblock, 0xFE)[0] if incompat & 0x80else32
        self.volume = superblock[0x78:0x88].split(b\"\\x00\", 1)[0].decode()

    def inode_raw(self, number: int) -> tuple[bytes, int]:
        group = (number - 1) // self.inodes_per_group
        index = (number - 1) % self.inodes_per_group
        desc_off = self.block_size + group * self.desc_size
        desc = self.blob[desc_off:desc_off + self.desc_size]
        table = struct.unpack_from(\"<I\", desc, 8)[0]
        if self.desc_size >= 64:
            table |= struct.unpack_from(\"<I\", desc, 40)[0] << 32
        off = table * self.block_size + index * self.inode_size
        raw = self.blob[off:off + self.inode_size]
        size = struct.unpack_from(\"<I\", raw, 4)[0]
        if self.inode_size >= 0x70:
            size |= struct.unpack_from(\"<I\", raw, 0x6C)[0] << 32
        return raw, size

    def read_inode(self, number: int) -> bytes:
        raw, size = self.inode_raw(number)
        magic, entries, _, depth, _ = struct.unpack_from(\"<HHHHI\", raw, 0x28)
        if magic != 0xF30Aor depth != 0:
            raise RuntimeError(f\"inode {number}: unexpected extent header {magic:#x} depth {depth}\")
        out = bytearray()
        for entry in range(entries):
            logical, length, start_hi, start_lo = struct.unpack_from(\"<IHHI\", raw, 0x34 + entry * 12)
            del logical
            start = start_lo | (start_hi << 32)
            for block in range(length & 0x7FFF):
                at = (start + block) * self.block_size
                out += self.blob[at:at + self.block_size]
                if len(out) >= size:
                    return bytes(out[:size])
        return bytes(out[:size])

def directory_entries(data: bytes) -> list[tuple[int, str, bool]]:
    \"\"\"Return (inode, name, deleted). Deleted names survive inside an enlarged rec_len.\"\"\"
    found = []
    position = 0
    while position + 8 <= len(data):
        inode, rec_len, name_len, _ = struct.unpack_from(\"<IHBB\", data, position)
        if rec_len < 8or position + rec_len > len(data):
            break
        if inode and name_len:
            found.append((inode, data[position + 8:position + 8 + name_len].decode(\"latin1\"), False))
        slack = position + ((8 + name_len + 3) & ~3)
        end = position + rec_len
        while slack + 8 <= end:
            hidden_inode, _, hidden_len, _ = struct.unpack_from(\"<IHBB\", data, slack)
            if hidden_len == 0or slack + 8 + hidden_len > end ornot hidden_inode:
                break
            found.append((hidden_inode, data[slack + 8:slack + 8 + hidden_len].decode(\"latin1\"), True))
            slack += (8 + hidden_len + 3) & ~3
        position += rec_len
    return found

def walk(fs: Ext4) -> dict[str, bytes]:
    files: dict[str, bytes] = {}

    def visit(inode: int, path: str) -> None:
        raw, size = fs.inode_raw(inode)
        mode = struct.unpack_from(\"<H\", raw, 0)[0]
        body = fs.read_inode(inode)
        if mode & 0x4000:
            for child, name, _deleted in directory_entries(body):
                if name in (\".\", \"..\"):
                    continue
                visit(child, f\"{path}/{name}\")
            return
        if mode & 0x8000:
            files[path] = body
            del size

    visit(2, \"\")
    return files

def elf_section(blob: bytes, wanted: str) -> bytes:
    shoff = struct.unpack_from(\"<Q\", blob, 40)[0]
    shentsize = struct.unpack_from(\"<H\", blob, 58)[0]
    shnum = struct.unpack_from(\"<H\", blob, 60)[0]
    shstrndx = struct.unpack_from(\"<H\", blob, 62)[0]

    def header(index: int) -> tuple:
        return struct.unpack_from(\"<IIQQQQIIQQ\", blob, shoff + index * shentsize)

    names = blob[header(shstrndx)[4]:header(shstrndx)[4] + header(shstrndx)[5]]
    for index in range(shnum):
        name_off, _, _, _, offset, size, *_ = header(index)
        name = names[name_off:names.index(b\"\\x00\", name_off)].decode()
        if name == wanted:
            return blob[offset:offset + size]
    raise KeyError(wanted)

def ror8(value: int, count: int) -> int:
    count &= 7
    value &= 0xFF
    if count == 0:
        return value
    return ((value >> count) | (value << (8 - count))) & 0xFF

def consolation(data: bytes, table: bytes) -> bytes:
    \"\"\"dial == 0: sub, rotate right, xor. This is the planted fake flag.\"\"\"
    state = 3
    out = bytearray(data)
    for index, value in enumerate(out):
        value = (value - table[state % 20]) & 0xFF
        value = ror8(value, index)
        value ^= table[index % 20]
        out[index] = value
        state += 7
    return bytes(out)

def derive_key(material: bytes) -> bytes:
    digest = hashlib.sha256(material).digest()
    for _ in range(0x1FFFFFF):
        digest = hashlib.sha256(digest).digest()
    return digest

def decrypt_bytecode(blob: bytes, key_lo: int) -> bytes:
    # mov ebx, eax zeros the high half, then xor rbx, 0x5a5a5a5a5a5a5a5a.
    state = ((key_lo & 0xFFFFFFFF) ^ 0x5A5A5A5A) | (0x5A5A5A5A << 32)
    multiplier = 0x5851F42D4C957F2D
    increment = 0x14057B7EF767814F
    mask = (1 << 64) - 1
    out = bytearray(blob)
    for index in range(len(out)):
        state = (state * multiplier + increment) & mask
        out[index] ^= (state >> 56) & 0xFF
    return bytes(out)

def run_vm(code: bytes, memory: bytearray) -> int:
    acc = 0xFFFFFFA5
    pc = 0
    sp = 0
    flag = 0
    stack = [0] * 64
    steps = 0
    whileTrue:
        steps += 1
        if steps > 0x30D40or pc > 0x1BF:
            return-1
        op = code[pc] ^ (acc & 0xFF)
        if op == 0x1:
            stack[sp] = struct.unpack_from(\"<I\", code, pc + 1)[0]
            sp += 1
            pc += 5
        elif op == 0x2:
            stack[sp] = code[pc + 1]
            sp += 1
            pc += 2
        elif op == 0x3:
            sp -= 1
            pc += 1
        elif op == 0x4:
            stack[sp] = stack[sp - 1]
            sp += 1
            pc += 1
        elif op == 0x5:
            stack[sp - 1], stack[sp - 2] = stack[sp - 2], stack[sp - 1]
            pc += 1
        elif op == 0x6:
            stack[sp] = stack[sp - 2]
            sp += 1
            pc += 1
        elif op == 0x7:
            stack[sp - 1] = memory[stack[sp - 1] & 0x3F]
            pc += 1
        elif op == 0x8:
            stack[sp - 1] = struct.unpack_from(\"<I\", memory, stack[sp - 1] & 0x3F)[0]
            pc += 1
        elif op == 0x9:
            memory[stack[sp - 2] & 0x3F] = stack[sp - 1] & 0xFF
            sp -= 2
            pc += 1
        elif op == 0xA:
            struct.pack_into(\"<I\", memory, stack[sp - 2] & 0x3F, stack[sp - 1] & 0xFFFFFFFF)
            sp -= 2
            pc += 1
        elif0xB <= op <= 0x13:
            left = stack[sp - 2] & 0xFFFFFFFF
            right = stack[sp - 1] & 0xFFFFFFFF
            count = right & 31
            if op == 0xB:
                result = (left + right) & 0xFFFFFFFF
            elif op == 0xC:
                result = (left - right) & 0xFFFFFFFF
            elif op == 0xD:
                result = left ^ right
            elif op == 0xE:
                result = left & right
            elif op == 0xF:
                result = left | right
            elif op == 0x10:
                result = (left << count) & 0xFFFFFFFF
            elif op == 0x11:
                result = left >> count
            elif op == 0x12:
                result = ((left >> count) | (left << ((32 - count) & 31))) & 0xFFFFFFFFif count else left
            else:
                result = (left * right) & 0xFFFFFFFF
            sp -= 1
            stack[sp - 1] = result
            pc += 1
        elif op in (0x15, 0x16, 0x17, 0x18):
            target = pc + 2 + struct.unpack_from(\"b\", code, pc + 1)[0]
            if op == 0x15:
                pc = target
            elif op == 0x16:
                sp -= 1
                pc = target if stack[sp] == 0else pc + 2
            elif op == 0x17:
                sp -= 1
                pc = target if stack[sp] != 0else pc + 2
            else:
                flag = (flag - 1) & 0xFF
                pc = target if flag != 0else pc + 2
        elif op == 0x19:
            flag = code[pc + 1]
            pc += 2
        elif op == 0x1A:
            return0
        elif op == 0x1B:
            stack[sp] = flag
            sp += 1
            pc += 1
        else:
            return-1
        acc = (acc * 0x6D + 0x3B) & 0xFFFFFFFF

def vm_decrypt(code: bytes, key: bytes, blob: bytes) -> bytes:
    out = bytearray(blob)
    for offset in range(0, len(out) - 7, 8):
        memory = bytearray(64)
        memory[:16] = key
        memory[16:24] = out[offset:offset + 8]
        if run_vm(code, memory) == 0:
            out[offset:offset + 8] = memory[16:24]
    return bytes(out)

def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument(\"--image\", type=pathlib.Path, default=DEFAULT_IMAGE)
    args = parser.parse_args()
    fs = Ext4(args.image.read_bytes())
    files = walk(fs)
    print(f\"volume {fs.volume} block {fs.block_size}\")
    cache = {path: body for path, body in files.items() if\"/.kmod/\"in path or path.endswith(\"flag.txt\")}
    for path in sorted(cache):
        print(f\"file {path} {len(cache[path])}\")

    rodata = elf_section(files[\"/usr/local/bin/kcompat.ko\"], \".rodata\")
    data = elf_section(files[\"/usr/local/bin/kcompat.ko\"], \".data\")
    dial = struct.unpack_from(\"<i\", data, 0x9B8)[0]
    table = rodata[0x100:0x115]
    paths = []
    cursor = 0x130
    while cursor < 0x1F0:
        end = rodata.index(b\"\\x00\", cursor)
        paths.append(bytes(byte ^ 0xC3for byte in rodata[cursor:end]))
        cursor = end + 1
        while cursor < 0x1F0and rodata[cursor] == 0:
            cursor += 1
    print(\"dial\", dial)
    print(\"paths\", [item.decode() for item in paths])

    s1 = files[\"/var/.cache/.kmod/.s1\"]
    s2 = files[\"/var/.cache/.kmod/.s2\"]
    s3 = files[\"/var/.cache/.kmod/.s3\"]
    shard = s2[0] & 3
    key_file = files[f\"/var/.cache/.kmod/.k{shard}\"]
    material = s1[:4] + s2[:4] + s3[:4] + key_file[:4]
    digest = derive_key(material)
    expect = bytes(byte ^ 0xA6for byte in rodata[0x3C0:0x3D0])
    print(f\"selector s2[0]&3 = {shard}\")
    print(\"material\", material.hex())
    print(\"digest\", digest.hex())
    print(\"check\", digest[16:].hex(), \"match\", digest[16:] == expect)

    code = decrypt_bytecode(rodata[0x200:0x3C0], struct.unpack_from(\"<I\", digest, 0)[0])
    key = digest[:16]
    real = vm_decrypt(code, key, files[\"/var/.cache/.kmod/.real_cache\"])
    draft = vm_decrypt(code, key, files[\"/var/.cache/.kmod/.real_cache.bak\"])
    fake = consolation(files[\"/home/ctf/flag.txt\"], table)
    print(\"real\", real)
    print(\"draft\", draft)
    print(\"consolation\", fake)

if __name__ == \"__main__\":
    main()

题目描述：
我们从一台失陷主机上抓到了一段流量：capture.pcap。抓包抓到一半主机断电了。
附件：capture.zip

我们从一台失陷主机上抓到了一段流量：capture.pcap。抓包抓到一半主机断电了。

Header (24 bytes): d4c3b2a9 0200 0400 00000000 00000000 ffff0000 37130000
Pkt 47: offset=7001, incl_len=65, orig_len=65
Pkt 48: offset=7082, incl_len=3997706, orig_len=3997696, first16=00000200000000050200000000990800

import struct

with open(\"capture.pcap\", \"rb\") as f:
    data = bytearray(f.read())

data[0:4] = struct.pack(\"<I\", 0xA1B2C3D4)
data[20:24] = struct.pack(\"<I\", 1)

offset = 24
pkt_idx = 0
while offset + 16 <= len(data):
    ts_sec, ts_usec, incl_len, orig_len = struct.unpack(\"<IIII\", data[offset:offset+16])
    pkt_start = offset + 16
    eth_type = struct.unpack(\">H\", data[pkt_start+12:pkt_start+14])[0]
    real_len = 14 + struct.unpack(\">H\", data[pkt_start+16:pkt_start+18])[0] if eth_type == 0x0800else42
    if incl_len != real_len:
        print(f\"Mismatch at Pkt {pkt_idx} (offset {offset}): header={incl_len}, real={real_len}\")
        data[offset+8:offset+16] = struct.pack(\"<II\", real_len, real_len)
    offset = pkt_start + real_len
    pkt_idx += 1
print(f\"Total packets: {pkt_idx}, final offset: {offset}/{len(data)}\")

Mismatch at Pkt 47 (offset 7001): header=65, real=67
Mismatch at Pkt 53 (offset 7491): header=64, real=71
Mismatch at Pkt 59 (offset 7983): header=64, real=71
Mismatch at Pkt 67 (offset 8639): header=71, real=75
Mismatch at Pkt 75 (offset 9337): header=86, real=83
Mismatch at Pkt 83 (offset 10035): header=76, real=83
Total packets: 4561, final offset: 5954555/5954555

KV-VM REFERENCE MANUAL -- firmware v2.4
=======================================
16 registers R0-RF, mod 256. operands follow the opcode byte.

0x00 HALT        0x2C LOADI r,imm   0x4A MOVR rd,rs    0xB4 LOADC rD,[rA]
0x58 STOREC [rA],rV  0x37 PUSH r    0x91 POP r         0x61 EMIT r
0x83 ADDI r,imm  0x6E SUBI r,imm    0x0F XORI r,imm    0xA2 ANDI r,imm
0x1D ROLI r,imm  0xC9 XORR rd,rs    0x53 ADDR rd,rs    0xD7 JMP a16le
0x19 JNZ r,a16le 0xEB JZ r,a16le

00111 0000 01000001 | 10010 0000 | 11111 | 0 (padding)
LOADI R0,  0x41     | EMIT  R0   | HALT  |

flag{tw0_c4t
PX rev.C k=9f3cb712aa55d04e

pkt=117 src=10.0.0.14 seq=0 len=10 cksum=0x50 enc=2d2087dc8b32f1bd270e
pkt=119 src=10.0.0.5  seq=1 len=10 cksum=0x30 enc=ab120bc19c6cf61f7713
pkt=121 src=10.0.0.5  seq=2 len=10 cksum=0x0e enc=0650160eaa2ca91aa952
pkt=123 src=10.0.0.14 seq=3 len=10 cksum=0x37 enc=10feaff6916921ac04b9
pkt=125 src=10.0.0.5  seq=4 len=2  cksum=0x3b enc=80bb

b21c30ce216721f3b8321c00a1944c226923c001ac05c64035101e080307c0b030ca267b8bf9d4f71f87

s_0ne_tr4p}

import struct
import sys

def parse_and_repair_pcap(path):
    with open(path, \"rb\") as f:
        data = f.read()

    offset = 24
    packets = []
    while offset + 16 <= len(data):
        ts_sec, ts_usec, incl_len, orig_len = struct.unpack(\"<IIII\", data[offset:offset+16])
        pkt_start = offset + 16
        eth_type = struct.unpack(\">H\", data[pkt_start+12:pkt_start+14])[0]
        # 根据二层协议与 IPv4 Total Length 计算真实帧长，绕过损坏的 incl_len
        if eth_type == 0x0800:
            ip_len = struct.unpack(\">H\", data[pkt_start+16:pkt_start+18])[0]
            real_len = 14 + ip_len
        elif eth_type == 0x0806:
            real_len = 42
        else:
            real_len = incl_len
        packets.append(data[pkt_start:pkt_start+real_len])
        offset = pkt_start + real_len
    return packets

def run_kv_vm(code_bytes, rev=\"B\"):
    mem = bytearray(code_bytes) + bytearray(256)
    regs = [0] * 16
    stack = []
    out = bytearray()
    pc = 0

    def get_bits(pos, n):
        val = 0
        for i in range(n):
            val = (val << 1) | ((mem[(pos + i) >> 3] >> (7 - ((pos + i) & 7))) & 1)
        return val

    whileTrue:
        op = get_bits(pc, 5)
        if op == 0b11111:  # HALT
            break
        elif op == 0b00111:  # LOADI r, imm
            regs[get_bits(pc + 5, 4)] = get_bits(pc + 9, 8)
            pc += 17
        elif op == 0b10010:  # EMIT r
            out.append(regs[get_bits(pc + 5, 4)])
            pc += 9
        elif op == 0b00011:  # ADDI r, imm
            r = get_bits(pc + 5, 4)
            regs[r] = (regs[r] + get_bits(pc + 9, 8)) & 0xFF
            pc += 17
        elif op == 0b10101:  # SUBI r, imm
            r = get_bits(pc + 5, 4)
            regs[r] = (regs[r] - get_bits(pc + 9, 8)) & 0xFF
            pc += 17
        elif op == 0b01011:  # XORI r, imm
            r = get_bits(pc + 5, 4)
            regs[r] = (regs[r] ^ get_bits(pc + 9, 8)) & 0xFF
            pc += 17
        elif op == 0b01110:  # ANDI r, imm
            r = get_bits(pc + 5, 4)
            regs[r] = (regs[r] & get_bits(pc + 9, 8)) & 0xFF
            pc += 17
        elif op == 0b11000:  # ROLI (rev.B) / RORI (rev.C)
            r = get_bits(pc + 5, 4)
            imm = get_bits(pc + 9, 8) & 7
            v = regs[r]
            if rev == \"B\":
                regs[r] = ((v << imm) | (v >> (8 - imm))) & 0xFF
            else:
                regs[r] = ((v >> imm) | (v << (8 - imm))) & 0xFF
            pc += 17
        elif op == 0b01001:  # MOVR rd, rs
            regs[get_bits(pc + 5, 4)] = regs[get_bits(pc + 9, 4)]
            pc += 13
        elif op == 0b10001:  # XORR rd, rs
            rd = get_bits(pc + 5, 4)
            regs[rd] ^= regs[get_bits(pc + 9, 4)]
            pc += 13
        elif op == 0b00110:  # ADDR rd, rs
            rd = get_bits(pc + 5, 4)
            regs[rd] = (regs[rd] + regs[get_bits(pc + 9, 4)]) & 0xFF
            pc += 13
        elif op == 0b01101:  # PUSH r
            stack.append(regs[get_bits(pc + 5, 4)])
            pc += 9
        elif op == 0b10110:  # POP r
            r = get_bits(pc + 5, 4)
            ifnot stack:
                if rev == \"B\":
                    raise RuntimeError(\"Stack underflow\")
                regs[r] = 0
            else:
                regs[r] = stack.pop()
            pc += 9
        elif op == 0b00001:  # STOREC [rA], rV
            mem[regs[get_bits(pc + 5, 4)]] = regs[get_bits(pc + 9, 4)]
            pc += 13
        elif op == 0b10100:  # LOADC rD, [rA]
            regs[get_bits(pc + 5, 4)] = mem[regs[get_bits(pc + 9, 4)]]
            pc += 13
        elif op == 0b01010:  # JMP addr16
            pc = get_bits(pc + 5, 16)
        elif op == 0b11100:  # JNZ r, addr16
            r = get_bits(pc + 5, 4)
            a = get_bits(pc + 9, 16)
            pc = a if regs[r] != 0else pc + 25
        elif op == 0b10000:  # JZ r, addr16
            r = get_bits(pc + 5, 4)
            a = get_bits(pc + 9, 16)
            pc = a if regs[r] == 0else pc + 25
        elif op == 0b01000:  # JMPR r
            pc = regs[get_bits(pc + 5, 4)] * (8if rev == \"B\"else16)
        else:
            raise RuntimeError(f\"Unknown opcode {op:05b} at pc={pc}\")

    return bytes(out)

def main():
    pcap_path = sys.argv[1] if len(sys.argv) > 1else\"capture.pcap\"
    packets = parse_and_repair_pcap(pcap_path)
    print(f\"[*] Parsed {len(packets)} packets from {pcap_path}\")

    # 1. 提取 KV 与 PX 两个 ICMP Echo Request 通道载荷
    kv_pkts = []
    px_pkts = []
    for pkt in packets:
        if struct.unpack(\">H\", pkt[12:14])[0] == 0x0800and pkt[23] == 1:
            ihl = (pkt[14] & 0x0F) * 4
            icmp = pkt[14 + ihl:]
            if icmp[0] == 8:  # Echo Request
                payload = icmp[8:]
                if payload.startswith(b\"KV\"):
                    seq, length, cksum = payload[2], payload[3], payload[4]
                    kv_pkts.append((seq, length, cksum, payload[5:5+length]))
                elif payload.startswith(b\"PX\"):
                    seq = int.from_bytes(payload[2:4], \"big\")
                    length, cksum = payload[4], payload[5]
                    px_pkts.append((seq, length, cksum, payload[6:6+length]))

    # 2. 解密并重组 KV 通道 (K_0 = 0xA5)
    K = [0xA5]
    for _ in range(256):
        K.append((K[-1] * 31 + 7) & 0xFF)

    kv_by_seq = {}
    for seq, length, cksum, enc in kv_pkts:
        plain = bytes(enc[i] ^ ((K[seq] + i) & 0xFF) for i in range(length))
        if (sum(plain) & 0xFF) == cksum and seq notin kv_by_seq:
            kv_by_seq[seq] = plain

    kv_rom = b\"\".join(kv_by_seq[s] for s in sorted(kv_by_seq))
    print(f\"[*] Reassembled KV bytecode: {len(kv_rom)} bytes (seqs 0..{max(kv_by_seq)})\")

    kv_out = run_kv_vm(kv_rom, rev=\"B\").decode()
    print(\"[*] Stage 1 (KV rev.B) output:\")
    print(kv_out.strip())

    lines = kv_out.strip().splitlines()
    flag_part1 = lines[0]
    px_key = bytes.fromhex(lines[1].split(\"k=\")[1])

    # 3. 使用第一阶段输出的密钥解密 PX 通道并在 rev.C 虚拟机上执行
    px_enc = b\"\".join(enc for _, _, _, enc in sorted(px_pkts, key=lambda x: x[0]))
    px_rom = bytes(px_enc[i] ^ px_key[i % len(px_key)] for i in range(len(px_enc)))
    print(f\"[*] Reassembled PX bytecode: {len(px_rom)} bytes\")

    flag_part2 = run_kv_vm(px_rom, rev=\"C\").decode()
    print(\"[*] Stage 2 (PX rev.C) output:\")
    print(flag_part2)

    print(f\"\
[+] Flag: {flag_part1}{flag_part2}\")

if __name__ == \"__main__\":
    main()