QUIC and HTTP/3 Security: Comprehensive Security Analysis and Implementation#

1
# QUIC Protocol Architecture Analysis
2
import struct
3
import socket
4
import time
5
import ssl
6
import hashlib
7
from dataclasses import dataclass
8
from typing import Dict, List, Optional, Tuple
9
from enum import Enum
10

11
class QUICPacketType(Enum):
12
    INITIAL = 0x00
13
    ZERO_RTT = 0x01
14
    HANDSHAKE = 0x02
15
    RETRY = 0x03
16
    VERSION_NEGOTIATION = 0xff
17
    SHORT_HEADER = 0x40
18

19
@dataclass
20
class QUICPacket:
21
    packet_type: QUICPacketType
22
    version: int
23
    destination_conn_id: bytes
24
    source_conn_id: bytes
25
    payload_length: int
26
    packet_number: int
27
    payload: bytes
28
    encrypted: bool = True
29
    timestamp: float = 0.0
30

31
@dataclass
32
class QUICConnection:
33
    connection_id: bytes
34
    version: int
35
    peer_address: Tuple[str, int]
36
    state: str
37
    encryption_level: str
38
    rtt_estimate: float
39
    congestion_window: int
40
    created_at: float
41
    last_activity: float
42

43
class QUICSecurityAnalyzer:
44
    """Analyze QUIC traffic for security threats"""
45

46
    def __init__(self):
47
        self.connections = {}  # connection_id -> QUICConnection
48
        self.packet_history = []
49
        self.security_events = []
50

51
        # Attack detection thresholds
52
        self.max_connections_per_ip = 1000
53
        self.max_packets_per_second = 10000
54
        self.suspicious_packet_sizes = [1, 2, 3, 4, 65535]  # Common probe sizes
55

56
    def analyze_packet(self, packet: QUICPacket, source_ip: str) -> Dict:
57
        """Analyze a QUIC packet for security threats"""
58
        threats_detected = []
59

60
        # 1. Connection flooding detection
61
        if self._detect_connection_flooding(source_ip):
62
            threats_detected.append("connection_flooding")
63

64
        # 2. Amplification attack detection
65
        if self._detect_amplification_attack(packet, source_ip):
66
            threats_detected.append("amplification_attack")
67

68
        # 3. 0-RTT replay attack detection
69
        if self._detect_0rtt_replay(packet):
70
            threats_detected.append("0rtt_replay")
71

72
        # 4. Version downgrade detection
73
        if self._detect_version_downgrade(packet):
74
            threats_detected.append("version_downgrade")
75

76
        # 5. Connection migration abuse
77
        if self._detect_migration_abuse(packet):
78
            threats_detected.append("migration_abuse")
79

80
        # Update packet history
81
        packet.timestamp = time.time()
82
        self.packet_history.append(packet)
83

84
        # Maintain sliding window
85
        cutoff = time.time() - 300  # 5 minutes
86
        self.packet_history = [p for p in self.packet_history if p.timestamp > cutoff]
87

88
        analysis_result = {
89
            'threats': threats_detected,
90
            'risk_score': self._calculate_risk_score(threats_detected),
91
            'timestamp': packet.timestamp,
92
            'packet_type': packet.packet_type.name,
93
            'source_ip': source_ip
94
        }
95

96
        if threats_detected:
97
            self.security_events.append(analysis_result)
98
            self._trigger_security_response(analysis_result)
99

100
        return analysis_result
101

102
    def _detect_connection_flooding(self, source_ip: str) -> bool:
103
        """Detect connection flooding attacks"""
104
        recent_connections = [
105
            conn for conn in self.connections.values()
106
            if conn.peer_address[0] == source_ip and
107
            time.time() - conn.created_at < 60  # Last minute
108
        ]
109

110
        return len(recent_connections) > self.max_connections_per_ip / 60
111

112
    def _detect_amplification_attack(self, packet: QUICPacket, source_ip: str) -> bool:
113
        """Detect UDP amplification attacks"""
114
        # Small initial packets that might trigger large responses
115
        if (packet.packet_type == QUICPacketType.INITIAL and
116
            packet.payload_length in self.suspicious_packet_sizes):
117
            return True
118

119
        # Check for spoofed source addresses (simplified)
120
        if packet.packet_type == QUICPacketType.VERSION_NEGOTIATION:
121
            return True  # Version negotiation can be abused
122

123
        return False
124

125
    def _detect_0rtt_replay(self, packet: QUICPacket) -> bool:
126
        """Detect 0-RTT replay attacks"""
127
        if packet.packet_type != QUICPacketType.ZERO_RTT:
128
            return False
129

130
        # Check for duplicate 0-RTT packets
131
        recent_0rtt = [
132
            p for p in self.packet_history
133
            if (p.packet_type == QUICPacketType.ZERO_RTT and
134
                p.destination_conn_id == packet.destination_conn_id and
135
                time.time() - p.timestamp < 10)  # Within 10 seconds
136
        ]
137

138
        return len(recent_0rtt) > 1
139

140
    def _detect_version_downgrade(self, packet: QUICPacket) -> bool:
141
        """Detect version downgrade attacks"""
142
        if packet.packet_type == QUICPacketType.INITIAL:
143
            # Check if version is suspiciously old
144
            known_versions = [0x00000001, 0x1234abcd]  # QUIC v1 and draft versions
145
            return packet.version not in known_versions
146

147
        return False
148

149
    def _detect_migration_abuse(self, packet: QUICPacket) -> bool:
150
        """Detect connection migration abuse"""
151
        conn_id = packet.destination_conn_id
152
        if conn_id in self.connections:
153
            conn = self.connections[conn_id]
154

155
            # Too many migrations in short time
156
            migration_window = 60  # 1 minute
157
            if time.time() - conn.created_at < migration_window:
158
                # Count recent activity from different addresses would be checked here
159
                pass
160

161
        return False
162

163
    def _calculate_risk_score(self, threats: List[str]) -> int:
164
        """Calculate risk score based on detected threats"""
165
        threat_scores = {
166
            'connection_flooding': 40,
167
            'amplification_attack': 60,
168
            '0rtt_replay': 80,
169
            'version_downgrade': 30,
170
            'migration_abuse': 20
171
        }
172

173
        total_score = sum(threat_scores.get(threat, 10) for threat in threats)
174
        return min(total_score, 100)
175

176
    def _trigger_security_response(self, analysis_result: Dict):
177
        """Trigger security response based on analysis"""
178
        risk_score = analysis_result['risk_score']
179
        source_ip = analysis_result['source_ip']
180

181
        if risk_score >= 80:
182
            print(f"🚨 CRITICAL: High-risk QUIC traffic from {source_ip} "
183
                  f"(Risk: {risk_score}, Threats: {analysis_result['threats']})")
184
        elif risk_score >= 40:
185
            print(f"⚠️ WARNING: Suspicious QUIC traffic from {source_ip} "
186
                  f"(Risk: {risk_score}, Threats: {analysis_result['threats']})")

1
# UDP-based Attack Detection and Mitigation
2
import socket
3
import struct
4
import time
5
from collections import defaultdict
6

7
class QUICFirewall:
8
    """Advanced QUIC-aware firewall implementation"""
9

10
    def __init__(self):
11
        self.rate_limits = defaultdict(lambda: {'packets': 0, 'window_start': time.time()})
12
        self.blocked_ips = {}  # IP -> block_until_timestamp
13
        self.connection_states = {}
14

15
        # Security policies
16
        self.policies = {
17
            'max_packets_per_second': 1000,
18
            'max_new_connections_per_minute': 100,
19
            'block_duration': 300,  # 5 minutes
20
            'enable_amplification_protection': True,
21
            'require_address_validation': True
22
        }
23

24
    def process_quic_packet(self, packet_data: bytes, source_addr: Tuple[str, int],
25
                           dest_addr: Tuple[str, int]) -> bool:
26
        """Process QUIC packet and determine if it should be allowed"""
27
        source_ip = source_addr[0]
28

29
        # Check if IP is blocked
30
        if self._is_blocked(source_ip):
31
            return False
32

33
        # Parse QUIC packet header
34
        try:
35
            packet_info = self._parse_quic_header(packet_data)
36
        except Exception:
37
            # Malformed packet
38
            self._record_violation(source_ip, "malformed_packet")
39
            return False
40

41
        # Rate limiting
42
        if not self._check_rate_limit(source_ip):
43
            self._block_ip(source_ip, "rate_limit_exceeded")
44
            return False
45

46
        # Amplification protection
47
        if self.policies['enable_amplification_protection']:
48
            if not self._check_amplification_protection(packet_info, source_addr):
49
                self._record_violation(source_ip, "amplification_attempt")
50
                return False
51

52
        # Address validation for new connections
53
        if (packet_info['type'] == 'INITIAL' and
54
            self.policies['require_address_validation']):
55
            if not self._validate_address(packet_info, source_addr):
56
                return False
57

58
        # Update connection state
59
        self._update_connection_state(packet_info, source_addr)
60

61
        return True
62

63
    def _parse_quic_header(self, data: bytes) -> Dict:
64
        """Parse QUIC packet header"""
65
        if len(data) < 1:
66
            raise ValueError("Packet too short")
67

68
        first_byte = data[0]
69
        packet_info = {}
70

71
        # Determine packet type
72
        if first_byte & 0x80:  # Long header
73
            if len(data) < 6:
74
                raise ValueError("Long header packet too short")
75

76
            packet_type = (first_byte & 0x30) >> 4
77
            type_names = {0: 'INITIAL', 1: '0RTT', 2: 'HANDSHAKE', 3: 'RETRY'}
78
            packet_info['type'] = type_names.get(packet_type, 'UNKNOWN')
79

80
            # Extract version
81
            version = struct.unpack('!I', data[1:5])[0]
82
            packet_info['version'] = version
83

84
            # Extract connection IDs
85
            dcid_len = data[5]
86
            if len(data) < 6 + dcid_len:
87
                raise ValueError("Destination connection ID too short")
88

89
            packet_info['dest_conn_id'] = data[6:6+dcid_len]
90

91
            scid_offset = 6 + dcid_len
92
            if len(data) <= scid_offset:
93
                raise ValueError("Source connection ID missing")
94

95
            scid_len = data[scid_offset]
96
            packet_info['source_conn_id'] = data[scid_offset+1:scid_offset+1+scid_len]
97

98
        else:  # Short header
99
            packet_info['type'] = 'SHORT'
100
            # Extract destination connection ID (variable length, needs connection context)
101
            # For simplicity, assume first 8 bytes after flags
102
            packet_info['dest_conn_id'] = data[1:9] if len(data) >= 9 else data[1:]
103

104
        packet_info['size'] = len(data)
105
        return packet_info
106

107
    def _is_blocked(self, ip: str) -> bool:
108
        """Check if IP address is currently blocked"""
109
        if ip in self.blocked_ips:
110
            if time.time() < self.blocked_ips[ip]:
111
                return True
112
            else:
113
                # Block expired
114
                del self.blocked_ips[ip]
115
        return False
116

117
    def _check_rate_limit(self, ip: str) -> bool:
118
        """Check and update rate limiting for IP address"""
119
        now = time.time()
120
        rate_info = self.rate_limits[ip]
121

122
        # Reset window if needed
123
        if now - rate_info['window_start'] >= 1.0:  # 1 second window
124
            rate_info['packets'] = 0
125
            rate_info['window_start'] = now
126

127
        rate_info['packets'] += 1
128

129
        return rate_info['packets'] <= self.policies['max_packets_per_second']
130

131
    def _check_amplification_protection(self, packet_info: Dict,
132
                                       source_addr: Tuple[str, int]) -> bool:
133
        """Check for amplification attack patterns"""
134
        # Small initial packets that could trigger large responses
135
        if (packet_info['type'] == 'INITIAL' and
136
            packet_info['size'] < 10):  # Suspiciously small
137
            return False
138

139
        # Version negotiation abuse
140
        if (packet_info['type'] == 'INITIAL' and
141
            packet_info.get('version') == 0):  # Version negotiation trigger
142
            return False
143

144
        return True
145

146
    def _validate_address(self, packet_info: Dict,
147
                         source_addr: Tuple[str, int]) -> bool:
148
        """Implement address validation for new connections"""
149
        # In a real implementation, this would use:
150
        # 1. Retry packets with address validation tokens
151
        # 2. Rate limiting of new connections per IP
152
        # 3. Geolocation checks
153
        # 4. Reputation-based filtering
154

155
        conn_id = packet_info.get('dest_conn_id', b'')
156

157
        # Simple rate limiting for new connections
158
        ip = source_addr[0]
159
        now = time.time()
160

161
        if ip not in self.connection_states:
162
            self.connection_states[ip] = {'new_connections': [], 'validated': set()}
163

164
        state = self.connection_states[ip]
165

166
        # Clean old entries
167
        cutoff = now - 60  # 1 minute window
168
        state['new_connections'] = [t for t in state['new_connections'] if t > cutoff]
169

170
        # Check rate limit
171
        if len(state['new_connections']) >= self.policies['max_new_connections_per_minute']:
172
            return False
173

174
        # Add to new connections
175
        state['new_connections'].append(now)
176

177
        return True
178

179
    def _update_connection_state(self, packet_info: Dict,
180
                                source_addr: Tuple[str, int]):
181
        """Update connection state tracking"""
182
        ip = source_addr[0]
183
        conn_id = packet_info.get('dest_conn_id', b'')
184

185
        if ip not in self.connection_states:
186
            self.connection_states[ip] = {'new_connections': [], 'validated': set()}
187

188
        # Mark connection as having activity
189
        if packet_info['type'] in ['HANDSHAKE', 'SHORT']:
190
            self.connection_states[ip]['validated'].add(conn_id)
191

192
    def _record_violation(self, ip: str, violation_type: str):
193
        """Record security violation"""
194
        print(f"🚨 QUIC Security Violation: {violation_type} from {ip}")
195

196
        # Implement violation scoring and automatic blocking
197
        # For now, just log the violation
198

199
    def _block_ip(self, ip: str, reason: str):
200
        """Block IP address temporarily"""
201
        block_until = time.time() + self.policies['block_duration']
202
        self.blocked_ips[ip] = block_until
203

204
        print(f"🚫 Blocked {ip} for {self.policies['block_duration']} seconds "
205
              f"(Reason: {reason})")
206

207
    def get_firewall_stats(self) -> Dict:
208
        """Get firewall statistics"""
209
        now = time.time()
210
        active_blocks = len([ip for ip, until in self.blocked_ips.items() if until > now])
211

212
        total_connections = sum(
213
            len(state['validated']) for state in self.connection_states.values()
214
        )
215

216
        return {
217
            'active_blocks': active_blocks,
218
            'tracked_ips': len(self.rate_limits),
219
            'total_connections': total_connections,
220
            'policies': self.policies
221
        }
222

223
# Example usage
224
firewall = QUICFirewall()
225

226
# Simulate processing QUIC packets
227
sample_packets = [
228
    # Normal INITIAL packet
229
    b'\x80\x00\x00\x00\x01\x08\x12\x34\x56\x78\x9a\xbc\xde\xf0\x04\x11\x22\x33\x44',
230
    # Suspicious small packet
231
    b'\x80\x00\x00\x00\x01\x00\x00',
232
    # Short header packet
233
    b'\x40\x12\x34\x56\x78\x9a\xbc\xde\xf0'
234
]
235

236
for packet in sample_packets:
237
    allowed = firewall.process_quic_packet(packet, ('192.168.1.100', 12345), ('server.com', 443))
238
    print(f"Packet allowed: {allowed}")
239

240
print("\nFirewall Stats:", firewall.get_firewall_stats())

1
# QUIC TLS 1.3 Security Validation
2
import hashlib
3
import hmac
4
import secrets
5
from cryptography.hazmat.primitives import hashes, serialization
6
from cryptography.hazmat.primitives.asymmetric import rsa, padding
7
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
8
from cryptography.hazmat.backends import default_backend
9

10
class QUICTLSValidator:
11
    """Validate QUIC TLS 1.3 implementation security"""
12

13
    def __init__(self):
14
        self.supported_cipher_suites = [
15
            'TLS_AES_128_GCM_SHA256',
16
            'TLS_AES_256_GCM_SHA384',
17
            'TLS_CHACHA20_POLY1305_SHA256',
18
            'TLS_AES_128_CCM_SHA256',
19
            'TLS_AES_128_CCM_8_SHA256'
20
        ]
21

22
        self.supported_groups = [
23
            'secp256r1', 'secp384r1', 'secp521r1',
24
            'x25519', 'x448',
25
            'ffdhe2048', 'ffdhe3072', 'ffdhe4096'
26
        ]
27

28
        self.min_tls_version = '1.3'
29

30
    def validate_handshake(self, client_hello: Dict, server_hello: Dict) -> Dict:
31
        """Validate QUIC TLS handshake security"""
32
        issues = []
33

34
        # 1. TLS version validation
35
        if not self._validate_tls_version(server_hello):
36
            issues.append("TLS version < 1.3 not allowed in QUIC")
37

38
        # 2. Cipher suite validation
39
        cipher_issues = self._validate_cipher_suite(server_hello)
40
        issues.extend(cipher_issues)
41

42
        # 3. Key exchange validation
43
        kx_issues = self._validate_key_exchange(client_hello, server_hello)
44
        issues.extend(kx_issues)
45

46
        # 4. Certificate validation
47
        cert_issues = self._validate_certificates(server_hello)
48
        issues.extend(cert_issues)
49

50
        # 5. 0-RTT security validation
51
        if client_hello.get('early_data'):
52
            zero_rtt_issues = self._validate_0rtt_security(client_hello)
53
            issues.extend(zero_rtt_issues)
54

55
        # 6. QUIC transport parameters
56
        transport_issues = self._validate_transport_parameters(server_hello)
57
        issues.extend(transport_issues)
58

