📘 Network Protocols: Understanding Layers

🎯 Introduction

Welcome to this exciting tutorial on network protocols and layers! 🎉 In this guide, we’ll explore how networks communicate using the OSI and TCP/IP models, and how to work with different protocol layers in Python.

You’ll discover how understanding network layers can transform your Python networking skills. Whether you’re building web applications 🌐, creating network tools 🛠️, or debugging connectivity issues 🐛, understanding protocol layers is essential for writing robust network code.

By the end of this tutorial, you’ll feel confident working with different network layers in your Python projects! Let’s dive in! 🏊‍♂️

📚 Understanding Network Protocols

🤔 What are Network Protocols?

Network protocols are like languages that computers use to talk to each other 🗣️. Think of them as a set of rules and formats that ensure devices can communicate reliably across networks.

In Python terms, protocols define how data is:

• ✨ Formatted and structured for transmission
• 🚀 Sent across different network layers
• 🛡️ Verified and error-checked
• 📦 Packaged and unpacked at each layer

💡 The OSI Model vs TCP/IP Model

Here’s why developers need to understand both models:

1. OSI Model (7 Layers) 🏗️: Theoretical framework for understanding
2. TCP/IP Model (4 Layers) 🌐: Practical implementation used on the Internet
3. Clear Abstraction 📖: Each layer has specific responsibilities
4. Debugging Power 🔧: Isolate issues to specific layers

Real-world example: Imagine sending a package 📦. The OSI model is like having different departments handle packaging, addressing, routing, and delivery - each with its own protocols!

🔧 Basic Syntax and Usage

📝 Working with Different Layers

Let’s start with exploring network layers in Python:

# 👋 Hello, Network Layers!
import socket
import struct
import binascii

# 🎨 Layer 7: Application Layer (HTTP)
def application_layer_example():
    # 🌐 Simple HTTP request
    http_request = (
        "GET / HTTP/1.1\r\n"
        "Host: example.com\r\n"
        "User-Agent: Python Network Explorer 🐍\r\n"
        "\r\n"
    )
    print(f"📱 Application Layer (HTTP):\n{http_request}")
    return http_request.encode()

# 🔄 Layer 4: Transport Layer (TCP)
def transport_layer_example():
    # 🚀 Create TCP socket
    tcp_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    
    # 📊 Get socket info
    print(f"🔌 Transport Layer (TCP):")
    print(f"  Protocol: {tcp_socket.proto}")
    print(f"  Type: {tcp_socket.type}")
    print(f"  Family: {tcp_socket.family}")
    
    return tcp_socket

# 🌐 Layer 3: Network Layer (IP)
def network_layer_example():
    # 🎯 Parse IP header
    def parse_ip_header(data):
        # 📦 First 20 bytes are IP header
        ip_header = struct.unpack('!BBHHHBBH4s4s', data[:20])
        
        version = ip_header[0] >> 4
        header_length = (ip_header[0] & 0xF) * 4
        ttl = ip_header[5]
        protocol = ip_header[6]
        src_ip = socket.inet_ntoa(ip_header[8])
        dst_ip = socket.inet_ntoa(ip_header[9])
        
        return {
            'version': version,
            'header_length': header_length,
            'ttl': ttl,
            'protocol': protocol,
            'source_ip': src_ip,
            'destination_ip': dst_ip
        }
    
    print("🌍 Network Layer (IP) Parser Ready!")
    return parse_ip_header

💡 Explanation: Notice how each layer has its own responsibilities! The application layer handles HTTP, transport handles TCP connections, and network handles IP addressing.

🎯 Layer Interaction Patterns

Here are patterns for working with protocol layers:

