Crypto Kindergarten

Full Source Code

from Crypto.Util.number import getPrime
import os
import secrets

FLAG = os.getenv("GZCTF_FLAG", "BKISC{local_test_flag}")

def menu():
    print("1. Query")
    print("2. Verify")
    print("3. Exit")
    return input("Option: ")
def main():
    secret_size = 256
    p_size = 32
    secret = os.urandom(secret_size)
    secret_num = int(secret.hex(), 16)
    for _ in range(75):
        option = menu()
        if option == "1":
            p = getPrime(p_size)
            r = secret_num % p
            if secrets.randbits(1):
                r = -r % p
            print(f'(p, r) = ({p}, {r})')
        elif option == "2":
            secret_input = input("What is the secret? ")
            secret_input = bytes.fromhex(secret_input)
            if secret_input == secret:
                print("Correct!")
                print("Here is your reward: ", FLAG)
            else:
                print("SUSSSSSS")
            exit()
        elif option == "3":
            print("Exiting...")
            break
        else:
            print("Invalid option, please try again.")

main()

Code Explanation

FLAG = os.getenv("GZCTF_FLAG", "BKISC{local_test_flag}") loads the real flag from the GZCTF_FLAG environment variable on the remote server. When we run the challenge locally, it falls back to BKISC{local_test_flag}.

menu() prints the three available actions and reads our choice from stdin. This matters because the main loop only runs 75 times, so every menu choice consumes one turn.

secret_size = 256 means the secret is 256 bytes long. As an integer, the secret is smaller than 2^2048.

p_size = 32 means every query uses a fresh 32-bit prime.

secret = os.urandom(secret_size) generates the random session secret.

secret_num = int(secret.hex(), 16) converts the secret bytes into a large integer so the program can compute modular residues.

The Query branch is the vulnerable part:

p = getPrime(p_size)
r = secret_num % p
if secrets.randbits(1):
    r = -r % p
print(f'(p, r) = ({p}, {r})')

For each query, the server returns a prime p and a residue r. However, r is randomly either secret_num mod p or -secret_num mod p. Therefore each query leaks:

secret_num ≡ ±r_i (mod p_i)

The Verify branch asks us to submit the exact secret in hex. If the submitted bytes match the session secret, the server prints the flag.

Exploitation Workflow

We cannot use all 75 turns for queries, because we still need one final turn to choose Verify. So the exploit sends 74 queries, recovers the secret, and uses the 75th menu action to verify it.

If the sign were known, this would be a direct CRT problem. The random sign turns it into a signed CRT problem:

x ≡ ±r_1  (mod p_1)
x ≡ ±r_2  (mod p_2)
...
x ≡ ±r_74 (mod p_74)
0 <= x < 2^2048

Let:

M = p_1 * p_2 * ... * p_n

With 74 independent 32-bit primes, M is around 2330 bits, which is larger than the 2048-bit secret.

For each residue, we build a CRT lift R_i:

R_i ≡ r_i (mod p_i)
R_i ≡ 0   (mod p_j), j != i

Then there are hidden signs s_i ∈ {+1, -1} such that:

secret_num ≡ s_1 R_1 + s_2 R_2 + ... + s_n R_n (mod M)

The correct signed sum has a very small representative modulo M, because secret_num < 2^2048 while M is about 2330 bits. This is a low-density subset-sum style problem, which can be solved with lattice reduction.

We build the lattice with K = 2^2048:

[K, 0, 0, ..., R_1]
[0, K, 0, ..., R_2]
...
[0, 0, 0, ..., M  ]

The correct sign combination creates this short lattice vector:

s_1 row_1 + s_2 row_2 + ... + s_n row_n - k row_M
= (s_1 K, s_2 K, ..., s_n K, secret_num)

So after running LLL, we scan the reduced basis for a row where all coordinates except the last one are exactly ±K. The absolute value of the last coordinate is the recovered secret.

Finally, we convert it back to 256 bytes and send:

secret_num.to_bytes(256, "big").hex()

Exploit Script

The exploit is saved as solve.py.

#!/usr/bin/env python3
import argparse
import re
import socket
import subprocess
import sys

try:
    from fpylll import IntegerMatrix, LLL
except ImportError as exc:
    raise SystemExit("Missing dependency: python3 -m pip install fpylll cysignals") from exc


SECRET_SIZE = 256
SECRET_BITS = SECRET_SIZE * 8
QUERY_COUNT = 74
PROMPT = b"Option:"
SECRET_PROMPT = b"What is the secret?"
PAIR_RE = re.compile(rb"\(p, r\) = \((\d+), (\d+)\)")


