parser for anubis log

parse_anubis.py

#!/usr/bin/env python3
"""
High-performance parser for Anubis WAF logs.
Designed for extreme speed and memory efficiency on huge log files.
"""

import sys
import os
import json
import time
from collections import Counter, deque
import argparse

# Attempt to load high-performance JSON parsers if available
try:
    import orjson as json_lib
    HAS_ORJSON = True
except ImportError:
    try:
        import ujson as json_lib
        HAS_ORJSON = False
    except ImportError:
        json_lib = json
        HAS_ORJSON = False


def format_size(bytes_size):
    """Formats bytes into human-readable sizes."""
    for unit in ['B', 'KB', 'MB', 'GB', 'TB']:
        if bytes_size < 1024.0:
            return f"{bytes_size:.2f} {unit}"
        bytes_size /= 1024.0
    return f"{bytes_size:.2f} PB"


class AnubisLogParser:
    def __init__(self, log_path, limit=10):
        self.log_path = log_path
        self.limit = limit
        
        # Aggregation structures
        self.total_lines = 0
        self.malformed_lines = 0
        self.blocked_count = 0
        self.challenges_count = 0
        
        # Counters for stats
        self.reasons_counter = Counter()  # check_result.name
        self.rules_counter = Counter()    # check_result.rule
        self.ip_counter = Counter()       # client ip address
        self.path_counter = Counter()     # request path
        self.ua_counter = Counter()       # user agent
        
        # Deque for last N blocks to keep memory bounded and O(1) operations
        self.recent_blocks = deque(maxlen=limit)

    def parse(self):
        """Streams the log file line by line with fast pre-filtering for maximum performance."""
        if not os.path.exists(self.log_path):
            raise FileNotFoundError(f"Log file not found: {self.log_path}")

        file_size = os.path.getsize(self.log_path)
        start_time = time.perf_counter()

        # Open file in streaming mode for memory safety
        with open(self.log_path, 'r', encoding='utf-8', errors='ignore') as f:
            for line in f:
                self.total_lines += 1
                
                # Fast pre-filtering: Check substring before executing JSON parser
                # This bypasses JSON deserialization overhead for 95%+ of lines in typical logs
                is_challenge = "new challenge issued" in line
                is_deny = "check_result" in line or '"rule":"DENY"' in line
                
                if not (is_challenge or is_deny):
                    continue

                try:
                    # Deserialize using the fastest available JSON library
                    if HAS_ORJSON:
                        entry = json_lib.loads(line.encode('utf-8'))
                    else:
                        entry = json_lib.loads(line)
                except Exception:
                    self.malformed_lines += 1
                    continue

                # Process challenge issues
                if entry.get("msg") == "new challenge issued":
                    self.challenges_count += 1
                    continue

                # Process blocks / rules checks
                check_result = entry.get("check_result")
                if check_result:
                    rule_action = check_result.get("rule")
                    
                    # Track rules and reasons
                    self.rules_counter[rule_action] += 1
                    
                    rule_name = check_result.get("name", "unknown")
                    self.reasons_counter[rule_name] += 1
                    
                    # We classify "DENY" as blocked
                    if rule_action == "DENY":
                        self.blocked_count += 1
                        
                        # Extract client IP
                        # x-forwarded-for can be a list or a single string; we prefer it, fallback to x-real-ip
                        xff = entry.get("x-forwarded-for")
                        real_ip = entry.get("x-real-ip", "unknown")
                        
                        client_ip = real_ip
                        if xff and xff.strip():
                            # Take the first IP if xff contains comma-separated values
                            client_ip = xff.split(',')[0].strip()
                        
                        self.ip_counter[client_ip] += 1
                        
                        # Track other parameters
                        path = entry.get("path", "/")
                        self.path_counter[path] += 1
                        
                        ua = entry.get("user_agent", "unknown")
                        self.ua_counter[ua] += 1
                        
                        # Append details for recent blocks list
                        self.recent_blocks.append({
                            "time": entry.get("time"),
                            "ip": client_ip,
                            "xff": xff,
                            "real_ip": real_ip,
                            "path": path,
                            "method": entry.get("method", "GET"),
                            "user_agent": ua,
                            "rule_name": rule_name,
                            "rule_action": rule_action,
                            "host": entry.get("host", "unknown")
                        })

        elapsed_time = time.perf_counter() - start_time
        return {
            "elapsed_time": elapsed_time,
            "file_size": file_size,
            "lines_per_second": self.total_lines / elapsed_time if elapsed_time > 0 else 0
        }

    def print_pretty(self, metrics):
        """Prints a highly polished, aesthetic summary report to the terminal."""
        # ANSI Colors
        C_BLUE = "\033[94m"
        C_CYAN = "\033[96m"
        C_GREEN = "\033[92m"
        C_YELLOW = "\033[93m"
        C_RED = "\033[91m"
        C_BOLD = "\033[1m"
        C_DIM = "\033[2m"
        C_END = "\033[0m"

        print(f"\n{C_BOLD}{C_CYAN}📊 ANUBIS LOG ANALYSIS REPORT{C_END}")
        print(f"{C_DIM}=" * 60 + f"{C_END}")
        print(f"{C_BOLD}Archivo:{C_END}      {self.log_path}")
        print(f"{C_BOLD}Tamaño:{C_END}       {format_size(metrics['file_size'])}")
        print(f"{C_BOLD}Registros:{C_END}    {self.total_lines:,} líneas leídas")
        print(f"{C_BOLD}Velocidad:{C_END}    {metrics['lines_per_second']:,.0f} líneas/seg ({metrics['elapsed_time']:.4f} segundos)")
        if self.malformed_lines > 0:
            print(f"{C_BOLD}Malformadas:{C_END}  {C_RED}{self.malformed_lines:,}{C_END}")
        print(f"{C_DIM}-" * 60 + f"{C_END}")

        # Metrics Summary
        print(f"{C_BOLD}📝 RESUMEN DE ACTIVIDAD WAF:{C_END}")
        block_pct = (self.blocked_count / self.total_lines * 100) if self.total_lines > 0 else 0
        print(f"  🛑 {C_BOLD}Total Bloqueados (DENY):{C_END}  {C_RED}{self.blocked_count:,}{C_END} ({block_pct:.2f}%)")
        print(f"  🛡️  {C_BOLD}Desafíos Emitidos (CAPTCHA):{C_END} {C_YELLOW}{self.challenges_count:,}{C_END}")
        print(f"{C_DIM}-" * 60 + f"{C_END}")

        # Reasons (Rules Matched)
        if self.reasons_counter:
            print(f"\n{C_BOLD}🔍 RAZONES DE BLOQUEO (check_result.name):{C_END}")
            for name, count in self.reasons_counter.most_common():
                pct = (count / self.blocked_count * 100) if self.blocked_count > 0 else 0
                print(f"  • {C_YELLOW}{name:<35}{C_END} => {C_BOLD}{count:>5,}{C_END} ({pct:.1f}%)")
        else:
            print(f"\n{C_BOLD}🔍 RAZONES DE BLOQUEO (check_result.name):{C_END} {C_DIM}Ninguno{C_END}")

        # Rules / Actions Breakdowns
        if self.rules_counter:
            print(f"\n{C_BOLD}⚙️  REGLAS / ACCIONES (check_result.rule):{C_END}")
            for rule, count in self.rules_counter.most_common():
                print(f"  • {C_GREEN}{rule:<15}{C_END} => {C_BOLD}{count:>5,}{C_END}")

        # Top IPs
        if self.ip_counter:
            print(f"\n{C_BOLD}🌐 TOP IPS BLOQUEADAS:{C_END}")
            for ip, count in self.ip_counter.most_common(10):
                pct = (count / self.blocked_count * 100) if self.blocked_count > 0 else 0
                print(f"  • {C_CYAN}{ip:<20}{C_END} => {C_BOLD}{count:>5,}{C_END} ({pct:.1f}%)")

        # Top Paths
        if self.path_counter:
            print(f"\n{C_BOLD}📂 TOP RUTAS BLOQUEADAS (Paths):{C_END}")
            for path, count in self.path_counter.most_common(5):
                print(f"  • {C_BLUE}{path:<40}{C_END} => {C_BOLD}{count:>5,}{C_END}")

        # Top User Agents
        if self.ua_counter:
            print(f"\n{C_BOLD}🤖 TOP USER AGENTS BLOQUEADOS:{C_END}")
            for ua, count in self.ua_counter.most_common(10):
                pct = (count / self.blocked_count * 100) if self.blocked_count > 0 else 0
                print(f"  • {C_YELLOW}{ua:<40}{C_END} => {C_BOLD}{count:>5,}{C_END} ({pct:.1f}%)")

        # Recent Blocks
        if self.recent_blocks:
            print(f"\n{C_BOLD}🚨 LISTADO DE ÚLTIMOS {len(self.recent_blocks)} BLOQUEOS:{C_END}")
            print(f"{C_DIM}" + "-" * 100 + f"{C_END}")
            print(f"{C_BOLD}{'TIMESTAMP':<25} {'IP ORIGEN':<18} {'MÉTODO':<8} {'RUTA':<25} {'REGLA':<20}{C_END}")
            print(f"{C_DIM}" + "-" * 100 + f"{C_END}")
            
            # Show in reverse order (newest first)
            for b in reversed(self.recent_blocks):
                # Clean timestamp (e.g. 2026-05-28T20:54:16Z)
                t_str = b['time'].split('.')[0] if b['time'] else 'unknown'
                t_str = t_str.replace('T', ' ')
                
                # Truncate path if too long
                path_str = b['path']
                if len(path_str) > 23:
                    path_str = path_str[:20] + "..."
                
                print(f"{C_DIM}{t_str:<25}{C_END} "
                      f"{C_CYAN}{b['ip']:<18}{C_END} "
                      f"{C_GREEN}{b['method']:<8}{C_END} "
                      f"{C_BLUE}{path_str:<25}{C_END} "
                      f"{C_RED}{b['rule_name']:<20}{C_END}")
            print(f"{C_DIM}" + "-" * 100 + f"{C_END}")
        else:
            print(f"\n{C_BOLD}🚨 LISTADO DE ÚLTIMOS BLOQUEOS:{C_END} {C_DIM}Ninguno{C_END}")
        print()

    def get_json_output(self, metrics):
        """Returns the complete analytics in a machine-readable JSON structure."""
        return {
            "summary": {
                "log_file": self.log_path,
                "file_size_bytes": metrics["file_size"],
                "total_lines": self.total_lines,
                "malformed_lines": self.malformed_lines,
                "blocked_count": self.blocked_count,
                "challenges_count": self.challenges_count,
                "elapsed_time_seconds": metrics["elapsed_time"],
                "lines_per_second": metrics["lines_per_second"]
            },
            "reasons": dict(self.reasons_counter),
            "rules": dict(self.rules_counter),
            "top_ips": dict(self.ip_counter.most_common(20)),
            "top_paths": dict(self.path_counter.most_common(20)),
            "top_user_agents": dict(self.ua_counter.most_common(20)),
            "recent_blocks": list(reversed(self.recent_blocks))
        }


