Automated Threat Intelligence Alerting in Kestra

id: ioc-threat-intel-isolation
namespace: company.security
description: |
  Query AbuseIPDB API for threat intelligence on suspicious IPs, evaluate risk confidence
  scores via Python, and trigger automated Slack alerts for SecOps triage when thresholds are breached.

inputs:
  - id: suspicious_ip
    type: STRING
    displayName: Suspicious IP Address
    description: IPv4 or IPv6 address to query against threat intelligence feeds.
    defaults: "118.25.6.39"

  - id: confidence_threshold
    type: INT
    displayName: Abuse Confidence Threshold
    description: Minimum abuse confidence score (0-100) that triggers a security
      containment alert.
    defaults: "75"

  - id: fail_on_high_risk
    type: BOOL
    displayName: Fail on High Risk
    description: When true, halts execution with an error status when the IP is
      confirmed malicious. When false, logs and alerts without failing the flow
      execution.
    defaults: "false"

triggers:
  - id: hourly_threat_audit
    type: io.kestra.plugin.core.trigger.Schedule
    description: Hourly scheduled threat intelligence check against active
      indicators of compromise; shipped disabled.
    cron: "0 * * * *"
    disabled: true

tasks:
  - id: check_threat_intel
    type: io.kestra.plugin.scripts.shell.Commands
    description: Query AbuseIPDB API to fetch real-time reputation and threat
      intelligence for a suspicious IP address using curl.
    containerImage: curlimages/curl:latest
    timeout: PT2M
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
    commands:
      - |
        curl -f -s -G https://api.abuseipdb.com/api/v2/check \
          --data-urlencode "ipAddress={{ inputs.suspicious_ip }}" \
          -d maxAgeInDays=30 \
          -H "Key: {{ secret('ABUSEIPDB_API_KEY') }}" \
          -H "Accept: application/json" > threat-intel.json
    outputFiles:
      - threat-intel.json

  - id: analyze_threat_score
    type: io.kestra.plugin.scripts.python.Script
    description: Parse AbuseIPDB intelligence payload with Python and evaluate
      malicious confidence score against threshold.
    containerImage: python:3.11-slim
    timeout: PT2M
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
    dependencies:
      - kestra
    inputFiles:
      threat-intel.json: "{{ outputs.check_threat_intel.outputFiles['threat-intel.json'] }}"
    script: |
      import json
      from kestra import Kestra

      with open("threat-intel.json") as f:
          payload = json.load(f)

      data = payload.get("data", {})
      abuse_score = data.get("abuseConfidenceScore", 0)
      country_code = data.get("countryCode", "Unknown")
      total_reports = data.get("totalReports", 0)
      domain = data.get("domain", "N/A")
      isp = data.get("isp", "N/A")

      is_malicious = abuse_score >= {{ inputs.confidence_threshold }}

      Kestra.outputs({
          "is_malicious": is_malicious,
          "abuse_score": abuse_score,
          "country_code": country_code,
          "total_reports": total_reports,
          "domain": domain,
          "isp": isp
      })

  - id: log_threat_summary
    type: io.kestra.plugin.core.log.Log
    description: Print an itemized audit log of the threat intelligence check.
    message: |
      Threat Intelligence Analysis Summary:
      - Target IP: {{ inputs.suspicious_ip }}
      - Abuse Confidence Score: {{ outputs.analyze_threat_score.vars.abuse_score }}% (Threshold: {{ inputs.confidence_threshold }}%)
      - Geolocation: {{ outputs.analyze_threat_score.vars.country_code }}
      - ISP / Domain: {{ outputs.analyze_threat_score.vars.isp }} ({{ outputs.analyze_threat_score.vars.domain }})
      - Total Abuse Reports: {{ outputs.analyze_threat_score.vars.total_reports }}
      - Malicious Flag: {{ outputs.analyze_threat_score.vars.is_malicious }}

  - id: evaluate_quarantine_policy
    type: io.kestra.plugin.core.flow.If
    description: Evaluate risk confidence score against security policy and branch
      workflow execution.
    condition: "{{ outputs.analyze_threat_score.vars.is_malicious }}"
    then:
      - id: notify_secops_incident
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Non-core plugin task notifying SecOps incident response team of
          blocked malicious IP via Slack.
        url: "{{ secret('SLACK_SECURITY_WEBHOOK') }}"
        payload: |
          {
            "text": "*SOAR Threat Intelligence Alert!*\n*Suspicious IP:* `{{ inputs.suspicious_ip }}`\n*Abuse Confidence Score:* `{{ outputs.analyze_threat_score.vars.abuse_score }}%`\n*Country:* `{{ outputs.analyze_threat_score.vars.country_code }}`\n*ISP:* `{{ outputs.analyze_threat_score.vars.isp }}`\n*Total Reports:* `{{ outputs.analyze_threat_score.vars.total_reports }}`\n*Action:* Escalated for SecOps review; no automated blocking performed."
          }

      - id: check_fail_policy
        type: io.kestra.plugin.core.flow.If
        description: Enforce execution failure when fail_on_high_risk toggle is enabled.
        condition: "{{ inputs.fail_on_high_risk }}"
        then:
          - id: fail_threat_gate
            type: io.kestra.plugin.core.execution.Fail
            description: Halt workflow execution with error state when malicious indicator
              of compromise is confirmed.
            errorMessage: "Threat Gate FAILED: Target IP {{ inputs.suspicious_ip }}
              evaluated with abuse confidence score of {{
              outputs.analyze_threat_score.vars.abuse_score }}%, exceeding
              maximum threshold of {{ inputs.confidence_threshold }}%."
    else:
      - id: log_benign_ip
        type: io.kestra.plugin.core.log.Log
        description: Log confirmation when target IP sits below threat confidence threshold.
        level: INFO
        message: "Threat gate PASSED: IP {{ inputs.suspicious_ip }} evaluated with abuse
          score {{ outputs.analyze_threat_score.vars.abuse_score }}% (below
          threshold of {{ inputs.confidence_threshold }}%). No containment
          action required."

