UDP Header Format: Anatomy of Connectionless Transport

UDP Header (8 bytes total)
═══════════════════════════════════════════════════════════════════
 
 0                   1                   2                   3
 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
├─┴─┴─┴─┴─┴─┴─┴─┴─┴─┴─┴─┴─┴─┴─┴─┴─┴─┴─┴─┴─┴─┴─┴─┴─┴─┴─┴─┴─┴─┴─┴─┤
│         Source Port           │       Destination Port        │
│         (16 bits)             │         (16 bits)             │
│         Bytes 0-1             │         Bytes 2-3             │
├───────────────────────────────┴───────────────────────────────┤
│            Length             │           Checksum            │
│          (16 bits)            │          (16 bits)            │
│          Bytes 4-5            │          Bytes 6-7            │
└───────────────────────────────┴───────────────────────────────┘
 
Destination Port Field Details:
───────────────────────────────────────────────────────────────────
• Offset: 16 bits (2 bytes) from start of UDP header
• Size: 16 bits (2 bytes)
• Range: 0 - 65535 (2^16 - 1)
• Byte Order: Network byte order (big-endian)
• Purpose: Identifies the receiving application/process

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
109
110
111

"""
Using ephemeral range ports for private services.
Common pattern for internal microservices, testing, and development.
"""
 
import socket
import logging
 
# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
 
class PrivateUDPService:
    """
    Example of a UDP service using an ephemeral-range port.
    Suitable for internal services, microservices, or development.
    """
    
    # Common port choices for private services
    COMMON_DEV_PORTS = [
        49200,  # Development instance 1
        49201,  # Development instance 2
        50000,  # Testing services
        55555,  # Easy to remember
        60000,  # Higher range
    ]
    
    def __init__(self, port: int = 0, host: str = "0.0.0.0"):
        """
        Initialize the service.
        
        Args:
            port: Specific port or 0 for OS assignment
            host: Interface to bind to
        """
        self.socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        
        if port == 0:
            # Let OS choose from ephemeral range
            self.socket.bind((host, 0))
            self.port = self.socket.getsockname()[1]
            logger.info(f"OS assigned port: {self.port}")
        else:
            # Use specified port
            try:
                self.socket.bind((host, port))
                self.port = port
                logger.info(f"Bound to specified port: {self.port}")
            except OSError as e:
                logger.error(f"Cannot bind to port {port}: {e}")
                # Fallback to OS assignment
                self.socket.bind((host, 0))
                self.port = self.socket.getsockname()[1]
                logger.info(f"Fallback to OS-assigned port: {self.port}")
        
        self.host = host
    
    def get_service_address(self) -> tuple:
        """Return the address clients should use to reach this service."""
        return (self.host, self.port)
    
    def receive_datagram(self, buffer_size: int = 4096) -> tuple:
        """Receive a datagram, returning (data, sender_address)."""
        return self.socket.recvfrom(buffer_size)
    
    def send_response(self, data: bytes, address: tuple) -> int:
        """Send a response datagram to the specified address."""
        return self.socket.sendto(data, address)
    
    def close(self):
        """Release the socket."""
        self.socket.close()
 
def demonstrate_port_selection():
    """Show different approaches to destination port selection."""
    
    print("=== Port Selection Strategies ===\n")
    
    # Strategy 1: OS-assigned (maximum flexibility)
    service1 = PrivateUDPService(port=0)
    print(f"Strategy 1 (OS-assigned): Port {service1.port}")
    service1.close()
    
    # Strategy 2: Fixed private port (predictable for configuration)
    service2 = PrivateUDPService(port=55555)
    print(f"Strategy 2 (Fixed private): Port {service2.port}")
    service2.close()
    
    # Strategy 3: Check common ports, fallback
    chosen_port = None
    for port in PrivateUDPService.COMMON_DEV_PORTS:
        test_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        try:
            test_socket.bind(("0.0.0.0", port))
            chosen_port = port
            test_socket.close()
            break
        except OSError:
            test_socket.close()
            continue
    
    if chosen_port:
        service3 = PrivateUDPService(port=chosen_port)
        print(f"Strategy 3 (First available): Port {service3.port}")
        service3.close()
    else:
        print("Strategy 3: All preferred ports in use, would use OS assignment")
 
if __name__ == "__main__":
    demonstrate_port_selection()

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
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207

"""
Comprehensive UDP Server Binding and Listening Patterns.
 
Demonstrates:
1. Basic single-address binding
2. Wildcard binding (all interfaces)
3. Interface-specific binding
4. Non-blocking operation
5. Timeout handling
"""
 
import socket
import select
import logging
from typing import Optional, Tuple
 
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
 
