Posted on
• Artificial Intelligence

Artificial Intelligence Endpoint Security

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

AI for Linux Endpoint Security: From Logs to Real-Time Defense

Your Linux servers are already telling you when something’s wrong—the problem is separating the signal from the noise before an attacker blends in. Artificial intelligence can help you do exactly that on the endpoint you control most: the Linux host. In this guide, you’ll wire up high-fidelity telemetry, train a lightweight anomaly model, and score events in real time—with installation steps for apt, dnf, and zypper along the way.

What you’ll get:

  • A practical, host-first approach to AI-driven detection

  • Minimal dependencies, all open source

  • Copy-pasteable commands and configs you can try today

Why AI belongs on the endpoint

  • Volume and variety: Endpoint activity (processes, syscalls, file changes) is rich but noisy. Unsupervised ML excels at surfacing outliers without handcrafting every rule.

  • Faster detection: Anomaly scores can flag never-before-seen behaviors (e.g., “curl | bash” under an unusual service account) in seconds.

  • Privacy and simplicity: Models run locally on your server—no need to ship sensitive logs to the cloud.

This is not “replace your rules.” It’s “stack the deck”: keep rule-based detection where it’s strong, and add AI to catch the weird stuff.

Step 1: Turn on the right telemetry with auditd

auditd gives you kernel-level visibility into process execution, file writes, and module loads—exactly the signals an AI model can learn from.

Install auditd:

  • Debian/Ubuntu (apt):

    sudo apt update
sudo apt install -y auditd audispd-plugins

  • Fedora/RHEL/CentOS (dnf):

    sudo dnf install -y audit

  • openSUSE/SLE (zypper):

    sudo zypper refresh
sudo zypper install -y audit

Enable and start:

sudo systemctl enable --now auditd

Add a lean, high-signal ruleset (works on most 64-bit hosts; add b32 if you run 32-bit binaries too):

sudo bash -c 'cat >/etc/audit/rules.d/ai-endpoint.rules' << "EOF"
# Log process execution
-a always,exit -F arch=b64 -S execve,execveat -k exec
-a always,exit -F arch=b32 -S execve,execveat -k exec

# Watch sensitive directories for writes
-w /etc -p wa -k etc_changes
-w /usr/local/bin -p wa -k bin_changes
-w /root -p wa -k root_changes

# 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

# Outbound network connects (may be chatty; adjust later)
-a always,exit -F arch=b64 -S connect -k net
-a always,exit -F arch=b32 -S connect -k net
EOF

sudo augenrules --load
sudo systemctl restart auditd

Tips:

  • Start minimal to manage noise, then iterate.

  • Keep /var/log/audit/audit.log on fast local storage.

Optional utilities for later:

  • Debian/Ubuntu:

    sudo apt install -y jq lsof python3 python3-pip

  • Fedora/RHEL/CentOS:

    sudo dnf install -y jq lsof python3 python3-pip

  • openSUSE/SLE:

    sudo zypper install -y jq lsof python3 python3-pip

Step 2: Train a quick anomaly model from your host’s history

We’ll use an Isolation Forest to learn “normal” process behavior from audit logs, then flag outliers. You can use distro packages (faster, no compile) or pip.

Install Python ML libraries (choose one path):

  • Distro packages:

    • Debian/Ubuntu:
    sudo apt install -y python3-sklearn python3-pandas
    • Fedora:
    sudo dnf install -y python3-scikit-learn python3-pandas
    • RHEL/CentOS (may require EPEL on some versions):
    sudo dnf install -y epel-release || true
sudo dnf install -y python3-scikit-learn python3-pandas
    • openSUSE/SLE:
    sudo zypper install -y python3-scikit-learn python3-pandas

  • Or via pip (in a venv):

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

Train from existing audit logs. The script below:

  • Parses audit events grouped by msg id

  • Extracts features: exec path, argc, hour of day, uid, path depth

  • Trains an Isolation Forest

  • Saves model and a simple encoder for categorical features

Save as train_model.py:

#!/usr/bin/env python3
import re, os, sys, json, time, math
from collections import defaultdict
from datetime import datetime
import pandas as pd
from sklearn.ensemble import IsolationForest
from sklearn.preprocessing import OneHotEncoder
from sklearn.compose import ColumnTransformer
from sklearn.pipeline import Pipeline
from joblib import dump

AUDIT_LOG = "/var/log/audit/audit.log"

msg_re = re.compile(r'msg=audit\(([^)]+)\)')
kv_re  = re.compile(r'([a-zA-Z0-9_]+)=(".*?"|\S+)')

