Playbook (SOAR)

Playbook is the SOAR (Security Orchestration, Automation and Response) engine of CSIRT-Pro. Build visual automation flows with a drag-and-drop editor to automate detection, triage, notification, and response workflows.

Response actions are approval-gated

A playbook can enrich, evaluate, and notify on its own. It does not block traffic, remediate, or close a case automatically. Any change a playbook makes to an external system is something you script explicitly, and SOAR write actions in CSIRT-Pro follow a request, approve, and apply flow.

---

Screen Layout

The Playbook editor has the following components.

1. Flow editor, a visual canvas for placing and connecting nodes.
2. Code editor for writing Python or JavaScript in each node.
3. Test panel for running the playbook with sample data.
4. Execution history for past execution results and logs.
5. Version control for browsing and restoring playbook versions.
6. Environment variables for secrets and configuration values.

---

Core Concepts

Node Types

Type	Type ID	Description
Action	1	Execute logic such as HTTP requests, data transformations, API calls, and notifications
Condition (If/Else)	2	Branch the flow on a true or false evaluation
Loop	4	Iterate over a collection of items

Trigger Types

A playbook can be triggered automatically by the following events.

Trigger	Type ID	Description
Manual	--	Run from the UI or through the API
Webhook	--	Triggered by an external HTTP POST request
Schedule	104	Cron-based periodic execution
Case Creation	--	Runs automatically when a new case is created

---

Creating a Playbook

1. Create a New Playbook

From the Playbook list screen, click Create New and assign a name.

2. Add Nodes

1. Right-click the flow editor canvas and select Add Node.
2. Choose a node type: Action, Condition, or Loop.
3. Assign a name to the node.

3. Write Code

Click a node to open the code editor. Write Python or JavaScript to implement the node's logic.

Key variables available in every node:

Variable	Description
input_data	Data passed from the previous node, or the trigger payload
env	Dictionary of environment variables
output	Assign the node's output to this variable

4. Connect Nodes

Drag connections between nodes to define the execution flow.

5. Configure a Trigger

Set the trigger for the playbook.

# Schedule examples (cron format)
*/5 * * * *    # Every 5 minutes
0 * * * *      # Every hour
0 0 * * *      # Every day at midnight
0 9 * * 1      # Every Monday at 9:00 AM

---

Node Type Details

Action Node (type = 1)

Runs arbitrary code. The result is stored in output and passed to the next node as input_data.

{
  "type": 1,
  "name": "send_slack_alert",
  "code": "import requests\n\nresponse = requests.post(\n    'https://slack.com/api/chat.postMessage',\n    headers={'Authorization': f'Bearer {env[\"SLACK_TOKEN\"]}'},\n    json={'channel': '#alerts', 'text': f'Alert: {input_data[\"message\"]}'}\n)\n\noutput = response.json()"
}

Condition Node (type = 2)

Evaluates a boolean expression. The flow branches to the true or false path based on the result.

{
  "type": 2,
  "name": "check_severity",
  "code": "output = input_data['severity'] == 'high'",
  "true": [
    {"type": 1, "name": "escalate", "code": "..."}
  ],
  "false": [
    {"type": 1, "name": "log_only", "code": "..."}
  ]
}

Loop Node (type = 4)

Iterates over an array. Each item is processed in turn by the child nodes.

{
  "type": 4,
  "name": "process_each_ip",
  "code": "output = input_data['ip_list']"
}

---

Code Examples

Python: HTTP Request

import requests

response = requests.post(
    "https://slack.com/api/chat.postMessage",
    headers={"Authorization": f"Bearer {env['SLACK_TOKEN']}"},
    json={
        "channel": "#security-alerts",
        "text": f"New alert: {input_data['message']}"
    }
)

output = response.json()

Python: Query an External API

import requests

ip = input_data['src_ip']
response = requests.get(
    f"https://api.abuseipdb.com/api/v2/check",
    headers={
        "Key": env['ABUSEIPDB_KEY'],
        "Accept": "application/json"
    },
    params={"ipAddress": ip, "maxAgeInDays": 90}
)

