UDP Checksum

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

# Demonstrating the tunnel exception for IPv6 UDP checksums
 
class IPv6TunnelChecksumPolicy:
    """
    Policy engine for determining when zero checksum is allowed
    per RFC 6935/6936.
    """
    
    # Protocols explicitly allowed to use zero checksum
    ZERO_CHECKSUM_ALLOWED = {
        'VXLAN',
        'GENEVE',
        'GUE',
        # Add other explicitly permitted protocols
    }
    
    @classmethod
    def can_use_zero_checksum(
        cls,
        protocol_name: str,
        endpoints_configured: bool,
        inner_has_integrity: bool
    ) -> tuple[bool, str]:
        """
        Determine if zero checksum is permitted for this tunnel.
        
        Returns:
            (allowed: bool, reason: str)
        """
        # Check 1: Protocol must be in allowed list
        if protocol_name not in cls.ZERO_CHECKSUM_ALLOWED:
            return (False, f"Protocol '{protocol_name}' not in allowed list")
        
        # Check 2: Endpoints must be explicitly configured
        if not endpoints_configured:
            return (False, "Endpoints not mutually configured for zero checksum")
        
        # Check 3: Inner protocol must provide integrity
        if not inner_has_integrity:
            return (False, "Inner protocol lacks integrity check")
        
        return (True, f"Zero checksum allowed for {protocol_name}")
    
    @classmethod
    def validate_received_zero_checksum(
        cls,
        protocol_name: str,
        expected_protocol: str
    ) -> tuple[bool, str]:
        """
        Validate a received packet with zero checksum.
        
        The receiver must be configured to expect this specific 
        tunnel type with zero checksum.
        """
        if protocol_name != expected_protocol:
            return (False, "Unexpected protocol for zero checksum")
        
        if protocol_name not in cls.ZERO_CHECKSUM_ALLOWED:
            return (False, "Protocol not authorized for zero checksum")
        
        return (True, "Packet accepted per RFC 6936")
 
# Example usage
if __name__ == "__main__":
    # VXLAN tunnel - allowed
    allowed, reason = IPv6TunnelChecksumPolicy.can_use_zero_checksum(
        protocol_name="VXLAN",
        endpoints_configured=True,
        inner_has_integrity=True  # Ethernet frames inside
    )
    print(f"VXLAN: {reason}")
    
    # DNS - not allowed
    allowed, reason = IPv6TunnelChecksumPolicy.can_use_zero_checksum(
        protocol_name="DNS",
        endpoints_configured=False,
        inner_has_integrity=False
    )
    print(f"DNS: {reason}")

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

import struct
import socket
 
