The messaging protocol that survives anything.
Cockroach Protocol makes it easy to build apps on mesh networks.
You handle the app logic. We handle:
- ✅ Message signing (Ed25519)
- ✅ Chunking (any size message → 200-byte packets)
- ✅ Sync (lost packets recovered automatically)
- ✅ Transport abstraction (mesh, HTTP, Bluetooth, same code)
from cockroach import Node, Identity
# Create identity (once)
identity = Identity.generate()
# Create node
node = Node(identity, transport="meshtastic", port="/dev/ttyUSB0")
# Send any size message - chunking is automatic
node.send(b":tech: This is my post about mesh networking...")
# Receive complete messages - reassembly is automatic
@node.on_message
def handle(msg):
print(f"From: {msg.author.hex()[:8]}")
print(f"Data: {msg.payload}")pip install cockroach-protocolFor Meshtastic support:
pip install cockroach-protocol[meshtastic]from cockroach import Identity
identity = Identity.generate()
identity.save("~/.cockroach/identity.key")
# Later...
identity = Identity.load("~/.cockroach/identity.key")from cockroach import Node
# For Meshtastic
node = Node(identity, transport="meshtastic", port="/dev/ttyUSB0")
# For HTTP relay
node = Node(identity, transport="http", url="https://relay.example.com")
# For testing
node = Node(identity, transport="memory")# Small message (≤96 bytes) - single packet
node.send(b"Hello mesh!")
# Large message - automatically chunked
node.send(b"x" * 10000) # Protocol handles splitting@node.on_message
def handle(msg):
print(f"Timestamp: {msg.timestamp}")
print(f"Author: {msg.author.hex()}")
print(f"Payload: {msg.payload}")
node.start() # Start receiving# Sync happens automatically when nodes connect
# You can also trigger manual sync:
node.sync()See PROTOCOL.md for the full specification.
┌──────────────────────────────────────────────┐
│ timestamp (8 bytes) - μs since epoch │
│ author (32 bytes) - ed25519 pubkey │
│ payload (≤96 bytes) - your data │
│ signature (64 bytes) - ed25519 sig │
└──────────────────────────────────────────────┘
Max packet: 200 bytes
Messages identified by (timestamp, author) tuple.
Messages >96 bytes automatically split:
- Header packet: announces chunk count + content hash
- Data packets: sequential timestamps, raw data
- Receiver reassembles + verifies hash
- Missing chunks recovered via sync
from cockroach import Node, Identity
identity = Identity.load("~/.cockroach/identity.key")
node = Node(identity, transport="meshtastic", port="/dev/ttyUSB0")
def send_chat(room: str, message: str):
payload = f"_{room}: {message}".encode()
node.send(payload)
@node.on_message
def on_chat(msg):
text = msg.payload.decode()
if text.startswith("_"):
# Parse: _room: message
room, message = text[1:].split(": ", 1)
print(f"[{room}] {message}")
node.start()
send_chat("local", "Hello neighbors!")def send_file(filename: str):
with open(filename, "rb") as f:
data = f.read()
# Protocol automatically chunks files up to 24KB
node.send(data)Identity.generate() -> Identity # Create new identity
Identity.load(path) -> Identity # Load from file
identity.save(path) # Save to file
identity.public_key -> bytes # 32-byte pubkey
identity.sign(data) -> bytes # Sign dataMessage.create(identity, payload, timestamp=None) -> Message
message.timestamp -> int # Microseconds since epoch
message.author -> bytes # 32-byte pubkey
message.payload -> bytes # App data
message.signature -> bytes # 64-byte signature
message.verify() -> bool # Check signature
message.serialize() -> bytes # Wire format
Message.deserialize(data) -> Message # From wire formatNode(identity, transport, **kwargs)
node.send(payload: bytes) # Send message (auto-chunk)
node.on_message(callback) # Register handler
node.start() # Start receiving
node.stop() # Stop
node.sync() # Trigger syncStore(path: str) # SQLite store
store.append(message) -> bool # Store if not duplicate
store.get(timestamp, author) -> Msg # Get by key
store.after(timestamp) -> [Msg] # For sync
store.latest() -> int # Newest timestampnode = Node(identity,
transport="meshtastic",
port="/dev/ttyUSB0", # or "tcp:192.168.1.100"
channel=1, # Meshtastic channel
)node = Node(identity,
transport="http",
url="https://relay.example.com",
)node = Node(identity, transport="memory")Cockroach Protocol delivers raw bytes. Define your own payload format:
:flair: Post content → Post with flair tag
_room: Chat message → Chat message to room
@timestamp Reply text → Reply to message
~name~bio → Profile update
[type:1][...type-specific data...]
0x01 = post
0x02 = comment (includes 8-byte timestamp reference)
0x03 = chat (includes room name)
0x04 = profile
import msgpack
payload = msgpack.packb({"type": "post", "body": "Hello!"})Your app, your format.
Q: Max message size? A: 24KB (255 chunks × 96 bytes)
Q: What if packets are lost? A: Sync recovers them. No action needed.
Q: Can I use on regular internet? A: Yes! HTTP transport works. Same protocol everywhere.
Q: How do I identify users? A: Public key = identity. First 8 chars of hex is often enough for display.
Q: Is it encrypted? A: Signed, not encrypted. Add encryption at app layer if needed.
🚧 v0.1 - In Development
- Protocol specification
- Python implementation
- Meshtastic transport
- HTTP transport
- CLI tool
- Example apps
Contributions welcome! See CONTRIBUTING.md.
MIT License - use it for anything.
- Protocol Specification
- LaTerminal - Social network built on Cockroach
- Meshtastic - LoRa mesh firmware
"The protocol that survives nuclear war" 🪳
Built by CM64.studio