outputs:
  - id: abuse_score
    type: INT
    value: "{{ outputs.analyze_threat_score.vars.abuse_score }}"
  - id: is_malicious
    type: BOOL
    value: "{{ outputs.analyze_threat_score.vars.is_malicious }}"
  - id: raw_report
    type: URI
    value: "{{ outputs.check_threat_intel.outputFiles['threat-intel.json'] }}"

Security Operations Center (SOC) teams frequently face influxes of suspicious incoming requests, automated vulnerability scans, and brute-force access attempts. Manually looking up threat indicators across disparate APIs delays incident response and leaves cloud infrastructure exposed to active attack vectors.

This blueprint provides an automated Security Orchestration, Automation, and Response (SOAR) workflow using AbuseIPDB threat intelligence. It queries IP reputation metrics, evaluates risk confidence scores in a containerized Python environment, logs audit details, and alerts the SecOps team via Slack when confidence thresholds are breached.

How it works

check_threat_intel (io.kestra.plugin.scripts.shell.Commands) queries the AbuseIPDB API using curlimages/curl:latest with credentials retrieved safely from Kestra Secrets.
analyze_threat_score (io.kestra.plugin.scripts.python.Script) parses the JSON payload, extracts key indicators (abuse confidence score, country code, total reports, ISP, domain), and outputs variables for flow logic.
log_threat_summary (io.kestra.plugin.core.log.Log) prints an itemized breakdown of threat indicators to the execution log.
evaluate_quarantine_policy (io.kestra.plugin.core.flow.If) evaluates whether the abuse score meets or exceeds inputs.confidence_threshold:
- If malicious: notify_secops_incident (io.kestra.plugin.slack.notifications.SlackIncomingWebhook) triggers an incident report in Slack.
- If fail_on_high_risk is true: fail_threat_gate (io.kestra.plugin.core.execution.Fail) halts execution with an error status.
- If benign: log_benign_ip (io.kestra.plugin.core.log.Log) logs a PASSED confirmation.
outputs exposes the raw JSON threat report URI and calculated confidence metrics for downstream security flows.

What you get

Automated real-time IP reputation lookup without commercial SOAR licensing overhead.
Configurable risk threshold gating (confidence_threshold) to eliminate false positives.
Rich Slack alerting containing actionable context (ISP, Geolocation, Abuse Score, and Report Count).
Non-blocking audit mode (fail_on_high_risk: false) for passive security logging.

Who it's for

Security Operations (SOC) Engineers automating incident triage and threat hunting.
SecOps and Cloud Security teams building threat-intelligence alerting and triage workflows.
DevOps engineers setting up automated perimeter checks for incoming webhook integrations.

Prerequisites

A Kestra worker capable of executing Docker container runners.
Outbound internet access on the worker to communicate with api.abuseipdb.com.

Secrets

ABUSEIPDB_API_KEY: API key for authentication with AbuseIPDB v2 endpoints.
SLACK_SECURITY_WEBHOOK: Incoming Webhook URL for your SecOps Slack channel.

Quick start

Configure secrets ABUSEIPDB_API_KEY and SLACK_SECURITY_WEBHOOK in your Kestra namespace.
Execute the flow with default inputs (suspicious_ip: 118.25.6.39, confidence_threshold: 75).
Observe the flow fetching real-time threat intelligence, evaluating the IP risk score, and sending a formatted Slack notification if the IP exceeds the threshold.

Outputs

outputs.abuse_score: AbuseIPDB confidence score from 0 to 100.
outputs.is_malicious: Boolean indicating whether the score meets or exceeds confidence_threshold.
outputs.raw_report: URI of the raw AbuseIPDB JSON response.
Example references: {{ outputs.abuse_score }}, {{ outputs.is_malicious }}, and {{ outputs.raw_report }}.

How to integrate with your SIEM or WAF

To automate quarantine actions directly in cloud infrastructure, extend notify_secops_incident with a cloud-specific task (for example, an AWS WAF or Cloudflare integration) that adds malicious IPs to an active blocklist or IP set.

Trigger this workflow dynamically via Kestra Webhook triggers (io.kestra.plugin.core.trigger.Webhook) whenever your web server or SIEM detects abnormal traffic anomalies.

Automated Incident Response Threat Intelligence Alerting

How it works

What you get

Who it's for

Prerequisites

Secrets

Quick start

Outputs

How to integrate with your SIEM or WAF

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

Automated Incident Response Threat Intelligence Alerting

How it works

What you get

Who it's for

Prerequisites

Secrets

Quick start

Outputs

How to integrate with your SIEM or WAF

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Core Blueprints

Namespace Scoped Failure Alerts to Slack with KV Enrichment

Microservices and APIs workflow example, Getting started with Kestra

Microsoft Fabric Warehouse Data Quality Assertion Gate

Migrate a Cron Script to Kestra Without Rewriting It, Using the Strangler Fig Pattern

Run Multiple Python Versions Side by Side in One Flow

Sync namespace resources between Git and Kestra at regular intervals

New to Kestra?