Streaming

Cowrie supports two streaming protocols for efficient record-by-record processing: Gen1 (simple length-prefixed) and Gen2 (master stream with metadata and checksums).

Gen1 Stream (Legacy)

Simple length-prefixed Cowrie records, similar to JSON Lines (JSONL).

Wire Format

[u32 length] [cowrie bytes]
[u32 length] [cowrie bytes]
...
[0x00 0x00 0x00 0x00]  // optional end marker

Length: 4-byte little-endian uint32
Payload: Complete Cowrie v2 document (header + data)
End Marker: Zero length signals end of stream (optional)

StreamWriter

import (
    "os"
    "github.com/Neumenon/cowrie/codec"
)

// Create writer
f, _ := os.Create("data.cowrie")
defer f.Close()

writer := codec.NewStreamWriter(f)

// Write records
for _, item := range items {
    err := writer.Write(item)
    if err != nil {
        return err
    }
}

// Flush if needed
writer.Sync()

StreamReader

// Read from file
data, _ := os.ReadFile("data.cowrie")
reader := codec.NewStreamReader(data)

// Read all records
for {
    var item MyStruct
    err := reader.Next(&item)
    if err == io.EOF {
        break
    }
    if err != nil {
        return err
    }
    process(item)
}

// Or read all at once
items, err := codec.ReadAllStream[MyStruct](data)

Use Cases

Log Shipping: Append-only event logs
Message Queues: Kafka-like record streams
Database Exports: Streaming table dumps
API Pagination: Cursor-based result sets

Example: Event Log

type Event struct {
    Timestamp int64
    UserID    string
    Action    string
    Metadata  map[string]any
}

// Write events
writer := codec.NewStreamWriter(logFile)
for _, event := range events {
    writer.Write(event)
}

// Read events
reader := codec.NewStreamReader(logData)
for {
    var event Event
    if err := reader.Next(&event); err == io.EOF {
        break
    }
    processEvent(event)
}

Gen2 Master Stream

Advanced streaming protocol with metadata, compression, checksums, and type routing.

Wire Format

Magic:       'S' 'J' 'S' 'T' (4 bytes)
Version:     0x02 (1 byte)
Flags:       u8 (compression, CRC, deterministic, metadata)
HeaderLen:   u16 LE (24 bytes for v2)
TypeID:      u32 LE (schema fingerprint)
PayloadLen:  u32 LE (compressed length)
RawLen:      u32 LE (original length, 0 if not compressed)
MetaLen:     u32 LE (metadata length, 0 if none)
[Metadata:   cowrie bytes (if MetaLen > 0)]
[Payload:    cowrie bytes or compressed]
[CRC32:      u32 LE (if FlagMasterCRC set)]

Flags

Bit	Flag	Meaning
0	Compressed	Payload is compressed
1-2	Compression type	0=none, 1=gzip, 2=zstd
3	Deterministic	Keys sorted for reproducible encoding
4	CRC	CRC32 checksum follows payload
5	Meta	Metadata present before payload

MasterWriter

import "github.com/Neumenon/cowrie/codec"

// Configure writer
opts := codec.DefaultMasterWriterOptions()
// opts.Deterministic = true  (default)
// opts.Compression = cowrie.CompressionZstd  (default)
// opts.EnableCRC = true  (default)

writer := codec.NewMasterWriter(file, opts)

// Write with metadata
meta := cowrie.Object(
    cowrie.Member{Key: "version", Value: cowrie.Int64(1)},
    cowrie.Member{Key: "source", Value: cowrie.String("api")},
)

err := writer.WriteWithMeta(data, meta)

MasterReader

// Configure reader
opts := codec.DefaultMasterReaderOptions()
// opts.MaxDecompressedSize = 100 * 1024 * 1024  // 100MB default
// opts.AllowLegacy = true  // Auto-detect Gen1 streams

reader := codec.NewMasterReader(data, opts)

// Read frames
for {
    frame, err := reader.Next()
    if err == io.EOF {
        break
    }
    if err != nil {
        return err
    }
    
    // Access frame data
    fmt.Println("TypeID:", frame.TypeID)
    fmt.Println("Compressed:", frame.Header.Compression)
    
    // Access metadata (if present)
    if frame.Meta != nil {
        version := frame.Meta.Get("version").Int64()
    }
    
    // Access payload
    payload := frame.Payload
    processData(payload)
}

Example: Type Routing

// Define type handlers
handlers := map[uint32]func(*cowrie.Value) error{
    12345: handleUserEvent,
    67890: handleSystemEvent,
}

reader := codec.NewMasterReader(data, codec.MasterReaderOptions{
    TypeHandlers: handlers,
})

for {
    frame, err := reader.Next()
    if err == io.EOF {
        break
    }
    
    // Automatic dispatch based on TypeID
    if handler, ok := handlers[frame.TypeID]; ok {
        handler(frame.Payload)
    }
}

