Artificial Intelligence Networking Projects you can build on Linux (with Bash)

If you could predict congestion before it hits, classify traffic without DPI, or spot anomalies in seconds—all with a handful of shell commands and a few lines of Python—would you try it? This post shows you how to stitch together classic Linux networking tools with lightweight machine learning to build useful, real-world “AI for networking” projects on your own workstation or lab host.

You’ll get:

• A reproducible lab using network namespaces (no VMs required)

• A traffic classification model built from packet captures

• A simple anomaly detector for live interfaces

And yes—every tool comes with installation steps for apt, dnf, and zypper.

Why AI for Networking is worth your time

• Visibility: ML can summarize millions of packets into features and patterns faster than humans or static scripts.

• Speed: Simple models (think Random Forests, Isolation Forests) can run on a laptop and still surface actionable insights in seconds.

• Repeatability: Once you wire Bash + Python together, you can automate capture, labeling, training, and deployment in a loop.

• Low cost: You already have most of the tools; the rest are a few apt/dnf/zypper commands away.

Note: The goal here is practical outcomes with minimal overhead, not bleeding-edge deep learning.

---

Prerequisites and installation

You’ll need:

• A recent Linux distribution (Debian/Ubuntu, Fedora/RHEL, or openSUSE)

• Sudo rights

• Basic familiarity with Bash

We’ll use:

• Core networking tools: iproute2 (ip, tc), tcpdump, iperf3

• Python for ML and packet parsing: venv, pip, scapy, scikit-learn

Install the OS packages:

• Debian/Ubuntu (apt)

sudo apt update
sudo apt install -y python3 python3-venv python3-pip tcpdump iperf3 iproute2

• Fedora/RHEL (dnf)

sudo dnf install -y python3 python3-pip tcpdump iperf3 iproute

• openSUSE (zypper)

sudo zypper install -y python3 python3-pip tcpdump iperf3 iproute2

Create and activate a Python virtual environment (recommended):

python3 -m venv ~/.venvs/ainet
source ~/.venvs/ainet/bin/activate
python3 -m pip install --upgrade pip
python3 -m pip install scapy scikit-learn numpy pandas matplotlib

If you see build errors for Python packages on older systems, add development tools:

• Debian/Ubuntu:

sudo apt install -y build-essential python3-dev

• Fedora/RHEL:

sudo dnf install -y @development-tools python3-devel

• openSUSE:

sudo zypper install -y -t pattern devel_C_C++ python3-devel

---

Project 1: Spin up a tiny, repeatable network lab with ip netns

Network namespaces let you create routers and hosts on one machine. This script stands up two hosts (ns1, ns2) and a router (rtr), wires them with veth pairs, and applies latency/loss for testing.

Create lab.sh:

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

# Clean up from any prior run
for ns in ns1 ns2 rtr; do
  ip netns del "$ns" 2>/dev/null || true
done
ip link del veth-n1 2>/dev/null || true
ip link del veth-n2 2>/dev/null || true

# Create namespaces
ip netns add ns1
ip netns add ns2
ip netns add rtr

# veth pairs: ns1<->rtr and ns2<->rtr
ip link add veth-n1 type veth peer name veth-r1
ip link add veth-n2 type veth peer name veth-r2

ip link set veth-n1 netns ns1
ip link set veth-r1 netns rtr
ip link set veth-n2 netns ns2
ip link set veth-r2 netns rtr

# Addresses
ip netns exec ns1 ip addr add 10.0.1.2/24 dev veth-n1
ip netns exec rtr ip addr add 10.0.1.1/24 dev veth-r1
ip netns exec ns2 ip addr add 10.0.2.2/24 dev veth-n2
ip netns exec rtr ip addr add 10.0.2.1/24 dev veth-r2

# Bring links up
for n in ns1 ns2 rtr; do
  ip netns exec "$n" ip link set lo up
done
ip netns exec ns1 ip link set veth-n1 up
ip netns exec ns2 ip link set veth-n2 up
ip netns exec rtr ip link set veth-r1 up
ip netns exec rtr ip link set veth-r2 up