class IPv6UDPChecksum:
    """
    Mandatory UDP checksum implementation for IPv6.
    """
    
    @staticmethod
    def build_pseudo_header(
        src_ip: str,
        dst_ip: str,
        udp_length: int
    ) -> bytes:
        """
        Build the IPv6 pseudo-header for checksum calculation.
        
        IPv6 pseudo-header format (40 bytes):
        - Source Address (16 bytes)
        - Destination Address (16 bytes)
        - Upper-Layer Packet Length (4 bytes)
        - Zero padding (3 bytes)
        - Next Header (1 byte) = 17 for UDP
        """
        src_bytes = socket.inet_pton(socket.AF_INET6, src_ip)
        dst_bytes = socket.inet_pton(socket.AF_INET6, dst_ip)
        
        return struct.pack(
            '!16s16sI3xB',
            src_bytes,
            dst_bytes,
            udp_length,
            17  # UDP protocol number
        )
    
    @staticmethod
    def compute_checksum(data: bytes) -> int:
        """Compute one's complement sum checksum."""
        if len(data) % 2 == 1:
            data += b'\x00'
        
        total = 0
        for i in range(0, len(data), 2):
            word = (data[i] << 8) + data[i + 1]
            total += word
            while total > 0xFFFF:
                total = (total & 0xFFFF) + (total >> 16)
        
        checksum = ~total & 0xFFFF
        
        # CRITICAL: In IPv6, if checksum computes to 0, use 0xFFFF
        # This is the ONLY case where 0xFFFF is substituted
        if checksum == 0:
            checksum = 0xFFFF
        
        return checksum
    
    @classmethod
    def create_udp_segment(
        cls,
        src_ip: str,
        dst_ip: str,
        src_port: int,
        dst_port: int,
        payload: bytes
    ) -> bytes:
        """
        Create a complete UDP segment with mandatory checksum.
        """
        udp_length = 8 + len(payload)
        
        # Build pseudo-header
        pseudo_header = cls.build_pseudo_header(src_ip, dst_ip, udp_length)
        
        # Build UDP header with checksum placeholder
        udp_header = struct.pack(
            '!HHHH',
            src_port,
            dst_port,
            udp_length,
            0  # Checksum placeholder
        )
        
        # Calculate checksum over all
        checksum_data = pseudo_header + udp_header + payload
        checksum = cls.compute_checksum(checksum_data)
        
        # Rebuild header with actual checksum
        udp_header = struct.pack(
            '!HHHH',
            src_port,
            dst_port,
            udp_length,
            checksum
        )
        
        return udp_header + payload
    
    @classmethod
    def verify_received_segment(
        cls,
        src_ip: str,
        dst_ip: str,
        udp_segment: bytes,
        allow_tunnel_exception: bool = False
    ) -> tuple[bool, str]:
        """
        Verify checksum of a received IPv6 UDP segment.
        
        Args:
            src_ip: Source IPv6 address from IP header
            dst_ip: Destination IPv6 address from IP header
            udp_segment: Complete UDP segment
            allow_tunnel_exception: If True, accept zero checksum
        
        Returns:
            (valid: bool, message: str)
        """
        # Extract checksum from UDP header
        received_checksum = struct.unpack('!H', udp_segment[6:8])[0]
        
        # Check for zero checksum
        if received_checksum == 0:
            if allow_tunnel_exception:
                return (True, "Accepted: Tunnel zero-checksum exception")
            else:
                # In IPv6, zero checksum is NOT allowed for regular traffic
                return (False, "REJECTED: Zero checksum invalid in IPv6")
        
        # Extract UDP length and verify
        udp_length = struct.unpack('!H', udp_segment[4:6])[0]
        
        # Build pseudo-header
        pseudo_header = cls.build_pseudo_header(src_ip, dst_ip, udp_length)
        
        # Verify checksum (sum everything including checksum)
        verify_data = pseudo_header + udp_segment
        if len(verify_data) % 2 == 1:
            verify_data += b'\x00'
        
        total = 0
        for i in range(0, len(verify_data), 2):
            word = (verify_data[i] << 8) + verify_data[i + 1]
            total += word
            while total > 0xFFFF:
                total = (total & 0xFFFF) + (total >> 16)
        
        if total == 0xFFFF:
            return (True, "Valid checksum")
        else:
            return (False, f"REJECTED: Invalid checksum (sum=0x{total:04X})")
 
# Demonstration
if __name__ == "__main__":
    segment = IPv6UDPChecksum.create_udp_segment(
        src_ip="2001:db8::1",
        dst_ip="2001:db8::2",
        src_port=12345,
        dst_port=53,
        payload=b"Hello IPv6!"
    )
    
    print(f"UDP segment hex: {segment.hex()}")
    print(f"Checksum bytes: {segment[6:8].hex()}")
    
    # Verify
    valid, msg = IPv6UDPChecksum.verify_received_segment(
        "2001:db8::1",
        "2001:db8::2",
        segment
    )
    print(f"Verification: {msg}")

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

import socket
import struct
from enum import Enum
from typing import Tuple
 
class IPVersion(Enum):
    IPv4 = 4
    IPv6 = 6
 
