diff --git a/CONFIGURATION.md b/CONFIGURATION.md index d046945..696857d 100644 --- a/CONFIGURATION.md +++ b/CONFIGURATION.md @@ -106,6 +106,7 @@ Set by `sudo baudbot broker register` when using brokered Slack OAuth flow. | `SLACK_BROKER_SIGNING_PUBLIC_KEY` | Broker Ed25519 public signing key (base64) | | `SLACK_BROKER_POLL_INTERVAL_MS` | Inbox poll interval in milliseconds (default: `3000`) | | `SLACK_BROKER_MAX_MESSAGES` | Max leased messages per poll request (default: `10`) | +| `SLACK_BROKER_WAIT_SECONDS` | Long-poll wait window for `/api/inbox/pull` (default: `20`, set `0` for immediate short-poll, max `25`) | | `SLACK_BROKER_DEDUPE_TTL_MS` | Dedupe cache TTL in milliseconds (default: `1200000`) | ### Kernel (Cloud Browsers) @@ -190,6 +191,7 @@ SLACK_BROKER_URL=https://broker.example.com SLACK_BROKER_WORKSPACE_ID=T0123ABCD SLACK_BROKER_POLL_INTERVAL_MS=3000 SLACK_BROKER_MAX_MESSAGES=10 +SLACK_BROKER_WAIT_SECONDS=20 SLACK_BROKER_DEDUPE_TTL_MS=1200000 # Experimental features (required for email) diff --git a/README.md b/README.md index 1d67de1..82648a2 100644 --- a/README.md +++ b/README.md @@ -98,6 +98,8 @@ sudo baudbot broker register \ --registration-token ``` +Broker pull mode uses long-polling by default (`SLACK_BROKER_WAIT_SECONDS=20`, max `25`; set `0` for immediate short-poll behavior). + Need to rotate/update a key later? ```bash diff --git a/bin/config.sh b/bin/config.sh index 09e9920..1bf244f 100755 --- a/bin/config.sh +++ b/bin/config.sh @@ -451,6 +451,7 @@ else SLACK_BROKER_SIGNING_PUBLIC_KEY \ SLACK_BROKER_POLL_INTERVAL_MS \ SLACK_BROKER_MAX_MESSAGES \ + SLACK_BROKER_WAIT_SECONDS \ SLACK_BROKER_DEDUPE_TTL_MS prompt_secret "SLACK_BOT_TOKEN" \ @@ -613,6 +614,7 @@ ordered_keys=( SLACK_BROKER_SIGNING_PUBLIC_KEY SLACK_BROKER_POLL_INTERVAL_MS SLACK_BROKER_MAX_MESSAGES + SLACK_BROKER_WAIT_SECONDS SLACK_BROKER_DEDUPE_TTL_MS BAUDBOT_AGENT_USER BAUDBOT_AGENT_HOME diff --git a/slack-bridge/broker-bridge.mjs b/slack-bridge/broker-bridge.mjs index 41f8f2a..8258b7e 100755 --- a/slack-bridge/broker-bridge.mjs +++ b/slack-bridge/broker-bridge.mjs @@ -24,17 +24,32 @@ import { } from "./security.mjs"; import { canonicalizeEnvelope, - canonicalizeOutbound, + canonicalizeProtocolRequest, canonicalizeSendRequest, } from "./crypto.mjs"; const SOCKET_DIR = path.join(homedir(), ".pi", "session-control"); const AGENT_TIMEOUT_MS = 120_000; -const API_PORT = parseInt(process.env.BRIDGE_API_PORT || "7890", 10); -const POLL_INTERVAL_MS = parseInt(process.env.SLACK_BROKER_POLL_INTERVAL_MS || "3000", 10); -const MAX_MESSAGES = parseInt(process.env.SLACK_BROKER_MAX_MESSAGES || "10", 10); -const DEDUPE_TTL_MS = parseInt(process.env.SLACK_BROKER_DEDUPE_TTL_MS || String(20 * 60 * 1000), 10); + +function clampInt(value, min, max, fallback) { + const parsed = Number.parseInt(String(value ?? ""), 10); + if (!Number.isFinite(parsed)) return fallback; + return Math.min(max, Math.max(min, parsed)); +} + +const API_PORT = clampInt(process.env.BRIDGE_API_PORT || "7890", 0, 65535, 7890); +const POLL_INTERVAL_MS = clampInt(process.env.SLACK_BROKER_POLL_INTERVAL_MS || "3000", 0, 60_000, 3000); +const MAX_MESSAGES = clampInt(process.env.SLACK_BROKER_MAX_MESSAGES || "10", 1, 100, 10); +const MAX_WAIT_SECONDS = 25; +const BROKER_WAIT_SECONDS = clampInt(process.env.SLACK_BROKER_WAIT_SECONDS || "20", 0, MAX_WAIT_SECONDS, 20); +const DEDUPE_TTL_MS = clampInt( + process.env.SLACK_BROKER_DEDUPE_TTL_MS || String(20 * 60 * 1000), + 1_000, + 7 * 24 * 60 * 60 * 1000, + 20 * 60 * 1000, +); const MAX_BACKOFF_MS = 30_000; +const INBOX_PROTOCOL_VERSION = "2026-02-1"; const BROKER_HEALTH_PATH = path.join(homedir(), ".pi", "agent", "broker-health.json"); function ts() { @@ -351,12 +366,25 @@ function getThreadId(channel, threadTs) { return id; } -function signRequest(action, timestamp, payloadField) { - const canonical = canonicalizeOutbound(workspaceId, action, timestamp, payloadField); +function signProtocolRequest(action, timestamp, payload) { + const canonical = canonicalizeProtocolRequest( + workspaceId, + INBOX_PROTOCOL_VERSION, + action, + timestamp, + payload, + ); const sig = sodium.crypto_sign_detached(canonical, cryptoState.serverSignSecretKey); return toBase64(sig); } +function signPullRequest(timestamp, maxMessages, waitSeconds) { + return signProtocolRequest("inbox.pull", timestamp, { + max_messages: maxMessages, + wait_seconds: waitSeconds, + }); +} + async function brokerFetch(pathname, body) { const url = `${brokerBaseUrl}${pathname}`; const response = await fetch(url, { @@ -389,14 +417,18 @@ async function brokerFetch(pathname, body) { async function pullInbox() { const timestamp = Math.floor(Date.now() / 1000); - const signature = signRequest("inbox.pull", timestamp, String(MAX_MESSAGES)); + const signature = signPullRequest(timestamp, MAX_MESSAGES, BROKER_WAIT_SECONDS); - const payload = await brokerFetch("/api/inbox/pull", { + const body = { workspace_id: workspaceId, + protocol_version: INBOX_PROTOCOL_VERSION, max_messages: MAX_MESSAGES, + wait_seconds: BROKER_WAIT_SECONDS, timestamp, signature, - }); + }; + + const payload = await brokerFetch("/api/inbox/pull", body); return Array.isArray(payload.messages) ? payload.messages : []; } @@ -404,11 +436,11 @@ async function pullInbox() { async function ackInbox(messageIds) { if (messageIds.length === 0) return; const timestamp = Math.floor(Date.now() / 1000); - const joined = messageIds.join(","); - const signature = signRequest("inbox.ack", timestamp, joined); + const signature = signProtocolRequest("inbox.ack", timestamp, { message_ids: messageIds }); await brokerFetch("/api/inbox/ack", { workspace_id: workspaceId, + protocol_version: INBOX_PROTOCOL_VERSION, message_ids: messageIds, timestamp, signature, @@ -918,7 +950,9 @@ async function startPollLoop() { } backoffMs = POLL_INTERVAL_MS; - await sleep(POLL_INTERVAL_MS); + if (BROKER_WAIT_SECONDS <= 0) { + await sleep(POLL_INTERVAL_MS); + } } catch (err) { if (!pollSucceeded) { markHealth("poll", false, err); @@ -960,7 +994,11 @@ async function startPollLoop() { logInfo(` outbound mode: ${outboundMode} ${outboundMode === "direct" ? "(using SLACK_BOT_TOKEN)" : "(via broker)"}`); logInfo(` broker: ${brokerBaseUrl}`); logInfo(` workspace: ${workspaceId}`); - logInfo(` poll interval: ${POLL_INTERVAL_MS}ms, max messages: ${MAX_MESSAGES}`); + logInfo(` inbox protocol: ${INBOX_PROTOCOL_VERSION}`); + logInfo( + ` poll mode: ${BROKER_WAIT_SECONDS > 0 ? `long-poll (${BROKER_WAIT_SECONDS}s)` : "short-poll"}, ` + + `interval: ${POLL_INTERVAL_MS}ms, max messages: ${MAX_MESSAGES}`, + ); logInfo(` allowed users: ${ALLOWED_USERS.length || "all"}`); logInfo(` pi socket: ${socketPath || "(not found — will retry on message)"}`); await startPollLoop(); diff --git a/slack-bridge/crypto.mjs b/slack-bridge/crypto.mjs index ec22bfc..70aa4f1 100644 --- a/slack-bridge/crypto.mjs +++ b/slack-bridge/crypto.mjs @@ -46,6 +46,24 @@ export function canonicalizeOutbound(workspace, action, timestamp, encryptedBody return utf8Bytes(`${workspace}|${action}|${timestamp}|${encryptedBody}`); } +/** + * Construct canonical bytes for protocol-versioned inbox pull/ack signing. + * + * Uses deterministic JSON serialization (sorted keys) to match broker + * json-stable-stringify canonicalization. + */ +export function canonicalizeProtocolRequest(workspace, protocolVersion, action, timestamp, payload) { + return utf8Bytes( + stableStringify({ + workspace_id: workspace, + protocol_version: protocolVersion, + action, + timestamp, + payload, + }), + ); +} + /** * Construct canonical bytes for /api/send request signing. * diff --git a/slack-bridge/crypto.test.mjs b/slack-bridge/crypto.test.mjs index ee038a3..f3f2927 100644 --- a/slack-bridge/crypto.test.mjs +++ b/slack-bridge/crypto.test.mjs @@ -10,6 +10,7 @@ import { stableStringify, canonicalizeEnvelope, canonicalizeOutbound, + canonicalizeProtocolRequest, canonicalizeSendRequest, } from "./crypto.mjs"; @@ -151,6 +152,30 @@ describe("canonicalizeOutbound", () => { }); }); +// ── canonicalizeProtocolRequest ───────────────────────────────────────────── + +describe("canonicalizeProtocolRequest", () => { + it("produces stable JSON payload for protocol-versioned inbox.pull", () => { + const result = decode( + canonicalizeProtocolRequest("T123", "2026-02-1", "inbox.pull", 1700000000, { + max_messages: 10, + wait_seconds: 20, + }), + ); + assert.equal( + result, + '{"action":"inbox.pull","payload":{"max_messages":10,"wait_seconds":20},"protocol_version":"2026-02-1","timestamp":1700000000,"workspace_id":"T123"}', + ); + }); + + it("returns Uint8Array", () => { + const result = canonicalizeProtocolRequest("T123", "2026-02-1", "inbox.ack", 1700000000, { + message_ids: ["m1", "m2"], + }); + assert.ok(result instanceof Uint8Array); + }); +}); + // ── canonicalizeSendRequest ───────────────────────────────────────────────── describe("canonicalizeSendRequest", () => { diff --git a/test/broker-bridge.integration.test.mjs b/test/broker-bridge.integration.test.mjs index 7cb6c89..7208c64 100644 --- a/test/broker-bridge.integration.test.mjs +++ b/test/broker-bridge.integration.test.mjs @@ -7,7 +7,10 @@ import { fileURLToPath } from "node:url"; import { mkdtempSync, mkdirSync, rmSync } from "node:fs"; import { tmpdir } from "node:os"; import sodium from "libsodium-wrappers-sumo"; -import { canonicalizeEnvelope } from "../slack-bridge/crypto.mjs"; +import { + canonicalizeEnvelope, + canonicalizeProtocolRequest, +} from "../slack-bridge/crypto.mjs"; function b64(bytes = 32, fill = 1) { return Buffer.alloc(bytes, fill).toString("base64"); @@ -50,6 +53,8 @@ describe("broker pull bridge semi-integration", () => { }); it("acks poison messages from broker to avoid infinite retry loops", async () => { + await sodium.ready; + let pullCount = 0; let ackPayload = null; @@ -158,7 +163,16 @@ describe("broker pull bridge semi-integration", () => { await Promise.race([ackWait, bridgeExited]); expect(ackPayload.workspace_id).toBe("T123BROKER"); + expect(ackPayload.protocol_version).toBe("2026-02-1"); expect(ackPayload.message_ids).toContain("m-poison-1"); + + const signKeypair = sodium.crypto_sign_seed_keypair(new Uint8Array(Buffer.alloc(32, 13))); + const canonical = canonicalizeProtocolRequest("T123BROKER", "2026-02-1", "inbox.ack", ackPayload.timestamp, { + message_ids: ackPayload.message_ids, + }); + const sigBytes = new Uint8Array(Buffer.from(ackPayload.signature, "base64")); + const valid = sodium.crypto_sign_verify_detached(sigBytes, canonical, signKeypair.publicKey); + expect(valid).toBe(true); }); it("forwards user messages to agent in fire-and-forget mode without get_message/turn_end RPCs", async () => { @@ -347,4 +361,264 @@ describe("broker pull bridge semi-integration", () => { expect(sendPayloads.some((payload) => payload.action === "chat.postMessage")).toBe(false); expect(sendPayloads.some((payload) => payload.action === "reactions.add")).toBe(false); }); + + it("uses protocol-versioned inbox.pull signatures with wait_seconds by default", async () => { + await sodium.ready; + + const workspaceId = "T123BROKER"; + const signingSeed = Buffer.alloc(32, 21); + const signKeypair = sodium.crypto_sign_seed_keypair(new Uint8Array(signingSeed)); + let pullPayload = null; + + const broker = createServer(async (req, res) => { + if (req.method === "POST" && req.url === "/api/inbox/pull") { + let raw = ""; + for await (const chunk of req) raw += chunk; + pullPayload = JSON.parse(raw); + + res.writeHead(200, { "Content-Type": "application/json" }); + res.end(JSON.stringify({ ok: true, messages: [] })); + return; + } + + if (req.method === "POST" && req.url === "/api/inbox/ack") { + res.writeHead(200, { "Content-Type": "application/json" }); + res.end(JSON.stringify({ ok: true, acked: 0 })); + return; + } + + if (req.method === "POST" && req.url === "/api/send") { + res.writeHead(200, { "Content-Type": "application/json" }); + res.end(JSON.stringify({ ok: true, ts: "1234.5678" })); + return; + } + + res.writeHead(404, { "Content-Type": "application/json" }); + res.end(JSON.stringify({ ok: false, error: "not found" })); + }); + + await new Promise((resolve) => broker.listen(0, "127.0.0.1", resolve)); + servers.push(broker); + + const address = broker.address(); + if (!address || typeof address === "string") { + throw new Error("failed to get broker test server address"); + } + const brokerUrl = `http://127.0.0.1:${address.port}`; + + const testFileDir = path.dirname(fileURLToPath(import.meta.url)); + const repoRoot = path.dirname(testFileDir); + const bridgePath = path.join(repoRoot, "slack-bridge", "broker-bridge.mjs"); + const bridgeCwd = path.join(repoRoot, "slack-bridge"); + + const bridge = spawn("node", [bridgePath], { + cwd: bridgeCwd, + env: { + ...process.env, + SLACK_BROKER_URL: brokerUrl, + SLACK_BROKER_WORKSPACE_ID: workspaceId, + SLACK_BROKER_SERVER_PRIVATE_KEY: b64(32, 11), + SLACK_BROKER_SERVER_PUBLIC_KEY: b64(32, 12), + SLACK_BROKER_SERVER_SIGNING_PRIVATE_KEY: signingSeed.toString("base64"), + SLACK_BROKER_PUBLIC_KEY: b64(32, 14), + SLACK_BROKER_SIGNING_PUBLIC_KEY: b64(32, 15), + SLACK_ALLOWED_USERS: "U_ALLOWED", + SLACK_BROKER_POLL_INTERVAL_MS: "50", + BRIDGE_API_PORT: "0", + }, + stdio: ["ignore", "pipe", "pipe"], + }); + children.push(bridge); + + await waitFor(() => pullPayload !== null, 10_000, 50, "timeout waiting for inbox pull request"); + + expect(pullPayload.workspace_id).toBe(workspaceId); + expect(pullPayload.protocol_version).toBe("2026-02-1"); + expect(pullPayload.max_messages).toBe(10); + expect(pullPayload.wait_seconds).toBe(20); + + const canonical = canonicalizeProtocolRequest(workspaceId, "2026-02-1", "inbox.pull", pullPayload.timestamp, { + max_messages: 10, + wait_seconds: 20, + }); + const sigBytes = new Uint8Array(Buffer.from(pullPayload.signature, "base64")); + const valid = sodium.crypto_sign_verify_detached(sigBytes, canonical, signKeypair.publicKey); + expect(valid).toBe(true); + + bridge.kill("SIGTERM"); + }); + + it("uses protocol-versioned inbox.pull signature with wait_seconds=0", async () => { + await sodium.ready; + + const workspaceId = "T123BROKER"; + const signingSeed = Buffer.alloc(32, 22); + const signKeypair = sodium.crypto_sign_seed_keypair(new Uint8Array(signingSeed)); + let pullPayload = null; + + const broker = createServer(async (req, res) => { + if (req.method === "POST" && req.url === "/api/inbox/pull") { + let raw = ""; + for await (const chunk of req) raw += chunk; + pullPayload = JSON.parse(raw); + + res.writeHead(200, { "Content-Type": "application/json" }); + res.end(JSON.stringify({ ok: true, messages: [] })); + return; + } + + if (req.method === "POST" && req.url === "/api/inbox/ack") { + res.writeHead(200, { "Content-Type": "application/json" }); + res.end(JSON.stringify({ ok: true, acked: 0 })); + return; + } + + if (req.method === "POST" && req.url === "/api/send") { + res.writeHead(200, { "Content-Type": "application/json" }); + res.end(JSON.stringify({ ok: true, ts: "1234.5678" })); + return; + } + + res.writeHead(404, { "Content-Type": "application/json" }); + res.end(JSON.stringify({ ok: false, error: "not found" })); + }); + + await new Promise((resolve) => broker.listen(0, "127.0.0.1", resolve)); + servers.push(broker); + + const address = broker.address(); + if (!address || typeof address === "string") { + throw new Error("failed to get broker test server address"); + } + const brokerUrl = `http://127.0.0.1:${address.port}`; + + const testFileDir = path.dirname(fileURLToPath(import.meta.url)); + const repoRoot = path.dirname(testFileDir); + const bridgePath = path.join(repoRoot, "slack-bridge", "broker-bridge.mjs"); + const bridgeCwd = path.join(repoRoot, "slack-bridge"); + + const bridge = spawn("node", [bridgePath], { + cwd: bridgeCwd, + env: { + ...process.env, + SLACK_BROKER_URL: brokerUrl, + SLACK_BROKER_WORKSPACE_ID: workspaceId, + SLACK_BROKER_SERVER_PRIVATE_KEY: b64(32, 11), + SLACK_BROKER_SERVER_PUBLIC_KEY: b64(32, 12), + SLACK_BROKER_SERVER_SIGNING_PRIVATE_KEY: signingSeed.toString("base64"), + SLACK_BROKER_PUBLIC_KEY: b64(32, 14), + SLACK_BROKER_SIGNING_PUBLIC_KEY: b64(32, 15), + SLACK_ALLOWED_USERS: "U_ALLOWED", + SLACK_BROKER_POLL_INTERVAL_MS: "50", + SLACK_BROKER_WAIT_SECONDS: "0", + BRIDGE_API_PORT: "0", + }, + stdio: ["ignore", "pipe", "pipe"], + }); + children.push(bridge); + + await waitFor(() => pullPayload !== null, 10_000, 50, "timeout waiting for protocol inbox pull request"); + + expect(pullPayload.workspace_id).toBe(workspaceId); + expect(pullPayload.protocol_version).toBe("2026-02-1"); + expect(pullPayload.max_messages).toBe(10); + expect(pullPayload.wait_seconds).toBe(0); + + const canonical = canonicalizeProtocolRequest(workspaceId, "2026-02-1", "inbox.pull", pullPayload.timestamp, { + max_messages: 10, + wait_seconds: 0, + }); + const sigBytes = new Uint8Array(Buffer.from(pullPayload.signature, "base64")); + const valid = sodium.crypto_sign_verify_detached(sigBytes, canonical, signKeypair.publicKey); + expect(valid).toBe(true); + + bridge.kill("SIGTERM"); + }); + + it("clamps max_messages before signing pull requests", async () => { + await sodium.ready; + + const workspaceId = "T123BROKER"; + const signingSeed = Buffer.alloc(32, 23); + const signKeypair = sodium.crypto_sign_seed_keypair(new Uint8Array(signingSeed)); + let pullPayload = null; + + const broker = createServer(async (req, res) => { + if (req.method === "POST" && req.url === "/api/inbox/pull") { + let raw = ""; + for await (const chunk of req) raw += chunk; + pullPayload = JSON.parse(raw); + + res.writeHead(200, { "Content-Type": "application/json" }); + res.end(JSON.stringify({ ok: true, messages: [] })); + return; + } + + if (req.method === "POST" && req.url === "/api/inbox/ack") { + res.writeHead(200, { "Content-Type": "application/json" }); + res.end(JSON.stringify({ ok: true, acked: 0 })); + return; + } + + if (req.method === "POST" && req.url === "/api/send") { + res.writeHead(200, { "Content-Type": "application/json" }); + res.end(JSON.stringify({ ok: true, ts: "1234.5678" })); + return; + } + + res.writeHead(404, { "Content-Type": "application/json" }); + res.end(JSON.stringify({ ok: false, error: "not found" })); + }); + + await new Promise((resolve) => broker.listen(0, "127.0.0.1", resolve)); + servers.push(broker); + + const address = broker.address(); + if (!address || typeof address === "string") { + throw new Error("failed to get broker test server address"); + } + const brokerUrl = `http://127.0.0.1:${address.port}`; + + const testFileDir = path.dirname(fileURLToPath(import.meta.url)); + const repoRoot = path.dirname(testFileDir); + const bridgePath = path.join(repoRoot, "slack-bridge", "broker-bridge.mjs"); + const bridgeCwd = path.join(repoRoot, "slack-bridge"); + + const bridge = spawn("node", [bridgePath], { + cwd: bridgeCwd, + env: { + ...process.env, + SLACK_BROKER_URL: brokerUrl, + SLACK_BROKER_WORKSPACE_ID: workspaceId, + SLACK_BROKER_SERVER_PRIVATE_KEY: b64(32, 11), + SLACK_BROKER_SERVER_PUBLIC_KEY: b64(32, 12), + SLACK_BROKER_SERVER_SIGNING_PRIVATE_KEY: signingSeed.toString("base64"), + SLACK_BROKER_PUBLIC_KEY: b64(32, 14), + SLACK_BROKER_SIGNING_PUBLIC_KEY: b64(32, 15), + SLACK_ALLOWED_USERS: "U_ALLOWED", + SLACK_BROKER_POLL_INTERVAL_MS: "50", + SLACK_BROKER_MAX_MESSAGES: "999", + BRIDGE_API_PORT: "0", + }, + stdio: ["ignore", "pipe", "pipe"], + }); + children.push(bridge); + + await waitFor(() => pullPayload !== null, 10_000, 50, "timeout waiting for clamped inbox pull request"); + + expect(pullPayload.workspace_id).toBe(workspaceId); + expect(pullPayload.protocol_version).toBe("2026-02-1"); + expect(pullPayload.max_messages).toBe(100); + expect(pullPayload.wait_seconds).toBe(20); + + const canonical = canonicalizeProtocolRequest(workspaceId, "2026-02-1", "inbox.pull", pullPayload.timestamp, { + max_messages: 100, + wait_seconds: 20, + }); + const sigBytes = new Uint8Array(Buffer.from(pullPayload.signature, "base64")); + const valid = sodium.crypto_sign_verify_detached(sigBytes, canonical, signKeypair.publicKey); + expect(valid).toBe(true); + + bridge.kill("SIGTERM"); + }); });