Posted on
• Artificial Intelligence

Artificial Intelligence Packet Capture Analysis

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

AI-Powered Packet Capture Analysis on Linux (A Bash-First Workflow)

Drowning in PCAPs? Modern networks generate more traffic than humans can manually review. What if a few shell commands and a tiny ML model could sift millions of packets and flag the suspicious stuff while you sleep?

In this post, you’ll learn a practical, Bash-first way to:

  • Capture and rotate packet data safely

  • Extract useful features with CLI tools

  • Train a lightweight anomaly detector

  • Score new traffic continuously and surface likely issues

This isn’t about replacing signatures or analysts. It’s about using AI to reduce noise, highlight outliers, and give you fast triage on encrypted and novel traffic patterns.

Why AI for PCAPs is worth your time

  • Scale and speed: Terabytes of traffic can be summarized into flow-level features. ML can quickly highlight oddities (port scans, beacons, data exfil) you’d otherwise miss.

  • Encryption reality: With TLS everywhere, payload signatures lose power. Feature-based detection (size, timing, flags, connection counts) still works.

  • Complements your stack: Blend anomalies with Zeek/Suricata alerts for context, not conflict.

Install the prerequisites

We’ll use:

  • tcpdump (capture)

  • tshark (CLI Wireshark; parsing)

  • Zeek (optional, flow logs)

  • Suricata (optional, signatures)

  • Python 3 with scikit-learn, pandas (ML)

Run the commands for your distro.

Debian/Ubuntu (apt):

sudo apt update
sudo apt install -y tcpdump tshark python3 python3-pip python3-venv zeek suricata

Note: apt may ask about allowing non-superusers to capture packets. Choose as appropriate.

Fedora/RHEL/CentOS Stream (dnf):

sudo dnf install -y tcpdump wireshark-cli python3 python3-pip python3-virtualenv zeek suricata

Note: If zeek isn’t available on your RHEL base, enable EPEL or use the Zeek project packages.

openSUSE (zypper):

sudo zypper install -y tcpdump wireshark-cli python3 python3-pip python3-virtualenv zeek suricata

If your distro doesn’t provide wireshark-cli, try installing tshark or wireshark instead:

  • apt: sudo apt install tshark

  • dnf: sudo dnf install wireshark-cli (tshark is provided by wireshark-cli)

  • zypper: sudo zypper install wireshark or wireshark-cli

Create a Python virtual environment and install ML libs:

python3 -m venv .venv
. .venv/bin/activate
pip install --upgrade pip
pip install pandas scikit-learn numpy pyshark joblib

Optional but useful:

  • Allow non-root captures (adjust per your policy):
sudo usermod -aG wireshark $USER
sudo setcap cap_net_raw,cap_net_admin=eip $(command -v dumpcap)  # requires libcap

Only capture traffic you are authorized to monitor.

Step 1: Capture traffic into a rotating ring buffer

Keep disk usage predictable and avoid gigantic files. This example rotates every 5 minutes and keeps 12 files (~1 hour).

sudo mkdir -p /var/log/pcap
sudo tcpdump -i eth0 -s 96 -G 300 -W 12 -w /var/log/pcap/cap-%Y%m%d%H%M%S.pcap 'tcp or udp'

  • -i eth0: choose your interface (or use any)

  • -s 96: snaplen; capture headers not payload to reduce size

  • -G 300: rotate every 300 seconds

  • -W 12: keep 12 files max

  • Filter: capture only TCP/UDP

Tip: Use dumpcap for very high-rate capture:

sudo dumpcap -i eth0 -b duration:300 -b files:12 -w /var/log/pcap/cap.pcapng

Step 2: Extract structured features with tshark (fast and scriptable)

We’ll turn packets into a simple CSV for ML. This example focuses on TCP over IPv4.

pcap_to_packets.sh

#!/usr/bin/env bash
set -euo pipefail
pcap="${1:?usage: pcap_to_packets.sh file.pcap [out.csv]}"
out="${2:-${pcap%.pcap}.tcp.csv}"