Schema Fingerprinting

Type IDs are computed using FNV-1a hash of the value’s type structure:

value := cowrie.Object(
    cowrie.Member{Key: "name", Value: cowrie.String("Alice")},
    cowrie.Member{Key: "age", Value: cowrie.Int64(30)},
)

typeID := cowrie.SchemaFingerprint32(value)  // Stable hash

Use cases:

Type Routing: Dispatch to correct handler
Schema Validation: Detect schema changes
Multiplexing: Route records by type
Schema Registry: Track schema evolution

Compression

Master stream automatically compresses payloads ≥ 256 bytes when enabled:

opts := codec.MasterWriterOptions{
    Compression: cowrie.CompressionZstd,  // or CompressionGzip
}

Compression is applied when:

Payload ≥ 256 bytes
Compressed size < original size

Security: Decompression is size-limited to prevent bombs:

opts := codec.MasterReaderOptions{
    MaxDecompressedSize: 100 * 1024 * 1024,  // 100MB default
}

CRC32 Checksums

Enable data integrity verification:

opts := codec.MasterWriterOptions{
    EnableCRC: true,  // default
}

CRC32 (IEEE polynomial) covers the entire frame from magic to payload. Reader automatically validates and returns ErrMasterCRCMismatch if corrupted.

Deterministic Encoding

Ensures reproducible output for content addressing:

opts := codec.MasterWriterOptions{
    Deterministic: true,  // default
}

Object keys sorted lexicographically
Same input → same output bytes
Essential for caching, deduplication, Merkle trees

Comparison

Feature	Gen1 Stream	Gen2 Master Stream
Overhead	4 bytes	24 bytes + optional metadata
Compression	No	Gzip, Zstd
Checksums	No	CRC32
Metadata	No	Per-frame metadata
Type Routing	No	Schema fingerprinting
Legacy Support	N/A	Auto-detect Gen1

Column Readers (Advanced)

Gen2 supports columnar reading with column hints for efficient partial decoding.

Column Hints Format

HintCount: varint
For each hint:
  Field:     [len:varint][utf8 bytes]
  Type:      u8
  ShapeLen:  varint
  ShapeDims: ShapeLen * varint
  Flags:     u8

Hints appear after header flags, before dictionary (only if FlagHasColumnHints set).

Use Cases

Parquet-like Access: Read specific columns without full decode
Query Push-down: Filter on indexed columns
Partial Hydration: Load only needed fields
Wide Tables: Skip unused columns in large records

Security Limits

opts := cowrie.DecodeOptions{
    MaxHintCount: 10_000,  // Max column hints
}

See Security Limits for details.

Performance Tips

Gen1 Stream

Batch Writes: Group small records to reduce syscalls

var buffer bytes.Buffer
writer := codec.NewStreamWriter(&buffer)
for _, item := range batch {
    writer.Write(item)
}
file.Write(buffer.Bytes())  // Single write

Streaming Read: Process records without loading entire file

// Good: Streaming
reader := codec.NewStreamReader(data)
for {
    var item Item
    if err := reader.Next(&item); err == io.EOF {
        break
    }
    process(item)  // Constant memory
}

// Avoid: Load all
items, _ := codec.ReadAllStream[Item](data)  // Allocates array

Zero-Copy Decoding: Reuse structs

var item Item
reader := codec.NewStreamReader(data)
for {
    err := reader.Next(&item)
    if err == io.EOF {
        break
    }
    process(item)
    // 'item' reused on next iteration
}

Gen2 Master Stream

Enable Compression: 30-70% size reduction for mixed data

opts := codec.MasterWriterOptions{
    Compression: cowrie.CompressionZstd,
}

Deterministic for Dedup: Cache compressed frames by hash

opts := codec.MasterWriterOptions{
    Deterministic: true,
    Compression:   cowrie.CompressionZstd,
}
frame, _ := encodeFrame(data, opts)
hash := sha256.Sum256(frame)
cache[hash] = frame  // Deduplication

Type-Based Routing: Avoid decoding unneeded frames

frame, _ := reader.Next()
if frame.TypeID == expectedTypeID {
    processPayload(frame.Payload)
} else {
    skipPayload(frame)
}

Metadata for Filtering: Skip frames without full decode

for {
    frame, _ := reader.Next()
    
    // Check metadata first
    if frame.Meta != nil {
        partition := frame.Meta.Get("partition").Int64()
        if partition != targetPartition {
            continue  // Skip without decoding payload
        }
    }
    
    // Only decode relevant frames
    process(frame.Payload)
}

Real-World Examples

Event Streaming (Gen1)

// Producer
func streamEvents(events []Event, w io.Writer) error {
    writer := codec.NewStreamWriter(w)
    for _, event := range events {
        if err := writer.Write(event); err != nil {
            return err
        }
    }
    return writer.Sync()
}

