shadowhunter_logger.py

#!/usr/bin/env python3
"""
ShadowHunter - Dark Web Credential Intelligence Platform
Module: Winston-Style Logger with Multi-Level Logging
Author: Fevra
Version: 0.1.0

Provides enterprise-grade logging with:
- Multiple transport targets (console, file, JSON)
- Colored console output with severity-based formatting
- JSON structured logging for SIEM integration
- Log rotation and retention policies
- Performance metrics and audit trails
"""

import os
import sys
import json
import logging
import traceback
from datetime import datetime, timezone
from pathlib import Path
from typing import Optional, Dict, Any, List
from dataclasses import dataclass, field, asdict
from enum import Enum
from logging.handlers import RotatingFileHandler, TimedRotatingFileHandler
import threading

# ============================================================================
# CONFIGURATION & CONSTANTS
# ============================================================================

# ANSI Color codes for terminal output
class Colors:
    """ANSI escape codes for colored terminal output."""
    RESET = "\033[0m"
    BOLD = "\033[1m"
    DIM = "\033[2m"
    
    # Foreground colors
    BLACK = "\033[30m"
    RED = "\033[31m"
    GREEN = "\033[32m"
    YELLOW = "\033[33m"
    BLUE = "\033[34m"
    MAGENTA = "\033[35m"
    CYAN = "\033[36m"
    WHITE = "\033[37m"
    
    # Bright foreground colors
    BRIGHT_RED = "\033[91m"
    BRIGHT_GREEN = "\033[92m"
    BRIGHT_YELLOW = "\033[93m"
    BRIGHT_BLUE = "\033[94m"
    BRIGHT_MAGENTA = "\033[95m"
    BRIGHT_CYAN = "\033[96m"
    
    # Background colors
    BG_RED = "\033[41m"
    BG_YELLOW = "\033[43m"
    BG_BLUE = "\033[44m"


class LogLevel(str, Enum):
    """Log level enumeration matching Winston levels."""
    ERROR = "error"      # Critical errors requiring immediate attention
    WARN = "warn"        # Warning conditions
    INFO = "info"        # Informational messages
    HTTP = "http"        # HTTP request/response logging
    VERBOSE = "verbose"  # Verbose operational information
    DEBUG = "debug"      # Debug-level messages
    SILLY = "silly"      # Extremely detailed tracing


# Map Winston-style levels to Python logging levels
LEVEL_MAP = {
    LogLevel.ERROR: logging.ERROR,
    LogLevel.WARN: logging.WARNING,
    LogLevel.INFO: logging.INFO,
    LogLevel.HTTP: 25,  # Custom level between INFO and DEBUG
    LogLevel.VERBOSE: 15,  # Custom level between DEBUG and INFO
    LogLevel.DEBUG: logging.DEBUG,
    LogLevel.SILLY: 5,  # Lowest level for trace-level logging
}

# Register custom logging levels
logging.addLevelName(25, "HTTP")
logging.addLevelName(15, "VERBOSE")
logging.addLevelName(5, "SILLY")


# ============================================================================
# DATA MODELS
# ============================================================================

@dataclass
class LogEntry:
    """
    Structured log entry for consistent logging format.
    
    Attributes:
        level: Log severity level
        message: Human-readable log message
        timestamp: ISO 8601 formatted timestamp
        module: Source module name
        metadata: Additional contextual data
    """
    level: str
    message: str
    timestamp: str = ""
    module: str = "ShadowHunter"
    metadata: Dict[str, Any] = field(default_factory=dict)
    
    def __post_init__(self):
        """Set timestamp if not provided."""
        if not self.timestamp:
            self.timestamp = datetime.now(timezone.utc).isoformat()
    
    def to_dict(self) -> Dict[str, Any]:
        """Convert log entry to dictionary for JSON serialization."""
        return {
            "timestamp": self.timestamp,
            "level": self.level,
            "module": self.module,
            "message": self.message,
            **self.metadata
        }
    
    def to_json(self) -> str:
        """Serialize log entry to JSON string."""
        return json.dumps(self.to_dict(), default=str)


# ============================================================================
# CUSTOM FORMATTERS
# ============================================================================

