Posted on
• Artificial Intelligence

Artificial Intelligence Network Incident Response

Author
  • User
    linuxbash
    Posts by this author
    Posts by this author

Artificial Intelligence Network Incident Response (NIR) with Bash: From Packets to Protection

If your SOC still triages incidents like it’s 2012, you’re fighting a wildfire with a garden hose. Modern networks generate millions of events daily, attackers automate, and alert fatigue is real. The value of Artificial Intelligence in Network Incident Response is simple: shrink time-to-detect and time-to-contain by letting machines sift the noise, surface anomalies, and even take first-response actions—while you stay in control.

This post shows you how to stand up a lightweight, AI-assisted NIR pipeline on Linux using open-source tools you already love—Bash, Suricata, Python, and nftables. You’ll collect telemetry, detect anomalies, enrich context, and automatically block likely-bad actors, all with auditable shell-friendly building blocks.

What you’ll build:

  • Telemetry: Suricata EVE JSON logs (structured network events)

  • Detection: A small Isolation Forest anomaly detector (scikit-learn)

  • Enrichment: Quick whois lookups for triage context

  • Response: ipset/nftables blocking with timeouts

  • Automation: One-liners and scripts you can drop into cron or a systemd timer

Why AI belongs in NIR

  • Volume and variability: Signature-only approaches miss low-and-slow or novel behavior. Unsupervised models excel at spotting “weird” in oceans of normal.

  • Speed and consistency: ML-powered scoring helps prioritize what matters now, reducing alert queues and human context switching.

  • Measurable lift: Teams using anomaly detection for early triage report faster containment, fewer false negatives, and better analyst focus on root cause.

0) Install prerequisites

We’ll use Suricata for structured telemetry, tshark/tcpdump for quick captures, jq for JSON slicing, Python with scikit-learn for anomaly detection, whois for enrichment, and nftables/ipset for containment.

Debian/Ubuntu (apt):

sudo apt update
sudo apt install -y \
  suricata tshark tcpdump jq whois ipset nftables \
  python3 python3-pip python3-numpy python3-pandas python3-sklearn

Fedora/RHEL/CentOS Stream (dnf):

sudo dnf install -y \
  suricata wireshark-cli tcpdump jq whois ipset nftables \
  python3 python3-pip python3-numpy python3-pandas python3-scikit-learn

openSUSE (zypper):

sudo zypper refresh
sudo zypper install -y \
  suricata wireshark-cli tcpdump jq whois ipset nftables \
  python3 python3-pip python3-numpy python3-pandas python3-scikit-learn

Optional but recommended: update Suricata rules.

sudo suricata-update
sudo systemctl restart suricata || true

1) Turn on network telemetry (Suricata EVE JSON)

Suricata’s EVE output gives rich, structured events (flow, DNS, HTTP, TLS, alerts) in JSON at /var/log/suricata/eve.json.

Quick start (ad-hoc, discover the active interface and run in the foreground):

IFACE=$(ip route get 1.1.1.1 | awk '/dev/ {for(i=1;i<=NF;i++) if ($i=="dev") print $(i+1); exit}')
sudo suricata -i "$IFACE" -c /etc/suricata/suricata.yaml -D

Or run as a service and persist the interface:

  • On Debian/Ubuntu, set the interface in /etc/default/suricata, e.g.:
echo 'SURICATA_OPTIONS="-i '"$IFACE"'"' | sudo tee /etc/default/suricata
sudo systemctl enable --now suricata
  • On Fedora/openSUSE, use /etc/sysconfig/suricata similarly, then:
sudo systemctl enable --now suricata

Verify events are flowing:

sudo tail -f /var/log/suricata/eve.json | head -n 5

Tip: Ensure outputs.eve-log is enabled in /etc/suricata/suricata.yaml (it is by default on most distros).

2) Minimal anomaly detector (Isolation Forest)

We’ll score Suricata flow events and flag source IPs that generate anomalous flows. This script reads EVE JSON from stdin, trains on a recent slice, and prints IPs that cross a simple anomaly threshold.

Save as ai_nir.py:

#!/usr/bin/env python3
import sys, json, argparse, collections
import numpy as np
from sklearn.ensemble import IsolationForest