59
        return {
60
            'valid': len(issues) == 0,
61
            'issues': issues,
62
            'security_level': self._calculate_security_level(issues),
63
            'recommendations': self._generate_recommendations(issues)
64
        }
65

66
    def _validate_tls_version(self, server_hello: Dict) -> bool:
67
        """Validate TLS version is 1.3"""
68
        version = server_hello.get('version', '1.2')
69
        return version == '1.3'
70

71
    def _validate_cipher_suite(self, server_hello: Dict) -> List[str]:
72
        """Validate selected cipher suite"""
73
        issues = []
74

75
        cipher_suite = server_hello.get('cipher_suite')
76
        if not cipher_suite:
77
            issues.append("No cipher suite selected")
78
            return issues
79

80
        if cipher_suite not in self.supported_cipher_suites:
81
            issues.append(f"Unsupported cipher suite: {cipher_suite}")
82

83
        # Check for weak ciphers (should not exist in TLS 1.3, but double-check)
84
        weak_patterns = ['RC4', 'DES', 'MD5', 'SHA1', 'NULL']
85
        for pattern in weak_patterns:
86
            if pattern in cipher_suite.upper():
87
                issues.append(f"Weak cipher detected: {pattern} in {cipher_suite}")
88

89
        return issues
90

91
    def _validate_key_exchange(self, client_hello: Dict, server_hello: Dict) -> List[str]:
92
        """Validate key exchange parameters"""
93
        issues = []
94

95
        # Validate supported groups
96
        client_groups = client_hello.get('supported_groups', [])
97
        selected_group = server_hello.get('key_exchange_group')
98

99
        if not selected_group:
100
            issues.append("No key exchange group selected")
101
            return issues
102

103
        if selected_group not in self.supported_groups:
104
            issues.append(f"Unsupported key exchange group: {selected_group}")
105

106
        if selected_group not in client_groups:
107
            issues.append(f"Server selected group {selected_group} not offered by client")
108

109
        # Validate key share
110
        key_share = server_hello.get('key_share')
111
        if not key_share:
112
            issues.append("Missing key share in server hello")
113
        elif len(key_share) < 32:  # Minimum for reasonable security
114
            issues.append(f"Key share too short: {len(key_share)} bytes")
115

116
        return issues
117

118
    def _validate_certificates(self, server_hello: Dict) -> List[str]:
119
        """Validate certificate chain"""
120
        issues = []
121

122
        cert_chain = server_hello.get('certificate_chain', [])
123
        if not cert_chain:
124
            issues.append("No certificate chain provided")
125
            return issues
126

127
        # Validate each certificate in chain
128
        for i, cert_data in enumerate(cert_chain):
129
            cert_issues = self._validate_single_certificate(cert_data, i == 0)
130
            issues.extend([f"Cert {i}: {issue}" for issue in cert_issues])
131

132
        return issues
133

134
    def _validate_single_certificate(self, cert_data: Dict, is_leaf: bool) -> List[str]:
135
        """Validate a single certificate"""
136
        issues = []
137

138
        # Key size validation
139
        key_size = cert_data.get('key_size', 0)
140
        key_type = cert_data.get('key_type', 'unknown')
141

142
        if key_type == 'RSA' and key_size < 2048:
143
            issues.append(f"RSA key too small: {key_size} bits")
144
        elif key_type == 'ECDSA' and key_size < 256:
145
            issues.append(f"ECDSA key too small: {key_size} bits")
146

147
        # Expiration validation
148
        not_after = cert_data.get('not_after')
149
        if not_after and not_after < time.time():
150
            issues.append("Certificate has expired")
151

152
        # Algorithm validation
153
        signature_alg = cert_data.get('signature_algorithm', '')
154
        weak_algs = ['md5', 'sha1']
155
        if any(weak_alg in signature_alg.lower() for weak_alg in weak_algs):
156
            issues.append(f"Weak signature algorithm: {signature_alg}")
157

158
        return issues
159

160
    def _validate_0rtt_security(self, client_hello: Dict) -> List[str]:
161
        """Validate 0-RTT security parameters"""
162
        issues = []
163

164
        # Check for replay protection
165
        if not client_hello.get('anti_replay_token'):
166
            issues.append("0-RTT enabled without anti-replay protection")
167

168
        # Validate max early data size
169
        max_early_data = client_hello.get('max_early_data_size', 0)
170
        if max_early_data > 16384:  # 16KB reasonable limit
171
            issues.append(f"Max early data size too large: {max_early_data} bytes")
172

173
        # Check for forward secrecy implications
174
        psk_key_exchange_modes = client_hello.get('psk_key_exchange_modes', [])
175
        if 'psk_ke' in psk_key_exchange_modes:
176
            issues.append("PSK-only mode reduces forward secrecy")
177

178
        return issues
179

180
    def _validate_transport_parameters(self, server_hello: Dict) -> List[str]:
181
        """Validate QUIC transport parameters"""
182
        issues = []
183

184
        transport_params = server_hello.get('quic_transport_parameters', {})
185

186
        # Validate initial connection window
187
        initial_window = transport_params.get('initial_max_stream_data_bidi_local', 0)
188
        if initial_window > 1048576:  # 1MB reasonable limit
189
            issues.append(f"Initial stream window too large: {initial_window}")
190

191
        # Validate idle timeout
192
        idle_timeout = transport_params.get('idle_timeout', 0)
193
        if idle_timeout > 600000:  # 10 minutes max
194
            issues.append(f"Idle timeout too long: {idle_timeout}ms")
195

196
        # Check for connection migration parameters
197
        if transport_params.get('disable_active_migration', False):
198
            # This is actually good for security, but note it
199
            pass
200

201
        return issues
202

203
    def _calculate_security_level(self, issues: List[str]) -> str:
204
        """Calculate overall security level"""
205
        critical_patterns = ['expired', 'weak', 'unsupported', 'missing']
206
        critical_count = sum(
207
            1 for issue in issues
208
            if any(pattern in issue.lower() for pattern in critical_patterns)
209
        )
210

211
        if critical_count > 0:
212
            return 'LOW'
213
        elif len(issues) > 5:
214
            return 'MEDIUM'
215
        elif len(issues) > 0:
216
            return 'HIGH'
217
        else:
218
            return 'EXCELLENT'
219

220
    def _generate_recommendations(self, issues: List[str]) -> List[str]:
221
        """Generate security recommendations"""
222
        recommendations = []
223

224
        if any('cipher' in issue.lower() for issue in issues):
225
            recommendations.append("Use only AEAD cipher suites (AES-GCM, ChaCha20-Poly1305)")
226

227
        if any('key' in issue.lower() for issue in issues):
228
            recommendations.append("Use RSA keys ≥2048 bits or ECDSA keys ≥256 bits")
229

230
        if any('0-rtt' in issue.lower() for issue in issues):
231
            recommendations.append("Implement proper 0-RTT replay protection")
232

233
        if any('certificate' in issue.lower() for issue in issues):
234
            recommendations.append("Maintain valid certificate chain with strong algorithms")
235

236
        return recommendations
237

238
# Example validation
239
validator = QUICTLSValidator()
240

241
# Simulate handshake data
242
client_hello = {
243
    'version': '1.3',
244
    'supported_groups': ['x25519', 'secp256r1'],
245
    'cipher_suites': ['TLS_AES_256_GCM_SHA384', 'TLS_CHACHA20_POLY1305_SHA256'],
246
    'early_data': True,
247
    'max_early_data_size': 8192,
248
    'anti_replay_token': 'present'
249
}
250

251
server_hello = {
252
    'version': '1.3',
253
    'cipher_suite': 'TLS_AES_256_GCM_SHA384',
254
    'key_exchange_group': 'x25519',
255
    'key_share': b'x' * 32,  # 32 bytes key share
256
    'certificate_chain': [{
257
        'key_type': 'ECDSA',
258
        'key_size': 256,
259
        'signature_algorithm': 'sha256WithECDSA',
260
        'not_after': time.time() + 86400 * 365,  # Valid for 1 year
261
    }],
262
    'quic_transport_parameters': {
263
        'initial_max_stream_data_bidi_local': 65536,
264
        'idle_timeout': 30000,  # 30 seconds
265
        'disable_active_migration': True
266
    }
267
}
268

269
result = validator.validate_handshake(client_hello, server_hello)
270
print(f"Handshake validation result: {result}")

1
# QUIC Connection Migration Security Monitor
2
import time
3
import hashlib
4
from collections import defaultdict
5
from dataclasses import dataclass
6
from typing import Dict, List, Set, Tuple
7

8
@dataclass
9
class MigrationEvent:
10
    connection_id: bytes
11
    old_address: Tuple[str, int]
12
    new_address: Tuple[str, int]
13
    timestamp: float
14
    validation_token: bytes
15
    authenticated: bool
16

17
class ConnectionMigrationMonitor:
18
    """Monitor and secure QUIC connection migrations"""
19

20
    def __init__(self):
21
        self.active_connections = {}  # conn_id -> connection info
22
        self.migration_history = []
23
        self.suspicious_patterns = []
24

25
        # Security policies
26
        self.max_migrations_per_hour = 10
