From 5cf2201690e4a47b45714cfe330bc2853fd19cd0 Mon Sep 17 00:00:00 2001 From: Ben Vinegar Date: Sun, 22 Feb 2026 15:02:53 -0500 Subject: [PATCH 1/3] bridge: add broker long-poll v2 pull signing --- CONFIGURATION.md | 2 + README.md | 2 + bin/config.sh | 2 + slack-bridge/broker-bridge.mjs | 54 ++++- slack-bridge/crypto.mjs | 17 ++ slack-bridge/crypto.test.mjs | 26 +++ test/broker-bridge.integration.test.mjs | 260 +++++++++++++++++++++++- 7 files changed, 353 insertions(+), 10 deletions(-) diff --git a/CONFIGURATION.md b/CONFIGURATION.md index d046945..63c8a63 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 legacy 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..6213ae6 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 legacy short-poll). + 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..2cea6a9 100755 --- a/slack-bridge/broker-bridge.mjs +++ b/slack-bridge/broker-bridge.mjs @@ -25,15 +25,30 @@ import { import { canonicalizeEnvelope, canonicalizeOutbound, + canonicalizeOutboundV2, 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", 1, 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 BROKER_HEALTH_PATH = path.join(homedir(), ".pi", "agent", "broker-health.json"); @@ -357,6 +372,19 @@ function signRequest(action, timestamp, payloadField) { return toBase64(sig); } +function signPullRequest(timestamp, maxMessages, waitSeconds) { + if (waitSeconds <= 0) { + return signRequest("inbox.pull", timestamp, String(maxMessages)); + } + + const canonical = canonicalizeOutboundV2(workspaceId, "inbox.pull.v2", timestamp, { + max_messages: maxMessages, + wait_seconds: waitSeconds, + }); + const sig = sodium.crypto_sign_detached(canonical, cryptoState.serverSignSecretKey); + return toBase64(sig); +} + async function brokerFetch(pathname, body) { const url = `${brokerBaseUrl}${pathname}`; const response = await fetch(url, { @@ -389,14 +417,17 @@ 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, max_messages: MAX_MESSAGES, + ...(BROKER_WAIT_SECONDS > 0 ? { wait_seconds: BROKER_WAIT_SECONDS } : {}), timestamp, signature, - }); + }; + + const payload = await brokerFetch("/api/inbox/pull", body); return Array.isArray(payload.messages) ? payload.messages : []; } @@ -918,7 +949,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 +993,10 @@ 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( + ` 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..33ec37f 100644 --- a/slack-bridge/crypto.mjs +++ b/slack-bridge/crypto.mjs @@ -46,6 +46,23 @@ export function canonicalizeOutbound(workspace, action, timestamp, encryptedBody return utf8Bytes(`${workspace}|${action}|${timestamp}|${encryptedBody}`); } +/** + * Construct canonical bytes for v2 outbound request signing. + * + * Uses deterministic JSON serialization (sorted keys) to match broker + * json-stable-stringify canonicalization. + */ +export function canonicalizeOutboundV2(workspace, action, timestamp, payload) { + return utf8Bytes( + stableStringify({ + workspace_id: workspace, + 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..64c2a3b 100644 --- a/slack-bridge/crypto.test.mjs +++ b/slack-bridge/crypto.test.mjs @@ -10,6 +10,7 @@ import { stableStringify, canonicalizeEnvelope, canonicalizeOutbound, + canonicalizeOutboundV2, canonicalizeSendRequest, } from "./crypto.mjs"; @@ -151,6 +152,31 @@ describe("canonicalizeOutbound", () => { }); }); +// ── canonicalizeOutboundV2 ────────────────────────────────────────────────── + +describe("canonicalizeOutboundV2", () => { + it("produces stable JSON payload for inbox.pull.v2", () => { + const result = decode( + canonicalizeOutboundV2("T123", "inbox.pull.v2", 1700000000, { + max_messages: 10, + wait_seconds: 20, + }), + ); + assert.equal( + result, + '{"action":"inbox.pull.v2","payload":{"max_messages":10,"wait_seconds":20},"timestamp":1700000000,"workspace_id":"T123"}', + ); + }); + + it("returns Uint8Array", () => { + const result = canonicalizeOutboundV2("T123", "inbox.pull.v2", 1700000000, { + max_messages: 10, + wait_seconds: 20, + }); + 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..64deb74 100644 --- a/test/broker-bridge.integration.test.mjs +++ b/test/broker-bridge.integration.test.mjs @@ -7,7 +7,11 @@ 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, + canonicalizeOutbound, + canonicalizeOutboundV2, +} from "../slack-bridge/crypto.mjs"; function b64(bytes = 32, fill = 1) { return Buffer.alloc(bytes, fill).toString("base64"); @@ -347,4 +351,258 @@ 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 inbox.pull.v2 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.max_messages).toBe(10); + expect(pullPayload.wait_seconds).toBe(20); + + const canonical = canonicalizeOutboundV2(workspaceId, "inbox.pull.v2", 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("falls back to legacy inbox.pull signature when SLACK_BROKER_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 legacy inbox pull request"); + + expect(pullPayload.workspace_id).toBe(workspaceId); + expect(pullPayload.max_messages).toBe(10); + expect(Object.prototype.hasOwnProperty.call(pullPayload, "wait_seconds")).toBe(false); + + const canonical = canonicalizeOutbound(workspaceId, "inbox.pull", pullPayload.timestamp, "10"); + 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.max_messages).toBe(100); + expect(pullPayload.wait_seconds).toBe(20); + + const canonical = canonicalizeOutboundV2(workspaceId, "inbox.pull.v2", 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"); + }); }); From 22addf17b36cf4668bf7bac5f9b5adb9fdd02e7d Mon Sep 17 00:00:00 2001 From: Ben Vinegar Date: Sun, 22 Feb 2026 15:24:53 -0500 Subject: [PATCH 2/3] bridge: adopt protocol-versioned inbox pull/ack signing --- CONFIGURATION.md | 2 +- README.md | 2 +- slack-bridge/broker-bridge.mjs | 30 ++++++++++++---------- slack-bridge/crypto.mjs | 5 ++-- slack-bridge/crypto.test.mjs | 17 ++++++------- test/broker-bridge.integration.test.mjs | 34 ++++++++++++++++++------- 6 files changed, 54 insertions(+), 36 deletions(-) diff --git a/CONFIGURATION.md b/CONFIGURATION.md index 63c8a63..696857d 100644 --- a/CONFIGURATION.md +++ b/CONFIGURATION.md @@ -106,7 +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 legacy short-poll, max `25`) | +| `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) diff --git a/README.md b/README.md index 6213ae6..82648a2 100644 --- a/README.md +++ b/README.md @@ -98,7 +98,7 @@ sudo baudbot broker register \ --registration-token ``` -Broker pull mode uses long-polling by default (`SLACK_BROKER_WAIT_SECONDS=20`, max `25`; set `0` for legacy short-poll). +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? diff --git a/slack-bridge/broker-bridge.mjs b/slack-bridge/broker-bridge.mjs index 2cea6a9..8617995 100755 --- a/slack-bridge/broker-bridge.mjs +++ b/slack-bridge/broker-bridge.mjs @@ -24,8 +24,7 @@ import { } from "./security.mjs"; import { canonicalizeEnvelope, - canonicalizeOutbound, - canonicalizeOutboundV2, + canonicalizeProtocolRequest, canonicalizeSendRequest, } from "./crypto.mjs"; @@ -50,6 +49,7 @@ const DEDUPE_TTL_MS = clampInt( 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() { @@ -366,23 +366,23 @@ 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) { - if (waitSeconds <= 0) { - return signRequest("inbox.pull", timestamp, String(maxMessages)); - } - - const canonical = canonicalizeOutboundV2(workspaceId, "inbox.pull.v2", timestamp, { + return signProtocolRequest("inbox.pull", timestamp, { max_messages: maxMessages, wait_seconds: waitSeconds, }); - const sig = sodium.crypto_sign_detached(canonical, cryptoState.serverSignSecretKey); - return toBase64(sig); } async function brokerFetch(pathname, body) { @@ -421,8 +421,9 @@ async function pullInbox() { const body = { workspace_id: workspaceId, + protocol_version: INBOX_PROTOCOL_VERSION, max_messages: MAX_MESSAGES, - ...(BROKER_WAIT_SECONDS > 0 ? { wait_seconds: BROKER_WAIT_SECONDS } : {}), + wait_seconds: BROKER_WAIT_SECONDS, timestamp, signature, }; @@ -435,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, @@ -993,6 +994,7 @@ async function startPollLoop() { logInfo(` outbound mode: ${outboundMode} ${outboundMode === "direct" ? "(using SLACK_BOT_TOKEN)" : "(via broker)"}`); logInfo(` broker: ${brokerBaseUrl}`); logInfo(` workspace: ${workspaceId}`); + 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}`, diff --git a/slack-bridge/crypto.mjs b/slack-bridge/crypto.mjs index 33ec37f..70aa4f1 100644 --- a/slack-bridge/crypto.mjs +++ b/slack-bridge/crypto.mjs @@ -47,15 +47,16 @@ export function canonicalizeOutbound(workspace, action, timestamp, encryptedBody } /** - * Construct canonical bytes for v2 outbound request signing. + * 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 canonicalizeOutboundV2(workspace, action, timestamp, payload) { +export function canonicalizeProtocolRequest(workspace, protocolVersion, action, timestamp, payload) { return utf8Bytes( stableStringify({ workspace_id: workspace, + protocol_version: protocolVersion, action, timestamp, payload, diff --git a/slack-bridge/crypto.test.mjs b/slack-bridge/crypto.test.mjs index 64c2a3b..f3f2927 100644 --- a/slack-bridge/crypto.test.mjs +++ b/slack-bridge/crypto.test.mjs @@ -10,7 +10,7 @@ import { stableStringify, canonicalizeEnvelope, canonicalizeOutbound, - canonicalizeOutboundV2, + canonicalizeProtocolRequest, canonicalizeSendRequest, } from "./crypto.mjs"; @@ -152,26 +152,25 @@ describe("canonicalizeOutbound", () => { }); }); -// ── canonicalizeOutboundV2 ────────────────────────────────────────────────── +// ── canonicalizeProtocolRequest ───────────────────────────────────────────── -describe("canonicalizeOutboundV2", () => { - it("produces stable JSON payload for inbox.pull.v2", () => { +describe("canonicalizeProtocolRequest", () => { + it("produces stable JSON payload for protocol-versioned inbox.pull", () => { const result = decode( - canonicalizeOutboundV2("T123", "inbox.pull.v2", 1700000000, { + canonicalizeProtocolRequest("T123", "2026-02-1", "inbox.pull", 1700000000, { max_messages: 10, wait_seconds: 20, }), ); assert.equal( result, - '{"action":"inbox.pull.v2","payload":{"max_messages":10,"wait_seconds":20},"timestamp":1700000000,"workspace_id":"T123"}', + '{"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 = canonicalizeOutboundV2("T123", "inbox.pull.v2", 1700000000, { - max_messages: 10, - wait_seconds: 20, + const result = canonicalizeProtocolRequest("T123", "2026-02-1", "inbox.ack", 1700000000, { + message_ids: ["m1", "m2"], }); assert.ok(result instanceof Uint8Array); }); diff --git a/test/broker-bridge.integration.test.mjs b/test/broker-bridge.integration.test.mjs index 64deb74..7208c64 100644 --- a/test/broker-bridge.integration.test.mjs +++ b/test/broker-bridge.integration.test.mjs @@ -9,8 +9,7 @@ import { tmpdir } from "node:os"; import sodium from "libsodium-wrappers-sumo"; import { canonicalizeEnvelope, - canonicalizeOutbound, - canonicalizeOutboundV2, + canonicalizeProtocolRequest, } from "../slack-bridge/crypto.mjs"; function b64(bytes = 32, fill = 1) { @@ -54,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; @@ -162,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 () => { @@ -352,7 +362,7 @@ describe("broker pull bridge semi-integration", () => { expect(sendPayloads.some((payload) => payload.action === "reactions.add")).toBe(false); }); - it("uses inbox.pull.v2 signatures with wait_seconds by default", async () => { + it("uses protocol-versioned inbox.pull signatures with wait_seconds by default", async () => { await sodium.ready; const workspaceId = "T123BROKER"; @@ -423,10 +433,11 @@ describe("broker pull bridge semi-integration", () => { 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 = canonicalizeOutboundV2(workspaceId, "inbox.pull.v2", pullPayload.timestamp, { + const canonical = canonicalizeProtocolRequest(workspaceId, "2026-02-1", "inbox.pull", pullPayload.timestamp, { max_messages: 10, wait_seconds: 20, }); @@ -437,7 +448,7 @@ describe("broker pull bridge semi-integration", () => { bridge.kill("SIGTERM"); }); - it("falls back to legacy inbox.pull signature when SLACK_BROKER_WAIT_SECONDS=0", async () => { + it("uses protocol-versioned inbox.pull signature with wait_seconds=0", async () => { await sodium.ready; const workspaceId = "T123BROKER"; @@ -506,13 +517,17 @@ describe("broker pull bridge semi-integration", () => { }); children.push(bridge); - await waitFor(() => pullPayload !== null, 10_000, 50, "timeout waiting for legacy inbox pull request"); + 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(Object.prototype.hasOwnProperty.call(pullPayload, "wait_seconds")).toBe(false); + expect(pullPayload.wait_seconds).toBe(0); - const canonical = canonicalizeOutbound(workspaceId, "inbox.pull", pullPayload.timestamp, "10"); + 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); @@ -592,10 +607,11 @@ describe("broker pull bridge semi-integration", () => { 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 = canonicalizeOutboundV2(workspaceId, "inbox.pull.v2", pullPayload.timestamp, { + const canonical = canonicalizeProtocolRequest(workspaceId, "2026-02-1", "inbox.pull", pullPayload.timestamp, { max_messages: 100, wait_seconds: 20, }); From f19b29ad817d1f27c8bb63dfa8a950198792be00 Mon Sep 17 00:00:00 2001 From: Ben Vinegar Date: Sun, 22 Feb 2026 15:27:06 -0500 Subject: [PATCH 3/3] bridge: allow BRIDGE_API_PORT=0 for test ephemeral binds --- slack-bridge/broker-bridge.mjs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/slack-bridge/broker-bridge.mjs b/slack-bridge/broker-bridge.mjs index 8617995..8258b7e 100755 --- a/slack-bridge/broker-bridge.mjs +++ b/slack-bridge/broker-bridge.mjs @@ -37,7 +37,7 @@ function clampInt(value, min, max, fallback) { return Math.min(max, Math.max(min, parsed)); } -const API_PORT = clampInt(process.env.BRIDGE_API_PORT || "7890", 1, 65535, 7890); +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;