DOCSEARCHbyAlgolia
Submit Docs

Agentic SOC Parallel Simulation (ASPS)

repository·main·Indexed 41 days ago

https://github.com/safelineman/agentic-soc-simulation

A multi-agent security operations center simulation platform using Large Action Models (LAM) and the DeepSeek reasoning engine. ASPS automates the security lifecycle from attack generation to incident response using seven specialized agent roles, an MCP server for tool execution (including VirusTotal and firewall simulation), and the OCSF standard for data modeling. It features a Streamlit dashboard and a Human-in-the-Loop (HITL) mechanism for approving high-risk operations.

Tokens
4K
Snippets
11
Records
19
Agent score
51%

What's inside ASPS

  1. How the Multi-Agent SOC Team works

    main

    ASPS simulates a complete SOC ecosystem using seven specialized AI agents that collaborate via shared context. This "swarm" intelligence allows the system to cover the entire security lifecycle.

    The Seven Agent Roles:

    • Purple Team (😈 紫队工程师): Generates high-fidelity attack traffic and background noise using the OCSF standard based on playbooks.
    • Detection Eng (🛠️ 检测工程师): Analyzes telemetry data to automatically develop and validate detection rules.
    • Analysis Engine (⚙️ 分析引擎): Scans the security data warehouse and matches telemetry against detection rules to trigger alerts.
    • Triage (🔍 分诊): Filters false positives and assesses the credibility of alerts.
    • Forensics (🕵️‍♂️ 取证): Performs deep investigations by calling tools (e.g., IP reputation, payload analysis) to build attack chains.
    • Commander (🛡️ 指挥官): Formulates response strategies and executes actions like blocking IPs.
    • Reporter (📝 报告): Summarizes investigation findings into structured security reports.

    Human-in-the-Loop (HITL) Mechanism

    The system uses an asynchronous PENDING_APPROVAL mechanism for high-risk actions (e.g., network-wide blocking). The AI submits a request and continues other tasks, while the human analyst approves or denies the action via the dashboard.

  2. How the Human-in-the-Loop (HITL) Mechanism Works

    main

    ASPS implements an Async Human-AI Symbiosis model to ensure safety during high-risk operations (e.g., network-wide bans).

    When an agent (like the Commander) determines a response is necessary, it does not execute the action immediately. Instead:

    1. The agent triggers a PENDING_APPROVAL status.
    2. The request is submitted to the dashboard's "Human Decision" area.
    3. The AI continues with other non-blocking tasks.
    4. A human analyst must manually click "Approve" or "Reject" on the interface to finalize the action.

  3. Understand the ASPS System Architecture

    main

    The ASPS platform is organized into three primary layers that simulate a real-world enterprise SOC environment:

    1. Dashboard (Streamlit): The visual interface (app/main.py) used by analysts to monitor the Detection Engineering Workbench, the Security Data Warehouse, and the War Room (Knowledge Graph).
    2. MCP Server (FastAPI): The "hands and feet" of the agents (mcp_server/main.py). It provides standard APIs for security tools like VirusTotal, Firewall simulation, and Graph management.
    3. AI Agent Core (Python): The "brain" (agent/) containing the logic for the 7 specialized roles. It uses the OCSF (Open Cybersecurity Schema Framework) standard for data modeling and an Analysis Engine for rule matching.

    The 7 Specialized Agent Roles

    • Purple Team: Generates high-fidelity attack traffic (OCSF standard).
    • Detection Eng: Analyzes telemetry to develop and verify detection rules.
    • Analysis Engine: Scans the data warehouse and triggers alerts.
    • Triage: Filters false positives and assesses alert credibility.
    • Forensics: Investigates alerts and constructs attack chains using MCP tools.
    • Commander: Formulates and executes response strategies.
    • Reporter: Generates structured security reports.

  4. System Architecture and Components

    main

    The ASPS platform is organized into a layered architecture consisting of a User Interface, an Agent Core, a Data Layer, and an MCP Tool Layer.

    Core Components:

    • Streamlit Dashboard (app/main.py): The visual command center for analysts. It includes the Detection Engineering Workbench, Security Data Warehouse view, and the War Room for real-time investigation and threat graphing.
    • MCP Server (mcp_server/main.py): A FastAPI-based server that provides standard APIs for agents to interact with the outside world (e.g., VirusTotal, Firewall simulation, Knowledge Graph engine).
    • AI Agent Core (agent/):
      • core.py: Defines the BaseAgent class and the seven specialized roles.
      • engine.py: The analysis engine responsible for rule matching and alert triggering.
      • ocsf.py: Defines the data models based on the Open Cybersecurity Schema Framework (OCSF).
    • Security Data Warehouse: A lightweight internal storage supporting OCSF standard data for telemetry and attack logs.

  5. Quick Start: Install and Run ASPS

    main

    To get the Agentic SOC Parallel Simulation (ASPS) platform running, follow these steps to prepare your environment and start the necessary services.

    1. Environment Preparation

    Clone the repository and install the required Python dependencies:

    pip install -r requirements.txt

    2. Start Services

    You must run two separate services in two different terminal windows.

    Terminal 1: Start the MCP Tool Server This server acts as the "hands and eyes" for the agents, providing APIs for IP reputation, firewall control, and graph management.

    python3 -m uvicorn mcp_server.main:app --reload --port 8000

    Terminal 2: Start the Web Dashboard This launches the Streamlit-based user interface for monitoring and interaction.

    python3 -m streamlit run app/main.py

    3. Configuration and Usage

    1. Access the UI: Open http://localhost:8501 in your browser.
    2. Configure API Keys: In the left sidebar under "系统配置" (System Configuration), enter your DeepSeek API Key and VirusTotal API Key (optional). These are session-based and do not require local file modifications.
    3. Run a Simulation: In the "♾️ 平行仿真演练 (Parallel Simulation)" section, enter an attack playbook or use the default, then click "🚀 启动全流程演练" (Start Full Process Drill).
    pip install -r requirements.txt