class LocalTube:
    def __init__(self, path):
        self.proc = subprocess.Popen(
            [sys.executable, "-u", path],
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,
        )

    def recv(self, size):
        return self.proc.stdout.read(size)

    def sendline(self, data):
        if isinstance(data, str):
            data = data.encode()
        self.proc.stdin.write(data + b"\n")
        self.proc.stdin.flush()

    def recv_remaining(self):
        out, _ = self.proc.communicate(timeout=3)
        return out


class SocketTube:
    def __init__(self, host, port):
        self.sock = socket.create_connection((host, port))

    def recv(self, size):
        return self.sock.recv(size)

    def sendline(self, data):
        if isinstance(data, str):
            data = data.encode()
        self.sock.sendall(data + b"\n")

    def recv_remaining(self):
        self.sock.settimeout(2)
        chunks = []
        while True:
            try:
                chunk = self.sock.recv(4096)
            except TimeoutError:
                break
            if not chunk:
                break
            chunks.append(chunk)
        return b"".join(chunks)


def recv_until(tube, marker):
    data = b""
    while marker not in data:
        chunk = tube.recv(1)
        if not chunk:
            raise EOFError(f"connection closed while waiting for {marker!r}")
        data += chunk
    return data


def collect_pairs(tube):
    recv_until(tube, PROMPT)
    pairs = []

    for _ in range(QUERY_COUNT):
        tube.sendline("1")
        response = recv_until(tube, PROMPT)
        match = PAIR_RE.search(response)
        if not match:
            raise ValueError(f"could not parse query response: {response!r}")
        pairs.append((int(match.group(1)), int(match.group(2))))

    return pairs


def deduplicate_pairs(pairs):
    unique = {}
    for p, r in pairs:
        r %= p
        if p in unique:
            previous = unique[p]
            if (previous - r) % p != 0 and (previous + r) % p != 0:
                raise ValueError(f"inconsistent residues for repeated prime {p}")
            continue
        unique[p] = r
    return [(p, r) for p, r in unique.items()]


def recover_secret_num(pairs):
    pairs = deduplicate_pairs(pairs)

    modulus = 1
    for p, _ in pairs:
        modulus *= p

    if modulus.bit_length() <= SECRET_BITS:
        raise ValueError(f"not enough CRT modulus bits: {modulus.bit_length()}")

    lifts = []
    for p, r in pairs:
        partial = modulus // p
        lift = (r * partial * pow(partial, -1, p)) % modulus
        lifts.append(lift)

    scale = 1 << SECRET_BITS
    dim = len(lifts) + 1
    basis = IntegerMatrix(dim, dim)

    for i, lift in enumerate(lifts):
        basis[i, i] = scale
        basis[i, dim - 1] = lift
    basis[dim - 1, dim - 1] = modulus

    LLL.reduction(basis, delta=0.99)

    for row in range(dim):
        vector = [int(basis[row, col]) for col in range(dim)]
        if not all(abs(x) == scale for x in vector[:-1]):
            continue

        candidate = abs(vector[-1])
        if candidate >= (1 << SECRET_BITS):
            continue
        if all((candidate - r) % p == 0 or (candidate + r) % p == 0 for p, r in pairs):
            return candidate

    raise ValueError("LLL did not recover the secret")


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--host", help="remote host")
    parser.add_argument("--port", type=int, help="remote port")
    parser.add_argument("--path", default="chall.py", help="local challenge path")
    args = parser.parse_args()

    if args.host and args.port:
        tube = SocketTube(args.host, args.port)
    else:
        tube = LocalTube(args.path)

    pairs = collect_pairs(tube)
    secret_num = recover_secret_num(pairs)
    secret_hex = secret_num.to_bytes(SECRET_SIZE, "big").hex()

    tube.sendline("2")
    recv_until(tube, SECRET_PROMPT)
    tube.sendline(secret_hex)
    print(tube.recv_remaining().decode(errors="replace"), end="")


if __name__ == "__main__":
    main()

Running the Exploit

Install the dependency:

python3 -m pip install fpylll cysignals

Run against the local challenge:

python3 solve.py

Local output:

Correct!
Here is your reward:  BKISC{local_test_flag}

Run against the remote service:

python3 solve.py --host <host> --port <port>

Local Flag

BKISC{local_test_flag}

The real flag is stored only on the remote service in GZCTF_FLAG, so the same exploit should be run with the challenge host and port to retrieve it.