Log Analyzer

A log-parsing tool that scans SSH auth.log (and generic access logs) with regular expressions, tallies failures per source IP with collections.Counter, and flags brute-force sources over a threshold.

Project Overview

Authentication and access logs are the first place a defender looks for brute-force and credential-stuffing activity. This project parses auth.log-style lines, extracts failed-login source IPs with a compiled regex, counts them with Counter, and reports any IP exceeding a configurable threshold within the file. It generalizes to web access logs by swapping the pattern, teaching the core "parse → count → threshold → alert" detection loop.

[!warning] Analyze logs you are authorized to read System logs can contain personal data. Only process logs from systems you own or administer, and handle any extracted IPs/usernames responsibly.

Objectives

Stream large log files with constant memory.
Extract structured fields from unstructured lines with compiled regular expressions.
Aggregate events by source, user, and time window.
Detect brute-force patterns and unusual activity.
Produce a sorted, actionable report.

Requirements

Item	Detail
Python	3.9 or newer
Dependencies	`re`, `collections`, `datetime`, `argparse` (standard library)
Privileges	Read access to the log files; often root for `/var/log`
Input	Synthetic logs, or logs you are authorized to analyse
Formats	Supports syslog-style auth logs and common web access logs

Architecture

CLI — argparse for log path, failure threshold, and pattern preset (ssh/http).
Reader — streams the file line by line (safe for multi-GB logs).
Matcher — a compiled re pattern pulls the source IP (and optionally username) from failure lines.
Counter — collections.Counter tallies failures per IP; .most_common() ranks them.
Detector — any IP at/above the threshold is reported as a suspected brute-force source.

log lines ─▶ regex match failures ─▶ Counter[ip] += 1
                                          │
                                          ▼
                              ip >= threshold ? ─▶ alert

Directory Structure

log-analyzer/
├── log_analyzer.py     # CLI + parsing/detection
└── sample-auth.log     # sample data for testing

Implementation Plan

Stream the file. Iterate the handle directly and count lines; confirm memory stays flat on a large file.
Compile the pattern once. Build a regex with named groups for timestamp, host, user, and source IP, compiled outside the loop.
Classify events. Separate successful authentications, failures, and everything else.
Aggregate. Count per source IP and per username with Counter.
Apply a time window. Keep recent timestamps per IP in a deque and flag N failures within M minutes — this distinguishes a burst from slow background noise.
Detect spraying. Also aggregate per username across many sources, which a per-IP threshold misses entirely.
Support compressed logs transparently with gzip.
Report. Sort by count descending and emit both a table and JSON.

Complete Example Implementation

#!/usr/bin/env python3
"""log_analyzer.py - Detect brute-force sources in auth/access logs.

Analyze only logs you are authorized to read.
"""
import argparse
import re
import sys
from collections import Counter

# Preset regexes: capture the offending source IP from a failure line.
PATTERNS = {
    # sshd: "Failed password for invalid user bob from 10.0.0.5 port 22 ssh2"
    "ssh": re.compile(
        r"Failed password for (?:invalid user )?(?P<user>\S+) "
        r"from (?P<ip>\d{1,3}(?:\.\d{1,3}){3})"
    ),
    # Common access log 401/403: capture leading IP on auth-failure lines.
    "http": re.compile(
        r"^(?P<ip>\d{1,3}(?:\.\d{1,3}){3}).*\"\s(?:401|403)\s"
    ),
}


def analyze(path: str, pattern: re.Pattern) -> tuple[Counter, Counter]:
    """Return (failures_by_ip, failures_by_user) Counters."""
    by_ip: Counter = Counter()
    by_user: Counter = Counter()
    with open(path, encoding="utf-8", errors="ignore") as fh:
        for line in fh:
            m = pattern.search(line)
            if m:
                by_ip[m.group("ip")] += 1
                if "user" in m.groupdict() and m.group("user"):
                    by_user[m.group("user")] += 1
    return by_ip, by_user


