Posted on
• Artificial Intelligence

Artificial Intelligence Threat Detection on Linux

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

Artificial Intelligence Threat Detection on Linux: From Logs to Live Alerts

If an attacker landed on one of your Linux hosts right now, how long would it take you to notice? Minutes matter. Traditional rule-only detection (signatures, blocklists) can miss new tactics, while sifting through endless logs by hand is unrealistic. The good news: with the telemetry Linux already gives you and a small dose of machine learning, you can surface suspicious behavior quickly—often in minutes.

This guide shows you a practical, Bash-friendly path to AI-driven threat detection on Linux. You’ll instrument your hosts, normalize logs, train a lightweight anomaly detector, and wire it to alert automatically. No heavy SIEM required to get started.

Why AI on Linux, and why now?

  • Attackers adapt quickly: Rules lag behind new tools and “living off the land” techniques.

  • Linux is noisy (in a good way): Syscalls, process execs, and network flows contain strong behavioral signals that anomaly detection can leverage.

  • You can start small: Combine a few core sensors (auditd, Suricata) with a local Python model to flag outliers without full-blown data pipelines.

The goal is not replacing rules. It’s augmenting them—using AI to bubble up the “weird stuff” you’d otherwise miss or review too late.

What you’ll build

  • Host sensors: auditd for process/syscall activity; Suricata for network activity.

  • Normalized, queryable data: JSON-friendly outputs you can grep, jq, and pipe.

  • An anomaly detector: A small Python script using Isolation Forest to flag unusual processes or network events.

  • Automation: A systemd timer to run detection periodically and write alerts you can review or forward.

1) Instrument your Linux hosts (process + network)

Start by enabling two low-friction sensors: auditd (host activity) and Suricata (network IDS). You’ll need sudo/root.

Install auditd:

  • apt (Debian/Ubuntu):
sudo apt update
sudo apt install -y auditd
  • dnf (Fedora/RHEL/CentOS Stream):
sudo dnf install -y audit
  • zypper (openSUSE/SLE):
sudo zypper refresh
sudo zypper install -y audit

Enable and start auditd:

sudo systemctl enable --now auditd
sudo systemctl status auditd

Add a focused set of audit rules to watch execs, sensitive file reads, and kernel module operations. Save as /etc/audit/rules.d/ai-threat.rules:

# Track all execve/execveat calls (can be noisy; tune in prod)
-a always,exit -F arch=b64 -S execve,execveat -k exec_monitor
-a always,exit -F arch=b32 -S execve,execveat -k exec_monitor

# Watch reads of /etc/shadow (credential file)
-w /etc/shadow -p r -k shadow_read

# Kernel module load/unload
-a always,exit -F arch=b64 -S init_module,finit_module,delete_module -k kmod
-a always,exit -F arch=b32 -S init_module,finit_module,delete_module -k kmod

Load rules:

sudo augenrules --load
sudo systemctl restart auditd

Install Suricata for network telemetry:

  • apt:
sudo apt update
sudo apt install -y suricata jq
  • dnf:
sudo dnf install -y suricata jq
  • zypper: sudo zypper refresh sudo zypper install -y suricata jq

Quick-start Suricata in AF_PACKET mode (lab-friendly). Replace eth0 with your interface:

sudo sed -i 's/interface: .*/interface: eth0/' /etc/suricata/suricata.yaml
sudo systemctl enable --now suricata
sudo systemctl status suricata

If you prefer not to change config yet, you can run it ad-hoc:

sudo suricata -i eth0 -D -c /etc/suricata/suricata.yaml

Notes:

  • auditd logs: /var/log/audit/audit.log

  • Suricata JSON: /var/log/suricata/eve.json

  • jq helps you inspect and filter JSON quickly.

2) Normalize and preview your data

Before modeling, confirm you’re getting the signals you expect.

Tail audit logs (process/syscall activity):