tshark -r "$pcap" -Y "tcp && ip" \
  -T fields -E header=y -E separator=, \
  -e frame.time_epoch \
  -e ip.src -e ip.dst \
  -e tcp.srcport -e tcp.dstport \
  -e frame.len \
  -e tcp.flags \
  > "$out"

echo "Wrote $out"

Make executable:

chmod +x pcap_to_packets.sh

Optional (Zeek flow logs): Zeek already aggregates flows. For offline PCAPs:

zeek -r sample.pcap
# Flow data in conn.log (TSV). For a CSV-like cut:
zeek-cut id.orig_h id.resp_h id.orig_p id.resp_p proto service duration orig_bytes resp_bytes conn_state < conn.log > conn.tsv

Step 3: Train a tiny anomaly detector (Isolation Forest)

We’ll aggregate per 5‑tuple flows from the packet CSV and train an IsolationForest. Start with a “known-good” day to form your baseline.

score_flows.py

#!/usr/bin/env python3
import argparse, json
import numpy as np
import pandas as pd
from sklearn.ensemble import IsolationForest
from joblib import dump, load

def parse_flags(x):
    x = str(x)
    try:
        return int(x, 16) if x.startswith('0x') else int(x)
    except Exception:
        return np.nan

def build_features(csv_path):
    df = pd.read_csv(csv_path)
    cols = ['frame.time_epoch','ip.src','ip.dst','tcp.srcport','tcp.dstport','frame.len','tcp.flags']
    df = df[cols].dropna()
    df['t'] = pd.to_numeric(df['frame.time_epoch'], errors='coerce')
    df['len'] = pd.to_numeric(df['frame.len'], errors='coerce')
    df['flags'] = df['tcp.flags'].map(parse_flags)
    df['tcp.srcport'] = pd.to_numeric(df['tcp.srcport'], errors='coerce')
    df['tcp.dstport'] = pd.to_numeric(df['tcp.dstport'], errors='coerce')
    df = df.dropna()

    gcols = ['ip.src','ip.dst','tcp.srcport','tcp.dstport']
    def count_mask(series, mask):
        return int(np.sum(series.apply(lambda v: 1 if (int(v) & mask) != 0 else 0)))

    agg = df.groupby(gcols).agg(
        pkts=('len','count'),
        bytes=('len','sum'),
        mean_len=('len','mean'),
        std_len=('len','std'),
        t_min=('t','min'),
        t_max=('t','max'),
        syn=('flags', lambda s: count_mask(s, 0x02)),
        rst=('flags', lambda s: count_mask(s, 0x04)),
        fin=('flags', lambda s: count_mask(s, 0x01)),
        ack=('flags', lambda s: count_mask(s, 0x10)),
    ).fillna(0)
    agg['duration'] = agg['t_max'] - agg['t_min']
    X = agg[['pkts','bytes','mean_len','std_len','syn','rst','fin','ack','duration']].to_numpy()
    meta = agg.reset_index()  # keep who/where ports
    return X, meta

def main():
    ap = argparse.ArgumentParser()
    ap.add_argument('mode', choices=['train','score'])
    ap.add_argument('--features', required=True, help='CSV from pcap_to_packets.sh')
    ap.add_argument('--model', default='iforest.joblib')
    ap.add_argument('--contamination', type=float, default=0.02)
    args = ap.parse_args()

    X, meta = build_features(args.features)

    if args.mode == 'train':
        clf = IsolationForest(
            n_estimators=200,
            max_samples='auto',
            contamination=args.contamination,
            random_state=42,
            n_jobs=-1
        )
        clf.fit(X)
        dump({'model': clf}, args.model)
        print(f"Saved model -> {args.model}  (n={len(X)})")
    else:
        bundle = load(args.model)
        clf = bundle['model']
        scores = clf.decision_function(X)  # higher = less anomalous
        preds = clf.predict(X)             # -1 anomalous, 1 normal
        out = []
        for i, p in enumerate(preds):
            if p == -1:
                row = meta.iloc[i].to_dict()
                row['anomaly_score'] = float(-scores[i])  # invert for readability
                out.append(row)
        # Print top anomalies as JSON Lines
        out = sorted(out, key=lambda r: r['anomaly_score'], reverse=True)
        for r in out[:50]:
            print(json.dumps(r))
        print(f"Anomalies: {len(out)} / {len(X)}", flush=True)

