From 5f1216bf07ce19a06aede0d9b940aca0597be9ce Mon Sep 17 00:00:00 2001 From: "Evgeny @ SimpleX Chat" <259188159+evgeny-simplex@users.noreply.github.com> Date: Mon, 1 Jun 2026 18:24:20 +0000 Subject: [PATCH] agent layer functions (untested) --- .../2026-05-31-agent-client.md | 35 +- smp-web/src/agent/client.ts | 3 +- smp-web/src/agent/smp-ops.ts | 367 ++++++++++++++++++ smp-web/tests/infra-test.ts | 72 ++++ 4 files changed, 465 insertions(+), 12 deletions(-) create mode 100644 smp-web/src/agent/smp-ops.ts diff --git a/rfcs/2026-03-20-smp-agent-web/2026-05-31-agent-client.md b/rfcs/2026-03-20-smp-agent-web/2026-05-31-agent-client.md index 540c51f2f..9d045360f 100644 --- a/rfcs/2026-03-20-smp-agent-web/2026-05-31-agent-client.md +++ b/rfcs/2026-03-20-smp-agent-web/2026-05-31-agent-client.md @@ -55,19 +55,32 @@ Each step produces a testable artifact. - Store wrappers (withStore, withStore', storeError) - **Test**: TS tests — create agent client, test worker lifecycle, test server selection -### Step 3: SMP session management + queue operations (cross-language via agent-repl) -- getSMPServerClient, smpConnectClient, smpClientDisconnected -- getSMPProxyClient, withProxySession -- withClient_, withClient, withSMPClient -- sendOrProxySMPMessage, sendOrProxySMPCommand -- newRcvQueue, newRcvQueue_ -- agentCbEncrypt, agentCbEncryptOnce, agentCbDecrypt -- sendConfirmation, sendInvitation, sendAgentMessage +### Step 3a: SMP session management + direct queue operations +- getSMPServerClient, smpConnectClient, waitForSMPClient +- agentCbEncrypt, agentCbEncryptOnce, agentCbDecrypt (using existing ClientMsgEnvelope from protocol.ts) +- sendAgentMessage, sendConfirmation, sendInvitation - secureQueue, secureSndQueue, sendAck - decryptSMPMessage -- subscribeQueues, subscribeQueues_, subscribeSessQueues_, processSubResults -- addNewQueueSubscription, resubscribeSMPSession -- **Test via agent-repl**: Haskell creates queue → TS subscribes, TS creates queue → Haskell sends → TS receives+decrypts, TS sends → Haskell receives +- newRcvQueue +- subscribeQueues, subscribeServerQueues +- addNewQueueSubscription +- **Test**: TS-only round-trip test for agentCbEncrypt (encode → decode → decrypt), then agent-repl cross-language tests (TS creates queue → Haskell sends → TS receives, TS sends → Haskell receives) + +### Step 3b: Disconnect handling + resubscription +- smpClientDisconnected (full: remove proxied relays, notify DOWN, trigger resubscribe) +- resubscribeSMPSession, resubscribeSessQueues +- processSubResults (partition into failed/subscribed) +- subscribeSessQueues_ (batch SUB + process results) +- **Test**: agent-repl: establish connection → kill WebSocket → verify subs move to pending → reconnect → verify resubscribed + +### Step 3c: Proxy operations +- getSMPProxyClient (get/create proxied relay session) +- withProxySession (bracket for proxied operations) +- sendOrProxySMPMessage (decide direct vs proxy, delegate) +- sendOrProxySMPCommand (decide direct vs proxy for SKEY etc) +- ipAddressProtected, shouldUseProxy logic +- withClient_, withClient, withSMPClient, withLogClient_ +- **Test**: agent-repl: TS sends via proxy → Haskell receives ### Step 4: Agent message flow (cross-language end-to-end) - agentRatchetEncrypt, agentRatchetEncryptHeader, agentRatchetDecrypt diff --git a/smp-web/src/agent/client.ts b/smp-web/src/agent/client.ts index 970faac16..9f760a400 100644 --- a/smp-web/src/agent/client.ts +++ b/smp-web/src/agent/client.ts @@ -132,7 +132,7 @@ export type ATransmission = [string, Uint8Array, any] // (corrId, connId, event export interface AgentClient { active: boolean subQ: ABQueue - // msgQ is per-client, stored in smpClients + msgQ: ABQueue // ServerMsg from SMP clients, processed by subscriber loop config: AgentConfig store: AgentStore smpServers: Map // userId → servers @@ -159,6 +159,7 @@ export function newAgentClient(config: AgentConfig, store: AgentStore, smpServer return { active: true, subQ: new ABQueue(config.tbqSize), + msgQ: new ABQueue(config.tbqSize), config, store, smpServers, diff --git a/smp-web/src/agent/smp-ops.ts b/smp-web/src/agent/smp-ops.ts new file mode 100644 index 000000000..e46a153db --- /dev/null +++ b/smp-web/src/agent/smp-ops.ts @@ -0,0 +1,367 @@ +// SMP session management and queue operations. +// Transpilation of Agent/Client.hs: getSMPServerClient, agentCbEncrypt/Decrypt, +// sendConfirmation, sendAgentMessage, newRcvQueue, subscribeQueues, etc. + +import type {SMPClient, ProxiedRelay} from "../client.js" +import {createSMPClient} from "../client.js" +import type {AuthKey, SMPResponse} from "../protocol.js" +import { + encodeClientMsgEnvelope, encodeClientMessage, + type ClientMsgEnvelope, type ClientMessage, type PubHeader, type PrivHeader, +} from "../protocol.js" +import {getSessVar, removeSessVar, tryReadSessVar, type SessionVar} from "./session.js" +import {AgentError, type AgentClient, type AgentErrorType} from "./client.js" +import {ABQueue} from "./queue.js" +import type {RcvQueueSub} from "./subscriptions.js" +import {cbEncrypt, cbDecrypt} from "@simplex-chat/xftp-web/dist/crypto/secretbox.js" +import {generateX25519KeyPair, generateEd25519KeyPair, dh, encodePubKeyX25519} from "@simplex-chat/xftp-web/dist/crypto/keys.js" +import {concatBytes, encodeBytes} from "@simplex-chat/xftp-web/dist/protocol/encoding.js" + +// -- Transport session key (simplified: one session per server, no TSMEntity) + +export function tSessKey(userId: number, server: string): string { + return `${userId}:${server}` +} + +// -- SMP connected client with proxy relay sessions + +export interface SMPConnectedClient { + client: SMPClient + proxiedRelays: Map> +} + +// -- Server message for msgQ (dispatched by subscriber loop) + +export interface ServerMsg { + userId: number + server: string + sessionId: Uint8Array + entityId: Uint8Array + msg: SMPResponse +} + +// -- getSMPServerClient (Client.hs:642-651) +// Get or create SMP client for the given server. +// Returns existing if connected, waits if pending, connects if new. +export async function getSMPServerClient( + c: AgentClient, + userId: number, + server: string, + keyHash: Uint8Array, + wsUrl: string, +): Promise { + if (!c.active) throw new AgentError({tag: "INACTIVE"}) + const key = tSessKey(userId, server) + const clients = c.smpClients as Map> + const {isNew, v} = getSessVar({val: c.workerSeq}, key, clients) + if (isNew) { + return smpConnectClient(c, userId, server, keyHash, wsUrl, key, v) + } + return waitForSMPClient(server, v) +} + +// smpConnectClient (Client.hs:704-718) +async function smpConnectClient( + c: AgentClient, + userId: number, + server: string, + keyHash: Uint8Array, + wsUrl: string, + key: string, + v: SessionVar, +): Promise { + const clients = c.smpClients as Map> + try { + const smp = await createSMPClient( + wsUrl, keyHash, + // onMessage (Client.hs:716 — messages go to msgQ) + (entityId: Uint8Array, msg: SMPResponse) => { + const serverMsg: ServerMsg = {userId, server, sessionId: smp.sessionId, entityId, msg} + c.msgQ.enqueue(serverMsg) + }, + // onDisconnected (Client.hs:720-754 — smpClientDisconnected) + () => smpClientDisconnected(c, userId, server, key, v, smp), + ) + // setSessionId in subscription tracker (Client.hs:717) + c.currentSubs.setSessionId(tSessKey(userId, server), smp.sessionId) + const connected: SMPConnectedClient = {client: smp, proxiedRelays: new Map()} + v.resolve(connected) + // Notify CONNECT (Client.hs:884) + c.subQ.enqueue(["", new Uint8Array(0), {tag: "CONNECT", server}]) + return connected + } catch (e) { + // Connection failed (Client.hs:886-895) + removeSessVar(v, key, clients) + v.reject(e instanceof Error ? e : new Error(String(e))) + throw new AgentError({tag: "BROKER", addr: server, err: "NETWORK"}) + } +} + +// waitForProtocolClient (Client.hs:847-868) +async function waitForSMPClient( + server: string, + v: SessionVar, +): Promise { + try { + return await v.promise + } catch { + throw new AgentError({tag: "BROKER", addr: server, err: "NETWORK"}) + } +} + +// smpClientDisconnected (Client.hs:720-754) +// Handle WebSocket disconnect: move subs to pending, notify DOWN, remove client. +function smpClientDisconnected( + c: AgentClient, + userId: number, + server: string, + key: string, + v: SessionVar, + smp: SMPClient, +): void { + const clients = c.smpClients as Map> + // removeSessVar (only if this is still the current client) + removeSessVar(v, key, clients) + if (!c.active) return + // Move active subs to pending (Client.hs:731-737) + const tSess = tSessKey(userId, server) + const moved = c.currentSubs.setSubsPending(tSess, smp.sessionId) + // Notify DISCONNECT (Client.hs:746) + c.subQ.enqueue(["", new Uint8Array(0), {tag: "DISCONNECT", server}]) + if (moved.size > 0) { + // Notify DOWN with affected connIds (Client.hs:747) + const connIds = [...new Set([...moved.values()].map(rq => toHex(rq.connId)))] + c.subQ.enqueue(["", new Uint8Array(0), {tag: "DOWN", server, connIds}]) + // TODO: trigger resubscription (Client.hs:750-754) + } +} + +// -- agentCbEncrypt (Client.hs:2074-2082) +// Per-queue E2E encrypt with stored DH secret. +// Returns encoded ClientMsgEnvelope. +export function agentCbEncrypt( + e2eDhSecret: Uint8Array, + smpClientVersion: number, + e2ePubKey: Uint8Array | null, + msg: Uint8Array, +): Uint8Array { + const cmNonce = crypto.getRandomValues(new Uint8Array(24)) + // paddedLen: e2eEncConfirmationLength (15904) for confirmations, e2eEncMessageLength (16000) for messages + // Protocol.hs:316-320 + const paddedLen = e2ePubKey !== null ? 15904 : 16000 + const cmEncBody = cbEncrypt(e2eDhSecret, cmNonce, msg, paddedLen) + const env: ClientMsgEnvelope = { + cmHeader: {phVersion: smpClientVersion, phE2ePubDhKey: e2ePubKey}, + cmNonce, + cmEncBody, + } + return encodeClientMsgEnvelope(env) +} + +// agentCbEncryptOnce (Client.hs:2085-2095) +// Per-queue E2E encrypt with ephemeral DH key (for invitations). +export function agentCbEncryptOnce( + clientVersion: number, + dhRcvPubKey: Uint8Array, + msg: Uint8Array, +): Uint8Array { + const {publicKey: dhSndPubKey, privateKey: dhSndPrivKey} = generateX25519KeyPair() + const e2eDhSecret = dh(dhRcvPubKey, dhSndPrivKey) + return agentCbEncrypt(e2eDhSecret, clientVersion, dhSndPubKey, msg) +} + +// agentCbDecrypt (Client.hs:2099-2102) +export function agentCbDecrypt( + dhSecret: Uint8Array, + nonce: Uint8Array, + msg: Uint8Array, +): Uint8Array { + const result = cbDecrypt(dhSecret, nonce, msg) + if (result === null) throw new AgentError({tag: "AGENT", err: {tag: "A_CRYPTO", err: "DECRYPT_CB"}}) + return result +} + +// -- sendAgentMessage (Client.hs:1948-1952) +// Per-queue E2E encrypt message + SEND. +// sq is raw IDB SndQueue row. +export async function sendAgentMessage( + c: AgentClient, + sq: any, + msgFlags: {notification: boolean}, + agentMsg: Uint8Array, +): Promise { + const clientMsg: ClientMessage = {privHeader: {type: "PHEmpty"}, body: agentMsg} + const msg = agentCbEncrypt(sq.e2e_dh_secret, sq.smp_client_version, null, encodeClientMessage(clientMsg)) + const smp = await getClientForQueue(c, sq) + const privKey: AuthKey = {type: "ed25519", key: sq.snd_private_key} + await smp.sendMessage(privKey, sq.snd_id, msgFlags.notification, msg) +} + +// sendConfirmation (Client.hs:1788-1794) +export async function sendConfirmation( + c: AgentClient, + sq: any, + agentConfirmation: Uint8Array, +): Promise { + if (!sq.e2e_pub_key) throw new AgentError({tag: "INTERNAL", msg: "sendConfirmation: no e2e pub key"}) + const senderCanSecure_ = sq.queue_mode === "M" + const privHeader: PrivHeader = senderCanSecure_ + ? {type: "PHEmpty"} + : {type: "PHConfirmation", key: toPublicEd25519(sq.snd_private_key)} + const spKey: AuthKey | null = senderCanSecure_ ? {type: "ed25519", key: sq.snd_private_key} : null + const clientMsg: ClientMessage = {privHeader, body: agentConfirmation} + const msg = agentCbEncrypt(sq.e2e_dh_secret, sq.smp_client_version, sq.e2e_pub_key, encodeClientMessage(clientMsg)) + const smp = await getClientForQueue(c, sq) + await smp.sendMessage(spKey, sq.snd_id, true, msg) +} + +// secureQueue (Client.hs:1830-1833) +export async function secureQueue( + c: AgentClient, + rq: any, + senderKey: Uint8Array, +): Promise { + const smp = await getClientForQueue(c, rq) + await smp.secureQueue({type: "ed25519", key: rq.rcv_private_key}, rq.rcv_id, senderKey) +} + +// secureSndQueue (Client.hs:1835-1841) +export async function secureSndQueue( + c: AgentClient, + sq: any, +): Promise { + const smp = await getClientForQueue(c, sq) + await smp.secureSndQueue({type: "ed25519", key: sq.snd_private_key}, sq.snd_id) +} + +// sendAck (Client.hs:1904-1907) +export async function sendAck( + c: AgentClient, + rq: any, + msgId: Uint8Array, +): Promise { + const smp = await getClientForQueue(c, rq) + await smp.ackMessage({type: "ed25519", key: rq.rcv_private_key}, rq.rcv_id, msgId) +} + +// -- subscribeQueues (Client.hs:1543-1556) +export async function subscribeQueues( + c: AgentClient, + userId: number, + queues: RcvQueueSub[], +): Promise { + const byServer = new Map() + for (const q of queues) { + const list = byServer.get(q.server) ?? [] + list.push(q) + byServer.set(q.server, list) + } + for (const [server, qs] of byServer) { + c.currentSubs.batchAddPendingSubs(tSessKey(userId, server), qs) + } + for (const [server, qs] of byServer) { + await subscribeServerQueues(c, userId, server, qs) + } +} + +async function subscribeServerQueues( + c: AgentClient, + userId: number, + server: string, + queues: RcvQueueSub[], +): Promise { + const key = tSessKey(userId, server) + const existing = tryReadSessVar(key, c.smpClients as Map>) + if (!existing) return + const smp = existing.client + const subReqs = queues.map(q => ({rcvId: q.rcvId, privKey: {type: "ed25519" as const, key: q.rcvPrivateKey}})) + try { + await smp.subscribeQueues(subReqs) + c.currentSubs.batchAddActiveSubs(key, smp.sessionId, queues) + } catch { + // On error, subs stay pending + } +} + +// addNewQueueSubscription (Client.hs:1724-1728) +export function addNewQueueSubscription( + c: AgentClient, + rq: RcvQueueSub, + userId: number, + server: string, + sessionId: Uint8Array, +): void { + c.currentSubs.addActiveSub(tSessKey(userId, server), sessionId, rq) +} + +// -- newRcvQueue (Client.hs:1373-1435, simplified: no short links, no ntf credentials) + +export interface NewRcvQueueResult { + rcvQueue: any + sndId: Uint8Array + e2eDhKey: Uint8Array + sessionId: Uint8Array +} + +export async function newRcvQueue( + c: AgentClient, + userId: number, + connId: Uint8Array, + server: string, + keyHash: Uint8Array, + wsUrl: string, + subscribe: boolean, +): Promise { + const {publicKey: rcvPubKey, privateKey: rcvPrivateKey} = generateEd25519KeyPair() + const {publicKey: dhPubKey, privateKey: dhPrivKey} = generateX25519KeyPair() + const {publicKey: e2eDhKey, privateKey: e2ePrivKey} = generateX25519KeyPair() + + const smpConn = await getSMPServerClient(c, userId, server, keyHash, wsUrl) + const smp = smpConn.client + + const ids = await smp.createQueue({publicKey: rcvPubKey, privateKey: rcvPrivateKey}, dhPubKey, subscribe) + + const rcvDhSecret = dh(ids.srvDhKey, dhPrivKey) + const rcvQueue = { + host: server, port: "443", + rcv_id: ids.rcvId, + conn_id: connId, + rcv_private_key: rcvPrivateKey, + rcv_dh_secret: rcvDhSecret, + e2e_priv_key: e2ePrivKey, + e2e_dh_secret: null, + snd_id: ids.sndId, + snd_key: null, + status: "new", + smp_client_version: smp.smpVersion, + rcv_queue_id: 0, + rcv_primary: 1, + replace_rcv_queue_id: null, + queue_mode: ids.queueMode, + server_key_hash: keyHash, + last_broker_ts: null, + to_subscribe: subscribe ? 0 : 1, + deleted: 0, + } + + return {rcvQueue, sndId: ids.sndId, e2eDhKey, sessionId: smp.sessionId} +} + +// -- Helpers + +async function getClientForQueue(c: AgentClient, q: any): Promise { + const clients = c.smpClients as Map> + for (const [key, sv] of clients) { + if (sv.value && key.endsWith(":" + q.host)) return sv.value.client + } + throw new AgentError({tag: "INTERNAL", msg: "no SMP client for " + q.host}) +} + +// Ed25519 public key from 64-byte private key (NaCl convention: last 32 bytes) +function toPublicEd25519(privateKey: Uint8Array): Uint8Array { + return privateKey.slice(32, 64) +} + +function toHex(b: Uint8Array): string { + return Array.from(b, x => x.toString(16).padStart(2, "0")).join("") +} diff --git a/smp-web/tests/infra-test.ts b/smp-web/tests/infra-test.ts index 30f39d520..f3b29b52d 100644 --- a/smp-web/tests/infra-test.ts +++ b/smp-web/tests/infra-test.ts @@ -443,6 +443,77 @@ async function testOperationState() { throwWhenInactive(c) // should not throw } +// -- agentCbEncrypt / ClientMsgEnvelope round-trip tests + +import {agentCbEncrypt, agentCbDecrypt, agentCbEncryptOnce} from "../dist/agent/smp-ops.js" +import {decodeClientMsgEnvelope, decodeClientMessage, type ClientMsgEnvelope} from "../dist/protocol.js" +import {Decoder} from "@simplex-chat/xftp-web/dist/protocol/encoding.js" +import {cbDecrypt} from "@simplex-chat/xftp-web/dist/crypto/secretbox.js" +import {generateX25519KeyPair, dh} from "@simplex-chat/xftp-web/dist/crypto/keys.js" + +async function testAgentCbEncrypt() { + console.log(" agentCbEncrypt...") + + // Generate a DH secret (simulating queue E2E setup) + const {publicKey: rcvPub, privateKey: rcvPriv} = generateX25519KeyPair() + const {publicKey: sndPub, privateKey: sndPriv} = generateX25519KeyPair() + const dhSecret = dh(rcvPub, sndPriv) + const dhSecretRcv = dh(sndPub, rcvPriv) + // Both sides should compute the same secret + assertEq(hex(dhSecret), hex(dhSecretRcv), "DH secrets match") + + const plaintext = new Uint8Array([1, 2, 3, 4, 5]) + const smpVersion = 18 + + // Encrypt: agentCbEncrypt wraps plaintext in ClientMsgEnvelope + const envelope = agentCbEncrypt(dhSecret, smpVersion, null, plaintext) + assert(envelope.length > 0, "agentCbEncrypt produces output") + + // Decode the ClientMsgEnvelope + const d = new Decoder(envelope) + const env = decodeClientMsgEnvelope(d) + assertEq(env.cmHeader.phVersion, smpVersion, "envelope version matches") + assertEq(env.cmHeader.phE2ePubDhKey, null, "no pub key for message (not confirmation)") + assertEq(env.cmNonce.length, 24, "nonce is 24 bytes") + assert(env.cmEncBody.length > 0, "encrypted body is non-empty") + + // Decrypt with receiver's DH secret + const decrypted = cbDecrypt(dhSecretRcv, env.cmNonce, env.cmEncBody) + assert(decrypted !== null, "cbDecrypt succeeds") + // The decrypted content is the padded plaintext (with padding) + // First bytes should match our plaintext + let match = true + for (let i = 0; i < plaintext.length; i++) { + if (decrypted![i] !== plaintext[i]) { match = false; break } + } + assert(match, "decrypted content starts with plaintext") + + // Encrypt with e2ePubKey (confirmation mode — has DH pub key in header) + const envConf = agentCbEncrypt(dhSecret, smpVersion, sndPub, plaintext) + const d2 = new Decoder(envConf) + const env2 = decodeClientMsgEnvelope(d2) + assertEq(env2.cmHeader.phVersion, smpVersion, "confirmation envelope version") + assert(env2.cmHeader.phE2ePubDhKey !== null, "confirmation has pub key") + assertEq(env2.cmHeader.phE2ePubDhKey!.length, 32, "pub key is 32 bytes") + + // agentCbDecrypt + const decrypted2 = agentCbDecrypt(dhSecretRcv, env2.cmNonce, env2.cmEncBody) + for (let i = 0; i < plaintext.length; i++) { + if (decrypted2[i] !== plaintext[i]) { match = false; break } + } + assert(match, "agentCbDecrypt matches plaintext") + + // agentCbEncryptOnce — ephemeral DH + const envOnce = agentCbEncryptOnce(smpVersion, rcvPub, plaintext) + const d3 = new Decoder(envOnce) + const env3 = decodeClientMsgEnvelope(d3) + assert(env3.cmHeader.phE2ePubDhKey !== null, "encryptOnce has ephemeral pub key") + // Receiver can decrypt using their private key + sender's ephemeral pub key + const ephDhSecret = dh(env3.cmHeader.phE2ePubDhKey!, rcvPriv) + const decrypted3 = cbDecrypt(ephDhSecret, env3.cmNonce, env3.cmEncBody) + assert(decrypted3 !== null, "encryptOnce: receiver can decrypt") +} + // -- Run all async function main() { @@ -458,6 +529,7 @@ async function main() { await testLocking() await testServerSelection() await testOperationState() + await testAgentCbEncrypt() console.log(`\n${passed} passed, ${failed} failed`) if (failed > 0) process.exit(1) }