Running-lab operations

Running-lab operations (Scenario API)

This guide covers operations on a running lab via the Scenario API: launching a lab wired to a defense integration (a log-collector pipeline such as Splunk), then retrieving the attack report and the security alerts the attacks triggered.

Prerequisites: an access token and a workspace id — see the Authentication guide. Throughout:

export BASE="https://app.mantis-platform.io/api/scenario/lab"
export TOKEN="<your access token>"
export WORKSPACE_ID="<your workspace id>"

1. Launch a lab with a defense integration (Splunk)

Defense integrations are deployed by passing a log_collectors list inside lab_config. A collector pipeline chains agents (that ship logs) → an aggregator (e.g. logstash) → a SIEM (e.g. Splunk).

Discover the available collectors

GET /log_collector returns the catalog, including each collector's allowed location types, mandatory_inputs, and any user_config keys.

cURL

curl "$BASE/log_collector" \
  -H "Authorization: Bearer $TOKEN" \
  -H "X-Workspace-Id: $WORKSPACE_ID"

Python

import requests

BASE = "https://app.mantis-platform.io/api/scenario/lab"
HEADERS = {"Authorization": f"Bearer {TOKEN}", "X-Workspace-Id": WORKSPACE_ID}

for c in requests.get(f"{BASE}/log_collector", headers=HEADERS).json():
    print(c["collector_name"], c["collector_type"], "requires:", c["mandatory_inputs"])

Build the Splunk pipeline

A working Splunk setup is winlogbeat (agent, on Windows) → logstash (aggregator) → splunk (aggregator). Splunk requires a logstash input and takes no user_config (its default UI credentials are admin / AmossysSPLUNK35;). Each entry's output names the next instance in the chain.

[
  {
    "instance_name": "winlogbeat_windows",
    "collector_name": "winlogbeat",
    "collector_type": "agent",
    "location": [{ "location_type": "system_type", "value": "windows" }],
    "output": [{ "instance_name": "logstash01", "collector_name": "logstash", "collector_type": "aggregator" }]
  },
  {
    "instance_name": "logstash01",
    "collector_name": "logstash",
    "collector_type": "aggregator",
    "location": [{ "location_type": "new_node", "value": "logstash01" }],
    "output": [{ "instance_name": "splunk01", "collector_name": "splunk", "collector_type": "aggregator" }]
  },
  {
    "instance_name": "splunk01",
    "collector_name": "splunk",
    "collector_type": "aggregator",
    "location": [{ "location_type": "new_node", "value": "splunk01" }],
    "output": []
  }
]

Optionally validate the wiring with POST /log_collector/config_check (send the bare array above; it returns {"requirements_met": bool, "errors": {...}}).

Launch with the pipeline

Embed the array as lab_config.log_collectors and call run_lab:

cURL

curl -X POST "$BASE/scenario/run_lab?workspace_id=$WORKSPACE_ID" \
  -H "Authorization: Bearer $TOKEN" \
  -H "X-Workspace-Id: $WORKSPACE_ID" \
  -H "Content-Type: application/json" \
  -d '{
        "lab_config": {
          "content_type": "KILLCHAIN",
          "content_name": "my-scenario",
          "scenario_profile": "default",
          "log_collectors": [ /* the array above */ ]
        }
      }'

Python

splunk_pipeline = [ ... ]  # the array above

lab_config = {
    "content_type": "KILLCHAIN",
    "content_name": "my-scenario",
    "scenario_profile": "default",
    "log_collectors": splunk_pipeline,
}
resp = requests.post(
    f"{BASE}/scenario/run_lab",
    params={"workspace_id": WORKSPACE_ID},
    json={"lab_config": lab_config},
    headers=HEADERS,
)
resp.raise_for_status()
lab_id = resp.json()
print("Lab launched:", lab_id)

2. Retrieve the attack report

GET /runner/{lab_id}/attack_report returns a JSON array of attack steps that have played. Each entry carries an id, status, started_date / last_update, the target_nodes involved, and (for attack steps) a worker describing the technique (MITRE data, title, …).

It is populated once attacks start running (lab status SCENARIO_EXECUTION or later) and is empty ([]) before that — there is no error while "not ready", so poll on the lab status.

cURL

curl "$BASE/runner/$LAB_ID/attack_report" -H "Authorization: Bearer $TOKEN"

Python

import time

def wait_for_report(lab_id):
    while True:
        report = requests.get(f"{BASE}/runner/{lab_id}/attack_report", headers=HEADERS).json()
        if report:
            return report
        time.sleep(10)

for step in wait_for_report(lab_id):
    print(step["id"], step["status"], step.get("worker", {}).get("title"))

3. Retrieve the security alerts

GET /runner/{lab_id}/security_alerts returns {"<product>": [<SecurityAlert>, ...]} — one key per configured SIEM/EDR product (e.g. "Splunk"). Each alert links back to the attack report via correlated_attacks (a list of attack-step ids); an empty list can indicate a false positive.

Alerts are produced by post-execution correlation, so they appear only after the lab finishes (SCENARIO_FINISHED), only if an alert-capable SIEM/EDR was deployed, and on a periodic cadence (~every 50s) subject to SIEM ingestion latency. Until then the endpoint returns {} — poll.

cURL

curl "$BASE/runner/$LAB_ID/security_alerts" -H "Authorization: Bearer $TOKEN"

Python

alerts = requests.get(f"{BASE}/runner/{lab_id}/security_alerts", headers=HEADERS).json()
for product, product_alerts in alerts.items():
    for a in product_alerts:
        print(product, a["alert_name"], a["alert_status"], "-> attacks", a["correlated_attacks"])

Example response:

{
  "Splunk": [
    {
      "alert_name": "Suspicious PowerShell Execution",
      "alert_status": "NEW",
      "signature_name": "Encoded PowerShell command",
      "mitre_data": { "technique": {"id": "T1059.001", "name": "PowerShell"} },
      "asset_hostname": "WIN10",
      "start_time": "2025-10-08T20:51:25+00:00",
      "end_time": "2025-10-08T20:51:59+00:00",
      "correlated_attacks": [3828]
    }
  ]
}

Next step

To play red-team actions (custom commands, Atomic Red Team) on the running lab, see Red-team operations.