Stream - peaq

The stream module ships in @peaqos/peaq-os-sdk and peaq-os-sdk, and its exports are flagged @experimental. It is the cryptographic core of Stream: it signs the data a machine produces, chunks and encrypts it, and re-wraps chunk keys so a buyer can decrypt only what they bought. You hand it a payload and it returns signed, verifiable, encrypted artifacts. Topic subscription and transport on a robot are handled by the ROS 2 Edge Agent, which produces its own peaqos-stream-envelope@v1 packages (Ed25519 signatures, X25519 field encryption) from a Data Event Map. Those envelopes are not the EIP-191 / AES-256-GCM data packages this module produces, so verify each with its own pipeline. Selling and shipping the encrypted chunks (purchases, payment rails, and S3 / P2P delivery) is the distribution surface.

Module location

SDK	Import
JavaScript / TypeScript	`@peaqos/peaq-os-sdk` → `src/stream/`
Python	`peaq_os_sdk` → `peaq_os_sdk.stream`

Functions

Function	Purpose
`buildSignedDataPackage`	Apply field rules to a topic payload, bundle it with the machine’s identity (DID, timestamp, schema version, sequence number), and sign it EIP-191. Async.
`verifyDataPackage`	Recover the EIP-191 signer and check it against the address in the package DID (`did:peaq:0x<address>`). No network needed. Async.
`buildChunkChain`	Split data into bounded chunks, encrypt each with a fresh per-chunk key wrapped to the owner, operator, and machine, and link each chunk to the previous one. One caller-supplied Ed25519 key signs the whole chain; only the encryption key is per-chunk.
`verifyChunkChain`	Confirm the chain is intact: no gaps, reordering, or edits.
`createBuyerAccessEntry`	Unwrap a chunk’s key with the owner’s private key and re-wrap it to a buyer’s public key. This is the access-grant primitive.
`buildBuyerAccessFiles`	Re-wrap a chunk chain’s keys for a buyer and return distributable access-file documents (split by size, 500 KiB default). Takes the chunks, not pre-built entries.
`decryptChunk`	Decrypt a chunk with a recipient’s private key and wrapped-key entry.
`computeChunkId`	Recompute a chunk’s content-derived ID for integrity checks.

Python exposes the same surface in snake_case (build_signed_data_package, verify_data_package, build_chunk_chain, verify_chunk_chain, create_buyer_access_entry, build_buyer_access_files, decrypt_chunk, compute_chunk_id).

Types

type FieldRuleAction = "include" | "exclude" | "encrypt" | "anonymize";
type AnonymizeStrategy = "hash" | "generalize" | "redact";
type KeyStorageBackend = "tee" | "software";

type UnsignedDataPayload = {
  topic: string;                       // ROS 2 topic or feed name
  fields: Record<string, unknown>;
  capturedAt: string;                  // ISO 8601
};

type TopicFieldRules = {
  topic: string;
  rules: FieldRule[];                  // ordered; first matching fieldPath wins
  defaultAction?: FieldRuleAction;
};

type FieldRule = {
  fieldPath: string;                   // dot-notation, e.g. "gps.latitude"
  action: FieldRuleAction;
  anonymizeStrategy?: AnonymizeStrategy;
};

type DataPackage = {
  did: string;                 // did:peaq:0x...
  timestamp: string;           // ISO 8601
  schemaVersion: string;
  sequenceNumber: number;
  payload: Record<string, unknown>;
  signature: string;           // EIP-191 personal_sign
  signingKeyId: string;
};

type VerificationResult = {
  valid: boolean;
  signerDid: string;
  reason?: string;
};

Signing credentials ride in a SigningContext ({ privateKey, backend: "tee" | "software", did, keyId }, whose toJSON redacts the key so it never leaks through serialization) and field-level encryption in an EncryptionContext ({ encryptionKey, algorithm: "AES-256-GCM" }). The field layer uses AES-256-GCM; the per-chunk layer (inside buildChunkChain) uses XChaCha20-Poly1305. Chunking defaults: chunkSize 262144 bytes (256 KiB), hashAlgorithm "sha-256", via ChunkingConfig. The full envelope shape is documented under Data streams → The chunk envelope.

Example

Sign a reading, verify it, chunk-and-encrypt, then grant a buyer access:

import {
  buildSignedDataPackage,
  verifyDataPackage,
  buildChunkChain,
  createBuyerAccessEntry,
  buildBuyerAccessFiles,
  SigningContext,
  EncryptionContext,
} from "@peaqos/peaq-os-sdk";

