Artificial Intelligence Packet Capture Automation with Bash

Smarter PCAPs, less noise, faster answers.

If you’ve ever tailed gigabytes of packet captures at 3 a.m. hunting for a weird spike, you know the pain: sprawling PCAPs, short-lived indicators, and never enough time. The value of packet capture is indisputable, but “always-on, capture-everything” quickly becomes “store-nothing-useful.”

This post shows how to use Linux + Bash to automate packet capture with AI-assisted triage so you:

• Capture the right packets at the right time (ring buffers and triggers).

• Extract flow features in seconds (tshark).

• Surface outliers automatically (IsolationForest).

• Summarize and escalate when it matters.

You’ll get a working, end-to-end pipeline you can schedule with cron/systemd, with installation commands for apt, dnf, and zypper.

---

Why AI-driven capture automation is worth it

• Relevance over volume: Ring buffers and event-driven triggers (burst detection, port scans) keep high-signal traffic while shedding noise.

• Reduced toil: AI flags outliers from recent captures, pointing you to the odd flows instead of making you skim everything.

• Faster incident response: Automatic pivoting—tighten filters around suspicious hosts/ports, extend capture windows, and archive relevant slices.

---

Prerequisites and installation

We’ll use tcpdump (capture), tshark (parsing), jq (JSON), and Python with scikit-learn (AI anomaly detection). Optional: Zeek for richer logs.

Packages and service names vary a bit by distro; use the matching block below.

• Debian/Ubuntu (apt)

sudo apt update
sudo apt install -y tcpdump tshark jq python3 python3-venv python3-pip cron
# Optional (if available)
sudo apt install -y zeek
# Enable cron (usually enabled by default)
sudo systemctl enable --now cron

• Fedora/RHEL/CentOS Stream (dnf)

sudo dnf install -y tcpdump wireshark-cli jq python3 python3-pip python3-virtualenv cronie
# Optional (availability varies by distro/version; use COPR/EPEL if not found)
sudo dnf install -y zeek || echo "Zeek not in repo; install from Zeek.org or COPR/EPEL"
# Enable cron
sudo systemctl enable --now crond

• openSUSE Leap/Tumbleweed (zypper)

sudo zypper refresh
sudo zypper install -y tcpdump wireshark-cli jq python310 python310-pip python310-virtualenv cron
# Optional (if present)
sudo zypper install -y zeek || echo "Zeek not in repo; install from Zeek.org or build from source"
# Enable cron
sudo systemctl enable --now cron

Allow non-root capture (optional but recommended):

# tcpdump capability
sudo setcap cap_net_raw,cap_net_admin=eip /usr/sbin/tcpdump 2>/dev/null || \
sudo setcap cap_net_raw,cap_net_admin=eip /usr/bin/tcpdump

# tshark uses dumpcap for capture paths
sudo setcap cap_net_raw,cap_net_admin=eip /usr/bin/dumpcap

Create a project directory:

mkdir -p ~/ai-pcap/{pcap,out,logs,scripts,venv}
cd ~/ai-pcap

Set up a Python environment and install ML deps:

# Ubuntu/Debian
python3 -m venv venv
# openSUSE with python310
python3 -m venv venv
source venv/bin/activate
pip install --upgrade pip
pip install numpy pandas scikit-learn
deactivate

---

Step 1: Safe, continuous capture with a ring buffer

We’ll start a rolling capture that:

• Keeps a small time window (e.g., last 60 minutes).

• Excludes routine noise (e.g., SSH to your jump host).

• Uses a safe snaplen to cut payload bloat.

Create scripts/capture.sh:

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

IFACE="${IFACE:-eth0}"
OUTDIR="${OUTDIR:-$HOME/ai-pcap/pcap}"
ROTATE_SEC="${ROTATE_SEC:-300}"   # rotate every 5 min
WINDOW_FILES="${WINDOW_FILES:-12}" # keep 12 files => 60 min window
SNAPLEN="${SNAPLEN:-128}"

# Example BPF: ignore SSH to/from 10.0.0.0/8; capture everything else IP
BPF_FILTER="${BPF_FILTER:-ip and not port 22 and not net 10.0.0.0/8}"

mkdir -p "$OUTDIR"

