Encrypted Traffic Analysis

# Capture encrypted traffic with complete metadata
tcpdump -i eth0 -s 0 -w encrypted_capture.pcap 'port 443 or port 22 or port 993 or port 995'
# Captures HTTPS, SSH, IMAPS, POP3S traffic
# -s 0: Full packet capture including headers
# Preserves all metadata for analysis

# Analyze encrypted protocol distribution
tshark -r encrypted_capture.pcap -q -z conv,tcp
# Shows TCP conversation statistics
# Identifies most active encrypted connections
# Reveals communication patterns between hosts

# Extract TLS connection metadata
tshark -r encrypted_capture.pcap -Y "tls.handshake.type == 1" \
-T fields -e ip.src -e ip.dst -e tls.handshake.ciphersuites -e tls.handshake.extensions_server_name
# TLS Client Hello analysis
# Shows cipher suite preferences and SNI information
# Reveals application and server identification

#!/usr/bin/env python3
import pyshark
from collections import defaultdict
import matplotlib.pyplot as plt
import numpy as np

def analyze_traffic_patterns(capture_file):
    """Analyze encrypted traffic patterns and flows"""
    
    capture = pyshark.FileCapture(capture_file, display_filter="tls")
    
    flows = defaultdict(list)
    packet_sizes = defaultdict(list)
    timing_data = defaultdict(list)
    
    for packet in capture:
        try:
            flow_key = f"{packet.ip.src}:{packet.tcp.srcport}->{packet.ip.dst}:{packet.tcp.dstport}"
            
            # Collect packet sizes
            packet_size = int(packet.length)
            packet_sizes[flow_key].append(packet_size)
            
            # Collect timing information
            timestamp = float(packet.sniff_timestamp)
            timing_data[flow_key].append(timestamp)
            
            # Track flow characteristics
            flows[flow_key].append({
                'timestamp': timestamp,
                'size': packet_size,
                'direction': 'out' if packet.ip.src < packet.ip.dst else 'in'
            })
            
        except AttributeError:
            continue
    
    return flows, packet_sizes, timing_data

def generate_flow_statistics(flows):
    """Generate statistical analysis of traffic flows"""
    
    statistics = {}
    
    for flow, packets in flows.items():
        sizes = [p['size'] for p in packets]
        times = [p['timestamp'] for p in packets]
        
        if len(times) > 1:
            intervals = [times[i+1] - times[i] for i in range(len(times)-1)]
            
            statistics[flow] = {
                'total_packets': len(packets),
                'total_bytes': sum(sizes),
                'avg_packet_size': np.mean(sizes),
                'std_packet_size': np.std(sizes),
                'avg_interval': np.mean(intervals),
                'std_interval': np.std(intervals),
                'duration': max(times) - min(times)
            }
    
    return statistics

# Analyze encrypted traffic patterns
flows, sizes, timing = analyze_traffic_patterns("encrypted_capture.pcap")
stats = generate_flow_statistics(flows)

print(f"Analyzed {len(flows)} encrypted flows")
for flow, stat in list(stats.items())[:5]:
    print(f"Flow {flow}: {stat['total_packets']} packets, {stat['total_bytes']} bytes")

# Extract TLS certificate information
tshark -r encrypted_capture.pcap -Y "tls.handshake.type == 11" \
-T fields -e ip.dst -e tls.handshake.certificate
# Server certificate analysis
# Reveals server identity and organization
# Identifies specific services and applications

# Analyze HTTP/2 over TLS patterns
tshark -r encrypted_capture.pcap -Y "http2" \
-T fields -e ip.src -e ip.dst -e http2.stream.id -e http2.headers.name
# HTTP/2 header analysis (if visible)
# Shows application-layer protocol usage
# Identifies web application characteristics

#!/usr/bin/env python3
import pyshark
import statistics