class DualStackUDPChecksum:
    """
    Unified checksum handling for dual-stack UDP.
    """
    
    @staticmethod
    def detect_version(ip_address: str) -> IPVersion:
        """Detect IP version from address string."""
        try:
            socket.inet_pton(socket.AF_INET, ip_address)
            return IPVersion.IPv4
        except socket.error:
            pass
        
        try:
            socket.inet_pton(socket.AF_INET6, ip_address)
            return IPVersion.IPv6
        except socket.error:
            raise ValueError(f"Invalid IP address: {ip_address}")
    
    @classmethod
    def build_pseudo_header(
        cls,
        src_ip: str,
        dst_ip: str,
        udp_length: int
    ) -> Tuple[bytes, IPVersion]:
        """
        Build appropriate pseudo-header based on IP version.
        
        Returns:
            (pseudo_header_bytes, ip_version)
        """
        version = cls.detect_version(src_ip)
        
        if version == IPVersion.IPv4:
            # 12-byte IPv4 pseudo-header
            pseudo = struct.pack(
                '!4s4sBBH',
                socket.inet_aton(src_ip),
                socket.inet_aton(dst_ip),
                0,      # Zero padding
                17,     # Protocol
                udp_length
            )
        else:
            # 40-byte IPv6 pseudo-header
            pseudo = struct.pack(
                '!16s16sI3xB',
                socket.inet_pton(socket.AF_INET6, src_ip),
                socket.inet_pton(socket.AF_INET6, dst_ip),
                udp_length,
                17
            )
        
        return (pseudo, version)
    
    @classmethod
    def calculate_checksum(
        cls,
        src_ip: str,
        dst_ip: str,
        udp_header_no_checksum: bytes,
        payload: bytes
    ) -> int:
        """
        Calculate checksum with automatic version detection.
        Always computes checksum (recommended for both versions).
        """
        udp_length = 8 + len(payload)
        pseudo_header, version = cls.build_pseudo_header(
            src_ip, dst_ip, udp_length
        )
        
        # Combine all data for checksum
        data = pseudo_header + udp_header_no_checksum + payload
        if len(data) % 2 == 1:
            data += b'\x00'
        
        # One's complement sum
        total = 0
        for i in range(0, len(data), 2):
            total += (data[i] << 8) + data[i + 1]
            while total > 0xFFFF:
                total = (total & 0xFFFF) + (total >> 16)
        
        checksum = ~total & 0xFFFF
        
        # Handle zero checksum
        if checksum == 0:
            checksum = 0xFFFF  # Required for both versions when computing
        
        return checksum
    
    @classmethod
    def verify_checksum(
        cls,
        src_ip: str,
        dst_ip: str,
        udp_segment: bytes
    ) -> Tuple[bool, str]:
        """
        Verify checksum with version-appropriate rules.
        """
        received_checksum = struct.unpack('!H', udp_segment[6:8])[0]
        udp_length = struct.unpack('!H', udp_segment[4:6])[0]
        _, version = cls.build_pseudo_header(src_ip, dst_ip, udp_length)
        
        # Handle zero checksum based on version
        if received_checksum == 0:
            if version == IPVersion.IPv6:
                return (False, "REJECT: Zero checksum invalid for IPv6")
            else:
                # IPv4: zero means "no checksum" - accept per spec
                return (True, "ACCEPT: IPv4 zero-checksum (no verification)")
        
        # Verify the checksum
        pseudo_header, _ = cls.build_pseudo_header(
            src_ip, dst_ip, udp_length
        )
        data = pseudo_header + udp_segment
        if len(data) % 2 == 1:
            data += b'\x00'
        
        total = 0
        for i in range(0, len(data), 2):
            total += (data[i] << 8) + data[i + 1]
            while total > 0xFFFF:
                total = (total & 0xFFFF) + (total >> 16)
        
        if total == 0xFFFF:
            return (True, f"VALID: {version.name} checksum verified")
        else:
            return (False, f"INVALID: {version.name} checksum failed")
 
# Test both versions
if __name__ == "__main__":
    # IPv4 test
    print("IPv4 pseudo-header:")
    ph4, v4 = DualStackUDPChecksum.build_pseudo_header(
        "192.168.1.1", "192.168.1.2", 100
    )
    print(f"  Length: {len(ph4)} bytes, Version: {v4}")
    
    # IPv6 test
    print("IPv6 pseudo-header:")
    ph6, v6 = DualStackUDPChecksum.build_pseudo_header(
        "2001:db8::1", "2001:db8::2", 100
    )
    print(f"  Length: {len(ph6)} bytes, Version: {v6}")

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

import unittest
import struct
import socket
 