# 🏗️ Pattern 1: Socket Programming at Multiple Layers
class NetworkLayerExplorer:
    def __init__(self):
        self.layers = {
            7: "Application 📱",
            6: "Presentation 🎨",
            5: "Session 🔄",
            4: "Transport 🚀",
            3: "Network 🌐",
            2: "Data Link 🔗",
            1: "Physical ⚡"
        }
    
    # 🎮 Create raw socket for lower layer access
    def create_raw_socket(self):
        try:
            # 🛡️ Requires admin privileges
            raw_socket = socket.socket(
                socket.AF_INET, 
                socket.SOCK_RAW, 
                socket.IPPROTO_TCP
            )
            print("✅ Raw socket created for layer 3-4 access!")
            return raw_socket
        except PermissionError:
            print("⚠️ Need admin privileges for raw sockets!")
            return None
    
    # 📊 Analyze packet at different layers
    def analyze_packet(self, data):
        analysis = {}
        
        # 🌐 Layer 3: IP Header
        if len(data) >= 20:
            ip_version = (data[0] >> 4) & 0xF
            ip_header_length = (data[0] & 0xF) * 4
            
            analysis['layer_3'] = {
                'name': 'IP',
                'version': ip_version,
                'header_length': ip_header_length
            }
        
        # 🚀 Layer 4: TCP/UDP Header
        if len(data) >= 40:  # IP + TCP headers
            protocol = data[9]
            if protocol == 6:  # TCP
                tcp_start = ip_header_length
                src_port = struct.unpack('!H', data[tcp_start:tcp_start+2])[0]
                dst_port = struct.unpack('!H', data[tcp_start+2:tcp_start+4])[0]
                
                analysis['layer_4'] = {
                    'name': 'TCP',
                    'source_port': src_port,
                    'destination_port': dst_port
                }
        
        return analysis

# 🎨 Pattern 2: Protocol Encapsulation
class ProtocolStack:
    def __init__(self):
        self.layers = []
    
    # 📦 Encapsulate data through layers
    def encapsulate(self, data, layer_name):
        header = f"[{layer_name} Header]"
        encapsulated = f"{header}{data}"
        self.layers.append({
            'layer': layer_name,
            'data': encapsulated
        })
        print(f"📦 {layer_name} encapsulated: {encapsulated[:50]}...")
        return encapsulated
    
    # 🎁 Decapsulate through layers
    def decapsulate(self, data):
        for layer in reversed(self.layers):
            header = f"[{layer['layer']} Header]"
            if data.startswith(header):
                data = data[len(header):]
                print(f"🎁 {layer['layer']} decapsulated")
        return data

💡 Practical Examples

🛒 Example 1: HTTP Protocol Analyzer

Let’s build a practical HTTP protocol analyzer:

# 🌐 HTTP Protocol Analyzer
import socket
import re
from datetime import datetime

