Glen Larkin

Project Overview

loganalyzer.py is a Python-based log anomaly detection tool that parses Linux authentication logs and flags suspicious activity — brute force attempts, privilege escalation, and off-hours logins. Findings are exported to a timestamped CSV report sorted by severity, ready to feed into a ticketing or incident response workflow.

Why I Built This

How It Works

Usage

python3 loganalyzer.py /var/log/auth.log

python3 loganalyzer.py /var/log/auth.log /var/log/syslog

python3 loganalyzer.py /var/log/auth.log -o incident_report.csv

Parsing /var/log/auth.log...
Parsed 9 events. Running detection...

=== ANOMALY SUMMARY ===
  CRITICAL: 1
  HIGH: 1
  MEDIUM: 1
  TOTAL: 3

Report saved → anomalies_20260528_201500.csv

Full Source Code

~160 lines of pure Python, no external dependencies. Available on GitHub.

#!/usr/bin/env python3
"""Log anomaly detector: parses auth/syslog logs and flags suspicious events."""

import re, csv, sys
from collections import defaultdict
from datetime import datetime
from pathlib import Path

BRUTE_FORCE_THRESHOLD = 5
BRUTE_FORCE_WINDOW_SEC = 60
OFF_HOURS = range(0, 6)

PATTERNS = {
    "ssh_fail": re.compile(
        r"(?P<month>\w+)\s+(?P<day>\d+)\s+(?P<time>\S+).*Failed password for "
        r"(?:invalid user )?(?P<user>\S+) from (?P<ip>\S+)"
    ),
    "ssh_success": re.compile(
        r"(?P<month>\w+)\s+(?P<day>\d+)\s+(?P<time>\S+).*Accepted \S+ for "
        r"(?P<user>\S+) from (?P<ip>\S+)"
    ),
    "sudo": re.compile(
        r"(?P<month>\w+)\s+(?P<day>\d+)\s+(?P<time>\S+).*sudo.*USER=(?P<user>\S+)"
        r".*COMMAND=(?P<cmd>.+)"
    ),
}


def parse_syslog_time(month, day, time_str):
    try:
        return datetime.strptime(f"{datetime.now().year} {month} {day} {time_str}",
                                 "%Y %b %d %H:%M:%S")
    except ValueError:
        return datetime.now()


def parse_log(path):
    events = []
    with open(path, errors="replace") as f:
        for line in f:
            for etype, pat in PATTERNS.items():
                m = pat.search(line)
                if m:
                    d = m.groupdict()
                    events.append({
                        "type": etype,
                        "timestamp": parse_syslog_time(
                            d.get("month", "Jan"), d.get("day", "1"), d.get("time", "00:00:00")),
                        "ip": d.get("ip", ""),
                        "user": d.get("user", ""),
                        "cmd": d.get("cmd", ""),
                    })
                    break
    return events


def detect_anomalies(events):
    findings = []
    fails_by_ip = defaultdict(list)

    for e in events:
        if e["type"] == "ssh_fail":
            fails_by_ip[e["ip"]].append(e["timestamp"])

        elif e["type"] == "ssh_success" and e["timestamp"].hour in OFF_HOURS:
            findings.append({"severity": "MEDIUM", "rule": "Off-Hours Login",
                "timestamp": e["timestamp"].isoformat(), "ip": e["ip"],
                "user": e["user"], "detail": f"Login at {e['timestamp'].strftime('%H:%M')}"})

        elif e["type"] == "sudo":
            findings.append({"severity": "HIGH", "rule": "Privilege Escalation",
                "timestamp": e["timestamp"].isoformat(), "ip": "",
                "user": e["user"], "detail": f"sudo: {e['cmd'][:80]}"})

    for ip, times in fails_by_ip.items():
        times.sort()
        for i in range(len(times)):
            window = [t for t in times[i:]
                      if (t - times[i]).total_seconds() <= BRUTE_FORCE_WINDOW_SEC]
            if len(window) >= BRUTE_FORCE_THRESHOLD:
                findings.append({"severity": "CRITICAL", "rule": "Brute Force Attempt",
                    "timestamp": times[i].isoformat(), "ip": ip, "user": "",
                    "detail": f"{len(window)} failed logins in {BRUTE_FORCE_WINDOW_SEC}s"})
                break

    return sorted(findings, key=lambda x: x["severity"])


def main():
    import argparse
    parser = argparse.ArgumentParser(description="Log anomaly detector")
    parser.add_argument("logs", nargs="+")
    parser.add_argument("-o", "--output")
    args = parser.parse_args()

    output = args.output or f"anomalies_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv"
    all_events = []
    for path in args.logs:
        print(f"Parsing {path}...")
        all_events.extend(parse_log(path))

    findings = detect_anomalies(all_events)

    with open(output, "w", newline="") as f:
        writer = csv.DictWriter(f,
            fieldnames=["severity", "rule", "timestamp", "ip", "user", "detail"])
        writer.writeheader()
        writer.writerows(findings)

    counts = defaultdict(int)
    for f in findings:
        counts[f["severity"]] += 1
    print("\n=== ANOMALY SUMMARY ===")
    for sev in ["CRITICAL", "HIGH", "MEDIUM"]:
        if counts[sev]:
            print(f"  {sev}: {counts[sev]}")
    print(f"  TOTAL: {len(findings)}\nReport saved → {output}")


if __name__ == "__main__":
    main()