def main() -> int:
    parser = argparse.ArgumentParser(description="Brute-force log analyzer.")
    parser.add_argument("logfile", help="path to the log file")
    parser.add_argument("-p", "--preset", choices=sorted(PATTERNS), default="ssh",
                        help="log format preset (default: ssh)")
    parser.add_argument("-t", "--threshold", type=int, default=5,
                        help="failures per IP to flag (default: 5)")
    parser.add_argument("--top", type=int, default=10,
                        help="show top N offenders (default: 10)")
    args = parser.parse_args()

    try:
        by_ip, by_user = analyze(args.logfile, PATTERNS[args.preset])
    except FileNotFoundError:
        print(f"[!] No such file: {args.logfile}", file=sys.stderr)
        return 2

    print(f"[*] {sum(by_ip.values())} failure(s) from {len(by_ip)} unique IP(s)\n")
    print(f"[*] Top {args.top} source IPs:")
    for ip, count in by_ip.most_common(args.top):
        flag = "  <-- BRUTE-FORCE" if count >= args.threshold else ""
        print(f"    {count:>5}  {ip}{flag}")

    if by_user:
        print("\n[*] Most targeted usernames:")
        for user, count in by_user.most_common(5):
            print(f"    {count:>5}  {user}")

    flagged = [ip for ip, c in by_ip.items() if c >= args.threshold]
    print(f"\n[*] {len(flagged)} IP(s) over threshold ({args.threshold}).")
    return 1 if flagged else 0


if __name__ == "__main__":
    raise SystemExit(main())

Create sample data and run:

cat > sample-auth.log <<'LOG'
Jan 10 10:00:01 host sshd[111]: Failed password for invalid user admin from 10.0.0.5 port 22 ssh2
Jan 10 10:00:02 host sshd[112]: Failed password for root from 10.0.0.5 port 22 ssh2
Jan 10 10:00:03 host sshd[113]: Failed password for root from 10.0.0.5 port 22 ssh2
Jan 10 10:00:04 host sshd[114]: Failed password for invalid user test from 10.0.0.5 port 22 ssh2
Jan 10 10:00:05 host sshd[115]: Failed password for root from 10.0.0.5 port 22 ssh2
Jan 10 10:01:00 host sshd[120]: Failed password for bob from 192.168.1.20 port 22 ssh2
LOG
python log_analyzer.py sample-auth.log -p ssh -t 5

Output

[*] 6 failure(s) from 2 unique IP(s)

[*] Top 10 source IPs:
        5  10.0.0.5  <-- BRUTE-FORCE
        1  192.168.1.20

[*] Most targeted usernames:
        3  root
        1  admin
        1  test
        1  bob

[*] 1 IP(s) over threshold (5).

Usage

# Basic analysis
python3 loganalyze.py /var/log/auth.log

# Tune the detection window
python3 loganalyze.py auth.log --threshold 10 --window 5

# Web access log format
python3 loganalyze.py access.log --format access

# Compressed input
python3 loganalyze.py auth.log.1.gz

# JSON output for a SIEM pipeline
python3 loganalyze.py auth.log --json > findings.json

Example Output

$ python3 loganalyze.py sample-auth.log --threshold 5 --window 5

[*] 24,817 lines parsed  (312 auth failures, 1,204 successes)

TOP SOURCES BY FAILURE
SOURCE IP        FAILURES  USERS           FIRST SEEN            LAST SEEN
10.0.0.66             147  root, admin     2026-03-20T02:11:04   2026-03-20T02:13:58
10.0.0.41              38  admin           2026-03-20T04:02:11   2026-03-20T04:09:47

[!] BRUTE FORCE  10.0.0.66  147 failures in 174s
[!] SPRAYING     'admin' targeted from 14 distinct sources

[*] 2 findings

Error Handling

Condition	Exception	Response
Log file missing	`FileNotFoundError`	Report the path and exit 2
No read permission	`PermissionError`	Explain that `/var/log` usually needs root; exit 2
Malformed line	—	Skip it, increment a counter, and report the total at the end
Unparseable timestamp	`ValueError`	Keep the line but mark the timestamp unknown
Non-UTF-8 bytes	`UnicodeDecodeError`	Open with `errors="ignore"`
Truncated gzip file	`EOFError`	Report partial analysis rather than discarding everything
Empty file	—	Report zero events rather than crashing on an empty aggregate

Silently discarding malformed lines is the dangerous failure mode here — always report how many were skipped.

Security Considerations

