alert_monitor.py

#!/usr/bin/env python3
"""
Alert Monitor - Security Alert Triage System
==============================================

A professional security alert monitoring and triage system for SOC analysts.
Tracks, prioritizes, and escalates security alerts based on severity.

Author: [Joshua Guda]
GitHub: [https://github.com/joshuaguda281-stack]
Created: 2024
License: MIT

Features:
- Alert database with SQLite storage
- Severity-based classification (1-5)
- Automatic escalation for critical alerts
- Report generation
- Email/Slack notifications (configurable)

Usage:
    python3 alert_monitor.py              # Interactive mode
    python3 alert_monitor.py monitor      # Start monitoring
    python3 alert_monitor.py report       # Generate report
"""

import json
import sqlite3
import sys
import time
from datetime import datetime
from collections import defaultdict

# Try to import optional dependencies
try:
    import requests
    HAS_REQUESTS = True
except ImportError:
    HAS_REQUESTS = False

try:
    import smtplib
    from email.mime.text import MIMEText
    HAS_SMTP = True
except ImportError:
    HAS_SMTP = False


class AlertMonitor:
    """
    Security Alert Monitoring System
    
    This class handles the complete lifecycle of security alerts:
    - Storage in SQLite database
    - Severity classification
    - Automated escalation
    - Notification delivery
    - Report generation
    """
    
    def __init__(self, db_path="alerts.db", config_path=None):
        """
        Initialize the Alert Monitor
        
        Args:
            db_path (str): Path to SQLite database file
            config_path (str): Path to configuration JSON file
        """
        self.db_path = db_path
        self.config = self.load_config(config_path)
        self.severity_levels = {
            1: "Informational",
            2: "Low", 
            3: "Medium",
            4: "High",
            5: "Critical"
        }
        self.init_database()
        print(f"[+] Alert Monitor initialized. Database: {db_path}")
    
    def load_config(self, config_path):
        """
        Load configuration from JSON file
        
        Default configuration includes:
        - Notification settings
        - Escalation thresholds
        - Integration endpoints
        """
        default_config = {
            'escalation_threshold': 4,  # Severity 4 and above escalate
            'enable_email': False,
            'enable_slack': False,
            'email_recipient': 'soc@example.com',
            'slack_webhook': None,
            'smtp_server': 'smtp.gmail.com',
            'smtp_port': 587
        }
        
        if config_path and self._file_exists(config_path):
            try:
                with open(config_path, 'r') as f:
                    user_config = json.load(f)
                    default_config.update(user_config)
                print(f"[+] Loaded config from {config_path}")
            except Exception as e:
                print(f"[-] Error loading config: {e}")
        
        return default_config
    
    def _file_exists(self, path):
        """Check if file exists"""
        try:
            with open(path, 'r'):
                return True
        except:
            return False
    
    def init_database(self):
        """Initialize SQLite database with alerts table"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS alerts (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                timestamp DATETIME,
                source_ip TEXT,
                destination_ip TEXT,
                signature TEXT,
                severity INTEGER,
                status TEXT DEFAULT 'new',
                analyst_notes TEXT,
                resolution TEXT,
                escalated INTEGER DEFAULT 0
            )
        ''')
        
        # Create index for faster queries
        cursor.execute('''
            CREATE INDEX IF NOT EXISTS idx_timestamp 
            ON alerts(timestamp)
        ''')
        
        cursor.execute('''
            CREATE INDEX IF NOT EXISTS idx_severity 
            ON alerts(severity)
        ''')
        
        conn.commit()
        conn.close()
        print("[+] Database initialized successfully")
    
    def add_alert(self, source_ip, dest_ip, signature, severity, auto_escalate=True):
        """
        Add a new security alert to the system
        
        Args:
            source_ip (str): Source IP address
            dest_ip (str): Destination IP address
            signature (str): Alert signature/description
            severity (int): 1-5 severity level
            auto_escalate (bool): Automatically escalate critical alerts
        
        Returns:
            int: Alert ID
        """
        # Validate severity
        if severity not in self.severity_levels:
            raise ValueError(f"Invalid severity: {severity}. Must be 1-5")
        
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        cursor.execute('''
            INSERT INTO alerts (timestamp, source_ip, destination_ip, signature, severity)
            VALUES (?, ?, ?, ?, ?)
        ''', (datetime.now(), source_ip, dest_ip, signature, severity))
        
        conn.commit()
        alert_id = cursor.lastrowid
        conn.close()
        
        # Print to console
        severity_name = self.severity_levels[severity]
        print(f"\n[!] ALERT #{alert_id}: {signature}")
        print(f"    Source: {source_ip} -> Destination: {dest_ip}")
        print(f"    Severity: {severity_name} ({severity}/5)")
        
        # Auto-escalate critical alerts
        if auto_escalate and severity >= self.config['escalation_threshold']:
            self.escalate_alert(alert_id)
        
        return alert_id
    
    def escalate_alert(self, alert_id):
        """
        Escalate a high-severity alert to incident response team
        
        Args:
            alert_id (int): Alert ID to escalate
        """
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        cursor.execute('''
            UPDATE alerts 
            SET escalated = 1, status = 'escalated'
            WHERE id = ?
        ''', (alert_id,))
        
        conn.commit()
        conn.close()
        
        print(f"[!] ALERT #{alert_id} ESCALATED to Incident Response Team!")
        self.send_notification(alert_id)
    
    def send_notification(self, alert_id):
        """
        Send notifications via configured channels
        
        Supported channels:
        - Email (SMTP)
        - Slack (Webhook)
        - Console (always)
        """
        # Get alert details
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        cursor.execute('SELECT * FROM alerts WHERE id = ?', (alert_id,))
        alert = cursor.fetchone()
        conn.close()
        
        if not alert:
            print(f"[-] Alert {alert_id} not found")
            return
        
        # Format notification
        notification = f"""
