Binary Data and Network Protocols (struct)

Binary Data and Network Protocols (struct)

After reading this chapter, you will master low-level binary serialization and deserialization in Python using the struct module, pack primitive Python types into dense C-compatible byte arrays, control byte endianness and memory alignment, unpack fixed-width binary protocols and telemetry streams, perform zero-copy buffer modifications using struct.pack_into and struct.unpack_from, and engineer resilient wire framing for high-throughput network applications.

Mental model

Higher-level formats like JSON, XML, and YAML serialize data as UTF-8 human-readable text. While flexible, textual formats incur substantial overhead: a 64-bit integer like 18446744073709551615 takes 20 ASCII bytes plus delimiter bytes, parsing requires string scanning and memory allocation, and floating-point conversions lose IEEE precision.

In contrast, the struct module translates directly between Python values and raw binary representations matching C structs:

Textual (JSON):
  {"id": 42, "temp": 98.6}  ──▶ 25+ ASCII bytes, variable width, slow parsing

Binary (struct):
  struct.pack("!Id", 42, 98.6)
  ┌───────────────────────┬──────────────────────────────────┐
  │ 4 bytes: uint32 (42)  │ 8 bytes: IEEE double-float (98.6)│
  └───────────────────────┴──────────────────────────────────┘
  ──▶ Exactly 12 bytes! Constant-time O(1) packing/unpacking, zero allocations.

Endianness & Byte Order Modifiers

When exchanging binary packets across networks or persisting files to disk, you must explicitly specify byte order:

Modifier	Meaning	Size & Alignment	Use Case
@	Native byte order	Native C struct alignment & padding	Internal C interop (default)
=	Native byte order	Standard (packed, no padding)	System-specific serialization
<	Little-endian	Standard (packed, no padding)	x86/ARM disk formats, NVMe
>	Big-endian	Standard (packed, no padding)	Network protocols
!	Network byte order	Standard (packed, big-endian)	RFC standard for all TCP/UDP headers

---

Minimal example

Save as struct_overview.py:

# struct_overview.py
import struct

def main() -> None:
    # Format String:
    #   '!' = Network byte order (big-endian)
    #   'H' = unsigned 16-bit short (2 bytes)
    #   'I' = unsigned 32-bit int   (4 bytes)
    #   'f' = single-precision float (4 bytes)
    #   '6s' = 6-byte string        (6 bytes)
    FMT = "!HIf6s"
    
    expected_size = struct.calcsize(FMT)
    print(f"Calculated struct size: {expected_size} bytes\n")

    # 1. Pack Python values into raw binary bytes
    packet_type = 101
    sequence_id = 98452
    latency_ms = 4.25
    device_tag = b"NODE01"

    binary_payload = struct.pack(FMT, packet_type, sequence_id, latency_ms, device_tag)
    print(f"Packed bytes ({len(binary_payload)} bytes): {binary_payload.hex()}")

    # 2. Unpack raw bytes back into Python primitives
    unpacked_type, unpacked_seq, unpacked_lat, unpacked_tag = struct.unpack(FMT, binary_payload)
    print("\nUnpacked fields:")
    print(f"  Packet Type : {unpacked_type}")
    print(f"  Sequence ID : {unpacked_seq}")
    print(f"  Latency (ms): {unpacked_lat:.2f}")
    print(f"  Device Tag  : {unpacked_tag.decode()}")

if __name__ == "__main__":
    main()

Run via uv run python struct_overview.py:

Calculated struct size: 16 bytes

Packed bytes (16 bytes): 006500018094408800004e4f44453031

Unpacked fields:
  Packet Type : 101
  Sequence ID : 98452
  Latency (ms): 4.25
  Device Tag  : NODE01

---

Worked examples

Case 1: Custom Binary RPC Wire Framing with Checksum Validation

In custom network protocols, every message starts with a fixed-size header specifying magic bytes, sequence IDs, flags, and the length of the following dynamic payload:

# binary_rpc_framer.py
import struct
import zlib
from typing import NamedTuple

class Packet(NamedTuple):
    sequence: int
    flags: int
    payload: bytes

# Header: 4-byte Magic, 4-byte Seq, 4-byte Payload Length, 2-byte Flags, 4-byte CRC32
# Total header length: 18 bytes
HEADER_FORMAT = "!4sIIHI"
HEADER_SIZE = struct.calcsize(HEADER_FORMAT)
MAGIC_BYTES = b"RPC1"

def encode_packet(sequence: int, flags: int, payload: bytes) -> bytes:
    """Encodes a payload into a binary wire packet with CRC32 integrity check."""
    crc = zlib.crc32(payload)
    header = struct.pack(HEADER_FORMAT, MAGIC_BYTES, sequence, len(payload), flags, crc)
    return header + payload

