1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101

import h2.connection
import h2.events
import socket
import ssl

class HTTP2Client:
    def __init__(self, host, port=443):
        self.host = host
        self.port = port
        self.connection = None
        self.socket = None
    
    def connect(self):
        """Connect to HTTP/2 server"""
        # Create socket
        self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        
        # Configure TLS
        context = ssl.create_default_context()
        context.set_alpn_protocols(['h2'])
        
        # Connect
        self.socket = context.wrap_socket(self.socket, server_hostname=self.host)
        self.socket.connect((self.host, self.port))
        
        # Create HTTP/2 connection
        self.connection = h2.connection.H2Connection()
        self.connection.initiate_connection()
        
        # Send initial configuration
        self.socket.send(self.connection.data_to_send())
    
    def send_request(self, method, path, headers=None):
        """Send HTTP/2 request"""
        if headers is None:
            headers = []
        
        # Basic headers
        request_headers = [
            (':method', method),
            (':path', path),
            (':scheme', 'https'),
            (':authority', self.host)
        ]
        
        # Add additional headers
        request_headers.extend(headers)
        
        # Send headers
        stream_id = self.connection.get_next_available_stream_id()
        self.connection.send_headers(stream_id, request_headers)
        
        # Send data
        self.socket.send(self.connection.data_to_send())
        
        return stream_id
    
    def read_response(self, stream_id):
        """Read HTTP/2 response"""
        response_data = b''
        
        while True:
            data = self.socket.recv(65536)
            if not data:
                break
            
            events = self.connection.receive_data(data)
            for event in events:
                if isinstance(event, h2.events.DataReceived):
                    if event.stream_id == stream_id:
                        response_data += event.data
                        self.connection.acknowledge_received_data(
                            event.flow_controlled_length, 
                            event.stream_id
                        )
                elif isinstance(event, h2.events.StreamEnded):
                    if event.stream_id == stream_id:
                        break
            
            # Send pending data
            data_to_send = self.connection.data_to_send()
            if data_to_send:
                self.socket.send(data_to_send)
        
        return response_data
    
    def close(self):
        """Close connection"""
        if self.socket:
            self.socket.close()

# Usage example
client = HTTP2Client('httpbin.org')
client.connect()

# Send request
stream_id = client.send_request('GET', '/get')
response = client.read_response(stream_id)
print(f"Response: {response.decode()}")

client.close()

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108

import h2.connection
import h2.events
import socket
import ssl
import threading

class HTTP2Server:
    def __init__(self, host='localhost', port=8443):
        self.host = host
        self.port = port
        self.socket = None
        self.running = False
    
    def start(self):
        """Start HTTP/2 server"""
        # Create socket
        self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        self.socket.bind((self.host, self.port))
        self.socket.listen(5)
        
        # Configure TLS
        context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
        context.load_cert_chain('server.crt', 'server.key')
        context.set_alpn_protocols(['h2'])
        
        self.running = True
        print(f"HTTP/2 server started on {self.host}:{self.port}")
        
        while self.running:
            try:
                client_socket, addr = self.socket.accept()
                tls_socket = context.wrap_socket(client_socket, server_side=True)
                
                # Handle client in separate thread
                thread = threading.Thread(
                    target=self.handle_client, 
                    args=(tls_socket, addr)
                )
                thread.start()
                
            except KeyboardInterrupt:
                self.running = False
                break
    
    def handle_client(self, client_socket, addr):
        """Handle HTTP/2 client"""
        connection = h2.connection.H2Connection()
        connection.initiate_connection()
        
        try:
            while True:
                data = client_socket.recv(65536)
                if not data:
                    break
                
                events = connection.receive_data(data)
                for event in events:
                    if isinstance(event, h2.events.RequestReceived):
                        self.handle_request(connection, event)
                    elif isinstance(event, h2.events.DataReceived):
                        self.handle_data(connection, event)
                
                # Send pending data
                data_to_send = connection.data_to_send()
                if data_to_send:
                    client_socket.send(data_to_send)
        
        except Exception as e:
            print(f"Error handling client: {e}")
        finally:
            client_socket.close()
    
    def handle_request(self, connection, event):
        """Handle HTTP/2 request"""
        headers = dict(event.headers)
        method = headers.get(':method')
        path = headers.get(':path')
        
        # Generate response
        response_headers = [
            (':status', '200'),
            ('content-type', 'text/html'),
            ('content-length', '0')
        ]
        
        # Send headers
        connection.send_headers(event.stream_id, response_headers)
        
        # Send data
        response_body = f"<h1>HTTP/2 Response</h1><p>Method: {method}</p><p>Path: {path}</p>"
        connection.send_data(event.stream_id, response_body.encode())
        connection.end_stream(event.stream_id)
    
    def handle_data(self, connection, event):
        """Handle HTTP/2 data"""
        # Process received data
        pass
    
    def stop(self):
        """Stop server"""
        self.running = False
        if self.socket:
            self.socket.close()