class HTTPAnalyzer:
    def __init__(self):
        self.methods = ['GET', 'POST', 'PUT', 'DELETE', 'HEAD', 'OPTIONS', 'PATCH']
        self.versions = ['HTTP/1.0', 'HTTP/1.1', 'HTTP/2']
        self.emoji_status = {
            200: '✅',
            404: '❌',
            500: '💥',
            301: '↪️',
            403: '🔒'
        }
    
    # 📱 Parse HTTP request
    def parse_request(self, request_data):
        lines = request_data.split('\r\n')
        if not lines:
            return None
        
        # 🎯 Parse request line
        request_line = lines[0].split(' ')
        if len(request_line) != 3:
            return None
        
        method, path, version = request_line
        
        # 📊 Parse headers
        headers = {}
        for i in range(1, len(lines)):
            if lines[i] == '':
                break
            if ':' in lines[i]:
                key, value = lines[i].split(':', 1)
                headers[key.strip()] = value.strip()
        
        # 🎨 Create analysis
        analysis = {
            'method': method,
            'path': path,
            'version': version,
            'headers': headers,
            'timestamp': datetime.now().isoformat(),
            'emoji': '🌐'
        }
        
        return analysis
    
    # 🚀 Analyze HTTP response
    def analyze_response(self, response_data):
        lines = response_data.split('\r\n')
        if not lines:
            return None
        
        # 📈 Parse status line
        status_line = lines[0].split(' ', 2)
        if len(status_line) < 2:
            return None
        
        version = status_line[0]
        status_code = int(status_line[1])
        status_text = status_line[2] if len(status_line) > 2 else ''
        
        # 🎯 Get emoji for status
        emoji = self.emoji_status.get(status_code, '📄')
        
        return {
            'version': version,
            'status_code': status_code,
            'status_text': status_text,
            'emoji': emoji,
            'meaning': self._get_status_meaning(status_code)
        }
    
    # 💡 Get status meaning
    def _get_status_meaning(self, code):
        meanings = {
            200: "Success! 🎉 Everything worked perfectly",
            404: "Not Found 🔍 The resource doesn't exist",
            500: "Server Error 💥 Something went wrong on the server",
            301: "Moved Permanently ↪️ Resource has a new location",
            403: "Forbidden 🔒 You don't have permission"
        }
        return meanings.get(code, f"Status code {code}")
    
    # 🛠️ Simple HTTP client
    def make_request(self, host, port=80, path='/', method='GET'):
        try:
            # 🔌 Create socket
            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            sock.settimeout(5)
            
            # 🌐 Connect
            print(f"🔗 Connecting to {host}:{port}...")
            sock.connect((host, port))
            
            # 📤 Send request
            request = f"{method} {path} HTTP/1.1\r\n"
            request += f"Host: {host}\r\n"
            request += "User-Agent: Python HTTP Analyzer 🐍\r\n"
            request += "Connection: close\r\n"
            request += "\r\n"
            
            sock.send(request.encode())
            print(f"📤 Request sent!")
            
            # 📥 Receive response
            response = b''
            while True:
                data = sock.recv(1024)
                if not data:
                    break
                response += data
            
            sock.close()
            
            # 🎨 Analyze response
            response_text = response.decode('utf-8', errors='ignore')
            analysis = self.analyze_response(response_text)
            
            if analysis:
                print(f"📥 Response: {analysis['emoji']} {analysis['status_code']} - {analysis['meaning']}")
            
            return response_text
            
        except Exception as e:
            print(f"❌ Error: {e}")
            return None

# 🎮 Let's use it!
analyzer = HTTPAnalyzer()

# 📊 Analyze a sample request
sample_request = """GET /api/users HTTP/1.1
Host: api.example.com
User-Agent: Mozilla/5.0
Accept: application/json
Authorization: Bearer token123"""

result = analyzer.parse_request(sample_request)
if result:
    print(f"🎯 Parsed Request:")
    print(f"  Method: {result['method']}")
    print(f"  Path: {result['path']}")
    print(f"  Headers: {len(result['headers'])} found")

🎯 Try it yourself: Add support for parsing HTTP response bodies and handling different content types!

🎮 Example 2: Multi-Layer Network Monitor

Let’s create a network monitor that works across layers:

# 🏆 Multi-Layer Network Monitor
import socket
import struct
import threading
import time
from collections import defaultdict

