Internet 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

"""
TCP/UDP Pseudo-Header Structure for IPv4
 
The pseudo-header is prepended to TCP/UDP data for checksum calculation only.
It is never transmitted on the wire.
 
+--------+--------+--------+--------+
|          Source Address          |
+--------+--------+--------+--------+
|       Destination Address        |
+--------+--------+--------+--------+
|  Zero  |Protocol|   TCP Length   |
+--------+--------+--------+--------+
 
For IPv6, the pseudo-header is similar but with 128-bit addresses
and a 32-bit length field.
"""
 
import struct
 
def build_tcp_pseudoheader(
    src_ip: str,
    dst_ip: str,
    tcp_length: int
) -> bytes:
    """
    Construct the IPv4 TCP pseudo-header for checksum calculation.
    
    Args:
        src_ip: Source IP address as dotted-decimal string
        dst_ip: Destination IP address as dotted-decimal string
        tcp_length: Length of TCP header + data in bytes
    
    Returns:
        12-byte pseudo-header
    """
    import socket
    
    src_addr = socket.inet_aton(src_ip)  # 4 bytes
    dst_addr = socket.inet_aton(dst_ip)  # 4 bytes
    reserved = 0                          # 1 byte
    protocol = 6                          # TCP protocol number
    length = tcp_length                   # 2 bytes
    
    # Pack in network byte order (big-endian)
    pseudoheader = struct.pack(
        '!4s4sBBH',
        src_addr,
        dst_addr,
        reserved,
        protocol,
        length
    )
    
    return pseudoheader
 
 
# Example: Build pseudo-header for a TCP segment
pseudo = build_tcp_pseudoheader('192.168.1.100', '10.0.0.1', 32)
print(f"Pseudo-header ({len(pseudo)} bytes): {pseudo.hex()}")

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

def ones_complement_add(a: int, b: int, bits: int = 16) -> int:
    """
    Add two numbers using one's complement arithmetic.
    
    In one's complement:
    - Addition may produce a carry-out bit
    - This carry is added back to the result (end-around carry)
    - Mathematically equivalent to addition modulo (2^bits - 1)
    
    Args:
        a, b: 16-bit unsigned integers
        bits: Number of bits (default 16)
    
    Returns:
        One's complement sum as unsigned integer
    """
    mask = (1 << bits) - 1  # 0xFFFF for 16 bits
    
    # Standard addition
    result = a + b
    
    # Extract carry bits (everything above 'bits' position)
    carry = result >> bits
    
    # Fold carry back into result (end-around carry)
    result = (result & mask) + carry
    
    # May need another fold if the above addition also caused carry
    if result > mask:
        result = (result & mask) + (result >> bits)
    
    return result
 
 
def ones_complement(value: int, bits: int = 16) -> int:
    """
    Compute the one's complement (bitwise NOT within bit width).
    
    Args:
        value: Integer to complement
        bits: Number of bits
    
    Returns:
        One's complement of value
    """
    mask = (1 << bits) - 1
    return (~value) & mask
 
 
# Demonstration
print("One's Complement Addition Examples:")
print(f"0x8001 + 0x8001 = 0x{ones_complement_add(0x8001, 0x8001):04X}")
print(f"0xFFFF + 0x0001 = 0x{ones_complement_add(0xFFFF, 0x0001):04X}")
print(f"0xFFFF + 0xFFFF = 0x{ones_complement_add(0xFFFF, 0xFFFF):04X}")
 
