""" FastAPI Server for OpenSecOpsEnv ================================= Exposes reset(), step(), state() over HTTP. Endpoints --------- POST /reset – begin new episode (creates/replaces session) POST /step – execute one action within a session GET /state – get full debug state for a session POST /grade – grade the current episode GET /health – liveness probe GET /tasks – list available tasks GET /web – simple debug UI GET /dashboard – 🔥 Live AI Demo Dashboard (Attacker vs Defender) GET /demo/stream – SSE stream: single-agent episode (for basic demo) GET /battle/stream – SSE stream: Red Attacker vs Blue Defender live battle Multi-Agent Architecture ------------------------ - Red Agent (Attacker) – escalates the incident, tries to worsen metrics - Blue Agent (Defender) – investigates and mitigates the incident - Both share the same environment; turn order is interleaved - Each agent has a distinct reward signal (zero-sum-like) - Curriculum: Blue agent auto-levels up as it solves harder tasks Session model ------------- Each reset() call creates a named session keyed by `session_id` (defaults to the task_id). This allows multiple concurrent agents to run different tasks without colliding. """ from __future__ import annotations import asyncio import copy import json import os import random import re from dataclasses import dataclass, field from typing import Any, Optional # Auto-load .env file so HF_TOKEN etc. are always available try: from dotenv import load_dotenv load_dotenv() except ImportError: pass # python-dotenv not installed — fall back to env vars only from fastapi import FastAPI, HTTPException from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import HTMLResponse, StreamingResponse, RedirectResponse from pydantic import BaseModel from opensecops_env.env import OpenSecOpsEnv from opensecops_env.grader import grade from opensecops_env.models import SecOpsAction from opensecops_env.tasks.task_definitions import TASKS # --------------------------------------------------------------------------- # 🤖 Live AI Inference Engine # --------------------------------------------------------------------------- # Set these environment variables to connect to your trained Hugging Face # Inference Endpoints. When not set, falls back to heuristic playbooks. # # TRAINED_MODEL_ENDPOINT — your fine-tuned Qwen2.5-7B-GRPO endpoint # UNTRAINED_MODEL_ENDPOINT — base Qwen2.5-7B for contrast (optional) # HF_API_TOKEN — your Hugging Face API token # --------------------------------------------------------------------------- # Hardcoded live defaults — override via env var at any time _TRAINED_ENDPOINT: Optional[str] = ( os.environ.get("TRAINED_MODEL_ENDPOINT", "").strip() or "https://hk5m5hadqtjiqg53.us-east-1.aws.endpoints.huggingface.cloud" ) _UNTRAINED_ENDPOINT: Optional[str] = os.environ.get("UNTRAINED_MODEL_ENDPOINT", "").strip() or None _HF_API_TOKEN: str = os.environ.get("HF_API_TOKEN", os.environ.get("HF_TOKEN", "")) if not _HF_API_TOKEN: # ASCII-only: Windows consoles often use cp1252 and choke on emoji in print(). print( "\nWARNING: HF_API_TOKEN / HF_TOKEN not set. " "The live AI endpoint will get 401 Unauthorized and fall back to heuristic playbooks.\n" " Fix: set HF_TOKEN=hf_xxxx before starting uvicorn.\n" ) _AI_SYSTEM_PROMPT = """You are an expert on-call security engineer responding to a production incident. At each step you receive: alerts, metrics, logs, topology, last_action_result. Your goal: investigate the root cause and apply targeted mitigations. RESPOND ONLY with a valid JSON object: {"action_type": "", "parameters": {}} Actions: query_logs, inspect_metrics, restart_service, scale_service, block_ip, rollback_deployment, run_security_scan, isolate_service, submit_diagnosis Diagnosis labels: infra_failure:memory_leak | infra_failure:service_crash misconfiguration:bad_config cyber_attack:ddos | cyber_attack:data_exfiltration | cyber_attack:privilege_escalation""" def _obs_to_text(obs: dict, step: int) -> str: """Format an observation dict into the prompt text the model was trained on.""" parts = [f"=== Step {step} ==="] parts.append(f"Last result: {obs.get('last_action_result', '')}") if obs.get("alerts"): parts.append("\nALERTS:") for a in obs["alerts"]: parts.append(f" [{a.get('severity','').upper()}] {a.get('service')} - {a.get('message','')}") parts.append("\nMETRICS:") for svc, m in obs.get("metrics", {}).items(): if isinstance(m, dict): parts.append( f" {svc}: cpu={m.get('cpu',0):.1f}% mem={m.get('memory',0):.1f}% " f"lat={m.get('latency',0):.0f}ms err={m.get('error_rate',0):.2f}%" ) parts.append("\nLOGS:") for line in obs.get("logs", [])[:5]: parts.append(f" {line}") parts.append("\nTOPOLOGY:") for svc, deps in obs.get("topology", {}).items(): parts.append(f" {svc} -> {deps}") parts.append("\nRespond with JSON action:") return "\n".join(parts) def _parse_ai_action(text: str) -> Optional[SecOpsAction]: """ Parse JSON action from LLM response text. Handles multiple output formats the model may produce. """ if isinstance(text, list): text = text[-1].get("content", "") if text else "" if not isinstance(text, str): text = str(text) # Strip markdown code fences text = re.sub(r"```[a-z]*\n?", "", text.strip()).strip() text = re.sub(r"```", "", text).strip() # Valid action types this environment accepts VALID_ACTIONS = { "query_logs", "inspect_metrics", "restart_service", "scale_service", "block_ip", "rollback_deployment", "run_security_scan", "isolate_service", "submit_diagnosis", } # Map common model-generated synonyms → valid action types ACTION_SYNONYMS = { "investigate": "query_logs", "query": "query_logs", "log_query": "query_logs", "check_logs": "query_logs", "metrics": "inspect_metrics", "check_metrics": "inspect_metrics", "inspect": "inspect_metrics", "monitor": "inspect_metrics", "restart": "restart_service", "reboot": "restart_service", "scale": "scale_service", "block": "block_ip", "ban": "block_ip", "rollback": "rollback_deployment", "revert": "rollback_deployment", "scan": "run_security_scan", "security_scan": "run_security_scan", "isolate": "isolate_service", "quarantine": "isolate_service", "diagnose": "submit_diagnosis", "diagnosis": "submit_diagnosis", "submit": "submit_diagnosis", } # Diagnosis label synonyms DIAGNOSIS_SYNONYMS = { "memory_leak": "infra_failure:memory_leak", "service_crash": "infra_failure:service_crash", "bad_config": "misconfiguration:bad_config", "ddos": "cyber_attack:ddos", "data_exfiltration": "cyber_attack:data_exfiltration", "exfiltration": "cyber_attack:data_exfiltration", "privilege_escalation": "cyber_attack:privilege_escalation", } def _normalise_action(raw: str) -> str: raw = raw.lower().strip().replace("-", "_").replace(" ", "_") if raw in VALID_ACTIONS: return raw return ACTION_SYNONYMS.get(raw, "inspect_metrics") # --- Try: extract first JSON object --- m = re.search(r"\{.*?\}", text, re.DOTALL) if m: try: d = json.loads(m.group(0)) # Standard format: {"action_type": "...", "parameters": {...}} if "action_type" in d: atype = _normalise_action(d["action_type"]) params = d.get("parameters", {}) # Normalise diagnosis label if present if "label" in params: params["label"] = DIAGNOSIS_SYNONYMS.get(params["label"], params["label"]) return SecOpsAction(action_type=atype, parameters=params) # Alternate format: {"action": "INVESTIGATE", "details": {"hosts": ["db"]}} if "action" in d: raw_action = str(d.get("action", "")).lower() details = d.get("details", d.get("parameters", {})) atype = _normalise_action(raw_action) # Extract service from details["hosts"] or details["services"] params: dict[str, Any] = {} hosts = details.get("hosts", details.get("services", [])) if hosts and isinstance(hosts, list) and hosts: svc = hosts[0] if atype in ("query_logs", "run_security_scan", "inspect_metrics", "restart_service", "isolate_service"): params["service"] = svc elif atype == "block_ip": params["ip"] = svc ips = details.get("ips", details.get("ip", [])) if ips: if isinstance(ips, list): params["ip"] = ips[0] else: params["ip"] = str(ips) label = details.get("label", details.get("diagnosis", "")) if label: params["label"] = DIAGNOSIS_SYNONYMS.get(label, label) return SecOpsAction(action_type=atype, parameters=params) except Exception: pass # --- Fallback: keyword scan of raw text --- text_lower = text.lower() # Check for diagnosis first (highest value) for kw, label in DIAGNOSIS_SYNONYMS.items(): if kw.replace("_", " ") in text_lower or kw in text_lower: return SecOpsAction( action_type="submit_diagnosis", parameters={"label": label}, ) # Then mitigations for kw in ["isolate", "quarantine"]: if kw in text_lower: svc = "db" if "db" in text_lower else "auth" return SecOpsAction(action_type="isolate_service", parameters={"service": svc}) for kw in ["block", "ban"]: if kw in text_lower: return SecOpsAction(action_type="block_ip", parameters={"ip": "10.0.0.99"}) if "restart" in text_lower or "reboot" in text_lower: svc = "auth" if "auth" in text_lower else "api" return SecOpsAction(action_type="restart_service", parameters={"service": svc}) if "scan" in text_lower or "security" in text_lower: svc = "db" if "db" in text_lower else "api" return SecOpsAction(action_type="run_security_scan", parameters={"target": svc}) if "log" in text_lower or "query" in text_lower or "investigate" in text_lower: svc = "db" if "db" in text_lower else "auth" return SecOpsAction(action_type="query_logs", parameters={"service": svc}) # Ultimate fallback return SecOpsAction(action_type="inspect_metrics", parameters={}) async def _query_ai_model( endpoint_url: str, obs: dict, step: int, timeout: float = 30.0, ) -> Optional[SecOpsAction]: """ Call a Hugging Face Inference Endpoint with the current observation. Returns a parsed SecOpsAction, or None if the call fails. """ try: import httpx except ImportError: return None prompt_text = _obs_to_text(obs, step) payload = { "inputs": prompt_text, "parameters": { "max_new_tokens": 128, "temperature": 0.1, "do_sample": True, "return_full_text": False, }, } headers = {"Content-Type": "application/json"} if _HF_API_TOKEN: headers["Authorization"] = f"Bearer {_HF_API_TOKEN}" try: async with httpx.AsyncClient(timeout=timeout) as client: resp = await client.post(endpoint_url, json=payload, headers=headers) resp.raise_for_status() data = resp.json() # HF text-generation returns [{"generated_text": "..."}] if isinstance(data, list) and data: generated = data[0].get("generated_text", "") elif isinstance(data, dict): generated = data.get("generated_text", str(data)) else: generated = str(data) return _parse_ai_action(generated) except Exception as e: print(f"[AI] endpoint call failed: {type(e).__name__}: {e}") return None def _randomise_env_seed(env: Any, episode_seed: Optional[int] = None) -> None: """ Override the env's internal RNG with a fresh random seed after reset. This makes each episode unique — different metric drift, different log ordering, different alert shuffling — so the dashboard never shows the same episode twice. """ seed = episode_seed if episode_seed is not None else random.randint(0, 2**31) env._rng = random.Random(seed) # Also add small random noise to initial metrics so starting values vary for svc, metrics in env._metrics.items(): jitter = random.Random(seed + hash(svc)) metrics.cpu = max(0.0, min(100.0, metrics.cpu + jitter.uniform(-8, 8))) metrics.memory = max(0.0, min(100.0, metrics.memory + jitter.uniform(-5, 5))) metrics.latency = max(0.0, min(5000.0, metrics.latency + jitter.uniform(-15, 30))) # --------------------------------------------------------------------------- # Session registry (session_id → OpenSecOpsEnv instance) # --------------------------------------------------------------------------- _sessions: dict[str, OpenSecOpsEnv] = {} _default_session: str = "default" def _get_session(session_id: str) -> OpenSecOpsEnv: if session_id not in _sessions: raise HTTPException( status_code=400, detail=f"Session '{session_id}' not found. Call /reset first.", ) return _sessions[session_id] # --------------------------------------------------------------------------- # ██████╗ ███████╗██████╗ ███████╗███████╗ ██████╗ ███████╗███████╗ # ██╔══██╗██╔════╝██╔══██╗ ██╔════╝██╔════╝██╔════╝ ██╔════╝██╔════╝ # ██████╔╝█████╗ ██║ ██║ ███████╗█████╗ ██║ ███╗█████╗ ███████╗ # ██╔══██╗██╔══╝ ██║ ██║ ╚════██║██╔══╝ ██║ ██║██╔══╝ ╚════██║ # ██║ ██║███████╗██████╔╝ ███████║███████╗╚██████╔╝███████╗███████║ # Multi-Agent: Red (Attacker) vs Blue (Defender) # --------------------------------------------------------------------------- # Red agent actions: escalate attacks, confuse defender, corrupt metrics RED_ACTIONS = [ "inject_noise", # inject misleading log noise "amplify_attack", # boost attack_progress "corrupt_metric", # spike a random healthy service's cpu/latency "create_false_alert", # add a misleading critical alert "accelerate_spread", # spread attack to adjacent service ] @dataclass class RedAgentState: """Lightweight internal state for the Red agent.""" task_id: str = "" round: int = 0 actions_taken: list[str] = field(default_factory=list) total_damage: float = 0.0 class MultiAgentSecOpsEnv: """ Wraps OpenSecOpsEnv to expose separate blue_step() and red_step() interfaces. Blue Agent (Defender): uses the standard SecOpsAction API. Red Agent (Attacker): uses RedAgent heuristics to make the incident worse. Turn order: red acts first each round (escalate), then blue responds (mitigate). This models real adversarial incident response. """ def __init__(self) -> None: self._env = OpenSecOpsEnv() self._red_state = RedAgentState() self._rng = random.Random() # Unseeded — random each episode self._cumulative_blue_reward = 0.0 self._cumulative_red_reward = 0.0 def reset(self, task_id: str) -> dict[str, Any]: obs = self._env.reset(task_id) # Randomise the env seed so each episode differs episode_seed = random.randint(0, 2**31) _randomise_env_seed(self._env, episode_seed) self._rng = random.Random(episode_seed) # Red agent also differently seeded self._red_state = RedAgentState(task_id=task_id) self._cumulative_blue_reward = 0.0 self._cumulative_red_reward = 0.0 return self._build_ma_state(obs) # --- Blue (Defender) step --- def blue_step( self, action: SecOpsAction ) -> tuple[dict[str, Any], float, bool, dict[str, Any]]: obs, reward, done, info = self._env.step(action) self._cumulative_blue_reward += reward info["blue_cumulative"] = round(self._cumulative_blue_reward, 4) info["red_cumulative"] = round(self._cumulative_red_reward, 4) return self._build_ma_state(obs), reward, done, info # --- Red (Attacker) step --- def red_step( self, action_type: Optional[str] = None ) -> tuple[dict[str, Any], float, bool, dict[str, Any]]: """ Heuristic Red Agent step. Reward is the negative of what the Blue agent would gain — i.e., Red benefits when the environment is harder to diagnose and recover from. """ self._red_state.round += 1 red_reward = 0.0 message = "" env = self._env hidden = env._hidden metrics = env._metrics # Choose action heuristically if not specified if action_type is None or action_type == "auto": action_type = self._heuristic_red_action() self._red_state.actions_taken.append(action_type) if action_type == "inject_noise": # Add misleading log entries to obscure the real issue env._task_cfg.setdefault("initial_logs", []) noise_logs = [ "[cache] WARN Memory spike (harmless GC event) detected", "[db] INFO Routine VACUUM running – may spike CPU temporarily", "[auth] INFO Token rotation in progress – latency may increase", "[api] WARN Connection pool resize – client may see 503 briefly", ] entry = self._rng.choice(noise_logs) env._task_cfg["initial_logs"].append(entry) red_reward = 0.1 message = f"Red injected misleading log: {entry[:60]}..." elif action_type == "amplify_attack": if hidden.true_root_cause == "cyber_attack": boost = self._rng.uniform(0.05, 0.15) hidden.attack_progress = min(1.0, hidden.attack_progress + boost) red_reward = 0.2 message = f"Red amplified attack progress by {boost:.2f} → {hidden.attack_progress:.2f}" else: red_reward = 0.0 message = "Red tried to amplify — no active attack to boost." elif action_type == "corrupt_metric": # Spike a *healthy* (non-affected) service to mislead the defender healthy = [ s for s in metrics if s not in hidden.affected_services ] if healthy: svc = self._rng.choice(healthy) metrics[svc].cpu = min(100.0, metrics[svc].cpu + self._rng.uniform(20, 40)) metrics[svc].latency = min(5000.0, metrics[svc].latency + self._rng.uniform(150, 400)) red_reward = 0.15 message = f"Red spiked metrics on healthy service '{svc}' to create false alarm." else: red_reward = 0.0 message = "Red tried to corrupt metrics — no healthy services available." elif action_type == "create_false_alert": healthy = [s for s in metrics if s not in hidden.affected_services] if healthy: svc = self._rng.choice(healthy) false_alert = { "service": svc, "type": self._rng.choice(["high_cpu", "high_memory", "high_latency"]), "severity": "critical", "message": f"[RED INJECTED] {svc} critical threshold exceeded", "_red_injected": True, } env._task_cfg.setdefault("initial_alerts", []).append(false_alert) red_reward = 0.15 message = f"Red injected a false CRITICAL alert on '{svc}'." else: red_reward = 0.0 message = "No healthy services to inject false alert on." elif action_type == "accelerate_spread": # Try to spread the attack to an adjacent service topology = env._task_cfg.get("topology", {}) new_victims = set() for affected in list(hidden.affected_services): for neighbor in topology.get(affected, []): if neighbor not in hidden.affected_services: new_victims.add(neighbor) if new_victims: new_svc = self._rng.choice(list(new_victims)) hidden.affected_services.append(new_svc) metrics[new_svc].cpu = min(100.0, metrics[new_svc].cpu + 25.0) metrics[new_svc].error_rate = min(100.0, metrics[new_svc].error_rate + 8.0) red_reward = 0.25 message = f"Red spread attack to new service '{new_svc}'!" else: red_reward = 0.05 message = "Red tried to spread — no adjacent services to infect." else: message = f"Unknown red action '{action_type}'" red_reward = 0.0 self._cumulative_red_reward += red_reward self._red_state.total_damage += red_reward done = env._state.done obs = env._build_observation(f"[ATTACKER] {message}") info = { "red_action": action_type, "red_reward": round(red_reward, 4), "red_cumulative": round(self._cumulative_red_reward, 4), "blue_cumulative": round(self._cumulative_blue_reward, 4), "message": message, } return self._build_ma_state(obs), red_reward, done, info def _heuristic_red_action(self) -> str: """ Adaptive Red Agent strategy — observes what Blue has been doing and actively counter-strategises. This simulates theory-of-mind: Red models Blue's beliefs and tries to mislead them. Strategy tiers (in priority order): 1. Counter-investigate: if Blue queries a service, plant false alert on different service 2. Amplify active cyber attacks before Blue can contain them 3. Accelerate spread if Blue is ignoring adjacent services 4. Corrupt metrics to create new false alarms 5. Inject noise as default confusion """ hidden = self._env._hidden metrics = self._env._metrics blue_actions = list(self._env._state.investigation_actions or []) + \ list(self._env._state.mitigation_actions or []) healthy = [s for s in metrics if s not in hidden.affected_services] # --- Tier 1: Counter-investigate (theory-of-mind) --- # If Blue just queried logs on a service, plant a false CRITICAL on a DIFFERENT service # to pull attention away from what Blue is zeroing in on if blue_actions and healthy: last_blue_action = blue_actions[-1] if blue_actions else "" # Extract service Blue is investigating blue_target = None for act in reversed(blue_actions[-3:]): # check last 3 actions parts = act.split(":", 1) if len(parts) == 2 and parts[1] in metrics: blue_target = parts[1] break # If Blue is focusing somewhere, create noise elsewhere if blue_target and healthy: distraction_targets = [s for s in healthy if s != blue_target] if distraction_targets and self._rng.random() < 0.55: return "create_false_alert" # --- Tier 2: Amplify active cyber attacks --- if hidden.true_root_cause == "cyber_attack" and hidden.attack_progress < 0.85: # Amplify more aggressively early in the episode if self._red_state.round <= 3: return "amplify_attack" # Continue amplifying if Blue hasn't isolated the affected service yet affected_isolated = any( f"isolate_service:{s}" in blue_actions for s in hidden.affected_services ) if not affected_isolated: return "amplify_attack" # --- Tier 3: Spread to adjacent services if Blue isn't checking them --- if hidden.true_root_cause == "cyber_attack": topology = self._env._task_cfg.get("topology", {}) spreadable = [] for affected in hidden.affected_services: for neighbor in topology.get(affected, []): if neighbor not in hidden.affected_services: # Only spread to services Blue hasn't investigated blue_investigated = any(neighbor in a for a in blue_actions) if not blue_investigated: spreadable.append(neighbor) if spreadable and self._rng.random() < 0.45: return "accelerate_spread" # --- Tier 4: Corrupt metrics if Blue is gaining a clear picture --- if healthy and self._rng.random() < 0.40: # Corrupt the service that Blue has inspected LEAST (anti-investigation) blue_investigated_counts = {s: 0 for s in metrics} for act in blue_actions: parts = act.split(":", 1) if len(parts) == 2 and parts[1] in blue_investigated_counts: blue_investigated_counts[parts[1]] += 1 # Target a healthy service Blue looked at (plant doubt) healthy_investigated = [s for s in healthy if blue_investigated_counts.get(s, 0) > 0] if healthy_investigated: return "corrupt_metric" # --- Tier 5: Default — inject noise --- return "inject_noise" def _build_ma_state(self, obs) -> dict[str, Any]: state = self._env.state.to_dict() state["multi_agent"] = { "red_round": self._red_state.round, "red_actions": self._red_state.actions_taken, "red_total_damage": round(self._red_state.total_damage, 4), "blue_cumulative_reward": round(self._cumulative_blue_reward, 4), "red_cumulative_reward": round(self._cumulative_red_reward, 4), } state["observation"] = { "alerts": obs.alerts, "metrics": obs.metrics, "logs": obs.logs, "topology": obs.topology, "last_action_result": obs.last_action_result, "time_step": obs.time_step, } return state @property def state(self) -> dict[str, Any]: obs = self._env._build_observation("state_query") return self._build_ma_state(obs) # --------------------------------------------------------------------------- # Curriculum — tracks Blue agent improvement across episodes # --------------------------------------------------------------------------- CURRICULUM_LEVELS = [ {"level": 1, "tasks": ["easy_memory_leak"], "threshold": 0.65}, {"level": 2, "tasks": ["easy_memory_leak", "medium_ddos_cascade"], "threshold": 0.70}, {"level": 3, "tasks": ["medium_ddos_cascade", "medium_hard_bad_deployment"], "threshold": 0.72}, {"level": 4, "tasks": ["medium_hard_bad_deployment", "hard_data_exfiltration"], "threshold": 0.75}, {"level": 5, "tasks": ["hard_data_exfiltration"], "threshold": 0.80}, ] class CurriculumManager: """ Self-Improvement loop: - Blue agent starts at level 1 (easy tasks) - When average score over last N episodes exceeds threshold → level up - Reward curves, improvements, and level transitions are logged """ def __init__(self) -> None: self.current_level: int = 1 self.episode_count: int = 0 self.score_history: list[dict[str, Any]] = [] self.level_up_history: list[dict[str, Any]] = [] self._rng = random.Random(0) self._window = 5 # rolling window for level-up check def get_next_task(self) -> dict[str, Any]: lvl_cfg = CURRICULUM_LEVELS[min(self.current_level - 1, len(CURRICULUM_LEVELS) - 1)] task_id = self._rng.choice(lvl_cfg["tasks"]) return TASKS[task_id] def record_score(self, task_id: str, score: float) -> None: self.episode_count += 1 self.score_history.append({ "episode": self.episode_count, "task_id": task_id, "score": round(score, 4), "level": self.current_level, }) self._maybe_level_up() def _maybe_level_up(self) -> None: if self.current_level >= len(CURRICULUM_LEVELS): return lvl_cfg = CURRICULUM_LEVELS[self.current_level - 1] window = [r["score"] for r in self.score_history[-self._window:]] if len(window) < self._window: return avg = sum(window) / len(window) if avg >= lvl_cfg["threshold"]: old_level = self.current_level self.current_level += 1 self.level_up_history.append({ "from_level": old_level, "to_level": self.current_level, "episode": self.episode_count, "avg_score": round(avg, 4), }) def summary(self) -> str: if not self.score_history: return "No episodes recorded yet." recent = self.score_history[-10:] avg = sum(r["score"] for r in recent) / len(recent) return ( f"Level {self.current_level}/{len(CURRICULUM_LEVELS)} | " f"{self.episode_count} episodes | " f"Recent avg score: {avg:.3f} | " f"Level-ups: {len(self.level_up_history)}" ) # --------------------------------------------------------------------------- # Global multi-agent state # --------------------------------------------------------------------------- _ma_sessions: dict[str, MultiAgentSecOpsEnv] = {} _curriculum = CurriculumManager() def _get_ma_session(session_id: str) -> MultiAgentSecOpsEnv: if session_id not in _ma_sessions: raise HTTPException( status_code=400, detail=f"Multi-agent session '{session_id}' not found. Call /multi/reset first.", ) return _ma_sessions[session_id] # --------------------------------------------------------------------------- # Request / Response schemas (Pydantic v2 compatible) # --------------------------------------------------------------------------- class ResetRequest(BaseModel): task_id: str = "easy_memory_leak" session_id: str = "" # defaults to task_id when empty class StepRequest(BaseModel): action_type: str parameters: dict[str, Any] = {} session_id: str = "default" class StepResponse(BaseModel): observation: dict[str, Any] reward: float done: bool info: dict[str, Any] = {} class StateResponse(BaseModel): state: dict[str, Any] class GradeResponse(BaseModel): score: float diagnosis_correct: float action_efficiency: float investigation_quality: float details: dict[str, Any] class MultiAgentResetRequest(BaseModel): task_id: str | None = None session_id: str = "default_ma" class RedActionRequest(BaseModel): action_type: str | None = None parameters: dict[str, Any] = {} session_id: str = "default_ma" class MultiAgentStepResponse(BaseModel): observation: dict[str, Any] reward: float done: bool info: dict[str, Any] = {} class CurriculumSummaryResponse(BaseModel): current_level: int total_episodes: int summary_text: str level_up_history: list[dict] # --------------------------------------------------------------------------- # App # --------------------------------------------------------------------------- app = FastAPI( title="OpenSecOpsEnv — Attacker vs Defender", description=( "SecOps incident response environment for RL agent evaluation. " "Simulates memory leaks, DDoS attacks, misconfiguration, and " "data exfiltration scenarios across 4 tasks of increasing difficulty. " "Features a live Red (Attacker) vs Blue (Defender) multi-agent battle " "with self-improving curriculum learning." ), version="0.4.0", ) app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"], ) @app.get("/", include_in_schema=False) async def root_redirect(): """Redirect root to the live dashboard.""" return RedirectResponse(url="/dashboard") @app.get("/health") def health() -> dict[str, str]: """Liveness probe.""" return {"status": "ok", "environment": "opensecops", "version": "0.4.0"} @app.get("/tasks") def list_tasks() -> dict[str, Any]: """List all available tasks with metadata.""" return { "tasks": [ { "id": tid, "difficulty": cfg["difficulty"], "max_steps": cfg["max_steps"], "description": cfg["description"][:120] + "...", "noise_level": cfg["noise_level"], "correct_label": cfg["correct_label"], } for tid, cfg in TASKS.items() ] } @app.post("/reset") def reset(req: Optional[ResetRequest] = None) -> dict[str, Any]: """Start a new episode. Returns the initial observation.""" if req is None: req = ResetRequest() session_id = req.session_id or req.task_id try: env = OpenSecOpsEnv() obs = env.reset(task_id=req.task_id) except ValueError as e: raise HTTPException(status_code=400, detail=str(e)) _sessions[session_id] = env global _default_session _default_session = session_id return { "session_id": session_id, "observation": { "alerts": obs.alerts, "metrics": obs.metrics, "logs": obs.logs, "topology": obs.topology, "last_action_result": obs.last_action_result, "time_step": obs.time_step, "available_actions": obs.available_actions, }, } @app.post("/step", response_model=StepResponse) def step(req: StepRequest) -> StepResponse: """Execute one action and receive the next observation + reward.""" sid = req.session_id if req.session_id != "default" else _default_session env = _get_session(sid) action = SecOpsAction( action_type=req.action_type, parameters=req.parameters, ) obs, reward, done, info = env.step(action) return StepResponse( observation={ "alerts": obs.alerts, "metrics": obs.metrics, "logs": obs.logs, "topology": obs.topology, "last_action_result": obs.last_action_result, "time_step": obs.time_step, "available_actions": obs.available_actions, }, reward=reward, done=done, info=info, ) @app.get("/state", response_model=StateResponse) def state(session_id: str = "") -> StateResponse: """Return the full internal state (for debugging / graders).""" sid = session_id or _default_session env = _get_session(sid) return StateResponse(state=env.state.to_dict()) @app.post("/grade", response_model=GradeResponse) def grade_episode(session_id: str = "") -> GradeResponse: """Grade the current (or most recently completed) episode.""" sid = session_id or _default_session env = _get_session(sid) result = grade(env.state.to_dict()) return GradeResponse( score=result.score, diagnosis_correct=result.diagnosis_correct, action_efficiency=result.action_efficiency, investigation_quality=result.investigation_quality, details=result.details, ) # --------------------------------------------------------------------------- # Multi-Agent Endpoints # --------------------------------------------------------------------------- @app.post("/multi/reset", response_model=StateResponse) def multi_reset(req: MultiAgentResetRequest) -> StateResponse: """Reset a multi-agent (Attacker vs Defender) session.""" sid = req.session_id _ma_sessions[sid] = MultiAgentSecOpsEnv() env = _ma_sessions[sid] # If task_id is "auto", curriculum decides the next task if req.task_id == "auto": task_cfg = _curriculum.get_next_task() tid = task_cfg["task_id"] else: tid = req.task_id or "hard_data_exfiltration" state_dict = env.reset(tid) return StateResponse(state=state_dict) @app.post("/multi/blue/step", response_model=MultiAgentStepResponse) def multi_blue_step(req: StepRequest) -> MultiAgentStepResponse: """Blue (Defender) agent step — investigate and mitigate.""" env = _get_ma_session(req.session_id) action = SecOpsAction( action_type=req.action_type, parameters=req.parameters, ) obs_dict, reward, done, info = env.blue_step(action) if done: # Record final score in curriculum for self-improvement g = grade(env._env.state.to_dict()) _curriculum.record_score(env._red_state.task_id, g.score) return MultiAgentStepResponse( observation=obs_dict, reward=reward, done=done, info=info, ) @app.post("/multi/red/step", response_model=MultiAgentStepResponse) def multi_red_step(req: RedActionRequest) -> MultiAgentStepResponse: """Red (Attacker) agent step — escalate / confuse / spread.""" env = _get_ma_session(req.session_id) # Use specified action or let heuristic agent decide action_type = None if req.action_type and req.action_type != "auto": action_type = req.action_type obs_dict, reward, done, info = env.red_step(action_type) return MultiAgentStepResponse( observation=obs_dict, reward=reward, done=done, info=info, ) @app.get("/multi/state") def multi_state(session_id: str = "default_ma") -> StateResponse: """Get the current multi-agent session state.""" env = _get_ma_session(session_id) return StateResponse(state=env.state) # --------------------------------------------------------------------------- # Curriculum Endpoints # --------------------------------------------------------------------------- @app.get("/curriculum/summary", response_model=CurriculumSummaryResponse) def curriculum_summary() -> CurriculumSummaryResponse: """Return cursor on the Blue agent's self-improvement journey.""" return CurriculumSummaryResponse( current_level=_curriculum.current_level, total_episodes=_curriculum.episode_count, summary_text=_curriculum.summary(), level_up_history=_curriculum.level_up_history, ) # --------------------------------------------------------------------------- # SSE Demo Stream — powers the live dashboard (single-agent) # --------------------------------------------------------------------------- # Heuristic playbooks (deterministic, expert agent) _HEURISTIC_PLAYBOOKS: dict[str, list[dict]] = { "easy_memory_leak": [ {"action_type": "inspect_metrics", "parameters": {}}, {"action_type": "query_logs", "parameters": {"service": "auth"}}, {"action_type": "inspect_metrics", "parameters": {"service": "auth"}}, {"action_type": "restart_service", "parameters": {"service": "auth"}}, {"action_type": "submit_diagnosis", "parameters": {"label": "infra_failure:memory_leak"}}, ], "medium_ddos_cascade": [ {"action_type": "inspect_metrics", "parameters": {}}, {"action_type": "query_logs", "parameters": {"service": "gateway"}}, {"action_type": "query_logs", "parameters": {"service": "api"}}, {"action_type": "run_security_scan", "parameters": {"target": "api"}}, {"action_type": "block_ip", "parameters": {"ip": "203.0.113.45"}}, {"action_type": "block_ip", "parameters": {"ip": "198.51.100.12"}}, {"action_type": "scale_service", "parameters": {"service": "api", "replicas": 5}}, {"action_type": "submit_diagnosis", "parameters": {"label": "cyber_attack:ddos"}}, ], "medium_hard_bad_deployment": [ {"action_type": "inspect_metrics", "parameters": {}}, {"action_type": "query_logs", "parameters": {"service": "api"}}, {"action_type": "query_logs", "parameters": {"service": "cache"}}, {"action_type": "rollback_deployment", "parameters": {"service": "api", "version": "previous"}}, {"action_type": "restart_service", "parameters": {"service": "cache"}}, {"action_type": "submit_diagnosis", "parameters": {"label": "misconfiguration:bad_config"}}, ], "hard_data_exfiltration": [ {"action_type": "inspect_metrics", "parameters": {}}, {"action_type": "query_logs", "parameters": {"service": "db"}}, {"action_type": "query_logs", "parameters": {"service": "auth"}}, {"action_type": "run_security_scan", "parameters": {"target": "db"}}, {"action_type": "run_security_scan", "parameters": {"target": "auth"}}, {"action_type": "isolate_service", "parameters": {"service": "db"}}, {"action_type": "block_ip", "parameters": {"ip": "10.0.0.99"}}, {"action_type": "submit_diagnosis", "parameters": {"label": "cyber_attack:data_exfiltration"}}, ], } # Bad agent: makes wrong choices to show contrast _BAD_AGENT_PLAYBOOKS: dict[str, list[dict]] = { "easy_memory_leak": [ {"action_type": "inspect_metrics", "parameters": {}}, {"action_type": "restart_service", "parameters": {"service": "api"}}, # wrong service {"action_type": "scale_service", "parameters": {"service": "db", "replicas": 3}}, # useless {"action_type": "block_ip", "parameters": {"ip": "8.8.8.8"}}, # harmful {"action_type": "submit_diagnosis", "parameters": {"label": "cyber_attack:ddos"}}, # wrong ], "medium_ddos_cascade": [ {"action_type": "inspect_metrics", "parameters": {}}, {"action_type": "restart_service", "parameters": {"service": "db"}}, # wrong service {"action_type": "block_ip", "parameters": {"ip": "8.8.8.8"}}, # wrong ip {"action_type": "isolate_service", "parameters": {"service": "cache"}}, # wrong service {"action_type": "submit_diagnosis", "parameters": {"label": "infra_failure:memory_leak"}}, ], "medium_hard_bad_deployment": [ {"action_type": "inspect_metrics", "parameters": {}}, {"action_type": "query_logs", "parameters": {"service": "auth"}}, {"action_type": "restart_service", "parameters": {"service": "auth"}}, # wrong {"action_type": "block_ip", "parameters": {"ip": "10.0.0.1"}}, # harmful {"action_type": "submit_diagnosis", "parameters": {"label": "cyber_attack:ddos"}}, ], "hard_data_exfiltration": [ {"action_type": "inspect_metrics", "parameters": {}}, {"action_type": "restart_service", "parameters": {"service": "cache"}}, # chasing false alert {"action_type": "query_logs", "parameters": {"service": "cache"}}, {"action_type": "isolate_service", "parameters": {"service": "cache"}}, # wrong, causes outage {"action_type": "submit_diagnosis", "parameters": {"label": "infra_failure:memory_leak"}}, ], } # Red (Attacker) heuristic playbook — interleaved with blue steps _RED_PLAYBOOK_BY_TASK: dict[str, list[str]] = { "easy_memory_leak": ["inject_noise", "corrupt_metric", "inject_noise"], "medium_ddos_cascade": ["amplify_attack", "create_false_alert", "amplify_attack", "inject_noise"], "medium_hard_bad_deployment": ["inject_noise", "corrupt_metric", "create_false_alert", "inject_noise"], "hard_data_exfiltration": ["amplify_attack", "accelerate_spread", "create_false_alert", "amplify_attack", "inject_noise"], } # --------------------------------------------------------------------------- # Debug: test the live AI endpoint directly # --------------------------------------------------------------------------- @app.get("/debug/ai") async def debug_ai(): """Test the live AI endpoint with a sample observation. Shows exact response.""" if not _TRAINED_ENDPOINT: return {"error": "No endpoint configured", "token_set": bool(_HF_API_TOKEN)} test_obs = { "alerts": [{"service": "db", "severity": "critical", "message": "Unusual outbound traffic"}], "metrics": {"db": {"cpu": 82.0, "memory": 71.0, "latency": 95.0, "error_rate": 1.2}}, "logs": ["[db] CRIT Unusual 8GB export to external host 10.0.0.99"], "topology": {"db": ["auth", "api"]}, "last_action_result": "", "time_step": 1, } import httpx prompt = _obs_to_text(test_obs, 1) payload = {"inputs": prompt, "parameters": {"max_new_tokens": 128, "temperature": 0.1, "return_full_text": False}} headers = {"Content-Type": "application/json"} if _HF_API_TOKEN: headers["Authorization"] = f"Bearer {_HF_API_TOKEN}" try: async with httpx.AsyncClient(timeout=30.0) as client: resp = await client.post(_TRAINED_ENDPOINT, json=payload, headers=headers) return { "status_code": resp.status_code, "token_set": bool(_HF_API_TOKEN), "endpoint": _TRAINED_ENDPOINT[:50], "response": resp.json() if resp.status_code == 200 else resp.text[:500], } except Exception as e: return {"error": str(e), "token_set": bool(_HF_API_TOKEN), "endpoint": _TRAINED_ENDPOINT[:50]} def _sse(data: dict) -> str: return f"data: {json.dumps(data)}\n\n" @app.get("/demo/stream") async def demo_stream( task_id: str = "easy_memory_leak", mode: str = "trained", # "trained" | "untrained" speed: float = 1.5, # seconds between steps ): """ Server-Sent Events stream of a full single-agent episode. Powers the live dashboard. mode=trained → calls TRAINED_MODEL_ENDPOINT (live AI) or heuristic fallback mode=untrained → calls UNTRAINED_MODEL_ENDPOINT (base model) or bad-agent fallback Fallback logic: if no endpoint is configured, uses deterministic playbooks so the dashboard always works — even before the endpoint is live. """ async def event_gen(): try: # Determine which endpoint to use if mode == "trained": endpoint = _TRAINED_ENDPOINT fallback_playbook = _HEURISTIC_PLAYBOOKS.get(task_id, []) agent_label = "🤖 Trained AI (Qwen2.5-7B-GRPO)" else: endpoint = _UNTRAINED_ENDPOINT fallback_playbook = _BAD_AGENT_PLAYBOOKS.get(task_id, []) agent_label = "⚠️ Untrained Base Model" live_ai = endpoint is not None env = OpenSecOpsEnv() obs = env.reset(task_id) # Randomise seed so every run differs (different metric drift, alert order, etc.) episode_seed = random.randint(0, 2**31) _randomise_env_seed(env, episode_seed) obs_dict = { "alerts": obs.alerts, "metrics": obs.metrics, "logs": obs.logs, "topology": obs.topology, "last_action_result": obs.last_action_result, "time_step": obs.time_step, } # Send initial state yield _sse({ "type": "reset", "task_id": task_id, "mode": mode, "agent_label": agent_label, "live_ai": live_ai, "observation": obs_dict, }) cumulative = 0.0 rewards_history: list[float] = [] step = 0 max_steps = TASKS[task_id]["max_steps"] done = False while not done and step < max_steps: await asyncio.sleep(max(0.5, speed)) step += 1 # --- Decide action --- action: Optional[SecOpsAction] = None ai_raw: str = "" if live_ai: action = await _query_ai_model(endpoint, obs_dict, step) if action is None: # AI call failed — fall back gracefully ai_raw = "[AI timeout — using fallback]" pb_idx = step - 1 if pb_idx < len(fallback_playbook): d = fallback_playbook[pb_idx] action = SecOpsAction( action_type=d["action_type"], parameters=d.get("parameters", {}), ) else: action = SecOpsAction(action_type="inspect_metrics", parameters={}) else: ai_raw = f"{{\"action_type\": \"{action.action_type}\", \"parameters\": {json.dumps(action.parameters)}}}" else: pb_idx = step - 1 if pb_idx < len(fallback_playbook): d = fallback_playbook[pb_idx] action = SecOpsAction( action_type=d["action_type"], parameters=d.get("parameters", {}), ) else: break obs, reward, done, info = env.step(action) obs_dict = { "alerts": obs.alerts, "metrics": obs.metrics, "logs": obs.logs, "topology": obs.topology, "last_action_result": obs.last_action_result, "time_step": obs.time_step, } cumulative += reward rewards_history.append(round(reward, 4)) yield _sse({ "type": "step", "step": step, "action_type": action.action_type, "parameters": action.parameters, "reward": round(reward, 4), "cumulative_reward": round(cumulative, 4), "rewards_history": rewards_history, "done": done, "live_ai": live_ai, "ai_raw": ai_raw, "observation": obs_dict, }) # Final grade await asyncio.sleep(0.5) result = grade(env.state.to_dict()) # Record in curriculum so the Learning tab advances _curriculum.record_score(task_id, result.score) yield _sse({ "type": "grade", "score": round(result.score, 4), "diagnosis_correct": result.diagnosis_correct, "action_efficiency": round(result.action_efficiency, 4), "investigation_quality": round(result.investigation_quality, 4), "details": result.details, "rewards_history": rewards_history, "cumulative_reward": round(cumulative, 4), "live_ai": live_ai, "agent_label": agent_label, "curriculum_level": _curriculum.current_level, "curriculum_episodes": _curriculum.episode_count, }) except Exception as e: yield _sse({"type": "error", "message": str(e)}) return StreamingResponse( event_gen(), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", "X-Accel-Buffering": "no", }, ) @app.get("/battle/stream") async def battle_stream( task_id: str = "hard_data_exfiltration", speed: float = 1.5, ): """ 🔴 vs 🔵 Attacker vs Defender live battle SSE stream. Interleaves Red (Attacker) and Blue (Defender) turns. Blue Agent uses TRAINED_MODEL_ENDPOINT when set (live AI), otherwise falls back to the heuristic expert playbook. """ async def event_gen(): try: live_ai = _TRAINED_ENDPOINT is not None ma_env = MultiAgentSecOpsEnv() state_dict = ma_env.reset(task_id) obs_dict = state_dict["observation"] yield _sse({ "type": "battle_reset", "task_id": task_id, "observation": obs_dict, "multi_agent": state_dict.get("multi_agent", {}), "live_ai": live_ai, "blue_label": "🤖 Trained AI" if live_ai else "🧠 Expert Heuristic", }) blue_playbook = copy.deepcopy( _HEURISTIC_PLAYBOOKS.get(task_id, _HEURISTIC_PLAYBOOKS["hard_data_exfiltration"]) ) red_actions = _RED_PLAYBOOK_BY_TASK.get( task_id, ["inject_noise", "amplify_attack", "corrupt_metric"] ) blue_rewards: list[float] = [] red_rewards: list[float] = [] blue_cum = 0.0 red_cum = 0.0 done = False max_steps = TASKS[task_id]["max_steps"] for round_num in range(max_steps): if done: break # ── Red turn (Attacker escalates) ──────────────────────────── red_action_type = ( red_actions[round_num % len(red_actions)] if red_actions else "inject_noise" ) await asyncio.sleep(max(0.3, speed * 0.4)) red_state, red_r, done, red_info = ma_env.red_step(red_action_type) red_cum += red_r red_rewards.append(round(red_r, 4)) obs_dict = red_state.get("observation", obs_dict) yield _sse({ "type": "red_step", "round": round_num + 1, "action": red_action_type, "reward": round(red_r, 4), "red_cumulative": round(red_cum, 4), "blue_cumulative": round(blue_cum, 4), "message": red_info.get("message", ""), "observation": obs_dict, "multi_agent": red_state.get("multi_agent", {}), "red_rewards": red_rewards, "blue_rewards": blue_rewards, }) if done: break # ── Blue turn (Defender responds) ──────────────────────────── await asyncio.sleep(max(0.5, speed * 0.6)) action: Optional[SecOpsAction] = None ai_raw: str = "" if live_ai: action = await _query_ai_model(_TRAINED_ENDPOINT, obs_dict, round_num + 1) if action: ai_raw = f'{{"action_type": "{action.action_type}", "parameters": {json.dumps(action.parameters)}}}' if action is None: # Fallback to heuristic for this round if round_num < len(blue_playbook): d = blue_playbook[round_num] action = SecOpsAction( action_type=d["action_type"], parameters=d.get("parameters", {}), ) else: break blue_state, blue_r, done, blue_info = ma_env.blue_step(action) blue_cum += blue_r blue_rewards.append(round(blue_r, 4)) obs_dict = blue_state.get("observation", obs_dict) yield _sse({ "type": "blue_step", "round": round_num + 1, "action": action.action_type, "parameters": action.parameters, "reward": round(blue_r, 4), "blue_cumulative": round(blue_cum, 4), "red_cumulative": round(red_cum, 4), "result": obs_dict.get("last_action_result", ""), "observation": obs_dict, "multi_agent": blue_state.get("multi_agent", {}), "red_rewards": red_rewards, "blue_rewards": blue_rewards, "live_ai": live_ai, "ai_raw": ai_raw, }) if done: break # Final grade await asyncio.sleep(0.5) result = grade(ma_env._env.state.to_dict()) winner = "defender" if blue_cum > red_cum else "attacker" # Record in curriculum so the Learning tab advances _curriculum.record_score(task_id, result.score) yield _sse({ "type": "battle_end", "score": round(result.score, 4), "diagnosis_correct": result.diagnosis_correct, "action_efficiency": round(result.action_efficiency, 4), "investigation_quality": round(result.investigation_quality, 4), "blue_cumulative": round(blue_cum, 4), "red_cumulative": round(red_cum, 4), "winner": winner, "blue_rewards": blue_rewards, "red_rewards": red_rewards, "details": result.details, "live_ai": live_ai, "curriculum_level": _curriculum.current_level, "curriculum_episodes": _curriculum.episode_count, }) except Exception as e: yield _sse({"type": "error", "message": str(e)}) return StreamingResponse( event_gen(), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", "X-Accel-Buffering": "no", }, ) # --------------------------------------------------------------------------- # Dashboard (served at /dashboard) # --------------------------------------------------------------------------- @app.get("/ai/status") async def ai_status(): """Returns whether the live AI endpoint is configured and reachable.""" endpoint = _TRAINED_ENDPOINT model_ready = False if endpoint: try: import httpx async with httpx.AsyncClient(timeout=5.0) as client: r = await client.get(endpoint.rstrip('/') + '/health') model_ready = r.status_code < 400 except Exception: model_ready = True # Endpoint is set, assume alive (TGI has no /health sometimes) return { "live_ai": endpoint is not None, "endpoint_configured": endpoint is not None, "endpoint_url": (endpoint or "").replace("https://", "")[:40] + "..." if endpoint else None, "model_name": "Qwen2.5-7B-GRPO (fine-tuned)", "model_url": "https://huggingface.co/SapphireGaze429/opensecops-qwen2.5-7b-grpo", "model_ready": model_ready, } @app.get("/dashboard", response_class=HTMLResponse) def dashboard() -> HTMLResponse: """Live AI Demo Dashboard — Attacker vs Defender + Self-Improvement.""" return HTMLResponse(content=_DASHBOARD_HTML) # --------------------------------------------------------------------------- # Optional simple debug Web UI # --------------------------------------------------------------------------- ENABLE_WEB = os.environ.get("ENABLE_WEB_INTERFACE", "false").lower() == "true" @app.get("/web", response_class=HTMLResponse, include_in_schema=ENABLE_WEB) def web_ui() -> HTMLResponse: """Simple debug interface.""" if not ENABLE_WEB: return HTMLResponse( content='