27
        self.max_distance_km = 500  # Geographic distance
28
        self.require_validation = True
29
        self.block_tor_exits = True
30

31
    def handle_migration_event(self, event: MigrationEvent) -> bool:
32
        """Process a connection migration event"""
33
        # Validate the migration
34
        if not self._validate_migration(event):
35
            self._record_suspicious_migration(event, "validation_failed")
36
            return False
37

38
        # Check for abuse patterns
39
        if self._detect_migration_abuse(event):
40
            self._record_suspicious_migration(event, "abuse_pattern")
41
            return False
42

43
        # Check geographic constraints
44
        if not self._validate_geographic_migration(event):
45
            self._record_suspicious_migration(event, "geographic_violation")
46
            return False
47

48
        # Update connection state
49
        self._update_connection_state(event)
50

51
        # Record successful migration
52
        self.migration_history.append(event)
53

54
        print(f"✅ Connection migration allowed: "
55
              f"{event.old_address[0]} -> {event.new_address[0]}")
56

57
        return True
58

59
    def _validate_migration(self, event: MigrationEvent) -> bool:
60
        """Validate migration authentication"""
61
        if not self.require_validation:
62
            return True
63

64
        # Check if connection exists and is authenticated
65
        if event.connection_id not in self.active_connections:
66
            return False
67

68
        conn_info = self.active_connections[event.connection_id]
69

70
        # Validate migration token
71
        expected_token = self._generate_validation_token(
72
            event.connection_id,
73
            event.new_address,
74
            conn_info['migration_key']
75
        )
76

77
        return hmac.compare_digest(event.validation_token, expected_token)
78

79
    def _detect_migration_abuse(self, event: MigrationEvent) -> bool:
80
        """Detect migration abuse patterns"""
81
        # Count recent migrations for this connection
82
        recent_migrations = [
83
            m for m in self.migration_history
84
            if (m.connection_id == event.connection_id and
85
                time.time() - m.timestamp < 3600)  # Last hour
86
        ]
87

88
        if len(recent_migrations) >= self.max_migrations_per_hour:
89
            return True
90

91
        # Detect rapid back-and-forth migrations
92
        if len(recent_migrations) >= 2:
93
            last_migration = recent_migrations[-1]
94
            if (event.new_address == last_migration.old_address and
95
                time.time() - last_migration.timestamp < 60):  # Within 1 minute
96
                return True
97

98
        # Detect migration to known malicious networks
99
        if self._is_malicious_network(event.new_address[0]):
100
            return True
101

102
        return False
103

104
    def _validate_geographic_migration(self, event: MigrationEvent) -> bool:
105
        """Validate geographic feasibility of migration"""
106
        old_location = self._get_location(event.old_address[0])
107
        new_location = self._get_location(event.new_address[0])
108

109
        if not old_location or not new_location:
110
            # Can't determine locations, allow migration
111
            return True
112

113
        # Calculate distance
114
        distance = self._calculate_distance(old_location, new_location)
115

116
        # Check if distance is feasible given time since last migration
117
        conn_info = self.active_connections.get(event.connection_id, {})
118
        last_migration_time = conn_info.get('last_migration', 0)
119
        time_diff = event.timestamp - last_migration_time
120

121
        # Assume maximum speed of 1000 km/h (commercial aircraft)
122
        max_distance = (time_diff / 3600) * 1000
123

124
        return distance <= max_distance
125

126
    def _generate_validation_token(self, conn_id: bytes,
127
                                 new_address: Tuple[str, int],
128
                                 migration_key: bytes) -> bytes:
129
        """Generate validation token for migration"""
130
        data = conn_id + new_address[0].encode() + new_address[1].to_bytes(2, 'big')
131
        return hmac.new(migration_key, data, hashlib.sha256).digest()
132

133
    def _is_malicious_network(self, ip: str) -> bool:
134
        """Check if IP belongs to malicious network"""
135
        # In practice, this would check against:
136
        # - Tor exit nodes
137
        # - Known botnets
138
        # - Malware C&C networks
139
        # - Open proxies
140

141
        # Simplified check for Tor (example ranges)
142
        tor_ranges = [
143
            '103.251.167.0/24',
144
            '185.220.100.0/24',
145
            '199.87.154.0/24'
146
        ]
147

148
        for tor_range in tor_ranges:
149
            if self._ip_in_range(ip, tor_range):
150
                return True
151

152
        return False
153

154
    def _get_location(self, ip: str) -> Optional[Tuple[float, float]]:
155
        """Get geographic location for IP address"""
156
        # In practice, this would use MaxMind GeoIP or similar
157
        # For demo, return sample coordinates