# File naming with UTC timestamp
FNAME="$OUTDIR/cap-%Y%m%dT%H%M%SZ.pcap"

# Run as root or with setcap; use nice/ionice to be gentle
exec nice -n 10 ionice -c2 -n7 tcpdump \
  -i "$IFACE" -s "$SNAPLEN" -nn -tt \
  -G "$ROTATE_SEC" -W "$WINDOW_FILES" \
  -w "$FNAME" \
  $BPF_FILTER

Run it:

chmod +x scripts/capture.sh
sudo IFACE=eth0 OUTDIR=$HOME/ai-pcap/pcap ./scripts/capture.sh

Tip: Adjust BPF filters to your environment. Good defaults often exclude chatty internal nets and known management ports.

---

Step 2: Event-driven capture triggers (optional, powerful)

Instead of always-on, you can start/stop or widen/narrow capture based on real-time signals: bandwidth spikes, sudden port fan-out, or connection bursts.

A lightweight approach: watch /proc/net/dev bytes per second to trigger “enhanced” capture for a short window.

Create scripts/triggered-capture.sh:

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

IFACE="${IFACE:-eth0}"
THRESHOLD_MBPS="${THRESHOLD_MBPS:-100}"     # trigger when >100 Mbps
CHECK_INTERVAL="${CHECK_INTERVAL:-5}"        # seconds between checks
BURST_WINDOW_SEC="${BURST_WINDOW_SEC:-120}"  # enhanced capture duration
OUTDIR="${OUTDIR:-$HOME/ai-pcap/pcap}"

read_bytes() {
  awk -v iface="$IFACE" '$1 ~ iface":" {print $2}' /proc/net/dev
}

prev=$(read_bytes)
prev_t=$(date +%s)

while true; do
  sleep "$CHECK_INTERVAL"
  now=$(read_bytes)
  now_t=$(date +%s)
  diff=$((now - prev))
  dt=$((now_t - prev_t))
  bps=$((diff / dt))
  mbps=$((bps * 8 / 1000000))
  echo "$(date -Is) $IFACE throughput ~ ${mbps}Mbps"

  if (( mbps > THRESHOLD_MBPS )); then
    echo "$(date -Is) Burst detected: ${mbps}Mbps > ${THRESHOLD_MBPS}Mbps; starting enhanced capture..."
    # Start a short, tighter-snaplen capture with full IP payload for key ports
    ENHANCED_BPF='ip and (tcp port 443 or tcp port 22 or udp port 53)'
    timeout "$BURST_WINDOW_SEC" sudo env IFACE="$IFACE" OUTDIR="$OUTDIR" SNAPLEN=256 \
      BPF_FILTER="$ENHANCED_BPF" "$HOME/ai-pcap/scripts/capture.sh" &
  fi

  prev=$now
  prev_t=$now_t
done

Run it in a tmux/screen session or as a service to layer bursts on top of your base ring buffer.

---

Step 3: Extract features with tshark and flag anomalies with AI

We’ll convert the most recent PCAP to lightweight flow features, then run an IsolationForest to score outliers.

Create scripts/pcap_to_csv.sh:

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

PCAP="$1"
OUTCSV="$2"

# Extract per-packet fields; we’ll aggregate in Python.
tshark -r "$PCAP" -T fields -E header=y -E separator=, -E quote=d \
  -e frame.time_epoch \
  -e ip.src -e ip.dst -e ip.proto \
  -e tcp.srcport -e tcp.dstport \
  -e udp.srcport -e udp.dstport \
  -e frame.len \
  2>/dev/null > "$OUTCSV"

Create scripts/ai_detect.py:

#!/usr/bin/env python3
import sys, json, os
import pandas as pd
import numpy as np
from sklearn.ensemble import IsolationForest

if len(sys.argv) < 2:
    print("Usage: ai_detect.py features.csv", file=sys.stderr)
    sys.exit(2)

csv_path = sys.argv[1]
df = pd.read_csv(csv_path)

# Normalize columns
for col in ['tcp.srcport','tcp.dstport','udp.srcport','udp.dstport']:
    if col not in df.columns: df[col] = np.nan