class TestIPv6UDPChecksum(unittest.TestCase):
    """Comprehensive test suite for IPv6 UDP checksum compliance."""
    
    def ones_complement_sum(self, data: bytes) -> int:
        """Reference implementation for testing."""
        if len(data) % 2:
            data += b'\x00'
        
        total = 0
        for i in range(0, len(data), 2):
            total += (data[i] << 8) + data[i + 1]
            while total > 0xFFFF:
                total = (total & 0xFFFF) + (total >> 16)
        return total
    
    def build_ipv6_pseudo_header(self, src, dst, length):
        return struct.pack(
            '!16s16sI3xB',
            socket.inet_pton(socket.AF_INET6, src),
            socket.inet_pton(socket.AF_INET6, dst),
            length,
            17  # UDP
        )
    
    def test_pseudo_header_length(self):
        """IPv6 pseudo-header must be 40 bytes."""
        ph = self.build_ipv6_pseudo_header(
            '2001:db8::1', '2001:db8::2', 100
        )
        self.assertEqual(len(ph), 40)
    
    def test_checksum_never_zero(self):
        """Transmitted checksum must never be 0x0000."""
        # Find a payload that would produce checksum = 0
        # and verify it becomes 0xFFFF
        # (In practice, this is rare; we test the logic)
        
        pseudo = self.build_ipv6_pseudo_header(
            '::1', '::1', 8
        )
        # UDP header with checksum = 0 placeholder
        udp = struct.pack('!HHHH', 1234, 5678, 8, 0)
        
        data = pseudo + udp
        sum_val = self.ones_complement_sum(data)
        checksum = ~sum_val & 0xFFFF
        
        # If checksum is 0, it should be changed to 0xFFFF
        if checksum == 0:
            checksum = 0xFFFF
        
        self.assertNotEqual(checksum, 0,
            "Checksum must never be transmitted as zero")
    
    def test_reject_zero_checksum(self):
        """Receiver must reject zero checksum in IPv6."""
        # Simulate receiving a packet with checksum = 0
        received_checksum = 0x0000
        
        # Per IPv6 spec, this MUST be rejected
        is_valid = (received_checksum != 0)
        self.assertFalse(
            received_checksum == 0 and is_valid,
            "Zero checksum must be rejected for IPv6"
        )
    
    def test_valid_checksum_verification(self):
        """Valid checksum should verify to 0xFFFF."""
        pseudo = self.build_ipv6_pseudo_header(
            '2001:db8::1', '2001:db8::2', 20
        )
        
        # Build UDP with checksum=0 placeholder
        udp_no_cs = struct.pack('!HHHH', 1234, 53, 20, 0)
        payload = b'Hello World!'  # 12 bytes
        
        # Calculate correct checksum
        data = pseudo + udp_no_cs + payload
        sum_val = self.ones_complement_sum(data)
        checksum = ~sum_val & 0xFFFF
        if checksum == 0:
            checksum = 0xFFFF
        
        # Now verify: include checksum in segment
        udp_with_cs = struct.pack('!HHHH', 1234, 53, 20, checksum)
        verify_data = pseudo + udp_with_cs + payload
        verify_sum = self.ones_complement_sum(verify_data)
        
        self.assertEqual(verify_sum, 0xFFFF,
            "Valid checksum should verify to 0xFFFF")
    
    def test_invalid_checksum_detection(self):
        """Corrupted packet should not verify to 0xFFFF."""
        pseudo = self.build_ipv6_pseudo_header(
            '2001:db8::1', '2001:db8::2', 20
        )
        
        udp_no_cs = struct.pack('!HHHH', 1234, 53, 20, 0)
        payload = b'Hello World!"
        
        # Calculate correct checksum
        data = pseudo + udp_no_cs + payload
        sum_val = self.ones_complement_sum(data)
        checksum = ~sum_val & 0xFFFF
        if checksum == 0:
            checksum = 0xFFFF
        
        # Corrupt the payload
        corrupted_payload = b'Hello World?'  # '!' -> '?'
        
        udp_with_cs = struct.pack('!HHHH', 1234, 53, 20, checksum)
        verify_data = pseudo + udp_with_cs + corrupted_payload
        verify_sum = self.ones_complement_sum(verify_data)
        
        self.assertNotEqual(verify_sum, 0xFFFF,
            "Corrupted packet must not verify")
    
    def test_odd_length_payload(self):
        """Odd-length payloads must be handled correctly."""
        pseudo = self.build_ipv6_pseudo_header(
            '2001:db8::1', '2001:db8::2', 13  # 8 + 5 (odd payload)
        )
        
        udp_no_cs = struct.pack('!HHHH', 1234, 53, 13, 0)
        payload = b'Hello'  # 5 bytes - odd!
        
        data = pseudo + udp_no_cs + payload
        # Should compute without error
        sum_val = self.ones_complement_sum(data)
        checksum = ~sum_val & 0xFFFF
        
        self.assertIsNotNone(checksum,
            "Odd-length handling must work")
 
if __name__ == '__main__':
    unittest.main()