def decode_packet(raw_bytes: bytes) -> Packet:
    """Decodes and validates a binary wire packet."""
    if len(raw_bytes) < HEADER_SIZE:
        raise ValueError(f"Incomplete packet: expected at least {HEADER_SIZE} bytes")

    magic, sequence, length, flags, expected_crc = struct.unpack(
        HEADER_FORMAT, raw_bytes[:HEADER_SIZE]
    )

    if magic != MAGIC_BYTES:
        raise ValueError(f"Protocol violation: invalid magic bytes {magic!r}")

    payload = raw_bytes[HEADER_SIZE : HEADER_SIZE + length]
    if len(payload) != length:
        raise ValueError(f"Truncated payload: expected {length} bytes, got {len(payload)}")

    actual_crc = zlib.crc32(payload)
    if actual_crc != expected_crc:
        raise ValueError(f"Corrupted packet! Checksum mismatch: 0x{actual_crc:08X} != 0x{expected_crc:08X}")

    return Packet(sequence=sequence, flags=flags, payload=payload)

def main() -> None:
    # 1. Encode message
    msg = b'{"command": "ACQUIRE_LOCK", "resource": "db_writer"}'
    wire_data = encode_packet(sequence=4201, flags=0x0001, payload=msg)
    print(f"Encoded wire packet size: {len(wire_data)} bytes (Payload: {len(msg)} bytes)")

    # 2. Decode valid packet
    pkt = decode_packet(wire_data)
    print(f"Decoded successfully: seq={pkt.sequence}, flags=0x{pkt.flags:04X}, body={pkt.payload.decode()}")

    # 3. Simulate bit-flip corruption in transmission
    corrupted_data = bytearray(wire_data)
    corrupted_data[-1] ^= 0xFF  # Flip bits in last payload byte
    try:
        decode_packet(bytes(corrupted_data))
    except ValueError as err:
        print(f"\nCaught transmission corruption:\n  {err}")

if __name__ == "__main__":
    main()

Run:

uv run python binary_rpc_framer.py

Output:

Encoded wire packet size: 70 bytes (Payload: 52 bytes)
Decoded successfully: seq=4201, flags=0x0001, body={"command": "ACQUIRE_LOCK", "resource": "db_writer"}

Caught transmission corruption:
  Corrupted packet! Checksum mismatch: 0x... != 0x...

---

Case 2: High-Throughput Stream Parsing with struct.iter_unpack

When parsing continuous streams of sensor records or financial ticks, slicing strings creates millions of ephemeral Python objects that overwhelm the garbage collector. struct.iter_unpack iterates directly through a continuous memory buffer:

# telemetry_stream_parser.py
import struct
import time

# Record Format:
#   '!' = Big-endian
#   'I' = uint32 timestamp (seconds)
#   'H' = uint16 sensor_id
#   'f' = float32 reading
RECORD_FMT = "!IHf"
RECORD_SIZE = struct.calcsize(RECORD_FMT)

def generate_telemetry_batch(count: int = 4) -> bytes:
    """Simulates a raw binary sensor telemetry stream received over UDP."""
    base_time = 1773000000
    buffer = bytearray()
    for i in range(count):
        buffer.extend(struct.pack(RECORD_FMT, base_time + i, 100 + i, 23.5 + (i * 0.4)))
    return bytes(buffer)

def main() -> None:
    raw_telemetry = generate_telemetry_batch(4)
    print(f"Stream size: {len(raw_telemetry)} bytes ({len(raw_telemetry) // RECORD_SIZE} records)\n")

    print("Streaming records via iter_unpack (zero intermediate slices):")
    for ts, sensor_id, temp in struct.iter_unpack(RECORD_FMT, raw_telemetry):
        print(f"  Timestamp: {ts} | Sensor: {sensor_id} | Temperature: {temp:5.2f} C")

if __name__ == "__main__":
    main()

Run:

uv run python telemetry_stream_parser.py

Output:

Stream size: 40 bytes (4 records)

Streaming records via iter_unpack (zero intermediate slices):
  Timestamp: 1773000000 | Sensor: 100 | Temperature: 23.50 C
  Timestamp: 1773000001 | Sensor: 101 | Temperature: 23.90 C
  Timestamp: 1773000002 | Sensor: 102 | Temperature: 24.30 C
  Timestamp: 1773000003 | Sensor: 103 | Temperature: 24.70 C

---

Case 3: Zero-Copy In-Place Buffer Updates with pack_into and unpack_from

When writing packet routers or video transcoders, allocating new bytes objects on every modification degrades throughput. struct.pack_into and struct.unpack_from modify pre-allocated mutable buffers (bytearray or memoryview) directly:

# zero_copy_packet_rewriter.py
import struct