const MACHINE_DID = process.env.MACHINE_DID!; // did:peaq:0x<machine address>, matches MACHINE_DATA_KEY

// X25519 recipient keys and the Ed25519 chunk-signing key pair, hex-encoded
const {
  OWNER_X25519_PUB, OWNER_X25519_PRIV, OPERATOR_X25519_PUB, MACHINE_X25519_PUB,
  BUYER_X25519_PUB, ED25519_PRIV, ED25519_PUB,
} = process.env as Record<string, string>;

const pkg = await buildSignedDataPackage({
  unsignedPayload: {
    topic: "/vehicle/telemetry",
    fields: { speed: 12, gps: { latitude: 47.1, longitude: 8.5 } },
    capturedAt: new Date().toISOString(),
  },
  topicRules: {
    topic: "/vehicle/telemetry",
    rules: [
      { fieldPath: "gps.latitude", action: "encrypt" },
      { fieldPath: "gps.longitude", action: "encrypt" },
    ],
    defaultAction: "include",
  },
  signingCtx: new SigningContext({
    privateKey: process.env.MACHINE_DATA_KEY!,
    backend: "software",
    did: MACHINE_DID,
    keyId: `${MACHINE_DID}#keys-1`,
  }),
  encryptionCtx: new EncryptionContext({
    encryptionKey: process.env.FIELD_KEY!,
    algorithm: "AES-256-GCM",
  }),
  schemaVersion: "1",
  sequenceNumber: 1,
});

const result = await verifyDataPackage(pkg);
if (!result.valid) throw new Error(result.reason);

// Chunk + encrypt; keys wrapped to owner, operator, and machine
const { chain, encryptedData } = buildChunkChain({
  input: new TextEncoder().encode(JSON.stringify(pkg)),
  ownerPublicKeyHex: OWNER_X25519_PUB,
  operatorPublicKeyHex: OPERATOR_X25519_PUB,
  machinePublicKeyHex: MACHINE_X25519_PUB,
  signingPrivateKeyHex: ED25519_PRIV,
  signingPublicKeyHex: ED25519_PUB,
  machineDid: MACHINE_DID,
  machineKeyId: `${MACHINE_DID}#keys-1`,
  did: MACHINE_DID,
  // config: { chunkSize: 262144 }
});

// Grant a buyer: re-wrap the purchased chunk keys to the buyer's public key
const accessFiles = buildBuyerAccessFiles({
  chunks: chain.chunks,
  ownerPrivateKeyHex: OWNER_X25519_PRIV,
  buyerRecipientId: "did:peaq:0xBUYER",
  buyerPublicKeyHex: BUYER_X25519_PUB,
});

// createBuyerAccessEntry is the per-chunk primitive behind it
const oneEntry = createBuyerAccessEntry({
  chunk: chain.chunks[0],
  ownerPrivateKeyHex: OWNER_X25519_PRIV,
  buyerRecipientId: "did:peaq:0xBUYER",
  buyerPublicKeyHex: BUYER_X25519_PUB,
});

import json
import os
from dataclasses import asdict
from datetime import datetime, timezone

from peaq_os_sdk.stream import (
    BuildChunkChainParams,
    EncryptionContext,
    SigningContext,
    build_signed_data_package,
    verify_data_package,
    build_chunk_chain,
    create_buyer_access_entry,
    build_buyer_access_files,
)
from peaq_os_sdk.types.stream import FieldRule, TopicFieldRules, UnsignedDataPayload

MACHINE_DID = os.environ["MACHINE_DID"]  # did:peaq:0x<machine address>, matches MACHINE_DATA_KEY

# X25519 recipient keys and the Ed25519 chunk-signing key pair, hex-encoded
OWNER_X25519_PUB = os.environ["OWNER_X25519_PUB"]
OWNER_X25519_PRIV = os.environ["OWNER_X25519_PRIV"]
OPERATOR_X25519_PUB = os.environ["OPERATOR_X25519_PUB"]
MACHINE_X25519_PUB = os.environ["MACHINE_X25519_PUB"]
BUYER_X25519_PUB = os.environ["BUYER_X25519_PUB"]
ED25519_PRIV = os.environ["ED25519_PRIV"]
ED25519_PUB = os.environ["ED25519_PUB"]