[!warning] Authorized use only Analyse only synthetic logs or logs you are explicitly authorized to access.

Logs are personal data. Usernames, source addresses, and timestamps identify individuals in most jurisdictions. Handle output as confidential, do not paste it into issue trackers, and apply your organisation's retention policy.
Thresholds create false positives. A NAT gateway, VPN concentrator, or misconfigured service account will generate failure bursts without being an attack. Always corroborate before acting.
And false negatives. A patient attacker staying below your threshold, or spraying one password across many accounts, will not trip a per-IP failure count. That is why the per-username aggregation matters.
Never build a regex from untrusted input — a user-supplied pattern is a denial-of-service vector via catastrophic backtracking.
Log data itself is untrusted input. Attacker-controlled fields (usernames, user agents) can contain injection payloads; never interpolate them into a shell command or SQL query, and be careful rendering them in HTML reports.
This is detection, not enforcement. Auto-blocking on this signal alone will eventually lock out legitimate users.

Testing

import pytest

import loganalyze


AUTH_LINES = [
    "Mar 20 02:11:04 host sshd[1]: Failed password for root from 10.0.0.66 port 5 ssh2",
    "Mar 20 02:11:06 host sshd[2]: Failed password for root from 10.0.0.66 port 6 ssh2",
    "Mar 20 02:11:09 host sshd[3]: Accepted password for analyst from 10.0.0.9 port 7 ssh2",
    "this line is not parseable at all",
]


def test_counts_failures_and_successes():
    result = loganalyze.analyse(AUTH_LINES)
    assert result.failures == 2
    assert result.successes == 1
    assert result.skipped == 1


def test_aggregates_by_source():
    result = loganalyze.analyse(AUTH_LINES)
    assert result.by_source["10.0.0.66"] == 2


def test_threshold_flags_brute_force():
    lines = AUTH_LINES[:2] * 5
    findings = loganalyze.detect(loganalyze.analyse(lines), threshold=5, window_s=300)
    assert any(f.kind == "brute-force" for f in findings)


def test_malformed_lines_do_not_raise():
    loganalyze.analyse(["", "   ", "garbage"])       # must not raise

Use synthetic fixture lines — never commit real log data.

Extension Ideas

Time-windowing — parse timestamps and flag N failures within M minutes, not just per file.
GeoIP & reverse DNS — enrich offender IPs with country/ASN and PTR records.
Auto-block hook — emit iptables/fail2ban actions (with a dry-run default).
More formats — add Apache/Nginx combined, Windows Security EVTX (via python-evtx).
Streaming/tail mode — follow the file in real time and alert live.

Troubleshooting

Symptom	Likely cause	Fix
No events parsed	Pattern does not match your log format	Print a few raw lines; auth log formats differ between distributions
Very slow	Regex compiled inside the loop	Move `re.compile()` above the loop
Appears to hang	Catastrophic backtracking	Anchor the pattern and avoid nested quantifiers
Memory grows steadily	`readlines()` or a whole-file comprehension	Iterate the file handle directly
`ValueError` from `strptime`	Format string mismatch	Syslog omits the year; supply it or use a tolerant parser
Every source flagged	Threshold too low, or window not applied	Raise the threshold and confirm the time window filters correctly
`PermissionError` on `/var/log`	Insufficient privileges	Run as root, or copy a log to a readable location

References

[[File-Integrity-Monitor]]
[[Network-Inventory-Tool]]
[[Mini-Projects/Readme|Mini-Projects]] — module index
[[Readme|Python for Security Professionals]] — course home

Navigation Menu

Uh oh!

Log Analyzer

Project Overview

Objectives

Requirements

Architecture

Directory Structure

Implementation Plan

Complete Example Implementation

Output

Usage

Example Output

Error Handling

Security Considerations

Testing

Extension Ideas

Troubleshooting

References

Related

Uh oh!

FilesExpand file tree

Log-Analyzer.md

Latest commit

History

Log-Analyzer.md

File metadata and controls

Log Analyzer

Project Overview

Objectives

Requirements

Architecture

Directory Structure

Implementation Plan

Complete Example Implementation

Output

Usage

Example Output

Error Handling

Security Considerations

Testing

Extension Ideas

Troubleshooting

References

Related