class ColoredFormatter(logging.Formatter):
    """
    Custom formatter with colored output for console display.
    
    Provides visual distinction between log levels using ANSI colors
    and includes emojis for quick severity recognition.
    """
    
    # Level-specific formatting configuration
    LEVEL_FORMATS = {
        logging.ERROR: {
            "color": Colors.BRIGHT_RED,
            "icon": "🔴",
            "label": "ERROR"
        },
        logging.WARNING: {
            "color": Colors.BRIGHT_YELLOW,
            "icon": "🟠",
            "label": "WARN "
        },
        logging.INFO: {
            "color": Colors.BRIGHT_GREEN,
            "icon": "🟢",
            "label": "INFO "
        },
        25: {  # HTTP level
            "color": Colors.BRIGHT_CYAN,
            "icon": "🔵",
            "label": "HTTP "
        },
        15: {  # VERBOSE level
            "color": Colors.BRIGHT_BLUE,
            "icon": "💬",
            "label": "VERB "
        },
        logging.DEBUG: {
            "color": Colors.BRIGHT_MAGENTA,
            "icon": "🐛",
            "label": "DEBUG"
        },
        5: {  # SILLY level
            "color": Colors.DIM + Colors.WHITE,
            "icon": "🔍",
            "label": "SILLY"
        },
    }
    
    def format(self, record: logging.LogRecord) -> str:
        """Format log record with colors and icons."""
        # Get level-specific formatting
        level_config = self.LEVEL_FORMATS.get(record.levelno, {
            "color": Colors.WHITE,
            "icon": "⚪",
            "label": record.levelname[:5].ljust(5)
        })
        
        # Format timestamp
        timestamp = datetime.now().strftime("%H:%M:%S")
        
        # Build formatted message
        color = level_config["color"]
        icon = level_config["icon"]
        label = level_config["label"]
        module = getattr(record, 'module_name', record.name)
        
        # Base formatted message
        formatted = (
            f"{Colors.DIM}{timestamp}{Colors.RESET} "
            f"{icon} {color}{Colors.BOLD}{label}{Colors.RESET} "
            f"{Colors.CYAN}[{module}]{Colors.RESET} "
            f"{record.getMessage()}"
        )
        
        # Add metadata if present
        if hasattr(record, 'metadata') and record.metadata:
            meta_str = json.dumps(record.metadata, indent=2, default=str)
            formatted += f"\n{Colors.DIM}{meta_str}{Colors.RESET}"
        
        # Add exception info if present
        if record.exc_info:
            formatted += f"\n{Colors.RED}{self.formatException(record.exc_info)}{Colors.RESET}"
        
        return formatted


class JSONFormatter(logging.Formatter):
    """
    JSON formatter for structured logging.
    
    Outputs logs in JSON format for easy parsing by log aggregators,
    SIEM systems, and analytics pipelines.
    """
    
    def format(self, record: logging.LogRecord) -> str:
        """Format log record as JSON."""
        log_entry = {
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "level": record.levelname.lower(),
            "module": getattr(record, 'module_name', record.name),
            "message": record.getMessage(),
            "logger": record.name,
            "path": record.pathname,
            "line": record.lineno,
            "function": record.funcName,
        }
        
        # Add metadata if present
        if hasattr(record, 'metadata') and record.metadata:
            log_entry["metadata"] = record.metadata
        
        # Add exception info if present
        if record.exc_info:
            log_entry["exception"] = {
                "type": record.exc_info[0].__name__ if record.exc_info[0] else None,
                "message": str(record.exc_info[1]) if record.exc_info[1] else None,
                "traceback": self.formatException(record.exc_info)
            }
        
        return json.dumps(log_entry, default=str)


# ============================================================================
# SHADOWHUNTER LOGGER
# ============================================================================