Debug UI disabled. Set ENABLE_WEB_INTERFACE=true to enable.

', status_code=200, ) return HTMLResponse(content=_build_web_ui()) def _build_web_ui() -> str: return """ OpenSecOpsEnv – Debug UI

OpenSecOpsEnv Debug Console

→ Open Live Dashboard

Reset

Step

Action type: Params (JSON):

Last Result

""" # --------------------------------------------------------------------------- # Dashboard HTML — Attacker vs Defender + Self-Improvement Theme # --------------------------------------------------------------------------- _DASHBOARD_HTML = r""" OpenSecOpsEnv — Attacker vs Defender
OpenEnv Curriculum RL
Server online
AI Endpoint may be cold-starting. The inference endpoint scales to zero when idle to save costs. If the first episode shows "AI timeout — using fallback", wait ~2 minutes and run again. Live AI inference will work once the endpoint is warm.
Scenario
Agent Mode
Speed
System State
Select a scenario and click
Run Episode to start.
Agent Action Feed
Agent actions appear here
during incident analysis.
// System logs will stream here...
Reward Curve
Final Score / 1.0
0.00
Cumul. Reward
0
Steps Taken
Diagnosis
Action Efficiency
Investigation Quality
Scenario
Speed
Defender vs Attacker
Live System State
Click Start Battle
to launch live simulation.
Battle Feed
Attacker and defender actions
stream here in real time.
// Combat logs will appear here...
Battle Score
0.00
Defender
0.00
Attacker
Episode Score / 1.0
Defender Advantage
Attack Suppression
Battle Rounds0
Curriculum Progress
Run episodes to see
the self-improvement curriculum.
Learning Curve & Score History
Level-Up Events
No level-ups yet. Run episodes to progress!
How Self-Improvement Works
1. Start at Level 1 with foundational scenarios.
2. Track score over a rolling window of five episodes.
3. Promote level automatically when thresholds are met.
4. Increase attacker pressure as defender capability improves.
5. Level 5 reflects expert readiness for disguised exfiltration scenarios.
Result
Episode Complete
"""