# Usage example
server = HTTP2Server()
# server.start()  # Run in separate thread

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78

import aioquic
import asyncio
import ssl
from aioquic.h3.connection import H3_ALPN
from aioquic.quic.configuration import QuicConfiguration

class HTTP3Client:
    def __init__(self, host, port=443):
        self.host = host
        self.port = port
        self.configuration = QuicConfiguration(
            alpn_protocols=H3_ALPN,
            is_client=True
        )
    
    async def connect(self):
        """Connect to HTTP/3 server"""
        # Configure SSL
        ssl_context = ssl.create_default_context()
        ssl_context.check_hostname = False
        ssl_context.verify_mode = ssl.CERT_NONE
        
        # Connect
        async with aioquic.connect(
            self.host,
            self.port,
            configuration=self.configuration,
            ssl_context=ssl_context
        ) as protocol:
            return protocol
    
    async def send_request(self, protocol, method, path, headers=None):
        """Send HTTP/3 request"""
        if headers is None:
            headers = {}
        
        # Basic headers
        request_headers = {
            ':method': method,
            ':path': path,
            ':scheme': 'https',
            ':authority': self.host
        }
        request_headers.update(headers)
        
        # Send request
        stream_id = protocol.get_next_available_stream_id()
        protocol.send_headers(stream_id, request_headers)
        protocol.send_data(stream_id, b'', end_stream=True)
        
        return stream_id
    
    async def read_response(self, protocol, stream_id):
        """Read HTTP/3 response"""
        response_data = b''
        
        while True:
            event = await protocol.wait_for_event()
            
            if event.stream_id == stream_id:
                if hasattr(event, 'data'):
                    response_data += event.data
                if hasattr(event, 'stream_ended') and event.stream_ended:
                    break
        
        return response_data

# Usage example
async def main():
    client = HTTP3Client('httpbin.org')
    protocol = await client.connect()
    
    # Send request
    stream_id = await client.send_request(protocol, 'GET', '/get')
    response = await client.read_response(protocol, stream_id)
    print(f"HTTP/3 Response: {response.decode()}")

# asyncio.run(main())

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79

import asyncio
import aiohttp

class HTTP2Multiplexing:
    def __init__(self, base_url):
        self.base_url = base_url
        self.session = None
    
    async def create_session(self):
        """Create HTTP/2 session"""
        connector = aiohttp.TCPConnector(
            limit=100,
            limit_per_host=30,
            ttl_dns_cache=300,
            use_dns_cache=True
        )
        
        self.session = aiohttp.ClientSession(
            connector=connector,
            timeout=aiohttp.ClientTimeout(total=30)
        )
    
    async def send_multiple_requests(self, requests):
        """Send multiple simultaneous requests"""
        if not self.session:
            await self.create_session()
        
        # Create tasks for simultaneous requests
        tasks = []
        for method, path, headers in requests:
            task = self.send_request(method, path, headers)
            tasks.append(task)
        
        # Execute simultaneous requests
        responses = await asyncio.gather(*tasks)
        return responses
    
    async def send_request(self, method, path, headers=None):
        """Send individual request"""
        url = f"{self.base_url}{path}"
        
        async with self.session.request(
            method, 
            url, 
            headers=headers
        ) as response:
            data = await response.text()
            return {
                'status': response.status,
                'headers': dict(response.headers),
                'data': data
            }
    
    async def close(self):
        """Close session"""
        if self.session:
            await self.session.close()

# Usage example
async def test_multiplexing():
    client = HTTP2Multiplexing('https://httpbin.org')
    
    # Multiple simultaneous requests
    requests = [
        ('GET', '/get', {}),
        ('GET', '/headers', {}),
        ('GET', '/user-agent', {}),
        ('GET', '/ip', {}),
        ('GET', '/json', {})
    ]
    
    responses = await client.send_multiple_requests(requests)
    
    for i, response in enumerate(responses):
        print(f"Request {i+1}: {response['status']}")
    
    await client.close()

# asyncio.run(test_multiplexing())

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51

