Posted on
• Artificial Intelligence

Artificial Intelligence Network Observability

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

AI-Powered Network Observability on Linux (with Bash): From Packets to Anomalies

Packets don’t lie—but they do overwhelm. If you’ve ever tailed a pcap at 2 a.m., you know the pain: too many events, too little signal. AI-driven network observability flips the script. Instead of hand-sifting, let the machine summarize behavior and flag anomalies. In this post, you’ll build a minimal, Linux-first pipeline that:

  • Collects packet/flow metadata with standard CLI tools

  • Featurizes traffic into CSV

  • Trains a tiny anomaly detector

  • Scores new traffic and surfaces “weird” events automatically

  • Automates the whole thing with Bash

No proprietary black boxes. Just open tools, scripts you can read, and knobs you can tune.

What problem are we solving?

  • Networks are fast, encrypted, and dynamic; humans can’t reliably spot subtle anomalies at scale.

  • Traditional alert rules miss new/unknown behaviors.

  • Full packet capture is heavy; metadata + ML is often enough to detect scans, exfiltration, beaconing, and lateral movement.

AI network observability brings:

  • Unsupervised detection of “odd” behavior without exhaustive rule-writing

  • Faster triage by ranking suspicious flows

  • Repeatable workflows you can automate and audit

What you’ll build (high level)

1) Collect flow-like metadata with tshark
2) Turn it into features (CSV)
3) Train a lightweight anomaly detector (Isolation Forest)
4) Score new captures and print top anomalies
5) Automate with a Bash wrapper and cron/systemd

Everything runs locally on Linux with common packages.

1) Install the toolkit

You’ll need packet capture and Python ML bits. Use one of the following based on your distro.

  • Debian/Ubuntu (apt):
sudo apt update
sudo apt install -y tcpdump tshark bpftrace nfdump python3 python3-pip python3-venv
  • Fedora/RHEL/CentOS Stream (dnf):
sudo dnf install -y tcpdump wireshark-cli bpftrace nfdump python3 python3-pip
# If you need virtualenv support explicitly:
# sudo dnf install -y python3-virtualenv
  • openSUSE Leap/Tumbleweed (zypper):
sudo zypper refresh
sudo zypper install -y tcpdump wireshark bpftrace nfdump python3 python3-pip

Verify you have TShark:

tshark -v

Note:

  • You generally need root/capabilities to capture. Using sudo is fine for a lab.

  • To capture as non-root, configure dumpcap capabilities and/or the wireshark group per your distro’s guidance.

Create a Python virtual environment for the AI bits:

python3 -m venv ~/ai-netobs-venv
. ~/ai-netobs-venv/bin/activate
pip install --upgrade pip
pip install pandas scikit-learn joblib

2) Capture and featurize traffic (CSV)

Pick an interface:

ip -br link show up

Export it (replace with your active interface, e.g., eth0 or enp0s31f6):

export IFACE=eth0

Capture 60 seconds of IP traffic and extract useful fields into CSV:

sudo tshark -i "$IFACE" -a duration:60 -Y "ip" -T fields \
  -e frame.time_epoch -e ip.proto -e ip.src -e ip.dst \
  -e tcp.srcport -e tcp.dstport -e udp.srcport -e udp.dstport \
  -e frame.len -e tcp.flags \
  -E header=y -E separator=, -E quote=d -E occurrence=f \
  > flows.csv

What you just did:

  • Limited capture to 60s for a small, rotating baseline

  • Output fields that work even with encrypted payloads (metadata still shines)

  • Produced flows.csv, which we’ll use for training and detection

Optional: quick NetFlow/IPFIX if you already have an exporter. nfdump can read flow files or listen, but that’s outside this minimal pipeline.

3) Train a tiny anomaly detector (unsupervised)

First collect a “baseline” capture during a quiet, known-good period—e.g., a few 1–5 minute samples concatenated. Then train:

Save as train_isoforest.py:

#!/usr/bin/env python3
import sys, os, math
import pandas as pd
from sklearn.ensemble import IsolationForest
from joblib import dump

if len(sys.argv) != 3:
    print("Usage: train_isoforest.py <baseline_flows.csv> <model_out.joblib>")
    sys.exit(1)

in_csv, out_model = sys.argv[1], sys.argv[2]
df = pd.read_csv(in_csv)

# Normalize port fields: pick TCP port if present, else UDP, else 0
def to_int(x):
    try:
        if isinstance(x, str) and x.startswith("0x"):
            return int(x, 16)
        return int(float(x))
    except Exception:
        return 0

for col in ["ip.proto","tcp.srcport","tcp.dstport","udp.srcport","udp.dstport","frame.len","tcp.flags"]:
    if col in df.columns:
        df[col] = df[col].apply(to_int)
    else:
        df[col] = 0