python3 -m uvicorn mcp_server.main:app --reload --port 8000
python3 -m streamlit run app/main.py

  6. Configure API Keys and Run Simulations

    main

    Once the services are running, access the dashboard at http://localhost:8501 to configure the system and begin a simulation.

    Configuration

    In the "System Configuration" sidebar, enter your credentials:

    • DeepSeek API Key: Required for the reasoning engine.
    • VirusTotal API Key: Optional, used for IP reputation tools. Note: Configuration is session-based and does not require modifying local files.

    Running a Parallel Simulation

    1. Locate the "♾️ Parallel Simulation" sidebar area.
    2. Enter an attack scenario (e.g., Lazarus APT attack) or use the default.
    3. Click "🚀 Start Full Simulation".

    The system will autonomously execute the full lifecycle: Data Generation $\rightarrow$ Rule Development $\rightarrow$ Engine Detection $\rightarrow$ SOC Response workflow.

  7. Handle high-risk actions with `ask_human_approval`

    main

    Certain operations, such as blocking an IP via firewall_block_ip, are considered high-risk.

    • firewall_block_ip: Always returns a status of PENDING_APPROVAL: Block IP <ip> | Reason: High-risk operation.... It does not perform the block automatically.
    • ask_human_approval: A dedicated tool to request permission for any action.
      • action (string): The intended operation.
      • reason (string): The justification.

    Agents should interpret any result starting with PENDING_APPROVAL as a signal to wait for human intervention.

  8. Project Structure Reference

    main

    The repository is organized as follows:

    • app/: Frontend Interface (Streamlit)
      • main.py: UI Layout, State Management, Visualization
    • mcp_server/: Tool Service (FastAPI)
      • main.py: APIs for IP Reputation, Firewall, Graph Management, etc.
    • agent/: Agent Core Logic
      • core.py: BaseAgent and specialized Agent roles
      • engine.py: Analysis Engine (rule matching and alert triggering)
      • ocsf.py: OCSF Telemetry Data Model Definition
    • requirements.txt: Project Dependencies

  9. Project Directory Structure

    main
    .
├── app/                # Frontend (Streamlit)
│   └── main.py         # UI layout, state management, and visualization
├── mcp_server/         # Tooling Service (FastAPI)
│   └── main.py         # APIs for IP reputation, firewall, and graph management
├── agent/              # Agent Core Logic
│   ├── core.py         # BaseAgent class and the seven specialized roles
│   ├── engine.py       # Analysis engine for rule matching and alerts
│   └── ocsf.py         # OCSF telemetry data model definitions
├── requirements.txt    # Project dependencies
└── README.md           # Project documentation

  10. Implement a Stream Callback for Real-time Logging

    main

    To capture and display the agent's reasoning process in real-time (e.g., in a Streamlit UI), provide a callback function that accepts a single log entry string. This allows the UI to update the 'Chain of Thought' or 'Agent History' components as the agent works.

    def stream_callback(log_entry):
    # Append log to session state or print to console
    print(f"[LOG]: {log_entry}")
    # In Streamlit, you might do:
    # st.session_state['agent_history'].append(log_entry)

  11. Run a Full Purple Team Simulation Workflow

    main

    The simulation follows a three-step 'Purple Team' workflow to validate detection capabilities:

    1. Generate Telemetry: Use run_purple_step1_gen_data(scenario, stream_callback) to generate attack telemetry based on a provided scenario string.
    2. Develop Rules: Use run_purple_step2_dev_rules(telemetry, scenario, stream_callback) to automatically create detection rules from the generated telemetry.
    3. Engine Scan: Use run_purple_step3_engine_scan(rules, telemetry, stream_callback) to run the analysis engine and identify alerts that trigger the newly developed rules.

    Each method accepts a stream_callback function to provide real-time logging updates to the UI.

    # Example workflow
telemetry = agent.run_purple_step1_gen_data(scenario, stream_callback=my_callback)
rules = agent.run_purple_step2_dev_rules(telemetry, scenario, stream_callback=my_callback)
alerts = agent.run_purple_step3_engine_scan(rules, telemetry, stream_callback=my_callback)

  12. Process Alerts using the Agent's Chain of Thought

    main

    Once alerts are identified, you can process them through the agent's reasoning engine using the think method. This simulates the SOC team's investigation process.

    Method Signature: think(alert, stream_callback=None, skip_report=False) -> (status, history, report)

    • alert: The alert object/dictionary to investigate.
    • stream_callback: A function called during the reasoning process to stream logs.
    • skip_report: Boolean to determine if a report should be generated immediately.

    Returns:

    • status: A string indicating the current state (e.g., "PENDING_APPROVAL_ASYNC: ..." if human intervention is needed).
    • history: The reasoning history.
    • report: The generated investigation report (if applicable).
    status, history, report = agent.think(alert, stream_callback=stream_callback, skip_report=True)

if "PENDING_APPROVAL_ASYNC" in status:
    # Handle human-in-the-loop requirement
    pass