sudo tail -f /var/log/audit/audit.log | sed -E 's/type=([A-Z_]+)/\1/g' | head -n 20

Preview Suricata events:

sudo tail -f /var/log/suricata/eve.json | jq '.'

Journald to JSON (handy for system/service metadata):

journalctl -n 100 -o json | jq '.[["_HOSTNAME","MESSAGE","PRIORITY","SYSLOG_IDENTIFIER"]]'

You should see EXECVE/SYSCALL records for commands you run, and Suricata should log flows/DNS/alerts if the host has traffic.

3) Add a small anomaly detector

We’ll use Python’s scikit-learn IsolationForest to score uncommon process and network behaviors. This is a lightweight, local “bubble up the weird” model that complements your rules.

Install Python and pip:

  • apt:
sudo apt update
sudo apt install -y python3 python3-pip
  • dnf:
sudo dnf install -y python3 python3-pip
  • zypper:
sudo zypper refresh
sudo zypper install -y python3 python3-pip

Install Python packages (user-scoped):

pip3 install --user pandas scikit-learn ujson

Create the detector at /usr/local/bin/ai_linux_detector.py:

#!/usr/bin/env python3
import os, re, json, time, math, sys
from collections import deque
from datetime import datetime
import pandas as pd
from sklearn.ensemble import IsolationForest

AUDIT_LOG = "/var/log/audit/audit.log"
SURICATA_EVE = "/var/log/suricata/eve.json"

def tail_lines(path, max_lines=5000):
    if not os.path.exists(path):
        return []
    dq = deque(maxlen=max_lines)
    with open(path, "r", errors="ignore") as f:
        for line in f:
            dq.append(line.rstrip("\n"))
    return list(dq)

# Extract key=value pairs from audit SYSCALL lines
KV_RE = re.compile(r'(\w+)=(".*?"|\S+)')
TS_RE = re.compile(r'audit\((\d+)(?:\.\d+)?')

def parse_audit_syscalls(lines):
    rows = []
    for ln in lines:
        if "SYSCALL" not in ln:
            continue
        ts_match = TS_RE.search(ln)
        ts = int(ts_match.group(1)) if ts_match else int(time.time())
        kv = dict((k, v.strip('"')) for k, v in KV_RE.findall(ln))
        # We care about execs; syscall=59 (execve) on x86_64 is common
        sc = kv.get("syscall", "")
        exe = kv.get("exe", "")
        comm = kv.get("comm", "")
        uid = int(kv.get("uid", "-1"))
        auid = int(kv.get("auid", "-1"))
        pid = int(kv.get("pid", "-1"))
        ppid = int(kv.get("ppid", "-1")) if "ppid" in kv else -1
        key = kv.get("key", "")
        success = 1 if kv.get("success") == "yes" else 0

        # Featureize
        hour = datetime.fromtimestamp(ts).hour
        path_depth = exe.count("/") if exe else 0
        exe_len = len(exe)
        comm_len = len(comm)
        is_exec_syscall = 1 if (sc in ("59", "322") or key == "exec_monitor") else 0  # 322=execveat on some arches

        rows.append({
            "ts": ts, "hour": hour, "uid": uid, "auid": auid,
            "pid": pid, "ppid": ppid, "success": success,
            "exe": exe, "comm": comm, "exe_len": exe_len,
            "comm_len": comm_len, "path_depth": path_depth,
            "is_exec": is_exec_syscall, "key": key
        })
    return pd.DataFrame(rows) if rows else pd.DataFrame()

