Essential Security Tools Every Beginner Should Know (2026...

python3 -m venv .venv-keyboard-monitor
source .venv-keyboard-monitor/bin/activate
pip install --upgrade pip
pip install pynput numpy scikit-learn pandas matplotlib flask

#!/usr/bin/env python3
"""
Keyboard Event Monitor
Captures typing patterns while preserving privacy
"""

from pynput import keyboard
import time
import json
from datetime import datetime
from collections import deque
import hashlib

class KeyboardMonitor:
    def __init__(self, max_samples=1000):
        self.max_samples = max_samples
        self.keystrokes = deque(maxlen=max_samples)
        self.session_start = time.time()
        self.last_key_time = None
        self.is_monitoring = False
        
    def on_press(self, key):
        """Handle key press events"""
        try:
            current_time = time.time()
            
            # Calculate time since last key
            interval = None
            if self.last_key_time:
                interval = current_time - self.last_key_time
            
            # Record keystroke pattern
            key_data = {
                "timestamp": current_time,
                "key": str(key).replace("'", ""),
                "interval": interval,
                "session_time": current_time - self.session_start
            }
            
            self.keystrokes.append(key_data)
            self.last_key_time = current_time
            
        except Exception as e:
            print(f"Error capturing key: {e}")
    
    def get_typing_features(self):
        """Extract typing pattern features"""
        if len(self.keystrokes) < 10:
            return None
        
        intervals = [k["interval"] for k in self.keystrokes if k["interval"] is not None]
        
        if not intervals:
            return None
        
        features = {
            "mean_interval": sum(intervals) / len(intervals),
            "std_interval": (sum((x - sum(intervals)/len(intervals))**2 for x in intervals) / len(intervals))**0.5,
            "min_interval": min(intervals),
            "max_interval": max(intervals),
            "keystroke_count": len(self.keystrokes),
            "session_duration": time.time() - self.session_start
        }
        
        return features
    
    def start_monitoring(self):
        """Start keyboard monitoring"""
        self.is_monitoring = True
        self.listener = keyboard.Listener(on_press=self.on_press)
        self.listener.start()
        print("Keyboard monitoring started. Press Ctrl+C to stop.")
    
    def stop_monitoring(self):
        """Stop keyboard monitoring"""
        self.is_monitoring = False
        if hasattr(self, 'listener'):
            self.listener.stop()
        print("Keyboard monitoring stopped.")
    
    def save_session(self, filename="typing_session.json"):
        """Save typing session data (privacy-preserving)"""
        # Hash sensitive data
        session_data = {
            "session_id": hashlib.sha256(str(self.session_start).encode()).hexdigest()[:16],
            "features": self.get_typing_features(),
            "keystroke_count": len(self.keystrokes),
            "timestamp": datetime.now().isoformat()
        }
        
        with open(filename, 'w') as f:
            json.dump(session_data, f, indent=2)
        
        print(f"Session saved to {filename}")
        return session_data

# Example usage
if __name__ == "__main__":
    monitor = KeyboardMonitor()
    
    try:
        monitor.start_monitoring()
        # Monitor for 60 seconds
        time.sleep(60)
    except KeyboardInterrupt:
        pass
    finally:
        monitor.stop_monitoring()
        features = monitor.get_typing_features()
        if features:
            print("\nTyping Pattern Features:")
            print(json.dumps(features, indent=2))
        monitor.save_session()

#!/usr/bin/env python3
"""
Typing Pattern Analyzer
Analyzes typing patterns to build user baseline
"""

import json
import numpy as np
from collections import defaultdict
import os

class TypingPatternAnalyzer:
    def __init__(self):
        self.baseline_features = None
        self.session_history = []
        
    def load_baseline(self, baseline_file="user_baseline.json"):
        """Load user baseline typing pattern"""
        if os.path.exists(baseline_file):
            with open(baseline_file, 'r') as f:
                data = json.load(f)
                self.baseline_features = data.get("features")
                self.session_history = data.get("history", [])
            print(f"Loaded baseline from {baseline_file}")
            return True
        return False
    
    def build_baseline(self, session_files):
        """Build baseline from multiple sessions"""
        all_features = []
        
        for session_file in session_files:
            if os.path.exists(session_file):
                with open(session_file, 'r') as f:
                    data = json.load(f)
                    features = data.get("features")
                    if features:
                        all_features.append(features)
        
        if not all_features:
            return None
        
        # Calculate average baseline
        baseline = {
            "mean_interval": np.mean([f["mean_interval"] for f in all_features]),
            "std_interval": np.mean([f["std_interval"] for f in all_features]),
            "min_interval": np.mean([f["min_interval"] for f in all_features]),
            "max_interval": np.mean([f["max_interval"] for f in all_features]),
            "session_count": len(all_features)
        }
        
        self.baseline_features = baseline
        return baseline
    
    def compare_pattern(self, current_features):
        """Compare current pattern with baseline"""
        if not self.baseline_features or not current_features:
            return None
        
        # Calculate deviations
        deviations = {
            "mean_interval_dev": abs(current_features["mean_interval"] - self.baseline_features["mean_interval"]) / self.baseline_features["mean_interval"],
            "std_interval_dev": abs(current_features["std_interval"] - self.baseline_features["std_interval"]) / (self.baseline_features["std_interval"] + 0.001),
            "min_interval_dev": abs(current_features["min_interval"] - self.baseline_features["min_interval"]) / (self.baseline_features["min_interval"] + 0.001),
            "max_interval_dev": abs(current_features["max_interval"] - self.baseline_features["max_interval"]) / (self.baseline_features["max_interval"] + 0.001),
        }
        
        # Calculate anomaly score (0-1, higher = more anomalous)
        anomaly_score = (
            deviations["mean_interval_dev"] * 0.4 +
            deviations["std_interval_dev"] * 0.3 +
            deviations["min_interval_dev"] * 0.15 +
            deviations["max_interval_dev"] * 0.15
        )
        
        return {
            "anomaly_score": min(anomaly_score, 1.0),
            "deviations": deviations,
            "is_anomaly": anomaly_score > 0.3  # Threshold
        }
    
    def save_baseline(self, baseline_file="user_baseline.json"):
        """Save baseline to file"""
        data = {
            "features": self.baseline_features,
            "history": self.session_history,
            "updated": datetime.now().isoformat()
        }
        
        with open(baseline_file, 'w') as f:
            json.dump(data, f, indent=2)
        
        print(f"Baseline saved to {baseline_file}")

# Example usage
if __name__ == "__main__":
    analyzer = TypingPatternAnalyzer()
    
    # Build baseline from multiple sessions
    session_files = ["typing_session.json", "typing_session_2.json"]
    baseline = analyzer.build_baseline(session_files)
    
    if baseline:
        print("\nUser Baseline:")
        print(json.dumps(baseline, indent=2))
        analyzer.save_baseline()
        
        # Test with new session
        with open("typing_session.json", 'r') as f:
            current = json.load(f)
            comparison = analyzer.compare_pattern(current.get("features"))
            if comparison:
                print("\nAnomaly Detection Result:")
                print(json.dumps(comparison, indent=2))

#!/usr/bin/env python3
"""
ML-Based Anomaly Detection
Uses Isolation Forest for typing pattern anomaly detection
"""

import numpy as np
import pandas as pd
from sklearn.ensemble import IsolationForest
from sklearn.preprocessing import StandardScaler
import joblib
import json
import os

class MLAnomalyDetector:
    def __init__(self):
        self.model = None
        self.scaler = StandardScaler()
        self.is_trained = False
        
    def prepare_features(self, features_list):
        """Prepare features for ML model"""
        if not features_list:
            return None
        
        # Extract feature vectors
        feature_vectors = []
        for features in features_list:
            if features:
                vector = [
                    features.get("mean_interval", 0),
                    features.get("std_interval", 0),
                    features.get("min_interval", 0),
                    features.get("max_interval", 0),
                    features.get("keystroke_count", 0),
                    features.get("session_duration", 0)
                ]
                feature_vectors.append(vector)
        
        return np.array(feature_vectors)
    
    def train(self, normal_sessions):
        """Train anomaly detection model on normal typing patterns"""
        # Prepare training data
        X = self.prepare_features(normal_sessions)
        
        if X is None or len(X) < 10:
            print("Need at least 10 normal sessions for training")
            return False
        
        # Scale features
        X_scaled = self.scaler.fit_transform(X)
        
        # Train Isolation Forest (good for anomaly detection)
        self.model = IsolationForest(
            contamination=0.1,  # Expect 10% anomalies
            random_state=42,
            n_estimators=100
        )
        
        self.model.fit(X_scaled)
        self.is_trained = True
        
        print(f"Model trained on {len(X)} normal sessions")
        return True
    
    def predict(self, features):
        """Predict if typing pattern is anomalous"""
        if not self.is_trained or not features:
            return None
        
        # Prepare features
        X = self.prepare_features([features])
        if X is None:
            return None
        
        # Scale features
        X_scaled = self.scaler.transform(X)
        
        # Predict (-1 = anomaly, 1 = normal)
        prediction = self.model.predict(X_scaled)[0]
        anomaly_score = self.model.score_samples(X_scaled)[0]
        
        # Convert to probability (lower score = more anomalous)
        # Normalize to 0-1 range (higher = more anomalous)
        normalized_score = 1 / (1 + np.exp(anomaly_score))
        
        return {
            "is_anomaly": prediction == -1,
            "anomaly_score": float(normalized_score),
            "raw_score": float(anomaly_score)
        }
    
    def save_model(self, model_file="anomaly_model.pkl", scaler_file="scaler.pkl"):
        """Save trained model"""
        if self.is_trained:
            joblib.dump(self.model, model_file)
            joblib.dump(self.scaler, scaler_file)
            print(f"Model saved to {model_file} and {scaler_file}")
    
    def load_model(self, model_file="anomaly_model.pkl", scaler_file="scaler.pkl"):
        """Load trained model"""
        if os.path.exists(model_file) and os.path.exists(scaler_file):
            self.model = joblib.load(model_file)
            self.scaler = joblib.load(scaler_file)
            self.is_trained = True
            print(f"Model loaded from {model_file}")
            return True
        return False

# Example usage
if __name__ == "__main__":
    detector = MLAnomalyDetector()
    
    # Simulate normal sessions for training
    normal_sessions = []
    for i in range(20):
        normal_sessions.append({
            "mean_interval": 0.15 + np.random.normal(0, 0.02),
            "std_interval": 0.05 + np.random.normal(0, 0.01),
            "min_interval": 0.05 + np.random.normal(0, 0.01),
            "max_interval": 0.5 + np.random.normal(0, 0.1),
            "keystroke_count": 100 + np.random.randint(-20, 20),
            "session_duration": 60 + np.random.randint(-10, 10)
        })
    
    # Train model
    if detector.train(normal_sessions):
        detector.save_model()
        
        # Test with new pattern
        test_features = {
            "mean_interval": 0.5,  # Unusually slow
            "std_interval": 0.2,
            "min_interval": 0.1,
            "max_interval": 1.0,
            "keystroke_count": 50,
            "session_duration": 30
        }
        
        result = detector.predict(test_features)
        print("\nAnomaly Detection Result:")
        print(json.dumps(result, indent=2))

#!/usr/bin/env python3
"""
Real-Time Keyboard Behavior Monitor
Combines monitoring, analysis, and ML detection
"""

import time
import json
from datetime import datetime
from keyboard_monitor import KeyboardMonitor
from pattern_analyzer import TypingPatternAnalyzer
from ml_anomaly_detector import MLAnomalyDetector

class RealTimeMonitor:
    def __init__(self, check_interval=30):
        self.monitor = KeyboardMonitor()
        self.analyzer = TypingPatternAnalyzer()
        self.detector = MLAnomalyDetector()
        self.check_interval = check_interval
        self.alerts = []
        
    def initialize(self):
        """Initialize components"""
        # Load baseline if exists
        self.analyzer.load_baseline()
        
        # Load ML model if exists
        if not self.detector.load_model():
            print("Warning: No trained model found. Run training first.")
        
    def check_anomaly(self):
        """Check for typing anomalies"""
        features = self.monitor.get_typing_features()
        
        if not features:
            return None
        
        # Method 1: Baseline comparison
        baseline_result = None
        if self.analyzer.baseline_features:
            baseline_result = self.analyzer.compare_pattern(features)
        
        # Method 2: ML detection
        ml_result = None
        if self.detector.is_trained:
            ml_result = self.detector.predict(features)
        
        # Combine results
        anomaly_detected = False
        confidence = 0.0
        
        if baseline_result and baseline_result["is_anomaly"]:
            anomaly_detected = True
            confidence = baseline_result["anomaly_score"]
        
        if ml_result and ml_result["is_anomaly"]:
            anomaly_detected = True
            confidence = max(confidence, ml_result["anomaly_score"])
        
        if anomaly_detected:
            alert = {
                "timestamp": datetime.now().isoformat(),
                "anomaly_detected": True,
                "confidence": confidence,
                "baseline_result": baseline_result,
                "ml_result": ml_result,
                "features": features
            }
            
            self.alerts.append(alert)
            self.generate_alert(alert)
        
        return {
            "anomaly_detected": anomaly_detected,
            "confidence": confidence,
            "baseline": baseline_result,
            "ml": ml_result
        }
    
    def generate_alert(self, alert_data):
        """Generate alert for detected anomaly"""
        print(f"\n⚠️  ALERT: Anomalous Typing Pattern Detected!")
        print(f"   Confidence: {alert_data['confidence']:.2%}")
        print(f"   Timestamp: {alert_data['timestamp']}")
        
        # In production: Send to SIEM, email, Slack, etc.
        # Example: send_to_siem(alert_data)
    
    def run(self, duration=300):
        """Run real-time monitoring"""
        print(f"Starting real-time monitoring for {duration} seconds...")
        self.monitor.start_monitoring()
        
        start_time = time.time()
        last_check = start_time
        
        try:
            while time.time() - start_time < duration:
                current_time = time.time()
                
                # Check for anomalies periodically
                if current_time - last_check >= self.check_interval:
                    result = self.check_anomaly()
                    if result and result["anomaly_detected"]:
                        print(f"Anomaly detected at {datetime.now().strftime('%H:%M:%S')}")
                    last_check = current_time
                
                time.sleep(1)
        
        except KeyboardInterrupt:
            print("\nStopping monitoring...")
        finally:
            self.monitor.stop_monitoring()
            
            # Save session
            session_data = self.monitor.save_session()
            
            # Print summary
            print(f"\n📊 Monitoring Summary:")
            print(f"   Total alerts: {len(self.alerts)}")
            print(f"   Session duration: {session_data.get('keystroke_count', 0)} keystrokes")
            
            if self.alerts:
                print(f"\n⚠️  Alerts Generated:")
                for i, alert in enumerate(self.alerts, 1):
                    print(f"   {i}. {alert['timestamp']} - Confidence: {alert['confidence']:.2%}")

# Example usage
if __name__ == "__main__":
    monitor = RealTimeMonitor(check_interval=30)
    monitor.initialize()
    monitor.run(duration=300)  # Monitor for 5 minutes

#!/usr/bin/env python3
"""
Dashboard for Keyboard Behavior Monitoring
Simple web interface for visualization
"""

from flask import Flask, render_template_string, jsonify
import json
import os
from datetime import datetime

app = Flask(__name__)

DASHBOARD_HTML = """
<!DOCTYPE html>
<html>
<head>
    <title>Keyboard Behavior Monitor Dashboard</title>
    <style>
        body { font-family: Arial, sans-serif; margin: 20px; background: #f5f5f5; }
        .container { max-width: 1200px; margin: 0 auto; background: white; padding: 20px; border-radius: 8px; }
        .alert { background: #fee; border-left: 4px solid #f00; padding: 10px; margin: 10px 0; }
        .normal { background: #efe; border-left: 4px solid #0f0; padding: 10px; margin: 10px 0; }
        .metric { display: inline-block; margin: 10px; padding: 15px; background: #f0f0f0; border-radius: 4px; }
        .metric-value { font-size: 24px; font-weight: bold; color: #333; }
        .metric-label { font-size: 12px; color: #666; }
        h1 { color: #333; }
        button { padding: 10px 20px; background: #007bff; color: white; border: none; border-radius: 4px; cursor: pointer; }
        button:hover { background: #0056b3; }
    </style>
</head>
<body>
    <div class="container">
        <h1>🔐 Keyboard Behavior Anomaly Detector</h1>
        
        <div>
            <div class="metric">
                <div class="metric-value" id="total-keystrokes">0</div>
                <div class="metric-label">Total Keystrokes</div>
            </div>
            <div class="metric">
                <div class="metric-value" id="anomaly-count">0</div>
                <div class="metric-label">Anomalies Detected</div>
            </div>
            <div class="metric">
                <div class="metric-value" id="session-time">0s</div>
                <div class="metric-label">Session Duration</div>
            </div>
        </div>
        
        <div id="alerts-container"></div>
        
        <button onclick="refreshData()">Refresh</button>
        <button onclick="startMonitoring()">Start Monitoring</button>
        <button onclick="stopMonitoring()">Stop Monitoring</button>
    </div>
    
    <script>
        function refreshData() {
            fetch('/api/status')
                .then(r => r.json())
                .then(data => {
                    document.getElementById('total-keystrokes').textContent = data.keystrokes || 0;
                    document.getElementById('anomaly-count').textContent = data.anomalies || 0;
                    document.getElementById('session-time').textContent = (data.duration || 0) + 's';
                    
                    const container = document.getElementById('alerts-container');
                    container.innerHTML = '';
                    
                    if (data.recent_alerts) {
                        data.recent_alerts.forEach(alert => {
                            const div = document.createElement('div');
                            div.className = alert.is_anomaly ? 'alert' : 'normal';
                            div.innerHTML = `
                                <strong>${alert.timestamp}</strong><br>
                                Confidence: ${(alert.confidence * 100).toFixed(1)}%<br>
                                Status: ${alert.is_anomaly ? '⚠️ Anomaly Detected' : '✓ Normal'}
                            `;
                            container.appendChild(div);
                        });
                    }
                });
        }
        
        function startMonitoring() {
            fetch('/api/start', {method: 'POST'})
                .then(r => r.json())
                .then(data => alert(data.message || 'Monitoring started'));
        }
        
        function stopMonitoring() {
            fetch('/api/stop', {method: 'POST'})
                .then(r => r.json())
                .then(data => alert(data.message || 'Monitoring stopped'));
        }
        
        // Auto-refresh every 5 seconds
        setInterval(refreshData, 5000);
        refreshData();
    </script>
</body>
</html>
"""

@app.route('/')
def dashboard():
    return render_template_string(DASHBOARD_HTML)

@app.route('/api/status')
def get_status():
    """Get current monitoring status"""
    # Read from session file if exists
    status = {
        "keystrokes": 0,
        "anomalies": 0,
        "duration": 0,
        "recent_alerts": []
    }
    
    if os.path.exists("typing_session.json"):
        with open("typing_session.json", 'r') as f:
            data = json.load(f)
            status["keystrokes"] = data.get("keystroke_count", 0)
            status["duration"] = data.get("features", {}).get("session_duration", 0)
    
    # Read alerts if exists
    if os.path.exists("alerts.json"):
        with open("alerts.json", 'r') as f:
            alerts = json.load(f)
            status["anomalies"] = len(alerts)
            status["recent_alerts"] = alerts[-5:]  # Last 5 alerts
    
    return jsonify(status)

@app.route('/api/start', methods=['POST'])
def start_monitoring():
    return jsonify({"message": "Monitoring started (run realtime_monitor.py in separate terminal)"})

@app.route('/api/stop', methods=['POST'])
def stop_monitoring():
    return jsonify({"message": "Monitoring stopped"})

if __name__ == '__main__':
    print("Starting dashboard on http://localhost:5000")
    app.run(debug=True, port=5000)