A deterministic, in-memory TCP model for testing network protocols.

tcpsim wraps the excellent smoltcp TCP/IP stack and exposes a simple Python API suitable for protocol testing, simulation, and network emulation.

The goal is to provide realistic TCP behavior—including congestion control, retransmissions, acknowledgements, flow control, and connection state transitions—without requiring:

• Root privileges
• TUN/TAP devices
• Network namespaces
• Ethernet frames
• Real sockets

Instead, applications exchange raw IP packets directly through a Sans-I/O API.

When testing network protocols such as multipath transports, FEC systems, packet schedulers, or custom reliability layers, it is often useful to observe how TCP behaves under:

• Packet loss
• Delay
• Reordering
• Duplication
• Bandwidth restrictions
• Path switching

Using the host operating system's TCP stack makes these tests difficult to control and reproduce.

tcpsim provides a deterministic TCP endpoint that can be embedded directly into a test suite.

• Deterministic execution
• Explicit time control
• No background threads
• No system networking required
• Real TCP state machine
• Real retransmission behavior
• Real flow control
• Python package
• Built on top of smoltcp

from tcpsim import TcpEndpoint

client = TcpEndpoint(
    local_ip="10.0.0.1",
    local_port=10000,
)

server = TcpEndpoint(
    local_ip="10.0.0.2",
    local_port=10001,
)

server.listen(port=5000)

client.connect(
    remote_ip="10.0.0.2",
    remote_port=5000,
)

for now_us in range(0, 1_000_000, 1_000):
    client.step(now_us)
    server.step(now_us)
    for packet in client.poll():
        server.consume(packet)
    for packet in server.poll():
        client.consume(packet)
    client.step(now_us)
    server.step(now_us)

client.write(b"hello world")

for now_us in range(1_000_000, 2_000_000, 1_000):
    client.step(now_us)
    server.step(now_us)
    for packet in client.poll():
        server.consume(packet)
    for packet in server.poll():
        client.consume(packet)
    client.step(now_us)
    server.step(now_us)

assert server.read() == b"hello world"

The library follows a Sans-I/O design.

The user is responsible for:

• Providing time
• Delivering packets
• Scheduling events
• Simulating the network

The library is responsible for:

• TCP state machine
• ACK generation
• Retransmissions
• Congestion control
• Flow control
• Connection management

Application
    │
    ▼
TcpEndpoint
    │
    ▼
IP Packet
    │
    ▼
Test Network
(loss / delay / reorder)
    │
    ▼
IP Packet
    │
    ▼
TcpEndpoint
    │
    ▼
Application

endpoint = TcpEndpoint(
    local_ip="10.0.0.1",
    local_port=10000,
)

endpoint.step(now_us=0)

Runs the smoltcp event loop at the given simulated time. Mutating calls do not step automatically; call step explicitly after enqueueing work.

delay_us = endpoint.poll_delay(now_us=0)

Returns None when no timer is pending, or the number of microseconds until the next stack event (retransmission, keepalive, etc.). Use this to avoid polling on a fixed time grid.

endpoint.listen(port=5000)

endpoint.connect(
    remote_ip="10.0.0.2",
    remote_port=5000,
)

accepted = endpoint.write(b"hello")

Returns the number of bytes accepted into the send buffer. When the send buffer is full or the peer receive window is closed, write returns 0. Retry after exchanging packets (poll / consume) and/or reading on the peer.

data = endpoint.read()

endpoint.close()

Initiates a graceful shutdown by sending FIN. The peer must also call close() to complete the four-way handshake.

endpoint.abort()

Immediately resets the connection (RST).

packets = endpoint.poll()

for packet in packets:
    network.send(packet)

Call step before poll to process the stack and enqueue outbound packets. poll only drains the transmit queue.

endpoint.consume(packet)

After delivering one or more packets, call step to process ingress.

state = endpoint.state()
stats = endpoint.stats()

stats() returns a dictionary with queue depths, capacities, congestion control algorithm, and local_endpoint / remote_endpoint strings (e.g. "10.0.0.1:10000").

Possible states:

CLOSED
LISTEN
SYN_SENT
SYN_RECEIVED
ESTABLISHED
FIN_WAIT_1
FIN_WAIT_2
CLOSE_WAIT
CLOSING
LAST_ACK
TIME_WAIT

Simulated time is provided only to step and poll_delay.

The library never reads the system clock.

endpoint.step(now_us=0)
endpoint.consume(packet)
endpoint.write(data)

This makes tests:

• Deterministic
• Fast
• Reproducible

python3 waf configure

python3 waf

python3 waf --run_tests

python3 waf prepare_release

This project intentionally does not provide:

• Raw sockets
• Ethernet support
• IPv6
• Routing
• DNS
• TLS
• AsyncIO integration
• Production networking

The focus is deterministic protocol testing.

MIT