print("
One's Complement (negation):")
print(f"~0x0001 = 0x{ones_complement(0x0001):04X}")
print(f"~0xFFFF = 0x{ones_complement(0xFFFF):04X}")  # Note: ~(-0) = +0

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

"""
Demonstration that Internet checksum is byte-order independent.
 
We compute the checksum two ways:
1. Treating first byte of each pair as MSB (big-endian / network order)
2. Treating first byte of each pair as LSB (little-endian / host order on x86)
 
Both produce the same checksum value!
"""
 
def internet_checksum(data: bytes, native_order: str = 'big') -> int:
    """
    Compute Internet checksum with specified byte interpretation.
    
    Args:
        data: Bytes to checksum
        native_order: 'big' for big-endian, 'little' for little-endian
    
    Returns:
        16-bit checksum value
    """
    if len(data) % 2 == 1:
        data = data + b'\x00'
    
    total = 0
    
    for i in range(0, len(data), 2):
        if native_order == 'big':
            word = (data[i] << 8) | data[i + 1]
        else:  # little
            word = data[i] | (data[i + 1] << 8)
        total += word
    
    # Fold to 16 bits with end-around carry
    while total > 0xFFFF:
        total = (total & 0xFFFF) + (total >> 16)
    
    # One's complement
    return (~total) & 0xFFFF
 
 
# Test with sample data
test_data = b'\x45\x00\x00\x73\x00\x00\x40\x00\x40\x11\x00\x00\xc0\xa8\x00\x01\xc0\xa8\x00\xc7'
 
checksum_be = internet_checksum(test_data, 'big')
checksum_le = internet_checksum(test_data, 'little')
 
print(f"Test data: {test_data.hex()}")
print(f"Big-endian computation:    0x{checksum_be:04X}")
print(f"Little-endian computation: 0x{checksum_le:04X}")
 
# The checksums are byte-swapped versions of each other
checksum_le_swapped = ((checksum_le & 0xFF) << 8) | (checksum_le >> 8)
print(f"Little-endian (swapped):   0x{checksum_le_swapped:04X}")
print(f"Match after swap: {checksum_be == checksum_le_swapped}")
 
# Key insight: When placed in the header (in network byte order),
# verification works regardless of host byte order!

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

"""
Complete Internet Checksum Implementation (RFC 1071)
 
This implementation prioritizes clarity and correctness.
For production use, consider assembly-level or hardware-offloaded versions.
"""
 
from typing import Union
 
 
def compute_internet_checksum(
    data: Union[bytes, bytearray],
    initial: int = 0
) -> int:
    """
    Compute the Internet checksum per RFC 1071.
    
    Args:
        data: Binary data to checksum
        initial: Initial sum value (for incremental checksums)
    
    Returns:
        16-bit checksum value in network byte order
    """
    # Start with initial value (for combining with pseudo-header)
    total = initial
    
    # Process complete 16-bit words
    length = len(data)
    i = 0
    
    while i < length - 1:
        # Combine two bytes into 16-bit word (network byte order)
        word = (data[i] << 8) | data[i + 1]
        total += word
        i += 2
    
    # Handle trailing odd byte by zero-padding
    if length % 2 == 1:
        total += data[-1] << 8
    
    # Fold 32-bit sum to 16 bits using end-around carry
    # Keep folding until all overflow bits are absorbed
    while total > 0xFFFF:
        total = (total & 0xFFFF) + (total >> 16)
    
    # Return one's complement of the sum
    return (~total) & 0xFFFF
 
 
def verify_internet_checksum(data: bytes) -> bool:
    """
    Verify data with embedded checksum.
    
    For correct data, summing all words (including checksum)
    and complementing should yield zero (or 0xFFFF in one's complement).
    
    Args:
        data: Data with checksum already embedded
    
    Returns:
        True if checksum is valid
    """
    total = 0
    length = len(data)
    i = 0
    
    while i < length - 1:
        word = (data[i] << 8) | data[i + 1]
        total += word
        i += 2
    
    if length % 2 == 1:
        total += data[-1] << 8
    
    while total > 0xFFFF:
        total = (total & 0xFFFF) + (total >> 16)
    
    # If all is correct, total should be 0xFFFF (all ones)
    # The complement of 0xFFFF is 0x0000
    return total == 0xFFFF or total == 0x0000
 
 
# === Demonstration ===
 
# Create a sample IP header (with checksum field set to zero)
ip_header = bytearray([
    0x45, 0x00,  # Version, IHL, DSCP, ECN
    0x00, 0x73,  # Total length: 115 bytes
    0x00, 0x00,  # Identification
    0x40, 0x00,  # Flags, Fragment offset
    0x40, 0x11,  # TTL: 64, Protocol: UDP (17)
    0x00, 0x00,  # Checksum (initially zero)
    0xc0, 0xa8, 0x00, 0x01,  # Source: 192.168.0.1
    0xc0, 0xa8, 0x00, 0xc7   # Dest: 192.168.0.199
])
 
# Compute checksum
checksum = compute_internet_checksum(ip_header)
print(f"Computed checksum: 0x{checksum:04X}")
 
# Insert checksum into header (bytes 10-11, big-endian)
ip_header[10] = (checksum >> 8) & 0xFF
ip_header[11] = checksum & 0xFF
print(f"IP header with checksum: {ip_header.hex()}")
 
# Verify
is_valid = verify_internet_checksum(bytes(ip_header))
print(f"Verification result: {is_valid}")
 
# Test error detection
corrupted = bytearray(ip_header)
corrupted[15] ^= 0x01  # Flip one bit in destination IP
is_valid_corrupted = verify_internet_checksum(bytes(corrupted))
print(f"Corrupted data verification: {is_valid_corrupted}")

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

/*
 * Optimized Internet Checksum (C/Assembly hybrid)
 * 
 * This version uses a 64-bit accumulator and processes
 * 8 bytes at a time for better performance.
 */
 
#include <stdint.h>
#include <stddef.h>
 
uint16_t fast_internet_checksum(const void* data, size_t len) {
    const uint16_t* ptr = (const uint16_t*) data;
    uint64_t sum = 0;  // Wide accumulator for deferred folding
    
    // Process 8 bytes (4 words) at a time
    while (len >= 8) {
        sum += ptr[0];
        sum += ptr[1];
        sum += ptr[2];
        sum += ptr[3];
        ptr += 4;
        len -= 8;
    }
    
    // Process remaining complete words
    while (len >= 2) {
        sum += *ptr++;
        len -= 2;
    }
    
    // Handle trailing byte
    if (len > 0) {
        sum += *(const uint8_t*)ptr << 8;  // Network byte order
    }
    
    // Fold 64-bit sum to 16 bits
    // First fold 64 -> 32
    sum = (sum >> 32) + (sum & 0xFFFFFFFF);
    sum = (sum >> 32) + (sum & 0xFFFFFFFF);
    
    // Then fold 32 -> 16
    sum = (sum >> 16) + (sum & 0xFFFF);
    sum = (sum >> 16) + (sum & 0xFFFF);
    
    return (uint16_t)(~sum);
}
 
/*
 * For maximum performance on x86-64, assembly can use:
 * - ADC (Add with Carry) instruction for end-around carry
 * - SIMD for parallel processing
 * 
 * Linux kernel uses assembly versions optimized per architecture.
 */

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

"""
Incremental Checksum Update (RFC 1624)
 
When modifying a small part of checksummed data,
we can update the checksum without reprocessing all data.
"""
 
def incremental_checksum_update(
    old_checksum: int,
    old_value: int,
    new_value: int
) -> int:
    """
    Update checksum when a single 16-bit word changes.
    
    Based on RFC 1624:
    new_checksum = ~(~old_checksum + ~old_value + new_value)
    
    In one's complement arithmetic with end-around carry.
    
    Args:
        old_checksum: The current checksum value
        old_value: The 16-bit word being replaced
        new_value: The new 16-bit word value
    
    Returns:
        Updated checksum
    """
    # Convert to "sum form" by complementing the checksum
    # (checksum = ~sum, so ~checksum = sum)
    sum_val = (~old_checksum) & 0xFFFF
    
    # Subtract old value (add its complement in one's complement)
    sum_val += (~old_value) & 0xFFFF
    
    # Add new value
    sum_val += new_value
    
    # Fold overflows
    while sum_val > 0xFFFF:
        sum_val = (sum_val & 0xFFFF) + (sum_val >> 16)
    
    # Complement back to checksum form
    return (~sum_val) & 0xFFFF
 
 
def decrement_ttl_update_checksum(
    ip_header: bytearray
) -> None:
    """
    Decrement IP header TTL and update checksum in-place.
    
    This is what routers do for every forwarded packet.
    """
    # TTL is at byte 8, checksum at bytes 10-11
    old_ttl_word = (ip_header[8] << 8) | ip_header[9]  # TTL + Protocol
    old_checksum = (ip_header[10] << 8) | ip_header[11]
    
    # Decrement TTL (upper byte of the word)
    new_ttl = ip_header[8] - 1
    if new_ttl < 0:
        raise ValueError("TTL expired!")
    
    ip_header[8] = new_ttl
    new_ttl_word = (ip_header[8] << 8) | ip_header[9]
    
    # Update checksum incrementally
    new_checksum = incremental_checksum_update(
        old_checksum, old_ttl_word, new_ttl_word
    )
    
    # Write back updated checksum
    ip_header[10] = (new_checksum >> 8) & 0xFF
    ip_header[11] = new_checksum & 0xFF
 
 
# === Demonstration ===
 
# Sample IP header with valid checksum
ip_header = bytearray([
    0x45, 0x00, 0x00, 0x73,  # Version, IHL, DSCP, Total length
    0x00, 0x00, 0x40, 0x00,  # ID, Flags, Fragment
    0x40, 0x11,              # TTL=64, Protocol=UDP
    0xB8, 0x61,              # Checksum (pre-computed)
    0xC0, 0xA8, 0x00, 0x01,  # Source: 192.168.0.1
    0xC0, 0xA8, 0x00, 0xC7   # Dest: 192.168.0.199
])
 
print(f"Before:  TTL={ip_header[8]}, Checksum=0x{ip_header[10]:02X}{ip_header[11]:02X}")
 
# Decrement TTL and update checksum
decrement_ttl_update_checksum(ip_header)
 
print(f"After:   TTL={ip_header[8]}, Checksum=0x{ip_header[10]:02X}{ip_header[11]:02X}")
 
# Verify the updated checksum is correct
from typing import Union
def verify(data: bytes) -> bool:
    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 == 0xFFFF
 
print(f"Checksum valid: {verify(bytes(ip_header))}")