def parse_args():
    p = argparse.ArgumentParser(description="AI-assisted NIR: flag anomalous source IPs from Suricata EVE JSON (flow events).")
    p.add_argument("--contamination", type=float, default=0.02, help="Estimated outlier fraction (0.01–0.10 typical)")
    p.add_argument("--min-samples", type=int, default=200, help="Minimum flow records to train")
    p.add_argument("--min-anom-count", type=int, default=3, help="Min outlier flows per src_ip to flag")
    p.add_argument("--verbose", action="store_true")
    return p.parse_args()

def main():
    args = parse_args()
    features, ips = [], []

    for line in sys.stdin:
        line = line.strip()
        if not line: continue
        try:
            ev = json.loads(line)
        except Exception:
            continue
        if ev.get("event_type") != "flow": 
            continue
        f = ev.get("flow", {})
        # Extract numeric features robustly; missing -> 0
        btos = f.get("bytes_toserver", 0)
        btoc = f.get("bytes_toclient", 0)
        ptos = f.get("pkts_toserver", 0)
        ptoc = f.get("pkts_toclient", 0)
        # directionality ratio (avoid div by zero)
        vol = btos + btoc
        dir_ratio = float(btos) / vol if vol else 0.0
        features.append([btos, btoc, ptos, ptoc, dir_ratio])
        ips.append(ev.get("src_ip", "0.0.0.0"))

    if len(features) < args.min_samples:
        if args.verbose:
            print(f"# Not enough flow records ({len(features)}) to train; need >= {args.min_samples}", file=sys.stderr)
        sys.exit(0)

    X = np.array(features, dtype=float)
    clf = IsolationForest(
        n_estimators=200,
        contamination=args.contamination,
        random_state=42,
        n_jobs=-1,
    )
    clf.fit(X)
    preds = clf.predict(X)  # -1 outlier, 1 inlier

    counts = collections.Counter()
    for ip, pred in zip(ips, preds):
        if pred == -1 and ip:
            counts[ip] += 1

    # Print just the IPs (one per line) for easy piping
    for ip, c in counts.most_common():
        if c >= args.min_anom_count:
            print(ip)

if __name__ == "__main__":
    main()

Make it executable:

chmod +x ai_nir.py

Run a quick test on the most recent flows:

sudo tail -n 50000 /var/log/suricata/eve.json | ./ai_nir.py --contamination 0.02 --min-samples 300 --min-anom-count 3

You’ll get a list of source IPs that look anomalous based on recent flow behavior.

Notes:

  • Start in “monitor mode” (print only) to evaluate noise and tune contamination/min-anom-count.

  • Add more features later (e.g., ports, proto, app_proto one-hot) for stronger signals.

3) Fast enrichment for triage (optional)

Give analysts instant context with simple whois lookups. Save as enrich.sh:

#!/usr/bin/env bash
set -euo pipefail
while read -r ip; do
  [[ -z "$ip" ]] && continue
  asn=$(whois -H "$ip" 2>/dev/null | awk -F: '/origin/ {print $2; exit}' | xargs)
  org=$(whois -H "$ip" 2>/dev/null | awk -F: '/OrgName|org-name|descr/ {print $2; exit}' | xargs)
  echo -e "$ip\t${asn:-NA}\t${org:-NA}"
done

chmod +x enrich.sh

Use it:

sudo tail -n 50000 /var/log/suricata/eve.json | ./ai_nir.py | ./enrich.sh

4) Automated containment with nftables/ipset

We’ll maintain a dynamic nftables set “blackhole” and drop traffic from flagged IPs with a timeout. First-time setup (idempotent-ish; errors are safe to ignore):

# Create table/chain if missing
sudo nft list table inet filter >/dev/null 2>&1 || sudo nft add table inet filter
sudo nft list chain inet filter input >/dev/null 2>&1 || sudo nft 'add chain inet filter input { type filter hook input priority 0; }'

# Create IPv4 and IPv6 sets with timeout support
sudo nft list set inet filter blackhole >/dev/null 2>&1 || \
  sudo nft 'add set inet filter blackhole { type ipv4_addr; flags timeout; }'
sudo nft list set inet filter blackhole6 >/dev/null 2>&1 || \
  sudo nft 'add set inet filter blackhole6 { type ipv6_addr; flags timeout; }'

# Add drop rules if missing
sudo nft list ruleset | grep -q 'ip saddr @blackhole drop' || \
  sudo nft add rule inet filter input ip saddr @blackhole drop
sudo nft list ruleset | grep -q 'ip6 saddr @blackhole6 drop' || \
  sudo nft add rule inet filter input ip6 saddr @blackhole6 drop

