Posted on
• Artificial Intelligence

Artificial Intelligence Security Automation Projects

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

Bash-first AI: 4 Security Automation Projects You Can Ship This Week

If your terminal feels like a firehose—alerts, logs, pcap, repeat—you’re not alone. The volume and velocity of security data outpaces what any analyst can triage by hand. The good news: a bit of Bash glue and lightweight machine learning can turn raw noise into actionable signal.

In this article, you’ll build four small, composable AI-driven automations that run happily on a Linux box. Each one is practical, auditable, and deployable with your existing tooling. You’ll get copy-pasteable scripts, clear install commands, and tips to productionize with cron or systemd.

What you’ll get:

  • Why AI belongs in your Bash toolbox

  • 4 real projects: auth-log anomaly detection, Suricata alert triage, CLI phishing URL detection, and process anomaly hunting

  • Universal install snippets for apt, dnf, and zypper

  • Minimal Python used as a filter; Bash remains the orchestrator

Why AI/ML belongs in your Linux security toolbox

  • Scale and prioritization: ML can surface the weird from thousands of “normal” events so humans spend time where it matters.

  • Bash-friendly: You don’t need a data lake. Stream logs through jq, pipe to a Python filter, and alert when it spikes.

  • Incremental wins: Start with simple models (Isolation Forest, Logistic Regression). Add labels and feedback later to improve.

  • Reproducible and debuggable: Everything is transparent text pipelines and small scripts. Easy to review, diff, and version-control.

Prerequisites and setup

These projects rely on standard CLI tools plus a Python virtual environment.

Install base packages:

# Debian/Ubuntu
sudo apt update
sudo apt install -y python3 python3-venv python3-pip jq git suricata tshark

# Fedora/RHEL/CentOS
sudo dnf install -y python3 python3-virtualenv python3-pip jq git suricata wireshark-cli

# openSUSE/SLES
sudo zypper refresh
sudo zypper install -y python3 python3-venv python3-pip jq git suricata wireshark-cli

Create and activate a Python virtual environment, then install ML libraries:

python3 -m venv ~/secai-venv
source ~/secai-venv/bin/activate
pip install --upgrade pip
pip install scikit-learn pandas numpy joblib ujson tldextract

Optional (for Project 3 blocking via ipset):

# Debian/Ubuntu
sudo apt install -y ipset

# Fedora/RHEL/CentOS
sudo dnf install -y ipset

# openSUSE/SLES
sudo zypper install -y ipset

Note: Some logs require root to read (e.g., /var/log/secure, /var/log/auth.log) and Suricata may need elevated privileges. Use sudo where indicated.

Project 1: Anomaly detection on SSH/auth logs

Goal: Identify unusual authentication activity (new source IPs, odd times, rare users) using an unsupervised model. Output only the odd stuff.

1) Create the Python scorer (reads from stdin, trains if no model yet):

cat > score_authlog.py << 'PY'
#!/usr/bin/env python3
import sys, re, os, json, time
from datetime import datetime
import numpy as np
from sklearn.ensemble import IsolationForest
from joblib import dump, load

MODEL_PATH = os.environ.get("AUTH_MODEL", "models/iforest_auth.joblib")
os.makedirs("models", exist_ok=True)

log_re = re.compile(r'(?P<month>\w{3})\s+(?P<day>\d{1,2})\s(?P<time>\d{2}:\d{2}:\d{2}).*?(?:(Failed|Accepted)\s\w+)\sfor\s(?:(?:invalid\suser\s)?)(?P<user>[\w\-\.\$]+)\sfrom\s(?P<ip>\d{1,3}(?:\.\d{1,3}){3})')

month_map = {m:i+1 for i,m in enumerate(["Jan","Feb","Mar","Apr","May","Jun","Jul","Aug","Sep","Oct","Nov","Dec"])}

def featurize(line):
    m = log_re.search(line)
    if not m:
        return None, None
    month = month_map.get(m.group("month"), 1)
    day = int(m.group("day"))
    hh,mm,ss = map(int, m.group("time").split(":"))
    tod = hh*3600 + mm*60 + ss
    user = m.group("user")
    ip = m.group("ip")
    octs = [int(x) for x in ip.split(".")]
    # Simple features: time of day, day, month, IP octets, user hash bucket
    user_bucket = (hash(user) % 97) / 97.0
    x = [tod/86400.0, day/31.0, month/12.0] + [o/255.0 for o in octs] + [user_bucket]
    return np.array(x, dtype=float), {"user":user, "ip":ip, "time":f"{month:02d}-{day:02d} {hh:02d}:{mm:02d}:{ss:02d}"}