def parse_suricata(lines):
    rows = []
    for ln in lines:
        try:
            ev = json.loads(ln)
        except Exception:
            continue
        # Feature subset for anomaly scoring
        ts = ev.get("timestamp")
        try:
            epoch = int(datetime.fromisoformat(ts.replace("Z","+00:00")).timestamp())
        except Exception:
            epoch = int(time.time())
        et = ev.get("event_type", "unknown")
        proto = ev.get("proto") if "proto" in ev else ev.get("app_proto", "na")
        dst_p = ev.get("dest_port", 0)
        src_p = ev.get("src_port", 0)
        b_in = ev.get("in_iface", "")
        sev = ev.get("alert", {}).get("severity", 0) if "alert" in ev else 0
        rows.append({
            "ts": epoch, "hour": datetime.fromtimestamp(epoch).hour,
            "etype": et, "proto": str(proto), "dport": int(dst_p or 0),
            "sport": int(src_p or 0), "sev": int(sev)
        })
    return pd.DataFrame(rows) if rows else pd.DataFrame()

def fit_score(df, num_cols, extra_info_cols, contamination=0.01):
    if df.empty:
        return df
    X = df[num_cols].fillna(0)
    model = IsolationForest(n_estimators=200, contamination=contamination, random_state=42)
    model.fit(X)
    scores = -model.score_samples(X)  # higher = more anomalous
    out = df.copy()
    out["anomaly_score"] = scores
    out = out.sort_values("anomaly_score", ascending=False)
    return out[extra_info_cols + num_cols + ["anomaly_score"]]

def main(out_path=None, top_n=25, contam=0.01):
    audit_df = parse_audit_syscalls(tail_lines(AUDIT_LOG, 8000))
    suri_df  = parse_suricata(tail_lines(SURICATA_EVE, 8000))

    findings = []

    if not audit_df.empty:
        audit_num = ["hour","uid","auid","pid","ppid","success","exe_len","comm_len","path_depth","is_exec"]
        audit_show = ["ts","exe","comm","key"]
        scored = fit_score(audit_df, audit_num, audit_show, contamination=contam).head(top_n)
        for _, r in scored.iterrows():
            findings.append({
                "source":"auditd",
                "ts": int(r["ts"]),
                "exe": r["exe"],
                "comm": r["comm"],
                "key": r["key"],
                "score": round(float(r["anomaly_score"]), 4),
                "explanation": "Unusual process/syscall pattern vs host baseline"
            })

    if not suri_df.empty:
        # Encode categories minimally
        suri_df["etype_code"] = suri_df["etype"].astype("category").cat.codes
        suri_df["proto_code"] = suri_df["proto"].astype("category").cat.codes
        suri_num = ["hour","dport","sport","sev","etype_code","proto_code"]
        suri_show = ["ts"]
        scored = fit_score(suri_df, suri_num, suri_show, contamination=contam).head(top_n)
        for _, r in scored.iterrows():
            findings.append({
                "source":"suricata",
                "ts": int(r["ts"]),
                "dport": int(r.get("dport",0)),
                "sport": int(r.get("sport",0)),
                "score": round(float(r["anomaly_score"]), 4),
                "explanation": "Unusual network event vs host baseline"
            })

    findings = sorted(findings, key=lambda x: x["score"], reverse=True)[:top_n]

    if out_path:
        os.makedirs(os.path.dirname(out_path), exist_ok=True)
        with open(out_path, "a") as f:
            for fnd in findings:
                f.write(json.dumps(fnd) + "\n")

    # Also print to stdout for quick runs
    for fnd in findings:
        print(json.dumps(fnd))

if __name__ == "__main__":
    out = None
    if "--output" in sys.argv:
        out = sys.argv[sys.argv.index("--output")+1]
    main(out_path=out, top_n=25, contam=0.01)

Make it executable:

sudo chmod +x /usr/local/bin/ai_linux_detector.py

Test-run:

sudo /usr/local/bin/ai_linux_detector.py --output /var/log/ai-detector/alerts.json | jq '.'

You should see top anomalous events (process or network) with scores.

Tip: Run it daily at first to build a baseline as normal activity accumulates.

4) Automate and alert with systemd

Create a service that writes alerts to a file you can tail or forward.

Service unit /etc/systemd/system/ai-detector.service:

[Unit]
Description=AI Linux Threat Detector (Anomaly Scoring)