class HTTP2ServerPush:
    def __init__(self, connection, stream_id):
        self.connection = connection
        self.stream_id = stream_id
    
    def push_resource(self, path, content, content_type='text/html'):
        """Send resource via server push"""
        # Headers for push
        push_headers = [
            (':method', 'GET'),
            (':path', path),
            (':scheme', 'https'),
            (':authority', 'example.com'),
            ('content-type', content_type),
            ('content-length', str(len(content)))
        ]
        
        # Create new stream for push
        push_stream_id = self.connection.get_next_available_stream_id()
        
        # Send headers
        self.connection.send_headers(push_stream_id, push_headers)
        
        # Send content
        self.connection.send_data(push_stream_id, content.encode())
        self.connection.end_stream(push_stream_id)
        
        return push_stream_id
    
    def push_css(self, css_content):
        """Push CSS file"""
        return self.push_resource('/style.css', css_content, 'text/css')
    
    def push_js(self, js_content):
        """Push JavaScript file"""
        return self.push_resource('/script.js', js_content, 'application/javascript')
    
    def push_image(self, image_path, image_data):
        """Push image"""
        content_type = 'image/jpeg'
        if image_path.endswith('.png'):
            content_type = 'image/png'
        elif image_path.endswith('.gif'):
            content_type = 'image/gif'
        
        return self.push_resource(image_path, image_data, content_type)

# Usage example
# push_handler = HTTP2ServerPush(connection, stream_id)
# push_handler.push_css("body { color: red; }")
# push_handler.push_js("console.log('Hello World');")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55

import ssl
import h2.connection

class SecureHTTP2:
    def __init__(self):
        self.ssl_context = None
        self.connection = None
    
    def create_secure_context(self):
        """Create secure SSL context"""
        context = ssl.create_default_context(ssl.Purpose.SERVER_AUTH)
        
        # Security configurations
        context.set_ciphers('ECDHE+AESGCM:ECDHE+CHACHA20:DHE+AESGCM:DHE+CHACHA20:!aNULL:!MD5:!DSS')
        context.options |= ssl.OP_NO_SSLv2
        context.options |= ssl.OP_NO_SSLv3
        context.options |= ssl.OP_NO_TLSv1
        context.options |= ssl.OP_NO_TLSv1_1
        context.options |= ssl.OP_NO_COMPRESSION
        context.options |= ssl.OP_SINGLE_DH_USE
        context.options |= ssl.OP_SINGLE_ECDH_USE
        
        # Certificate verification
        context.check_hostname = True
        context.verify_mode = ssl.CERT_REQUIRED
        
        return context
    
    def create_secure_connection(self, host, port=443):
        """Create secure connection"""
        # Create SSL context
        self.ssl_context = self.create_secure_context()
        
        # Create socket
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        
        # Configure ALPN for HTTP/2
        self.ssl_context.set_alpn_protocols(['h2'])
        
        # Connect with TLS
        secure_sock = self.ssl_context.wrap_socket(
            sock, 
            server_hostname=host
        )
        secure_sock.connect((host, port))
        
        # Create HTTP/2 connection
        self.connection = h2.connection.H2Connection()
        self.connection.initiate_connection()
        
        return secure_sock

# Usage example
secure_http2 = SecureHTTP2()
secure_sock = secure_http2.create_secure_connection('httpbin.org')

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

class SecurityHeaders:
    def __init__(self):
        self.security_headers = {
            'Strict-Transport-Security': 'max-age=31536000; includeSubDomains',
            'X-Content-Type-Options': 'nosniff',
            'X-Frame-Options': 'DENY',
            'X-XSS-Protection': '1; mode=block',
            'Referrer-Policy': 'strict-origin-when-cross-origin',
            'Content-Security-Policy': "default-src 'self'",
            'Permissions-Policy': 'geolocation=(), microphone=(), camera=()'
        }
    
    def add_security_headers(self, response_headers):
        """Add security headers"""
        for header, value in self.security_headers.items():
            response_headers.append((header, value))
        
        return response_headers
    
    def validate_headers(self, headers):
        """Validate security headers"""
        required_headers = [
            'Strict-Transport-Security',
            'X-Content-Type-Options',
            'X-Frame-Options'
        ]
        
        missing_headers = []
        for header in required_headers:
            if not any(h[0] == header for h in headers):
                missing_headers.append(header)
        
        return missing_headers

# Usage example
security = SecurityHeaders()
headers = [(':status', '200'), ('content-type', 'text/html')]
secure_headers = security.add_security_headers(headers)
print(f"Secure headers: {secure_headers}")