# Simulated IPv4 Header Fragment:
# Offset 0: Version & IHL (1B)
# Offset 8: TTL (1B)
# Offset 12: Source IP (4B)
# Offset 16: Destination IP (4B)

def main() -> None:
    # Allocate a mutable 20-byte packet buffer
    packet_buf = bytearray(20)

    # 1. Write headers directly at specific offsets
    struct.pack_into("!B", packet_buf, 0, 0x45)         # IPv4, 5 words
    struct.pack_into("!B", packet_buf, 8, 64)           # TTL = 64
    struct.pack_into("!4B", packet_buf, 12, 192, 168, 1, 10)  # Src: 192.168.1.10
    struct.pack_into("!4B", packet_buf, 16, 10, 0, 0, 1)      # Dst: 10.0.0.1

    print(f"Initial Packet Hex: {packet_buf.hex()}")

    # 2. Read fields directly from offset without slicing
    ttl = struct.unpack_from("!B", packet_buf, 8)[0]
    src_ip = ".".join(str(b) for b in struct.unpack_from("!4B", packet_buf, 12))
    print(f"Read before NAT: TTL={ttl}, Src={src_ip}")

    # 3. Apply NAT translation: rewrite Source IP and decrement TTL in place!
    struct.pack_into("!B", packet_buf, 8, ttl - 1)
    struct.pack_into("!4B", packet_buf, 12, 172, 16, 0, 99)

    # 4. Verify in-place mutation
    new_ttl = struct.unpack_from("!B", packet_buf, 8)[0]
    new_src_ip = ".".join(str(b) for b in struct.unpack_from("!4B", packet_buf, 12))
    print(f"\nRead after NAT : TTL={new_ttl}, Src={new_src_ip}")
    print(f"Mutated Packet Hex: {packet_buf.hex()}")

if __name__ == "__main__":
    main()

Run:

uv run python zero_copy_packet_rewriter.py

Output:

Initial Packet Hex: 450000000000000040000000c0a8010a0a000001
Read before NAT: TTL=64, Src=192.168.1.10

Read after NAT : TTL=63, Src=172.16.0.99
Mutated Packet Hex: 45000000000000003f000000ac1000630a000001

---

Pitfalls

Pitfall 1: Relying on Native @ Alignment for Network or Disk Formats

By default, omitting the byte order character or using @ enables native C-compiler struct padding. On 64-bit x86/ARM systems, fields are padded with invisible filler bytes to align with 4- or 8-byte boundaries. If sent over the network, different architectures misinterpret the offsets!

# THE TRAP:
import struct

# Native format without explicit endianness prefix:
native_data = struct.pack("ci", b"A", 42)
print("Native size with C padding:", len(native_data))

# THE FIX:
# Always use '!' or '<' to enforce packed, non-padded representation:
standard_data = struct.pack("!ci", b"A", 42)
print("Standard packed size     :", len(standard_data))

Output:

Native size with C padding: 8
Standard packed size     : 5

---

Pitfall 2: String Length Mismatches in s Format Strings

In struct format strings, 8s specifies a fixed-length 8-byte buffer. If the Python byte string is shorter than 8 bytes, struct.pack automatically pads it with null bytes (\x00). If longer, it silently truncates the string, discarding trailing data without error!

# THE TRAP:
import struct

long_tag = b"ENTERPRISE_SERVER"
truncated = struct.pack("!8s", long_tag)
print("Truncated bytes:", truncated)  # Silent data loss: only b'ENTERPRI'!

Output:

Truncated bytes: b'ENTERPRI'

The Fix: Always validate len(byte_string) <= max_length before calling struct.pack, or use variable-length prefixes like length-prefixed strings (!H<len>s).

---

Exercises

1. Write an encoder and decoder for a binary key-value protocol where keys are length-prefixed strings (uint8 length followed by bytes) and values are length-prefixed blobs (uint32 length followed by bytes).
2. Use struct.unpack_from() to parse the header of a WAV audio file or BMP image, extracting width, height, and bits-per-pixel metadata.
3. Benchmark the throughput of serializing 1,000,000 numeric tuples (int, float, int) using json.dumps() versus struct.pack("!IfI", ...).
4. Implement a binary ring-buffer logger using struct.pack_into and a fixed-size bytearray(65536).
5. Build a DNS response parser that unpacks DNS header flags (QR, Opcode, AA, TC, RD, RA, RCODE) using bitwise masks and struct.unpack("!HHHHHH", header).

---

Further reading

• Python Documentation: struct — Interpret bytes as packed binary data.
• David Beazley: Python Cookbook (Recipe 6.11: Reading and Writing Binary Arrays of Structures).
• RFC 791: Internet Protocol DARPA Internet Program Protocol Specification (Binary Packet Structure).
• W. Richard Stevens: TCP/IP Illustrated, Volume 1 (Network Byte Order and Framing).