Behavioral Monitoring for AI Systems

import asyncio
import json
import logging
from datetime import datetime, timedelta
from typing import Dict, List, Any, Optional
from dataclasses import dataclass
import redis
import sqlite3
from concurrent.futures import ThreadPoolExecutor

class BehavioralMonitoringSystem:
    def __init__(self, config: Dict[str, Any]):
        self.config = config
        self.redis_client = redis.Redis(**config['redis'])
        self.db_path = config['database']['path']
        self.event_queue = asyncio.Queue(maxsize=10000)
        self.anomaly_detector = BehaviorAnomalyDetector()
        self.pattern_detector = SequentialPatternDetector()
        self.collector = BehaviorCollector(self.event_queue)
        self.executor = ThreadPoolExecutor(max_workers=4)
        self.running = False
        
        # Initialize database
        self._init_database()
        
        # Load existing model if available
        self._load_model()
    
    def _init_database(self):
        """Initialize SQLite database for storing events and patterns"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS behavior_events (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                timestamp REAL NOT NULL,
                user_id TEXT NOT NULL,
                session_id TEXT NOT NULL,
                event_type TEXT NOT NULL,
                event_data TEXT NOT NULL,
                risk_score REAL DEFAULT 0.0,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        ''')
        
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS anomalies (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                user_id TEXT NOT NULL,
                anomaly_type TEXT NOT NULL,
                severity TEXT NOT NULL,
                details TEXT NOT NULL,
                resolved BOOLEAN DEFAULT FALSE,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        ''')
        
        cursor.execute('''
            CREATE INDEX IF NOT EXISTS idx_events_user_time 
            ON behavior_events(user_id, timestamp)
        ''')
        
        conn.commit()
        conn.close()
    
    async def start_monitoring(self):
        """Start the behavioral monitoring system"""
        self.running = True
        
        # Start background tasks
        tasks = [
            asyncio.create_task(self._event_processor()),
            asyncio.create_task(self._periodic_analysis()),
            asyncio.create_task(self._model_retraining())
        ]
        
        logging.info("Behavioral monitoring system started")
        
        try:
            await asyncio.gather(*tasks)
        except Exception as e:
            logging.error(f"Monitoring system error: {e}")
        finally:
            self.running = False
    
    async def _event_processor(self):
        """Process events from the queue"""
        while self.running:
            try:
                # Get event with timeout
                event = await asyncio.wait_for(
                    self.event_queue.get(), timeout=1.0
                )
                
                # Store event in database
                await self._store_event(event)
                
                # Real-time anomaly detection
                await self._check_real_time_anomalies(event)
                
            except asyncio.TimeoutError:
                continue
            except Exception as e:
                logging.error(f"Event processing error: {e}")
    
    async def _store_event(self, event: BehaviorEvent):
        """Store event in database and cache"""
        # Store in database
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        event_data = {
            'request_data': event.request_data,
            'response_data': event.response_data,
            'processing_time': event.processing_time,
            'resource_usage': event.resource_usage,
            'metadata': event.metadata
        }
        
        cursor.execute('''
            INSERT INTO behavior_events 
            (timestamp, user_id, session_id, event_type, event_data)
            VALUES (?, ?, ?, ?, ?)
        ''', (
            event.timestamp,
            event.user_id,
            event.session_id,
            event.event_type,
            json.dumps(event_data)
        ))
        
        conn.commit()
        conn.close()
        
        # Cache recent events for real-time analysis
        cache_key = f"recent_events:{event.user_id}"
        self.redis_client.lpush(cache_key, json.dumps(asdict(event)))
        self.redis_client.ltrim(cache_key, 0, 99)  # Keep last 100 events
        self.redis_client.expire(cache_key, 3600)  # 1 hour TTL
    
    async def _check_real_time_anomalies(self, event: BehaviorEvent):
        """Perform real-time anomaly detection on incoming events"""
        try:
            # Get recent events for this user
            cache_key = f"recent_events:{event.user_id}"
            recent_events_data = self.redis_client.lrange(cache_key, 0, -1)
            
            if len(recent_events_data) < 5:
                return  # Need sufficient data
            
            # Convert to BehaviorEvent objects
            recent_events = []
            for event_data in recent_events_data:
                try:
                    data = json.loads(event_data)
                    recent_events.append(BehaviorEvent(**data))
                except Exception:
                    continue
            
            # Run anomaly detection
            if self.anomaly_detector.is_trained:
                anomalies = self.anomaly_detector.detect_anomalies(recent_events)
                
                for anomaly in anomalies:
                    await self._handle_anomaly(event.user_id, anomaly)
            
            # Run pattern detection
            pattern_analysis = self.pattern_detector.analyze_sequence(
                recent_events, event.user_id
            )
            
            if pattern_analysis['risk_score'] > 0.7:
                await self._handle_suspicious_pattern(
                    event.user_id, pattern_analysis
                )
                
        except Exception as e:
            logging.error(f"Real-time analysis error: {e}")
    
    async def _handle_anomaly(self, user_id: str, anomaly: Dict[str, Any]):
        """Handle detected anomaly"""
        # Store anomaly
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        cursor.execute('''
            INSERT INTO anomalies (user_id, anomaly_type, severity, details)
            VALUES (?, ?, ?, ?)
        ''', (
            user_id,
            'behavioral_anomaly',
            anomaly['severity'],
            json.dumps(anomaly)
        ))
        
        conn.commit()
        conn.close()
        
        # Trigger alerts for high-severity anomalies
        if anomaly['severity'] in ['critical', 'high']:
            await self._send_alert({
                'type': 'behavioral_anomaly',
                'user_id': user_id,
                'severity': anomaly['severity'],
                'details': anomaly,
                'timestamp': datetime.now().isoformat()
            })
    
    async def _send_alert(self, alert_data: Dict[str, Any]):
        """Send alert to configured channels"""
        # Log alert
        logging.warning(f"SECURITY ALERT: {alert_data}")
        
        # Send to external systems (webhook, Slack, etc.)
        alert_channels = self.config.get('alerts', {}).get('channels', [])
        
        for channel in alert_channels:
            try:
                if channel['type'] == 'webhook':
                    # Send webhook (implementation depends on your setup)
                    pass
                elif channel['type'] == 'email':
                    # Send email alert
                    pass
            except Exception as e:
                logging.error(f"Alert sending failed: {e}")
    
    async def get_user_behavior_summary(self, 
                                      user_id: str, 
                                      hours: int = 24) -> Dict[str, Any]:
        """Get behavioral summary for a user"""
        start_time = datetime.now() - timedelta(hours=hours)
        
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        # Get events
        cursor.execute('''
            SELECT * FROM behavior_events 
            WHERE user_id = ? AND timestamp >= ?
            ORDER BY timestamp DESC
        ''', (user_id, start_time.timestamp()))
        
        events = cursor.fetchall()
        
        # Get anomalies
        cursor.execute('''
            SELECT * FROM anomalies 
            WHERE user_id = ? AND created_at >= ?
            ORDER BY created_at DESC
        ''', (user_id, start_time.isoformat()))
        
        anomalies = cursor.fetchall()
        conn.close()
        
        return {
            'user_id': user_id,
            'period_hours': hours,
            'total_events': len(events),
            'total_anomalies': len(anomalies),
            'risk_level': self._calculate_user_risk_level(events, anomalies),
            'recent_events': events[:10],  # Last 10 events
            'recent_anomalies': anomalies[:5]  # Last 5 anomalies
        }
    
    def _calculate_user_risk_level(self, events: List, anomalies: List) -> str:
        """Calculate overall risk level for user"""
        if not events:
            return 'unknown'
        
        anomaly_rate = len(anomalies) / len(events)
        
        if anomaly_rate > 0.3:
            return 'critical'
        elif anomaly_rate > 0.1:
            return 'high'
        elif anomaly_rate > 0.05:
            return 'medium'
        else:
            return 'low'

# Usage example
async def main():
    config = {
        'redis': {'host': 'localhost', 'port': 6379, 'db': 0},
        'database': {'path': 'behavioral_monitoring.db'},
        'alerts': {
            'channels': [
                {'type': 'webhook', 'url': 'https://your-webhook.com/alerts'},
                {'type': 'email', 'recipients': ['security@yourcompany.com']}
            ]
        }
    }
    
    monitoring_system = BehavioralMonitoringSystem(config)
    await monitoring_system.start_monitoring()

if __name__ == "__main__":
    asyncio.run(main())