Python SDK

Cowrie binary JSON codec implementation for Python

Python SDK

The Python implementation provides an intuitive API with NumPy integration and optional compression support.

Installation

pip install cowrie

# Or from source:
cd python
pip install -e .

Optional Dependencies

# For compression support
pip install zstandard

# For NumPy integration
pip install numpy

Imports

Gen2 (Full)

from cowrie import gen2

# Or specific imports
from cowrie.gen2 import encode, decode, Value

Gen1 (Lightweight)

from cowrie import gen1

# Or specific imports
from cowrie.gen1 import encode, decode

Basic Usage

Gen2: Encode & Decode

Encode

from cowrie.gen2 import encode, Value

# From Python dict/list
data = encode({
    "name": "Alice",
    "age": 30,
    "score": 98.5,
})

# Or using Value API
val = Value.object([
    ("name", Value.string("Alice")),
    ("age", Value.int64(30)),
    ("score", Value.float64(98.5)),
])
data = encode(val)

Gen1: Lightweight Encoding

from cowrie.gen1 import encode, decode

# Encode any JSON-compatible data
data = encode({
    "name": "Alice",
    "scores": [1.0, 2.0, 3.0],
})

# Decode to Python types
result = decode(data)
print(result["name"])  # "Alice"

Type System

Creating Values (Gen2)

Primitives

from cowrie.gen2 import Value

# Scalars
null_val = Value.null()
bool_val = Value.bool(True)
int_val = Value.int64(42)
uint_val = Value.uint64(100)
float_val = Value.float64(3.14)
str_val = Value.string("hello")
bytes_val = Value.bytes(b"\x01\x02")

ML Types

import numpy as np
from cowrie.gen2 import Value, DType

# Tensor (from NumPy)
array = np.array([[1.0, 2.0], [3.0, 4.0]], dtype=np.float32)
tensor = Value.tensor(
    dtype=DType.FLOAT32,
    shape=(2, 2),
    data=array.tobytes(),
)

# Or use from_numpy helper
tensor = Value.from_numpy(array)

# Image
with open("image.jpg", "rb") as f:
    image_data = f.read()

img = Value.image(
    format="jpeg",
    width=1920,
    height=1080,
    data=image_data,
)

# Audio
audio = Value.audio(
    encoding="pcm_int16",
    sample_rate=44100,
    channels=2,
    data=audio_bytes,
)

Graph Types

from cowrie.gen2 import Value

# Node
node = Value.node(
    id="person_42",
    labels=["Person", "Employee"],
    props={
        "name": Value.string("Alice"),
        "age": Value.int64(30),
    },
)

# Edge
edge = Value.edge(
    src="person_42",
    dst="company_1",
    edge_type="WORKS_AT",
    props={
        "since": Value.int64(2020),
    },
)

# Graph Shard
shard = Value.graph_shard(
    nodes=[node],
    edges=[edge],
    metadata={"version": 1},
)

Type Conversion

Automatic Conversion

from cowrie.gen2 import encode, decode

# Python types are automatically converted
data = encode({
    "name": "Alice",
    "age": 30,
    "scores": [98.5, 87.3, 92.1],
    "active": True,
    "metadata": None,
})

# Decode back to Python types
result = decode(data)
print(type(result))  # <class 'dict'>

Using from_any / to_any

from cowrie.gen2 import Value

# Convert Python value to Cowrie Value
val = Value.from_any({
    "name": "Alice",
    "age": 30,
})

# Convert Cowrie Value back to Python
python_val = val.to_any()
print(python_val)  # {'name': 'Alice', 'age': 30}

JSON Bridge

To JSON

from cowrie.gen2 import to_json, Value

val = Value.object([
    ("name", Value.string("Alice")),
    ("age", Value.int64(30)),
])

json_str = to_json(val)
print(json_str)  # {"name":"Alice","age":30}

Compression

from cowrie.gen2 import encode, decode, Compression

# Encode with compression
data = encode(
    {"name": "Alice", "data": [1.0] * 10000},
    compression=Compression.ZSTD,  # or Compression.GZIP
)

# Decode (automatically detects compression)
result = decode(data)

Streaming

Write to File

from cowrie.gen2 import encode

data = encode({"name": "Alice", "age": 30})

with open("output.cowrie", "wb") as f:
    f.write(data)

Read from File