# Build 5-tuple-like flow key (handle udp/tcp variants)
def pick_sport(row):
    return row['tcp.srcport'] if not np.isnan(row['tcp.srcport']) else row['udp.srcport']
def pick_dport(row):
    return row['tcp.dstport'] if not np.isnan(row['tcp.dstport']) else row['udp.dstport']

df['sport'] = df.apply(pick_sport, axis=1)
df['dport'] = df.apply(pick_dport, axis=1)

# Drop rows without IPs
df = df.dropna(subset=['ip.src','ip.dst','ip.proto'])

# Aggregate per flow
grp = df.groupby(['ip.src','ip.dst','ip.proto','sport','dport'])
flows = grp.agg(
    pkt_count=('frame.len','count'),
    byte_sum=('frame.len','sum'),
    byte_mean=('frame.len','mean'),
    byte_std=('frame.len','std')
).reset_index()
flows['byte_std'] = flows['byte_std'].fillna(0.0)

# Feature matrix
X = flows[['pkt_count','byte_sum','byte_mean','byte_std']].astype(float)

# IsolationForest for unsupervised outlier detection
seed = int(os.environ.get('AI_SEED', '42'))
model = IsolationForest(
    n_estimators=200,
    contamination=float(os.environ.get('AI_CONTAM', '0.02')),
    random_state=seed
)
model.fit(X)
scores = -model.score_samples(X)  # higher => more anomalous
flows['anomaly_score'] = scores

# Top-N anomalies
topn = int(os.environ.get('AI_TOPN', '10'))
sus = flows.sort_values('anomaly_score', ascending=False).head(topn)

# Produce a compact JSON summary
out = []
for _, r in sus.iterrows():
    out.append(dict(
        src=str(r['ip.src']),
        dst=str(r['ip.dst']),
        proto=int(r['ip.proto']),
        sport=None if pd.isna(r['sport']) else int(r['sport']),
        dport=None if pd.isna(r['dport']) else int(r['dport']),
        pkt_count=int(r['pkt_count']),
        byte_sum=int(r['byte_sum']),
        score=float(r['anomaly_score'])
    ))

print(json.dumps({"anomalies": out}, indent=2))

Make scripts executable:

chmod +x scripts/pcap_to_csv.sh scripts/ai_detect.py

Create scripts/triage.sh to tie it together:

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

BASE="$HOME/ai-pcap"
PCAP_DIR="$BASE/pcap"
OUT_DIR="$BASE/out"
LOG_DIR="$BASE/logs"
VENVDIR="$BASE/venv"

mkdir -p "$OUT_DIR" "$LOG_DIR"

# Find the newest PCAP
PCAP=$(ls -1t "$PCAP_DIR"/cap-*.pcap 2>/dev/null | head -n1 || true)
if [[ -z "${PCAP}" ]]; then
  echo "$(date -Is) No PCAPs found in $PCAP_DIR" | tee -a "$LOG_DIR/triage.log"
  exit 0
fi

CSV="$OUT_DIR/$(basename "$PCAP" .pcap).csv"
JSON="$OUT_DIR/$(basename "$PCAP" .pcap)-anomalies.json"

# Convert to CSV
"$BASE/scripts/pcap_to_csv.sh" "$PCAP" "$CSV"

# Run AI
source "$VENVDIR/bin/activate"
AI_TOPN="${AI_TOPN:-10}" AI_CONTAM="${AI_CONTAM:-0.02}" python3 "$BASE/scripts/ai_detect.py" "$CSV" > "$JSON"
deactivate

# Pretty print summary with jq
echo "Latest anomalies from $(basename "$PCAP"):"
jq -r '.anomalies[] | "score=\(.score|tostring) src=\(.src) sport=\(.sport) -> dst=\(.dst) dport=\(.dport) proto=\(.proto) pkts=\(.pkt_count) bytes=\(.byte_sum)"' "$JSON" \
  | tee -a "$LOG_DIR/triage.log"

Run triage manually:

chmod +x scripts/triage.sh
./scripts/triage.sh

Expected output example:

Latest anomalies from cap-20240722T144000Z.pcap:
score=0.812 src=192.0.2.55 sport=51514 -> dst=203.0.113.10 dport=53 proto=17 pkts=184 bytes=22308
score=0.701 src=10.1.2.3 sport=445 -> dst=10.8.9.40 dport=62222 proto=6 pkts=12 bytes=1894
...