# Same flow in snake_case: sign the payload with the machine's
# signing context, verify, build the encrypted chunk chain, then
# re-wrap chunk keys for the buyer.
# build_signed_data_package takes keyword-only args: there is no params
# object (BuildSignedDataPackageParams is JS-only). Mirrors the TS call above.
pkg = build_signed_data_package(
    unsigned_payload=UnsignedDataPayload(
        topic="/vehicle/telemetry",
        fields={"speed": 12, "gps": {"latitude": 47.1, "longitude": 8.5}},
        captured_at=datetime.now(timezone.utc).isoformat(),
    ),
    topic_rules=TopicFieldRules(
        topic="/vehicle/telemetry",
        rules=(
            FieldRule(field_path="gps.latitude", action="encrypt"),
            FieldRule(field_path="gps.longitude", action="encrypt"),
        ),
        default_action="include",
    ),
    signing_ctx=SigningContext(
        private_key=os.environ["MACHINE_DATA_KEY"],
        backend="software",
        did=MACHINE_DID,
        key_id=f"{MACHINE_DID}#keys-1",
    ),
    encryption_ctx=EncryptionContext(
        encryption_key=os.environ["FIELD_KEY"],
        algorithm="AES-256-GCM",
    ),
    schema_version="1",
    sequence_number=1,
)
result = verify_data_package(pkg)
assert result.valid, result.reason

# Chunk + encrypt; keys wrapped to owner, operator, and machine
chain_result = build_chunk_chain(
    BuildChunkChainParams(
        input=json.dumps(asdict(pkg)),
        owner_public_key_hex=OWNER_X25519_PUB,
        operator_public_key_hex=OPERATOR_X25519_PUB,
        machine_public_key_hex=MACHINE_X25519_PUB,
        signing_private_key_hex=ED25519_PRIV,
        signing_public_key_hex=ED25519_PUB,
        machine_did=MACHINE_DID,
        machine_key_id=f"{MACHINE_DID}#keys-1",
        did=MACHINE_DID,
        # config=ChunkingConfig(chunk_size=262144)
    )
)

# Batch helper: re-wrap every purchased chunk key to the buyer (takes chunks)
access_files = build_buyer_access_files(
    chunks=chain_result.chain.chunks,
    owner_private_key_hex=OWNER_X25519_PRIV,
    buyer_recipient_id="did:peaq:0xBUYER",
    buyer_public_key_hex=BUYER_X25519_PUB,
)

# create_buyer_access_entry is the per-chunk primitive behind it
one_entry = create_buyer_access_entry(
    chunk=chain_result.chain.chunks[0],
    owner_private_key_hex=OWNER_X25519_PRIV,
    buyer_recipient_id="did:peaq:0xBUYER",
    buyer_public_key_hex=BUYER_X25519_PUB,
)

buildChunkChain returns the chain plus encryptedData, a map of ciphertext bytes keyed by chunk index. The ciphertext is a pre-upload sidecar: callers store the bytes wherever they distribute from and set each chunk’s storageRef. The buyer decrypts with decryptChunk({ chunk, recipientPrivateKeyHex, recipientEntry, encryptedData }), or decrypt_chunk(chunk=…, recipient_private_key_hex=…, recipient_entry=…, encrypted_data=…) in Python, passing the chunk’s stored ciphertext bytes plus a KeyRecipient from chunk.encryption.keyRecipients or from a buyer access entry. The data itself is never re-encrypted when access is granted. From the terminal, the same publish and grant flows are peaqos stream publish and peaqos stream grant.

Errors

Error	When
`StreamValidationError`	Invalid input: bad DID, missing field, unknown rule, wrong key length, key-commitment mismatch.
`StreamSigningError`	Signing or verification failed. Messages never contain key material.
`StreamError`	Base class for the above.

Solana signing

OWS mnemonic-derived wallets carry a account (ed25519, derivation path m/44'/501'/0'/0') for cross-chain payments and for machines homed on Solana. This is wallet signing, separate from Stream data signing, which uses EIP-191. The vault signs Solana transactions natively. PeaqosClient.solanaSignerFromWallet(nameOrId) (Python: solana_signer_from_wallet()) returns a signer for that wallet’s Solana account. transferToken uses it for SPL and native transfers when paying for stream data (peaqos stream pay --chain solana from the terminal), and the same signer signs management writes and event submission for a machine homed on Solana. Solana onboarding (peaqos activate --chain solana) signs its owner stages with the same wallet. See SDK JS: Solana, SDK Python: Solana and Onboard a machine on Solana. Machine Market orders quoted in Solana tokens are paid externally with your Solana wallet; Scale is paused, so no order can be placed today.

​Module location

​Functions

​Types

​Example

​Errors

​Solana signing

​Related