Skip to content

Implementing a Length-Prefixed Framing Protocol

Delimiter framing stops working the moment a payload can contain the delimiter — which is immediately, for anything binary. The standard answer is a length prefix: every frame begins with a fixed-size header stating how many bytes follow, so the reader knows exactly how much to consume and payload bytes are never inspected for structure. It is the framing used by Kafka, Redis' bulk strings, gRPC's message framing, and most in-house binary protocols, and it is simple enough to implement in twenty lines.

The twenty lines have to be the right ones. A length taken from the wire and passed straight to a read call is a remote memory-exhaustion primitive; a header and payload written separately can interleave with another task's frame and corrupt the stream; and a reader that uses read(n) instead of readexactly(n) will silently truncate frames the first time TCP splits a packet. This guide builds a codec that gets all three right, plus versioning, and a fuzz harness that proves it.

Prerequisites

The eight-byte header, field by field A left-to-right frame layout of the header fields and payload. The eight-byte header, field by field version 1 byte msg_type 1 byte flags 2 bytes length 4 bytes payload length bytes Fixed header first, so the reader always knows how much to read next.

1. Choose the header layout

Fix the header once: size, byte order, and any fields beyond the length. Changing it later is a protocol break.

import struct

# ! = network byte order (big-endian), B = uint8, H = uint16, I = uint32
HEADER = struct.Struct("!BBHI")        # version, msg_type, flags, payload_length
HEADER_SIZE = HEADER.size              # 8 bytes
MAX_PAYLOAD = 4 * 1024 * 1024          # hard ceiling, enforced on read

Use network byte order so the protocol is portable across architectures, and give yourself a version byte on day one — retrofitting versioning into a protocol already in production is far more expensive than the byte costs. Size the length field for the maximum you will ever allow, not the maximum the type can express: a uint32 can describe 4 GiB, and your MAX_PAYLOAD is what stops that from mattering. Verify: HEADER.size matches the constant, and pack/unpack round-trips a known tuple exactly.

2. Read with readexactly, never read

read(n) returns up to n bytes as soon as any are available. readexactly(n) loops until it has all of them, or raises.

import asyncio


async def read_frame(reader: asyncio.StreamReader) -> tuple[int, int, bytes]:
    header = await reader.readexactly(HEADER_SIZE)      # blocks until all 8 bytes
    version, msg_type, flags, length = HEADER.unpack(header)
    if version != 1:
        raise ValueError(f"unsupported protocol version {version}")
    if length > MAX_PAYLOAD:                            # validate BEFORE allocating
        raise ValueError(f"frame length {length} exceeds {MAX_PAYLOAD}")
    payload = await reader.readexactly(length) if length else b""
    return msg_type, flags, payload

The validation order is the security-relevant part: the check happens before any allocation, so a peer declaring a 4 GiB frame gets an exception rather than an out-of-memory kill. readexactly() raises IncompleteReadError when the stream ends early, with exc.partial holding what did arrive — useful for logging, never for parsing. Verify: feed the reader a frame delivered one byte per TCP segment and confirm it is reassembled correctly.

3. Write the frame atomically

Header and payload must reach the transport as one call, or two tasks sharing a writer can interleave.

async def write_frame(writer: asyncio.StreamWriter, msg_type: int,
                      payload: bytes, flags: int = 0) -> None:
    if len(payload) > MAX_PAYLOAD:
        raise ValueError(f"payload of {len(payload)} exceeds {MAX_PAYLOAD}")
    writer.write(HEADER.pack(1, msg_type, flags, len(payload)) + payload)
    await writer.drain()                                # backpressure

Concatenating before the single write() costs one copy and removes an entire class of corruption bug: with two write() calls, another task can slip its own frame between your header and your payload, and the peer will interpret your payload as that frame's header. For very large payloads where the copy matters, guard the writer with a per-connection asyncio.Lock instead. Verify: run two tasks writing different frame types on one writer for a few seconds and confirm the peer decodes every frame with a valid version byte.