What to do with that?

• Tighten your live capture BPF to focus on these hosts/ports for the next N minutes.

• Auto-archive the flagged PCAP slices.

• Notify via Slack/email.

Example: auto-archive when high-score anomalies include external destinations:

jq -r '.anomalies[] | select(.dst|startswith("10.")|not) | .dst' "$JSON" | while read dst; do
  echo "$(date -Is) External anomaly to $dst — archiving $PCAP"
  gzip -9 -c "$PCAP" > "$OUT_DIR/$(basename "$PCAP").gz"
done

---

Step 4: Automate with systemd timers or cron

• systemd timer (recommended)

Create /etc/systemd/system/ai-triage.service:

[Unit]
Description=AI PCAP triage

[Service]
Type=oneshot
User=%i
ExecStart=%h/ai-pcap/scripts/triage.sh

Create /etc/systemd/system/ai-triage@.timer:

[Unit]
Description=Run AI PCAP triage periodically for %i

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

[Install]
WantedBy=timers.target

Enable for your user (replace YOURUSER):

sudo systemctl enable --now ai-triage@YOURUSER.timer
systemctl status ai-triage@YOURUSER.timer

• Cron alternative

*/5 * * * * /home/YOURUSER/ai-pcap/scripts/triage.sh >> /home/YOURUSER/ai-pcap/logs/cron.log 2>&1

---

Step 5 (optional): Natural-language summaries with an LLM

If you want a quick human-readable take, you can send the JSON to an LLM. Keep secrets out of scripts and use env vars.

Example generic curl (provider-agnostic; adapt URL/headers to your LLM service):

export LLM_API_KEY="your_api_key_here"
jq -r '.' "$HOME/ai-pcap/out/"*-anomalies.json | \
curl -s https://api.your-llm-provider.example/v1/chat/completions \
  -H "Authorization: Bearer $LLM_API_KEY" \
  -H "Content-Type: application/json" \
  -d @- | jq -r '.choices[0].message.content'

Prompt hint:

Summarize these network anomalies for a SOC analyst. Flag likely port scans, DNS exfiltration, or suspicious egress. Provide concrete next steps and key IPs/ports.

For completely offline setups, consider a local model runner (e.g., llama.cpp or ollama) and point curl at localhost.

---

Real-world example scenarios

• Port scan burst: trigger detects bandwidth/conn burst; triage surfaces many src->dst flows with low packets per flow but broad dport fan-out. Tighten BPF to “host attacker and tcp” for 10 minutes and archive.

• DNS tunneling suspicion: anomalies show a single internal host with unusually high DNS byte_sum and pkt_count to a single resolver. Extend capture and add BPF focus “udp port 53 and host .”

• Data egress spike: triage spots new external dst with large byte_sum over TCP 443 from an unusual internal source. Pivot to full-payload snaplen (careful) on that tuple, notify, and preserve chain of custody.

---

Hardening and performance notes

• Snaplen tradeoffs: 96–256 bytes capture headers and a bit of payload for context; go full only on-demand.

• BPF filters first: Drop obvious noise early (backups, monitoring subnets).

• Storage: Use a dedicated filesystem with quotas. Consider gzip or zstd after rotation.

• Privileges: Prefer setcap over root; restrict script permissions.

• Reproducibility: Pin Python deps in a requirements.txt or lockfile.

---

Conclusion and Call to Action

Packet capture is priceless—but only when it’s timely, scoped, and explainable. With a thin layer of Bash automation and a small dose of AI, you can convert rivers of packets into prioritized leads for your next investigation.

Your next steps:

• Install the tooling and start the ring buffer capture on your busiest interface.

• Run triage.sh hourly and review the anomalies in logs/triage.log.

• Add one trigger (burst or port scan) and one action (archive or notify).

• Iterate on BPF filters that fit your environment.

Want a ready-to-fork skeleton repo of the scripts in this post? Copy the structure here into ~/ai-pcap and adapt to your network. If you get stuck or want a deeper dive (Zeek integrations, Suricata alerts -> targeted PCAP, local LLM summaries), reach out and I’ll help you extend this into a production playbook.