// Consumer
func consumeEvents(r io.Reader) error {
    data, _ := io.ReadAll(r)
    reader := codec.NewStreamReader(data)
    
    for {
        var event Event
        err := reader.Next(&event)
        if err == io.EOF {
            break
        }
        if err != nil {
            return err
        }
        handleEvent(event)
    }
    return nil
}

Multi-Tenant Data Stream (Gen2)

// Write with tenant metadata
opts := codec.DefaultMasterWriterOptions()
writer := codec.NewMasterWriter(stream, opts)

for _, record := range records {
    meta := cowrie.Object(
        cowrie.Member{Key: "tenant", Value: cowrie.String(record.TenantID)},
        cowrie.Member{Key: "timestamp", Value: cowrie.Int64(time.Now().Unix())},
    )
    writer.WriteWithMeta(record.Data, meta)
}

// Read filtered by tenant
reader := codec.NewMasterReader(data, codec.DefaultMasterReaderOptions())
targetTenant := "acme_corp"

for {
    frame, err := reader.Next()
    if err == io.EOF {
        break
    }
    
    // Filter by metadata (no payload decode!)
    if frame.Meta != nil {
        tenant := frame.Meta.Get("tenant").String()
        if tenant != targetTenant {
            continue
        }
    }
    
    // Process only relevant tenant's data
    process(frame.Payload)
}

Error Handling

Gen1 Stream Errors

reader := codec.NewStreamReader(data)
for {
    var item Item
    err := reader.Next(&item)
    if err == io.EOF {
        break  // Normal end
    }
    if err != nil {
        // Truncated frame, decode error, etc.
        log.Printf("Stream error at position %d: %v", reader.Position(), err)
        return err
    }
}

Gen2 Master Stream Errors

reader := codec.NewMasterReader(data, opts)
for {
    frame, err := reader.Next()
    switch err {
    case nil:
        // Success
    case io.EOF:
        // Normal end
    case codec.ErrMasterCRCMismatch:
        // Corrupted frame
        log.Printf("CRC mismatch at offset %d", reader.Position())
    case codec.ErrMasterTruncated:
        // Incomplete frame
        log.Printf("Truncated frame at offset %d", reader.Position())
    case cowrie.ErrDecompressedTooLarge:
        // Decompression bomb detected
        log.Printf("Suspicious payload at offset %d", reader.Position())
    default:
        // Other errors
        return err
    }
}

Compression - Compress streamed data
Security Limits - Protect against malicious streams
Graph Types - Stream graph shards
ML Types - Stream tensors and embeddings

Getting Started

Core Concepts

Language SDKs

Advanced Features

CLI Tool

Performance

Gen1 Stream (Legacy)

Wire Format

StreamWriter

StreamReader

Use Cases

Example: Event Log

Gen2 Master Stream

Wire Format

Flags

MasterWriter

MasterReader

Example: Type Routing

Schema Fingerprinting

Compression

CRC32 Checksums

Deterministic Encoding

Comparison

Column Readers (Advanced)

Column Hints Format

Use Cases

Security Limits

Performance Tips

Gen1 Stream

Gen2 Master Stream

Real-World Examples

Event Streaming (Gen1)

Multi-Tenant Data Stream (Gen2)

Error Handling

Gen1 Stream Errors

Gen2 Master Stream Errors

Getting Started

Core Concepts

Language SDKs

Advanced Features

CLI Tool

Performance

Documentation Index

​Gen1 Stream (Legacy)

​Wire Format

​StreamWriter

​StreamReader

​Use Cases

​Example: Event Log

​Gen2 Master Stream

​Wire Format

​Flags

​MasterWriter

​MasterReader

​Example: Type Routing

​Schema Fingerprinting

​Compression

​CRC32 Checksums

​Deterministic Encoding

​Comparison

​Column Readers (Advanced)

​Column Hints Format

​Use Cases

​Security Limits

​Performance Tips

​Gen1 Stream

​Gen2 Master Stream

​Real-World Examples

​Event Streaming (Gen1)

​Multi-Tenant Data Stream (Gen2)

​Error Handling

​Gen1 Stream Errors

​Gen2 Master Stream Errors

​Related Topics

Gen1 Stream (Legacy)

Wire Format

StreamWriter

StreamReader

Use Cases

Example: Event Log

Gen2 Master Stream

Wire Format

Flags

MasterWriter

MasterReader

Example: Type Routing

Schema Fingerprinting

Compression

CRC32 Checksums

Deterministic Encoding

Comparison

Column Readers (Advanced)

Column Hints Format

Use Cases

Security Limits

Performance Tips

Gen1 Stream

Gen2 Master Stream

Real-World Examples

Event Streaming (Gen1)

Multi-Tenant Data Stream (Gen2)

Error Handling

Gen1 Stream Errors

Gen2 Master Stream Errors

Related Topics