Why the header and payload go in one write Two columns contrasting split writes with a single concatenated write. Why the header and payload go in one write two writes corruptible Task A writes its header Task B writes a whole frame Task A writes its payload Peer decodes garbage one write atomic Header and payload concatenated Transport receives them together No interleaving window One extra copy, worth it Concurrent writers on one connection make this a correctness issue, not style.

4. Wrap it in a codec object

Give the protocol one home so both ends stay in sync, and so timeouts and metrics have somewhere to live.

import asyncio


class FrameCodec:
    def __init__(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter,
                 *, read_timeout: float = 30.0) -> None:
        self.reader, self.writer = reader, writer
        self.read_timeout = read_timeout
        self.frames_in = self.frames_out = self.bytes_in = 0

    async def recv(self) -> tuple[int, int, bytes]:
        async with asyncio.timeout(self.read_timeout):
            msg_type, flags, payload = await read_frame(self.reader)
        self.frames_in += 1
        self.bytes_in += HEADER_SIZE + len(payload)
        return msg_type, flags, payload

    async def send(self, msg_type: int, payload: bytes, flags: int = 0) -> None:
        await write_frame(self.writer, msg_type, payload, flags)
        self.frames_out += 1

    async def aclose(self) -> None:
        self.writer.close()
        try:
            async with asyncio.timeout(5):
                await self.writer.wait_closed()
        except (TimeoutError, ConnectionResetError):
            self.writer.transport.abort()

The read timeout belongs here rather than in each call site: a frame that starts arriving and then stalls mid-payload would otherwise park the task forever holding a partially filled buffer. Verify: open a connection, send only a header with a non-zero length, and confirm the server times out after read_timeout and closes.

5. Serve it with bounded resources

The server loop pairs the codec with the limits that make it safe under load.

import asyncio
import logging

log = logging.getLogger("frames")
CONNECTIONS = asyncio.Semaphore(2_000)          # fd budget


