docs-alpha.jayfield.org/scripts/alpha/attacker-check.py

#!/usr/bin/env python3
"""alpha.jayfield.org weekly attacker summary.

Parses fail2ban.log (current + recent rotations, gzipped or not) for the
past 7 days and produces two append-only CSVs for longitudinal analysis,
plus a human-readable summary emailed to root@jayfield.org and stored as
a dated report file on disk.

Run weekly via cron (/etc/cron.d/attacker-check). Safe to also run
manually at any time: sudo /usr/local/sbin/attacker-check.py
"""
import csv
import gzip
import re
import statistics
import subprocess
import sys
from collections import defaultdict
from datetime import datetime, timedelta
from pathlib import Path

# current + enough rotations to cover a week even if cron and logrotate's
# midnight timer land oddly relative to each other
LOG_FILES = [
	"/var/log/fail2ban.log",
	"/var/log/fail2ban.log.1",
	"/var/log/fail2ban.log.2.gz",
]
DATA_DIR = Path("/var/log/attacker-check")
PER_IP_CSV = DATA_DIR / "per_ip_weekly.csv"
PER_JAIL_CSV = DATA_DIR / "per_jail_weekly.csv"
REPORTS_DIR = DATA_DIR / "reports"
REPORT_EMAIL = "root@jayfield.org"

# distributed-scan detection thresholds: a /24 needs at least this many
# distinct IPs, each averaging no more than this many events, to be
# flagged as a likely coordinated single-shot-per-IP scan
CAMPAIGN_MIN_IPS = 5
CAMPAIGN_MAX_AVG_EVENTS = 3.0

# "regular timing" flag: repeat hits on the same IP/jail with a
# coefficient of variation below this are suspiciously scripted rather
# than organic retry jitter (this is exactly the low-and-slow evasion
# pattern found twice this engagement - ~22-33min spacing to just miss
# fail2ban's findtime window)
REGULAR_TIMING_MAX_CV = 0.15
REGULAR_TIMING_MIN_EVENTS = 3

LINE_RE = re.compile(
	r'^(?P<ts>\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}),\d+\s+\S+\s+\[\d+\]:\s+'
	r'\S+\s+\[(?P<jail>[\w-]+)\]\s+(?P<action>Found|Ban|Unban)\s+'
	r'(?P<ip>[0-9a-fA-F:.]+)'
)


def open_maybe_gzip(path: Path):
	if path.suffix == ".gz":
		return gzip.open(path, "rt", errors="replace")
	return path.open("r", errors="replace")


def parse_events(since: datetime):
	"""Returns list of (timestamp, jail, action, ip) for events >= since."""
	events = []
	for name in LOG_FILES:
		path = Path(name)
		if not path.exists():
			continue
		with open_maybe_gzip(path) as f:
			for line in f:
				m = LINE_RE.match(line)
				if not m:
					continue
				if "Restore Ban" in line:
					continue  # persisted state on restart, not a new event
				ts = datetime.strptime(m.group("ts"), "%Y-%m-%d %H:%M:%S")
				if ts < since:
					continue
				events.append((ts, m.group("jail"), m.group("action"), m.group("ip")))
	return events


def subnet24(ip: str) -> str:
	parts = ip.split(".")
	if len(parts) == 4:
		return ".".join(parts[:3]) + ".0/24"
	return ip  # IPv6 or malformed - leave as-is, not bucketed


def write_csv(path: Path, rows: list, fieldnames: list):
	if not rows:
		return
	write_header = not path.exists()
	with path.open("a", newline="") as f:
		writer = csv.DictWriter(f, fieldnames=fieldnames)
		if write_header:
			writer.writeheader()
		writer.writerows(rows)