# Load or init model
if os.path.exists(MODEL_PATH):
    model = load(MODEL_PATH)
else:
    model = IsolationForest(n_estimators=150, contamination=0.01, random_state=42, warm_start=True)
    # "Prime" with an empty fit to set structure on first batch
    model.fit(np.random.rand(64,7))

buf = []
meta_buf = []
BATCH = 128
THRESH = -0.15  # lower = more anomalous

def process_batch():
    global buf, meta_buf, model
    if not buf: return
    X = np.vstack(buf)
    # Partial refit by reusing estimators (cheap online-ish approach)
    model.set_params(n_estimators=model.n_estimators + 1, warm_start=True)
    model.fit(X)
    dump(model, MODEL_PATH)
    scores = model.decision_function(X)
    for s, meta in zip(scores, meta_buf):
        if s < THRESH:
            out = {"score": float(s), **meta}
            print(json.dumps(out), flush=True)
    buf.clear(); meta_buf.clear()

for line in sys.stdin:
    f, meta = featurize(line)
    if f is None: 
        continue
    buf.append(f); meta_buf.append(meta)
    if len(buf) >= BATCH:
        process_batch()

process_batch()
PY
chmod +x score_authlog.py

2) Create a Bash runner that tails the right auth log and sends anomalies to syslog:

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

LOG_FILE=""
for c in /var/log/auth.log /var/log/secure /var/log/messages; do
  [[ -r "$c" ]] && LOG_FILE="$c" && break
done
[[ -z "$LOG_FILE" ]] && { echo "No readable auth log found"; exit 1; }

# Requires sudo for most systems
sudo tail -F "$LOG_FILE" | ./score_authlog.py | while read -r json; do
  user=$(echo "$json" | jq -r '.user')
  ip=$(echo "$json" | jq -r '.ip')
  score=$(echo "$json" | jq -r '.score')
  ts=$(echo "$json" | jq -r '.time')
  logger -t auth-anom "Anomalous auth: user=$user ip=$ip score=$score ts=$ts"
  echo "$json"
done
SH
chmod +x log_anomaly.sh

3) Run it:

source ~/secai-venv/bin/activate
./log_anomaly.sh

Automation tip (cron):

crontab -e
# Add (adjust path):
@reboot cd /path/to/your/scripts && . ~/secai-venv/bin/activate && nohup ./log_anomaly.sh >/var/log/auth_anom.log 2>&1 &

Project 2: AI-assisted Suricata alert dedup and prioritization

Goal: Suricata can be chatty. We’ll stream EVE JSON, cluster “similar” alerts online, and raise priority for rare/new patterns.

1) Ensure Suricata writes EVE JSON:

  • Default path: /var/log/suricata/eve.json

  • If missing, enable EVE in /etc/suricata/suricata.yaml, then restart:

sudo suricata -T -c /etc/suricata/suricata.yaml
sudo systemctl restart suricata || sudo suricata -c /etc/suricata/suricata.yaml -D

2) Create the triage script:

cat > suri_triage.py << 'PY'
#!/usr/bin/env python3
import sys, json, math, hashlib, time
from collections import defaultdict, deque
from sklearn.cluster import MiniBatchKMeans
import numpy as np

# Online-ish clustering of signatures+addrs to reduce dupes
kmeans = MiniBatchKMeans(n_clusters=12, random_state=42, batch_size=64)
primed = False
buf = []
meta = []
WINDOW = deque(maxlen=2000)  # track recent hashes to detect novelty

def vec(sig, src, dst, sev):
    # Feature: sig hash bucket + IP buckets + severity
    sh = (int(hashlib.md5(sig.encode()).hexdigest(), 16) % 1000) / 1000.0
    try:
        so = [int(x) for x in src.split(".")]
        do = [int(x) for x in dst.split(".")]
    except:
        so = [0,0,0,0]; do=[0,0,0,0]
    v = [sh] + [s/255.0 for s in so] + [d/255.0 for d in do] + [sev/7.0]
    return np.array(v, dtype=float)

