Posted on
• Artificial Intelligence

Artificial Intelligence Network Monitoring

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

Artificial Intelligence Network Monitoring: From Packets to Actionable Anomalies (with Bash)

Your network is talking—are you listening? Traditional monitoring floods you with logs and signature hits. But attackers know the rules, encryption hides payloads, and your team is drowning in noise. AI-driven monitoring helps you surface what matters: unusual behaviors, outliers, and weak signals that often precede incidents. In this guide, you’ll build a lightweight, Linux-friendly pipeline that turns raw traffic into anomaly alerts using open-source tools and a sprinkle of machine learning—without leaving your terminal.

Why AI for Network Monitoring?

  • Scale and encryption: As east–west traffic grows and TLS becomes universal, signature-only IDS misses behavior-based threats. AI models can learn baselines from metadata (flow-level features) and flag deviations—no DPI required.

  • Unknown unknowns: Unsupervised models (e.g., Isolation Forest) are well-suited to find novel or low-and-slow patterns that don’t match static signatures.

  • Fit for Linux ops: You can glue together Zeek/Suricata, tshark, and Python ML with Bash into a resilient, automatable pipeline you control.

What follows: a practical, step-by-step setup—install, capture, learn, alert—plus real-world examples and automation tips.

1) Install and Prepare the Toolchain

We’ll use:

  • Packet/flow tools: Zeek (preferred) or Suricata, plus tcpdump/tshark

  • Python for ML: pandas, scikit-learn, joblib

  • Optional: jq for JSON manipulation

Package names can vary slightly by distro/version. Replace eth0 later with your NIC.

Apt (Debian/Ubuntu):

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

DNF (Fedora/RHEL/CentOS; for RHEL/CentOS you may need EPEL enabled for some packages):

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

Zypper (openSUSE/SLE; add Packman or relevant repos if needed):

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

Create a Python virtual environment for the ML bits:

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

Tip: tshark capture as non-root requires additional capabilities; otherwise use sudo. For a lab, sudo is fine.

2) Capture Flow-Level Features (Low-Overhead + AI-Friendly)

Zeek provides rich, structured logs that are perfect for ML. Let’s run Zeek and collect conn.log (connections summary).

Create a working directory and start Zeek:

mkdir -p ~/zeek-run && cd ~/zeek-run
sudo zeek -i eth0

Zeek will create ./logs/current/conn.log (tab-separated). It contains fields like:

  • ts, id.orig_h, id.resp_h, id.orig_p, id.resp_p, proto

  • duration, orig_bytes, resp_bytes, orig_pkts, resp_pkts, conn_state

Alternative (if you prefer tshark CSV directly):

sudo tshark -i eth0 -T fields \
  -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 -e ip.len \
  -E header=y -E separator=, \
  >> ~/flows.csv

But we’ll proceed with Zeek for better semantics and fewer parsing headaches.

3) Train and Score with a Lightweight Unsupervised Model

We’ll train a simple Isolation Forest on recent Zeek connection features and score the latest window for anomalies. This doesn’t require labeled data and works well as a starting point.

Save this as ~/ai-netmon/ai_netmon.py:

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

# Paths
ZEEK_CONN = os.environ.get("ZEEK_CONN", os.path.expanduser("~/zeek-run/logs/current/conn.log"))
MODEL_DIR = os.environ.get("MODEL_DIR", os.path.expanduser("~/ai-netmon"))
MODEL_PATH = os.path.join(MODEL_DIR, "isoforest.joblib")
SCALER_PATH = os.path.join(MODEL_DIR, "scaler.joblib")

os.makedirs(MODEL_DIR, exist_ok=True)

def load_zeek_conn(path, min_time_epoch=None):
    if not os.path.exists(path):
        return pd.DataFrame()
    # Zeek TSV with '#' comments
    df = pd.read_csv(path, sep="\t", comment="#", low_memory=False)
    # Ensure required columns exist
    needed = ["ts","id.orig_h","id.resp_h","id.orig_p","id.resp_p","proto",
              "duration","orig_bytes","resp_bytes","orig_pkts","resp_pkts","conn_state"]
    for n in needed:
        if n not in df.columns:
            df[n] = np.nan
    # Convert timestamp to epoch if string
    # Zeek ts is float(epoch)
    try:
        df["ts"] = pd.to_numeric(df["ts"], errors="coerce")
    except Exception:
        df["ts"] = np.nan
    if min_time_epoch is not None:
        df = df[df["ts"] >= float(min_time_epoch)]
    return df.dropna(subset=["id.orig_h","id.resp_h"])

