Computer Networks Deep Dive — TCP Congestion Control, DNS, TLS & QUIC

import time

class TCPReno:
    def __init__(self):
        self.cwnd = 1      # segments
        self.ssthresh = 64  # slow start threshold
        self.state = 'slow_start'

    def on_ack(self):
        if self.state == 'slow_start':
            self.cwnd *= 2
            if self.cwnd >= self.ssthresh:
                self.state = 'congestion_avoidance'
        elif self.state == 'congestion_avoidance':
            self.cwnd += 1 / self.cwnd  # Additive increase

    def on_loss(self):
        self.ssthresh = self.cwnd // 2
        self.cwnd = self.ssthresh
        self.state = 'congestion_avoidance'

    def simulate(self, rtts=20):
        for rtt in range(1, rtts + 1):
            time.sleep(0.1)
            if rtt % 5 == 0 and rtt > 2:
                self.on_loss()
                cause = 'loss'
            else:
                self.on_ack()
                cause = 'ack'
            print(f'RTT {rtt:2d}: cwnd={self.cwnd:5.1f} '
                  f'state={self.state:22s} ({cause})')

tcp = TCPReno()
tcp.simulate(15)

import time

def dns_resolve(domain, recursive=True):
    steps = []
    steps.append(f'Client asks: resolve {domain}')

    if recursive:
        steps.append('Recursive resolver queries root server')
        steps.append('Root server → .com TLD server')
        steps.append('TLD server → authoritative nameserver')
        steps.append(f'Authoritative: {domain} → 93.184.216.34')
        return '93.184.216.34', steps

    steps.append('Client asks root server: where is .com?')
    steps.append('Root → here are the .com TLD servers')
    steps.append('Client asks .com TLD: where is example.com?')
    steps.append('.com → authoritative: ns1.example.com')
    steps.append('Client asks ns1.example.com: what is example.com?')
    steps.append('ns1 → A record: 93.184.216.34')
    return '93.184.216.34', steps

for mode in [True, False]:
    label = 'recursive' if mode else 'iterative'
    ip, steps = dns_resolve('example.com', recursive=mode)
    print(f'\n=== {label.upper()} resolution ===')
    for s in steps:
        print(f'  {s}')

import hashlib

def tls_handshake(client_random, server_random, psk=None):
    print('=== TLS 1.3 Handshake ===')

    # Client Hello
    client_key = hashlib.sha256(b'client_key_share').hexdigest()[:16]
    print(f'[1] Client Hello')
    print(f'    Random: {client_random}')
    print(f'    Key share: {client_key}')
    print(f'    Supported: TLS 1.3, TLS 1.2')
    if psk:
        print(f'    PSK: {psk}')

    # Server Hello
    server_key = hashlib.sha256(b'server_key_share').hexdigest()[:16]
    print(f'[2] Server Hello')
    print(f'    Random: {server_random}')
    print(f'    Key share: {server_key}')
    print(f'    Cipher suite: TLS_AES_256_GCM_SHA384')

    # Derive shared secret (simplified ECDHE)
    shared_secret = hashlib.sha256(
        (client_random + server_random).encode()
    ).hexdigest()[:16]
    print(f'[3] Shared Secret (ECDHE): {shared_secret}')

    # Server sends encrypted extensions + finished
    print(f'[4] Server: EncryptedExtensions')
    print(f'[5] Server: Certificate + CertificateVerify')
    print(f'[6] Server: Finished')

    # Client sends finished
    print(f'[7] Client: Finished')
    print(f'\nSecure connection established!')
    print(f'Application data encrypted with AES-256-GCM')
    return shared_secret

secret = tls_handshake('abc123', 'xyz789')
print(f'Encryption key: {secret}')

class HTTP2Connection:
    def __init__(self):
        self.streams = {}
        self.tcp_buffer = []

    def send_stream(self, stream_id, data):
        # Over TCP — one lost packet blocks all streams
        print(f'[HTTP/2] Sending stream {stream_id} '
              f'({len(data)} bytes)')
        self.tcp_buffer.append((stream_id, data))

    def simulate_loss(self, stream_id):
        # Head-of-line blocking: all subsequent streams wait
        blocked = [sid for sid, _ in self.tcp_buffer
                   if sid >= stream_id]
        print(f'[HTTP/2] Packet loss! Streams {blocked} '
              f'blocked until retransmission')

class HTTP3Connection:
    def __init__(self):
        self.streams = {}

    def send_stream(self, stream_id, data):
        # QUIC — each stream is independent
        print(f'[HTTP/3] Sending stream {stream_id} '
              f'({len(data)} bytes)')

    def simulate_loss(self, stream_id):
        print(f'[HTTP/3] Stream {stream_id} lost — '
              f'retransmitting only stream {stream_id}')

h2 = HTTP2Connection()
h3 = HTTP3Connection()

print('=== HTTP/2 (TCP) ===')
h2.send_stream(1, 'index.html')
h2.send_stream(3, 'style.css')
h2.send_stream(5, 'app.js')
h2.simulate_loss(3)

print('\n=== HTTP/3 (QUIC) ===')
h3.send_stream(1, 'index.html')
h3.send_stream(3, 'style.css')
h3.send_stream(5, 'app.js')
h3.simulate_loss(3)

def cidr_info(cidr):
    ip_str, prefix = cidr.split('/')
    prefix = int(prefix)

    # Convert IP to integer
    octets = [int(x) for x in ip_str.split('.')]
    ip_int = sum(o << (24 - 8*i) for i, o in enumerate(octets))

    # Calculate subnet mask
    mask = (0xFFFFFFFF << (32 - prefix)) & 0xFFFFFFFF
    network = ip_int & mask
    broadcast = network | ~mask & 0xFFFFFFFF

    # Number of hosts
    hosts = (1 << (32 - prefix)) - 2

    # Convert back to dotted decimal
    def to_ip(n):
        return '.'.join(str((n >> (24 - 8*i)) & 0xFF) for i in range(4))

    return {
        'network': to_ip(network),
        'broadcast': to_ip(broadcast),
        'mask': to_ip(mask),
        'hosts': hosts,
        'first_host': to_ip(network + 1),
        'last_host': to_ip(broadcast - 1),
    }

info = cidr_info('192.168.1.42/24')
for k, v in info.items():
    print(f'{k:12s}: {v}')

def ice_connect(local, remote, stun_server='stun.example.com'):
    print(f'ICE connecting {local} ↔ {remote}')
    print(f'Using STUN server: {stun_server}')

    # Step 1: Gather candidates
    local_candidates = [
        {'type': 'host', 'ip': local['private'], 'port': local['port']},
        {'type': 'srflx', 'ip': local['public'], 'port': 3478},
    ]
    print(f'Local candidates: {[c["type"] for c in local_candidates]}')

    # Step 2: STUN binding to learn public IP
    public_ip = '203.0.113.42'
    print(f'STUN: your public IP is {public_ip}')

    # Step 3: Connectivity check (NAT traversal)
    print(f'ICE connectivity check: sending from '
          f'{local_candidates[1]["ip"]} to {remote["public"]}')

    # Step 4: If direct fails, use TURN relay
    direct_works = True
    if not direct_works:
        print('Direct connection failed!')
        print('TURN relay: 203.0.113.99:3478')
        print('All traffic relayed through TURN server')

    print('ICE: connection established!')
    return direct_works

local = {'private': '192.168.1.5', 'public': '203.0.113.42', 'port': 54321}
remote = {'public': '198.51.100.7', 'port': 12345}
ice_connect(local, remote)