def main():
    parser = argparse.ArgumentParser(
        description="Script de alto rendimiento para parsear logs de Anubis WAF."
    )
    parser.add_argument(
        "log_path", 
        nargs="?", 
        default="/datos/anubis/anubis.log",
        help="Ruta al archivo de log de Anubis (por defecto: /datos/anubis/anubis.log)"
    )
    parser.add_argument(
        "-l", "--limit", 
        type=int, 
        default=10,
        help="Número de últimos bloqueos a listar (por defecto: 10)"
    )
    parser.add_argument(
        "-j", "--json", 
        action="store_true",
        help="Imprimir los resultados en formato JSON para consumo automático"
    )

    args = parser.parse_args()

    parser_instance = AnubisLogParser(args.log_path, args.limit)
    
    try:
        metrics = parser_instance.parse()
        if args.json:
            print(json.dumps(parser_instance.get_json_output(metrics), indent=2))
        else:
            parser_instance.print_pretty(metrics)
    except FileNotFoundError as e:
        print(f"❌ Error: {e}", file=sys.stderr)
        sys.exit(1)
    except KeyboardInterrupt:
        print("\n⚠️ Proceso interrumpido por el usuario.", file=sys.stderr)
        sys.exit(130)


if __name__ == "__main__":
    main()

Sample output

📊 ANUBIS LOG ANALYSIS REPORT

Archivo: umaza.log
Tamaño: 7.56 MB
Registros: 12,571 líneas leídas
Velocidad: 73,590 líneas/seg (0.1708 segundos)

📝 RESUMEN DE ACTIVIDAD WAF:
🛑 Total Bloqueados (DENY): 10,679 (84.95%)
🛡 Desafíos Emitidos (CAPTCHA): 1,878

🔍 RAZONES DE BLOQUEO (check_result.name):
• bot/block-ai-scrapers => 10,679 (100.0%)
• bot/gene