class NetworkMonitor:
    def __init__(self):
        self.stats = defaultdict(lambda: {
            'packets': 0,
            'bytes': 0,
            'protocols': defaultdict(int)
        })
        self.running = False
        self.emoji_protocols = {
            6: '🚀 TCP',
            17: '📨 UDP',
            1: '🔔 ICMP',
            2: '🌐 IGMP'
        }
    
    # 🎯 Protocol number to name
    def get_protocol_name(self, protocol_num):
        return self.emoji_protocols.get(protocol_num, f'🔍 Protocol {protocol_num}')
    
    # 📊 Parse Ethernet frame (Layer 2)
    def parse_ethernet_frame(self, data):
        if len(data) < 14:
            return None
        
        # 🔗 Ethernet header is 14 bytes
        eth_header = struct.unpack('!6s6sH', data[:14])
        
        dst_mac = ':'.join(f'{b:02x}' for b in eth_header[0])
        src_mac = ':'.join(f'{b:02x}' for b in eth_header[1])
        eth_type = eth_header[2]
        
        return {
            'destination_mac': dst_mac,
            'source_mac': src_mac,
            'type': eth_type,
            'payload': data[14:]
        }
    
    # 🌐 Parse IP packet (Layer 3)
    def parse_ip_packet(self, data):
        if len(data) < 20:
            return None
        
        # 📦 IP header (first 20 bytes)
        ip_header = struct.unpack('!BBHHHBBH4s4s', data[:20])
        
        version = ip_header[0] >> 4
        header_length = (ip_header[0] & 0xF) * 4
        total_length = ip_header[2]
        protocol = ip_header[6]
        src_ip = socket.inet_ntoa(ip_header[8])
        dst_ip = socket.inet_ntoa(ip_header[9])
        
        return {
            'version': version,
            'header_length': header_length,
            'total_length': total_length,
            'protocol': protocol,
            'source_ip': src_ip,
            'destination_ip': dst_ip,
            'payload': data[header_length:]
        }
    
    # 🚀 Parse TCP segment (Layer 4)
    def parse_tcp_segment(self, data):
        if len(data) < 20:
            return None
        
        # 🎯 TCP header
        tcp_header = struct.unpack('!HHLLHHHH', data[:20])
        
        src_port = tcp_header[0]
        dst_port = tcp_header[1]
        sequence = tcp_header[2]
        acknowledgment = tcp_header[3]
        offset = (tcp_header[4] >> 12) * 4
        flags = tcp_header[4] & 0x1FF
        
        # 🎨 Parse flags
        flag_names = []
        if flags & 0x01: flag_names.append('FIN')
        if flags & 0x02: flag_names.append('SYN')
        if flags & 0x04: flag_names.append('RST')
        if flags & 0x08: flag_names.append('PSH')
        if flags & 0x10: flag_names.append('ACK')
        if flags & 0x20: flag_names.append('URG')
        
        return {
            'source_port': src_port,
            'destination_port': dst_port,
            'sequence': sequence,
            'acknowledgment': acknowledgment,
            'flags': flag_names,
            'payload': data[offset:]
        }
    
    # 📈 Update statistics
    def update_stats(self, src_ip, dst_ip, protocol, packet_size):
        # 🎯 Update source stats
        self.stats[src_ip]['packets'] += 1
        self.stats[src_ip]['bytes'] += packet_size
        self.stats[src_ip]['protocols'][protocol] += 1
        
        # 🎯 Update destination stats
        self.stats[dst_ip]['packets'] += 1
        self.stats[dst_ip]['bytes'] += packet_size
        self.stats[dst_ip]['protocols'][protocol] += 1
    
    # 📊 Display statistics
    def display_stats(self):
        print("\n📊 Network Statistics Dashboard")
        print("=" * 60)
        
        for ip, data in sorted(self.stats.items(), 
                              key=lambda x: x[1]['packets'], 
                              reverse=True)[:10]:
            print(f"\n🌐 IP: {ip}")
            print(f"  📦 Packets: {data['packets']:,}")
            print(f"  💾 Bytes: {data['bytes']:,}")
            print(f"  🔧 Protocols:")
            
            for proto, count in data['protocols'].items():
                proto_name = self.get_protocol_name(proto)
                print(f"    {proto_name}: {count:,} packets")
    
    # 🎮 Demo mode - simulate traffic
    def simulate_traffic(self):
        print("🎮 Starting network simulation...")
        
        # 🌟 Simulate some traffic
        test_ips = [
            ('192.168.1.100', '8.8.8.8', 6, 1500),    # TCP to Google DNS
            ('10.0.0.5', '93.184.216.34', 17, 512),   # UDP to example.com
            ('172.16.0.10', '1.1.1.1', 1, 64),        # ICMP to Cloudflare
            ('192.168.1.100', '140.82.114.4', 6, 8192) # TCP to GitHub
        ]
        
        for _ in range(10):
            for src, dst, proto, size in test_ips:
                self.update_stats(src, dst, proto, size)
                time.sleep(0.1)
        
        self.display_stats()

# 🚀 Run the monitor
monitor = NetworkMonitor()
monitor.simulate_traffic()

🚀 Advanced Concepts

🧙‍♂️ Advanced Topic 1: Raw Socket Programming

When you’re ready to level up, try working with raw sockets:

# 🎯 Advanced raw socket packet crafting
import socket
import struct
import random