def featurize(df):
    # Numeric-safe transforms with log1p to reduce skew
    for c in ["duration","orig_bytes","resp_bytes","orig_pkts","resp_pkts"]:
        df[c] = pd.to_numeric(df[c], errors="coerce").fillna(0.0)
    X = pd.DataFrame({
        "log1p_duration": np.log1p(df["duration"]),
        "log1p_orig_bytes": np.log1p(df["orig_bytes"]),
        "log1p_resp_bytes": np.log1p(df["resp_bytes"]),
        "log1p_orig_pkts": np.log1p(df["orig_pkts"]),
        "log1p_resp_pkts": np.log1p(df["resp_pkts"]),
        # Basic categorical encodings
        "proto_tcp": (df["proto"] == "tcp").astype(int),
        "proto_udp": (df["proto"] == "udp").astype(int),
        "dport_53": (pd.to_numeric(df["id.resp_p"], errors="coerce") == 53).astype(int),
        "dport_80": (pd.to_numeric(df["id.resp_p"], errors="coerce") == 80).astype(int),
        "dport_443": (pd.to_numeric(df["id.resp_p"], errors="coerce") == 443).astype(int),
    })
    return X

def ensure_model(X_ref):
    # Train a model if not present. Use a rolling unsupervised baseline.
    if os.path.exists(MODEL_PATH) and os.path.exists(SCALER_PATH):
        clf = joblib.load(MODEL_PATH)
        scaler = joblib.load(SCALER_PATH)
        return clf, scaler, False
    scaler = StandardScaler()
    Xs = scaler.fit_transform(X_ref)
    clf = IsolationForest(
        n_estimators=150,
        contamination=0.02,  # expected outlier proportion; tune per environment
        random_state=42,
        n_jobs=-1
    )
    clf.fit(Xs)
    joblib.dump(clf, MODEL_PATH)
    joblib.dump(scaler, SCALER_PATH)
    return clf, scaler, True

def main():
    # Score last N seconds. Train if model is missing using a longer window.
    lookback_secs = int(os.environ.get("LOOKBACK_SECS", "300"))
    train_window_secs = int(os.environ.get("TRAIN_WINDOW_SECS", "3600"))
    now = time.time()
    df_train = load_zeek_conn(ZEEK_CONN, min_time_epoch=now - train_window_secs)
    if df_train.empty:
        print("[]")
        return
    X_train = featurize(df_train)
    clf, scaler, trained_new = ensure_model(X_train)

    df_score = load_zeek_conn(ZEEK_CONN, min_time_epoch=now - lookback_secs)
    if df_score.empty:
        print("[]")
        return
    X_score = featurize(df_score)
    Xs = scaler.transform(X_score)
    scores = clf.decision_function(Xs)  # higher is more normal
    preds = clf.predict(Xs)  # -1 = anomaly, 1 = normal

    anomalies = []
    for i, p in enumerate(preds):
        if p == -1:
            row = df_score.iloc[i]
            anomalies.append({
                "ts": float(row.get("ts", 0)),
                "src": str(row.get("id.orig_h", "")),
                "sport": int(pd.to_numeric(row.get("id.orig_p", 0), errors="coerce") or 0),
                "dst": str(row.get("id.resp_h", "")),
                "dport": int(pd.to_numeric(row.get("id.resp_p", 0), errors="coerce") or 0),
                "proto": str(row.get("proto", "")),
                "duration": float(row.get("duration", 0) or 0),
                "orig_bytes": float(row.get("orig_bytes", 0) or 0),
                "resp_bytes": float(row.get("resp_bytes", 0) or 0),
                "score": float(scores[i])
            })
    print(json.dumps(anomalies, indent=2))

if __name__ == "__main__":
    main()

Make it executable:

chmod +x ~/ai-netmon/ai_netmon.py

Test it:

source ~/ai-netmon-venv/bin/activate
LOOKBACK_SECS=1200 TRAIN_WINDOW_SECS=7200 ZEEK_CONN=~/zeek-run/logs/current/conn.log \
  ~/ai-netmon/ai_netmon.py

You should see a JSON array. If empty, generate some traffic (browse, curl, dig) and rerun.

Tuning tips:

  • Increase TRAIN_WINDOW_SECS to stabilize baselines.

  • Adjust contamination (0.005–0.05 typical) for your environment’s variability.

  • Add more features (e.g., DNS query length stats from dns.log) as you mature.

4) Automate Scoring and Alerting with Bash + systemd

A tiny Bash wrapper runs the model, logs anomalies to syslog, and writes a file.

~/ai-netmon/run_ai_netmon.sh:

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

