""" FastAPI Server — SOC Simulator OpenEnv HTTP Interface ====================================================== Exposes the standard OpenEnv endpoints: POST /reset → SOCObservation POST /step → SOCObservation (with reward, done, info) GET /state → SOCState GET /score → final grader score + explanation GET /health → liveness probe Also serves the built-in web UI at GET /web (when ENABLE_WEB_INTERFACE=true). """ from __future__ import annotations import os import sys sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) from typing import Any, Dict, Optional import uvicorn from fastapi import FastAPI, HTTPException from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import HTMLResponse from pydantic import BaseModel from env.soc_environment import SOCEnvironment from env.dynamic_input import DynamicInputPipeline from env.red_agent import RedAgent, BlueMemory from models import SOCAction, SOCObservation, SOCState, CustomParams # --------------------------------------------------------------------------- # App setup # --------------------------------------------------------------------------- app = FastAPI( title="SOC Simulator — AI Cybersecurity Incident Response Environment", description=( "An OpenEnv-compliant environment that simulates a real-world Security " "Operations Center (SOC). The AI agent analyses logs, detects threats, " "and takes response actions across three difficulty levels." ), version="1.0.0", docs_url="/docs", redoc_url="/redoc", ) app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"], ) # Single shared environment instance (thread-safe for single-agent usage) _env = SOCEnvironment() # Shared dynamic-input pipeline and red agent (persist across episodes) _pipeline = DynamicInputPipeline() _red_agent = RedAgent() # --------------------------------------------------------------------------- # Request / Response schemas # --------------------------------------------------------------------------- class ResetRequest(BaseModel): task_id: str = "easy_phishing_login" episode_id: Optional[str] = None seed: Optional[int] = None custom_params: Optional[CustomParams] = None class StepRequest(BaseModel): action: SOCAction class StepResponse(BaseModel): observation: SOCObservation reward: float done: bool info: Dict[str, Any] # --------------------------------------------------------------------------- # Endpoints # --------------------------------------------------------------------------- @app.get("/", include_in_schema=False) async def root(): """Welcome point for the API.""" return { "status": "✅ SOC Simulator OpenEnv API is perfectly running!", "message": "Visit /docs to see the interactive API payload Swagger UI, or call /health for liveness probe." } @app.get("/health") async def health(): """Liveness probe — used by HF Space and Docker healthchecks.""" return {"status": "ok", "env": "soc_simulator", "version": "1.0.0"} @app.post("/reset", response_model=SOCObservation) async def reset(request: Optional[ResetRequest] = None): """ Reset the environment and start a new episode. **task_id options:** - `easy_phishing_login` — Detect phishing + malicious login - `medium_brute_force_geo` — Correlate brute-force + geo-anomaly - `hard_apt_multistage` — Multi-stage APT kill chain **custom_params (all optional):** - `attacker_ip` — Override primary attacker IP - `target_user` — Override compromised username - `attack_intensity` — [0–1] attack severity multiplier - `enable_red_agent` — Activate adaptive RedAgent scenario mutation - `use_live_threat_intel` — Pull attacker IPs from live threat feeds """ if request is None: request = ResetRequest() # --- Source C: build base kwargs dict and merge custom_params overrides --- reset_kwargs: Dict[str, Any] = { "task_id": request.task_id, "episode_id": request.episode_id, "seed": request.seed, } cp = request.custom_params or CustomParams() # Merge non-None custom param fields into reset_kwargs custom_dict = cp.model_dump(exclude_none=True) reset_kwargs = _pipeline.merge_custom_params(reset_kwargs, custom_dict) # --- Source A: live threat intel --- if cp.use_live_threat_intel: live_ips = _pipeline.get_live_threat_ips() # Surface live IPs in kwargs so env / red_agent can use them if live_ips: reset_kwargs["live_threat_ips"] = live_ips # --- RedAgent mutation --- if cp.enable_red_agent and _env.state is not None: # Extract blue memory from the *previous* episode's state prev_state = _env.state prev_score = 0.0 try: prev_score = _env.get_final_score() except Exception: pass blue_mem = RedAgent.extract_blue_memory( prev_state.agent_actions, prev_score ) mutations = _red_agent.get_mutated_scenario( request.task_id, blue_mem ) # Merge red-agent mutations (don't override explicit custom_params) for k, v in mutations.items(): if k not in reset_kwargs or reset_kwargs.get(k) is None: reset_kwargs[k] = v try: obs = _env.reset(**reset_kwargs) return obs except ValueError as e: raise HTTPException(status_code=422, detail=str(e)) @app.post("/step", response_model=StepResponse) async def step(request: StepRequest): """ Execute one action in the environment. **action_type options:** block_ip | flag_user | isolate_host | ignore | escalate_alert For block_ip, flag_user, isolate_host — set `target` to the IP / user / host. """ if _env.state is None: raise HTTPException( status_code=400, detail="Environment not initialised. Call /reset first.", ) obs = _env.step(request.action) return StepResponse( observation=obs, reward=obs.reward, done=obs.done, info=obs.metadata.get("step_info", {}), ) @app.get("/state", response_model=SOCState) async def state(): """Return the current internal episode state (for debugging / inspection).""" if _env.state is None: raise HTTPException(status_code=400, detail="No active episode. Call /reset first.") return _env.state @app.get("/score") async def score(): """Run the grader on the current episode and return a [0, 1] score.""" if _env.state is None: raise HTTPException(status_code=400, detail="No active episode. Call /reset first.") try: final_score = _env.get_final_score() explanation = _env.get_score_explanation() # Source B: record score into pipeline for rolling-avg difficulty adaptation _pipeline.record_episode_score(final_score) # Log undetected attack if the blue agent largely failed if final_score < 0.5: _red_agent.record_undetected_attack() # Red agent considers an attack undetected if it results in a false negative # for the blue agent. The environment's score effectively reflects blue agent success. # But for exact red agent score, since the script relies on tracking it: return { "score": final_score, "explanation": explanation, "red_score": _red_agent.red_score, "mitre_techniques": explanation.get("mitre_tactics", []) } except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @app.get("/explain") async def explain(): """Return a detailed breakdown of the score components and agent behavior.""" if _env.state is None: raise HTTPException(status_code=400, detail="No active episode. Call /reset first.") try: explanation = _env.get_score_explanation() return explanation except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @app.get("/threat-intel") async def threat_intel(): """ Inspect live threat-intel cache status and trigger a refresh. Returns the current malicious IP cache metadata and a sample of threat IPs from URLhaus / Feodo Tracker. """ ips = _pipeline.get_live_threat_ips() summary = _pipeline.get_threat_intel_summary() summary["total_ips_available"] = len(ips) return summary @app.get("/difficulty") async def difficulty(): """ Return the current performance-based difficulty recommendation. Based on the rolling average of the last ≤5 episode scores. """ params = _pipeline.adapt_difficulty() params["score_history"] = _pipeline.score_history return params # --------------------------------------------------------------------------- # Optional Web UI # --------------------------------------------------------------------------- ENABLE_WEB = os.getenv("ENABLE_WEB_INTERFACE", "true").lower() == "true" @app.get("/web", response_class=HTMLResponse) async def web_interface(): """Interactive web UI for manual environment exploration.""" if not ENABLE_WEB: return HTMLResponse( content="""