result = response.json()
output = {
    "ip": ip,
    "abuse_score": result['data']['abuseConfidenceScore'],
    "is_malicious": result['data']['abuseConfidenceScore'] > 80
}

Python: Create a Case via API

import requests

response = requests.post(
    f"https://{env['CSIRT_DOMAIN']}/api/user/{env['USER_ID']}/messages/",
    headers={"Cookie": f"jwt={env['ACCESS_TOKEN']}"},
    json={
        "title": f"Suspicious activity from {input_data['src_ip']}",
        "description": f"AbuseIPDB score: {input_data['abuse_score']}",
        "status": "open",
        "tags": {"severity": "high", "source": "playbook"}
    }
)

output = response.json()

JavaScript: Webhook Notification

const response = await fetch('https://hooks.example.com/webhook', {
    method: 'POST',
    headers: {'Content-Type': 'application/json'},
    body: JSON.stringify({
        event: 'security_alert',
        message: input_data.message,
        severity: input_data.severity
    })
});

output = await response.json();

JavaScript: Data Transformation

const logs = input_data.logs;

const summary = logs.reduce((acc, log) => {
    const ip = log.src_ip;
    if (!acc[ip]) acc[ip] = 0;
    acc[ip]++;
    return acc;
}, {});

const topIPs = Object.entries(summary)
    .sort((a, b) => b[1] - a[1])
    .slice(0, 10)
    .map(([ip, count]) => ({ ip, count }));

output = { top_ips: topIPs };

---

Environment Variables

Store secrets and configuration values securely as environment variables.

Setting Environment Variables

1. Open the Environment Variables tab in the Playbook editor.
2. Enter a key and value, then save.
3. Access it in code with env['KEY_NAME'].

Security

Environment variables are encrypted at rest on the server side. Values are masked in the UI.

Common Environment Variables

Key	Example Value	Usage
SLACK_TOKEN	xoxb-xxx-xxx	Slack Bot token for notifications
ABUSEIPDB_KEY	abc123...	AbuseIPDB API key for IP reputation checks
CSIRT_DOMAIN	app.csirt-pro.example.com	CSIRT-Pro domain for internal API calls
USER_ID	user-uuid	Service account user ID
ACCESS_TOKEN	eyJhbGciOi...	Service account access token

---

Test Execution

Test a playbook with sample data before deploying it to production.

Test Procedure

1. Open the Test tab.
2. Enter sample input data as JSON.

   {
     "src_ip": "10.0.0.50",
     "message": "Failed login attempt",
     "severity": "high"
   }
   

3. Click Run Test.
4. Review the output of each node.

---

Autotest

Automated test suites can be configured for a playbook.

• GET /api/user/<user_id>/playbook/<playbook_id>/autotest/ gets the test config.
• PUT /api/user/<user_id>/playbook/<playbook_id>/autotest/ updates the test config.
• POST /api/user/<user_id>/playbook/<playbook_id>/autotest/execute runs the autotest.

---

Execution History

Playbook execution results are recorded as history entries.

Field	Description
Execution ID	Unique identifier for each execution
Status	pending / running / success / failed
Start Time	Execution start timestamp
End Time	Execution end timestamp
Logs	Per-node execution logs and outputs
Error Message	Error details for a failed execution

Retry Failed Executions

A failed execution can be retried by specifying its execution ID.

---

Version Control

Every change to a playbook is recorded as a version.

• Browse past versions.
• Restore a specific version.
• Compare versions side by side.

---

Playbook Merge

Two playbooks can be merged into one. Nodes from the source playbook are added to the target playbook.

POST /api/user/<user_id>/playbook/<playbook_id>/merge/<target_id>/

---

Webhook Triggers

An external system can trigger a playbook through its webhook endpoint.

curl -X POST "https://<domain>/api/user/<user_id>/playbook/<playbook_id>/webhook/" \
  -H "Content-Type: application/json" \
  -H "Cookie: jwt=<token>" \
  -d '{
    "alert_type": "brute_force",
    "src_ip": "10.0.0.50",
    "count": 150
  }'

