Using AI to Analyze Log Files for Security Threats

{
  "timestamp": "2025-11-15T14:23:45.123Z",
  "source": "firewall-01",
  "event_type": "connection_denied",
  "src_ip": "203.0.113.42",
  "src_port": 45123,
  "dst_ip": "10.0.1.50",
  "dst_port": 22,
  "protocol": "TCP"
}

import re
import json
from datetime import datetime
from typing import Dict, Any

class LogNormalizer:
    """
    Normalize logs from multiple sources into consistent JSON format

    Supports: Syslog, Apache/Nginx, JSON, Windows Event Log
    """

    def __init__(self):
        # Regex patterns for common log formats
        self.syslog_pattern = re.compile(
            r'<(?P<priority>\d+)>(?P<version>\d+)\s+'
            r'(?P<timestamp>\S+)\s+(?P<hostname>\S+)\s+'
            r'(?P<app_name>\S+)\s+(?P<proc_id>\S+)\s+(?P<msg_id>\S+)\s+-\s+'
            r'(?P<message>.*)'
        )

        self.apache_pattern = re.compile(
            r'(?P<ip>[\d\.]+)\s+-\s+-\s+\[(?P<timestamp>[^\]]+)\]\s+'
            r'"(?P<method>\w+)\s+(?P<path>\S+)\s+HTTP/[\d\.]+"\s+'
            r'(?P<status>\d+)\s+(?P<size>\d+)\s+"(?P<referer>[^"]*)"\s+"(?P<user_agent>[^"]*)"'
        )

    def normalize_syslog(self, log_line: str) -> Dict[str, Any]:
        """Parse RFC 5424 syslog format"""
        match = self.syslog_pattern.match(log_line)
        if not match:
            return None

        return {
            'timestamp': match.group('timestamp'),
            'source': match.group('hostname'),
            'application': match.group('app_name'),
            'message': match.group('message'),
            'log_type': 'syslog',
            'priority': int(match.group('priority'))
        }

    def normalize_apache(self, log_line: str) -> Dict[str, Any]:
        """Parse Apache/Nginx combined log format"""
        match = self.apache_pattern.match(log_line)
        if not match:
            return None

        return {
            'timestamp': match.group('timestamp'),
            'source_ip': match.group('ip'),
            'http_method': match.group('method'),
            'http_path': match.group('path'),
            'http_status': int(match.group('status')),
            'response_size': int(match.group('size')),
            'user_agent': match.group('user_agent'),
            'log_type': 'http_access'
        }

    def normalize_json(self, log_line: str) -> Dict[str, Any]:
        """Parse JSON log format"""
        try:
            log_data = json.loads(log_line)
            log_data['log_type'] = 'json'
            return log_data
        except json.JSONDecodeError:
            return None

    def normalize(self, log_line: str) -> Dict[str, Any]:
        """
        Auto-detect format and normalize

        Returns normalized dict or None if unparseable
        """
        # Try JSON first (most structured)
        if log_line.strip().startswith('{'):
            return self.normalize_json(log_line)

        # Try syslog
        if log_line.startswith('<'):
            return self.normalize_syslog(log_line)

        # Try Apache/Nginx
        if re.match(r'^\d+\.\d+\.\d+\.\d+', log_line):
            return self.normalize_apache(log_line)

        # Unknown format
        return {
            'raw': log_line,
            'log_type': 'unknown',
            'timestamp': datetime.utcnow().isoformat()
        }

# Example usage
if __name__ == '__main__':
    normalizer = LogNormalizer()

    # Test logs from different sources
    test_logs = [
        '<134>1 2025-11-15T14:23:45.123Z firewall-01 kernel - - - Connection from 203.0.113.42 DENIED',
        '203.0.113.42 - - [15/Nov/2025:14:23:45 +0000] "GET /admin/login HTTP/1.1" 404 512 "-" "Mozilla/5.0"',
        '{"timestamp": "2025-11-15T14:23:45Z", "event": "auth_failure", "user": "admin", "source_ip": "203.0.113.42"}'
    ]

    for log in test_logs:
        normalized = normalizer.normalize(log)
        print(json.dumps(normalized, indent=2))
        print("---")

import pandas as pd
import numpy as np
from sklearn.ensemble import IsolationForest
from sklearn.preprocessing import StandardScaler
from datetime import datetime, timedelta

