File size: 5,294 Bytes
f0926cb
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
"""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()