[Service]
Type=oneshot
ExecStart=/usr/bin/env python3 /usr/local/bin/ai_linux_detector.py --output /var/log/ai-detector/alerts.json
User=root
Nice=10

Timer unit /etc/systemd/system/ai-detector.timer:

[Unit]
Description=Run AI Linux Threat Detector every 5 minutes

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

[Install]
WantedBy=timers.target

Enable and start:

sudo mkdir -p /var/log/ai-detector
sudo systemctl daemon-reload
sudo systemctl enable --now ai-detector.timer
sudo systemctl list-timers | grep ai-detector

Tail alerts:

sudo tail -f /var/log/ai-detector/alerts.json | jq '.'

Forwarding options:

  • Ship /var/log/ai-detector/alerts.json to your SIEM or log aggregation.

  • Add a simple mailer or webhook in the script for high-score events.

5) Validate with safe simulations

Run a few benign-but-suspicious actions to see detections. Expect higher anomaly scores immediately after action in a “clean” baseline.

  • Unusual read of a sensitive file (root only):
sudo bash -c 'head -n1 /etc/shadow >/dev/null'
  • Execution from uncommon path:
mkdir -p /tmp/testbin
cp /bin/sleep /tmp/testbin/
TMPDIR=/tmp/testbin /tmp/testbin/sleep 1
  • Odd network port connection:
nc -vz example.com 4444 || true
  • Burst of base64 decodes (often seen in obfuscated scripts):
for i in $(seq 1 50); do echo "SGVsbG8=" | base64 -d >/dev/null; done

Re-run the detector or wait for the timer cycle:

sudo /usr/local/bin/ai_linux_detector.py --output /var/log/ai-detector/alerts.json | jq '.'

Real-world patterns you’ll start catching

  • Crypto-miner dropper from /tmp: Rare exec path depth, unusual parent, sudden outbound to mining pool ports (e.g., 3333/tcp).

  • Living-off-the-land exfil via curl/wget: Spike in execs of network tools outside maintenance windows; rare destination ports/domains in Suricata EVE.

  • Kernel tampering attempts: init_module/delete_module syscalls lighting up auditd rules you set.

  • Unexpected privilege pivots: Rare uid/auid combinations, elevated execs by service accounts at odd hours.

Tuning tips

  • Start with contamination=0.01 and adjust. Higher means more alerts.

  • Split baselines by role (web, DB, CI runner) to reduce “unusual but normal” noise.

  • Add features: parent process, TTY presence, session ID, command-line length, nighttime weights.

  • Keep rules: AI is a complement, not a replacement. Your audit rules and Suricata signatures still catch known-bad fast.

  • Store a sliding window: Keep 1–7 days of recent events for the model so it adapts, but doesn’t forget long-term norms too quickly.

Conclusion and next steps

You’ve just built an end-to-end AI-assisted detection loop using native Linux telemetry:

  • auditd and Suricata supply rich signals

  • a compact ML model flags outliers

  • systemd automates the cycle and produces actionable alerts

Next steps: 1) Roll out to a test group of hosts and tune contamination/features per role.
2) Add alert routing (email, Slack/Webhook) for high-scoring events.
3) Enrich features: parse command-lines, parent-child chains, Suricata DNS/TLS SNI.
4) Combine with your SIEM: Stream alerts and raw signals for longer retention and investigations.

Security evolves. With a few packages and a small script, your Linux fleet can, too.

Appendix: Package installation summary (copy/paste)

  • auditd/audit:

    • apt: sudo apt install -y auditd
    • dnf: sudo dnf install -y audit
    • zypper: sudo zypper install -y audit

  • Suricata:

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

  • jq:

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

  • Python + pip:

    • apt: sudo apt install -y python3 python3-pip
    • dnf: sudo dnf install -y python3 python3-pip
    • zypper: sudo zypper install -y python3 python3-pip

Happy hunting—and may your anomalies be few and obvious.

Related posts