Posted on
Artificial Intelligence

Artificial Intelligence Networking Case Studies

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

Artificial Intelligence Networking Case Studies (for Bash-first Linux users)

Your network talks. Can you teach it to think? Between the floods of packets, logs, and alerts, even seasoned admins can miss the early signs of trouble. AI doesn’t have to mean complex clusters and unicorn data science—on a Linux shell, a few lines of Bash plus small ML helpers can turn raw traffic into decisions.

In this article, you’ll:

  • See why AI is a practical fit for day‑to‑day networking.

  • Reproduce 3 hands‑on case studies that you can run on a laptop or a jump host.

  • Use only standard Linux tooling plus a minimal Python ML stack.

  • Get copy/pasteable install commands for apt, dnf, and zypper.

All examples are designed to be run in lab/test environments or with proper authorization.


Why AI for Networking (and why now)

  • Data is already there. Packet captures (pcap), NetFlow/IPFIX, DNS logs, and syslogs are standard—no new agents required.

  • Telemetry is “AI‑ready.” Simple features like packet counts, byte rates, RTT, and domain name stats can drive effective anomaly and capacity models.

  • Tooling is mature. TShark, iperf3, iproute2 (tc), awk, jq, and Python’s scikit‑learn let you build repeatable pipelines without heavy dependencies.

  • Real payoff. Faster incident triage, fewer false positives, better QoS, and early detection of covert channels.


Prerequisites and setup

The following packages will be used across the case studies:

  • Packet capture and parsing: tshark, tcpdump

  • Traffic generation and shaping: iperf3, iproute2 (tc)

  • Scripting/processing: python3, python3-pip, jq, gawk

  • Optional but recommended: granting dumpcap capability to capture without root

Install on your distro:

  • Debian/Ubuntu (apt):
sudo apt update
sudo apt install -y wireshark-common tshark tcpdump iproute2 iperf3 python3 python3-pip jq gawk ca-certificates
  • Fedora/RHEL/CentOS Stream (dnf):
sudo dnf install -y wireshark-cli tcpdump iproute iperf3 python3 python3-pip jq gawk ca-certificates
  • openSUSE (zypper):
sudo zypper refresh
sudo zypper install -y wireshark-cli tcpdump iproute2 iperf3 python3 python3-pip jq gawk ca-certificates

Python ML dependencies (user‑local):

pip3 install --user numpy pandas scikit-learn

Optional: allow non‑root captures (adjust path if needed):

sudo setcap 'CAP_NET_RAW+eip CAP_NET_ADMIN+eip' /usr/bin/dumpcap

Note: On Debian/Ubuntu, sudo dpkg-reconfigure wireshark-common can also enable non‑root captures. Always follow your org’s security policy.


Case Study 1 — Rapid DDoS anomaly detection from live traffic

Goal: Identify sources exhibiting abnormal per-second packet/byte behavior using an unsupervised model (Isolation Forest).

What you’ll do: 1) Extract per‑second features (packets, bytes) per source IP with TShark + awk. 2) Fit an Isolation Forest and rank suspicious IPs.

Step 1: Extract features (from live interface or pcap)

  • Live (replace eth0 with your interface):
sudo tshark -i eth0 -Y "ip" -T fields -e frame.time_epoch -e ip.src -e frame.len \
| awk 'BEGIN{OFS=","} {sec=int($1); key=sec","$2; pkts[key]++; bytes[key]+=$3}
       END{for (k in pkts) print k, pkts[k], bytes[k]}' \
| sort -t, -k1,1n -k2,2 > features.csv
  • From a pcap:
tshark -r sample.pcap -Y "ip" -T fields -e frame.time_epoch -e ip.src -e frame.len \
| awk 'BEGIN{OFS=","} {sec=int($1); key=sec","$2; pkts[key]++; bytes[key]+=$3}
       END{for (k in pkts) print k, pkts[k], bytes[k]}' \
| sort -t, -k1,1n -k2,2 > features.csv

This produces CSV rows: timestamp_sec,src_ip,pkts,bytes.

Step 2: Detect anomalies with Isolation Forest

cat > detect_isoforest.py << 'PY'
import sys
import pandas as pd
from sklearn.ensemble import IsolationForest

