import type { ShadeEnvelope, StorageProvider, RatchetMessage } from '@shade/core'; import { ShadeSessionManager, ShadeEventEmitter, NoSessionError, StreamRatchet, StreamHandshakeError, } from '@shade/core'; import { FingerprintGateRegistry, type FingerprintGateHandler, } from './gates.js'; import { SubtleCryptoProvider, MemoryStorage, createWorkerCryptoProvider, type WorkerCryptoProvider, createEncryptStream, createDecryptStream, type CreateEncryptStreamOptions, type CreateDecryptStreamOptions, } from '@shade/crypto-web'; import { encodeEnvelope, decodeEnvelope, inspectEnvelopeType, encodeStreamOpen, encodeStreamOpenAck, decodeStreamHandshake, encodeStreamFrame, decodeStreamFrame, } from '@shade/proto'; import { ShadeFetchTransport, type KTVerifierOptions } from '@shade/transport'; import { LightWitness } from '@shade/key-transparency'; import type { SignedTreeHead, STHWire } from '@shade/key-transparency'; import { TransferEngine, ShadeTransferHttpTransport, MultiTransportFallback, type ITransferTransport, type IncomingTransfer, type TransferHandle, type TransferOptions, type TransferSummary, type OutboundQueue as OutboundQueueLike, type QueuedEventInput, } from '@shade/transfer'; import type { Hono } from 'hono'; import { BackgroundTasks } from './background.js'; import { exportBackup, applyBackupPayload, decryptBackup, backupToString, backupFromString, } from './backup.js'; import { computeFingerprint, deserializeIdentityKeyPair } from '@shade/core'; import { acceptBroadcastEnvelope, createBroadcastChannelImpl, getBroadcastChannelImpl, listBroadcastChannelsImpl, maybeHandleControlPlaintext, type BroadcastChannel, type BroadcastChannelSummary, type BroadcastSdkHooks, type MessageMeta, } from './broadcast.js'; import type { ResolvedConfig, StorageSpec } from './config.js'; import { ShadeControlChannel, ShadeTransferAuthenticator, type ControlEnvelopeTransport, } from './streams-bridge.js'; import { createFilesNamespace, type FilesNamespace } from '@shade/files'; import type { ObservabilityHook } from '@shade/observability'; import { isAllowedThumbnailMime, sha256Once, THUMBNAIL_MAX_BYTES, type StreamFileMetadata, type ThumbnailMime, } from '@shade/streams'; import { generateThumbnail as generateThumbnailFromBlob, type ThumbnailGenerationOptions, } from './thumbnail.js'; /** * V3.9 — extended upload options. The base `TransferOptions` is forwarded * verbatim to `@shade/transfer`; the extra fields control the thumbnail * companion-stream and never leak past the SDK boundary. */ /** * V3.11 — opt-in WebRTC P2P transport. Pass to `shade.configureWebRTC()` * before the engine is built. The shape mirrors `WebRtcConnectionManager`'s * options without forcing the SDK to import `@shade/transport-webrtc` at * the value layer (the import happens lazily inside `engine()`). */ export interface ShadeWebRtcConfig { /** * WebRTC adapter. Use `nativeRtcFactory()` from `@shade/transport-webrtc` * in browsers / Deno / Cloudflare Workers; supply your own * `IRtcFactory` for Node-class environments (`node-datachannel`, `wrtc`). */ factory: import('@shade/transport-webrtc').IRtcFactory; iceServers?: import('@shade/transport-webrtc').ShadeRtcConfig['iceServers']; iceTransportPolicy?: import('@shade/transport-webrtc').ShadeRtcConfig['iceTransportPolicy']; bundlePolicy?: import('@shade/transport-webrtc').ShadeRtcConfig['bundlePolicy']; /** Default 30s. */ connectTimeoutMs?: number; /** Default 30s. */ requestTimeoutMs?: number; /** Default 4 MiB. */ backpressureThresholdBytes?: number; } /** Live WebRTC runtime returned by `shade.getWebRtcRuntime()`. */ export interface ShadeWebRtcRuntime { signaling: import('@shade/transport-webrtc').WebRtcSignalingChannel; manager: import('@shade/transport-webrtc').WebRtcConnectionManager; transport: import('@shade/transport-webrtc').WebRtcTransferTransport; fallback: MultiTransportFallback; /** Internal — wires `engine` into the receiver hooks once it's built. */ attachEngine(engine: TransferEngine): void; /** Tear down the manager, the signaling channel, and any open peer connections. */ destroy(): void; } export interface ShadeUploadOptions extends TransferOptions { /** * Pre-generated thumbnail bytes + MIME. Use this on server runtimes * where in-process image processing is already part of your pipeline, * or when you have a richer thumbnail than the auto-generator would * produce (e.g. center-cropped, branded watermark, etc.). */ thumbnail?: { bytes: Uint8Array; mime: ThumbnailMime }; /** * Browser auto-generation. Set to `true` for defaults, or pass a * config object. Returns `null` (silently skips the thumbnail) on * runtimes lacking `OffscreenCanvas` + `createImageBitmap`. */ generateThumbnail?: boolean | ThumbnailGenerationOptions; } /** * The high-level Shade API. * * Wraps crypto, storage, session management, transport, and optional * observer into a single object. Provides magic auto-establish + auto- * publish + auto-replenish behavior. */ export class Shade { private readonly crypto = new SubtleCryptoProvider(); private readonly events = new ShadeEventEmitter(); private storage!: StorageProvider; private manager!: ShadeSessionManager; private transport!: ShadeFetchTransport; private _background: BackgroundTasks | null = null; private address!: string; private initialized = false; // Per-address mutex to serialize session establishment under concurrent sends private establishing = new Map>(); // Per-address encrypt queue to serialize ratchet mutations private encryptChains = new Map>(); // Per-`from` decrypt queue: serializes incoming receives so two concurrent // shade.receive(from, env) calls can't race the ratchet/storage. Without // this, parallel deliveries (relay duplicate fan-out, fast pipelined // sends) hit `database is locked` (sqlite) or transaction conflicts (IDB) // because the underlying StorageProvider isn't required to be a // concurrent-safe writer. See V4.8.2 changelog. private decryptChains = new Map>(); // Message handlers — may be sync or async; receive() awaits each. The // optional third arg distinguishes direct vs broadcast plaintexts; // handlers registered without it work unchanged (V4.6 back-compat). private messageHandlers: Array< (from: string, plaintext: string, meta?: MessageMeta) => void | Promise > = []; // Stream-transfer engine, lazily constructed on first use. private transferEngine: TransferEngine | null = null; private controlChannel: ShadeControlChannel | null = null; private peerBaseUrlResolver: ((peerAddress: string) => Promise) | null = null; private envelopeOutboxes: ControlEnvelopeTransport | null = null; private transferTransportOverride: ITransferTransport | null = null; private transferQueue: OutboundQueueLike | null = null; // `@shade/files` namespace, lazy + memoized. private filesNamespace: FilesNamespace | null = null; // V3.12 — light-witness for split-view detection. private ktWitness: LightWitness | null = null; // V3.11 WebRTC P2P transport. Lazy-built on first engine() if configured. private webrtcConfig: ShadeWebRtcConfig | null = null; private webrtcRuntime: ShadeWebRtcRuntime | null = null; // V3.8 Worker-Crypto. Lazy: configured via `configureWorkerCrypto()`, // spawned the first time `encryptStream`/`decryptStream` is used. private workerCryptoConfig: { workerUrl: URL | string; idleTimeoutMs?: number; } | null = null; private workerCrypto: WorkerCryptoProvider | null = null; private workerCryptoBoot: Promise | null = null; // V3.3 fingerprint gates. Created in `initialize()` once storage is up. private gates!: FingerprintGateRegistry; constructor(private readonly config: ResolvedConfig) {} /** * Initialize the SDK: * 1. Resolve storage backend * 2. Create session manager + generate identity if needed * 3. Create transport * 4. Generate initial one-time prekeys * 5. Register with prekey server * 6. Start background tasks */ async initialize(): Promise { if (this.initialized) return; // Step 1: Storage this.storage = await resolveStorage(this.config.storage); // Step 2: Session manager with event bus attached this.manager = new ShadeSessionManager(this.crypto, this.storage, { events: this.events, ...(this.config.observability !== undefined ? { observability: this.config.observability } : {}), }); await this.manager.initialize(); // Step 3: Address (user-provided or persisted UUID) this.address = this.config.address ?? (await resolveAddress(this.storage)); // Step 4: Transport with our signing key const identity = await this.storage.getIdentityKeyPair(); if (!identity) throw new Error('Identity not available after initialize'); // V3.12 — wire up KT verifier + light-witness if configured. let ktOpts: KTVerifierOptions | undefined; if (this.config.keyTransparency) { const baseUrl = this.config.prekeyServer; this.ktWitness = new LightWitness({ crypto: this.crypto, logPublicKey: this.config.keyTransparency.logPublicKey, maxStaleMs: this.config.keyTransparency.maxStaleMs, maxStored: this.config.keyTransparency.witnessMaxStored, fetcher: { async fetchLatestSTH(): Promise { const res = await fetch(`${baseUrl}/v1/kt/sth`); if (!res.ok) throw new Error(`KT /sth: ${res.status}`); return (await res.json()) as STHWire; }, async fetchConsistencyProof(from, to): Promise<{ proof: string[] }> { const res = await fetch(`${baseUrl}/v1/kt/consistency?from=${from}&to=${to}`); if (!res.ok) throw new Error(`KT /consistency: ${res.status}`); return (await res.json()) as { proof: string[] }; }, }, }); ktOpts = { mode: this.config.keyTransparency.mode, logPublicKey: this.config.keyTransparency.logPublicKey, maxStaleMs: this.config.keyTransparency.maxStaleMs, onObserveSth: async (sth: SignedTreeHead) => { if (this.ktWitness) { // The fetched STH was already verified by the transport; feed // it to the witness for split-view tracking. `observe` may also // throw on split-view — we let it propagate to the caller. await this.ktWitness.observe(sth); } }, }; } this.transport = new ShadeFetchTransport({ baseUrl: this.config.prekeyServer, crypto: this.crypto, signingPrivateKey: identity.signingPrivateKey, ...(ktOpts ? { keyTransparency: ktOpts } : {}), }); // Step 5: Initial prekeys + register const otpks = await this.manager.generateOneTimePreKeys(20); const bundle = await this.manager.createPreKeyBundle(); try { await this.transport.register( this.address, this.manager.getPublicIdentity(), bundle.signedPreKey, otpks, ); } catch (err) { console.warn( `[Shade] Failed to register with prekey server at ${this.config.prekeyServer}: ${(err as Error).message}. Will retry on next replenish.`, ); } // Step 6: Background tasks this._background = new BackgroundTasks( this.manager, this.transport, this.address, this.config, ); this._background.start(); // Step 7: V3.3 fingerprint gates this.gates = new FingerprintGateRegistry(this.storage); this.initialized = true; } /** Your identity's safety number (12 groups × 5 digits) */ get fingerprint(): Promise { if (!this.initialized) throw new Error('Not initialized'); return this.manager.getIdentityFingerprint(); } /** Your address on the prekey server */ get myAddress(): string { if (!this.initialized) throw new Error('Not initialized'); return this.address; } /** * The local device's Ed25519 identity public key (32 bytes). * * Stable for the lifetime of the identity. After {@link rotate} this * reflects the new key; the previous key is preserved in retired- * identities storage for the configured grace period. * * Hand this to your application's backend at enrollment time so it * can verify signatures from this device, compute its own safety- * number representation, or pin the key for later attestation. Use * {@link fingerprint} instead for human side-channel comparison. */ get identityPublicKey(): Promise { if (!this.initialized) throw new Error('Not initialized'); return this.storage.getIdentityKeyPair().then((kp) => { if (!kp) throw new Error('Identity not yet generated'); return kp.signingPublicKey; }); } /** * `@shade/files` namespace — high-level entry point for E2EE filesystem * RPC. Lazily creates the underlying channel + streams bridges on first * access; subsequent accesses return the same instance. * * ```ts * const files = shade.files; * const stop = await files.serve({ list: ..., write: ..., ... }); * const fs = await files.client('bob'); * await fs.list('/'); * ``` * * Requires `configureTransfers({ resolveBaseUrl })` to be called first * (same as `upload`/`onIncomingTransfer`). */ get files(): FilesNamespace { if (!this.initialized) throw new Error('Not initialized'); if (this.filesNamespace !== null) return this.filesNamespace; // `@shade/files` only imports `Shade` as a type, so the cyclic ESM // import is type-only at the value layer — safe to bind synchronously. this.filesNamespace = createFilesNamespace(this); return this.filesNamespace; } /** Internal — exposes the BackgroundTasks for `@shade/files` to wire prune. */ get background(): BackgroundTasks | null { return this._background; } /** Access the underlying event emitter (for observer integration) */ getEvents(): ShadeEventEmitter { return this.events; } /** Access the underlying session manager (for advanced usage) */ getManager(): ShadeSessionManager { return this.manager; } /** Access the underlying transport (for advanced usage) */ getTransport(): ShadeFetchTransport { return this.transport; } /** * V3.12 — access the configured Key-Transparency light-witness, or * `null` when KT was not configured. Useful for surfacing observed * STHs to UI / dashboards, or for manual gossip checks against * trusted peers. */ getKTWitness(): LightWitness | null { return this.ktWitness; } /** * Returns the OTel observability hook the SDK was configured with, or * `undefined` if observability is off. Used by `@shade/files` and other * sub-modules to inherit the same tracer the rest of the SDK uses. */ getObservability(): ObservabilityHook | undefined { return this.config.observability; } /** * Encrypt a message to a peer. Auto-establishes a session if none exists. * Returns the ShadeEnvelope ready to send over any transport. */ async send(address: string, plaintext: string): Promise { if (!this.initialized) throw new Error('Not initialized'); // Serialize all sends to the same peer: the SessionManager mutates // ratchet state in place, and interleaved mutations corrupt it. const previous = this.encryptChains.get(address) ?? Promise.resolve(); const next = previous .catch(() => {}) // don't propagate upstream failures to later sends .then(async () => { try { return await this.manager.encrypt(address, plaintext); } catch (err) { if (!(err instanceof NoSessionError)) throw err; await this.ensureSession(address); return this.manager.encrypt(address, plaintext); } }); this.encryptChains.set(address, next); return next as Promise; } /** * Decrypt an incoming envelope and notify registered message handlers. * Returns the plaintext. * * The caller provides the `from` address because the envelope itself * doesn't authenticate the sender — that's determined by your transport * layer (auth header, WebSocket peer, push notification metadata, etc.). * * V4.6: when the decrypted plaintext is a broadcast control message * (sender-key distribution / revocation), the SDK consumes it * internally and returns an empty string; user handlers do NOT fire. * Apps therefore see only direct plaintexts here. Broadcast payloads * arrive via {@link Shade.acceptBroadcast}. */ async receive(from: string, envelope: ShadeEnvelope): Promise { if (!this.initialized) throw new Error('Not initialized'); // Serialize ONLY the ratchet/storage write portion of receive (the // call into `manager.decrypt`). Concurrent decrypts race the // SessionManager ratchet (mutated in place) and the StorageProvider // (not required to be a concurrent-safe writer — `bun:sqlite` // throws `database is locked`, IDB throws transaction conflicts). // The Prism FR called this out: a relay-duplicated WS fan-out // dispatched 8 parallel `shade.receive(from, env)` calls, one won // the X3DH prekey race and the other 7 failed with // `database is locked` / `one-time prekey not found`. The fix is // to queue per-`from` decrypts so the ratchet step is sequential. // // Crucially the user-visible MESSAGE HANDLERS run *outside* the // queue. Streams + file-RPC issue nested `shade.receive` calls for // the same peer from inside their handlers (e.g. `stream-end` // arrives while a write-RPC is still waiting on chunks); holding // the queue across the handler would self-deadlock. The atomic // unit we have to protect is just the ratchet+storage step, not // the consumer's reaction to it. const previous = this.decryptChains.get(from) ?? Promise.resolve(); const decryptPromise = previous .catch(() => undefined) // don't propagate upstream failures .then(() => this.manager.decrypt(from, envelope)); // Store a never-rejecting copy so the next chained receive doesn't // see a rejection from this one (we still surface our own rejection // to *this* caller via the original `decryptPromise`). this.decryptChains.set(from, decryptPromise.catch(() => undefined)); const plaintext = await decryptPromise; const consumed = await maybeHandleControlPlaintext( this.broadcastHooks(), from, plaintext, ); if (consumed) return ''; const meta: MessageMeta = { kind: 'direct' }; for (const handler of this.messageHandlers) { try { await handler(from, plaintext, meta); } catch (err) { console.error('[Shade] Message handler threw:', err); } } return plaintext; } /** * Register a handler for incoming messages. Async handlers are awaited. * * V4.6: handlers may declare an optional `meta` parameter to discriminate * direct (`meta.kind === 'direct'`) from broadcast (`meta.kind === 'broadcast'`) * deliveries. Handlers that ignore the third arg keep working unchanged * for direct messages. */ onMessage( handler: (from: string, plaintext: string, meta?: MessageMeta) => void | Promise, ): () => void { this.messageHandlers.push(handler); return () => { this.messageHandlers = this.messageHandlers.filter((h) => h !== handler); }; } // ─── V4.6 Broadcast channels ─────────────────────────────── /** * Create a new broadcast channel owned by this device. Returns a * handle for adding/removing members, encrypting a single payload, * and rotating on revocation. The channel id is opaque, stable * across `shutdown()` / re-open, and persisted via the configured * `StorageProvider`. */ async createBroadcastChannel(opts: { label?: string } = {}): Promise { if (!this.initialized) throw new Error('Not initialized'); return createBroadcastChannelImpl(this.broadcastHooks(), opts); } /** * Look up an existing sender-side broadcast channel by id. Returns * `null` when the id is unknown OR when this device only holds a * receiver-side copy (the receiver path uses `onMessage` for delivery * — there is no app-facing handle on the receive side). */ async getBroadcastChannel(channelId: string): Promise { if (!this.initialized) throw new Error('Not initialized'); return getBroadcastChannelImpl(this.broadcastHooks(), channelId); } /** * Snapshot of every broadcast channel persisted on this device, * including receiver-side channels that we joined. Useful for * rebuilding UI state on startup. */ async listBroadcastChannels(): Promise { if (!this.initialized) throw new Error('Not initialized'); return listBroadcastChannelsImpl(this.broadcastHooks()); } /** * Hand a wire-encoded broadcast envelope (type 0x21) to the SDK. * Decrypts via the matching channel, advances the chain, and dispatches * the plaintext to `onMessage` handlers with `meta.kind === 'broadcast'`. * * Stale generations (sender's old chain after a rotation we already * received) are silently dropped. Future generations (we haven't seen * the rotation distribution yet) throw — the app should ensure the * distribution envelope is delivered before the broadcast. */ async acceptBroadcast(envelope: Uint8Array): Promise { if (!this.initialized) throw new Error('Not initialized'); const result = await acceptBroadcastEnvelope(this.broadcastHooks(), envelope); if (result === null) return; for (const handler of this.messageHandlers) { try { await handler(result.meta.kind === 'broadcast' ? result.meta.sender : '', result.plaintext, result.meta); } catch (err) { console.error('[Shade] Broadcast handler threw:', err); } } } private broadcastHooks(): BroadcastSdkHooks { return { bilateralSend: (peer, pt) => this.send(peer, pt), myAddress: () => this.address, crypto: this.crypto, storage: this.storage, }; } /** Get a peer's fingerprint (requires an existing session) */ async getFingerprintFor(address: string): Promise { if (!this.initialized) throw new Error('Not initialized'); return this.manager.getRemoteFingerprint(address); } /** Verify a fingerprint matches the pinned identity for an address */ async verify(address: string, fingerprint: string): Promise { const remote = await this.getFingerprintFor(address); return normalize(remote) === normalize(fingerprint); } // ─── V3.3 fingerprint gates ─────────────────────────────── /** * Register a handler that runs before `upload()` proceeds when the file * is at or above `threshold` bytes and the peer is not yet verified. * Return `true` to allow + persist the verification, `false` to abort. * * Default threshold (when this method is never called): 10 MiB. */ beforeFirstLargeFile(threshold: number, handler: FingerprintGateHandler): void { if (!this.initialized) throw new Error('Not initialized'); this.gates.registerFirstLargeFile(threshold, handler); } /** * Register a handler that runs before `importBackup()` writes to storage. * The handler receives the fingerprint of the identity *embedded in the * backup blob*, so the user can OOB-confirm the backup is theirs. */ beforeBackupImport(handler: FingerprintGateHandler): void { if (!this.initialized) throw new Error('Not initialized'); this.gates.registerBackupImport(handler); } /** * Register a handler that runs the first time a peer's rotated identity * is observed (via `acceptIdentityChange` or X3DH against a new bundle). */ beforeNewDeviceTrust(handler: FingerprintGateHandler): void { if (!this.initialized) throw new Error('Not initialized'); this.gates.registerNewDeviceTrust(handler); } /** * Register a handler that runs per-recipient before an inbox fan-out * delivery (V3.6). Reserved hook — wired here so apps can register it * today and have it active automatically when V3.6 ships. */ beforeInboxFanout(handler: FingerprintGateHandler): void { if (!this.initialized) throw new Error('Not initialized'); this.gates.registerInboxFanout(handler); } /** * Mark a peer as verified at their current fingerprint. Call this from * your own UI (e.g. after the user scans a QR code or reads the safety * number aloud) to satisfy any gate without going through the handler. */ async markPeerVerified(address: string): Promise { if (!this.initialized) throw new Error('Not initialized'); const fingerprint = await this.manager.getRemoteFingerprint(address); await this.gates.markVerified(address, fingerprint, 'user'); } /** * Returns whether `address` has a current verification (fingerprint and * identity-version both still match). */ async isPeerVerified(address: string): Promise { if (!this.initialized) throw new Error('Not initialized'); const fingerprint = await this.manager.getRemoteFingerprint(address); return this.gates.isVerified(address, fingerprint); } /** Drop any persisted verification for `address`. */ async unmarkPeerVerified(address: string): Promise { if (!this.initialized) throw new Error('Not initialized'); await this.gates.revoke(address); } /** * Move every per-peer storage row for `oldLabel` (session, trusted * identity, peer-verification, identity-version counter) to * `newLabel`. Use this when first-contact forced you to label a * session by the relay's sender-fingerprint hint * (`fp:` — see `IncomingMessage.from` / `FetchedBlob.from`) and * the just-decrypted plaintext announces the peer's canonical * address: alias once and every subsequent * `send`/`receive`/broadcast cross-check operates under the * announced label, no app-side fp ↔ address mapping needed for the * receive path. * * The rename is atomic from a per-peer-mutex perspective — both * labels are locked for the duration so concurrent encrypt/decrypt * can't observe a half-moved state. Throws if `oldLabel` has no * session, or if `newLabel` already does (refuses to overwrite — * call `resetSession` first if that's intentional). * * After alias, the SDK's internal serialization queues * (`encryptChains`, `decryptChains`) for `oldLabel` are dropped so * future operations don't queue behind a stale chain. * * V4.8.3 — Prism FR `session-label-asymmetry-v4.8.2.md`. */ async aliasSession(oldLabel: string, newLabel: string): Promise { if (!this.initialized) throw new Error('Not initialized'); await this.manager.aliasSession(oldLabel, newLabel); // The SDK's per-`from` chains are keyed by label; drop the old // entries so future `send`/`receive` to either label start with a // fresh queue rather than chaining off whatever was last in flight // for `oldLabel`. this.encryptChains.delete(oldLabel); this.decryptChains.delete(oldLabel); } /** * Resolve a session label to the label its state actually lives under, * following any alias left behind by `aliasSession`. * * Transports need this to route by the canonical address after a * restart, when the only sender hint they hold is the first-contact * `fp:` label. `encrypt`/`decrypt` resolve internally — this is * for callers that must know the address itself. * * V4.12 — durable session-label aliases. */ async resolveSessionLabel(label: string): Promise { if (!this.initialized) throw new Error('Not initialized'); return this.manager.resolveSessionLabel(label); } /** * Accept a peer's rotated identity. Bumps the per-peer identity-version * counter so any earlier verification automatically goes stale, then * runs the `beforeNewDeviceTrust` gate before the new key is pinned. */ async acceptIdentityChange(address: string, newIdentityKey: Uint8Array): Promise { if (!this.initialized) throw new Error('Not initialized'); await this.storage.bumpPeerIdentityVersion(address); const newFingerprint = await computeFingerprint( this.crypto, // X3DH stores DH-only "trusted identity"; in this SDK the trusted // entry IS the DH public key. We feed it as both args so the // fingerprint binds to the rotated key material the user is asked // to confirm. newIdentityKey, newIdentityKey, ); await this.gates.checkNewDeviceTrust(address, newFingerprint); await this.manager.acceptIdentityChange(address, newIdentityKey); } /** Manually rotate the identity (destructive — see docs) */ async rotate(): Promise { if (!this.initialized) throw new Error('Not initialized'); // Rotate locally first const newBundle = await this.manager.rotateIdentity(); // Rebuild the transport with the new signing key so subsequent // signed operations (replenish, delete, register) work const identity = await this.storage.getIdentityKeyPair(); if (!identity) throw new Error('Identity missing after rotate'); this.transport = new ShadeFetchTransport({ baseUrl: this.config.prekeyServer, crypto: this.crypto, signingPrivateKey: identity.signingPrivateKey, }); // Re-upload the new bundle await this.transport.register( this.address, this.manager.getPublicIdentity(), newBundle.signedPreKey, [], ); // Rebuild background tasks so they use the new transport if (this._background) { this._background.stop(); this._background = new BackgroundTasks( this.manager, this.transport, this.address, this.config, ); this._background.start(); } } /** Manually trigger replenishment (normally background task handles this) */ async replenish(): Promise { if (!this.initialized) throw new Error('Not initialized'); if (!this._background) return 0; return this._background.runReplenish(); } /** * Export an encrypted backup blob that can be restored to a new device. * * @param passphrase User passphrase (minimum 12 characters) * @param knownAddresses Peer addresses whose sessions should be included */ async exportBackup(passphrase: string, knownAddresses: string[] = []): Promise { if (!this.initialized) throw new Error('Not initialized'); const blob = await exportBackup(this.crypto, this.storage, passphrase, knownAddresses); return backupToString(blob); } /** * Restore state from a backup string. Overwrites existing state. * Call this BEFORE initialize() on a fresh device, or after shutdown() + re-init. * * V3.3: invokes the `beforeBackupImport` gate. The handler receives the * fingerprint of the identity *embedded in the backup* — this lets the * user OOB-confirm that the backup is theirs before sessions and * pinned-trust entries are written to disk. */ async importBackup(backupString: string, passphrase: string): Promise { if (!this.initialized) throw new Error('Not initialized'); const blob = backupFromString(backupString); const payload = await decryptBackup(this.crypto, blob, passphrase); const backupFingerprint = await fingerprintFromBackupPayload(this.crypto, payload); await this.gates.checkBackupImport(this.address, backupFingerprint); await applyBackupPayload(this.storage, payload); // Reload identity after restore const restored = await this.storage.getIdentityKeyPair(); if (restored) { // Rebuild the manager and transport with the restored identity this.manager = new ShadeSessionManager(this.crypto, this.storage, { events: this.events, ...(this.config.observability !== undefined ? { observability: this.config.observability } : {}), }); await this.manager.initialize(); this.transport = new ShadeFetchTransport({ baseUrl: this.config.prekeyServer, crypto: this.crypto, signingPrivateKey: restored.signingPrivateKey, }); } } /** Clean shutdown: stop timers, close storage if it supports it */ async shutdown(): Promise { this._background?.stop(); if (this.transferEngine !== null) this.transferEngine.destroy(); if (this.controlChannel !== null) this.controlChannel.destroy(); if (this.webrtcRuntime !== null) { this.webrtcRuntime.destroy(); this.webrtcRuntime = null; } if (this.workerCrypto !== null) { await this.workerCrypto.destroy(); this.workerCrypto = null; } // Close storage if it has a close method (SQLite) const closable = this.storage as unknown as { close?: () => void | Promise }; if (typeof closable.close === 'function') { await closable.close(); } this.initialized = false; } // ─── Worker-Crypto streams (V3.8) ────────────────────────── /** * Opt in to Web Workers crypto: subsequent `encryptStream` / * `decryptStream` calls offload all AEAD work to a dedicated worker so * the main thread stays under the 16 ms-per-frame budget for big * uploads. The worker is spawned on first use and self-terminates * after `idleTimeoutMs` of inactivity (default 30 s). * * Bundlers resolve worker URLs differently — the recommended idiom is: * * ```ts * shade.configureWorkerCrypto({ * workerUrl: new URL('@shade/crypto-web/worker', import.meta.url), * }); * ``` * * See `docs/web-workers.md` for Vite / Webpack / Rollup recipes and * Safari notes. */ configureWorkerCrypto(opts: { workerUrl: URL | string; idleTimeoutMs?: number; }): void { this.workerCryptoConfig = opts; } /** * Encrypt a `ReadableStream` of plaintext into stream-chunk * wire envelopes via a Web Worker. * * The caller pre-negotiates `streamId` + `streamSecret` with the peer * (typically through `shade.upload()` for HTTP-based delivery, or any * other channel). The returned `stream` is a TransformStream: * pipe plaintext in, get encrypted chunks out. * * `laneSha256` resolves once the stream finishes (final chunk emitted * with `isLast=true`). Compare it against the receiver's lane sha256 * for end-to-end integrity proof. * * Requires `configureWorkerCrypto()` to be called first. */ encryptStream( opts: Omit, ): Promise<{ stream: TransformStream; laneSha256: Promise; }> { return this.ensureWorkerCrypto().then((provider) => createEncryptStream({ provider, ...opts }), ); } /** * Inverse of {@link Shade.encryptStream} — decrypt incoming wire * envelopes back into plaintext. Each input chunk MUST be a complete * stream-chunk envelope (the wire framing is the caller's job). * * Requires `configureWorkerCrypto()` to be called first. */ decryptStream( opts: Omit, ): Promise<{ stream: TransformStream; laneSha256: Promise; }> { return this.ensureWorkerCrypto().then((provider) => createDecryptStream({ provider, ...opts }), ); } /** * Direct access to the worker-backed `CryptoProvider`. Use when you * want to run a one-off heavy crypto op (X25519 batch DH, big HKDF * derivation, etc.) off the main thread without setting up a stream. */ async getWorkerCrypto(): Promise { return this.ensureWorkerCrypto(); } private async ensureWorkerCrypto(): Promise { if (this.workerCrypto !== null) return this.workerCrypto; if (this.workerCryptoBoot !== null) return this.workerCryptoBoot; if (this.workerCryptoConfig === null) { throw new Error( 'Call shade.configureWorkerCrypto({ workerUrl }) before encryptStream()/decryptStream(). See docs/web-workers.md.', ); } const cfg = this.workerCryptoConfig; this.workerCryptoBoot = (async () => { const provider = await createWorkerCryptoProvider({ workerUrl: cfg.workerUrl, ...(cfg.idleTimeoutMs !== undefined ? { idleTimeoutMs: cfg.idleTimeoutMs } : {}), }); this.workerCrypto = provider; return provider; })(); try { return await this.workerCryptoBoot; } finally { this.workerCryptoBoot = null; } } // ─── Stream transfers (v0.2.0) ───────────────────────────── /** * Configure how stream-transfer chunks reach peers. Provide a resolver * that returns the peer's HTTP base URL (e.g. by looking up a * `transfer.baseUrl` field in your prekey-bundle metadata or a static * directory map). If unset, `upload()` rejects with a clear error. * * Optionally also override the control-envelope transport (defaults to * HTTP POSTs to `/v1/transfer/control`). */ configureTransfers(opts: { /** * Resolver for the peer's HTTP base URL (used by the default * `ShadeTransferHttpTransport` to POST chunks). Optional when a * custom `transport` and `envelopeTransport` are supplied — e.g. * for pull-mode browser servers (`@shade/files transferQueueRoute`) * which never POST chunks anywhere. */ resolveBaseUrl?: (peerAddress: string) => Promise; /** * Override the chunk-level transport. Defaults to * `ShadeTransferHttpTransport` (HTTP POSTs per chunk) when * `resolveBaseUrl` is supplied. Required when `resolveBaseUrl` * is omitted. */ transport?: ITransferTransport; /** * Override the control-envelope transport. Defaults to HTTP POSTs * to `/v1/transfer/control` when `resolveBaseUrl` is * supplied. Required when `resolveBaseUrl` is omitted. */ envelopeTransport?: ControlEnvelopeTransport; }): void { if (opts.resolveBaseUrl === undefined) { if (opts.transport === undefined || opts.envelopeTransport === undefined) { throw new Error( 'configureTransfers: resolveBaseUrl is required unless both `transport` and `envelopeTransport` are supplied (e.g. for pull-mode queue servers).', ); } this.peerBaseUrlResolver = async () => { throw new Error( 'resolveBaseUrl was not configured — this Shade is in queue/pull mode and does not POST chunks. Configure a custom transport instead.', ); }; } else { this.peerBaseUrlResolver = opts.resolveBaseUrl; } this.transferTransportOverride = opts.transport ?? null; if (opts.envelopeTransport !== undefined) { this.envelopeOutboxes = opts.envelopeTransport; } else if (opts.resolveBaseUrl !== undefined) { this.envelopeOutboxes = new HttpEnvelopeTransport(opts.resolveBaseUrl, this.address); } else { throw new Error( 'configureTransfers: envelopeTransport is required when resolveBaseUrl is omitted.', ); } } /** * Deliver a freshly-encrypted ratchet envelope to a peer using the * configured envelope transport (HTTP POST to `/v1/transfer/control` by * default). Used by `@shade/files` for RPC plaintext delivery. */ async deliverControlEnvelope(peerAddress: string, envelope: ShadeEnvelope): Promise { if (this.envelopeOutboxes === null) { throw new Error( 'Call shade.configureTransfers({ resolveBaseUrl }) before deliverControlEnvelope()', ); } await this.envelopeOutboxes.send(peerAddress, envelope); } /** * Upload bytes to a peer. Returns a `TransferHandle` that can be paused/ * aborted and awaited. Requires `configureTransfers` to be called first. * * V3.3: when the file size is at or above the configured threshold * (default 10 MiB) and the peer is not yet verified, the registered * `beforeFirstLargeFile` handler is invoked. Rejection throws * `FingerprintNotVerifiedError` before any bytes hit the wire. * * V3.9: pass `thumbnail: { bytes, mime }` to attach a pre-generated * preview, or `generateThumbnail: true` to auto-derive a 256x256 preview * from an image input in browser-class runtimes (no-op elsewhere). The * thumbnail is shipped as a *separate* E2EE stream and referenced from * the main stream's `fileMetadata`. */ async upload(opts: ShadeUploadOptions): Promise { if (!this.initialized) throw new Error('Not initialized'); const size = inferTransferSize(opts); if (size !== null && size >= this.gates.getFirstLargeFileThreshold()) { // Establish the session up-front so we have a fingerprint to gate on. // For peers we've never contacted, this is the TOFU moment where the // gate matters most. if ((await this.storage.getSession(opts.to)) === null) { await this.ensureSession(opts.to); } const fingerprint = await this.manager.getRemoteFingerprint(opts.to); await this.gates.checkFirstLargeFile(opts.to, fingerprint, size); } const engine = await this.engine(); const thumbnail = await this.resolveThumbnail(opts); if (thumbnail !== null) { const fileMeta: StreamFileMetadata = { ...(opts.metadata?.fileMetadata ?? {}), thumbnailStreamId: thumbnail.streamId, thumbnailHash: thumbnail.hashB64, thumbnailMime: thumbnail.mime, thumbnailBytes: thumbnail.bytes, }; const merged: TransferOptions = { ...opts, metadata: { ...(opts.metadata ?? {}), fileMetadata: fileMeta, }, }; return engine.upload(merged); } return engine.upload(opts); } /** * Coordinate the thumbnail-side of a V3.9 upload. Resolves to either * - `null` — no thumbnail will be attached (caller passed neither * `thumbnail` nor a generator that produced bytes), or * - the streamId + sha256 + mime + bytes of the thumbnail-stream that * has now been kicked off (it runs to completion in the background; * the main upload's `done()` is independent). */ private async resolveThumbnail(opts: ShadeUploadOptions): Promise<{ streamId: string; hashB64: string; mime: ThumbnailMime; bytes: number; } | null> { let bytes: Uint8Array | null = null; let mime: ThumbnailMime | null = null; if (opts.thumbnail !== undefined) { bytes = opts.thumbnail.bytes; mime = opts.thumbnail.mime; } else if (opts.generateThumbnail !== undefined && opts.generateThumbnail !== false) { const genOpts: ThumbnailGenerationOptions = opts.generateThumbnail === true ? {} : opts.generateThumbnail; const gen = await generateThumbnailFromBlob(opts.input, genOpts); if (gen !== null) { bytes = gen.bytes; mime = gen.mime; } } if (bytes === null || mime === null) return null; if (bytes.byteLength > THUMBNAIL_MAX_BYTES) { throw new Error( `thumbnail size ${bytes.byteLength} exceeds THUMBNAIL_MAX_BYTES (${THUMBNAIL_MAX_BYTES})`, ); } if (!isAllowedThumbnailMime(mime)) { throw new Error(`thumbnail mime ${mime} not in allowlist`); } const hash = sha256Once(bytes); const hashB64 = bytesToBase64Std(hash); const engine = await this.engine(); // Ship the thumbnail FIRST so the receiver can present a preview the // moment the main `stream-init` references it. Single lane, single // chunk — at ≤ 64 KiB the parallelism overhead would dominate. const handle = await engine.upload({ to: opts.to, input: bytes, lanes: 1, chunkSize: Math.max(1, bytes.byteLength), metadata: { contentType: mime, userMetadata: { shadeThumbnail: '1', }, }, }); // Don't await `done()` — the main upload should not block on the // thumbnail finishing. Errors on the preview are surfaced via the // returned handle's events (consumer can listen if they care). handle.done().catch((err) => { console.warn('[Shade] thumbnail transfer failed:', err); }); return { streamId: handle.streamId, hashB64, mime, bytes: bytes.byteLength, }; } /** * Subscribe to incoming transfers from peers. Handler is invoked when a * `stream-init` arrives; the handler MUST call `incoming.accept(...)` to * begin receiving (or `incoming.decline(...)` to reject). */ async onIncomingTransfer( handler: (incoming: IncomingTransfer) => void | Promise, ): Promise<() => void> { return (await this.engine()).onIncomingTransfer(handler); } /** * Mount the **pull-mode** transfer routes on a Hono app. Mount under * any base path: `app.route('/api/v1/shade-files', shade.transferQueueRoute())`. * * Configures this Shade instance to queue all outbound chunks + * control envelopes per peer instead of POSTing them. Browser-style * receivers drain the queue via long-polling — no inbound HTTP * listener required on the receiver. * * Routes mounted (relative to the base path): * POST /queue — long-poll the per-peer outbound queue * POST /v1/transfer/:streamId/chunk — receive incoming chunks (browser → server) * GET /v1/transfer/:streamId/state — resume-state lookup * POST /v1/transfer/control — receive incoming control envelopes * GET /v1/transfer/health — peer reachability probe * * **Idempotent**: calling twice returns a fresh `Hono` app each * time but reuses the underlying queue + transport (so the engine * stays single). * * **Ordering**: must be called **before** `shade.files.serve(...)` * (or any other path that builds the engine), because configuring * the queue transport mutates the transfer stack. Calling after the * engine is built throws. */ async transferQueueRoute(opts: TransferQueueRouteOptions = {}): Promise { if (this.transferEngine !== null && this.transferTransportOverride === null) { throw new Error( 'transferQueueRoute(): the transfer engine has already been built with the default HTTP transport. Call transferQueueRoute() before any upload()/onIncomingTransfer()/configureTransfers().', ); } const { OutboundQueue, QueueTransferTransport } = await import('@shade/transfer'); if (this.transferQueue === null) { this.transferQueue = new OutboundQueue({ ...(opts.maxEventsPerPeer !== undefined ? { maxEventsPerPeer: opts.maxEventsPerPeer } : {}), ...(opts.idleEvictionMs !== undefined ? { idleEvictionMs: opts.idleEvictionMs } : {}), }); } if (this.transferTransportOverride === null) { const queueTransport = new QueueTransferTransport(this.transferQueue); const queueEnvelopeTransport = new QueueEnvelopeTransport(this.transferQueue); this.configureTransfers({ transport: queueTransport, envelopeTransport: queueEnvelopeTransport, }); } const queue = this.transferQueue; const blockMs = opts.blockMs ?? 30_000; const maxBlockMs = opts.maxBlockMs ?? 55_000; const engine = await this.engine(); const { createTransferRoutes, PermissiveAuthenticator } = await import('@shade/transfer'); const app = await createTransferRoutes(engine, { authenticator: PermissiveAuthenticator, }); app.post('/v1/transfer/control', async (c) => { const senderAddress = c.req.header('X-Shade-Sender-Address'); if (senderAddress === undefined || senderAddress === '') { return c.json({ error: 'missing X-Shade-Sender-Address' }, 400); } const ab = await c.req.arrayBuffer(); const bytes = new Uint8Array(ab); try { await this.acceptTransferEnvelope(senderAddress, bytes); } catch (err) { return c.json({ error: (err as Error).message }, 400); } return c.json({ ok: true }); }); // Long-poll endpoint. app.post('/queue', async (c) => { const senderAddress = c.req.header('X-Shade-Sender-Address'); if (senderAddress === undefined || senderAddress === '') { return c.json({ error: 'missing X-Shade-Sender-Address' }, 400); } let body: { since?: unknown; blockMs?: unknown }; try { body = (await c.req.json()) as { since?: unknown; blockMs?: unknown }; } catch { return c.json({ error: 'invalid JSON body' }, 400); } const since = typeof body.since === 'number' && Number.isFinite(body.since) ? body.since : 0; const requestedBlockMs = typeof body.blockMs === 'number' && Number.isFinite(body.blockMs) ? Math.max(0, Math.min(maxBlockMs, body.blockMs)) : blockMs; // Bun-side short-circuit if the request was aborted while we // were holding the long-poll. AbortSignal from the request body // is already surfaced via `c.req.raw.signal` in Hono. const events = await queue.drain(senderAddress, since, requestedBlockMs, c.req.raw.signal); return c.json({ events: events.map((e) => ({ id: e.id, timestampMs: e.timestampMs, kind: e.kind, bytesB64: bytesToBase64Std(e.bytes), ...(e.kind === 'chunk' ? { meta: e.meta } : {}), })), nextSince: events.length > 0 ? events[events.length - 1]!.id : since, }); }); return app; } /** * Mount the receiver-side HTTP routes on a Hono app. Mount under any * base path: `app.route('/shade', await shade.transferRoute())`. * * Routes: * POST /v1/transfer/:streamId/chunk — wire-encoded 0x11 chunks * GET /v1/transfer/:streamId/state — resume-state lookup * POST /v1/transfer/control — wire-encoded 0x02 control envelopes * GET /v1/transfer/health — peer reachability probe */ async transferRoute(): Promise { const engine = await this.engine(); const { createTransferRoutes, PermissiveAuthenticator } = await import('@shade/transfer'); const app = await createTransferRoutes(engine, { authenticator: PermissiveAuthenticator, }); // Add the control-envelope POST route on top. app.post('/v1/transfer/control', async (c) => { const senderAddress = c.req.header('X-Shade-Sender-Address'); if (senderAddress === undefined || senderAddress === '') { return c.json({ error: 'missing X-Shade-Sender-Address' }, 400); } const ab = await c.req.arrayBuffer(); const bytes = new Uint8Array(ab); try { await this.acceptTransferEnvelope(senderAddress, bytes); } catch (err) { return c.json({ error: (err as Error).message }, 400); } return c.json({ ok: true }); }); return app; } /** * Low-level entry for custom transports: hand a `0x02` ratchet envelope * (control-plane) or a `0x11` stream-chunk envelope to the engine. * Used internally by `transferRoute()`. */ async acceptTransferEnvelope(from: string, env: ShadeEnvelope | Uint8Array): Promise { const engine = await this.engine(); if (env instanceof Uint8Array) { const kind = inspectEnvelopeType(env); if (kind === 'stream-chunk') { // Engine extracts laneId/seq from the wire bytes via decodeStreamChunk. const headers = parseChunkHeader(env); await engine.receiveChunk(from, headers.streamId, headers.laneId, headers.seq, env); return; } if (kind === 'ratchet' || kind === 'prekey') { const decoded = decodeEnvelope(env); await this.controlChannel!.acceptEnvelope(from, decoded); return; } throw new Error(`Unknown envelope type ${kind}`); } // Already-decoded envelope (ratchet or prekey) await this.controlChannel!.acceptEnvelope(from, env); } // ─── V3.11 WebRTC P2P transport ──────────────────────────── /** * Opt in to direct peer-to-peer chunk delivery via WebRTC. * * When configured, `upload()` builds a `[WebRTC, HTTP]` * {@link MultiTransportFallback}: P2P first, HTTP as automatic * fallback. Signaling (SDP offer/answer + trickle-ICE) rides on top * of `Shade.send` / `Shade.onMessage` — no out-of-band server needed. * * Must be called BEFORE the first `upload()` / `onIncomingTransfer()` * (those instantiate the transfer engine, which captures the * transport stack at construction time). Calling later throws. * * The `factory` argument is the WebRTC adapter — `nativeRtcFactory()` * for browsers, a custom one for Node-class environments * (`node-datachannel`, `wrtc`, etc.). Set `iceServers` to override the * default public STUN list, or supply TURN credentials for paranoid * NATs: * * ```ts * import { nativeRtcFactory } from '@shade/transport-webrtc'; * shade.configureWebRTC({ * factory: nativeRtcFactory(), * iceServers: [ * { urls: 'stun:stun.l.google.com:19302' }, * { urls: 'turn:turn.example.com:3478', username: 'u', credential: 'p' }, * ], * }); * ``` */ configureWebRTC(opts: ShadeWebRtcConfig): void { if (this.transferEngine !== null) { throw new Error( 'shade.configureWebRTC() must be called before upload()/onIncomingTransfer() builds the engine', ); } this.webrtcConfig = opts; } /** * Returns the live WebRTC runtime (signaling channel + connection * manager + transport) if `configureWebRTC` was called and `engine()` * has been instantiated. Useful for diagnostics: peek * `runtime.manager.isConnected('alice')` to see whether a P2P link is * live, or wire `runtime.fallback.onSwitch(...)` to log demotions. */ getWebRtcRuntime(): ShadeWebRtcRuntime | null { return this.webrtcRuntime; } // ─── Internals ───────────────────────────────────────────── private async engine(): Promise { if (this.transferEngine !== null) return this.transferEngine; if (this.peerBaseUrlResolver === null || this.envelopeOutboxes === null) { throw new Error( 'Call shade.configureTransfers({ resolveBaseUrl }) before using upload()/onIncomingTransfer()', ); } this.controlChannel = new ShadeControlChannel(this, this.envelopeOutboxes); let transport: ITransferTransport; let webrtcRuntime: ShadeWebRtcRuntime | null = null; if (this.transferTransportOverride !== null) { // Custom transport (queue, in-memory, custom adapter) — used as-is. // WebRTC fallback only attaches when the default HTTP transport is // active because WebRTC's `MultiTransportFallback` is HTTP-shaped. transport = this.transferTransportOverride; } else { const httpTransport: ITransferTransport = new ShadeTransferHttpTransport({ resolveBaseUrl: this.peerBaseUrlResolver, authenticator: await this.makeAuthenticator(), }); transport = httpTransport; if (this.webrtcConfig !== null) { webrtcRuntime = await this.buildWebRtcRuntime(this.webrtcConfig, httpTransport); transport = webrtcRuntime.fallback; } } this.transferEngine = new TransferEngine({ crypto: this.crypto, controlChannel: this.controlChannel, transport, myAddress: this.address, ...(this.config.observability !== undefined ? { observability: this.config.observability } : {}), }); if (webrtcRuntime !== null) { // Receiver-hooks need to dispatch into the freshly-built engine. webrtcRuntime.attachEngine(this.transferEngine); this.webrtcRuntime = webrtcRuntime; } return this.transferEngine; } /** * Dynamically import `@shade/transport-webrtc`, wire its signaling * channel onto our `Shade.send`/`Shade.onMessage`, and build a * MultiTransportFallback that prefers WebRTC then falls back to HTTP. */ private async buildWebRtcRuntime( cfg: ShadeWebRtcConfig, httpTransport: ITransferTransport, ): Promise { // `@shade/transport-webrtc` is an optional peer dep — keep the // import lazy so consumers that don't use WebRTC don't pay for it. const moduleId = '@shade/transport-webrtc'; const mod = (await import(moduleId)) as typeof import('@shade/transport-webrtc'); const { WebRtcSignalingChannel, WebRtcConnectionManager, WebRtcTransferTransport, createShadeBridgeFromShade, } = mod; const signaling = new WebRtcSignalingChannel(createShadeBridgeFromShade(this)); let engineRef: TransferEngine | null = null; const manager = new WebRtcConnectionManager({ factory: cfg.factory, signaling, ...(cfg.iceServers !== undefined || cfg.iceTransportPolicy !== undefined || cfg.bundlePolicy !== undefined ? { config: { ...(cfg.iceServers !== undefined ? { iceServers: cfg.iceServers } : {}), ...(cfg.iceTransportPolicy !== undefined ? { iceTransportPolicy: cfg.iceTransportPolicy } : {}), ...(cfg.bundlePolicy !== undefined ? { bundlePolicy: cfg.bundlePolicy } : {}), }, } : {}), ...(cfg.connectTimeoutMs !== undefined ? { connectTimeoutMs: cfg.connectTimeoutMs } : {}), receiver: { onChunk: async (from, streamId, laneId, seq, envelope) => { if (engineRef === null) { throw new Error('webrtc receiver hook fired before engine attached'); } return engineRef.receiveChunk(from, streamId, laneId, seq, envelope); }, onResumeQuery: async (from, streamId) => { if (engineRef === null) return null; return engineRef.getResumeState(from, streamId); }, }, }); const webrtcTransport = new WebRtcTransferTransport({ manager, ...(cfg.requestTimeoutMs !== undefined ? { requestTimeoutMs: cfg.requestTimeoutMs } : {}), ...(cfg.backpressureThresholdBytes !== undefined ? { backpressureThresholdBytes: cfg.backpressureThresholdBytes } : {}), }); const fallback = new MultiTransportFallback([ { name: 'webrtc', transport: webrtcTransport }, { name: 'http', transport: httpTransport }, ]); return { signaling, manager, transport: webrtcTransport, fallback, attachEngine(engine) { engineRef = engine; }, destroy() { manager.destroy(); signaling.destroy(); }, }; } private async makeAuthenticator(): Promise { const identity = await this.storage.getIdentityKeyPair(); if (identity === null) throw new Error('Identity not initialized'); return new ShadeTransferAuthenticator(this.crypto, this.address, identity.signingPrivateKey); } /** Returns a list of in-flight stream transfers from storage (resume support). */ async listTransfers(filter?: { direction?: 'send' | 'receive'; }): Promise { if (this.storage.listActiveStreamStates === undefined) return []; const rows = await this.storage.listActiveStreamStates(filter?.direction); return rows.map((s) => ({ streamId: s.streamId, direction: s.direction, peerAddress: s.peerAddress, status: s.status, bytesProcessed: 0, // computed from laneState createdAt: s.createdAt, updatedAt: s.updatedAt, metadata: tryParseMetadata(s.metadataJson), })); } /** * Drop persisted stream-state records whose `updatedAt` is strictly * less than `olderThan` (Unix ms). Idempotent. Returns silently when * the configured storage backend does not implement stream-state * persistence (e.g. memory storage in tests). * * Recommended usage: schedule on a daily cron with a 14-day horizon * — see `docs/streams.md` § Retention. The `bun-server` SDK template * wires this up by default. */ async pruneStreamStates(olderThan: number): Promise { if (this.storage.pruneStreamStates === undefined) return; await this.storage.pruneStreamStates(olderThan); } // ─── Streaming sub-sessions (V4.11) ──────────────────────── /** * Open a long-lived streaming Double-Ratchet sub-session to an * already-known peer, for wrapping individual frames on a * bidirectional, often server-heavy channel (e.g. a console-log * WebSocket) with the same confidentiality / forward-secrecy / * replay guarantees as the HTTP `send`/`receive` path. * * This is the **initiator** half. Like the rest of the SDK it is * transport-agnostic: it produces handshake/frame bytes you put on * your WebSocket, and consumes the bytes you receive from it. * * ```ts * const stream = await shade.openStream(peerAddr); * ws.send(stream.handshakeFrame()); // → STREAM_OPEN * // … first inbound WS frame is the peer's STREAM_OPEN_ACK … * await stream.handleHandshake(ackBytes); // stream now usable * ws.send(await stream.seal(utf8(line))); // outbound frame * onLog(await stream.open(inboundBytes)); // inbound frame * await stream.close(); // on ws close * ``` * * Independence (R5): this never touches the stored parent session, * its prekeys, or the per-peer `send`/`receive` queues — it runs * concurrently against the same peer. The ratchet lives only in * memory and is zeroized by {@link ShadeStream.close}; a dropped * connection is re-opened with a fresh `openStream`, never resumed * (persisting per-frame ratchet secrets would defeat forward * secrecy). * * Note (Double-Ratchet semantics): a responder cannot `seal` until * it has `open`ed at least one frame from the initiator (standard * Signal behaviour). For a server-heavy stream either make the bursty * sender the initiator, or have the initiator send one priming frame * right after the handshake. * * Requires an established parent session; one is auto-established * (same path as {@link send}) if missing. */ async openStream(peerAddress: string): Promise { if (!this.initialized) throw new Error('Not initialized'); let begun; try { begun = await this.manager.beginStream(peerAddress); } catch (err) { if (!(err instanceof NoSessionError)) throw err; await this.ensureSession(peerAddress); begun = await this.manager.beginStream(peerAddress); } return new ShadeStream({ peer: peerAddress, role: 'initiator', streamId: begun.streamId, events: this.events, handshakeOut: encodeStreamOpen(begun.streamId, begun.ephemeralPublicKey), complete: begun.complete, }); } /** * Accept an inbound stream — the **responder** half. Feed it the * peer's `STREAM_OPEN` bytes (route by {@link inspectEnvelopeType} * `=== 'stream-open'`). The returned stream is immediately usable for * `open()`; send `handshakeFrame()` (the `STREAM_OPEN_ACK`) back over * the transport so the initiator can complete its side. */ async acceptStream(peerAddress: string, openBytes: Uint8Array): Promise { if (!this.initialized) throw new Error('Not initialized'); const hs = decodeStreamHandshake(openBytes); if (hs.kind !== 'open') { throw new StreamHandshakeError(`expected STREAM_OPEN, got ${hs.kind}`); } let accepted; try { accepted = await this.manager.acceptStream(peerAddress, hs.streamId, hs.ephemeralPub); } catch (err) { if (!(err instanceof NoSessionError)) throw err; await this.ensureSession(peerAddress); accepted = await this.manager.acceptStream(peerAddress, hs.streamId, hs.ephemeralPub); } return new ShadeStream({ peer: peerAddress, role: 'responder', streamId: hs.streamId, events: this.events, handshakeOut: encodeStreamOpenAck(hs.streamId, accepted.ephemeralPublicKey), ratchet: accepted.stream, }); } private async ensureSession(address: string): Promise { // Deduplicate concurrent establishment requests const existing = this.establishing.get(address); if (existing) { await existing; return; } const promise = (async () => { const bundle = await this.transport.fetchBundle(address); await this.manager.initSessionFromBundle(address, bundle); })(); this.establishing.set(address, promise); try { await promise; } finally { this.establishing.delete(address); } } } // ─── ShadeStream (V4.11) ───────────────────────────────────── interface ShadeStreamInit { peer: string; role: 'initiator' | 'responder'; streamId: Uint8Array; events: ShadeEventEmitter; /** Bytes to put on the wire for our half of the handshake. */ handshakeOut: Uint8Array; /** Initiator only: continuation that derives the ratchet from the ACK. */ complete?: (peerEphemeralPub: Uint8Array) => Promise; /** Responder only: ratchet is ready at accept time. */ ratchet?: StreamRatchet; } function streamIdsEqual(a: Uint8Array, b: Uint8Array): boolean { if (a.length !== b.length) return false; for (let i = 0; i < a.length; i++) if (a[i] !== b[i]) return false; return true; } /** * A live streaming Double-Ratchet sub-session. Transport-agnostic: it * emits/consumes wire bytes, the caller owns the WebSocket (or any * other ordered frame transport). * * Lifecycle: * - **initiator**: `handshakeFrame()` → `STREAM_OPEN`; after the peer's * `STREAM_OPEN_ACK` arrives call `handleHandshake(ack)`; then * `seal`/`open`. * - **responder**: usable immediately; `handshakeFrame()` → * `STREAM_OPEN_ACK` to send back; `open` the initiator's first frame * before `seal` (standard Double-Ratchet ordering). */ export class ShadeStream { private readonly _streamId: Uint8Array; private readonly _peer: string; private readonly _role: 'initiator' | 'responder'; private readonly events: ShadeEventEmitter; private readonly handshakeOut: Uint8Array; private readonly complete?: (peerEphemeralPub: Uint8Array) => Promise; private ratchet: StreamRatchet | null; private state: 'await-ack' | 'open' | 'closed'; constructor(init: ShadeStreamInit) { this._streamId = init.streamId; this._peer = init.peer; this._role = init.role; this.events = init.events; this.handshakeOut = init.handshakeOut; if (init.role === 'initiator') { if (init.complete) this.complete = init.complete; this.ratchet = null; this.state = 'await-ack'; } else { this.ratchet = init.ratchet ?? null; this.state = 'open'; } } /** Peer address this stream is bound to. */ get peer(): string { return this._peer; } /** Which half of the handshake this end performed. */ get role(): 'initiator' | 'responder' { return this._role; } /** Lowercase-hex stream id (stable for the stream's lifetime). */ get streamId(): string { return Array.from(this._streamId, (b) => b.toString(16).padStart(2, '0')).join(''); } /** True once the ratchet is established and not yet closed. */ get isOpen(): boolean { return this.state === 'open' && this.ratchet !== null; } /** * The bytes for our half of the handshake to put on the transport * (`STREAM_OPEN` for an initiator, `STREAM_OPEN_ACK` for a responder). * Stable; safe to read once and send. */ handshakeFrame(): Uint8Array { return this.handshakeOut; } /** * Initiator only: consume the peer's `STREAM_OPEN_ACK` and derive the * ratchet. Idempotent-safe to call exactly once; throws if called on * a responder, out of order, or with a mismatched streamId. */ async handleHandshake(ackBytes: Uint8Array): Promise { if (this._role !== 'initiator') { throw new StreamHandshakeError('handleHandshake is initiator-only'); } if (this.state !== 'await-ack' || !this.complete) { throw new StreamHandshakeError('handshake already completed or stream closed'); } const hs = decodeStreamHandshake(ackBytes); if (hs.kind !== 'open-ack') { throw new StreamHandshakeError(`expected STREAM_OPEN_ACK, got ${hs.kind}`); } if (!streamIdsEqual(hs.streamId, this._streamId)) { throw new StreamHandshakeError('STREAM_OPEN_ACK streamId mismatch'); } this.ratchet = await this.complete(hs.ephemeralPub); this.state = 'open'; } /** * Seal one logical frame. Returns `STREAM_FRAME` wire bytes — put * exactly one in one WS frame. Advances the sending chain one step. */ async seal(plaintext: Uint8Array): Promise { if (!this.ratchet || this.state !== 'open') { throw new StreamHandshakeError('stream not open (complete the handshake first)'); } const msg = await this.ratchet.seal(plaintext); return encodeStreamFrame(this._streamId, msg); } /** * Open one inbound `STREAM_FRAME`. Correct and memory-bounded across * long one-directional bursts; replays / counter-rewinds are rejected * by the underlying ratchet. */ async open(wire: Uint8Array): Promise { if (!this.ratchet || this.state !== 'open') { throw new StreamHandshakeError('stream not open (complete the handshake first)'); } const frame: { streamId: Uint8Array; message: RatchetMessage } = decodeStreamFrame(wire); if (!streamIdsEqual(frame.streamId, this._streamId)) { throw new StreamHandshakeError('STREAM_FRAME streamId mismatch'); } return this.ratchet.open(frame.message); } /** Zeroize and drop the ratchet. Idempotent. */ async close(): Promise { if (this.state === 'closed') return; this.state = 'closed'; if (this.ratchet) { await this.ratchet.close(); this.ratchet = null; } this.events.emit('stream.closed', { address: this._peer }); } } function bytesToBase64Std(bytes: Uint8Array): string { let bin = ''; for (let i = 0; i < bytes.length; i++) bin += String.fromCharCode(bytes[i]!); return btoa(bin); } function tryParseMetadata(json: string): import('@shade/streams').StreamMetadata | null { try { return JSON.parse(json) as import('@shade/streams').StreamMetadata; } catch { return null; } } /** * Best-effort plaintext-size inference for a `TransferOptions.input`. * Returns null when the size is genuinely unknowable (raw `ReadableStream` * without a metadata hint), so the caller can decide whether to gate. */ function inferTransferSize(opts: TransferOptions): number | null { if (typeof opts.metadata?.sizeBytes === 'number') return opts.metadata.sizeBytes; const input = opts.input; if (input instanceof Uint8Array) return input.byteLength; // Blob and File both expose `.size`. Use a structural check so we don't // depend on lib.dom typings inside the SDK build. if (typeof (input as unknown as { size?: unknown }).size === 'number') { return (input as unknown as { size: number }).size; } return null; } /** * Compute the safety-number fingerprint for the identity embedded in a * decrypted backup payload. Used by `Shade.importBackup` to drive the * `beforeBackupImport` gate before any state is overwritten. */ async function fingerprintFromBackupPayload( crypto: SubtleCryptoProvider, payload: import('./backup.js').BackupPayload, ): Promise { if (payload.identity === null) { // No identity in the backup means there's nothing to fingerprint. // Return a stable sentinel so the gate handler can still display // something meaningful instead of throwing here. return 'no-identity-in-backup'; } const id = deserializeIdentityKeyPair(payload.identity); return computeFingerprint(crypto, id.signingPublicKey, id.dhPublicKey); } function parseChunkHeader(bytes: Uint8Array): { streamId: string; laneId: number; seq: bigint; } { // [0]=ver [1]=type [2..18]=streamId(16) [18..22]=laneId u32 [22..30]=seq u64 if (bytes.length < 30) throw new Error('truncated stream-chunk header'); const view = new DataView(bytes.buffer, bytes.byteOffset); const sidBytes = bytes.slice(2, 18); const laneId = view.getUint32(18, false); const seq = view.getBigUint64(22, false); // Encode streamId as base64url let bin = ''; for (let i = 0; i < sidBytes.length; i++) bin += String.fromCharCode(sidBytes[i]!); const streamId = btoa(bin).replace(/\+/g, '-').replace(/\//g, '_').replace(/=+$/, ''); return { streamId, laneId, seq }; } // ─── Queue-mode (pull) envelope transport ───────────────────── /** * Configuration for {@link Shade.transferQueueRoute}. All fields are * optional with sensible production defaults. */ export interface TransferQueueRouteOptions { /** * Long-poll timeout in milliseconds. Server holds the request open * up to this long before returning an empty `events` array. Default * 30_000. */ blockMs?: number; /** * Hard cap on long-poll timeout (clamps client-supplied `blockMs`). * Default 55_000 — under typical reverse-proxy idle thresholds (60s * on most CDNs). */ maxBlockMs?: number; /** * Per-peer ring-buffer size. When the queue is full, oldest events * are dropped on enqueue. Receivers detect the gap via missing * sequence numbers and re-resume from `since=0`. Default 1000. */ maxEventsPerPeer?: number; /** * Drop a peer's queue + reject pending pollers after this much * silence. Default 10 minutes. Setting to `0` disables idle-eviction. */ idleEvictionMs?: number; } /** * `ControlEnvelopeTransport` that enqueues outbound envelopes into an * `OutboundQueue` for browser-style receivers to long-poll. Mirrors * `HttpEnvelopeTransport` shape (one `send(peer, envelope)` method); * the difference is the destination — local queue, not remote HTTP. */ class QueueEnvelopeTransport implements ControlEnvelopeTransport { constructor(private readonly queue: OutboundQueueLike) {} async send(peerAddress: string, envelope: ShadeEnvelope): Promise { const bytes = encodeEnvelope(envelope); const event: QueuedEventInput = { kind: 'envelope', bytes }; this.queue.enqueue(peerAddress, event); } } // ─── Default HTTP envelope transport ────────────────────────── class HttpEnvelopeTransport implements ControlEnvelopeTransport { constructor( private readonly resolveBaseUrl: (peerAddress: string) => Promise, private readonly myAddress: string, ) {} async send(peerAddress: string, envelope: ShadeEnvelope): Promise { const base = (await this.resolveBaseUrl(peerAddress)).replace(/\/$/, ''); const url = `${base}/v1/transfer/control`; const bytes = encodeEnvelope(envelope); const res = await fetch(url, { method: 'POST', headers: { 'Content-Type': 'application/octet-stream', 'X-Shade-Sender-Address': this.myAddress, }, body: bytes as unknown as never, }); if (!res.ok) { throw new Error(`control envelope POST failed: ${res.status} ${await res.text()}`); } } } // ─── Helpers ───────────────────────────────────────────────── async function resolveStorage(spec: StorageSpec): Promise { if (typeof spec === 'object' && 'getIdentityKeyPair' in spec) { return spec; } if (spec === 'memory') { return new MemoryStorage(); } if (typeof spec === 'string' && spec.startsWith('sqlite:')) { const path = spec.slice('sqlite:'.length); const { SQLiteStorage } = await import('@shade/storage-sqlite'); return new SQLiteStorage(path); } if (typeof spec === 'object' && spec.type === 'postgres') { // Dynamic import keeps @shade/storage-postgres optional — consumers that // never use postgres don't need to install it. The string-form import // path makes the resolver lazy at type-check time too. const moduleId = '@shade/storage-postgres'; const mod = (await import(moduleId)) as { PostgresStorage: { create(url: string): Promise }; }; return mod.PostgresStorage.create(spec.url); } if (typeof spec === 'object' && spec.type === 'indexeddb') { // Dynamic import keeps @shade/storage-indexeddb optional — Node-only // consumers don't need to install a browser-only adapter. const moduleId = '@shade/storage-indexeddb'; const mod = (await import(moduleId)) as { IndexedDBStorage: { create(opts: { dbName?: string }): Promise }; }; const opts: { dbName?: string } = {}; if (spec.dbName !== undefined) opts.dbName = spec.dbName; return mod.IndexedDBStorage.create(opts); } throw new Error(`Unsupported storage spec: ${JSON.stringify(spec)}`); } async function resolveAddress(storage: StorageProvider): Promise { // Try to load a persisted address, else generate a random one and save it. // We reuse the config table by storing a special key. // Since StorageProvider doesn't expose a generic key-value, we just use // the local registration ID as a deterministic fallback. const id = await storage.getLocalRegistrationId(); return `device:${id}`; } function normalize(fp: string): string { return fp.replace(/\s+/g, ' ').trim(); }