Stream - peaq

The stream module ships in @peaqos/peaq-os-sdk and peaq-os-sdk 0.3.0+. Its exports are flagged @experimental and may change as the surface settles. This page covers the cryptographic core; selling and shipping the encrypted chunks — purchases, payment rails, and S3 / P2P delivery — is the 0.4.0+ distribution surface.

The stream module 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 are handled by the on-machine Edge Agent, which drives this same module from a Data Event Map.

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 signer and check it against the public key derived from the DID. 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 write distributable access files (sharded by size). 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 } — 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 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: "did:peaq:0xMACHINE",
    keyId: "did:peaq:0xMACHINE#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 } = await 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: "did:peaq:0xMACHINE",
  machineKeyId: "did:peaq:0xMACHINE#keys-1",
  did: "did:peaq:0xMACHINE",
  // 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,
});

from peaq_os_sdk.stream import (
    build_signed_data_package,
    verify_data_package,
    build_chunk_chain,
    create_buyer_access_entry,
    build_buyer_access_files,
)

# 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=unsigned_payload,
    topic_rules=topic_rules,
    signing_ctx=signing_ctx,
    encryption_ctx=encryption_ctx,
    schema_version="1",
    sequence_number=1,
)
result = verify_data_package(pkg)
assert result.valid, result.reason

chain_result = build_chunk_chain(params)       # BuildChunkChainParams

# 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 }) — in Python, decrypt_chunk(chunk=…, recipient_private_key_hex=…, recipient_entry=…, encrypted_data=…) — 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. This is wallet signing — separate from Stream data signing, which uses EIP-191. Since SDK 0.4.0, the vault signs Solana transactions natively: PeaqosClient.solanaSignerFromWallet() returns a signer for the active OWS wallet’s Solana account, and transferToken uses it for SPL and native transfers when paying for stream data (peaqos stream pay --chain solana from the terminal). Solana-quoted Machine Market orders are still paid externally — complete the transfer with your Solana wallet and pass --payment-tx-hash to peaqos scale order; there is no peaqos solana command group.

​Module location

​Functions

​Types

​Example

​Errors

​Solana signing

​Related

Module location

Functions

Types

Example

Errors

Solana signing

Related