def main():
	DATA_DIR.mkdir(parents=True, exist_ok=True)
	REPORTS_DIR.mkdir(parents=True, exist_ok=True)
	now = datetime.now()
	since = now - timedelta(days=7)
	week_ending = now.date().isoformat()

	events = parse_events(since)

	found_by_key = defaultdict(list)  # (jail, ip) -> [timestamps], sorted later
	ban_counts = defaultdict(int)     # jail -> new-ban count this week
	for ts, jail, action, ip in events:
		if action == "Found":
			found_by_key[(jail, ip)].append(ts)
		elif action == "Ban":
			ban_counts[jail] += 1

	per_ip_rows = []
	subnet_ips = defaultdict(set)  # (jail, subnet) -> {ip, ...}
	for (jail, ip), timestamps in found_by_key.items():
		timestamps.sort()
		intervals = [
			(timestamps[i + 1] - timestamps[i]).total_seconds()
			for i in range(len(timestamps) - 1)
		]
		mean_iv = statistics.mean(intervals) if intervals else 0.0
		stdev_iv = statistics.pstdev(intervals) if len(intervals) > 1 else 0.0
		cv = (stdev_iv / mean_iv) if mean_iv > 0 else None
		regular_timing = (
			len(timestamps) >= REGULAR_TIMING_MIN_EVENTS
			and cv is not None
			and cv < REGULAR_TIMING_MAX_CV
		)
		sn = subnet24(ip)
		subnet_ips[(jail, sn)].add(ip)
		per_ip_rows.append({
			"week_ending": week_ending,
			"jail": jail,
			"ip": ip,
			"subnet_24": sn,
			"event_count": len(timestamps),
			"first_seen": timestamps[0].isoformat(),
			"last_seen": timestamps[-1].isoformat(),
			"mean_interval_s": round(mean_iv, 1),
			"stdev_interval_s": round(stdev_iv, 1),
			"regular_timing_flag": regular_timing,
		})

	jail_totals = defaultdict(lambda: {"events": 0, "ips": set(), "subnets": set()})
	for (jail, ip), timestamps in found_by_key.items():
		jail_totals[jail]["events"] += len(timestamps)
		jail_totals[jail]["ips"].add(ip)
		jail_totals[jail]["subnets"].add(subnet24(ip))

	per_jail_rows = []
	for jail, data in jail_totals.items():
		per_jail_rows.append({
			"week_ending": week_ending,
			"jail": jail,
			"total_events": data["events"],
			"distinct_ips": len(data["ips"]),
			"distinct_24_subnets": len(data["subnets"]),
			"bans_this_week": ban_counts.get(jail, 0),
		})

	write_csv(PER_IP_CSV, per_ip_rows, [
		"week_ending", "jail", "ip", "subnet_24", "event_count",
		"first_seen", "last_seen", "mean_interval_s", "stdev_interval_s",
		"regular_timing_flag",
	])
	write_csv(PER_JAIL_CSV, per_jail_rows, [
		"week_ending", "jail", "total_events", "distinct_ips",
		"distinct_24_subnets", "bans_this_week",
	])

	# distributed single-shot-per-IP scan detection (same shape as the
	# apache-noscript campaign found 2026-07-19: many IPs, ~1 hit each)
	campaigns = []
	for (jail, sn), ips in subnet_ips.items():
		if len(ips) < CAMPAIGN_MIN_IPS:
			continue
		avg_events = sum(len(found_by_key[(jail, ip)]) for ip in ips) / len(ips)
		if avg_events <= CAMPAIGN_MAX_AVG_EVENTS:
			campaigns.append((jail, sn, len(ips), round(avg_events, 1)))

	lines = []
	lines.append(f"alpha.jayfield.org attacker summary - week ending {week_ending}")
	lines.append(f"Window: {since.isoformat()} .. {now.isoformat()}")
	lines.append("")

	lines.append("Per-jail totals:")
	if per_jail_rows:
		for row in sorted(per_jail_rows, key=lambda r: -r["total_events"]):
			lines.append(
				f"  {row['jail']:<20} events={row['total_events']:<6} "
				f"ips={row['distinct_ips']:<5} subnets={row['distinct_24_subnets']:<5} "
				f"new_bans={row['bans_this_week']}"
			)
	else:
		lines.append("  (no fail2ban activity this week)")
	lines.append("")

	top_ips = sorted(per_ip_rows, key=lambda r: -r["event_count"])[:15]
	if top_ips:
		lines.append("Top attacking IPs this week:")
		for row in top_ips:
			flag = " [REGULAR TIMING]" if row["regular_timing_flag"] else ""
			lines.append(
				f"  {row['ip']:<16} jail={row['jail']:<18} events={row['event_count']:<5} "
				f"mean_interval={row['mean_interval_s']}s{flag}"
			)
		lines.append("")

	if campaigns:
		lines.append(
			f"Possible distributed scan campaigns (>={CAMPAIGN_MIN_IPS} distinct IPs "
			f"in one /24, <={CAMPAIGN_MAX_AVG_EVENTS} avg events/IP):"
		)
		for jail, sn, n_ips, avg_ev in sorted(campaigns, key=lambda c: -c[2]):
			lines.append(f"  {jail}: {sn} - {n_ips} distinct IPs, avg {avg_ev} events/IP")
	else:
		lines.append("No distributed-scan campaign pattern detected this week.")
	lines.append("")

	regular_ips = [r for r in per_ip_rows if r["regular_timing_flag"]]
	if regular_ips:
		lines.append(
			"IPs with suspiciously regular timing (possible scripted "
			"low-and-slow evasion, e.g. spaced just past a jail's findtime):"
		)
		for row in sorted(regular_ips, key=lambda r: -r["event_count"]):
			lines.append(
				f"  {row['ip']:<16} jail={row['jail']:<18} events={row['event_count']:<5} "
				f"mean_interval={row['mean_interval_s']}s stdev={row['stdev_interval_s']}s"
			)
		lines.append("")

	report_path = REPORTS_DIR / f"{week_ending}.txt"
	lines.append(f"Data appended to {PER_IP_CSV.name} and {PER_JAIL_CSV.name} in {DATA_DIR}/ for trend analysis.")
	lines.append(f"This report saved to {report_path}.")

	report = "\n".join(lines)
	print(report)

	report_path.write_text(report + "\n")

	try:
		subprocess.run(
			["mail", "-s", f"alpha.jayfield.org attacker summary - {week_ending}", REPORT_EMAIL],
			input=report, text=True, check=True,
		)
	except Exception as e:
		print(f"WARNING: failed to send email: {e}", file=sys.stderr)


if __name__ == "__main__":
	main()