╔══════════════════════════════════════════════════════════════╗
║                    🚨 SECURITY ALERT 🚨                       ║
╠══════════════════════════════════════════════════════════════╣
║ Alert ID:     {alert[0]}
║ Timestamp:    {alert[1]}
║ Source:       {alert[2]}
║ Destination:  {alert[3]}
║ Signature:    {alert[4]}
║ Severity:     {self.severity_levels[alert[5]]} ({alert[5]}/5)
║ Status:       {alert[6]}
╚══════════════════════════════════════════════════════════════╝

ACTION REQUIRED: Immediate investigation recommended.
        """
        
        print(notification)
        
        # Send email if configured
        if self.config['enable_email'] and HAS_SMTP:
            self._send_email(alert)
        
        # Send Slack if configured
        if self.config['enable_slack'] and HAS_REQUESTS and self.config['slack_webhook']:
            self._send_slack(alert)
        
        # Save to file as backup
        with open(f"alert_{alert_id}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.txt", "w") as f:
            f.write(notification)
    
    def _send_email(self, alert):
        """Send email notification (requires SMTP configuration)"""
        try:
            msg = MIMEText(f"""
            Security Alert Detected
            
            Alert ID: {alert[0]}
            Timestamp: {alert[1]}
            Source: {alert[2]}
            Destination: {alert[3]}
            Signature: {alert[4]}
            Severity: {self.severity_levels[alert[5]]}
            
            Please investigate immediately.
            """)
            
            msg['Subject'] = f"Security Alert #{alert[0]} - {self.severity_levels[alert[5]]}"
            msg['From'] = "security@yourdomain.com"
            msg['To'] = self.config['email_recipient']
            
            # Uncomment and configure to actually send emails
            # server = smtplib.SMTP(self.config['smtp_server'], self.config['smtp_port'])
            # server.starttls()
            # server.login("your-email", "your-password")
            # server.send_message(msg)
            # server.quit()
            
            print(f"    📧 Email notification sent to {self.config['email_recipient']}")
        except Exception as e:
            print(f"    [-] Email failed: {e}")
    
    def _send_slack(self, alert):
        """Send Slack notification via webhook"""
        try:
            payload = {
                'text': f"🚨 *Security Alert #{alert[0]}*",
                'attachments': [{
                    'color': 'danger' if alert[5] >= 4 else 'warning',
                    'fields': [
                        {'title': 'Signature', 'value': alert[4], 'short': False},
                        {'title': 'Source', 'value': alert[2], 'short': True},
                        {'title': 'Destination', 'value': alert[3], 'short': True},
                        {'title': 'Severity', 'value': self.severity_levels[alert[5]], 'short': True}
                    ]
                }]
            }
            
            # Uncomment to actually send to Slack
            # requests.post(self.config['slack_webhook'], json=payload)
            print(f"    💬 Slack notification sent")
        except Exception as e:
            print(f"    [-] Slack failed: {e}")
    
    def triage_alert(self, alert_id, notes, resolution="investigated"):
        """
        Triage an alert (Tier 1 SOC function)
        
        Args:
            alert_id (int): Alert ID to triage
            notes (str): Analyst investigation notes
            resolution (str): Resolution status
        """
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        cursor.execute('''
            UPDATE alerts 
            SET status = 'triaged', analyst_notes = ?, resolution = ?
            WHERE id = ?
        ''', (notes, resolution, alert_id))
        
        conn.commit()
        conn.close()
        
        print(f"[+] Alert {alert_id} triaged by analyst")
        print(f"    Notes: {notes[:100]}...")
    
    def get_stats(self):
        """Get alert statistics"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        # Total alerts
        cursor.execute('SELECT COUNT(*) FROM alerts')
        total = cursor.fetchone()[0]
        
        # By severity
        cursor.execute('''
            SELECT severity, COUNT(*) 
            FROM alerts 
            GROUP BY severity
            ORDER BY severity DESC
        ''')
        by_severity = cursor.fetchall()
        
        # By status
        cursor.execute('''
            SELECT status, COUNT(*) 
            FROM alerts 
            GROUP BY status
        ''')
        by_status = cursor.fetchall()
        
        # Top sources
        cursor.execute('''
            SELECT source_ip, COUNT(*) as count
            FROM alerts
            GROUP BY source_ip
            ORDER BY count DESC
            LIMIT 5
        ''')
        top_sources = cursor.fetchall()
        
        conn.close()
        
        return {
            'total': total,
            'by_severity': by_severity,
            'by_status': by_status,
            'top_sources': top_sources
        }
    
    def generate_report(self):
        """Generate daily security report"""
        stats = self.get_stats()
        
        report = f"""
{'='*70}
SECURITY ALERT REPORT
{'='*70}
Generated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
{'='*70}

SUMMARY STATISTICS
{'='*70}
Total Alerts: {stats['total']}

Alerts by Severity:
"""
        
        for severity, count in stats['by_severity']:
            severity_name = self.severity_levels[severity]
            bar = '█' * min(count, 50)
            report += f"  {severity_name:12} ({severity}): {count:4} {bar}\n"
        
        report += f"""
{'='*70}
Alerts by Status:
"""
        for status, count in stats['by_status']:
            report += f"  {status:12}: {count}\n"
        
        report += f"""
{'='*70}
Top Alert Sources:
"""
        for source, count in stats['top_sources']:
            report += f"  {source:15}: {count} alerts\n"
        
        report += f"""
{'='*70}
RECOMMENDATIONS
{'='*70}
"""
        
        # Add recommendations based on data
        critical_count = sum(count for sev, count in stats['by_severity'] if sev >= 4)
        if critical_count > 0:
            report += f"  ⚠️  {critical_count} critical alerts require immediate attention\n"
        
        if any(source[1] > 100 for source in stats['top_sources']):
            report += "  ⚠️  Possible port scanning detected - investigate top sources\n"
        
        report += f"""
{'='*70}
End of Report
{'='*70}
"""
        
        # Save report to file
        filename = f"security_report_{datetime.now().strftime('%Y%m%d')}.txt"
        with open(filename, 'w') as f:
            f.write(report)
        
        print(f"[+] Report saved to {filename}")
        print(report)
        
        return report
    
    def interactive_mode(self):
        """Run in interactive mode for manual alert entry"""
        print("\n" + "="*60)
        print("Alert Monitor - Interactive Mode")
        print("Enter alert details (Ctrl+C to exit)")
        print("="*60 + "\n")
        
        try:
            while True:
                print("\n--- New Alert ---")
                source = input("Source IP: ").strip()
                dest = input("Destination IP: ").strip()
                sig = input("Signature/Description: ").strip()
                
                print("\nSeverity:")
                for level, name in self.severity_levels.items():
                    print(f"  {level}: {name}")
                severity = int(input("Severity (1-5): ").strip())
                
                self.add_alert(source, dest, sig, severity)
                
                # Ask to triage
                triage = input("\nTriage now? (y/n): ").strip().lower()
                if triage == 'y':
                    notes = input("Investigation notes: ").strip()
                    self.triage_alert(alert_id, notes)
                
        except KeyboardInterrupt:
            print("\n\n[*] Exiting interactive mode")
            self.generate_report()
    
    def monitor_mode(self):
        """Simulate monitoring mode with sample alerts"""
        print("\n" + "="*60)
        print("Alert Monitor - Monitoring Mode")
        print("Simulating security alerts... (Ctrl+C to stop)")
        print("="*60 + "\n")
        
        sample_alerts = [
            ("192.168.1.100", "10.0.0.1", "Multiple failed SSH logins", 3),
            ("45.33.22.11", "192.168.1.50", "SQL Injection attempt detected", 4),
            ("185.130.5.253", "192.168.1.50", "Port scan detected", 2),
            ("10.0.0.5", "10.0.0.10", "Suspicious PowerShell execution", 5),
            ("192.168.1.200", "192.168.1.1", "DNS tunneling detected", 4),
            ("94.102.61.78", "192.168.1.50", "Malware C2 communication", 5),
            ("8.8.8.8", "192.168.1.50", "Unusual DNS query volume", 3),
        ]
        
        try:
            for src, dst, sig, sev in sample_alerts:
                alert_id = self.add_alert(src, dst, sig, sev)
                time.sleep(2)  # Simulate real-time monitoring
        except KeyboardInterrupt:
            print("\n\n[*] Stopping monitor...")
            self.generate_report()


def main():
    """Main entry point"""
    import argparse
    
    parser = argparse.ArgumentParser(description='Security Alert Monitor')
    parser.add_argument('mode', nargs='?', default='interactive',
                       choices=['interactive', 'monitor', 'report'],
                       help='Operation mode')
    parser.add_argument('--db', default='alerts.db',
                       help='Database file path')
    parser.add_argument('--config', default=None,
                       help='Configuration file path')
    
    args = parser.parse_args()
    
    # Initialize monitor
    monitor = AlertMonitor(db_path=args.db, config_path=args.config)
    
    # Run requested mode
    if args.mode == 'interactive':
        monitor.interactive_mode()
    elif args.mode == 'monitor':
        monitor.monitor_mode()
    elif args.mode == 'report':
        monitor.generate_report()


if __name__ == "__main__":
    main()