# Simulated security log data
# In production, this comes from SIEM or log aggregator
log_data = pd.DataFrame({
    'hour_of_day': [14, 2, 15, 3, 13, 2, 14, 16, 3, 2],
    'failed_login_count': [2, 15, 3, 45, 1, 87, 2, 3, 25, 150],
    'successful_login_count': [50, 5, 48, 3, 52, 2, 49, 51, 8, 0],
    'bytes_transferred_mb': [120, 500, 115, 2000, 110, 5000, 125, 118, 800, 15000],
    'unique_source_ips': [25, 3, 28, 1, 26, 1, 27, 24, 2, 1],
    'privileged_command_count': [5, 1, 6, 20, 4, 35, 5, 6, 15, 50],
    'dns_query_count': [1500, 500, 1450, 5000, 1520, 8000, 1480, 1510, 3500, 20000]
})

# Feature engineering: Add derived features
log_data['failed_to_success_ratio'] = (
    log_data['failed_login_count'] / (log_data['successful_login_count'] + 1)
)
log_data['is_off_hours'] = (log_data['hour_of_day'] < 6) | (log_data['hour_of_day'] > 20)

# Prepare features for anomaly detection
features = log_data[[
    'failed_login_count',
    'successful_login_count',
    'bytes_transferred_mb',
    'unique_source_ips',
    'privileged_command_count',
    'dns_query_count',
    'failed_to_success_ratio'
]]

# Normalize features
scaler = StandardScaler()
features_scaled = scaler.fit_transform(features)

# Train Isolation Forest
# contamination=0.2 means we expect 20% anomalies (tune based on your data)
iso_forest = IsolationForest(
    contamination=0.2,
    random_state=42,
    n_estimators=100
)
iso_forest.fit(features_scaled)

# Predict anomalies
log_data['anomaly_score'] = iso_forest.decision_function(features_scaled)
log_data['is_anomaly'] = iso_forest.predict(features_scaled)

# Sort by anomaly score (most anomalous first)
anomalies = log_data[log_data['is_anomaly'] == -1].sort_values('anomaly_score')

print("Top Anomalous Log Patterns Detected:\n")
print(anomalies[[
    'hour_of_day',
    'failed_login_count',
    'bytes_transferred_mb',
    'privileged_command_count',
    'anomaly_score'
]].to_string(index=False))

# Generate alerts for highest confidence anomalies
critical_threshold = anomalies['anomaly_score'].quantile(0.25)  # Bottom 25%

critical_anomalies = anomalies[anomalies['anomaly_score'] < critical_threshold]

print(f"\n[ALERT] {len(critical_anomalies)} CRITICAL ANOMALIES DETECTED")
for idx, row in critical_anomalies.iterrows():
    print(f"\nAnomaly #{idx}:")
    print(f"  Time: {row['hour_of_day']:02d}:00 {'(OFF-HOURS)' if row['is_off_hours'] else ''}")
    print(f"  Failed Logins: {row['failed_login_count']}")
    print(f"  Data Transfer: {row['bytes_transferred_mb']} MB")
    print(f"  Privileged Commands: {row['privileged_command_count']}")
    print(f"  Anomaly Score: {row['anomaly_score']:.4f}")

from transformers import pipeline
import pandas as pd

# Initialize text classification pipeline
# Using DistilBERT for efficiency (smaller, faster than BERT)
classifier = pipeline(
    "text-classification",
    model="distilbert-base-uncased-finetuned-sst-2-english",
    device=-1  # CPU; use 0+ for GPU
)

# Sample security log messages
log_messages = [
    "User login successful from known IP address",
    "Multiple failed authentication attempts detected from 203.0.113.42",
    "CRITICAL: Buffer overflow detected in network daemon process",
    "Scheduled backup completed successfully",
    "WARNING: Unusual outbound network traffic to unknown destination",
    "System update installed without errors",
    "ALERT: Ransomware encryption pattern detected on file server",
    "Routine system health check passed all tests"
]

# Classify each log message
results = []
for msg in log_messages:
    result = classifier(msg)[0]
    results.append({
        'message': msg,
        'sentiment': result['label'],
        'confidence': result['score']
    })

# Convert to DataFrame for analysis
df = pd.DataFrame(results)

# Filter for negative sentiment (potential threats)
threats = df[df['sentiment'] == 'NEGATIVE'].sort_values('confidence', ascending=False)