class ShadowHunterLogger:
    """
    Winston-style logger for ShadowHunter platform.
    
    Provides multi-transport logging with:
    - Colored console output
    - Rotating file logs
    - JSON structured logs for SIEM integration
    - Performance metrics tracking
    - Audit trail capabilities
    
    Usage:
        logger = ShadowHunterLogger("ModuleName")
        logger.info("Operation completed", {"duration": 1.5, "items": 100})
        logger.error("Operation failed", exc_info=True)
    """
    
    # Singleton instances cache
    _instances: Dict[str, 'ShadowHunterLogger'] = {}
    _lock = threading.Lock()
    
    # Default log directory
    DEFAULT_LOG_DIR = Path("logs")
    
    def __new__(cls, module_name: str = "ShadowHunter", **kwargs):
        """
        Implement singleton pattern per module name.
        
        Ensures only one logger instance exists per module to prevent
        duplicate handlers and log entries.
        """
        with cls._lock:
            if module_name not in cls._instances:
                instance = super().__new__(cls)
                cls._instances[module_name] = instance
            return cls._instances[module_name]
    
    def __init__(
        self,
        module_name: str = "ShadowHunter",
        log_level: LogLevel = LogLevel.INFO,
        log_dir: Optional[Path] = None,
        enable_console: bool = True,
        enable_file: bool = True,
        enable_json: bool = True,
        max_file_size: int = 10 * 1024 * 1024,  # 10 MB
        backup_count: int = 5
    ):
        """
        Initialize ShadowHunter logger.
        
        Args:
            module_name: Name of the module using this logger
            log_level: Minimum log level to capture
            log_dir: Directory for log files (defaults to ./logs)
            enable_console: Enable colored console output
            enable_file: Enable rotating file logging
            enable_json: Enable JSON structured logging
            max_file_size: Maximum size of each log file before rotation
            backup_count: Number of backup log files to retain
        """
        # Prevent re-initialization of existing instances
        if hasattr(self, '_initialized') and self._initialized:
            return
        
        self.module_name = module_name
        self.log_level = log_level
        self.log_dir = log_dir or self.DEFAULT_LOG_DIR
        
        # Create log directory if it doesn't exist
        self.log_dir.mkdir(parents=True, exist_ok=True)
        
        # Create base logger
        self.logger = logging.getLogger(f"ShadowHunter.{module_name}")
        self.logger.setLevel(LEVEL_MAP.get(log_level, logging.INFO))
        self.logger.propagate = False  # Prevent duplicate logs
        
        # Clear existing handlers to prevent duplicates on re-init
        self.logger.handlers = []
        
        # Configure transports
        if enable_console:
            self._add_console_handler()
        
        if enable_file:
            self._add_file_handler(max_file_size, backup_count)
        
        if enable_json:
            self._add_json_handler(max_file_size, backup_count)
        
        self._initialized = True
        
        # Log initialization
        self.debug(f"Logger initialized for module: {module_name}", {
            "log_level": log_level.value,
            "log_dir": str(self.log_dir),
            "transports": {
                "console": enable_console,
                "file": enable_file,
                "json": enable_json
            }
        })
    
    def _add_console_handler(self):
        """Add colored console handler."""
        console_handler = logging.StreamHandler(sys.stdout)
        console_handler.setFormatter(ColoredFormatter())
        console_handler.setLevel(logging.DEBUG)
        self.logger.addHandler(console_handler)
    
    def _add_file_handler(self, max_size: int, backup_count: int):
        """Add rotating file handler for human-readable logs."""
        log_file = self.log_dir / f"shadowhunter.log"
        file_handler = RotatingFileHandler(
            log_file,
            maxBytes=max_size,
            backupCount=backup_count,
            encoding='utf-8'
        )
        file_handler.setFormatter(logging.Formatter(
            '%(asctime)s | %(levelname)-8s | %(name)s | %(message)s',
            datefmt='%Y-%m-%d %H:%M:%S'
        ))
        file_handler.setLevel(logging.DEBUG)
        self.logger.addHandler(file_handler)
    
    def _add_json_handler(self, max_size: int, backup_count: int):
        """Add rotating JSON handler for structured logs."""
        json_file = self.log_dir / f"shadowhunter.json.log"
        json_handler = RotatingFileHandler(
            json_file,
            maxBytes=max_size,
            backupCount=backup_count,
            encoding='utf-8'
        )
        json_handler.setFormatter(JSONFormatter())
        json_handler.setLevel(logging.DEBUG)
        self.logger.addHandler(json_handler)
    
    def _log(
        self,
        level: int,
        message: str,
        metadata: Optional[Dict[str, Any]] = None,
        exc_info: bool = False
    ):
        """
        Internal method to log messages with metadata.
        
        Args:
            level: Python logging level
            message: Log message
            metadata: Additional context data
            exc_info: Include exception traceback
        """
        # Create log record with metadata
        extra = {
            'module_name': self.module_name,
            'metadata': metadata or {}
        }
        
        self.logger.log(level, message, extra=extra, exc_info=exc_info)
    
    # ========================================================================
    # PUBLIC LOGGING METHODS
    # ========================================================================
    
    def error(
        self,
        message: str,
        metadata: Optional[Dict[str, Any]] = None,
        exc_info: bool = False
    ):
        """
        Log error level message.
        
        Use for critical errors requiring immediate attention.
        Examples: Database connection failures, API errors, security breaches.
        
        Args:
            message: Error description
            metadata: Additional context (error codes, affected resources)
            exc_info: Include exception traceback
        """
        self._log(logging.ERROR, message, metadata, exc_info)
    
    def warn(
        self,
        message: str,
        metadata: Optional[Dict[str, Any]] = None
    ):
        """
        Log warning level message.
        
        Use for warning conditions that should be monitored.
        Examples: Deprecated API usage, rate limiting warnings, config issues.
        
        Args:
            message: Warning description
            metadata: Additional context
        """
        self._log(logging.WARNING, message, metadata)
    
    def info(
        self,
        message: str,
        metadata: Optional[Dict[str, Any]] = None
    ):
        """
        Log info level message.
        
        Use for general operational information.
        Examples: Scan started/completed, new threat detected, connection established.
        
        Args:
            message: Informational message
            metadata: Additional context
        """
        self._log(logging.INFO, message, metadata)
    
    def http(
        self,
        message: str,
        metadata: Optional[Dict[str, Any]] = None
    ):
        """
        Log HTTP request/response.
        
        Use for API request logging with method, URL, status, duration.
        
        Args:
            message: HTTP request summary
            metadata: Request/response details
        """
        self._log(25, message, metadata)
    
    def verbose(
        self,
        message: str,
        metadata: Optional[Dict[str, Any]] = None
    ):
        """
        Log verbose operational information.
        
        Use for detailed operational info during normal execution.
        Examples: Configuration loaded, cache hit/miss, retry attempts.
        
        Args:
            message: Verbose message
            metadata: Additional context
        """
        self._log(15, message, metadata)
    
    def debug(
        self,
        message: str,
        metadata: Optional[Dict[str, Any]] = None
    ):
        """
        Log debug level message.
        
        Use for debugging information during development.
        Examples: Variable values, function entry/exit, decision branches.
        
        Args:
            message: Debug message
            metadata: Additional context
        """
        self._log(logging.DEBUG, message, metadata)
    
    def silly(
        self,
        message: str,
        metadata: Optional[Dict[str, Any]] = None
    ):
        """
        Log silly/trace level message.
        
        Use for extremely detailed tracing during development.
        Examples: Loop iterations, byte-level data, timing microseconds.
        
        Args:
            message: Trace message
            metadata: Additional context
        """
        self._log(5, message, metadata)
    
    # ========================================================================
    # SPECIALIZED LOGGING METHODS
    # ========================================================================
    
    def threat_detected(
        self,
        threat_type: str,
        severity: str,
        domain: str,
        source: str,
        details: Optional[Dict[str, Any]] = None
    ):
        """
        Log detected security threat.
        
        Args:
            threat_type: Type of threat (credential_leak, ransomware, iab, etc.)
            severity: Threat severity (CRITICAL, HIGH, MEDIUM, LOW)
            domain: Affected domain
            source: Detection source (Telegram, Tor, Pastebin, etc.)
            details: Additional threat details
        """
        metadata = {
            "threat_type": threat_type,
            "severity": severity,
            "domain": domain,
            "source": source,
            **(details or {})
        }
        
        level = logging.ERROR if severity == "CRITICAL" else logging.WARNING
        self._log(level, f"🚨 THREAT DETECTED: {threat_type} on {domain}", metadata)
    
    def scan_started(self, scan_type: str, targets: List[str]):
        """Log scan initiation."""
        self.info(f"🔍 Scan started: {scan_type}", {
            "scan_type": scan_type,
            "target_count": len(targets),
            "targets": targets[:10]  # Log first 10 targets
        })
    
    def scan_completed(
        self,
        scan_type: str,
        duration_seconds: float,
        results_count: int,
        errors_count: int = 0
    ):
        """Log scan completion."""
        self.info(f"✅ Scan completed: {scan_type}", {
            "scan_type": scan_type,
            "duration_seconds": round(duration_seconds, 2),
            "results_count": results_count,
            "errors_count": errors_count
        })
    
    def api_request(
        self,
        method: str,
        url: str,
        status_code: int,
        duration_ms: float
    ):
        """Log API request/response."""
        self.http(f"{method} {url} → {status_code}", {
            "method": method,
            "url": url,
            "status_code": status_code,
            "duration_ms": round(duration_ms, 2)
        })
    
    def credential_found(
        self,
        email: str,
        source: str,
        has_password: bool = False,
        domain: Optional[str] = None
    ):
        """Log discovered credential."""
        # Mask email for privacy in logs
        masked_email = email[:3] + "***@" + email.split("@")[-1] if "@" in email else email[:3] + "***"
        
        self.warn(f"🔑 Credential found: {masked_email}", {
            "source": source,
            "has_password": has_password,
            "domain": domain
        })
    
    def connection_status(self, service: str, status: str, details: Optional[str] = None):
        """Log service connection status."""
        icon = "🟢" if status == "connected" else "🔴" if status == "failed" else "🟡"
        level = logging.INFO if status == "connected" else logging.WARNING
        
        self._log(level, f"{icon} {service}: {status}", {
            "service": service,
            "status": status,
            "details": details
        })
    
    def performance_metric(
        self,
        operation: str,
        duration_ms: float,
        success: bool,
        metadata: Optional[Dict[str, Any]] = None
    ):
        """Log performance metric."""
        self.verbose(f"⏱️ {operation}: {duration_ms:.2f}ms", {
            "operation": operation,
            "duration_ms": duration_ms,
            "success": success,
            **(metadata or {})
        })
    
    def audit(
        self,
        action: str,
        user: str,
        resource: str,
        result: str,
        metadata: Optional[Dict[str, Any]] = None
    ):
        """Log audit trail entry."""
        self.info(f"📋 AUDIT: {user} → {action} → {resource} [{result}]", {
            "audit": True,
            "action": action,
            "user": user,
            "resource": resource,
            "result": result,
            **(metadata or {})
        })