def novelty(h):
    # Rarer hashes -> higher novelty
    count = sum(1 for x in WINDOW if x==h)
    return 1.0 / (1+count)

def priority(sev, nvt, cluster_size):
    # Combine vendor severity, novelty, and inverse cluster size
    base = (8 - sev) / 7.0  # sev 1 (high) -> 1.0, sev 7 (low) -> ~0.14
    density = 1.0 / (1 + cluster_size)
    return 0.5*base + 0.35*nvt + 0.15*density

clusters = defaultdict(int)

def process_batch():
    global primed, clusters
    if not buf: return
    X = np.vstack(buf)
    if not primed:
        # Seed with random points to stabilize centers
        kmeans.partial_fit(np.random.rand(64, X.shape[1]))
        primed = True
    kmeans.partial_fit(X)
    labels = kmeans.predict(X)
    # Update cluster sizes
    for lb in labels:
        clusters[lb] += 1
    # Output prioritized alerts
    for (v, m, lb) in zip(buf, meta, labels):
        sev = m["sev"]
        h = m["hash"]
        p = priority(sev, novelty(h), clusters[lb])
        out = dict(priority=round(float(p),3), cluster=int(lb), **m)
        print(json.dumps(out), flush=True)
    buf.clear(); meta.clear()

for line in sys.stdin:
    try:
        ev = json.loads(line)
    except:
        continue
    sig = ev.get("sig") or ev.get("signature") or ""
    src = ev.get("src") or ev.get("src_ip") or ""
    dst = ev.get("dst") or ev.get("dest_ip") or ev.get("dst_ip") or ""
    sev = ev.get("sev") or ev.get("severity") or 3
    try:
        sev = int(sev)
    except:
        sev = 3
    v = vec(sig, src, dst, sev)
    h = hashlib.sha1(f"{sig}|{src}|{dst}".encode()).hexdigest()[:12]
    WINDOW.append(h)
    meta.append({"sig":sig, "src":src, "dst":dst, "sev":sev, "hash":h, "ts":ev.get("ts") or ev.get("timestamp")})
    buf.append(v)
    if len(buf) >= 64:
        process_batch()
process_batch()
PY
chmod +x suri_triage.py

3) Stream Suricata alerts with jq into the triage script:

sudo tail -F /var/log/suricata/eve.json \
 | jq -rc 'select(.event_type=="alert") | {ts:.timestamp,src:.src_ip,dst:.dest_ip,sev:.alert.severity,sig:.alert.signature}' \
 | ./suri_triage.py \
 | tee -a suri_triage.out

4) Example filter: only page high-priority items

cat suri_triage.out | jq -rc 'select(.priority>0.75)' \
 | while read -r j; do
     sig=$(echo "$j" | jq -r .sig)
     src=$(echo "$j" | jq -r .src)
     dst=$(echo "$j" | jq -r .dst)
     logger -t suri-ai "HIGH PRI: $sig [$src -> $dst]"
   done

Automation tip: run under systemd with Restart=always; bind jq and Python via an ExecStart sh -c pipeline.

Project 3: CLI phishing URL detector for logs, mail, and proxy feeds

Goal: Extract URLs from any text stream and classify likely phishing based on lexical features (length, digits, entropy, punycode, suspicious words). Output a blocklist or alerts.

1) URL classifier:

cat > url_classify.py << 'PY'
#!/usr/bin/env python3
import sys, re, math, json, os
import numpy as np
from sklearn.linear_model import LogisticRegression
from joblib import dump, load
import tldextract

MODEL_PATH = os.environ.get("URL_MODEL", "models/url_lr.joblib")
os.makedirs("models", exist_ok=True)

sus_words = ["login","verify","update","secure","account","signin","wallet","invoice","pay","support"]

def shannon_entropy(s):
    if not s: return 0.0
    from collections import Counter
    c = Counter(s)
    n = len(s)
    return -sum((cnt/n)*math.log2(cnt/n) for cnt in c.values())