# Read features.csv (no header): ts,src,pkts,bytes
df = pd.read_csv('features.csv', header=None, names=['ts','src','pkts','bytes'])
# Keep numeric features
X = df[['pkts','bytes']].astype('float64')

# Fit unsupervised model
model = IsolationForest(n_estimators=200, contamination=0.02, random_state=42)
model.fit(X)

# Lower score => more anomalous
df['score'] = model.score_samples(X)

# Aggregate by source IP to get a stable ranking
agg = df.groupby('src').agg(
    obs=('src','count'),
    pkts_sum=('pkts','sum'),
    bytes_sum=('bytes','sum'),
    score_median=('score','median'),
    score_min=('score','min')
).reset_index()

# Rank by ascending median score (most suspicious first)
suspects = agg.sort_values('score_median').head(20)

print("src,obs,pkts_sum,bytes_sum,score_median,score_min")
for _, r in suspects.iterrows():
    print(f"{r.src},{r.obs},{int(r.pkts_sum)},{int(r.bytes_sum)},{r.score_median:.4f},{r.score_min:.4f}")
PY

python3 detect_isoforest.py

Output will show top candidates to investigate. In a real incident, you can cross-reference with firewall logs and apply temporary rate limits or blocks.

Pro tip: Run the feature extractor in a tmux session and rotate features.csv per minute for rolling detection.


Case Study 2 — Self‑tuning QoS with tc (bandit‑style exploration)

Goal: Automatically find the highest safe egress rate for a class while keeping latency under a target. This is a pragmatic “explore‑exploit” controller that you can later replace with a regression or reinforcement learner.

What you’ll do: 1) Create an HTB shaper with tc. 2) Probe latency under candidate rates and pick the best that meets your SLO (e.g., avg RTT < 30 ms).

Step 1: Basic HTB setup (egress on eth0)

DEV=eth0
sudo tc qdisc del dev "$DEV" root 2>/dev/null || true
sudo tc qdisc add dev "$DEV" root handle 1: htb default 10
sudo tc class add dev "$DEV" parent 1: classid 1:1 htb rate 100mbit
sudo tc class add dev "$DEV" parent 1:1 classid 1:10 htb rate 5mbit ceil 100mbit

Optionally, direct a flow (e.g., iperf3 TCP to a known host) to this class:

# Example filter: match dest TCP port 5201
sudo tc filter add dev "$DEV" protocol ip parent 1: prio 1 u32 \
  match ip dport 5201 0xffff flowid 1:10

Step 2: Explore candidate rates and measure RTT

cat > rate_search.sh << 'SH'
#!/usr/bin/env bash
set -euo pipefail

DEV="${1:-eth0}"
TARGET_HOST="${2:-8.8.8.8}"
TARGET_RTT_MS="${3:-30}"
CANDIDATE_RATES=("5mbit" "10mbit" "20mbit" "30mbit" "40mbit" "50mbit")

best_rate=""
for rate in "${CANDIDATE_RATES[@]}"; do
  sudo tc class change dev "$DEV" parent 1:1 classid 1:10 htb rate "$rate" ceil 100mbit
  # short settle time
  sleep 1
  # measure average RTT with 10 pings
  avg=$(ping -c 10 -i 0.2 -w 5 "$TARGET_HOST" | awk -F'/' '/^rtt|^round-trip/ {print $5}')
  # On some distros the line starts with "rtt", others with "round-trip"
  if [[ -z "$avg" ]]; then
    avg=$(ping -c 10 -i 0.2 -w 5 "$TARGET_HOST" | awk -F'/' '/^rtt|^round-trip/ {print $5}')
  fi
  echo "Rate $rate => avg RTT ${avg} ms"
  # Compare as integers
  avg_int=${avg%.*}
  if [[ -n "$avg_int" && "$avg_int" -le "$TARGET_RTT_MS" ]]; then
    best_rate="$rate"
  else
    # once we exceed target, break (assumes monotonic)
    break
  fi
done

if [[ -n "$best_rate" ]]; then
  echo "Selecting best rate: $best_rate"
  sudo tc class change dev "$DEV" parent 1:1 classid 1:10 htb rate "$best_rate" ceil 100mbit