print("Potential Security Threats Identified:\n")
for idx, row in threats.iterrows():
    print(f"[THREAT] Confidence: {row['confidence']:.2%}")
    print(f"Message: {row['message']}\n")

from transformers import pipeline

# Initialize zero-shot classifier
classifier = pipeline(
    "zero-shot-classification",
    model="facebook/bart-large-mnli",
    device=-1
)

# Security log event
security_event = """
PowerShell executed with encoded command flag (-EncodedCommand) at 03:47 UTC.
Parent process: outlook.exe. Network connection established to 185.220.101.42:443
immediately after execution. Process created scheduled task for persistence.
"""

# MITRE ATT&CK Tactics (14 total)
attack_tactics = [
    "initial access",
    "execution",
    "persistence",
    "privilege escalation",
    "defense evasion",
    "credential access",
    "discovery",
    "lateral movement",
    "collection",
    "command and control",
    "exfiltration",
    "impact"
]

# Classify event (multi-label: event can map to multiple tactics)
result = classifier(security_event, attack_tactics, multi_label=True)

# Display top 5 matching tactics
print("MITRE ATT&CK Tactic Classification:\n")
for tactic, score in zip(result['labels'][:5], result['scores'][:5]):
    print(f"{tactic.upper()}: {score:.2%}")

# Automated response based on classification
if result['scores'][0] > 0.75:  # High confidence
    print(f"\n[AUTOMATED ALERT] High confidence {result['labels'][0].upper()} detected")
    print("Recommended actions:")

    if result['labels'][0] in ['execution', 'command and control']:
        print("  1. Isolate affected host immediately")
        print("  2. Capture memory dump for forensics")
        print("  3. Block C2 IP at firewall")
        print("  4. Disable compromised user account")
    elif result['labels'][0] == 'persistence':
        print("  1. Enumerate all scheduled tasks")
        print("  2. Remove malicious persistence mechanisms")
        print("  3. Audit startup programs and services")

import numpy as np
import pandas as pd
from sklearn.preprocessing import MinMaxScaler
import matplotlib.pyplot as plt

# Note: In production, use TensorFlow/Keras for full LSTM implementation
# This example demonstrates the data preparation and pattern detection approach

# Simulated hourly login counts (24 hours)
# Normal pattern: Low at night (0-6), high during business hours (9-17), low evening
normal_pattern = [
    2, 1, 1, 0, 1, 2,        # 00:00-05:59 (night - very low)
    5, 15, 45, 60, 55, 58,   # 06:00-11:59 (morning - increasing)
    50, 52, 48, 55, 53, 50,  # 12:00-17:59 (afternoon - steady high)
    30, 15, 8, 5, 3, 2       # 18:00-23:59 (evening - decreasing)
]

# Anomalous pattern: Spike at 2 AM (potential credential stuffing attack)
anomalous_pattern = [
    2, 150, 145, 140, 130, 5,  # ANOMALY: Spike 01:00-04:59
    5, 15, 45, 60, 55, 58,
    50, 52, 48, 55, 53, 50,
    30, 15, 8, 5, 3, 2
]

def detect_temporal_anomaly(pattern, baseline, threshold=2.0):
    """
    Detect anomalies using statistical deviation from baseline

    Args:
        pattern: Current pattern to analyze
        baseline: Expected normal pattern
        threshold: Standard deviations from mean to flag as anomaly

    Returns:
        List of anomalous hours
    """
    pattern = np.array(pattern)
    baseline = np.array(baseline)

    # Calculate z-scores
    mean = np.mean(baseline)
    std = np.std(baseline)

    z_scores = (pattern - mean) / (std + 1e-10)  # Avoid division by zero

    # Identify anomalies
    anomalies = []
    for hour, z_score in enumerate(z_scores):
        if abs(z_score) > threshold:
            anomalies.append({
                'hour': hour,
                'count': pattern[hour],
                'expected': baseline[hour],
                'z_score': z_score,
                'deviation_pct': ((pattern[hour] - baseline[hour]) / (baseline[hour] + 1)) * 100
            })

    return anomalies

# Detect anomalies
anomalies = detect_temporal_anomaly(
    pattern=anomalous_pattern,
    baseline=normal_pattern,
    threshold=2.0
)