def parse_audit(path):
    by_id = defaultdict(list)
    with open(path, 'r', errors='ignore') as f:
        for line in f:
            m = msg_re.search(line)
            if not m: 
                continue
            msgid = m.group(1).split(':')[-1]  # the numeric id after colon
            by_id[msgid].append(line.strip())
    records = []
    for mid, lines in by_id.items():
        rec = {"argc": None, "exe": None, "comm": None, "uid": None, "auid": None, "ts": None, "cwd": None}
        # timestamp
        try:
            first = lines[0]
            ts_part = first.split('audit(')[1].split(':')[0]
            rec["ts"] = float(ts_part)
        except:
            pass
        # merge KVs
        allk = {}
        for ln in lines:
            for k,v in kv_re.findall(ln):
                if v.startswith('"') and v.endswith('"'):
                    v = v[1:-1]
                allk.setdefault(k, v)
        # pick interesting fields
        rec["argc"] = int(allk["argc"]) if "argc" in allk and allk["argc"].isdigit() else None
        rec["exe"]  = allk.get("exe") or allk.get("path") or allk.get("comm")
        rec["comm"] = allk.get("comm")
        rec["uid"]  = int(allk["uid"]) if allk.get("uid","").isdigit() else None
        rec["auid"] = int(allk["auid"]) if allk.get("auid","").isdigit() else None
        rec["cwd"]  = allk.get("cwd")
        if rec["exe"] or rec["comm"]:
            records.append(rec)
    return records

def featurize(df):
    df = df.copy()
    # Time features
    df["hour"] = df["ts"].apply(lambda x: datetime.fromtimestamp(x).hour if pd.notnull(x) else -1)
    # Path features
    path = df["exe"].fillna(df["comm"].fillna("unknown"))
    df["exe"] = path
    df["path_depth"] = path.apply(lambda p: p.count("/") if isinstance(p,str) else 0)
    df["in_system_dir"] = path.apply(lambda p: 1 if isinstance(p,str) and (p.startswith("/usr/") or p.startswith("/bin/") or p.startswith("/sbin/")) else 0)
    # Argc fallback
    df["argc"] = df["argc"].fillna(0)
    df["uid"]  = df["uid"].fillna(-1)
    df["auid"] = df["auid"].fillna(-1)
    # Keep columns
    keep = ["exe","argc","hour","uid","auid","path_depth","in_system_dir"]
    return df[keep]

def main():
    path = AUDIT_LOG if len(sys.argv)==1 else sys.argv[1]
    recs = parse_audit(path)
    if not recs:
        print("No audit records found; generate activity then retry.", file=sys.stderr)
        sys.exit(1)
    df = pd.DataFrame(recs)
    X = featurize(df)

    # Build pipeline: OneHot on exe, pass-through numerics
    cat_cols = ["exe"]
    num_cols = [c for c in X.columns if c not in cat_cols]
    pre = ColumnTransformer([
        ("exe_ohe", OneHotEncoder(handle_unknown="ignore", sparse=True), cat_cols),
        ("num", "passthrough", num_cols)
    ])
    iso = IsolationForest(n_estimators=200, contamination=0.01, random_state=42)
    pipe = Pipeline([("prep", pre), ("model", iso)])
    pipe.fit(X)

    os.makedirs("./model", exist_ok=True)
    dump(pipe, "./model/audit_iforest.joblib")
    print(f"Trained on {len(X)} events; model saved to ./model/audit_iforest.joblib")

if __name__ == "__main__":
    main()

Run it:

sudo python3 train_model.py

Real-world payoffs you’ll see fast:

  • Web server user (www-data) spawning package managers or shells

  • Odd-hour admin activity that never happened before on that host

  • Long command lines typical of crypto miners or one-liners like curl ... | bash

Let the server run a typical workload for a day, then retrain for a better baseline.

Step 3: Score in real time and alert

We’ll tail the audit log, build records per event id, and score them with the saved model. High anomaly scores will be printed as alerts.

Save as score_realtime.py:

#!/usr/bin/env python3
import sys, re, time, json, os
from collections import defaultdict, deque
from datetime import datetime
import pandas as pd
from joblib import load

LOG = "/var/log/audit/audit.log"
MODEL = "./model/audit_iforest.joblib"

msg_re = re.compile(r'msg=audit\(([^)]+)\)')
kv_re  = re.compile(r'([a-zA-Z0-9_]+)=(".*?"|\S+)')

def featurize_one(rec):
    # mirror features from training
    exe = rec.get("exe") or rec.get("comm") or "unknown"
    ts = rec.get("ts") or time.time()
    hour = datetime.fromtimestamp(ts).hour
    path_depth = exe.count("/") if isinstance(exe,str) else 0
    in_system = 1 if isinstance(exe,str) and (exe.startswith("/usr/") or exe.startswith("/bin/") or exe.startswith("/sbin/")) else 0
    return pd.DataFrame([{
        "exe": exe,
        "argc": int(rec.get("argc") or 0),
        "hour": hour,
        "uid": int(rec.get("uid") or -1),
        "auid": int(rec.get("auid") or -1),
        "path_depth": path_depth,
        "in_system_dir": in_system
    }])

