app.py

"""OpenRange HF Space — Gradio frontend proxying to local Kind/Helm backend."""

import json
import os

import gradio as gr
import httpx

BACKEND_URL = os.environ.get("BACKEND_URL", "").rstrip("/")
print(f"[openrange] BACKEND_URL = {BACKEND_URL[:50]}..." if BACKEND_URL else "[openrange] WARNING: BACKEND_URL not set")


def _get(path: str, timeout: float = 60.0) -> str:
    if not BACKEND_URL:
        return '{"error": "BACKEND_URL not configured"}'
    try:
        r = httpx.get(f"{BACKEND_URL}/{path}", timeout=timeout)
        return r.text
    except Exception as e:
        return json.dumps({"error": f"{type(e).__name__}: {e}"})


def _post(path: str, body: dict | None = None, timeout: float = 120.0) -> str:
    if not BACKEND_URL:
        return '{"error": "BACKEND_URL not configured"}'
    try:
        r = httpx.post(f"{BACKEND_URL}/{path}", json=body or {}, timeout=timeout)
        return r.text
    except Exception as e:
        return json.dumps({"error": f"{type(e).__name__}: {e}"})


def _pretty(text: str) -> str:
    try:
        return json.dumps(json.loads(text), indent=2)
    except Exception:
        return text


# ── Callbacks ──────────────────────────────────────────────────────────────


def check_health():
    return _pretty(_get("health"))


def get_metadata():
    return _pretty(_get("metadata"))


def get_state():
    return _pretty(_get("state"))


def get_console_snapshot():
    return _pretty(_get("console/api/snapshot"))


def get_console_episode():
    return _pretty(_get("console/api/episode"))


def get_console_history():
    return _pretty(_get("console/api/history"))


def do_reset():
    return _pretty(_post("reset"))


def do_step(command: str, mode: str):
    if not command.strip():
        return "Enter a command first"
    return _pretty(_post("step", {"command": command, "mode": mode}))


# ── UI ─────────────────────────────────────────────────────────────────────

with gr.Blocks(
    title="OpenRange",
    theme=gr.themes.Default(primary_hue="blue", neutral_hue="slate"),
    css="""
    .output textarea { font-family: 'SF Mono','Fira Code',Consolas,monospace !important;
                       font-size: 13px !important; }
    """,
) as demo:
    gr.Markdown(
        "# OpenRange\n"
        "Multi-agent cybersecurity gymnasium — proxied to local Kind/Helm backend"
    )

    with gr.Tab("Operator Console"):
        with gr.Row():
            health_btn = gr.Button("Health", size="sm")
            state_btn = gr.Button("State", size="sm")
            meta_btn = gr.Button("Metadata", size="sm")
            reset_btn = gr.Button("Reset Environment", variant="primary", size="sm")

        info_out = gr.Textbox(label="Response", lines=8, interactive=False, elem_classes="output")
        health_btn.click(fn=check_health, outputs=info_out)
        state_btn.click(fn=get_state, outputs=info_out)
        meta_btn.click(fn=get_metadata, outputs=info_out)
        reset_btn.click(fn=do_reset, outputs=info_out)

        gr.Markdown("---")
        gr.Markdown("### Snapshot & Episode")
        with gr.Row():
            snap_btn = gr.Button("Snapshot Info", size="sm")
            ep_btn = gr.Button("Episode Info", size="sm")
            hist_btn = gr.Button("Action History", size="sm")

        console_out = gr.Textbox(label="Console Data", lines=10, interactive=False, elem_classes="output")
        snap_btn.click(fn=get_console_snapshot, outputs=console_out)
        ep_btn.click(fn=get_console_episode, outputs=console_out)
        hist_btn.click(fn=get_console_history, outputs=console_out)

    with gr.Tab("Execute Step"):
        gr.Markdown("Send a command to the range as Red (attacker) or Blue (defender).")
        with gr.Row():
            cmd_input = gr.Textbox(
                label="Command",
                placeholder="nmap -sV 10.0.1.0/24",
                scale=4,
            )
            mode_input = gr.Dropdown(
                choices=["red", "blue"],
                value="red",
                label="Mode",
                scale=1,
            )
        step_btn = gr.Button("Execute", variant="primary")
        step_out = gr.Textbox(label="Observation", lines=15, interactive=False, elem_classes="output")
        step_btn.click(fn=do_step, inputs=[cmd_input, mode_input], outputs=step_out)

    with gr.Tab("API Info"):
        gr.Markdown(f"""
### Backend URL
`{BACKEND_URL or 'NOT SET — configure BACKEND_URL Space variable'}`

### OpenEnv Endpoints (proxied)
| Method | Path | Description |
|--------|------|-------------|
| GET | `/health` | Backend health check |
| GET | `/metadata` | Environment metadata |
| GET | `/schema` | Action/Observation JSON schemas |
| POST | `/reset` | Reset environment (select new snapshot) |
| POST | `/step` | Execute an action |
| GET | `/state` | Current episode state |
| GET | `/console/` | Operator debug console |

### Architecture
```
You (browser) -> HF Space (this Gradio app) -> bore tunnel -> local machine -> Kind cluster
```
""")

demo.launch()