158
        location_map = {
159
            '192.168.1.100': (37.7749, -122.4194),  # San Francisco
160
            '10.0.0.100': (40.7128, -74.0060),      # New York
161
            '172.16.0.100': (51.5074, -0.1278),     # London
162
        }
163

164
        return location_map.get(ip)
165

166
    def _calculate_distance(self, loc1: Tuple[float, float],
167
                          loc2: Tuple[float, float]) -> float:
168
        """Calculate distance between two geographic points"""
169
        import math
170

171
        lat1, lon1 = math.radians(loc1[0]), math.radians(loc1[1])
172
        lat2, lon2 = math.radians(loc2[0]), math.radians(loc2[1])
173

174
        dlat = lat2 - lat1
175
        dlon = lon2 - lon1
176

177
        # Haversine formula
178
        a = (math.sin(dlat/2)**2 +
179
             math.cos(lat1) * math.cos(lat2) * math.sin(dlon/2)**2)
180
        c = 2 * math.asin(math.sqrt(a))
181

182
        # Earth's radius in kilometers
183
        return 6371 * c
184

185
    def _ip_in_range(self, ip: str, cidr: str) -> bool:
186
        """Check if IP is in CIDR range"""
187
        import ipaddress
188
        try:
189
            return ipaddress.ip_address(ip) in ipaddress.ip_network(cidr)
190
        except:
191
            return False
192

193
    def _update_connection_state(self, event: MigrationEvent):
194
        """Update connection state after successful migration"""
195
        conn_id = event.connection_id
196

197
        if conn_id not in self.active_connections:
198
            self.active_connections[conn_id] = {
199
                'migration_key': secrets.token_bytes(32),
200
                'created_at': time.time()
201
            }
202

203
        conn_info = self.active_connections[conn_id]
204
        conn_info['current_address'] = event.new_address
205
        conn_info['last_migration'] = event.timestamp
206
        conn_info['migration_count'] = conn_info.get('migration_count', 0) + 1
207

208
    def _record_suspicious_migration(self, event: MigrationEvent, reason: str):
209
        """Record suspicious migration attempt"""
210
        suspicious_event = {
211
            'event': event,
212
            'reason': reason,
213
            'timestamp': time.time()
214
        }
215

216
        self.suspicious_patterns.append(suspicious_event)
217

218
        print(f"🚨 Suspicious migration blocked: {reason} "
219
              f"({event.old_address[0]} -> {event.new_address[0]})")
220

221
    def get_security_report(self) -> Dict:
222
        """Generate security report for connection migrations"""
223
        total_migrations = len(self.migration_history)
224
        suspicious_count = len(self.suspicious_patterns)
225

226
        # Analyze patterns
227
        by_reason = defaultdict(int)
228
        for pattern in self.suspicious_patterns:
229
            by_reason[pattern['reason']] += 1
230

231
        # Top migrating connections
232
        migration_counts = defaultdict(int)
233
        for migration in self.migration_history:
234
            migration_counts[migration.connection_id] += 1
235

236
        top_migrators = sorted(migration_counts.items(),
237
                              key=lambda x: x[1], reverse=True)[:5]
238

239
        return {
240
            'total_migrations': total_migrations,
241
            'suspicious_attempts': suspicious_count,
242
            'success_rate': ((total_migrations / (total_migrations + suspicious_count))
243
                           if total_migrations + suspicious_count > 0 else 0),
244
            'suspicious_by_reason': dict(by_reason),
245
            'top_migrating_connections': [
246
                {'connection_id': conn_id.hex(), 'count': count}
247
                for conn_id, count in top_migrators
248
            ],
249
            'active_connections': len(self.active_connections)
250
        }
251

252
# Example usage
253
monitor = ConnectionMigrationMonitor()
254

255
# Simulate migration events
256
events = [
257
    MigrationEvent(
258
        connection_id=b'\x12\x34\x56\x78',
259
        old_address=('192.168.1.100', 12345),
260
        new_address=('10.0.0.100', 12345),
261
        timestamp=time.time(),
262
        validation_token=b'valid_token',
263
        authenticated=True
264
    ),
265
    MigrationEvent(
266
        connection_id=b'\x12\x34\x56\x78',
267
        old_address=('10.0.0.100', 12345),
268
        new_address=('172.16.0.100', 12345),
269
        timestamp=time.time() + 30,  # 30 seconds later
270
        validation_token=b'invalid_token',
271
        authenticated=False
272
    )
273
]
274

275
for event in events:
276
    monitor.handle_migration_event(event)
277

278
print("\nSecurity Report:", monitor.get_security_report())