class UDPServer:
    """
    Production-ready UDP server demonstrating proper binding patterns.
    """
    
    def __init__(
        self,
        port: int,
        host: str = "0.0.0.0",
        buffer_size: int = 65535,
        timeout: Optional[float] = None
    ):
        """
        Initialize UDP server.
        
        Args:
            port: Port to listen on (required for servers)
            host: Interface to bind to
                  "0.0.0.0" = all IPv4 interfaces
                  "::" = all IPv6 interfaces (dual-stack)
                  "127.0.0.1" = localhost only
                  "192.168.1.100" = specific interface
            buffer_size: Maximum datagram size to receive
            timeout: Socket timeout in seconds (None = blocking)
        """
        self.port = port
        self.host = host
        self.buffer_size = buffer_size
        
        # Create IPv4 UDP socket
        self.socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        
        # Socket options for server operation
        self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        
        # Optional: Allow multiple processes to bind (for load balancing)
        # self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1)
        
        # Set timeout if specified
        if timeout is not None:
            self.socket.settimeout(timeout)
        
        # Bind to the specified address
        try:
            self.socket.bind((host, port))
            bound_addr = self.socket.getsockname()
            logger.info(f"Server bound to {bound_addr[0]}:{bound_addr[1]}")
        except PermissionError:
            logger.error(
                f"Permission denied binding to port {port}. "
                "Ports below 1024 require elevated privileges."
            )
            raise
        except OSError as e:
            logger.error(f"Cannot bind to {host}:{port} - {e}")
            raise
    
    def receive(self) -> Tuple[bytes, Tuple[str, int]]:
        """
        Receive a single datagram.
        
        Returns:
            Tuple of (data, (sender_ip, sender_port))
        
        Raises:
            socket.timeout: If timeout is set and no data arrives
        """
        data, addr = self.socket.recvfrom(self.buffer_size)
        logger.debug(f"Received {len(data)} bytes from {addr}")
        return data, addr
    
    def send_to(self, data: bytes, address: Tuple[str, int]) -> int:
        """
        Send a datagram to a specific address.
        
        Returns:
            Number of bytes sent
        """
        sent = self.socket.sendto(data, address)
        logger.debug(f"Sent {sent} bytes to {address}")
        return sent
    
    def run_echo_server(self):
        """Simple echo server demonstrating the receive/reply pattern."""
        logger.info(f"Echo server running on {self.host}:{self.port}")
        
        while True:
            try:
                data, client = self.receive()
                logger.info(f"Echo: {len(data)} bytes from {client}")
                self.send_to(data, client)
            except socket.timeout:
                logger.debug("Timeout - no data received")
            except KeyboardInterrupt:
                logger.info("Server shutting down")
                break
    
    def close(self):
        """Release server socket."""
        self.socket.close()
        logger.info("Server socket closed")
 
 
class MultiInterfaceUDPServer:
    """
    Advanced server that binds to multiple specific interfaces.
    Useful when you need different behavior per interface.
    """
    
    def __init__(self, port: int, interfaces: list):
        """
        Args:
            port: Port to listen on
            interfaces: List of IP addresses to bind to
        """
        self.sockets = {}
        
        for interface in interfaces:
            sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
            sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
            sock.bind((interface, port))
            sock.setblocking(False)
            self.sockets[interface] = sock
            logger.info(f"Bound to {interface}:{port}")
    
    def poll(self, timeout: float = 1.0) -> Optional[Tuple]:
        """
        Poll all sockets for incoming data.
        
        Returns:
            (data, sender_address, receiving_interface) or None
        """
        readable, _, _ = select.select(
            self.sockets.values(), [], [], timeout
        )
        
        for sock in readable:
            data, addr = sock.recvfrom(65535)
            # Find which interface received this
            interface = sock.getsockname()[0]
            return data, addr, interface
        
        return None
    
    def close(self):
        for sock in self.sockets.values():
            sock.close()
 
 
def demonstrate_binding_scenarios():
    """Show different binding configurations."""
    
    print("=== UDP Binding Demonstrations ===\n")
    
    # Scenario 1: Wildcard binding (most common)
    print("1. Wildcard Binding (0.0.0.0)")
    try:
        server1 = UDPServer(port=55001, host="0.0.0.0")
        print(f"   Accepts connections on ALL network interfaces")
        server1.close()
    except Exception as e:
        print(f"   Failed: {e}")
    
    # Scenario 2: Localhost only (secure for local services)
    print("\n2. Localhost Only (127.0.0.1)")
    try:
        server2 = UDPServer(port=55002, host="127.0.0.1")
        print(f"   Only accessible from this machine")
        server2.close()
    except Exception as e:
        print(f"   Failed: {e}")
    
    # Scenario 3: Show what happens with port in use
    print("\n3. Port Already in Use")
    server3a = UDPServer(port=55003)
    try:
        server3b = UDPServer(port=55003)
        print("   Unexpectedly succeeded (SO_REUSEADDR)")
        server3b.close()
    except OSError as e:
        print(f"   Expected failure: {e}")
    server3a.close()
 
if __name__ == "__main__":
    demonstrate_binding_scenarios()