async def handle(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
    peer = writer.get_extra_info("peername")
    codec = FrameCodec(reader, writer)
    async with CONNECTIONS:
        try:
            while True:
                try:
                    msg_type, flags, payload = await codec.recv()
                except (asyncio.IncompleteReadError, ConnectionResetError):
                    break                                    # peer closed
                except TimeoutError:
                    log.info("idle/stalled peer %s closed", peer)
                    break
                except ValueError as exc:                    # protocol violation
                    log.warning("bad frame from %s: %s", peer, exc)
                    await codec.send(0xFF, str(exc).encode()[:256])
                    break
                await codec.send(msg_type, await dispatch(msg_type, payload))
        finally:
            await codec.aclose()
            log.info("%s: in=%d out=%d bytes=%d", peer,
                     codec.frames_in, codec.frames_out, codec.bytes_in)


async def main() -> None:
    server = await asyncio.start_server(handle, "0.0.0.0", 9100,
                                        limit=HEADER_SIZE + MAX_PAYLOAD)
    async with server:
        await server.serve_forever()

Set the stream limit to at least one full frame; readexactly() needs to buffer the whole payload, and a limit below MAX_PAYLOAD turns legitimate large frames into errors. A protocol violation gets one error frame and a close — never a retry loop, because a desynchronised stream cannot be resynchronised without a resync marker. Verify: with 500 concurrent clients, the live-connection gauge stays at or below the semaphore's permit count and file descriptors stay flat.

6. Fuzz the parser

Framing bugs surface at boundaries, so generate the boundaries deliberately.

import asyncio
import random


async def fuzz_once(rng: random.Random) -> None:
    reader = asyncio.StreamReader()
    frames = [(rng.randrange(256), bytes(rng.randrange(256) for _ in range(rng.randrange(0, 5000))))
              for _ in range(20)]
    blob = b"".join(HEADER.pack(1, t, 0, len(p)) + p for t, p in frames)

    async def feed() -> None:
        i = 0
        while i < len(blob):                       # random split points
            chunk = rng.randrange(1, 97)
            reader.feed_data(blob[i:i + chunk])
            i += chunk
            await asyncio.sleep(0)
        reader.feed_eof()

    asyncio.get_running_loop().create_task(feed())
    for expected_type, expected_payload in frames:
        msg_type, _flags, payload = await read_frame(reader)
        assert (msg_type, payload) == (expected_type, expected_payload)


async def main() -> None:
    rng = random.Random(1234)                      # deterministic: reproducible
    for _ in range(200):
        await fuzz_once(rng)
    print("200 fuzz rounds passed")

StreamReader.feed_data() lets you exercise the parser without sockets, and random chunk sizes reproduce exactly the packet-boundary conditions that break naive implementations. Keep the seed fixed so a failure is reproducible. Verify: the harness passes 200 rounds, then deliberately swap readexactly for read and confirm it fails — proof the test is actually exercising the boundary case.

What the fuzz harness exercises Four bars showing the distribution of fuzz input shapes. What the fuzz harness exercises 1-byte segments worst-case split random 1–96 byte chunks realistic mix whole blob in one feed worst-case merge truncated final frame IncompleteReadError path Bar length shows the share of fuzz rounds spent in each shape across 200 seeded runs. Random split points with a fixed seed reproduce every boundary condition.

Verification

  • Boundary-independent. Frames delivered one byte at a time, and many frames in one segment, both decode identically.
  • Hostile lengths are rejected cheaply. A frame declaring MAX_PAYLOAD + 1 bytes produces an error and a close with no allocation spike.
  • No interleaving under concurrency. Two writers on one connection produce a stream where every frame decodes with the expected version byte.

Pitfalls and edge cases

  • Trusting the declared length. Always compare against a hard maximum before reading. This is the single most exploited bug in hand-written binary protocols.
  • Using read(n). It returns short reads whenever TCP splits the payload, so frames silently truncate under load and only under load.
  • Separate header and payload writes. Two tasks sharing a writer will interleave; concatenate, or hold a per-connection lock.
  • Native byte order. struct.Struct("I") uses host order and padding; always use an explicit ! prefix.
  • Stream limit below MAX_PAYLOAD. readexactly() needs to buffer a whole frame; a smaller limit rejects valid traffic.
  • No version field. Adding one after deployment requires a flag day; one byte now avoids it.
  • Attempting to resynchronise after a bad frame. Without an explicit resync marker in the protocol, the stream position is unknown — close the connection.
  • Zero-length payloads. readexactly(0) returns immediately, but only if you skip the call or handle the case; make sure the codec allows empty frames if the protocol does.

Frequently Asked Questions

Why use readexactly instead of read in a framing protocol?

read(n) returns as soon as any bytes are available and may return fewer than n, so a payload split across TCP segments is silently truncated — a bug that appears only under load or across a real network. readexactly(n) loops until exactly n bytes are collected and raises IncompleteReadError if the stream ends first, which is what frame decoding requires.

How large should the length prefix be?

Four bytes (uint32, network byte order) is the usual choice and describes any payload you should be willing to accept; two bytes caps you at 65,535 which many protocols outgrow. The field width is not the safety mechanism — an explicit maximum payload constant, validated before you read, is what protects memory.

How do I stop a malicious length prefix from exhausting memory?

Validate the decoded length against a hard maximum before calling readexactly, and reject the frame with an error and a connection close if it exceeds the ceiling. Also set the stream limit to roughly one maximum frame so the reader cannot buffer more than that, and cap concurrent connections so the worst case is bounded by connections times frame size.

Should the header and payload be written in one call?

Yes. Concatenate them into a single writer.write() so two tasks sharing the connection cannot interleave a header with another frame's payload. If the copy is too expensive for very large payloads, serialise access with a per-connection asyncio.Lock around the two writes instead.