The JSON body is available in the playbook as input_data.

---

Detailed Flow Examples

Example 1: Alert Triage Flow

Triage an incoming alert automatically: check IP reputation, evaluate severity, create a case for a high-severity alert, and notify the team.

[Webhook Trigger]
    │
    ▼
[Action: Check IP Reputation]
    │
    ▼
[Condition: Severity High?]
    │          │
   True       False
    │          │
    ▼          ▼
[Action:    [Action:
 Create      Log to
 Case]       Pipeline]
    │
    ▼
[Action: Notify Slack]

Node 1, Check IP Reputation (Action, type=1):

import requests

ip = input_data['src_ip']
response = requests.get(
    "https://api.abuseipdb.com/api/v2/check",
    headers={"Key": env['ABUSEIPDB_KEY'], "Accept": "application/json"},
    params={"ipAddress": ip, "maxAgeInDays": 90}
)

data = response.json()['data']
output = {
    "src_ip": ip,
    "abuse_score": data['abuseConfidenceScore'],
    "country": data.get('countryCode', 'Unknown'),
    "message": input_data.get('message', ''),
    "severity": "high" if data['abuseConfidenceScore'] > 80 else "low"
}

Node 2, Severity Check (Condition, type=2):

output = input_data['severity'] == 'high'

Node 3, Create Case (Action, type=1), true branch:

import requests

response = requests.post(
    f"https://{env['CSIRT_DOMAIN']}/api/user/{env['USER_ID']}/messages/",
    headers={"Cookie": f"jwt={env['ACCESS_TOKEN']}"},
    json={
        "title": f"High-risk IP detected: {input_data['src_ip']}",
        "description": f"AbuseIPDB score: {input_data['abuse_score']}, Country: {input_data['country']}",
        "status": "open",
        "tags": {"severity": "high", "source": "auto-triage"}
    }
)

output = {"case_id": response.json().get("id"), **input_data}

Node 4, Notify Slack (Action, type=1):

import requests

requests.post(
    "https://slack.com/api/chat.postMessage",
    headers={"Authorization": f"Bearer {env['SLACK_TOKEN']}"},
    json={
        "channel": "#security-alerts",
        "text": (
            f"*High-Severity Alert*\n"
            f"IP: `{input_data['src_ip']}` (Score: {input_data['abuse_score']})\n"
            f"Country: {input_data['country']}\n"
            f"Case ID: {input_data.get('case_id', 'N/A')}"
        )
    }
)

output = {"status": "notified"}

---

Example 2: IP Blocking Flow

When a case is created with a malicious IP, block the IP on the firewall and notify the team. The firewall write here is an action you script and operate under your own controls, not something CSIRT-Pro performs on its own.

[Case Creation Trigger]
    │
    ▼
[Action: Extract IOCs]
    │
    ▼
[Loop: For Each IP]
    │
    ▼
[Condition: Score > 90?]
    │          │
   True       False
    │          │
    ▼          ▼
[Action:    [Action:
 Block on    Skip]
 Firewall]
    │
    ▼
[Action: Log Action]

Node 1, Extract IOCs (Action, type=1):

import re

text = input_data.get('description', '') + ' ' + input_data.get('title', '')
ip_pattern = r'\b(?:\d{1,3}\.){3}\d{1,3}\b'
ips = list(set(re.findall(ip_pattern, text)))

output = {"ip_list": ips, "case_id": input_data.get("id")}

Node 2, Loop Over IPs (Loop, type=4):

output = input_data['ip_list']

Node 3, Check Score (Condition, type=2):

import requests

ip = input_data  # current loop item
response = requests.get(
    "https://api.abuseipdb.com/api/v2/check",
    headers={"Key": env['ABUSEIPDB_KEY'], "Accept": "application/json"},
    params={"ipAddress": ip, "maxAgeInDays": 90}
)

score = response.json()['data']['abuseConfidenceScore']
output = score > 90

Node 4, Block on Firewall (Action, type=1), true branch:

import requests