class PacketCrafter:
    def __init__(self):
        self.packet_id = random.randint(1, 65535)
    
    # 🪄 Calculate checksum
    def checksum(self, data):
        if len(data) % 2:
            data += b'\x00'
        
        s = 0
        for i in range(0, len(data), 2):
            w = (data[i] << 8) + data[i + 1]
            s = (s + w) & 0xFFFF
        
        return ~s & 0xFFFF
    
    # 🚀 Craft IP header
    def craft_ip_header(self, src_ip, dst_ip, protocol=6):
        # 📦 IP header fields
        version = 4
        ihl = 5  # Header length
        tos = 0
        total_length = 20  # Will be updated
        identification = self.packet_id
        flags = 0
        fragment_offset = 0
        ttl = 64
        header_checksum = 0
        
        # 🎨 Pack header
        header = struct.pack('!BBHHHBBH4s4s',
            (version << 4) | ihl,
            tos,
            total_length,
            identification,
            (flags << 13) | fragment_offset,
            ttl,
            protocol,
            header_checksum,
            socket.inet_aton(src_ip),
            socket.inet_aton(dst_ip)
        )
        
        # ✨ Calculate checksum
        header_checksum = self.checksum(header)
        
        # 🔧 Repack with checksum
        header = struct.pack('!BBHHHBBH4s4s',
            (version << 4) | ihl,
            tos,
            total_length,
            identification,
            (flags << 13) | fragment_offset,
            ttl,
            protocol,
            header_checksum,
            socket.inet_aton(src_ip),
            socket.inet_aton(dst_ip)
        )
        
        return header
    
    # 🌟 Craft TCP header
    def craft_tcp_header(self, src_port, dst_port, flags='S'):
        # 🎯 TCP header fields
        sequence = random.randint(0, 2**32 - 1)
        acknowledgment = 0
        data_offset = 5  # 20 bytes
        
        # 🎨 Convert flags
        flag_bits = 0
        if 'F' in flags: flag_bits |= 0x01  # FIN
        if 'S' in flags: flag_bits |= 0x02  # SYN
        if 'R' in flags: flag_bits |= 0x04  # RST
        if 'P' in flags: flag_bits |= 0x08  # PSH
        if 'A' in flags: flag_bits |= 0x10  # ACK
        if 'U' in flags: flag_bits |= 0x20  # URG
        
        window = 8192
        checksum = 0
        urgent_pointer = 0
        
        # 📦 Pack TCP header
        header = struct.pack('!HHLLHHHH',
            src_port,
            dst_port,
            sequence,
            acknowledgment,
            (data_offset << 12) | flag_bits,
            window,
            checksum,
            urgent_pointer
        )
        
        print(f"🚀 Crafted TCP packet with flags: {flags}")
        return header

🏗️ Advanced Topic 2: Protocol State Machines

For the brave developers, implement protocol state machines:

# 🚀 TCP State Machine Implementation
class TCPStateMachine:
    def __init__(self):
        self.states = {
            'CLOSED': '🔒',
            'LISTEN': '👂',
            'SYN_SENT': '📤',
            'SYN_RECEIVED': '📥',
            'ESTABLISHED': '✅',
            'FIN_WAIT_1': '👋',
            'FIN_WAIT_2': '🤝',
            'TIME_WAIT': '⏰',
            'CLOSE_WAIT': '⏳',
            'LAST_ACK': '🔚'
        }
        self.current_state = 'CLOSED'
        self.transitions = []
    
    # 🎯 State transition
    def transition(self, event):
        old_state = self.current_state
        new_state = self._get_next_state(event)
        
        if new_state:
            self.current_state = new_state
            emoji_old = self.states.get(old_state, '❓')
            emoji_new = self.states.get(new_state, '❓')
            
            self.transitions.append({
                'from': old_state,
                'to': new_state,
                'event': event,
                'emoji': f"{emoji_old} → {emoji_new}"
            })
            
            print(f"🔄 State transition: {emoji_old} {old_state} → {emoji_new} {new_state}")
            return True
        
        print(f"❌ Invalid transition from {old_state} with event {event}")
        return False
    
    # 🧠 State transition logic
    def _get_next_state(self, event):
        transitions = {
            ('CLOSED', 'LISTEN'): 'LISTEN',
            ('CLOSED', 'CONNECT'): 'SYN_SENT',
            ('LISTEN', 'SYN'): 'SYN_RECEIVED',
            ('SYN_SENT', 'SYN_ACK'): 'ESTABLISHED',
            ('SYN_RECEIVED', 'ACK'): 'ESTABLISHED',
            ('ESTABLISHED', 'FIN'): 'FIN_WAIT_1',
            ('FIN_WAIT_1', 'ACK'): 'FIN_WAIT_2',
            ('FIN_WAIT_2', 'FIN'): 'TIME_WAIT',
            ('ESTABLISHED', 'FIN_RECEIVED'): 'CLOSE_WAIT',
            ('CLOSE_WAIT', 'CLOSE'): 'LAST_ACK',
            ('LAST_ACK', 'ACK'): 'CLOSED',
            ('TIME_WAIT', 'TIMEOUT'): 'CLOSED'
        }
        
        return transitions.get((self.current_state, event))
    
    # 📊 Show connection lifecycle
    def show_lifecycle(self):
        print("\n🎯 TCP Connection Lifecycle:")
        for transition in self.transitions:
            print(f"  {transition['emoji']} on {transition['event']}")