from cowrie.gen2 import decode

with open("data.cowrie", "rb") as f:
    data = f.read()
    result = decode(data)

Stream Frames

from cowrie.gen2 import write_framed, read_framed
import io

# Write multiple values to a stream
buf = io.BytesIO()
write_framed(buf, {"type": "event1", "data": 42})
write_framed(buf, {"type": "event2", "data": 43})

# Read values back
buf.seek(0)
val1 = read_framed(buf)
val2 = read_framed(buf)

NumPy Integration

import numpy as np
from cowrie.gen2 import encode, decode, Value

# Encode NumPy array as tensor
array = np.random.rand(100, 100).astype(np.float32)
val = Value.from_numpy(array)
data = encode(val)

# Decode back to NumPy
result = decode(data)
if isinstance(result, dict) and result.get("__tensor__"):
    restored = Value.to_numpy(result)
    print(restored.shape)  # (100, 100)

Error Handling

from cowrie.gen2 import decode, SecurityLimitExceeded

try:
    result = decode(data)
except SecurityLimitExceeded as e:
    print(f"Security limit exceeded: {e}")
except ValueError as e:
    print(f"Invalid data: {e}")
except Exception as e:
    print(f"Decode error: {e}")

Security Options

from cowrie.gen2 import decode, DecodeOptions

opts = DecodeOptions(
    max_depth=500,
    max_array_len=1_000_000,
    max_object_len=100_000,
    max_string_len=10_000_000,
    max_bytes_len=100_000_000,
)

result = decode(data, options=opts)

Type Hints

The Python SDK includes full type hints:

from typing import Any, Dict, List
from cowrie.gen2 import encode, decode, Value

def process_data(data: bytes) -> Dict[str, Any]:
    result: Dict[str, Any] = decode(data)
    return result

def create_value(name: str, age: int) -> Value:
    return Value.object([
        ("name", Value.string(name)),
        ("age", Value.int64(age)),
    ])

Advanced Features

Custom Type Handlers

from datetime import datetime
from cowrie.gen2 import Value

def encode_custom(obj):
    """Convert custom Python objects to Cowrie values."""
    if isinstance(obj, datetime):
        return Value.datetime64(obj)
    elif hasattr(obj, "to_dict"):
        return Value.from_any(obj.to_dict())
    else:
        raise TypeError(f"Cannot encode {type(obj)}")

class Person:
    def __init__(self, name, age):
        self.name = name
        self.age = age
    
    def to_dict(self):
        return {"name": self.name, "age": self.age}

person = Person("Alice", 30)
val = encode_custom(person)

Batch Operations

from cowrie.gen2 import encode

# Encode multiple values efficiently
values = [
    {"id": i, "name": f"user_{i}"}
    for i in range(1000)
]

encoded_batch = [encode(v) for v in values]

Performance Tips

1. Use Gen1 for simple data: Faster for JSON-compatible data

2. Reuse buffers: For high-throughput scenarios

   from io import BytesIO
   buf = BytesIO()
   

3. Batch encode: Process multiple items together

4. Enable compression: For large, repetitive data

   encode(data, compression=Compression.ZSTD)
   

5. Use type hints: Helps with IDE autocomplete and catches errors early

Examples

Complete Example

from cowrie.gen2 import encode, decode, Value
import numpy as np

def main():
    # Create a complex value
    data = {
        "user": {
            "name": "Alice",
            "id": 12345,
        },
        "scores": [98.5, 87.3, 92.1],
        "embedding": np.random.rand(384).astype(np.float32),
        "active": True,
    }
    
    # Convert NumPy array to tensor
    val = Value.from_any(data)
    
    # Encode
    encoded = encode(val)
    print(f"Encoded size: {len(encoded)} bytes")
    
    # Decode
    decoded = decode(encoded)
    
    # Process
    print(f"User: {decoded['user']['name']}")
    print(f"Scores: {decoded['scores']}")

if __name__ == "__main__":
    main()

Django/Flask Integration

from flask import Flask, request, Response
from cowrie.gen2 import encode, decode

app = Flask(__name__)

@app.route("/api/data", methods=["POST"])
def receive_data():
    # Decode Cowrie payload
    data = decode(request.data)
    
    # Process data
    result = process(data)
    
    # Encode response
    response_data = encode(result)
    return Response(response_data, mimetype="application/x-cowrie")