Spaces:
Sleeping
Sleeping
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()
|