df["src_port"] = df[["tcp.srcport","udp.srcport"]].max(axis=1)
df["dst_port"] = df[["tcp.dstport","udp.dstport"]].max(axis=1)
df["is_tcp"] = (df["ip.proto"] == 6).astype(int)
df["is_udp"] = (df["ip.proto"] == 17).astype(int)

features = df[["ip.proto","src_port","dst_port","frame.len","tcp.flags","is_tcp","is_udp"]].fillna(0)

# Isolation Forest: great default for unsupervised anomaly detection
model = IsolationForest(
    n_estimators=200,
    contamination=0.01,  # tweak: expected fraction of anomalies in baseline
    random_state=42
)
model.fit(features)

os.makedirs(os.path.dirname(out_model) or ".", exist_ok=True)
dump(model, out_model)
print(f"Trained model saved to {out_model}")

Train it (assuming your venv is active):

python3 train_isoforest.py flows.csv model.joblib

Tip:

  • Re-train occasionally as your “normal” evolves.

  • Adjust contamination if you see too many/too few anomalies.

4) Detect anomalies in near real time

Create a scoring script to run on fresh captures.

Save as score_flows.py:

#!/usr/bin/env python3
import sys
import pandas as pd
from joblib import load

if len(sys.argv) != 3:
    print("Usage: score_flows.py <flows.csv> <model.joblib>")
    sys.exit(1)

in_csv, in_model = sys.argv[1], sys.argv[2]
df = pd.read_csv(in_csv)

def to_int(x):
    try:
        if isinstance(x, str) and x.startswith("0x"):
            return int(x, 16)
        return int(float(x))
    except Exception:
        return 0

for col in ["ip.proto","tcp.srcport","tcp.dstport","udp.srcport","udp.dstport","frame.len","tcp.flags"]:
    if col in df.columns:
        df[col] = df[col].apply(to_int)
    else:
        df[col] = 0

df["src_port"] = df[["tcp.srcport","udp.srcport"]].max(axis=1)
df["dst_port"] = df[["tcp.dstport","udp.dstport"]].max(axis=1)
df["is_tcp"] = (df["ip.proto"] == 6).astype(int)
df["is_udp"] = (df["ip.proto"] == 17).astype(int)

features = df[["ip.proto","src_port","dst_port","frame.len","tcp.flags","is_tcp","is_udp"]].fillna(0)

model = load(in_model)

# Predictions: -1 = anomaly, 1 = normal
pred = model.predict(features)
scores = model.decision_function(features)  # lower = more anomalous

df_out = df.copy()
df_out["anomaly"] = pred
df_out["score"] = scores

# Show top 20 most anomalous rows (lowest scores)
res = df_out.sort_values(by="score").head(20)

cols = ["frame.time_epoch","ip.src","ip.dst","src_port","dst_port","frame.len","ip.proto","tcp.flags","anomaly","score"]
cols = [c for c in cols if c in res.columns]
print(res[cols].to_csv(index=False))

Test it against a fresh short capture:

sudo tshark -i "$IFACE" -a duration:15 -Y "ip" -T fields \
  -e frame.time_epoch -e ip.proto -e ip.src -e ip.dst \
  -e tcp.srcport -e tcp.dstport -e udp.srcport -e udp.dstport \
  -e frame.len -e tcp.flags \
  -E header=y -E separator=, -E quote=d -E occurrence=f \
  > new_flows.csv

python3 score_flows.py new_flows.csv model.joblib

You’ll get a CSV of the most “weird” flows first.

5) Automate with a Bash wrapper (cron/systemd friendly)

Save as ai-net-observe.sh:

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

# Config
IFACE="${IFACE:-$(ip route show default 2>/dev/null | awk '/default/ {print $5; exit}')}"

DURATION="${DURATION:-30}"       # seconds per sample
WORKDIR="${WORKDIR:-/var/tmp/ai-netobs}"
MODEL="${MODEL:-$WORKDIR/model.joblib}"
VENV="${VENV:-$HOME/ai-netobs-venv}"

mkdir -p "$WORKDIR"

if [[ ! -f "$MODEL" ]]; then
  echo "Model not found at $MODEL"
  echo "Train once with: python3 train_isoforest.py <baseline.csv> $MODEL"
  exit 2
fi

STAMP=$(date +%Y%m%d-%H%M%S)
CSV="$WORKDIR/flows-$STAMP.csv"