# ============================================================================
# MODULE-LEVEL CONVENIENCE FUNCTIONS
# ============================================================================

# Default logger instance
_default_logger: Optional[ShadowHunterLogger] = None


def get_logger(module_name: str = "ShadowHunter", **kwargs) -> ShadowHunterLogger:
    """
    Get or create a logger instance for the specified module.
    
    Args:
        module_name: Name of the module
        **kwargs: Additional logger configuration
    
    Returns:
        ShadowHunterLogger instance
    """
    return ShadowHunterLogger(module_name, **kwargs)


def configure_logging(
    log_level: LogLevel = LogLevel.INFO,
    log_dir: Optional[Path] = None,
    enable_console: bool = True,
    enable_file: bool = True,
    enable_json: bool = True
):
    """
    Configure global logging settings.
    
    Args:
        log_level: Minimum log level to capture
        log_dir: Directory for log files
        enable_console: Enable colored console output
        enable_file: Enable rotating file logging
        enable_json: Enable JSON structured logging
    """
    global _default_logger
    _default_logger = ShadowHunterLogger(
        "ShadowHunter",
        log_level=log_level,
        log_dir=log_dir,
        enable_console=enable_console,
        enable_file=enable_file,
        enable_json=enable_json
    )
    return _default_logger


# ============================================================================
# DEMO & TESTING
# ============================================================================