def features(url):
    ex = tldextract.extract(url)
    domain = ".".join([ex.domain, ex.suffix]) if ex.suffix else ex.domain
    sub = ex.subdomain or ""
    host = ".".join([p for p in [sub, ex.domain, ex.suffix] if p])
    path = re.sub(r'^https?://[^/]+', '', url, flags=re.I)
    u_ascii = url.encode('idna', 'ignore').decode('ascii', 'ignore')
    feats = {}
    feats["len"] = len(url)
    feats["digits"] = sum(ch.isdigit() for ch in url)
    feats["dots"] = url.count(".")
    feats["hyphens"] = url.count("-")
    feats["entropy"] = shannon_entropy(u_ascii.lower())
    feats["puny"] = 1 if "xn--" in u_ascii.lower() else 0
    feats["path_len"] = len(path)
    feats["sub_len"] = len(sub)
    feats["sus_words"] = sum(1 for w in sus_words if w in url.lower())
    return np.array([feats[k] for k in ["len","digits","dots","hyphens","entropy","puny","path_len","sub_len","sus_words"]], dtype=float), domain

# Load or bootstrap a model
if os.path.exists(MODEL_PATH):
    clf = load(MODEL_PATH)
else:
    # start with a weak prior: short benign vs long suspicious
    X = np.array([[20,1,1,0,2.5,0,1,0,0],
                  [120,10,4,3,4.0,0,60,10,2],
                  [200,20,6,4,4.5,1,120,20,3],
                  [30,0,1,0,2.0,0,5,0,0]])
    y = np.array([0,1,1,0])  # 1=phishy
    clf = LogisticRegression(max_iter=200)
    clf.fit(X, y)
    dump(clf, MODEL_PATH)

for line in sys.stdin:
    url = line.strip()
    if not url:
        continue
    x, domain = features(url)
    p = float(clf.predict_proba([x])[0][1])
    label = "phish" if p >= 0.7 else ("suspect" if p >= 0.5 else "ok")
    print(json.dumps({"url":url, "domain":domain, "score":round(p,3), "label":label}))
PY
chmod +x url_classify.py

2) Bash wrapper to extract URLs from any file/stream and classify:

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

# Extract URLs with PCRE; requires grep -P (GNU grep)
extract_urls() {
  grep -oP '(?i)\bhttps?://[^\s\"\'\<\>]+'
}

# Usage: urls_scan.sh < file_with_text
extract_urls | ./url_classify.py
SH
chmod +x urls_scan.sh

3) Example: scan mail logs or a raw message

# Mail log example
sudo journalctl -u postfix -n 1000 | ./urls_scan.sh | jq -rc 'select(.label!="ok")'

# Raw file example
cat message.eml | ./urls_scan.sh | tee suspicious_urls.json

Optional: generate an ipset blocklist from high-confidence hits (score >= 0.85):

# Prepare the set once
sudo ipset create phish_domains hash:ip -exist || true

# Resolve and add
jq -rc 'select(.score>=0.85) | .domain' suspicious_urls.json | sort -u | while read -r d; do
  ip=$(getent ahostsv4 "$d" | awk '{print $1}' | head -n1 || true)
  [[ -n "${ip:-}" ]] && sudo ipset add phish_domains "$ip" -exist
done

Note: For proxies (e.g., Squid) you can instead generate a domain ACL file from the same JSON and deny by domain without IP resolution.

Project 4: Hunt anomalous processes from ps/ss data

Goal: Periodically model “normal” process behavior on the host and flag outliers by CPU/mem use and network fan-out.

1) Create the hunter:

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

# Snapshot features: pid, user, cpu, mem, start_time, listen_count, conn_count
snapshot() {
  # ps for CPU/MEM; ss for sockets
  ss_out=$(ss -tunap 2>/dev/null | awk 'NR>1 {print $0}')
  # Count connections per PID
  declare -A CONN LISTEN
  while IFS= read -r l; do
    pid=$(echo "$l" | grep -oP 'pid=\K[0-9]+' || true)
    [[ -z "$pid" ]] && continue
    if echo "$l" | grep -q LISTEN; then
      (( LISTEN[$pid]++ )) || true
    else
      (( CONN[$pid]++ )) || true
    fi
  done <<< "$ss_out"

  # Output CSV header + rows
  echo "pid,user,cpu,mem,rss_kb,start,listen,conns,cmd"
  ps -eo pid,user,%cpu,%mem,rss,lstart,cmd --no-headers | while read -r pid user cpu mem rss w1 w2 w3 w4 w5 w6 cmd_rest; do
    start="$w1 $w2 $w3 $w4 $w5 $w6"
    listen=${LISTEN[$pid]:-0}
    conns=${CONN[$pid]:-0}
    # Limit cmd length for CSV sanity
    cmd=$(echo "$cmd_rest" | cut -c1-120)
    echo "$pid,$user,$cpu,$mem,$rss,$start,$listen,$conns,\"$cmd\""
  done
}