# Capture slice
sudo tshark -i "$IFACE" -a "duration:$DURATION" -Y "ip" -T fields \
  -e frame.time_epoch -e ip.proto -e ip.src -e ip.dst \
  -e tcp.srcport -e tcp.dstport -e udp.srcport -e udp.dstport \
  -e frame.len -e tcp.flags \
  -E header=y -E separator=, -E quote=d -E occurrence=f \
  > "$CSV"

# Score
. "$VENV/bin/activate"
python3 -u - <<'PY' "$CSV" "$MODEL" | tee "$WORKDIR/alerts-$STAMP.csv" | head -n 25 | sed 's/^/ANOMALY: /' | logger -t ai-netobs
import sys
import pandas as pd
from joblib import load

in_csv, in_model = sys.argv[1], sys.argv[2]
df = pd.read_csv(in_csv)

def to_int(x):
    try:
        if isinstance(x, str) and x.startswith("0x"):
            return int(x, 16)
        return int(float(x))
    except Exception:
        return 0

for col in ["ip.proto","tcp.srcport","tcp.dstport","udp.srcport","udp.dstport","frame.len","tcp.flags"]:
    if col in df.columns:
        df[col] = df[col].apply(to_int)
    else:
        df[col] = 0

df["src_port"] = df[["tcp.srcport","udp.srcport"]].max(axis=1)
df["dst_port"] = df[["tcp.dstport","udp.dstport"]].max(axis=1)
df["is_tcp"] = (df["ip.proto"] == 6).astype(int)
df["is_udp"] = (df["ip.proto"] == 17).astype(int)

features = df[["ip.proto","src_port","dst_port","frame.len","tcp.flags","is_tcp","is_udp"]].fillna(0)
model = load(in_model)

pred = model.predict(features)
scores = model.decision_function(features)

df_out = df.copy()
df_out["anomaly"] = pred
df_out["score"] = scores

res = df_out[df_out["anomaly"] == -1].sort_values(by="score").head(50)

cols = ["frame.time_epoch","ip.src","ip.dst","src_port","dst_port","frame.len","ip.proto","tcp.flags","anomaly","score"]
cols = [c for c in cols if c in res.columns]
print(res[cols].to_csv(index=False))
PY

echo "Done. Check journalctl -t ai-netobs for recent anomalies."

Make it executable:

chmod +x ai-net-observe.sh

Run it ad-hoc:

./ai-net-observe.sh
journalctl -t ai-netobs -n 50 --no-pager

Cron it every 5 minutes:

( crontab -l 2>/dev/null; echo "*/5 * * * * IFACE=$IFACE DURATION=30 $PWD/ai-net-observe.sh >/dev/null 2>&1" ) | crontab -

Or create a simple systemd unit for steadier scheduling.

Real-world patterns this catches

  • Port scans: bursts of small TCP SYNs to many destination ports. Features: many flows with small frame.len, varying dst_port, tcp.flags set for SYN, unusual for baseline.

  • Data exfiltration spikes: large frame.len out to rare external IPs or ports not common for your environment.

  • Beaconing/malware callbacks: low-volume but periodic traffic to a single external ip.dst/dst_port, atypical relative to learned normal.

This pipeline won’t replace a full IDS, but it’s a strong, explainable signal booster.

Why this approach works (and trade-offs)

  • Encryption-proof: it uses headers and sizes, not payloads.

  • Lightweight: no full packet retention required for detection.

  • Unsupervised: flags “new” things without labels.

  • Trade-offs: false positives early on; model needs periodic retraining; metadata features are coarser than full DPI.

Tuning tips:

  • Adjust contamination in Isolation Forest.

  • Engineer more features (e.g., rolling rates per src/dst, entropy of dst_port, inter-arrival time deltas).

  • Maintain separate baselines for weekdays vs weekends, or office vs data center segments.

Optional: quick bpftrace sanity view

If you want a fast kernel-level sanity check (root required), this view counts connects by process:

sudo bpftrace -e 'tracepoint:syscalls:sys_enter_connect { @[comm] = count(); } interval:s:10 { print(@); clear(@); }'

Note: Tracepoint args vary by kernel and arch; keep this as a quick pulse, not a core dependency.

Conclusion and Call to Action

You just built a practical, Bash-first AI observability loop:

  • Capture metadata with TShark

  • Train a compact unsupervised model

  • Score new traffic and log anomalies

  • Automate it on your Linux host

Next steps:

  • Baseline on known-good periods and re-train weekly.

  • Add features (rates per IP, unique port counts, time-of-day buckets).

  • Push results to your SIEM, or alert with email/Slack from the wrapper script.

  • Roll this to a sensor on each segment for east–west visibility.

If you try this on your network, start small—one interface, one host—and iterate. Packets don’t lie, but they do appreciate good company: a little AI.

Related posts