# Routing
ip netns exec ns1 ip route add default via 10.0.1.1
ip netns exec ns2 ip route add default via 10.0.2.1
ip netns exec rtr sysctl -w net.ipv4.ip_forward=1 >/dev/null

# Optional: emulate WAN with 30ms latency + 0.1% loss on rtr -> ns2
ip netns exec rtr tc qdisc add dev veth-r2 root netem delay 30ms loss 0.1%

echo "Lab ready:

- ns1 10.0.1.2 -> rtr 10.0.1.1 -> ns2 10.0.2.2
Try:
  ip netns exec ns2 iperf3 -s
  ip netns exec ns1 iperf3 -c 10.0.2.2 -t 10
  ip netns exec ns1 ping -c 3 10.0.2.2
To teardown: ip netns del ns1 ns2 rtr; ip link del veth-n1; ip link del veth-n2"

Run it:

sudo bash lab.sh

Generate traffic:

sudo ip netns exec ns2 iperf3 -s
sudo ip netns exec ns1 iperf3 -c 10.0.2.2 -t 10
sudo ip netns exec ns1 ping -c 3 10.0.2.2

Tip: Adjust tc netem parameters to simulate different WAN conditions:

sudo ip netns exec rtr tc qdisc change dev veth-r2 root netem delay 80ms loss 0.5% rate 50mbit

---

Project 2: Build a simple traffic classifier from pcaps

We’ll capture traffic in the lab, extract flow-level features with Python (scapy), and train a lightweight Random Forest to distinguish “short flows” from “bulk transfers.” This is a starter pattern you can extend to app identification or QoS tagging.

1) Capture labeled data

• Short flows (1 MiB bursts):

sudo ip netns exec ns2 iperf3 -s -1
sudo ip netns exec ns1 bash -c 'for i in {1..10}; do iperf3 -c 10.0.2.2 -n 1M; done' &
sudo ip netns exec rtr tcpdump -i veth-r1 -w short_flows.pcap -c 100000

• Bulk transfer:

sudo ip netns exec ns2 iperf3 -s -1
sudo ip netns exec ns1 iperf3 -c 10.0.2.2 -t 30 &
sudo ip netns exec rtr tcpdump -i veth-r1 -w bulk_flows.pcap -c 200000

2) Extract features and train

Save as train_classifier.py:

#!/usr/bin/env python3
import argparse, os
from scapy.all import rdpcap, TCP, UDP, IP
import numpy as np
import pandas as pd
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import train_test_split
from sklearn.metrics import classification_report

def flows_from_pcap(pcap_path, label):
    pkts = rdpcap(pcap_path)
    flows = {}
    for p in pkts:
        if not p.haslayer(IP): 
            continue
        ip = p[IP]
        proto = 'TCP' if p.haslayer(TCP) else 'UDP' if p.haslayer(UDP) else str(ip.proto)
        sport = p[TCP].sport if p.haslayer(TCP) else p[UDP].sport if p.haslayer(UDP) else 0
        dport = p[TCP].dport if p.haslayer(TCP) else p[UDP].dport if p.haslayer(UDP) else 0
        key = (ip.src, ip.dst, sport, dport, proto)
        ts = float(p.time)
        length = len(p)
        rec = flows.get(key, {"times": [], "sizes": [], "proto": proto})
        rec["times"].append(ts)
        rec["sizes"].append(length)
        flows[key] = rec
    rows = []
    for key, rec in flows.items():
        times = np.array(rec["times"])
        sizes = np.array(rec["sizes"])
        if len(times) < 2:
            iats = np.array([0.0])
        else:
            iats = np.diff(np.sort(times))
        feat = {
            "pkts": len(sizes),
            "bytes": sizes.sum(),
            "pkt_mean": sizes.mean(),
            "pkt_std": sizes.std() if len(sizes) > 1 else 0.0,
            "iat_mean": iats.mean() if len(iats) > 0 else 0.0,
            "iat_std": iats.std() if len(iats) > 1 else 0.0,
            "duration": (times.max() - times.min()) if len(times) > 1 else 0.0,
            "proto_tcp": 1 if rec["proto"] == "TCP" else 0,
            "proto_udp": 1 if rec["proto"] == "UDP" else 0,
            "label": label
        }
        rows.append(feat)
    return pd.DataFrame(rows)