Web Interface Disabled

Set ENABLE_WEB_INTERFACE=true to enable.

→ Use Swagger UI instead

""", status_code=200, ) return HTMLResponse(content=_build_web_ui()) def _build_web_ui() -> str: """Build the interactive SOC dashboard HTML with a military-grade aesthetic.""" return """ SOC Simulator — Dark Ops Console
SOC SIMULATOR / DARK OPS
THREAT LEVEL: LOW
STANDBY --:--:-- USR: ANALYST-01
+0.0%
0%
Risk Level
+0
0
Active Connections
+0
0
Blocked IPs
+0
0
Isolated Hosts
Threat Map / Attack Surface LIVE
Alerts / Events Bulk Ops Ready
Severity Title Timestamp Description
No active anomalies detected.
Incident Timeline Expandable Nodes
Threat Analytics Adaptive Metrics
Risk Trajectory (Area)
Severity Distribution (Bars)
Current Score
—
Critical High Medium Low Info

Raw Event Stream LIVE 
Initialize the environment to begin stream capture...
""" # --------------------------------------------------------------------------- # Entry point # --------------------------------------------------------------------------- def main(): port = int(os.getenv("PORT", "8000")) uvicorn.run("server.app:app", host="0.0.0.0", port=port, reload=False) if __name__ == "__main__": main()