if __name__ == '__main__':
    main()

Make executable:

chmod +x score_flows.py

Train on baseline:

# Choose a pcap you believe is "normal"
./pcap_to_packets.sh /var/log/pcap/cap-BASELINE.pcap baseline.csv
./score_flows.py train --features baseline.csv --model iforest.joblib --contamination 0.02

Step 4: Score new captures automatically

Wrap everything in a small loop. This prints likely-bad flows as JSON Lines you can feed into jq, Loki, Splunk, or email.

scan_pcaps.sh

#!/usr/bin/env bash
set -euo pipefail
PCAP_DIR="/var/log/pcap"
MODEL="iforest.joblib"

for pcap in "$PCAP_DIR"/*.pcap; do
  csv="${pcap%.pcap}.tcp.csv"
  if [[ ! -f "$csv" ]]; then
    echo "Extracting: $pcap"
    ./pcap_to_packets.sh "$pcap" "$csv"
    echo "Scoring: $csv"
    ./score_flows.py score --features "$csv" --model "$MODEL" | tee -a anomalies.jsonl
  fi
done

Run it periodically (cron/systemd timer) after the capture job. Example cron (runs every 10 minutes):

*/10 * * * * cd /path/to/ai-pcap && . .venv/bin/activate && ./scan_pcaps.sh

Step 5: Interpret results (what you’ll likely see)

  • Port scans and probes: Many short flows with SYNs and few ACKs; high syn, low bytes, short duration.

  • Beaconing malware: Regular intervals, consistent small byte patterns to single IP; similar feature vectors repeating to an external host.

  • Data exfiltration: Flows with unusually large bytes or long durations to rare destinations, outside normal business patterns.

Tune the detector:

  • Adjust contamination (expected anomaly rate); 0.5–5% is common to start.

  • Whitelist known noisy services (e.g., backups, monitoring IPs).

  • Retrain periodically with curated “known-good” traffic.

Combine with signatures:

  • Run Suricata alongside this pipeline. Use its alerts to label flows and build a supervised classifier later if you desire.

Real-world example workflow (end-to-end)

1) Start capture:

sudo tcpdump -i eth0 -s 96 -G 300 -W 12 -w /var/log/pcap/cap-%Y%m%d%H%M%S.pcap 'tcp or udp'

2) Build baseline on a quiet period:

./pcap_to_packets.sh /var/log/pcap/cap-20260701000000.pcap baseline.csv
./score_flows.py train --features baseline.csv --model iforest.joblib --contamination 0.02

3) Run continuous scoring:

./scan_pcaps.sh

4) Investigate top anomalies:

tail -n +1 anomalies.jsonl | head -50

Conclusion and Call to Action

You don’t need a data lake or a GPU farm to get value from AI on packet data. With tcpdump, tshark, and a few dozen lines of Python, you can:

  • Summarize raw packets into meaningful flow features

  • Baseline normal behavior

  • Surface likely bad or unusual traffic within minutes

Your next step:

  • Install the prerequisites for your distro

  • Capture an hour of traffic, train the baseline, and run the scorer

  • Iterate: tune contamination, whitelist noisy services, and schedule the job

If you want more fidelity, swap the feature extractor for Zeek conn.log, add timing features (inter-arrival variance), or feed Suricata alerts as labels. Keep it Bash-first, keep it explainable—and let the AI trim your PCAP haystack down to a handful of needles.

Related posts