class ApplicationFingerprinter:
    def __init__(self):
        self.signatures = {
            'web_browsing': {'avg_size': (400, 1500), 'frequency': 'irregular'},
            'email_client': {'avg_size': (200, 800), 'frequency': 'periodic'},
            'file_transfer': {'avg_size': (1400, 1500), 'frequency': 'sustained'},
            'video_streaming': {'avg_size': (1200, 1500), 'frequency': 'constant'},
            'voip': {'avg_size': (100, 300), 'frequency': 'regular'}
        }
    
    def analyze_flow(self, packets):
        """Analyze flow characteristics for application identification"""
        
        sizes = [int(p.length) for p in packets]
        times = [float(p.sniff_timestamp) for p in packets]
        
        if len(sizes) < 10 or len(times) < 10:
            return "insufficient_data"
        
        avg_size = statistics.mean(sizes)
        size_variance = statistics.variance(sizes)
        
        # Calculate timing regularity
        intervals = [times[i+1] - times[i] for i in range(len(times)-1)]
        interval_variance = statistics.variance(intervals) if len(intervals) > 1 else 0
        
        # Match against signatures
        for app, signature in self.signatures.items():
            size_min, size_max = signature['avg_size']
            if size_min <= avg_size <= size_max:
                if signature['frequency'] == 'regular' and interval_variance < 1.0:
                    return app
                elif signature['frequency'] == 'irregular' and interval_variance > 5.0:
                    return app
                elif signature['frequency'] == 'constant' and interval_variance < 0.5:
                    return app
        
        return "unknown"
    
    def fingerprint_capture(self, capture_file):
        """Fingerprint all flows in capture"""
        
        capture = pyshark.FileCapture(capture_file, display_filter="tls")
        
        flows = {}
        current_flow = None
        flow_packets = []
        
        for packet in capture:
            try:
                flow_key = f"{packet.ip.src}:{packet.tcp.srcport}->{packet.ip.dst}:{packet.tcp.dstport}"
                
                if flow_key != current_flow:
                    if current_flow and flow_packets:
                        flows[current_flow] = self.analyze_flow(flow_packets)
                    current_flow = flow_key
                    flow_packets = []
                
                flow_packets.append(packet)
                
            except AttributeError:
                continue
        
        # Process last flow
        if current_flow and flow_packets:
            flows[current_flow] = self.analyze_flow(flow_packets)
        
        return flows

# Fingerprint applications in encrypted traffic
fingerprinter = ApplicationFingerprinter()
app_flows = fingerprinter.fingerprint_capture("encrypted_capture.pcap")

for flow, app in app_flows.items():
    print(f"{flow}: {app}")

#!/usr/bin/env python3
import pyshark
import matplotlib.pyplot as plt
from datetime import datetime
import numpy as np

def timing_correlation_analysis(capture_file, reference_events=None):
    """Correlate encrypted traffic timing with external events"""
    
    capture = pyshark.FileCapture(capture_file, display_filter="tls")
    
    connection_times = []
    data_volumes = []
    
    for packet in capture:
        try:
            if hasattr(packet.tls, 'handshake_type') and packet.tls.handshake_type == '1':
                # TLS Client Hello - new connection
                timestamp = float(packet.sniff_timestamp)
                connection_times.append(timestamp)
        except AttributeError:
            continue
    
    # Analyze connection timing patterns
    if len(connection_times) > 1:
        intervals = [connection_times[i+1] - connection_times[i] for i in range(len(connection_times)-1)]
        
        # Statistical analysis
        mean_interval = np.mean(intervals)
        std_interval = np.std(intervals)
        
        print(f"Connection analysis:")
        print(f"  Total connections: {len(connection_times)}")
        print(f"  Mean interval: {mean_interval:.2f} seconds")
        print(f"  Std deviation: {std_interval:.2f} seconds")
        
        # Identify potential automation vs human patterns
        if std_interval < mean_interval * 0.1:
            print("  Pattern: Likely automated (low variance)")
        elif std_interval > mean_interval * 0.5:
            print("  Pattern: Likely human-driven (high variance)")
        else:
            print("  Pattern: Mixed or unknown")
    
    return connection_times, intervals

# Perform timing correlation analysis
times, intervals = timing_correlation_analysis("encrypted_capture.pcap")

# Extract comprehensive TLS metadata
tshark -r encrypted_capture.pcap -Y "tls.handshake.type == 1" \
-T fields -e frame.time -e ip.src -e ip.dst -e tcp.dstport \
-e tls.handshake.version -e tls.handshake.extensions_server_name \
-e tls.handshake.ciphersuites > tls_metadata.csv
# Comprehensive TLS connection metadata
# Timing, endpoints, versions, server names
# Cipher suite preferences reveal client applications

# Analyze connection duration patterns
tshark -r encrypted_capture.pcap -Y "tcp.flags.fin == 1" \
-T fields -e tcp.stream -e frame.time > connection_endings.csv
# TCP connection termination times
# Combined with start times shows session durations
# Reveals usage patterns and session characteristics

#!/usr/bin/env python3
import csv
import pyshark
from collections import defaultdict
import json