def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--short", default="short_flows.pcap")
    ap.add_argument("--bulk", default="bulk_flows.pcap")
    args = ap.parse_args()
    df = pd.concat([
        flows_from_pcap(args.short, "short"),
        flows_from_pcap(args.bulk, "bulk")
    ], ignore_index=True).fillna(0)
    X = df.drop(columns=["label"])
    y = df["label"]
    Xtr, Xte, ytr, yte = train_test_split(X, y, test_size=0.3, random_state=42, stratify=y)
    clf = RandomForestClassifier(n_estimators=200, random_state=42, n_jobs=-1)
    clf.fit(Xtr, ytr)
    preds = clf.predict(Xte)
    print(classification_report(yte, preds, digits=4))
    # Save model and feature columns
    import joblib
    joblib.dump({"model": clf, "columns": X.columns.tolist()}, "traffic_rf.joblib")
    print("Saved model to traffic_rf.joblib with columns:", X.columns.tolist())

if __name__ == "__main__":
    main()

Run it:

source ~/.venvs/ainet/bin/activate
python3 train_classifier.py

You’ll see precision/recall. This is a toy dataset, but the pattern is production-friendly:

• Define labels you care about (e.g., interactive vs bulk; priority classes; suspected malware)

• Capture representative pcaps

• Extract robust, protocol-agnostic features

• Train, validate, iterate

3) Use the model on new captures

Save as classify_pcap.py:

#!/usr/bin/env python3
import sys, joblib
from scapy.all import rdpcap, IP, TCP, UDP
import numpy as np
import pandas as pd

model_bundle = joblib.load("traffic_rf.joblib")
clf = model_bundle["model"]
cols = model_bundle["columns"]

def df_from_pcap(pcap):
    pkts = rdpcap(pcap)
    flows = {}
    for p in pkts:
        if not p.haslayer(IP): 
            continue
        ip = p[IP]
        proto = 'TCP' if p.haslayer(TCP) else 'UDP' if p.haslayer(UDP) else str(ip.proto)
        sport = p[TCP].sport if p.haslayer(TCP) else p[UDP].sport if p.haslayer(UDP) else 0
        dport = p[TCP].dport if p.haslayer(TCP) else p[UDP].dport if p.haslayer(UDP) else 0
        key = (ip.src, ip.dst, sport, dport, proto)
        ts = float(p.time)
        length = len(p)
        rec = flows.get(key, {"times": [], "sizes": [], "proto": proto})
        rec["times"].append(ts)
        rec["sizes"].append(length)
        flows[key] = rec
    rows = []
    keys = []
    for key, rec in flows.items():
        times = np.array(rec["times"])
        sizes = np.array(rec["sizes"])
        iats = np.diff(np.sort(times)) if len(times) > 1 else np.array([0.0])
        feat = {
            "pkts": len(sizes),
            "bytes": sizes.sum(),
            "pkt_mean": sizes.mean(),
            "pkt_std": sizes.std() if len(sizes) > 1 else 0.0,
            "iat_mean": iats.mean() if len(iats) > 0 else 0.0,
            "iat_std": iats.std() if len(iats) > 1 else 0.0,
            "duration": (times.max() - times.min()) if len(times) > 1 else 0.0,
            "proto_tcp": 1 if rec["proto"] == "TCP" else 0,
            "proto_udp": 1 if rec["proto"] == "UDP" else 0
        }
        rows.append(feat)
        keys.append(key)
    df = pd.DataFrame(rows).fillna(0)
    return keys, df

if len(sys.argv) != 2:
    print("Usage: classify_pcap.py new_traffic.pcap")
    sys.exit(1)

keys, df = df_from_pcap(sys.argv[1])
# Align columns
for c in cols:
    if c not in df.columns:
        df[c] = 0
df = df[cols]
preds = clf.predict(df)
for k, p in zip(keys, preds):
    print(k, p)

Example usage:

sudo ip netns exec rtr tcpdump -i veth-r1 -w new_traffic.pcap -c 120000
python3 classify_pcap.py new_traffic.pcap

---

Project 3: Live anomaly detection with Isolation Forest

