Posted on
• Artificial Intelligence

Artificial Intelligence Log Correlation

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

Artificial Intelligence Log Correlation on Linux (with Bash-first workflows)

Ever feel like you only find out about incidents after the damage is done—and your logs didn’t “say it out loud” soon enough? The truth is: your logs did say it. Across SSH, web, and app logs, the signal was there—just scattered. AI-powered log correlation helps you stitch together weak signals from different sources into a single, timely warning you can act on.

In this post, you’ll learn a practical, Unix-friendly way to do AI log correlation using Bash and a small Python helper. You’ll normalize logs, extract features, score anomalies, and get actionable output you can automate via cron. No heavy stacks required.

Why AI log correlation is worth your time

  • Detect multi-vector attacks: SSH brute-force aligned with web scanning and DB errors in the same time window? That’s coordinated.

  • Reduce MTTR: Surface cross-system relationships automatically so you can jump straight to the likely root cause.

  • Scale beyond rules: Static rules miss novel behaviors; anomaly detection adapts as your environment changes.

  • Stay Unix-friendly: Keep your pipelines simple—pipe, parse, learn, alert—without committing to a heavyweight platform up front.

What we’ll build

A minimal, scriptable AI correlation pipeline you can run anywhere:

  • Input: systemd/journald (for sshd) + Nginx access logs

  • Transform: per-minute features per source IP (e.g., SSH failures, HTTP 5xx/404 counts)

  • Model: Isolation Forest (unsupervised anomaly detection)

  • Output: ranked anomalies with a short “why” (feature spikes) you can email, log, or page

Installation (Debian/Ubuntu, Fedora/RHEL, openSUSE)

We’ll use Python 3 with pip to install ML dependencies.

  • APT (Debian/Ubuntu and derivatives):
sudo apt update
sudo apt install -y python3 python3-pip
  • DNF (Fedora/RHEL/CentOS Stream):
sudo dnf install -y python3 python3-pip
  • Zypper (openSUSE/SLE):
sudo zypper refresh
sudo zypper install -y python3 python3-pip

Then install Python packages for the script:

python3 -m pip install --user pandas scikit-learn

Tip: If you prefer a virtual environment:

python3 -m venv ~/ai-log-corr-venv
. ~/ai-log-corr-venv/bin/activate
pip install pandas scikit-learn

Step-by-step: From raw logs to correlated anomalies

1) Save the correlation script

Create ai_log_corr.py with the following content:

#!/usr/bin/env python3
import argparse, json, re, subprocess, sys
from collections import defaultdict
from datetime import datetime
import pandas as pd
from sklearn.ensemble import IsolationForest

SSH_FAIL_RE = re.compile(r"Failed password .* from (?P<ip>\d+\.\d+\.\d+\.\d+)")
NGINX_CLF_RE = re.compile(
    r'^(?P<ip>\S+) \S+ \S+ \[(?P<time>[^\]]+)\] "(?P<req>[^"]*)" (?P<status>\d{3}) (?P<size>\S+) "(?P<ref>[^"]*)" "(?P<ua>[^"]*)"'
)

def epoch_minute(ts_seconds: float) -> int:
    return int(ts_seconds // 60)

def load_ssh_events(since: str):
    # journalctl outputs JSON per line with __REALTIME_TIMESTAMP in microseconds
    cmd = ["journalctl", "-u", "ssh.service", "--since", since, "-o", "json"]
    p = subprocess.Popen(cmd, stdout=subprocess.PIPE, text=True)
    by_key = defaultdict(lambda: {"ssh_fail": 0})
    for line in p.stdout:
        line = line.strip()
        if not line:
            continue
        try:
            obj = json.loads(line)
        except json.JSONDecodeError:
            continue
        msg = obj.get("MESSAGE", "")
        m = SSH_FAIL_RE.search(msg)
        if not m:
            continue
        ip = m.group("ip")
        ts_micro = int(obj.get("__REALTIME_TIMESTAMP", "0"))
        minute = epoch_minute(ts_micro / 1_000_000.0)
        by_key[(minute, ip)]["ssh_fail"] += 1
    p.wait()
    return by_key

def parse_nginx_time(t: str) -> int:
    # Example: 10/Jul/2026:13:45:01 +0000
    dt = datetime.strptime(t, "%d/%b/%Y:%H:%M:%S %z")
    return epoch_minute(dt.timestamp())

def load_nginx_events(path: str):
    by_key = defaultdict(lambda: {"nginx_5xx": 0, "nginx_404": 0})
    try:
        with open(path, "r", encoding="utf-8", errors="ignore") as f:
            for line in f:
                m = NGINX_CLF_RE.match(line)
                if not m:
                    continue
                ip = m.group("ip")
                t = m.group("time")
                status = int(m.group("status"))
                minute = parse_nginx_time(t)
                if 500 <= status <= 599:
                    by_key[(minute, ip)]["nginx_5xx"] += 1
                elif status == 404:
                    by_key[(minute, ip)]["nginx_404"] += 1
    except FileNotFoundError:
        pass
    return by_key

def merged_features(ssh_by_key, ngx_by_key):
    keys = set(ssh_by_key.keys()) | set(ngx_by_key.keys())
    rows = []
    for (minute, ip) in keys:
        ssh = ssh_by_key.get((minute, ip), {})
        ngx = ngx_by_key.get((minute, ip), {})
        rows.append({
            "minute": minute,
            "ip": ip,
            "ssh_fail": int(ssh.get("ssh_fail", 0)),
            "nginx_5xx": int(ngx.get("nginx_5xx", 0)),
            "nginx_404": int(ngx.get("nginx_404", 0)),
        })
    df = pd.DataFrame(rows)
    if df.empty:
        return df
    return df.fillna(0)

def explain_row(row):
    parts = []
    for k in ["ssh_fail", "nginx_5xx", "nginx_404"]:
        v = int(row[k])
        if v > 0:
            parts.append(f"{k}={v}")
    return "; ".join(parts) if parts else "no features > 0"

def main():
    ap = argparse.ArgumentParser(description="AI log correlation: SSH + Nginx anomaly detector")
    ap.add_argument("--nginx", default="/var/log/nginx/access.log", help="Path to Nginx access.log")
    ap.add_argument("--ssh-since", default="2 hours ago", help="journalctl --since window (e.g., '90 min ago', 'today')")
    ap.add_argument("--top", type=int, default=10, help="Show top-N anomalies")
    ap.add_argument("--contamination", type=float, default=0.01, help="Expected anomaly proportion (0.0-0.5)")
    args = ap.parse_args()

    ssh_by_key = load_ssh_events(args.ssh_since)
    ngx_by_key = load_nginx_events(args.nginx)
    df = merged_features(ssh_by_key, ngx_by_key)

    if df.empty:
        print("No data in the selected window/logs.", file=sys.stderr)
        sys.exit(0)

    X = df[["ssh_fail", "nginx_5xx", "nginx_404"]].values
    model = IsolationForest(
        n_estimators=200,
        contamination=args.contamination,
        random_state=0,
        n_jobs=-1,
    )
    model.fit(X)
    scores = -model.score_samples(X)  # higher => more anomalous
    df = df.assign(anomaly_score=scores)

    # Rank and display
    out = df.sort_values("anomaly_score", ascending=False).head(args.top)
    print("minute,ip,ssh_fail,nginx_5xx,nginx_404,anomaly_score,why")
    for _, r in out.iterrows():
        print(f"{int(r['minute'])},{r['ip']},{int(r['ssh_fail'])},{int(r['nginx_5xx'])},{int(r['nginx_404'])},{r['anomaly_score']:.4f},{explain_row(r)}")

if __name__ == "__main__":
    main()

Make it executable:

chmod +x ai_log_corr.py

Notes:

  • You may need sudo to read /var/log/nginx/access.log.

  • The script uses journalctl for sshd; if your unit is sshd.service instead of ssh.service, adjust the command accordingly.

2) Run it against live logs