class EncryptedTrafficIntelligence:
    def __init__(self):
        self.intelligence = {
            'connections': [],
            'servers': defaultdict(int),
            'timing_patterns': [],
            'cipher_preferences': defaultdict(int),
            'suspicious_patterns': []
        }
    
    def analyze_capture(self, capture_file):
        """Comprehensive encrypted traffic intelligence extraction"""
        
        capture = pyshark.FileCapture(capture_file)
        
        for packet in capture:
            try:
                # TLS connection analysis
                if hasattr(packet, 'tls') and hasattr(packet.tls, 'handshake_type'):
                    if packet.tls.handshake_type == '1':  # Client Hello
                        self.analyze_client_hello(packet)
                    elif packet.tls.handshake_type == '2':  # Server Hello
                        self.analyze_server_hello(packet)
                
                # General connection metadata
                if hasattr(packet, 'tcp') and hasattr(packet, 'ip'):
                    self.track_connection_metadata(packet)
                    
            except AttributeError:
                continue
    
    def analyze_client_hello(self, packet):
        """Analyze TLS Client Hello for intelligence"""
        
        connection = {
            'timestamp': float(packet.sniff_timestamp),
            'src': packet.ip.src,
            'dst': packet.ip.dst,
            'port': packet.tcp.dstport,
            'tls_version': getattr(packet.tls, 'handshake_version', 'unknown'),
            'server_name': getattr(packet.tls, 'handshake_extensions_server_name', 'unknown')
        }
        
        self.intelligence['connections'].append(connection)
        
        # Track server popularity
        if connection['server_name'] != 'unknown':
            self.intelligence['servers'][connection['server_name']] += 1
    
    def analyze_server_hello(self, packet):
        """Analyze TLS Server Hello for intelligence"""
        
        if hasattr(packet.tls, 'handshake_ciphersuite'):
            cipher = packet.tls.handshake_ciphersuite
            self.intelligence['cipher_preferences'][cipher] += 1
    
    def track_connection_metadata(self, packet):
        """Track general connection patterns"""
        
        # Look for suspicious patterns
        if hasattr(packet, 'tcp'):
            # Detect potential covert channels
            if int(packet.tcp.window_size) in [1337, 31337, 8080]:
                self.intelligence['suspicious_patterns'].append({
                    'type': 'suspicious_window_size',
                    'value': packet.tcp.window_size,
                    'src': packet.ip.src,
                    'dst': packet.ip.dst
                })
    
    def generate_report(self):
        """Generate intelligence report"""
        
        report = {
            'summary': {
                'total_connections': len(self.intelligence['connections']),
                'unique_servers': len(self.intelligence['servers']),
                'cipher_suites': len(self.intelligence['cipher_preferences']),
                'suspicious_indicators': len(self.intelligence['suspicious_patterns'])
            },
            'top_servers': dict(sorted(self.intelligence['servers'].items(), 
                                     key=lambda x: x[1], reverse=True)[:10]),
            'cipher_distribution': dict(self.intelligence['cipher_preferences']),
            'suspicious_activity': self.intelligence['suspicious_patterns']
        }
        
        return report

# Generate comprehensive encrypted traffic intelligence
intel = EncryptedTrafficIntelligence()
intel.analyze_capture("encrypted_capture.pcap")
report = intel.generate_report()

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

#!/usr/bin/env python3
import pyshark
import numpy as np

def side_channel_analysis(capture_file):
    """Analyze side-channel information in encrypted traffic"""
    
    capture = pyshark.FileCapture(capture_file, display_filter="tls")
    
    packet_intervals = []
    packet_sizes = []
    
    prev_time = None
    
    for packet in capture:
        try:
            current_time = float(packet.sniff_timestamp)
            size = int(packet.length)
            
            if prev_time:
                interval = current_time - prev_time
                packet_intervals.append(interval)
            
            packet_sizes.append(size)
            prev_time = current_time
            
        except (AttributeError, ValueError):
            continue
    
    # Statistical analysis for side-channel patterns
    if packet_intervals:
        # Timing analysis
        timing_entropy = np.var(packet_intervals)
        
        # Size analysis
        size_entropy = np.var(packet_sizes)
        
        print(f"Timing entropy: {timing_entropy:.4f}")
        print(f"Size entropy: {size_entropy:.4f}")
        
        # Look for patterns that might indicate specific applications
        if timing_entropy < 0.001:
            print("Highly regular timing - possible automated system")
        elif timing_entropy > 1.0:
            print("Irregular timing - likely human interaction")

side_channel_analysis("encrypted_capture.pcap")

# Analyze DoH traffic patterns
tshark -r encrypted_capture.pcap -Y "http2 and tcp.port == 443" \
-T fields -e ip.src -e ip.dst -e http2.stream.id
# DoH uses HTTP/2 over port 443
# Different pattern from regular HTTPS browsing
# Identifies DNS query behavior through encrypted channel

# Correlate DoH with subsequent connections
# Look for DoH queries followed by new TLS connections
# Reveals browsing patterns despite encryption

#!/usr/bin/env python3
import pyshark