We’ll read interface counters, turn them into per-second rates, and run an unsupervised model (IsolationForest) to flag outliers in near real time.

Save as live_anomaly.py:

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

def read_bytes(iface):
    base = f"/sys/class/net/{iface}/statistics"
    with open(os.path.join(base, "rx_bytes")) as f: rx = int(f.read())
    with open(os.path.join(base, "tx_bytes")) as f: tx = int(f.read())
    return rx, tx

def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--iface", required=True, help="Interface name (e.g., veth-r1)")
    ap.add_argument("--interval", type=float, default=1.0)
    ap.add_argument("--window", type=int, default=60)
    args = ap.parse_args()

    buf = []
    # Warm up buffer
    rx_prev, tx_prev = read_bytes(args.iface)
    t_prev = time.time()
    time.sleep(args.interval)
    for _ in range(args.window):
        rx, tx = read_bytes(args.iface)
        t = time.time()
        dt = max(t - t_prev, 1e-6)
        rx_rate = (rx - rx_prev) * 8.0 / dt  # bps
        tx_rate = (tx - tx_prev) * 8.0 / dt
        buf.append([rx_rate, tx_rate])
        rx_prev, tx_prev, t_prev = rx, tx, t
        time.sleep(args.interval)

    clf = IsolationForest(n_estimators=200, contamination=0.05, random_state=42)
    X = np.array(buf)
    clf.fit(X)

    print(f"Model trained on {len(buf)} seconds of baseline. Monitoring now...")
    while True:
        rx, tx = read_bytes(args.iface)
        t = time.time()
        dt = max(t - t_prev, 1e-6)
        rx_rate = (rx - rx_prev) * 8.0 / dt
        tx_rate = (tx - tx_prev) * 8.0 / dt
        score = clf.decision_function([[rx_rate, tx_rate]])[0]
        pred = clf.predict([[rx_rate, tx_rate]])[0]  # -1 anomaly, 1 normal
        status = "ANOMALY" if pred == -1 else "normal"
        print(f"{time.strftime('%X')} {args.iface} rx={rx_rate:,.0f}bps tx={tx_rate:,.0f}bps score={score:.4f} {status}")
        rx_prev, tx_prev, t_prev = rx, tx, t
        time.sleep(args.interval)

if __name__ == "__main__":
    main()

Run it against your router namespace interface:

# In one terminal, generate traffic
sudo ip netns exec ns2 iperf3 -s
sudo ip netns exec ns1 iperf3 -c 10.0.2.2 -t 120

# In another terminal, monitor from within the router namespace
source ~/.venvs/ainet/bin/activate
sudo ip netns exec rtr python3 live_anomaly.py --iface veth-r1 --interval 1 --window 30

You’ll see per-second rates and anomaly flags. Change load patterns to watch the detector respond.

Notes:

• Start with a clean baseline window that represents “normal.”

• Add simple features (e.g., EWMA, packet rates) for more robust detection.

• Pipe output to syslog with logger if desired:

sudo ip netns exec rtr python3 live_anomaly.py --iface veth-r1 2>&1 | logger -t ainet

---

Real-world tips and extensions

• Data quality beats model complexity. Well-chosen flow features often outperform deep models on small/medium datasets.

• Automate the pipeline. Cron or systemd timers can capture, retrain nightly, and redeploy models.

• Keep it explainable. Start with tree-based models; feature importance helps justify actions to ops teams.

• Safety first. Use namespaces and tc netem to test before touching production routers.

• Version artifacts. Save your pcaps, feature scripts, and *.joblib models with timestamps and Git tags.

---

Conclusion and your next step

You’ve built:

• A self-contained, Linux-native network lab

• A flow-based classifier trained from your own captures

• A live anomaly detector you can point at any interface

From here:

• Extend the feature set (TLS SNI counts, SYN/FIN ratios, burstiness)

• Add alerting and dashboards (Prometheus, Grafana, Loki)

• Integrate actions (tag flows, adjust tc/qdisc, or open tickets)

If you found this useful, clone these snippets into your repo, script the capture/train/deploy loop, and iterate on your own traffic. The most powerful AI networking projects start small, run fast, and evolve with your data.