Examples:

# Past 2 hours, default top-10
sudo ./ai_log_corr.py

# Past 30 minutes, show top-5 anomalies
sudo ./ai_log_corr.py --ssh-since "30 minutes ago" --top 5

# Custom Nginx path
sudo ./ai_log_corr.py --nginx /var/log/nginx/your_vhost.access.log

Sample output:

minute,ip,ssh_fail,nginx_5xx,nginx_404,anomaly_score,why
29123456,203.0.113.45,18,0,0,0.8921,ssh_fail=18
29123456,198.51.100.27,0,5,12,0.8443,nginx_5xx=5; nginx_404=12
29123457,203.0.113.45,6,2,7,0.8209,ssh_fail=6; nginx_5xx=2; nginx_404=7
...

Interpretation:

  • Each row is a minute bucket and source IP with feature counts.

  • anomaly_score is higher for “strange” combinations or spikes across features.

  • why summarizes which features fired, so you can act quickly.

3) Automate with cron

Schedule it to run every 5 minutes and log results:

sudo install -d -m 750 /var/log/ai-log-corr
sudo sh -c 'echo "*/5 * * * * root /usr/bin/python3 /opt/ai-log-corr/ai_log_corr.py --ssh-since \"30 minutes ago\" --top 5 >> /var/log/ai-log-corr/anomalies.log 2>&1" >/etc/cron.d/ai-log-corr'

Or with a user crontab (ensure the user can read the logs):

crontab -e
# Add:
*/5 * * * * /usr/bin/python3 /home/you/ai_log_corr.py --ssh-since "30 minutes ago" --top 5 >> /home/you/ai-log-corr.log 2>&1

4) Tune and extend features

  • Add more sources: parse app logs with a small regex and add columns (e.g., app_errors, db_timeouts).

  • Adjust contamination to control sensitivity (e.g., 0.02 for noisier environments).

  • Use per-service or per-path features in Nginx (e.g., 40x on /wp-login.php).

  • Whitelist known scanners or internal CIDRs before modeling.

  • Align time windows to known events (deploys, migrations) to catch causality.

5) Real-world correlation patterns to watch

  • Coordinated probe: SSH brute-force plus a spike of 404/403 on WordPress or admin paths from the same IP block within minutes.

  • Outage precursor: Rising 5xx on a service that maps to a rollout window, followed by elevated SSH activity from an automation IP.

  • Data exfil attempt: An unusual burst of large 200 responses after repeated 401/403s, accompanied by SSH auth noise.

Why this works (and where to go next)

Isolation Forest is good at flagging unusual combinations of features. By bucketing per minute and per IP, you get a simple “correlation surface” where spikes across sources stand out, even if each signal would be ignorable on its own. This is the essence of AI log correlation: fuse weak signals to find strong incidents.

Next steps if you want more:

  • Enrich with GeoIP, ASN, or user-agent classes to improve discriminative power.

  • Persist features to a time-series DB and feed a streaming model (e.g., online anomaly detection).

  • Consider an open-source stack with dashboards and built-in anomaly detection (e.g., OpenSearch + Anomaly Detection plugin) once you outgrow scripts.

Call to Action

  • Install the prerequisites, drop in the script, and run it against the last 2 hours of logs.

  • Put it on cron for a week and see what it surfaces; iterate your features.

  • Share a sanitized top anomaly with your team and decide what alert threshold makes sense for production.

Your logs already know what’s wrong. Correlate them with a dash of AI—and let your shell do the rest.

Related posts