def main():
    pipe = load(MODEL)
    # group lines by msg id
    buffers = defaultdict(list)
    with open(LOG, "r", errors="ignore") as f:
        # seek end
        f.seek(0,2)
        while True:
            line = f.readline()
            if not line:
                time.sleep(0.2)
                continue
            m = msg_re.search(line)
            if not m:
                continue
            msgid = m.group(1).split(':')[-1]
            buffers[msgid].append(line.strip())
            if "type=EOE" in line or "type=SYSCALL" in line and any("type=EXECVE" in l for l in buffers[msgid]):
                # parse event
                allk = {}
                ts = None
                for ln in buffers[msgid]:
                    for k,v in kv_re.findall(ln):
                        if v.startswith('"') and v.endswith('"'): v = v[1:-1]
                        allk.setdefault(k, v)
                    if ts is None and "audit(" in ln:
                        try:
                            ts_part = ln.split('audit(')[1].split(':')[0]
                            ts = float(ts_part)
                        except:
                            pass
                rec = {
                    "ts": ts,
                    "argc": allk.get("argc"),
                    "exe": allk.get("exe") or allk.get("path") or allk.get("comm"),
                    "comm": allk.get("comm"),
                    "uid": allk.get("uid"),
                    "auid": allk.get("auid")
                }
                X = featurize_one(rec)
                score = -pipe.decision_function(X)[0]  # larger = more anomalous
                if score > 0.6:  # tune threshold
                    print(json.dumps({
                        "time": datetime.utcnow().isoformat()+"Z",
                        "score": round(score,3),
                        "exe": X.iloc[0]["exe"],
                        "uid": int(X.iloc[0]["uid"]),
                        "argc": int(X.iloc[0]["argc"])
                    }), flush=True)
                buffers.pop(msgid, None)

if __name__ == "__main__":
    main()

Run it as a service so it survives reboots.

Create a systemd unit:

sudo bash -c 'cat >/etc/systemd/system/ai-edr.service' << "EOF"
[Unit]
Description=AI Endpoint Anomaly Scorer
After=network.target auditd.service

[Service]
Type=simple
WorkingDirectory=/root/ai-edr
ExecStart=/usr/bin/python3 /root/ai-edr/score_realtime.py
Restart=always
RestartSec=2
Nice=5
NoNewPrivileges=true
PrivateTmp=true
ProtectSystem=full
ProtectHome=true

[Install]
WantedBy=multi-user.target
EOF

sudo mkdir -p /root/ai-edr/model
# copy your train_model.py and score_realtime.py into /root/ai-edr and the model file into /root/ai-edr/model
sudo systemctl daemon-reload
sudo systemctl enable --now ai-edr
journalctl -u ai-edr -f

Now you’ll see JSON alerts when unusual executions occur.

Step 4: Automate a safe response

Start conservative: alert first, then selectively block repeat offenders.

Example: quarantine unknown binaries with very high anomaly scores.

quarantine.sh:

#!/usr/bin/env bash
set -euo pipefail
read -r event
exe=$(echo "$event" | jq -r '.exe')
score=$(echo "$event" | jq -r '.score')
if (( $(echo "$score > 0.9" | bc -l) )) && [[ -x "$exe" ]] && [[ ! "$exe" =~ ^/(usr|bin|sbin)/ ]]; then
  ts=$(date -u +%Y%m%dT%H%M%SZ)
  sudo mkdir -p /root/quarantine
  sudo mv "$exe" "/root/quarantine/$(basename "$exe").$ts"
  logger -t ai-edr "Quarantined $exe at score=$score"
fi

Wire it to the scorer:

# Requires jq and bc installed
# Debian/Ubuntu:
sudo apt install -y jq bc
# Fedora/RHEL/CentOS:
sudo dnf install -y jq bc
# openSUSE/SLE:
sudo zypper install -y jq bc

# Run side-by-side (or integrate into the service ExecStart with a tee)
journalctl -u ai-edr -f -o cat | ./quarantine.sh

Other response ideas:

  • Tag suspicious UIDs with a temporary outbound block:

    # block UID 1003 outgoing traffic (example with nftables)
sudo nft add rule inet filter output meta skuid 1003 drop

  • Notify your SIEM or chat via curl webhook (include the JSON event)

Real-world examples this catches early

  • Crypto miner dropped in /tmp and executed by an odd service user—uncommon path depth + command shape spikes score.

  • “curl https://… | bash” launched under nginx user—never seen before on that role/host.

  • Package manager or compiler invoked by www-data—unexpected executable with non-admin UID.

Maintenance: keep it healthy

  • Retrain after changes or weekly cron:

    (cd /root/ai-edr && sudo python3 train_model.py && sudo systemctl restart ai-edr)

  • Triage top alerts; whitelist known-good rare tasks by:

    • Lowering scores for approved paths
    • Adding explicit allow rules in audit (or excluding service accounts)

  • Rotate logs and watch disk:

    sudo du -sh /var/log/audit
sudo grep -A5 "^/var/log/audit" /etc/logrotate.d/auditd

Conclusion and next steps

You don’t need heavyweight EDR to get smarter at the endpoint. With auditd for truth, a small local model for “weirdness,” and a cautious response script, you’ll surface attacks that signatures miss—without shipping your host’s internals anywhere.

Your next 60 minutes: 1) Install and enable auditd (copy rules above). 2) Generate normal activity for a day and train the model. 3) Run the scorer, watch alerts, and add a simple quarantine or notification.

Want to go further? Feed these alerts into your SIEM, enrich with network context, or swap in more features (parent PID, cwd, remote IP) for stronger models. The foundation is now in place—owned by you, on your Linux box.

Related posts