Modern applications generate massive volumes of log data every second, and manually identifying suspicious patterns—brute-force attempts, credential stuffi
Modern applications generate massive volumes of log data every second, and manually identifying suspicious patterns—brute-force attempts, credential stuffing, scraping bots, or novel attack vectors—is impossible at scale. Traditional rule-based detectors (regex for known bad patterns) fail against zero-day threats and generate noisy false positives that cause alert fatigue.
This script solves that problem by combining unsupervised machine learning (Isolation Forest on TF-IDF features) with LLM-powered semantic analysis. The ML layer surfaces statistically anomalous log entries without needing labeled training data, while the OpenAI layer explains why a batch of anomalies looks suspicious in plain English—turning raw log noise into an actionable security report.
The script follows a two-stage detection pipeline. First, log lines are parsed into structured records (IP, timestamp, method, path, status, user agent) using a regex tuned for common web-server formats (nginx/Apache combined). Structured fields plus the raw line are vectorized with TfidfVectorizer (n-grams over tokens and character patterns), then scored by IsolationForest, which isolates outliers based on how quickly random partitions separate them from the rest. Records below a configurable contamination threshold are flagged as anomalies.
Second, the top-N anomalies are batched into a prompt for the OpenAI API, which classifies each as benign, suspicious, or malicious, explains the reasoning, and suggests a mitigation. The script caches the model, streams results to stdout as JSON or human-readable text, and exits with a non-zero code if any malicious verdict is returned—making it easy to wire into CI/CD or cron-based alerting.
#!/usr/bin/env python3
"""
AI-Powered Log Anomaly Detector
Parses web-server access logs, uses an Isolation Forest on TF-IDF features
to surface statistically anomalous entries, then asks an LLM to classify and
explain the top anomalies in plain English.
Usage:
python3 log_anomaly.py --log /var/log/nginx/access.log
"""
from __future__ import annotations
import argparse
import json
import os
import re
import sys
from dataclasses import dataclass, asdict
from pathlib import Path
from typing import Iterable
import numpy as np
from sklearn.ensemble import IsolationForest
from sklearn.feature_extraction.text import TfidfVectorizer
try:
from openai import OpenAI
except ImportError: # pragma: no cover
OpenAI = None # type: ignore[assignment]
# ---------------------------------------------------------------------------
# Configuration (environment variables with sensible defaults)
# ---------------------------------------------------------------------------
OPENAI_API_KEY = os.environ.get("OPENAI_API_KEY", "")
OPENAI_MODEL = os.environ.get("OPENAI_MODEL", "gpt-4o-mini")
CONTAMINATION = float(os.environ.get("ANOMALY_CONTAMINATION", "0.05"))
TOP_N = int(os.environ.get("ANOMALY_TOP_N", "15"))
MAX_LINES = int(os.environ.get("ANOMALY_MAX_LINES", "200000"))
REQUEST_TIMEOUT = float(os.environ.get("OPENAI_TIMEOUT", "30"))
# ---------------------------------------------------------------------------
# Log parsing
# ---------------------------------------------------------------------------
# Matches nginx/Apache "combined" format:
# 1.2.3.4 - - [10/Oct/2024:13:55:36 +0000] "GET /path HTTP/1.1" 200 1234 "-" "UA"
LOG_RE = re.compile(
r'^(?P<ip>\S+) \S+ \S+ '
r'\[(?P<ts>[^\]]+)\] '
r'"(?P<method>[A-Z]+) (?P<path>\S+) (?P<proto>[^"]+)" '
r'(?P<status>\d{3}) (?P<size>\S+) '
r'"(?P<referer>[^"]*)" "(?P<ua>[^"]*)"'
)
@dataclass
class LogRecord:
raw: str
ip: str
timestamp: str
method: str
path: str
status: int
size: int
user_agent: str
def parse_line(line: str) -> LogRecord | None:
"""Parse one log line; return None if it doesn't match the expected format."""
match = LOG_RE.match(line.strip())
if not match:
return None
g = match.groupdict()
try:
status = int(g["status"])
size = int(g["size"]) if g["size"].isdigit() else 0
except (TypeError, ValueError):
return None
return LogRecord(
raw=line.strip(),
ip=g["ip"],
timestamp=g["ts"],
method=g["method"],
path=g["path"],
status=status,
size=size,
user_agent=g["ua"],
)
def load_logs(path: Path) -> list[LogRecord]:
"""Read and parse up to MAX_LINES records from a log file."""
if not path.exists():
raise FileNotFoundError(f"Log file not found: {path}")
records: list[LogRecord] = []
with path.open("r", encoding="utf-8", errors="replace") as fh:
for i, line in enumerate(fh):
if i >= MAX_LINES:
break
rec = parse_line(line)
if rec is not None:
records.append(rec)
if not records:
raise ValueError("No parseable log lines found in the input file.")
return records
# ---------------------------------------------------------------------------
# Feature engineering + anomaly detection
# ---------------------------------------------------------------------------
def build_feature_text(rec: LogRecord) -> str:
"""Combine structured fields into a single text blob for TF-IDF."""
# Bucket status codes so 4xx/5xx cluster together semantically.
status_bucket = f"{rec.status // 100}xx"
return (
f"{rec.method} {rec.path} {status_bucket} "
f"ua={rec.user_agent} ip={rec.ip}"
)
def detect_anomalies(records: list[LogRecord]) -> list[tuple[LogRecord, float]]:
"""Return (record, anomaly_score) pairs sorted by descending score."""
corpus = [build_feature_text(r) for r in records]
# TF-IDF with word + character n-grams captures both token patterns
# (e.g. "/admin") and structural oddities (e.g. long paths, encoding).
vectorizer = TfidfVectorizer(
analyzer="word",
ngram_range=(1, 2),
min_df=1,
max_features=20_000,
sublinear_tf=True,
)
X = vectorizer.fit_transform(corpus)
model = IsolationForest(
n_estimators=200,
contamination=CONTAMINATION,
random_state=42,
n_jobs=-1,
)
model.fit(X)
# decision_function: higher = more normal, lower = more anomalous.
scores = model.decision_function(X)
predictions = model.predict(X) # -1 = anomaly, 1 = normal
flagged = [
(records[i], float(-scores[i]))
for i in range(len(records))
if predictions[i] == -1
]
flagged.sort(key=lambda pair: pair[1], reverse=True)
return flagged
# ---------------------------------------------------------------------------
# LLM enrichment
# ---------------------------------------------------------------------------
SYSTEM_PROMPT = (
"You are a senior security analyst. For each log entry, return a JSON "
"object with keys: verdict ('benign'|'suspicious'|'malicious'), "
"reason (one sentence), and mitigation (one sentence or empty string). "
"Respond with a JSON object of the form {\"results\": [ ... ]} in the "
"same order as the input entries."
)
def classify_with_llm(anomalies: list[tuple[LogRecord, float]]) -> list[dict]:
"""Send top-N anomalies to OpenAI and return structured verdicts."""
if OpenAI is None:
raise RuntimeError("openai package is not installed. Run: pip install openai")
if not OPENAI_API_KEY:
raise RuntimeError("OPENAI_API_KEY environment variable is not set.")
client = OpenAI(api_key=OPENAI_API_KEY, timeout=REQUEST_TIMEOUT)
payload = [
{
"ip": rec.ip,
"method": rec.method,
"path": rec.path,
"status": rec.status,
"user_agent": rec.user_agent,
"anomaly_score": round(score, 3),
}
for rec, score in anomalies[:TOP_N]
]
user_msg = (
"Analyze these anomalous web-server log entries. "
"Consider brute-force patterns, path traversal, scanner user agents, "
"and unusual status codes.\n\n"
f"{json.dumps(payload, indent=2)}"
)
response = client.chat.completions.create(
model=OPENAI_MODEL,
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": user_msg},
],
response_format={"type": "json_object"},
temperature=0.1,
)
content = response.choices[0].message.content or "{}"
try:
parsed = json.loads(content)
except json.JSONDecodeError as exc:
raise RuntimeError(f"LLM returned non-JSON content: {content!r}") from exc
return parsed.get("results", [])
# ---------------------------------------------------------------------------
# Output + entrypoint
# ---------------------------------------------------------------------------
def render_report(
anomalies: list[tuple[LogRecord, float]],
verdicts: list[dict],
as_json: bool,
) -> tuple[str, bool]:
"""Format the final report; return (text, has_malicious)."""
has_malicious = any(v.get("verdict") == "malicious" for v in verdicts)
if as_json:
out = {
"summary": {
"total_anomalies": len(anomalies),
"analyzed": len(verdicts),
"has_malicious": has_malicious,
},
"findings": [
{
"log": asdict(rec),
"anomaly_score": round(score, 4),
"analysis": verdicts[i] if i < len(verdicts) else None,
}
for i, (rec, score) in enumerate(anomalies[:TOP_N])
],
}
return json.dumps(out, indent=2), has_malicious
lines = [
f"Detected {len(anomalies)} anomalies (analyzing top {len(verdicts)}).",
"-" * 72,
]
for i, (rec, score) in enumerate(anomalies[:TOP_N]):
verdict = verdicts[i] if i < len(verdicts) else {}
lines.append(
f"[{i + 1}] score={score:.3f} {rec.method} {rec.path} "
f"status={rec.status} ip={rec.ip}"
)
lines.append(f" verdict: {verdict.get('verdict', 'n/a')}")
lines.append(f" reason: {verdict.get('reason', 'n/a')}")
if verdict.get("mitigation"):
lines.append(f" action: {verdict['mitigation']}")
lines.append("")
return "\n".join(lines), has_malicious
def parse_args(argv: Iterable[str] | None = None) -> argparse.Namespace:
parser = argparse.ArgumentParser(description="AI-Powered Log Anomaly Detector")
parser.add_argument("--log", required=True, type=Path, help="Path to access log")
parser.add_argument("--json", action="store_true", help="Emit JSON output")
parser.add_argument(
"--no-llm",
action="store_true",
help="Skip OpenAI enrichment (ML anomalies only)",
)
return parser.parse_args(argv)
def main(argv: Iterable[str] | None = None) -> int:
args = parse_args(argv)
try:
records = load_logs(args.log)
except (FileNotFoundError, ValueError) as exc:
print(f"error: {exc}", file=sys.stderr)
return 2
anomalies = detect_anomalies(records)
if not anomalies:
print("No anomalies detected.", file=sys.stderr)
return 0
verdicts: list[dict] = []
if not args.no_llm:
try:
verdicts = classify_with_llm(anomalies)
except RuntimeError as exc:
print(f"warning: LLM analysis skipped ({exc})", file=sys.stderr)
report, has_malicious = render_report(anomalies, verdicts, args.json)
print(report)
return 1 if has_malicious else 0
if __name__ == "__main__":
sys.exit(main())
| Variable | Default | Description |
|---|---|---|
OPENAI_API_KEY |
(empty) | Required for LLM enrichment. If unset, the script warns and prints ML-only results (unless --no-llm is passed). |
OPENAI_MODEL |
gpt-4o-mini |
Any chat-completions model that supports response_format={"type": "json_object"}. |
ANOMALY_CONTAMINATION |
0.05 |
Expected fraction of anomalies. Raise for noisier logs, lower for stricter detection. |
ANOMALY_TOP_N |
15 |
Number of top anomalies sent to the LLM and shown in the report. Keep ≤ 25 to control token cost. |
ANOMALY_MAX_LINES |
200000 |
Hard cap on lines read from the log file to bound memory and runtime. |
OPENAI_TIMEOUT |
30 |
Seconds before the OpenAI request aborts. |
CLI flags: --log (required), --json (machine-readable output), --no-llm (skip API calls).
1. Full analysis of an nginx access log:
export OPENAI_API_KEY=sk-...
python3 log_anomaly.py --log /var/log/nginx/access.log
Expected output:
Detected 47 anomalies (analyzing top 15).
------------------------------------------------------------------------
[1] score=0.412 GET /wp-login.php status=401 ip=185.220.101.5
verdict: malicious
reason: Repeated 401s on /wp-login.php from a known Tor exit node indicate credential stuffing.
action: Block the source IP and enable rate limiting on /wp-login.php.
...
2. ML-only mode for air-gapped environments:
python3 log_anomaly.py --log /var/log/apache2/access.log --no-llm
3. JSON output piped into a SIEM or alerting pipeline:
python3 log_anomaly.py --log ./access.log --json | jq '.summary'
# { "total_anomalies": 47, "analyzed": 15, "has_malicious": true }
The process exits with code 1 when any malicious verdict is returned, so cron or CI jobs can trigger alerts on failure.
FileNotFoundError caught in main, prints to stderr, exits 2.ValueError from load_logs, same handling.parse_line returns None; those lines are silently skipped so a few bad lines don't kill the run.openai package → ImportError is guarded at import time; the LLM call raises a clear RuntimeError that main downgrades to a warning, so ML results still print.json.JSONDecodeError is re-raised as RuntimeError with the raw content for debugging.RuntimeError branch; the OpenAI SDK's own exceptions bubble up as RuntimeError subclasses in most versions, and the report still prints ML findings.load_logs with a tail-follow reader (pygtail, watchdog) and run detect_anomalies on rolling windows so detection happens in near real-time.TfidfVectorizer + IsolationForest after a warm-up period, then load them at startup so scoring is O(1) per line and doesn't require re-fitting on every run.scipy.sparse.hstack for a stronger signal.