wp-statpress-analyze/bot.py

"""
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"<script|javascript:|onerror=|onload=|"
    r"c99\.php|r57\.php|webshell|shell\.php|cmd\.php|"
    r"\.bak|\.sql|\.zip|\.tar\.gz|config\.php"
    r")"
)

# Private / reserved IP ranges
PRIVATE_NETWORKS = [
    ipaddress.ip_network("10.0.0.0/8"),
    ipaddress.ip_network("172.16.0.0/12"),
    ipaddress.ip_network("192.168.0.0/16"),
    ipaddress.ip_network("127.0.0.0/8"),
    ipaddress.ip_network("169.254.0.0/16"),
    ipaddress.ip_network("100.64.0.0/10"),
    ipaddress.ip_network("0.0.0.0/8"),
    ipaddress.ip_network("240.0.0.0/4"),
    ipaddress.ip_network("::1/128"),
    ipaddress.ip_network("fc00::/7"),
    ipaddress.ip_network("fe80::/10"),
]

# ─── IP Frequency Tracker ─────────────────────────────────────────────────────

class IPFrequencyTracker:
    """
    Track IP frequency at multiple time windows:
    - Per day:   ip → {date → count}
    - Per hour:  ip → {date_hour → count}
    - Total:     ip → total count across range
    """

    def __init__(self):
        # ip → {YYYY-MM-DD → count}
        self.daily:  Dict[str, Dict[str, int]] = defaultdict(lambda: defaultdict(int))
        # ip → {YYYY-MM-DD HH → count}
        self.hourly: Dict[str, Dict[str, int]] = defaultdict(lambda: defaultdict(int))
        # ip → total hits
        self.total:  Dict[str, int]            = defaultdict(int)
        self.grand_total: int                  = 0

    def add(self, ip: str, visit_date):
        """
        visit_date can be:
        - datetime object  → use date + hour
        - date object      → use date only
        - string YYYY-MM-DD or YYYYMMDD
        """
        self.total[ip]   += 1
        self.grand_total += 1

        # Extract date string
        if hasattr(visit_date, "strftime"):
            date_str = visit_date.strftime("%Y-%m-%d")
            hour_str = visit_date.strftime("%Y-%m-%d %H") \
                       if hasattr(visit_date, "hour") else None
        else:
            s        = str(visit_date).strip()
            date_str = s[:10] if "-" in s else f"{s[:4]}-{s[4:6]}-{s[6:8]}"
            # Try to extract hour if datetime string
            hour_str = s[:13] if len(s) >= 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()