def analyze_vpn_traffic(capture_file):
    """Analyze VPN traffic for behavioral patterns"""
    
    capture = pyshark.FileCapture(capture_file, display_filter="udp.port == 1194 or esp")
    
    vpn_sessions = {}
    
    for packet in capture:
        try:
            # OpenVPN typically uses UDP 1194
            # IPSec uses ESP protocol
            
            if hasattr(packet, 'udp') and packet.udp.dstport == '1194':
                # OpenVPN traffic
                session_key = f"{packet.ip.src}->{packet.ip.dst}"
                if session_key not in vpn_sessions:
                    vpn_sessions[session_key] = []
                
                vpn_sessions[session_key].append({
                    'timestamp': float(packet.sniff_timestamp),
                    'size': int(packet.length),
                    'protocol': 'OpenVPN'
                })
            
            elif hasattr(packet, 'esp'):
                # IPSec ESP traffic
                session_key = f"{packet.ip.src}->{packet.ip.dst}"
                if session_key not in vpn_sessions:
                    vpn_sessions[session_key] = []
                
                vpn_sessions[session_key].append({
                    'timestamp': float(packet.sniff_timestamp),
                    'size': int(packet.length),
                    'protocol': 'IPSec'
                })
                
        except AttributeError:
            continue
    
    return vpn_sessions

# Analyze VPN usage patterns
vpn_data = analyze_vpn_traffic("encrypted_capture.pcap")
for session, packets in vpn_data.items():
    print(f"VPN Session {session}: {len(packets)} packets")

#!/usr/bin/env python3
import pyshark
import numpy as np
from sklearn.cluster import KMeans
from sklearn.preprocessing import StandardScaler

def ml_traffic_classification(capture_file):
    """Use machine learning for encrypted traffic classification"""
    
    capture = pyshark.FileCapture(capture_file, display_filter="tls")
    
    features = []
    
    for packet in capture:
        try:
            # Extract features for ML classification
            feature_vector = [
                int(packet.length),  # Packet size
                float(packet.sniff_timestamp) % 86400,  # Time of day
                int(packet.tcp.window_size),  # TCP window size
                len(getattr(packet.tls, 'handshake_extensions_server_name', ''))  # SNI length
            ]
            
            features.append(feature_vector)
            
        except (AttributeError, ValueError):
            continue
    
    if len(features) > 10:
        # Normalize features
        scaler = StandardScaler()
        features_normalized = scaler.fit_transform(features)
        
        # Cluster analysis
        kmeans = KMeans(n_clusters=5)
        clusters = kmeans.fit_predict(features_normalized)
        
        # Analyze clusters
        unique_clusters = np.unique(clusters)
        for cluster in unique_clusters:
            cluster_size = np.sum(clusters == cluster)
            print(f"Cluster {cluster}: {cluster_size} packets")
    
    return features, clusters if 'clusters' in locals() else None

# Apply machine learning to encrypted traffic
features, clusters = ml_traffic_classification("encrypted_capture.pcap")

#!/usr/bin/env python3
import json
import statistics
from datetime import datetime, timedelta

class EncryptedTrafficBaseline:
    def __init__(self):
        self.baseline = {
            'hourly_patterns': {},
            'application_signatures': {},
            'normal_ranges': {}
        }
    
    def establish_baseline(self, capture_files):
        """Establish normal encrypted traffic patterns"""
        
        all_connections = []
        
        for capture_file in capture_files:
            # Process each capture file
            # Collect connection metadata, timing, sizes
            pass  # Implementation details...
        
        # Statistical analysis to establish normal ranges
        if all_connections:
            connection_rates = [c['rate'] for c in all_connections]
            packet_sizes = [c['avg_size'] for c in all_connections]
            
            self.baseline['normal_ranges'] = {
                'connection_rate': {
                    'mean': statistics.mean(connection_rates),
                    'std': statistics.stdev(connection_rates),
                    'min': min(connection_rates),
                    'max': max(connection_rates)
                },
                'packet_size': {
                    'mean': statistics.mean(packet_sizes),
                    'std': statistics.stdev(packet_sizes),
                    'min': min(packet_sizes),
                    'max': max(packet_sizes)
                }
            }
    
    def detect_anomalies(self, new_capture):
        """Detect anomalies based on established baseline"""
        
        # Compare new traffic against baseline
        # Flag unusual patterns or behaviors
        pass  # Implementation details...

# Establish behavioral baseline for encrypted traffic
baseline = EncryptedTrafficBaseline()
baseline.establish_baseline(["day1.pcap", "day2.pcap", "day3.pcap"])

# Implement traffic mixing and obfuscation
# Use Tor or similar onion routing
# Deploy traffic analysis resistance measures
# Implement decoy traffic generation