if __name__ == "__main__":
    print("=" * 80)
    print("ShadowHunter Logger Demo")
    print("=" * 80)
    
    # Create logger
    logger = get_logger("Demo", log_level=LogLevel.DEBUG)
    
    print("\n📝 Testing all log levels:\n")
    
    # Test all log levels
    logger.error("Database connection failed", {
        "host": "localhost",
        "port": 5432,
        "error_code": "CONN_REFUSED"
    })
    
    logger.warn("Rate limit approaching", {
        "current": 450,
        "limit": 500,
        "resets_in": "5 minutes"
    })
    
    logger.info("Scan completed successfully", {
        "threats_found": 12,
        "duration": "45.3s"
    })
    
    logger.http("API request completed", {
        "method": "POST",
        "url": "/api/scan",
        "status": 200
    })
    
    logger.verbose("Cache hit for domain lookup", {
        "domain": "example.com",
        "cache_age": "2 minutes"
    })
    
    logger.debug("Processing credential entry", {
        "entry_id": 12345,
        "source": "telegram"
    })
    
    logger.silly("Loop iteration", {
        "iteration": 42,
        "total": 100
    })
    
    print("\n📝 Testing specialized logging methods:\n")
    
    # Test specialized methods
    logger.threat_detected(
        threat_type="credential_leak",
        severity="CRITICAL",
        domain="acmecorp.com",
        source="Telegram",
        details={"credential_count": 45, "has_passwords": True}
    )
    
    logger.scan_started("telegram_monitor", ["@stealer_logs", "@breach_alerts"])
    logger.scan_completed("telegram_monitor", 15.5, 23, 2)
    
    logger.credential_found(
        email="admin@example.com",
        source="Lumma Stealer Log",
        has_password=True,
        domain="example.com"
    )
    
    logger.connection_status("Neo4j", "connected", "localhost:7687")
    logger.connection_status("Tor", "failed", "SOCKS5 proxy not responding")
    
    logger.performance_metric(
        operation="credential_extraction",
        duration_ms=125.5,
        success=True,
        metadata={"records_processed": 1000}
    )
    
    logger.audit(
        action="DELETE",
        user="admin",
        resource="user:analyst_123",
        result="success"
    )
    
    print("\n" + "=" * 80)
    print(f"📁 Log files created in: {logger.log_dir}")
    print("=" * 80)