⚠️ Common Pitfalls and Solutions

😱 Pitfall 1: Forgetting Network Byte Order

# ❌ Wrong way - host byte order issues!
def pack_wrong():
    port = 8080
    # This will be interpreted differently on different systems!
    packed = struct.pack('H', port)  # Missing network byte order
    return packed

# ✅ Correct way - always use network byte order!
def pack_correct():
    port = 8080
    # Use '!' for network byte order (big-endian)
    packed = struct.pack('!H', port)  # ✅ Network byte order
    return packed

# 🛡️ Helper function for safety
def safe_pack_port(port):
    if not 0 <= port <= 65535:
        raise ValueError(f"⚠️ Invalid port: {port}")
    return struct.pack('!H', port)

🤯 Pitfall 2: Not Handling Partial Receives

# ❌ Dangerous - might not receive all data!
def receive_wrong(sock):
    data = sock.recv(1024)  # 💥 Might be partial!
    return data

# ✅ Safe - handle partial receives!
def receive_all(sock, expected_size):
    buffer = b''
    bytes_received = 0
    
    while bytes_received < expected_size:
        chunk = sock.recv(min(expected_size - bytes_received, 4096))
        if not chunk:
            print("⚠️ Connection closed by peer!")
            break
        
        buffer += chunk
        bytes_received += len(chunk)
        print(f"📥 Received {bytes_received}/{expected_size} bytes")
    
    return buffer

# 🎯 Even better - with timeout handling
def receive_with_timeout(sock, expected_size, timeout=5):
    sock.settimeout(timeout)
    try:
        return receive_all(sock, expected_size)
    except socket.timeout:
        print("⏰ Receive timeout!")
        return None

🛠️ Best Practices

1. 🎯 Always Use Network Byte Order: Use struct with ’!’ prefix
2. 📝 Document Protocol Formats: Clear documentation saves debugging time
3. 🛡️ Validate All Input: Never trust network data
4. 🎨 Use Meaningful Constants: Define protocol constants clearly
5. ✨ Handle Errors Gracefully: Networks are unreliable by nature

🧪 Hands-On Exercise

🎯 Challenge: Build a Protocol Analyzer

Create a multi-protocol network analyzer:

📋 Requirements:

• ✅ Support for TCP, UDP, and ICMP protocols
• 🏷️ Parse headers at layers 3 and 4
• 👤 Track connections and statistics
• 📅 Log timestamps for all packets
• 🎨 Display protocol information with emojis!

🚀 Bonus Points:

• Add support for HTTP parsing
• Implement connection tracking
• Create a real-time dashboard

💡 Solution

# 🎯 Multi-Protocol Network Analyzer
import socket
import struct
import threading
import time
from datetime import datetime
from collections import defaultdict

