""" Bot Analyzer - Select visitor records from a date range - Analyze which visits are likely non-human - Score each record with reasons - Export human / bot separated reports """ import os import re import csv import argparse import ipaddress import logging from datetime import datetime, date from collections import defaultdict from typing import Optional, List, Dict, Tuple # ─── Config ─────────────────────────────────────────────────────────────────── DB_CONFIG = { "host": "localhost", "user": "user", "password": "password", "database": "database", "charset": "utf8mb4", } TABLE_NAME = "wp_statpress" IP_COLUMN = "ip" DATE_COLUMN = "timestamp" UA_COLUMN = "agent" # set None if not available PATH_COLUMN = "urlrequested" # set None if not available BATCH_SIZE = 50_000 OUTPUT_DIR = "./bot_analysis" # Scoring thresholds BOT_SCORE_THRESHOLD = 5 # score >= this = bot SUSPICIOUS_THRESHOLD = 3 # score >= this = suspicious # IP frequency thresholds — scaled to time window HIGH_FREQ_PER_DAY = 100 # flag if IP hits > N times in a single day HIGH_FREQ_PER_HOUR = 30 # flag if IP hits > N times in a single hour (if datetime available) HIGH_FREQ_TOTAL_RATIO = 0.01 # flag if IP accounts for > 1% of total traffic in date range # ─── Logging ────────────────────────────────────────────────────────────────── logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", handlers=[ logging.FileHandler("bot_analyzer.log"), logging.StreamHandler() ] ) log = logging.getLogger(__name__) # ─── Scoring Rules ──────────────────────────────────────────────────────────── # Each entry: (score, reason, description) SCORING_RULES = { "private_ip": (10, "Private/Reserved IP", "IP is in private or reserved range"), "invalid_ip": (10, "Invalid IP", "IP address format is invalid"), "no_ua": (7, "No User Agent", "Empty or missing user agent"), "bot_ua": (8, "Bot User Agent", "User agent matches known bot/crawler pattern"), "tool_ua": (7, "Tool User Agent", "User agent matches automation tool (curl, wget, python...)"), "old_ie": (5, "Suspicious Old Browser", "IE6/IE7/IE8 + suspicious combination"), "attack_path": (9, "Attack Path", "Request path matches attack/scanner pattern"), "scanner_path": (8, "Scanner Path", "Request path matches vulnerability scanner"), "high_freq_day": (7, "High Freq Per Day", f"IP appears > {HIGH_FREQ_PER_DAY} times in a single day"), "high_freq_hour": (8, "High Freq Per Hour", f"IP appears > {HIGH_FREQ_PER_HOUR} times in a single hour"), "high_freq_ratio": (5, "High Traffic Ratio", f"IP accounts for > {HIGH_FREQ_TOTAL_RATIO*100:.1f}% of total traffic"), "empty_path": (3, "Empty Path", "Request path is empty or just /"), "suspicious_combo": (4, "Suspicious Combination", "Old UA + unusual path combination"), } # Known bot / crawler UA patterns BOT_UA_RE = re.compile( r"(?i)(" r"googlebot|bingbot|slurp|duckduckbot|baiduspider|yandexbot|" r"facebookexternalhit|twitterbot|linkedinbot|pinterest|" r"mj12bot|dotbot|rogerbot|semrushbot|ahrefsbot|" r"archive\.org_bot|ia_archiver|wayback|" r"petalbot|bytespider|gptbot|claudebot|anthropic|" r"crawler|spider|scraper|bot\b" r")" ) # Automation tool UA patterns TOOL_UA_RE = re.compile( r"(?i)(" r"curl|wget|python-requests|python-urllib|httpie|" r"go-http-client|java/|okhttp|axios|node-fetch|" r"libwww-perl|lwp-|ruby|perl|php/|" r"zgrab|masscan|nmap|nikto|sqlmap|nuclei|" r"dirbuster|gobuster|wfuzz|hydra|" r"headlesschrome|phantomjs|selenium|puppeteer|playwright|" r"postman|insomnia" r")" ) # Attack / scanner path patterns ATTACK_PATH_RE = re.compile( r"(?i)(" r"wp-login\.php|xmlrpc\.php|" r"\.env|\.git|\.svn|\.htaccess|\.htpasswd|" r"phpmyadmin|pma|adminer|" r"manager/html|solr/admin|jenkins|" r"actuator|/api/v1/pods|" r"/etc/passwd|/proc/self|/windows/win\.ini|" r"select\s+.+from|union\s+select|waitfor\s+delay|pg_sleep|" r"exec\s*\(|eval\s*\(|base64_decode|" r"\.\./|%2e%2e|%252e|\.\.%2f|" r"= 13 and ":" in s else None self.daily[ip][date_str] += 1 if hour_str: self.hourly[ip][hour_str] += 1 def max_daily(self, ip: str) -> Tuple[int, str]: """Return (max_count, date) for the busiest day.""" if ip not in self.daily or not self.daily[ip]: return 0, "" day, cnt = max(self.daily[ip].items(), key=lambda x: x[1]) return cnt, day def max_hourly(self, ip: str) -> Tuple[int, str]: """Return (max_count, hour) for the busiest hour.""" if ip not in self.hourly or not self.hourly[ip]: return 0, "" hour, cnt = max(self.hourly[ip].items(), key=lambda x: x[1]) return cnt, hour def traffic_ratio(self, ip: str) -> float: """Fraction of total traffic this IP accounts for.""" if self.grand_total == 0: return 0.0 return self.total[ip] / self.grand_total def freq_signals(self, ip: str) -> List[Tuple[str, str]]: """ Returns list of (rule_key, detail_string) for triggered freq signals. """ signals = [] max_day_cnt, max_day = self.max_daily(ip) max_hour_cnt, max_hour = self.max_hourly(ip) ratio = self.traffic_ratio(ip) if max_hour_cnt > HIGH_FREQ_PER_HOUR: signals.append(( "high_freq_hour", f"{max_hour_cnt} hits in {max_hour}" )) if max_day_cnt > HIGH_FREQ_PER_DAY: signals.append(( "high_freq_day", f"{max_day_cnt} hits on {max_day}" )) if ratio > HIGH_FREQ_TOTAL_RATIO: signals.append(( "high_freq_ratio", f"{ratio*100:.2f}% of total traffic" )) return signals def top(self, n: int = 20) -> List[Tuple[str, int, int, float]]: """Return top N IPs: (ip, total, max_daily, ratio%)""" result = [] for ip, total in sorted(self.total.items(), key=lambda x: -x[1])[:n]: max_day, _ = self.max_daily(ip) ratio = self.traffic_ratio(ip) * 100 result.append((ip, total, max_day, ratio)) return result # ─── Scorer ─────────────────────────────────────────────────────────────────── class BotScorer: """Score a single visit record and return reasons.""" def __init__(self, ip_freq: IPFrequencyTracker): self.ip_freq = ip_freq def is_private_ip(self, ip: str) -> bool: try: addr = ipaddress.ip_address(ip) return any(addr in net for net in PRIVATE_NETWORKS) except ValueError: return True def score( self, ip: str, ua: Optional[str] = None, path: Optional[str] = None ) -> Tuple[int, List[str]]: """ Returns (total_score, [list of reasons]) Higher score = more likely to be a bot. """ total = 0 reasons = [] # ── IP checks ──────────────────────────────────────────────────────── if not ip or ip.strip() in ("", "-", "unknown"): total += SCORING_RULES["invalid_ip"][0] reasons.append(SCORING_RULES["invalid_ip"][1]) elif self.is_private_ip(ip): total += SCORING_RULES["private_ip"][0] reasons.append(SCORING_RULES["private_ip"][1]) # Windowed frequency signals for rule_key, detail in self.ip_freq.freq_signals(ip): score_val, label, _ = SCORING_RULES[rule_key] total += score_val reasons.append(f"{label} ({detail})") # ── User Agent checks ──────────────────────────────────────────────── if ua is not None: ua_clean = (ua or "").strip() if not ua_clean or ua_clean == "-": total += SCORING_RULES["no_ua"][0] reasons.append(SCORING_RULES["no_ua"][1]) elif BOT_UA_RE.search(ua_clean): total += SCORING_RULES["bot_ua"][0] reasons.append(SCORING_RULES["bot_ua"][1]) elif TOOL_UA_RE.search(ua_clean): total += SCORING_RULES["tool_ua"][0] reasons.append(SCORING_RULES["tool_ua"][1]) elif re.search(r"(?i)MSIE [678]\.", ua_clean): # IE 6/7/8 is ancient — very suspicious in 2024+ total += SCORING_RULES["old_ie"][0] reasons.append(SCORING_RULES["old_ie"][1]) # ── Path checks ────────────────────────────────────────────────────── if path is not None: path_clean = (path or "").strip() if ATTACK_PATH_RE.search(path_clean): total += SCORING_RULES["attack_path"][0] reasons.append(SCORING_RULES["attack_path"][1]) if path_clean in ("", "/", "-"): total += SCORING_RULES["empty_path"][0] reasons.append(SCORING_RULES["empty_path"][1]) # ── Combination checks ─────────────────────────────────────────────── if ua is not None and path is not None: ua_clean = (ua or "").strip() path_clean = (path or "").strip() if re.search(r"(?i)MSIE", ua_clean) and ATTACK_PATH_RE.search(path_clean): total += SCORING_RULES["suspicious_combo"][0] reasons.append(SCORING_RULES["suspicious_combo"][1]) return total, reasons # ─── Database ───────────────────────────────────────────────────────────────── class Database: def __init__(self): import mysql.connector self.conn = mysql.connector.connect(**DB_CONFIG) self.cursor = self.conn.cursor(buffered=False) log.info("MySQL connected") def count_range(self, start: str, end: str) -> int: self.cursor.execute( f"SELECT COUNT(*) FROM {TABLE_NAME} " f"WHERE {DATE_COLUMN} BETWEEN %s AND %s", (start, end) ) return self.cursor.fetchone()[0] def fetch_batch(self, start: str, end: str, last_id: int) -> list: cols = ["id", IP_COLUMN, DATE_COLUMN] if UA_COLUMN: cols.append(UA_COLUMN) if PATH_COLUMN: cols.append(PATH_COLUMN) self.cursor.execute( f"SELECT {', '.join(cols)} FROM {TABLE_NAME} " f"WHERE {DATE_COLUMN} BETWEEN %s AND %s " f"AND id > %s " f"ORDER BY id ASC " f"LIMIT %s", (start, end, last_id, BATCH_SIZE) ) return self.cursor.fetchall() def close(self): self.cursor.close() self.conn.close() log.info("MySQL disconnected") # ─── Reporter ───────────────────────────────────────────────────────────────── class Reporter: def __init__(self, output_dir: str, date_range: str): os.makedirs(output_dir, exist_ok=True) ts = datetime.today().strftime("%Y%m%d_%H%M%S") self.human_path = os.path.join(output_dir, f"human_{date_range}_{ts}.csv") self.bot_path = os.path.join(output_dir, f"bot_{date_range}_{ts}.csv") self.suspicious_path = os.path.join(output_dir, f"suspicious_{date_range}_{ts}.csv") self.summary_path = os.path.join(output_dir, f"summary_{date_range}_{ts}.csv") cols_base = ["id", "ip", "date"] cols_ua = ["user_agent"] if UA_COLUMN else [] cols_path = ["path"] if PATH_COLUMN else [] cols_score = ["score", "reasons"] self.cols_human = cols_base + cols_ua + cols_path self.cols_bot = cols_base + cols_ua + cols_path + cols_score self._human_f = open(self.human_path, "w", newline="", encoding="utf-8") self._bot_f = open(self.bot_path, "w", newline="", encoding="utf-8") self._suspicious_f = open(self.suspicious_path, "w", newline="", encoding="utf-8") self._human_w = csv.writer(self._human_f) self._bot_w = csv.writer(self._bot_f) self._suspicious_w = csv.writer(self._suspicious_f) self._human_w.writerow(self.cols_human) self._bot_w.writerow(self.cols_bot) self._suspicious_w.writerow(self.cols_bot) self.stats = { "total": 0, "human": 0, "bot": 0, "suspicious": 0, } self.reason_counts: Dict[str, int] = defaultdict(int) def write(self, row_id, ip, visit_date, ua, path, score, reasons): self.stats["total"] += 1 base = [row_id, ip, visit_date] if UA_COLUMN: base.append(ua or "") if PATH_COLUMN: base.append(path or "") if score >= BOT_SCORE_THRESHOLD: self.stats["bot"] += 1 self._bot_w.writerow(base + [score, " | ".join(reasons)]) for r in reasons: self.reason_counts[r.split("(")[0].strip()] += 1 elif score >= SUSPICIOUS_THRESHOLD: self.stats["suspicious"] += 1 self._suspicious_w.writerow(base + [score, " | ".join(reasons)]) for r in reasons: self.reason_counts[r.split("(")[0].strip()] += 1 else: self.stats["human"] += 1 self._human_w.writerow(base) def close(self): self._human_f.close() self._bot_f.close() self._suspicious_f.close() def export_summary(self, ip_freq: IPFrequencyTracker): total = self.stats["total"] with open(self.summary_path, "w", newline="", encoding="utf-8") as f: writer = csv.writer(f) writer.writerow(["=== Overall Stats ==="]) writer.writerow(["Category", "Count", "Percentage"]) for key in ["human", "bot", "suspicious"]: pct = f"{self.stats[key]/total*100:.1f}%" if total else "0%" writer.writerow([key.capitalize(), self.stats[key], pct]) writer.writerow(["Total", total, "100%"]) writer.writerow([]) writer.writerow(["=== Bot Detection Reasons ==="]) writer.writerow(["Reason", "Count"]) for reason, count in sorted( self.reason_counts.items(), key=lambda x: -x[1] ): writer.writerow([reason, count]) writer.writerow([]) writer.writerow(["=== Top 20 High Frequency IPs ==="]) writer.writerow(["IP", "Total Hits", "Max Daily", "Traffic Ratio %", "Classification"]) for ip, total_hits, max_day, ratio in ip_freq.top(20): signals = ip_freq.freq_signals(ip) classification = "Bot" if signals else "Normal" writer.writerow([ip, total_hits, max_day, f"{ratio:.2f}%", classification]) log.info(f"Summary saved: {self.summary_path}") # ─── Console Output ─────────────────────────────────────────────────────────── def print_summary(stats: dict, reason_counts: dict, ip_freq: IPFrequencyTracker): total = stats["total"] print(f"\n{'═'*55}") print(" Bot Analysis Summary") print(f"{'═'*55}") print(f" {'Total records':<25} {total:>12,}") print(f" {'─'*52}") for key, label in [ ("human", "Human"), ("bot", "Bot"), ("suspicious", "Suspicious"), ]: count = stats[key] pct = f"{count/total*100:.1f}%" if total else "0%" bar = "█" * int(count/total*30) if total else "" print(f" {label:<25} {count:>12,} {pct:>6} {bar}") print(f"\n{'═'*55}") print(" Top Bot Detection Reasons") print(f"{'═'*55}") for reason, count in sorted(reason_counts.items(), key=lambda x: -x[1])[:10]: pct = f"{count/total*100:.1f}%" if total else "0%" print(f" {reason:<35} {count:>8,} {pct:>6}") print(f"\n{'═'*55}") print(f" Top 10 High Frequency IPs") print(f"{'═'*55}") print(f" {'IP':<20} {'Total':>8} {'Max/Day':>8} {'Ratio':>7} {'Status'}") print(f" {'─'*60}") for ip, total_hits, max_day, ratio in ip_freq.top(10): signals = ip_freq.freq_signals(ip) status = "Bot" if signals else "Normal" print(f" {ip:<20} {total_hits:>8,} {max_day:>8,} {ratio:>6.2f}% {status}") print() # ─── Main ───────────────────────────────────────────────────────────────────── def main(): global BOT_SCORE_THRESHOLD, HIGH_FREQ_PER_DAY, HIGH_FREQ_PER_HOUR, HIGH_FREQ_TOTAL_RATIO parser = argparse.ArgumentParser(description="Bot Analyzer — Detect non-human visits") parser.add_argument("--start", "-s", required=True, help="Start date YYYY-MM-DD") parser.add_argument("--end", "-e", required=True, help="End date YYYY-MM-DD") parser.add_argument("--output", "-o", default=OUTPUT_DIR, help="Output directory") parser.add_argument("--bot-threshold", type=int, default=BOT_SCORE_THRESHOLD, help=f"Bot score threshold (default: {BOT_SCORE_THRESHOLD})") parser.add_argument("--freq-per-day", type=int, default=HIGH_FREQ_PER_DAY, help=f"Max hits per day threshold (default: {HIGH_FREQ_PER_DAY})") parser.add_argument("--freq-per-hour", type=int, default=HIGH_FREQ_PER_HOUR, help=f"Max hits per hour threshold (default: {HIGH_FREQ_PER_HOUR})") parser.add_argument("--freq-ratio", type=float, default=HIGH_FREQ_TOTAL_RATIO, help=f"Max traffic ratio threshold (default: {HIGH_FREQ_TOTAL_RATIO})") args = parser.parse_args() BOT_SCORE_THRESHOLD = args.bot_threshold HIGH_FREQ_PER_DAY = args.freq_per_day HIGH_FREQ_PER_HOUR = args.freq_per_hour HIGH_FREQ_TOTAL_RATIO = args.freq_ratio start_date = args.start end_date = args.end date_range = f"{start_date}_to_{end_date}" log.info(f"Analyzing: {start_date} → {end_date}") db = Database() ip_freq = IPFrequencyTracker() reporter = Reporter(args.output, date_range) total = db.count_range(start_date, end_date) log.info(f"Total records in range: {total:,}") # ── Pass 1: Count IP frequencies ────────────────────────────────────────── log.info("Pass 1/2: Counting IP frequencies...") last_id = 0 processed = 0 while True: rows = db.fetch_batch(start_date, end_date, last_id) if not rows: break for row in rows: ip = str(row[1]).strip() if row[1] else "" visit_date = row[2] ip_freq.add(ip, visit_date) last_id = row[0] processed += len(rows) log.info(f"Pass 1: {processed:,}/{total:,} ({processed/total*100:.1f}%)") # ── Pass 2: Score each record ────────────────────────────────────────────── log.info("Pass 2/2: Scoring records...") scorer = BotScorer(ip_freq) last_id = 0 processed = 0 while True: rows = db.fetch_batch(start_date, end_date, last_id) if not rows: break for row in rows: idx = 0 row_id = row[idx]; idx += 1 ip = str(row[idx]).strip() if row[idx] else ""; idx += 1 visit_date = row[idx]; idx += 1 ua = str(row[idx]).strip() if UA_COLUMN and idx < len(row) else None if ua is not None: idx += 1 path = str(row[idx]).strip() if PATH_COLUMN and idx < len(row) else None score, reasons = scorer.score(ip, ua, path) reporter.write(row_id, ip, visit_date, ua, path, score, reasons) last_id = row_id processed += len(rows) log.info(f"Pass 2: {processed:,}/{total:,} ({processed/total*100:.1f}%)") # ── Finalize ────────────────────────────────────────────────────────────── reporter.close() reporter.export_summary(ip_freq) db.close() print_summary(reporter.stats, dict(reporter.reason_counts), ip_freq) print(f"{'═'*55}") print(" Output Files") print(f"{'═'*55}") print(f" Human : {reporter.human_path}") print(f" Bot : {reporter.bot_path}") print(f" Suspicious: {reporter.suspicious_path}") print(f" Summary : {reporter.summary_path}") print() if __name__ == "__main__": main()