VENV="${HOME}/ai-netmon-venv"
ZEEK_CONN="${HOME}/zeek-run/logs/current/conn.log"
OUT="${HOME}/ai-netmon/anomalies.json"
LOOKBACK_SECS="${LOOKBACK_SECS:-300}"
TRAIN_WINDOW_SECS="${TRAIN_WINDOW_SECS:-3600}"

source "${VENV}/bin/activate"
export ZEEK_CONN LOOKBACK_SECS TRAIN_WINDOW_SECS

ANOMALIES="$("${HOME}/ai-netmon/ai_netmon.py")"
echo "${ANOMALIES}" > "${OUT}"

COUNT="$(echo "${ANOMALIES}" | jq 'length')"
if [[ "${COUNT}" -gt 0 ]]; then
  logger -t ai-netmon "Detected ${COUNT} network anomalies (last ${LOOKBACK_SECS}s). See ${OUT}"
  # Optional: print top 3
  echo "${ANOMALIES}" | jq '.[0:3]'
fi

Make it executable:

chmod +x ~/ai-netmon/run_ai_netmon.sh

Install jq if missing:

  • apt:
sudo apt install -y jq
  • dnf:
sudo dnf -y install jq
  • zypper:
sudo zypper install -y jq

Run manually:

~/ai-netmon/run_ai_netmon.sh
journalctl -t ai-netmon -n 5

Optional: systemd timer to run every 5 minutes.

~/.config/systemd/user/ai-netmon.service:

[Unit]
Description=AI Network Monitoring Scorer

[Service]
Type=oneshot
ExecStart=%h/ai-netmon/run_ai_netmon.sh

~/.config/systemd/user/ai-netmon.timer:

[Unit]
Description=Run AI NetMon every 5 minutes

[Timer]
OnBootSec=2min
OnUnitActiveSec=5min
Persistent=true

[Install]
WantedBy=timers.target

Enable and start:

systemctl --user daemon-reload
systemctl --user enable --now ai-netmon.timer
systemctl --user list-timers | grep ai-netmon

Prefer system-wide? Drop the units in /etc/systemd/system/ and use sudo systemctl.

5) Real-World Uses You Can Try Today

  • DNS tunneling and exfiltration

    • Symptom: spikes in DNS volume, long/variable query lengths, lots of NXDOMAINs.
    • How: Extend features with dns.log stats (mean query length per src, QPS bursts). Unsupervised models flag outliers—investigate domains and hosts.

  • Lateral movement

    • Symptom: unusual SMB/RDP/WinRM connections from non-admin subnets; new lateral flows after hours.
    • How: Encode destination ports/services (445/3389/5985), time-of-day; combine with connection counts per src. Outliers surface rogue pivots.

  • Crypto-mining or C2 beacons

    • Symptom: long-lived low-throughput TCP sessions to unusual ASNs or fixed intervals.
    • How: Leverage duration, orig_bytes/resp_bytes ratios, and periodicity approximations (inter-arrival uniformity). High-duration/low-byte flows often stand out.

  • Port scanning and recon

    • Symptom: many short connections across increasing destination ports from a single host.
    • How: Aggregate connections per src over a short window; add “unique dports per src” as a feature. Spikes are flagged quickly.

Remember: AI signals are leads, not verdicts. Combine with IDS alerts (Suricata), asset context, and change windows.

Troubleshooting and Hardening

  • No data? Check interface name (ip link), permissions, and whether Zeek is writing to logs/current/.

  • Too many false positives? Increase TRAIN_WINDOW_SECS, reduce contamination, or enrich features to better reflect “normal” traffic.

  • Performance: Zeek is efficient; if traffic is heavy, rotate logs more frequently or sample (e.g., -s 1:10 in tcpdump-based pipelines).

  • Persistence: Periodically retrain the model (e.g., daily cron/systemd service) to follow environment drift.

  • Security: Restrict who can read logs and the model dir; consider running Zeek/ML under a dedicated user.

Conclusion and Next Steps (CTA)

You just built a functional AI-assisted network monitor using Linux-native tools and Bash. It ingests real traffic, learns a baseline, and raises anomaly alerts on a schedule—no black boxes required.

Where to go from here:

  • Enrich features with DNS/HTTP logs and time-based aggregations.

  • Add a small SQLite or Parquet store for rolling windows and faster training.

  • Correlate with Suricata alerts to prioritize anomalies with signature matches.

  • Ship anomalies to your SIEM via syslog or webhooks.

Ready to evolve your network visibility? Start by letting this run for a week, tune contamination, and review top anomalies daily. Your network will tell you what’s weird—now you can hear it.

Related posts