else
  echo "No candidate met the RTT target (${TARGET_RTT_MS} ms). Keeping minimum."
fi
SH

chmod +x rate_search.sh
./rate_search.sh eth0 8.8.8.8 30

Notes:

  • Replace 8.8.8.8 with a relevant probe (e.g., your upstream gateway).

  • To emulate load during probing, run an iperf3 flow simultaneously:

iperf3 -c <your-iperf3-server> -t 60 -R
  • Extend this into a smarter controller by logging (rate, load) → RTT data and training a small regression model to predict the best rate at current load.

Rollback:

sudo tc qdisc del dev eth0 root

Case Study 3 — Detecting DNS tunneling via entropy of query names

Goal: Find suspiciously long, high‑entropy DNS queries that may indicate tunneling or DGA activity.

What you’ll do: 1) Stream DNS queries with TShark. 2) Compute length and Shannon entropy of dns.qry.name. 3) Flag suspicious queries using a simple threshold or a one‑class model.

Step 1: Stream DNS queries (requests only)

sudo tshark -i eth0 -Y "dns.flags.response == 0 and udp.port == 53" \
  -T fields -e frame.time_epoch -e ip.src -e dns.qry.name > dns_stream.txt

Step 2: Score queries

cat > dns_tunnel_detect.py << 'PY'
import sys, math
from collections import Counter
import argparse

def shannon_entropy(s: str) -> float:
    if not s:
        return 0.0
    counts = Counter(s)
    n = len(s)
    return -sum((c/n)*math.log2(c/n) for c in counts.values())

parser = argparse.ArgumentParser()
parser.add_argument("--min-length", type=int, default=50, help="Minimum query length to consider")
parser.add_argument("--entropy-thresh", type=float, default=3.5, help="Entropy threshold")
args = parser.parse_args()

print("ts,src,query,length,entropy,suspicious")
with open('dns_stream.txt') as f:
    for line in f:
        parts = line.strip().split('\t')
        if len(parts) < 3:
            # TShark may separate with tabs or spaces; try split on whitespace
            parts = line.strip().split()
            if len(parts) < 3:
                continue
        ts, src, qname = parts[0], parts[1], parts[2]
        # Consider only the label portion; remove dots for entropy calc
        clean = qname.replace('.', '')
        length = len(clean)
        ent = shannon_entropy(clean.lower())
        suspicious = (length >= args.min_length and ent >= args.entropy_thresh)
        print(f"{ts},{src},{qname},{length},{ent:.2f},{int(suspicious)}")
PY

python3 dns_tunnel_detect.py | awk -F',' '$6==1 {print $0}' | head -n 50

Enhancements:

  • Aggregate by domain suffix (e.g., second‑level) to see persistent tunneling roots.

  • Replace the threshold with a one‑class SVM trained on your baseline:

pip3 install --user scikit-learn

Export baseline features (length, entropy, label count, digit ratio), fit a one‑class model, then score new queries.


Operational tips

  • Automate:

    • Convert each case study into a systemd service + timer, or a cron job, that writes CSV/JSON to /var/log/net-ml/.
  • Observe:

    • Ship outputs to a time‑series DB or flat files and visualize with Grafana.
  • Guardrails:

    • Always add a manual approval step before pushing automated mitigations (e.g., ACLs, tc changes) to production.
  • Version and document:

    • Store scripts in a repo with a README explaining data sources, thresholds, and rollback.

Conclusion and next steps

AI for networking doesn’t have to be a black box. With Bash, TShark, tc, and a modest Python stack, you can detect anomalies, tune QoS, and surface stealthy behavior—today.

Your next move:

  • Run Case Study 1 on a recent pcap or a lab SPAN port.

  • Put Case Study 2 on a test host to auto‑tune a class under load.

  • Point Case Study 3 at your resolver mirror or DNS taps and validate the signal.

When you’re ready, generalize these into pipelines:

  • Feature extractors (Bash/TShark) → Feature store (CSV/Parquet) → Models (Python) → Actions (tc/ACLs/alerts).

  • Add CI tests with synthetic pcaps so you trust the results.

If you found this useful, subscribe to the blog or star the repo where you keep these scripts. Your network is already speaking—let’s make it think.