print("Temporal Anomalies Detected:\n")
for anomaly in anomalies:
    print(f"Hour: {anomaly['hour']:02d}:00")
    print(f"  Count: {anomaly['count']} (Expected: {anomaly['expected']})")
    print(f"  Deviation: {anomaly['deviation_pct']:+.1f}%")
    print(f"  Z-Score: {anomaly['z_score']:.2f}")

    if anomaly['z_score'] > 4.0:
        print(f"  [CRITICAL ALERT] Severe anomaly detected")
        print(f"  Possible credential stuffing or brute force attack")
    print()

import networkx as nx
from collections import defaultdict

# Simulated network connection logs
connections = [
    {'src': '10.0.1.10', 'dst': '10.0.2.5', 'port': 445, 'user': 'alice'},
    {'src': '10.0.1.10', 'dst': '10.0.2.8', 'port': 445, 'user': 'alice'},
    {'src': '10.0.2.5', 'dst': '10.0.3.12', 'port': 445, 'user': 'alice'},
    {'src': '10.0.2.5', 'dst': '10.0.3.15', 'port': 445, 'user': 'alice'},
    {'src': '10.0.2.8', 'dst': '10.0.4.20', 'port': 445, 'user': 'alice'},
    {'src': '10.0.2.8', 'dst': '10.0.4.22', 'port': 445, 'user': 'alice'},
    {'src': '10.0.3.12', 'dst': '10.0.5.30', 'port': 445, 'user': 'alice'},
    # Normal admin connections
    {'src': '10.0.1.100', 'dst': '10.0.1.101', 'port': 22, 'user': 'admin'},
]

def detect_lateral_movement(connections, threshold=3):
    """
    Detect potential lateral movement using graph analysis

    Lateral movement indicators:
    - Single user authenticating to multiple hosts
    - Cascading connections (A→B→C→D pattern)
    - SMB/RDP connections (ports 445, 3389)

    Args:
        connections: List of connection dicts
        threshold: Minimum unique destinations to flag

    Returns:
        List of suspicious movement patterns
    """
    # Build graph
    G = nx.DiGraph()

    # Track user activity
    user_connections = defaultdict(list)

    for conn in connections:
        # Add edge to graph
        G.add_edge(
            conn['src'],
            conn['dst'],
            port=conn['port'],
            user=conn['user']
        )

        # Track per-user
        user_connections[conn['user']].append(conn)

    # Analyze for lateral movement patterns
    suspicious_users = []

    for user, conns in user_connections.items():
        unique_sources = len(set(c['src'] for c in conns))
        unique_destinations = len(set(c['dst'] for c in conns))

        # Check for lateral movement indicators
        if unique_destinations >= threshold:
            # Calculate path depth (how many hops from initial compromise)
            paths = []
            sources = list(set(c['src'] for c in conns))

            for source in sources:
                destinations = [c['dst'] for c in conns if c['src'] == source]
                for dest in destinations:
                    if nx.has_path(G, source, dest):
                        path = nx.shortest_path(G, source, dest)
                        paths.append(path)

            max_depth = max(len(p) - 1 for p in paths) if paths else 0

            suspicious_users.append({
                'user': user,
                'unique_sources': unique_sources,
                'unique_destinations': unique_destinations,
                'total_connections': len(conns),
                'max_hop_depth': max_depth,
                'connection_paths': paths,
                'smb_connections': sum(1 for c in conns if c['port'] == 445)
            })

    return suspicious_users

# Detect lateral movement
suspicious = detect_lateral_movement(connections, threshold=3)

print("Lateral Movement Detection Results:\n")
for user_data in suspicious:
    print(f"[ALERT] User: {user_data['user']}")
    print(f"  Unique Destinations: {user_data['unique_destinations']}")
    print(f"  Total Connections: {user_data['total_connections']}")
    print(f"  SMB Connections: {user_data['smb_connections']}")
    print(f"  Max Hop Depth: {user_data['max_hop_depth']}")

    if user_data['smb_connections'] >= 5 and user_data['max_hop_depth'] >= 3:
        print(f"\n  [CRITICAL] HIGH CONFIDENCE LATERAL MOVEMENT")
        print(f"  Indicators:")
        print(f"    - Cascading SMB connections across {user_data['max_hop_depth']} network hops")
        print(f"    - {user_data['unique_destinations']} unique destinations")
        print(f"    - MITRE ATT&CK: T1021.002 (SMB/Windows Admin Shares)")
        print(f"\n  Recommended Actions:")
        print(f"    1. Disable account '{user_data['user']}' immediately")
        print(f"    2. Isolate all {user_data['unique_destinations']} affected hosts")
        print(f"    3. Force password reset for '{user_data['user']}'")
        print(f"    4. Initiate incident response procedure")

    print("\n  Connection Path:")
    for path in user_data['connection_paths'][:3]:  # Show first 3 paths
        print(f"    {' → '.join(path)}")
    print()