response = requests.post(
    f"https://{env['FIREWALL_API']}/api/v1/block",
    headers={"Authorization": f"Bearer {env['FIREWALL_TOKEN']}"},
    json={"ip": input_data, "duration": "24h", "reason": "Auto-blocked by CSIRT-Pro"}
)

output = {"blocked_ip": input_data, "status": response.status_code}

---

Example 3: Scheduled Threat Hunt Flow

Run every 15 minutes to search recent logs for known malicious indicators and create a case for any match.

Trigger: Schedule (type=104), cron */15 * * * *.

[Schedule: Every 15 min]
    │
    ▼
[Action: Query Recent Logs]
    │
    ▼
[Condition: Matches Found?]
    │          │
   True       False
    │          │
    ▼          ▼
[Loop:      [Action:
 Each        Log No
 Match]      Findings]
    │
    ▼
[Action: Enrich with TI]
    │
    ▼
[Action: Create Case]

Node 1, Query Recent Logs (Action, type=1):

import requests

response = requests.post(
    f"https://{env['CSIRT_DOMAIN']}/api/user/{env['USER_ID']}/search/",
    headers={"Cookie": f"jwt={env['ACCESS_TOKEN']}"},
    json={
        "expr": "from `firewall_logs` filter action == \"DENY\" filter __time > @2025-01-15T10:00 sort {-__time} take 500"
    }
)

# The search API returns results in a columnar binary format; read them with pyarrow.
import pyarrow as pa
reader = pa.ipc.open_stream(response.content)
table = reader.read_all()
df = table.to_pandas()

suspicious = df[df['src_ip'].isin(env.get('WATCHLIST_IPS', '').split(','))]

output = {
    "matches": suspicious.to_dict('records'),
    "match_count": len(suspicious)
}

Node 2, Check Matches (Condition, type=2):

output = input_data['match_count'] > 0

Node 3, Loop Each Match (Loop, type=4):

output = input_data['matches']

Node 4, Enrich with Threat Intelligence (Action, type=1):

import requests

ip = input_data['src_ip']
response = requests.get(
    "https://api.abuseipdb.com/api/v2/check",
    headers={"Key": env['ABUSEIPDB_KEY'], "Accept": "application/json"},
    params={"ipAddress": ip, "maxAgeInDays": 90}
)

data = response.json()['data']
output = {
    **input_data,
    "abuse_score": data['abuseConfidenceScore'],
    "country": data.get('countryCode', 'Unknown'),
    "isp": data.get('isp', 'Unknown')
}

Node 5, Create Case (Action, type=1):

import requests

match = input_data
response = requests.post(
    f"https://{env['CSIRT_DOMAIN']}/api/user/{env['USER_ID']}/messages/",
    headers={"Cookie": f"jwt={env['ACCESS_TOKEN']}"},
    json={
        "title": f"Threat hunt match: {match['src_ip']}",
        "description": (
            f"Source IP: {match['src_ip']}\n"
            f"AbuseIPDB Score: {match['abuse_score']}\n"
            f"Country: {match['country']}\n"
            f"ISP: {match['isp']}\n"
            f"Action: {match.get('action', 'N/A')}"
        ),
        "status": "open",
        "tags": {"severity": "high", "source": "threat-hunt"}
    }
)

output = {"case_created": True, "case_id": response.json().get("id")}

---

API Operations

Playbooks can be managed and run through the REST API. See the API Reference for full endpoint documentation, including the following.

• GET /api/user/<user_id>/playbook/ lists playbooks.
• POST /api/user/<user_id>/playbook/ creates a playbook.
• POST /api/user/<user_id>/playbook/<playbook_id>/execute/ runs a playbook.
• POST /api/user/<user_id>/playbook/<playbook_id>/webhook/ is the webhook trigger.
• GET /api/user/<user_id>/playbook/<playbook_id>/histories/ returns execution history.
• GET /api/user/<user_id>/playbook/<playbook_id>/versions/ returns version history.
• GET /api/user/<user_id>/playbook/<playbook_id>/environments/ returns environment variables.
• POST /api/user/<user_id>/playbook/<playbook_id>/merge/<target_id>/ merges playbooks.