snapshot
SH
chmod +x proc_hunt.sh

2) Python anomaly scorer for process snapshots:

cat > proc_score.py << 'PY'
#!/usr/bin/env python3
import sys, os, csv, json, time
import numpy as np
from sklearn.ensemble import IsolationForest
from joblib import dump, load

MODEL_PATH = os.environ.get("PROC_MODEL", "models/iforest_proc.joblib")
os.makedirs("models", exist_ok=True)

def to_vec(row):
    cpu = float(row['cpu'])
    mem = float(row['mem'])
    rss = float(row['rss_kb'])/1024.0
    listen = float(row['listen'])
    conns = float(row['conns'])
    user_bucket = (hash(row['user']) % 127)/127.0
    return np.array([cpu/100.0, mem/100.0, rss/8192.0, listen/64.0, conns/256.0, user_bucket], dtype=float)

if os.path.exists(MODEL_PATH):
    model = load(MODEL_PATH)
else:
    model = IsolationForest(n_estimators=200, contamination=0.02, random_state=7)
    model.fit(np.random.rand(128,6))
    dump(model, MODEL_PATH)

rows = list(csv.DictReader(sys.stdin))
if not rows:
    sys.exit(0)

X = np.vstack([to_vec(r) for r in rows])
model.fit(X)
dump(model, MODEL_PATH)
scores = model.decision_function(X)

for r, s in zip(rows, scores):
    if s < -0.1:
        out = {
          "score": float(s),
          "pid": r['pid'], "user": r['user'],
          "cpu": r['cpu'], "mem": r['mem'],
          "listen": r['listen'], "conns": r['conns'],
          "cmd": r['cmd']
        }
        print(json.dumps(out))
PY
chmod +x proc_score.py

3) Run it and alert:

./proc_hunt.sh | ./proc_score.py | tee -a proc_anom.json | while read -r j; do
  pid=$(echo "$j" | jq -r .pid)
  cmd=$(echo "$j" | jq -r .cmd)
  s=$(echo "$j" | jq -r .score)
  logger -t proc-anom "Suspicious proc pid=$pid score=$s cmd=$cmd"
done

Automation tip (cron every 5 minutes):

crontab -e
# */5 * * * * cd /path && . ~/secai-venv/bin/activate && ./proc_hunt.sh | ./proc_score.py >> /var/log/proc_anom.json 2>&1

Operational notes and hardening tips

  • Baseline first: Let each model observe typical activity before trusting alerts. Start by logging only; add blocking later.

  • Version your models: Commit models/ directory or keep a simple “model registry” with timestamps. Roll back when needed.

  • Tune thresholds: decision_function thresholds vary by environment. Begin lenient (fewer alerts), tighten gradually.

  • Least privilege: Only grant read access to the specific logs/files required. Avoid running long-lived pipelines as root unless necessary.

  • Observability: Always tee raw outputs to a file and send summaries to syslog for correlation.

Conclusion and next steps

You don’t need a massive platform to get real value from AI in security. With Bash, jq, and a handful of Python models, you can:

  • Spot unusual logins as they happen

  • Cut Suricata noise into prioritized, deduplicated signals

  • Flag phishing URLs from any text stream

  • Hunt odd processes without an agent

Your next step:

  • Pick one project above, paste the scripts into a repo, and run it in “log-only” mode for a week.

  • Review the outputs, then adjust thresholds and add your first blocking or paging action.

  • Iterate: add domain allowlists, user risk scores, and feedback loops to improve models over time.

If you found this useful, share it with your team and start a small “AI + Bash” backlog. Small, composable wins add up fast.

Related posts