Now a small containment helper that reads IPs from stdin, whitelists known-good, and adds a 1-hour block. Save as contain.sh:

#!/usr/bin/env bash
set -euo pipefail

WHITELIST="/etc/ai-nir/whitelist.txt"
TIMEOUT="${TIMEOUT:-1h}"   # e.g., TIMEOUT=15m ./contain.sh

is_whitelisted() {
  local ip="$1"
  [[ -f "$WHITELIST" ]] && grep -Fxq "$ip" "$WHITELIST"
}

while read -r ip; do
  [[ -z "$ip" ]] && continue
  if is_whitelisted "$ip"; then
    echo "skip (whitelisted): $ip" >&2
    continue
  fi
  if [[ "$ip" =~ : ]]; then
    sudo nft add element inet filter blackhole6 "{ $ip timeout $TIMEOUT }" || true
  else
    sudo nft add element inet filter blackhole "{ $ip timeout $TIMEOUT }" || true
  fi
  echo "blocked $ip for $TIMEOUT" >&2
done

chmod +x contain.sh

Dry-run first:

sudo tail -n 50000 /var/log/suricata/eve.json | ./ai_nir.py | sort -u

When you’re comfortable, activate containment:

sudo tail -n 50000 /var/log/suricata/eve.json | ./ai_nir.py | sudo TIMEOUT=1h ./contain.sh

List or clear the sets:

sudo nft list set inet filter blackhole
sudo nft flush set inet filter blackhole

5) Wire it together on a schedule

Option A: cron (cronie)

crontab -e

Add:

*/2 * * * * sudo tail -n 80000 /var/log/suricata/eve.json | /opt/ai-nir/ai_nir.py --contamination 0.02 --min-samples 400 --min-anom-count 3 | /opt/ai-nir/contain.sh

Option B: systemd timer (more robust)

  • /etc/systemd/system/ai-nir.service
[Unit]
Description=AI NIR anomaly detect and contain

[Service]
Type=oneshot
ExecStart=/bin/bash -lc 'tail -n 80000 /var/log/suricata/eve.json | /opt/ai-nir/ai_nir.py --contamination 0.02 --min-samples 400 --min-anom-count 3 | /opt/ai-nir/contain.sh'
  • /etc/systemd/system/ai-nir.timer
[Unit]
Description=Run AI NIR every 2 minutes

[Timer]
OnCalendar=*:0/2
Persistent=true

[Install]
WantedBy=timers.target

Enable:

sudo systemctl daemon-reload
sudo systemctl enable --now ai-nir.timer
sudo systemctl list-timers | grep ai-nir

Real-world-style scenario

  • An externally exposed service is probed by an automated tool that makes many short, lopsided flows (small request bursts, near-zero responses).

  • Suricata logs these as flow events. Our model—trained on mostly normal traffic—flags multiple outlier flows from the probing IP.

  • The AI pipeline emits that IP, contain.sh adds it to nftables with a 1-hour timeout, and subsequent packets are dropped.

  • Analysts review enrich.sh output, confirm the behavior, and optionally extend block duration or tune thresholds.

Tuning tips and safeguards

  • Start passive: Run detection without contain.sh for at least a few days; measure false positives.

  • Whitelist generously: Internal scanners, monitoring, known partners.

  • Thresholds: Increase --min-anom-count in noisy networks; reduce contamination to be stricter.

  • Auditability: Log all blocks (stdout/stderr) to a file with a timestamped tee; consider git-tracking whitelist changes.

  • Safety: Prefer time-limited blocks; let IPs age out unless re-triggered.

Conclusion and next steps

You now have an end-to-end, Bash-first AI pipeline that:

  • Collects structured network telemetry

  • Detects anomalies with a compact ML model

  • Enriches context for faster triage

  • Automatically contains likely threats with reversible, time-limited blocks

Next steps:

  • Add features (ports, proto, app_proto one-hot, flow duration) to strengthen detection.

  • Feed in DNS/HTTP/TLS EVE events for richer context.

  • Promote to a proper service with metrics, dashboards, and alerting.

  • Integrate known-bad feeds (MISP, blocklists) to complement AI with intelligence.

If you found this useful, try it in monitor mode on a lab or low-risk segment, iterate on thresholds for a week, then turn on automated containment with short timeouts. Your future self—rested and less buried in alerts—will thank you.

Related posts