class ProtocolAnalyzer:
    def __init__(self):
        self.protocols = {
            1: {'name': 'ICMP', 'emoji': '🔔', 'parser': self.parse_icmp},
            6: {'name': 'TCP', 'emoji': '🚀', 'parser': self.parse_tcp},
            17: {'name': 'UDP', 'emoji': '📨', 'parser': self.parse_udp}
        }
        self.connections = defaultdict(lambda: {
            'packets': 0,
            'bytes': 0,
            'start_time': datetime.now(),
            'last_seen': datetime.now()
        })
        self.stats = defaultdict(int)
    
    # 🌐 Parse IP header
    def parse_ip_header(self, data):
        if len(data) < 20:
            return None
        
        header = struct.unpack('!BBHHHBBH4s4s', data[:20])
        
        version = header[0] >> 4
        ihl = (header[0] & 0xF) * 4
        total_length = header[2]
        protocol = header[6]
        src_ip = socket.inet_ntoa(header[8])
        dst_ip = socket.inet_ntoa(header[9])
        
        return {
            'version': version,
            'header_length': ihl,
            'total_length': total_length,
            'protocol': protocol,
            'source_ip': src_ip,
            'destination_ip': dst_ip,
            'payload': data[ihl:]
        }
    
    # 🚀 Parse TCP
    def parse_tcp(self, data, ip_info):
        if len(data) < 20:
            return None
        
        tcp_header = struct.unpack('!HHLLHHHH', data[:20])
        
        src_port = tcp_header[0]
        dst_port = tcp_header[1]
        sequence = tcp_header[2]
        flags = tcp_header[4] & 0x1FF
        
        # 🎨 Format connection key
        conn_key = f"{ip_info['source_ip']}:{src_port} -> {ip_info['destination_ip']}:{dst_port}"
        
        # 📊 Update connection tracking
        self.connections[conn_key]['packets'] += 1
        self.connections[conn_key]['bytes'] += ip_info['total_length']
        self.connections[conn_key]['last_seen'] = datetime.now()
        
        # 🎯 Parse flags
        flag_str = ""
        if flags & 0x02: flag_str += "S"  # SYN
        if flags & 0x10: flag_str += "A"  # ACK
        if flags & 0x01: flag_str += "F"  # FIN
        if flags & 0x04: flag_str += "R"  # RST
        
        return {
            'source_port': src_port,
            'destination_port': dst_port,
            'sequence': sequence,
            'flags': flag_str,
            'connection': conn_key
        }
    
    # 📨 Parse UDP
    def parse_udp(self, data, ip_info):
        if len(data) < 8:
            return None
        
        udp_header = struct.unpack('!HHHH', data[:8])
        
        src_port = udp_header[0]
        dst_port = udp_header[1]
        length = udp_header[2]
        
        return {
            'source_port': src_port,
            'destination_port': dst_port,
            'length': length
        }
    
    # 🔔 Parse ICMP
    def parse_icmp(self, data, ip_info):
        if len(data) < 8:
            return None
        
        icmp_header = struct.unpack('!BBHHH', data[:8])
        
        icmp_type = icmp_header[0]
        code = icmp_header[1]
        
        # 🎨 ICMP type meanings
        type_meanings = {
            0: 'Echo Reply 📣',
            3: 'Destination Unreachable ❌',
            8: 'Echo Request 📢',
            11: 'Time Exceeded ⏰'
        }
        
        return {
            'type': icmp_type,
            'code': code,
            'meaning': type_meanings.get(icmp_type, f'Type {icmp_type}')
        }
    
    # 📊 Analyze packet
    def analyze_packet(self, data):
        # 🌐 Parse IP header
        ip_info = self.parse_ip_header(data)
        if not ip_info:
            return None
        
        # 🎯 Get protocol info
        protocol_info = self.protocols.get(ip_info['protocol'])
        if not protocol_info:
            return None
        
        # 📈 Update stats
        self.stats[protocol_info['name']] += 1
        
        # 🔧 Parse protocol-specific data
        proto_data = protocol_info['parser'](ip_info['payload'], ip_info)
        
        # 🎨 Create analysis result
        result = {
            'timestamp': datetime.now().isoformat(),
            'emoji': protocol_info['emoji'],
            'protocol': protocol_info['name'],
            'source': ip_info['source_ip'],
            'destination': ip_info['destination_ip'],
            'size': ip_info['total_length'],
            'details': proto_data
        }
        
        return result
    
    # 📊 Display statistics
    def show_stats(self):
        print("\n📊 Protocol Statistics")
        print("=" * 40)
        
        total_packets = sum(self.stats.values())
        if total_packets == 0:
            print("No packets captured yet!")
            return
        
        for protocol, count in sorted(self.stats.items()):
            percentage = (count / total_packets) * 100
            proto_info = next((p for p in self.protocols.values() 
                             if p['name'] == protocol), {})
            emoji = proto_info.get('emoji', '📦')
            
            print(f"{emoji} {protocol}: {count:,} packets ({percentage:.1f}%)")
        
        print(f"\n📈 Total packets: {total_packets:,}")
        
        # 🔗 Show active connections
        print("\n🔗 Active TCP Connections")
        print("=" * 40)
        
        for conn, info in sorted(self.connections.items(), 
                               key=lambda x: x[1]['packets'], 
                               reverse=True)[:5]:
            duration = (info['last_seen'] - info['start_time']).total_seconds()
            print(f"\n{conn}")
            print(f"  📦 Packets: {info['packets']:,}")
            print(f"  💾 Bytes: {info['bytes']:,}")
            print(f"  ⏱️ Duration: {duration:.1f}s")
    
    # 🎮 Demo mode
    def demo(self):
        print("🎮 Running Protocol Analyzer Demo...")
        
        # Simulate some packets
        test_packets = [
            # TCP SYN
            {'protocol': 6, 'src': '192.168.1.100', 'dst': '8.8.8.8', 
             'src_port': 54321, 'dst_port': 80, 'flags': 'S'},
            
            # UDP DNS
            {'protocol': 17, 'src': '192.168.1.100', 'dst': '8.8.8.8',
             'src_port': 54322, 'dst_port': 53},
            
            # ICMP Echo
            {'protocol': 1, 'src': '192.168.1.100', 'dst': '8.8.8.8',
             'type': 8},
            
            # TCP ACK
            {'protocol': 6, 'src': '8.8.8.8', 'dst': '192.168.1.100',
             'src_port': 80, 'dst_port': 54321, 'flags': 'SA'}
        ]
        
        for packet in test_packets * 3:
            # Simulate packet analysis
            if packet['protocol'] == 6:
                conn_key = f"{packet['src']}:{packet['src_port']} -> {packet['dst']}:{packet['dst_port']}"
                self.connections[conn_key]['packets'] += 1
                self.connections[conn_key]['bytes'] += 1500
                self.stats['TCP'] += 1
            elif packet['protocol'] == 17:
                self.stats['UDP'] += 1
            elif packet['protocol'] == 1:
                self.stats['ICMP'] += 1
            
            time.sleep(0.2)
        
        self.show_stats()

# 🚀 Run the analyzer
analyzer = ProtocolAnalyzer()
analyzer.demo()

🎓 Key Takeaways

You’ve learned so much! Here’s what you can now do:

• ✅ Understand network protocol layers with confidence 💪
• ✅ Work with different OSI and TCP/IP layers in Python 🛡️
• ✅ Parse and analyze network protocols like a pro 🎯
• ✅ Build network monitoring tools using Python 🐛
• ✅ Handle low-level networking with raw sockets! 🚀

Remember: Network protocols are the foundation of all internet communication. Understanding them gives you superpowers! 🤝

🤝 Next Steps

Congratulations! 🎉 You’ve mastered network protocol layers!

Here’s what to do next:

1. 💻 Practice with the packet analyzer exercise above
2. 🏗️ Build a simple protocol implementation (like a chat protocol)
3. 📚 Move on to our next tutorial: UDP Sockets and Connectionless Communication
4. 🌟 Experiment with tools like Wireshark to see real protocols in action!

Remember: Every network expert started by understanding the basics. Keep exploring, keep learning, and most importantly, have fun with networking! 🚀

Happy networking! 🎉🚀✨