import json
from datetime import datetime, timedelta
from typing import List, Dict, Any

class SecurityLogAnalysisPipeline:
    """
    End-to-end AI-powered log analysis pipeline

    Stages:
    1. Ingest & normalize logs
    2. Feature extraction
    3. ML-based anomaly detection
    4. NLP-based threat classification
    5. Temporal pattern analysis
    6. Alert generation & prioritization
    """

    def __init__(self):
        self.normalizer = LogNormalizer()
        # In production, load pre-trained models here
        self.anomaly_detector = None  # IsolationForest
        self.threat_classifier = None  # Transformer model
        self.alert_threshold = 0.75

    def ingest_logs(self, log_file_path: str) -> List[Dict[str, Any]]:
        """Read and normalize logs from file"""
        normalized_logs = []

        with open(log_file_path, 'r') as f:
            for line in f:
                normalized = self.normalizer.normalize(line.strip())
                if normalized:
                    normalized_logs.append(normalized)

        return normalized_logs

    def extract_features(self, logs: List[Dict[str, Any]]) -> pd.DataFrame:
        """
        Extract numerical features for ML analysis

        Features extracted:
        - Failed login count per hour
        - Successful login count per hour
        - Unique source IPs per hour
        - Data transfer volume per hour
        - Privileged command count per hour
        """
        # Group logs by hour
        hourly_aggregates = defaultdict(lambda: {
            'failed_logins': 0,
            'successful_logins': 0,
            'unique_ips': set(),
            'data_bytes': 0,
            'privileged_commands': 0
        })

        for log in logs:
            # Parse timestamp
            try:
                ts = datetime.fromisoformat(log.get('timestamp', ''))
                hour_key = ts.strftime('%Y-%m-%d %H:00')
            except:
                continue

            # Aggregate metrics
            agg = hourly_aggregates[hour_key]

            if 'failed' in log.get('message', '').lower():
                agg['failed_logins'] += 1
            elif 'success' in log.get('message', '').lower():
                agg['successful_logins'] += 1

            if 'source_ip' in log:
                agg['unique_ips'].add(log['source_ip'])

            if 'response_size' in log:
                agg['data_bytes'] += log['response_size']

            if log.get('privileged', False):
                agg['privileged_commands'] += 1

        # Convert to DataFrame
        data = []
        for hour, metrics in hourly_aggregates.items():
            data.append({
                'timestamp': hour,
                'failed_logins': metrics['failed_logins'],
                'successful_logins': metrics['successful_logins'],
                'unique_ips': len(metrics['unique_ips']),
                'data_bytes': metrics['data_bytes'],
                'privileged_commands': metrics['privileged_commands']
            })

        return pd.DataFrame(data)

    def analyze(self, log_file_path: str) -> Dict[str, Any]:
        """
        Run complete analysis pipeline

        Returns analysis results with alerts
        """
        # Stage 1: Ingest
        logs = self.ingest_logs(log_file_path)

        # Stage 2: Feature extraction
        features = self.extract_features(logs)

        # Stage 3: Anomaly detection
        # (Implementation would use trained IsolationForest model)

        # Stage 4: NLP threat classification
        # (Implementation would use trained transformer model)

        # Stage 5: Alert generation
        alerts = self.generate_alerts(logs, features)

        return {
            'timestamp': datetime.utcnow().isoformat(),
            'logs_analyzed': len(logs),
            'alerts_generated': len(alerts),
            'critical_alerts': len([a for a in alerts if a['severity'] == 'critical']),
            'alerts': alerts
        }

    def generate_alerts(self, logs, features) -> List[Dict[str, Any]]:
        """Generate prioritized security alerts"""
        # Implementation would aggregate findings from all analysis stages
        # and generate actionable alerts with context
        return []

# Example usage
if __name__ == '__main__':
    pipeline = SecurityLogAnalysisPipeline()

    # In production, this would point to your log aggregation system
    analysis_results = pipeline.analyze('/var/log/security/combined.log')

    print(json.dumps(analysis_results, indent=2))