From 88a9a4cd722b01a36e4de33d72d9a8ed218673e1 Mon Sep 17 00:00:00 2001 From: Lily Shen <115414357+lilyshen0722@users.noreply.github.com> Date: Fri, 4 Sep 2026 21:53:02 -0700 Subject: [PATCH 01/11] feat(installables): add per-pod connector gates --- .../unit/routes/admin.installables.test.js | 127 ++++++++++++++++ .../unit/routes/installables.test.js | 42 +++++ .../routes/integrations.linkedUserId.test.js | 122 ++++++++++++++- .../routes/slack.webhook.signature.test.js | 46 ++++++ .../unit/routes/telegram.webhook.test.js | 99 ++++++++++++ .../scripts/migrate-connector-gates.test.js | 73 +++++++++ .../services/installableEventHandlers.test.js | 140 +++++++++++++++-- .../installableInstallationService.test.js | 99 +++++++++++- .../unit/services/slackBridgeService.test.js | 26 +++- .../services/telegramBridgeService.test.js | 61 +++++++- backend/models/InstallableInstallation.ts | 36 ++++- backend/models/Integration.ts | 56 ++++++- backend/routes/admin/installables.ts | 143 ++++++++++++++++++ backend/routes/installables.ts | 10 ++ backend/routes/integrations.ts | 60 +++++++- backend/routes/webhooks/slack.ts | 2 + backend/routes/webhooks/telegram.ts | 10 ++ backend/scripts/migrate-connector-gates.ts | 97 ++++++++++++ backend/server.ts | 2 + backend/services/connectorRelayPolicy.ts | 23 ++- backend/services/installable/eventHandlers.ts | 33 +++- .../installableInstallationService.ts | 37 ++++- .../installable/installableReconciler.ts | 90 ++++++++++- .../projectors/webhookProjector.ts | 7 + backend/services/slackBridgeService.ts | 18 ++- backend/services/telegramBridgeService.ts | 19 ++- 26 files changed, 1440 insertions(+), 38 deletions(-) create mode 100644 backend/__tests__/unit/routes/admin.installables.test.js create mode 100644 backend/__tests__/unit/scripts/migrate-connector-gates.test.js create mode 100644 backend/routes/admin/installables.ts create mode 100644 backend/scripts/migrate-connector-gates.ts diff --git a/backend/__tests__/unit/routes/admin.installables.test.js b/backend/__tests__/unit/routes/admin.installables.test.js new file mode 100644 index 0000000000..3b0391b422 --- /dev/null +++ b/backend/__tests__/unit/routes/admin.installables.test.js @@ -0,0 +1,127 @@ +const request = require('supertest'); +const express = require('express'); + +jest.mock('../../../middleware/auth', () => (req, _res, next) => { + req.user = { id: req.header('x-test-user') || 'admin-1' }; + next(); +}); +jest.mock('../../../middleware/adminAuth', () => (req, res, next) => { + if (req.header('x-test-admin') !== 'true') return res.status(403).json({ error: 'Admin access required' }); + return next(); +}); +jest.mock('../../../models/InstallableInstallation', () => ({ + findOne: jest.fn(), + findOneAndUpdate: jest.fn(), +})); +jest.mock('../../../models/Integration', () => ({ updateMany: jest.fn() })); + +const InstallableInstallation = require('../../../models/InstallableInstallation'); +const Integration = require('../../../models/Integration'); +const adminInstallableRoutes = require('../../../routes/admin/installables'); + +const app = express(); +app.use(express.json()); +app.use('/api/admin/installables', adminInstallableRoutes); + +const installableId = 'telegram'; +const installationId = '64b64c48c4f37a6b2f34c222'; +const admin = { 'x-test-admin': 'true' }; +const active = { + _id: installationId, + installableId, + targetType: 'user', + scope: 'user', + status: 'active', + installedBy: 'owner-1', +}; + +describe('admin installable pause/resume', () => { + beforeEach(() => jest.clearAllMocks()); + + it('requires the admin middleware before it can stop a private connector', async () => { + const res = await request(app) + .post(`/api/admin/installables/${installableId}/installations/${installationId}/pause`) + .send({ reason: 'Safety review' }); + + expect(res.status).toBe(403); + expect(InstallableInstallation.findOneAndUpdate).not.toHaveBeenCalled(); + }); + + it('requires a reason before any parent or projection write', async () => { + const res = await request(app) + .post(`/api/admin/installables/${installableId}/installations/${installationId}/pause`) + .set(admin) + .send({}); + + expect(res.status).toBe(400); + expect(InstallableInstallation.findOne).not.toHaveBeenCalled(); + expect(InstallableInstallation.findOneAndUpdate).not.toHaveBeenCalled(); + expect(Integration.updateMany).not.toHaveBeenCalled(); + }); + + it('pauses the parent before stamping its inbound projection', async () => { + const paused = { ...active, status: 'paused' }; + InstallableInstallation.findOne.mockResolvedValue(active); + InstallableInstallation.findOneAndUpdate.mockResolvedValue(paused); + Integration.updateMany.mockResolvedValue({ modifiedCount: 1 }); + + const res = await request(app) + .post(`/api/admin/installables/${installableId}/installations/${installationId}/pause`) + .set(admin) + .send({ reason: 'Safety review' }); + + expect(res.status).toBe(200); + expect(res.body).toMatchObject({ status: 'paused', projected: true }); + expect(InstallableInstallation.findOneAndUpdate).toHaveBeenCalledWith( + expect.objectContaining({ status: 'active', targetType: 'user', scope: 'user' }), + expect.objectContaining({ + $set: expect.objectContaining({ status: 'paused', adminPause: expect.objectContaining({ reason: 'Safety review' }) }), + $push: expect.objectContaining({ pauseAudit: expect.objectContaining({ action: 'pause', ownerId: 'owner-1' }) }), + }), + { new: true }, + ); + expect(Integration.updateMany).toHaveBeenCalledWith( + { installationId }, + expect.objectContaining({ $set: expect.objectContaining({ 'config.adminPause': expect.objectContaining({ reason: 'Safety review' }) }) }), + ); + }); + + it('reports a partial pause when child projection fails after the parent stopped', async () => { + InstallableInstallation.findOne.mockResolvedValue(active); + InstallableInstallation.findOneAndUpdate.mockResolvedValue({ ...active, status: 'paused' }); + Integration.updateMany.mockRejectedValue(new Error('db unavailable')); + + const res = await request(app) + .post(`/api/admin/installables/${installableId}/installations/${installationId}/pause`) + .set(admin) + .send({ reason: 'Safety review' }); + + expect(res.status).toBe(202); + expect(res.body).toMatchObject({ status: 'paused', projected: false }); + expect(InstallableInstallation.findOneAndUpdate).toHaveBeenCalledTimes(1); + }); + + it('clears children before resuming and leaves the parent paused if its completion CAS loses', async () => { + const paused = { ...active, status: 'paused' }; + InstallableInstallation.findOne.mockResolvedValue(paused); + Integration.updateMany.mockResolvedValue({ modifiedCount: 1 }); + InstallableInstallation.findOneAndUpdate.mockResolvedValue(null); + + const res = await request(app) + .post(`/api/admin/installables/${installableId}/installations/${installationId}/resume`) + .set(admin) + .send({ reason: 'Review complete' }); + + expect(res.status).toBe(202); + expect(res.body).toMatchObject({ status: 'paused', projected: false }); + expect(Integration.updateMany).toHaveBeenCalledWith( + { installationId }, + { $unset: { 'config.adminPause': 1 } }, + ); + expect(InstallableInstallation.findOneAndUpdate).toHaveBeenCalledWith( + expect.objectContaining({ status: 'paused' }), + expect.objectContaining({ $set: expect.objectContaining({ status: 'active' }) }), + { new: true }, + ); + }); +}); diff --git a/backend/__tests__/unit/routes/installables.test.js b/backend/__tests__/unit/routes/installables.test.js index 19cb7bcc58..cf3d6a348d 100644 --- a/backend/__tests__/unit/routes/installables.test.js +++ b/backend/__tests__/unit/routes/installables.test.js @@ -26,6 +26,10 @@ jest.mock('../../../services/installable/installableInstallationService', () => this.message = 'This install is still in progress; try again shortly.'; this.boundPodId = boundPodId; }, + InstallationPausedError: function InstallationPausedError(reason) { + this.message = 'This connector is paused by an administrator.'; + this.reason = reason; + }, })); const Pod = require('../../../models/Pod'); @@ -186,4 +190,42 @@ describe('installable connector routes', () => { expect(res.status).toBe(202); expect(res.body.status).toBe('uninstalling'); }); + + it('returns the paused refusal for install without claiming a replacement', async () => { + Pod.findById.mockResolvedValue({ + _id: podId, + createdBy: { toString: () => 'someone-else' }, + members: ['64b64c48c4f37a6b2f34c111'], + }); + installationService.install.mockRejectedValue( + new installationService.InstallationPausedError('Safety review in progress.'), + ); + + const res = await request(app) + .post('/api/installables/telegram/install') + .set(auth) + .send({ podId }); + + expect(res.status).toBe(409); + expect(res.body).toMatchObject({ + code: 'installation_paused', + reason: 'Safety review in progress.', + }); + }); + + it('returns the paused refusal for uninstall instead of a successful no-op', async () => { + installationService.uninstall.mockRejectedValue( + new installationService.InstallationPausedError('Safety review in progress.'), + ); + + const res = await request(app) + .delete('/api/installables/telegram/install') + .set(auth); + + expect(res.status).toBe(409); + expect(res.body).toMatchObject({ + code: 'installation_paused', + reason: 'Safety review in progress.', + }); + }); }); diff --git a/backend/__tests__/unit/routes/integrations.linkedUserId.test.js b/backend/__tests__/unit/routes/integrations.linkedUserId.test.js index 16215dac9a..5748fc6d01 100644 --- a/backend/__tests__/unit/routes/integrations.linkedUserId.test.js +++ b/backend/__tests__/unit/routes/integrations.linkedUserId.test.js @@ -8,7 +8,7 @@ const request = require('supertest'); const express = require('express'); jest.mock('../../../middleware/auth', () => (req, res, next) => { - req.user = { id: 'user-1' }; + req.user = { id: req.header('x-test-user') || 'user-1' }; next(); }); jest.mock('../../../middleware/adminAuth', () => (req, res, next) => next()); @@ -129,6 +129,126 @@ describe('PATCH /api/integrations/:id — linkedUserId guard', () => { }); }); +describe('PATCH /api/integrations/:id — user-scoped connector gates', () => { + const currentPodId = '64b64c1f7e5b8f0a12345670'; + const allowedPodId = '64b64c1f7e5b8f0a12345671'; + const forbiddenPodId = '64b64c1f7e5b8f0a12345672'; + + const userScopedIntegration = () => ({ + _id: 'integration-user', + scope: 'user', + type: 'telegram', + podId: currentPodId, + createdBy: { toString: () => 'user-1' }, + config: { + linkedUserId: 'user-1', + gates: { [currentPodId]: { enabled: true, since: new Date() } }, + toObject() { + return { + linkedUserId: 'user-1', + gates: { [currentPodId]: { enabled: true, since: new Date() } }, + }; + }, + }, + }); + + beforeEach(() => { + jest.clearAllMocks(); + Integration.findById.mockResolvedValue(userScopedIntegration()); + Integration.findByIdAndUpdate.mockResolvedValue({ _id: 'integration-user' }); + }); + + it('refuses a pod creator or admin who is not the linked owner', async () => { + const podCreator = await request(app) + .patch('/api/integrations/integration-user') + .set('x-test-user', 'pod-creator') + .send({ config: { liveRelay: false } }); + expect(podCreator.status).toBe(403); + + const admin = await request(app) + .patch('/api/integrations/integration-user') + .set('x-test-user', 'instance-admin') + .send({ config: { gates: {} } }); + expect(admin.status).toBe(403); + expect(Integration.findByIdAndUpdate).not.toHaveBeenCalled(); + }); + + it('checks every requested gate before writing any of them', async () => { + Pod.findById + .mockResolvedValueOnce({ _id: allowedPodId, createdBy: 'user-1', members: [] }) + .mockResolvedValueOnce({ _id: forbiddenPodId, createdBy: 'someone-else', members: [] }); + + const res = await request(app) + .patch('/api/integrations/integration-user') + .send({ + config: { + gates: { + [allowedPodId]: { enabled: true, since: new Date().toISOString() }, + [forbiddenPodId]: { enabled: true, since: new Date().toISOString() }, + }, + }, + }); + + expect(res.status).toBe(403); + expect(Integration.findByIdAndUpdate).not.toHaveBeenCalled(); + }); + + it('allows the linked owner to write gates for pods they belong to', async () => { + Pod.findById.mockResolvedValue({ _id: allowedPodId, createdBy: 'user-1', members: [] }); + + const res = await request(app) + .patch('/api/integrations/integration-user') + .send({ config: { gates: { [allowedPodId]: { enabled: true, since: new Date().toISOString() } } } }); + + expect(res.status).toBe(200); + const [, update] = Integration.findByIdAndUpdate.mock.calls[0]; + expect(update.config.gates[allowedPodId]).toMatchObject({ enabled: true }); + }); + + it.each(['a.b', '$ne'])('refuses an unsafe gate key before any membership lookup or write', async (key) => { + const res = await request(app) + .patch('/api/integrations/integration-user') + .send({ config: { gates: { [key]: { enabled: true } } } }); + + expect(res.status).toBe(400); + expect(Pod.findById).not.toHaveBeenCalled(); + expect(Integration.findByIdAndUpdate).not.toHaveBeenCalled(); + }); + + it('refuses an oversized gate body before any membership lookup or write', async () => { + const gates = Object.fromEntries(Array.from({ length: 101 }, (_, index) => [ + index.toString(16).padStart(24, '0'), { enabled: true }, + ])); + const res = await request(app) + .patch('/api/integrations/integration-user') + .send({ config: { gates } }); + + expect(res.status).toBe(400); + expect(Pod.findById).not.toHaveBeenCalled(); + expect(Integration.findByIdAndUpdate).not.toHaveBeenCalled(); + }); + + it('rejects a syntactically valid gate key whose pod does not exist', async () => { + Pod.findById.mockResolvedValue(null); + const missingPodId = '64b64c1f7e5b8f0a12345673'; + const res = await request(app) + .patch('/api/integrations/integration-user') + .send({ config: { gates: { [missingPodId]: { enabled: true } } } }); + + expect(res.status).toBe(403); + expect(Integration.findByIdAndUpdate).not.toHaveBeenCalled(); + }); + + it('refuses client writes to the administrator pause projection', async () => { + const res = await request(app) + .patch('/api/integrations/integration-user') + .send({ config: { adminPause: { reason: 'not owner controlled' } } }); + + expect(res.status).toBe(400); + expect(Integration.findByIdAndUpdate).not.toHaveBeenCalled(); + }); +}); + // POST /api/integrations — the create path had none of the PATCH guards: // config was spread verbatim (linkedUserId, connectCode, chatId all client- // settable) and there was no pod-membership check (ADR-025 review, 2026-08-26). diff --git a/backend/__tests__/unit/routes/slack.webhook.signature.test.js b/backend/__tests__/unit/routes/slack.webhook.signature.test.js index 90525833a2..422022b564 100644 --- a/backend/__tests__/unit/routes/slack.webhook.signature.test.js +++ b/backend/__tests__/unit/routes/slack.webhook.signature.test.js @@ -64,4 +64,50 @@ describe('installable Slack webhook signature and acknowledgement', () => { })); expect(receipts.markDone).toHaveBeenCalledWith('Ev1'); }); + + test('acks a paused connector event, records its receipt, and relays again after resume', async () => { + receipts.claim + .mockResolvedValueOnce('claimed') + .mockResolvedValueOnce('duplicate') + .mockResolvedValueOnce('claimed'); + receipts.markDone.mockResolvedValue(undefined); + const pausedIntegration = { + _id: 'integration-paused', type: 'slack', isActive: true, status: 'connected', + config: { + teamId: 'T1', chatId: 'D1', chatType: 'im', liveRelay: true, + adminPause: { reason: 'Safety review', at: new Date(), adminId: 'admin-1' }, + }, + }; + const resumedIntegration = { ...pausedIntegration, config: { ...pausedIntegration.config } }; + delete resumedIntegration.config.adminPause; + let paused = true; + Integration.findOne.mockImplementation((query) => ({ + lean: jest.fn().mockResolvedValue( + query['config.adminPause']?.$exists === false && paused ? null : resumedIntegration, + ), + })); + const bridge = require('../../../services/slackBridgeService'); + bridge.relaySlackMessageToPod.mockResolvedValue({ relayed: true }); + + const first = await request(app).post('/api/webhooks/slack/events').set(signatureHeaders(eventBody)).send(eventBody); + await new Promise((resolve) => setImmediate(resolve)); + const second = await request(app).post('/api/webhooks/slack/events').set(signatureHeaders(eventBody)).send(eventBody); + paused = false; // admin resume clears config.adminPause on the projection. + const resumedBody = { ...eventBody, event_id: 'Ev2' }; + const resumed = await request(app).post('/api/webhooks/slack/events').set(signatureHeaders(resumedBody)).send(resumedBody); + await new Promise((resolve) => setImmediate(resolve)); + + expect(first.status).toBe(200); + expect(second.status).toBe(200); + expect(resumed.status).toBe(200); + expect(Integration.findOne).toHaveBeenCalledWith(expect.objectContaining({ + 'config.adminPause': { $exists: false }, + })); + expect(receipts.markDone).toHaveBeenCalledWith('Ev1'); + expect(receipts.markDone).toHaveBeenCalledWith('Ev2'); + expect(bridge.relaySlackMessageToPod).toHaveBeenCalledTimes(1); + expect(bridge.relaySlackMessageToPod).toHaveBeenCalledWith(expect.objectContaining({ + integration: resumedIntegration, + })); + }); }); diff --git a/backend/__tests__/unit/routes/telegram.webhook.test.js b/backend/__tests__/unit/routes/telegram.webhook.test.js index fac989cccc..cd61a404b8 100644 --- a/backend/__tests__/unit/routes/telegram.webhook.test.js +++ b/backend/__tests__/unit/routes/telegram.webhook.test.js @@ -140,6 +140,33 @@ describe('Telegram webhook routes', () => { ); }); + it.each(['/summary', '/pod_summary', '/status'])('fails closed for %s when a user connector has no active pod', async (command) => { + const integration = { + _id: 'integration-no-pod', + scope: 'user', + config: { chatId: '42', chatType: 'private', messageBuffer: [{ content: 'hello' }] }, + }; + Integration.findOne.mockResolvedValue(integration); + + const res = await request(app) + .post('/api/webhooks/telegram') + .send({ + message: { + text: command, + chat: { id: 42, type: 'private' }, + from: { id: 7, first_name: 'Sam' }, + }, + }); + + expect(res.status).toBe(200); + expect(IntegrationSummaryService.createSummary).not.toHaveBeenCalled(); + expect(Summary.findOne).not.toHaveBeenCalled(); + expect(Pod.findById).not.toHaveBeenCalled(); + expect(telegramService.sendMessage).toHaveBeenCalledWith( + 'bot-token', '42', expect.stringContaining('no active pod'), + ); + }); + it('buffers non-command messages via provider', async () => { const integration = { _id: 'integration-1', @@ -262,6 +289,18 @@ describe('bridge command surface (/mode /status /mute /help)', () => { expect(until).toBeGreaterThan(Date.now()); }); + it('/unmute only clears the owner mute and cannot clear an admin pause', async () => { + const res = await post('/unmute'); + + expect(res.status).toBe(200); + expect(Integration.findByIdAndUpdate).toHaveBeenCalledWith( + 'integration-1', + { $unset: { 'config.relayMutedUntil': '' } }, + ); + const [, update] = Integration.findByIdAndUpdate.mock.calls[0]; + expect(update.$unset['config.adminPause']).toBeUndefined(); + }); + it('/help answers without an integration lookup dependency', async () => { Integration.findOne = jest.fn().mockResolvedValue(null); const res = await post('/help'); @@ -331,6 +370,66 @@ describe('bridge command surface (/mode /status /mute /help)', () => { expect(bridge.relayTelegramMessageToPod).not.toHaveBeenCalled(); }); + it('acks an update for a paused connector and posts nothing, then relays after resume', async () => { + const duplicate = Object.assign(new Error('duplicate'), { code: 11000 }); + WebhookDelivery.create + .mockResolvedValueOnce({}) + .mockRejectedValueOnce(duplicate) + .mockResolvedValueOnce({}); + const pausedIntegration = { + _id: 'integration-paused', type: 'telegram', podId: 'pod-1', isActive: true, + config: { + chatId: '42', chatType: 'private', liveRelay: true, linkedUserId: 'user-1', + adminPause: { reason: 'Safety review', at: new Date(), adminId: 'admin-1' }, + }, + }; + const resumedIntegration = { + ...pausedIntegration, + config: { ...pausedIntegration.config }, + }; + delete resumedIntegration.config.adminPause; + let paused = true; + // The paused projection is intentionally invisible to this query. That + // preserves Telegram's 200 acknowledgement and receipt semantics while + // ensuring no bridge/pod write happens before a later resume. + Integration.findOne = jest.fn().mockImplementation((query) => ( + query['config.adminPause']?.$exists === false && paused + ? Promise.resolve(null) + : Promise.resolve(resumedIntegration) + )); + bridge.relayTelegramMessageToPod.mockResolvedValue({ relayed: true }); + delete process.env.TELEGRAM_WEBHOOK_ALLOW_UNVERIFIED; + process.env.TELEGRAM_SECRET_TOKEN = 'pause-secret'; + + const first = await request(app) + .post('/api/webhooks/telegram') + .set('x-telegram-bot-api-secret-token', 'pause-secret') + .send(liveUpdate(33)); + const second = await request(app) + .post('/api/webhooks/telegram') + .set('x-telegram-bot-api-secret-token', 'pause-secret') + .send(liveUpdate(33)); + paused = false; // admin resume clears the child projection. + const resumed = await request(app) + .post('/api/webhooks/telegram') + .set('x-telegram-bot-api-secret-token', 'pause-secret') + .send(liveUpdate(34)); + + expect(first.status).toBe(200); + expect(second.status).toBe(200); + expect(resumed.status).toBe(200); + expect(WebhookDelivery.create).toHaveBeenCalledWith(expect.objectContaining({ + provider: 'telegram', deliveryId: '33', + })); + expect(Integration.findOne).toHaveBeenCalledWith(expect.objectContaining({ + 'config.adminPause': { $exists: false }, + })); + expect(bridge.relayTelegramMessageToPod).toHaveBeenCalledTimes(1); + expect(bridge.relayTelegramMessageToPod).toHaveBeenCalledWith(expect.objectContaining({ + integration: resumedIntegration, + })); + }); + it('releases the claim when processing throws, so redelivery retries', async () => { const integration = { _id: 'integration-1', diff --git a/backend/__tests__/unit/scripts/migrate-connector-gates.test.js b/backend/__tests__/unit/scripts/migrate-connector-gates.test.js new file mode 100644 index 0000000000..7c45c24868 --- /dev/null +++ b/backend/__tests__/unit/scripts/migrate-connector-gates.test.js @@ -0,0 +1,73 @@ +// @ts-nocheck + +// The migration suite needs testUtils' Mongo harness, not JWT behavior. +jest.mock('jsonwebtoken', () => ({})); + +const mongoose = require('mongoose'); + +const Integration = require('../../../models/Integration'); +const { migrateConnectorGates } = require('../../../scripts/migrate-connector-gates'); +const { + setupMongoDb, + closeMongoDb, + clearMongoDb, +} = require('../../utils/testUtils'); + +const id = () => new mongoose.Types.ObjectId().toString(); + +describe('migrate-connector-gates', () => { + beforeAll(async () => setupMongoDb()); + afterAll(async () => closeMongoDb()); + beforeEach(async () => clearMongoDb()); + + it('moves only installable projections to user scope with their initial gate', async () => { + const ownerId = id(); + const podId = id(); + const createdAt = new Date('2026-09-01T00:00:00.000Z'); + const projection = await Integration.create({ + installationId: id(), + podId, + type: 'telegram', + status: 'pending', + createdBy: ownerId, + isActive: true, + config: { liveRelay: true, linkedUserId: ownerId }, + createdAt, + updatedAt: createdAt, + }); + const legacy = await Integration.create({ + podId: id(), + type: 'telegram', + status: 'pending', + createdBy: ownerId, + isActive: true, + config: { liveRelay: true }, + }); + + await expect(migrateConnectorGates()).resolves.toEqual({ scanned: 1, migrated: 1, skipped: 0 }); + const migrated = await Integration.findById(projection._id); + const unchangedLegacy = await Integration.findById(legacy._id); + expect(migrated.scope).toBe('user'); + expect(migrated.config.gates.get(podId)).toMatchObject({ enabled: true, since: createdAt }); + expect(unchangedLegacy.scope).toBe('pod'); + expect(unchangedLegacy.config.gates).toBeUndefined(); + }); + + it('is idempotent and dry-run writes nothing', async () => { + const ownerId = id(); + const podId = id(); + const projection = await Integration.create({ + installationId: id(), podId, type: 'slack', status: 'pending', createdBy: ownerId, + isActive: true, config: { liveRelay: true, linkedUserId: ownerId }, + }); + + await expect(migrateConnectorGates({ dryRun: true })) + .resolves.toEqual({ scanned: 1, migrated: 1, skipped: 0 }); + expect((await Integration.findById(projection._id)).scope).toBe('pod'); + expect((await Integration.findById(projection._id)).config.gates).toBeUndefined(); + + await expect(migrateConnectorGates()).resolves.toEqual({ scanned: 1, migrated: 1, skipped: 0 }); + await expect(migrateConnectorGates()).resolves.toEqual({ scanned: 0, migrated: 0, skipped: 0 }); + expect((await Integration.findById(projection._id)).config.gates.get(podId)).toMatchObject({ enabled: true }); + }); +}); diff --git a/backend/__tests__/unit/services/installableEventHandlers.test.js b/backend/__tests__/unit/services/installableEventHandlers.test.js index fd25a63dfa..00f70d4a7e 100644 --- a/backend/__tests__/unit/services/installableEventHandlers.test.js +++ b/backend/__tests__/unit/services/installableEventHandlers.test.js @@ -1,10 +1,14 @@ // @ts-nocheck +// This suite does not exercise JWTs; keep testUtils loadable on Node 26. +jest.mock('jsonwebtoken', () => ({})); + const mongoose = require('mongoose'); const Installable = require('../../../models/Installable'); const InstallableInstallation = require('../../../models/InstallableInstallation'); const Integration = require('../../../models/Integration'); +const Pod = require('../../../models/Pod'); const { install, } = require('../../../services/installable/installableInstallationService'); @@ -22,6 +26,14 @@ const { const freshId = () => new mongoose.Types.ObjectId().toString(); +const createPod = async (podId, ownerId, memberIds = []) => Pod.create({ + _id: podId, + name: `Pod ${podId.slice(-4)}`, + type: 'team', + createdBy: ownerId, + members: memberIds, +}); + describe('installable event dispatcher', () => { let originalTelegramHandler; let originalSlackHandler; @@ -57,8 +69,12 @@ describe('installable event dispatcher', () => { it('selects only the event pod connector before invoking its handler', async () => { const podA = freshId(); const podB = freshId(); - await install({ installableId: 'telegram', installedBy: freshId(), podId: podA }); - await install({ installableId: 'telegram', installedBy: freshId(), podId: podB }); + const ownerA = freshId(); + const ownerB = freshId(); + await createPod(podA, ownerA); + await createPod(podB, ownerB); + await install({ installableId: 'telegram', installedBy: ownerA, podId: podA }); + await install({ installableId: 'telegram', installedBy: ownerB, podId: podB }); const relay = jest.fn().mockResolvedValue(undefined); eventHandlers['telegram.relay'] = relay; @@ -84,15 +100,23 @@ describe('installable event dispatcher', () => { it('relays each same-pod connector to its own selected Telegram chat', async () => { const podId = freshId(); - const first = await install({ installableId: 'telegram', installedBy: freshId(), podId }); - const second = await install({ installableId: 'telegram', installedBy: freshId(), podId }); + const firstOwner = freshId(); + const secondOwner = freshId(); + await createPod(podId, firstOwner, [secondOwner]); + const first = await install({ installableId: 'telegram', installedBy: firstOwner, podId }); + const second = await install({ installableId: 'telegram', installedBy: secondOwner, podId }); await Integration.updateOne( { _id: first.integration._id }, { $set: { 'config.chatId': 'chat-a', 'config.chatType': 'private' } }, ); await Integration.updateOne( { _id: second.integration._id }, - { $set: { 'config.chatId': 'chat-b', 'config.chatType': 'private' } }, + { + $set: { 'config.chatId': 'chat-b', 'config.chatType': 'private' }, + // User scope is allowed to omit its active inbound pod. Outbound must + // still fan out from the event pod through an enabled gate. + $unset: { podId: 1 }, + }, ); const previousToken = process.env.TELEGRAM_BOT_TOKEN; process.env.TELEGRAM_BOT_TOKEN = 'test-bot-token'; @@ -120,6 +144,53 @@ describe('installable event dispatcher', () => { } }); + it('selects a user connector only for an enabled gate whose owner remains a member', async () => { + const ownerId = freshId(); + const otherOwnerId = freshId(); + const firstPodId = freshId(); + const secondPodId = freshId(); + await createPod(firstPodId, ownerId); + await createPod(secondPodId, otherOwnerId, [ownerId]); + const installed = await install({ installableId: 'telegram', installedBy: ownerId, podId: firstPodId }); + const relay = jest.fn().mockResolvedValue(undefined); + eventHandlers['telegram.relay'] = relay; + + // Membership alone is deliberately not a subscription: a user joining a + // pod must opt into that pod's gate before anything reaches their DM. + await dispatch('chat.message', { + podId: secondPodId, agentUsername: 'kai', displayName: 'Kai', content: '[ESCALATE] no gate', + }); + expect(relay).not.toHaveBeenCalled(); + + await Integration.updateOne( + { _id: installed.integration._id }, + { $set: { [`config.gates.${secondPodId}`]: { enabled: false, since: new Date() } } }, + ); + + await dispatch('chat.message', { + podId: firstPodId, agentUsername: 'kai', displayName: 'Kai', content: '[ESCALATE] first', + }); + await dispatch('chat.message', { + podId: secondPodId, agentUsername: 'kai', displayName: 'Kai', content: '[ESCALATE] disabled', + }); + expect(relay).toHaveBeenCalledTimes(1); + + await Integration.updateOne( + { _id: installed.integration._id }, + { $set: { [`config.gates.${secondPodId}.enabled`]: true } }, + ); + await dispatch('chat.message', { + podId: secondPodId, agentUsername: 'kai', displayName: 'Kai', content: '[ESCALATE] enabled', + }); + expect(relay).toHaveBeenCalledTimes(2); + + await Pod.updateOne({ _id: secondPodId }, { $pull: { members: ownerId } }); + await dispatch('chat.message', { + podId: secondPodId, agentUsername: 'kai', displayName: 'Kai', content: '[ESCALATE] removed', + }); + expect(relay).toHaveBeenCalledTimes(2); + }); + it('does not invoke a handler when the event pod has no installation', async () => { const relay = jest.fn().mockResolvedValue(undefined); eventHandlers['telegram.relay'] = relay; @@ -137,11 +208,13 @@ describe('installable event dispatcher', () => { it('continues to dispatch an existing direct Telegram integration', async () => { const podId = freshId(); + const ownerId = freshId(); + await createPod(podId, ownerId); await Integration.create({ podId, type: 'telegram', status: 'pending', - createdBy: freshId(), + createdBy: ownerId, isActive: true, config: { liveRelay: true, chatId: 'chat-1', chatType: 'private' }, }); @@ -162,8 +235,10 @@ describe('installable event dispatcher', () => { it('selects installed Slack and Telegram rows independently for one pod', async () => { const podId = freshId(); - await install({ installableId: 'telegram', installedBy: freshId(), podId }); - await install({ installableId: 'slack', installedBy: freshId(), podId }); + const ownerId = freshId(); + await createPod(podId, ownerId); + await install({ installableId: 'telegram', installedBy: ownerId, podId }); + await install({ installableId: 'slack', installedBy: ownerId, podId }); const telegramRelay = jest.fn().mockResolvedValue(undefined); const slackRelay = jest.fn().mockResolvedValue(undefined); eventHandlers['telegram.relay'] = telegramRelay; @@ -184,7 +259,9 @@ describe('installable event dispatcher', () => { it('does not dispatch an error-gated Slack row while keeping its projection active for recovery', async () => { const podId = freshId(); - const installed = await install({ installableId: 'slack', installedBy: freshId(), podId }); + const ownerId = freshId(); + await createPod(podId, ownerId); + const installed = await install({ installableId: 'slack', installedBy: ownerId, podId }); await Integration.updateOne( { _id: installed.integration._id }, { $set: { status: 'error', isActive: true } }, @@ -203,9 +280,52 @@ describe('installable event dispatcher', () => { expect(slackRelay).not.toHaveBeenCalled(); }); + it('keeps fanning out when one member connector fails', async () => { + const podId = freshId(); + const firstOwner = freshId(); + const secondOwner = freshId(); + const thirdOwner = freshId(); + await createPod(podId, firstOwner, [secondOwner, thirdOwner]); + const [first, second, third] = await Promise.all([ + install({ installableId: 'telegram', installedBy: firstOwner, podId }), + install({ installableId: 'telegram', installedBy: secondOwner, podId }), + install({ installableId: 'telegram', installedBy: thirdOwner, podId }), + ]); + await Promise.all([ + Integration.updateOne({ _id: first.integration._id }, { $set: { 'config.chatId': 'chat-a', 'config.chatType': 'private' } }), + Integration.updateOne({ _id: second.integration._id }, { $set: { 'config.chatId': 'chat-b', 'config.chatType': 'private' } }), + Integration.updateOne({ _id: third.integration._id }, { $set: { 'config.chatId': 'chat-c', 'config.chatType': 'private' } }), + ]); + const previousToken = process.env.TELEGRAM_BOT_TOKEN; + process.env.TELEGRAM_BOT_TOKEN = 'test-bot-token'; + const sendMessage = jest.spyOn(telegramSend, 'sendMessage') + .mockResolvedValueOnce({ messageId: 1 }) + .mockRejectedValueOnce(new Error('provider throttled')) + .mockResolvedValueOnce({ messageId: 3 }); + + try { + await dispatch('chat.message', { + podId, agentUsername: 'kai', displayName: 'Kai', content: 'fan out despite one failure', + }); + + expect(sendMessage).toHaveBeenCalledTimes(3); + expect(new Set(sendMessage.mock.calls.map((call) => call[1]))).toEqual( + new Set(['chat-a', 'chat-b', 'chat-c']), + ); + const rows = await Integration.find({ _id: { $in: [first.integration._id, second.integration._id, third.integration._id] } }); + expect(rows.every((row) => row.status !== 'error')).toBe(true); + } finally { + sendMessage.mockRestore(); + if (previousToken === undefined) delete process.env.TELEGRAM_BOT_TOKEN; + else process.env.TELEGRAM_BOT_TOKEN = previousToken; + } + }); + it('contains an individual handler failure', async () => { const podId = freshId(); - await install({ installableId: 'telegram', installedBy: freshId(), podId }); + const ownerId = freshId(); + await createPod(podId, ownerId); + await install({ installableId: 'telegram', installedBy: ownerId, podId }); eventHandlers['telegram.relay'] = jest.fn().mockRejectedValue(new Error('provider unavailable')); await expect(dispatch('chat.message', { diff --git a/backend/__tests__/unit/services/installableInstallationService.test.js b/backend/__tests__/unit/services/installableInstallationService.test.js index b2061ae195..c0e5937475 100644 --- a/backend/__tests__/unit/services/installableInstallationService.test.js +++ b/backend/__tests__/unit/services/installableInstallationService.test.js @@ -1,10 +1,15 @@ // @ts-nocheck +// testUtils imports jsonwebtoken, whose Node 26-incompatible SlowBuffer +// dependency is irrelevant to this Mongo-backed service suite. +jest.mock('jsonwebtoken', () => ({})); + const mongoose = require('mongoose'); const Installable = require('../../../models/Installable'); const InstallableInstallation = require('../../../models/InstallableInstallation'); const Integration = require('../../../models/Integration'); +const Pod = require('../../../models/Pod'); const ConnectorSecret = require('../../../models/ConnectorSecret'); const { install, @@ -12,6 +17,7 @@ const { InstallLockLostError, InstallableAlreadyInstalledError, InstallableProjectionError, + InstallationPausedError, } = require('../../../services/installable/installableInstallationService'); const { sweep } = require('../../../services/installable/installableReconciler'); const { TELEGRAM_CONNECTOR, SLACK_CONNECTOR } = require('../../../scripts/seed-builtin-connectors'); @@ -58,7 +64,7 @@ describe('installable connector projection', () => { process.env.CONNECTOR_SECRET_ACTIVE_KEY = 'k1'; }); - it('projects Telegram once, mints last, and shares one Integration between components', async () => { + it('projects a fresh connector as user-scoped, mints last, and shares one Integration between components', async () => { const { userId, podId } = ids(); const result = await install({ installableId: 'telegram', installedBy: userId, podId }); @@ -78,9 +84,11 @@ describe('installable connector projection', () => { expect(String(integration.podId)).toBe(podId); expect(String(result.installation.boundPodId)).toBe(podId); expect(integration.createdBy.toString()).toBe(userId); + expect(integration.scope).toBe('user'); expect(integration.config.linkedUserId).toBe(userId); expect(integration.config.liveRelay).toBe(true); expect(integration.config.relayAllAgentMessages).toBe(true); + expect(integration.config.gates.get(podId)).toMatchObject({ enabled: true, since: expect.any(Date) }); expect(integration.config.connectCode).toMatch(/^[a-f0-9]{32}$/); expect(integration.config.connectCodeExpiresAt).toBeInstanceOf(Date); @@ -406,6 +414,61 @@ describe('installable connector projection', () => { expect(replacementAfterSecondUninstall.isActive).toBe(false); }); + it('refuses install and uninstall while an administrator has paused the parent without writing either row', async () => { + const { userId, podId } = ids(); + const installed = await install({ installableId: 'telegram', installedBy: userId, podId }); + const pausedAt = new Date(); + await InstallableInstallation.updateOne( + { _id: installed.installation._id }, + { + $set: { + status: 'paused', + errorMessage: 'Safety review in progress.', + adminPause: { reason: 'Safety review in progress.', at: pausedAt, adminId: 'admin' }, + }, + }, + ); + await Integration.updateOne( + { _id: installed.integration._id }, + { $set: { 'config.adminPause': { reason: 'Safety review in progress.', at: pausedAt, adminId: 'admin' } } }, + ); + const beforeParent = await InstallableInstallation.findById(installed.installation._id).lean(); + const beforeIntegration = await Integration.findById(installed.integration._id).lean(); + + await expect(install({ installableId: 'telegram', installedBy: userId, podId })) + .rejects.toMatchObject({ code: 'installation_paused', reason: 'Safety review in progress.' }); + await expect(uninstall({ installableId: 'telegram', installedBy: userId })) + .rejects.toBeInstanceOf(InstallationPausedError); + + expect(await InstallableInstallation.countDocuments({ installableId: 'telegram', targetId: userId })).toBe(1); + expect(await Integration.countDocuments({ installationId: String(installed.installation._id) })).toBe(1); + expect(await InstallableInstallation.findById(installed.installation._id).lean()).toMatchObject({ + status: beforeParent.status, + claimId: beforeParent.claimId, + claimedAt: beforeParent.claimedAt, + errorMessage: beforeParent.errorMessage, + }); + expect(await Integration.findById(installed.integration._id).lean()).toMatchObject({ + isActive: beforeIntegration.isActive, + config: expect.objectContaining({ adminPause: beforeIntegration.config.adminPause }), + }); + }); + + it('keeps a paused parent in the live unique set', async () => { + const { userId, podId } = ids(); + const installed = await install({ installableId: 'telegram', installedBy: userId, podId }); + await InstallableInstallation.updateOne( + { _id: installed.installation._id }, + { $set: { status: 'paused', errorMessage: 'Paused.' } }, + ); + + await expect(InstallableInstallation.create({ + installableId: 'telegram', installableVersion: '1.0.0', targetType: 'user', + targetId: new mongoose.Types.ObjectId(userId), scope: 'user', + installedBy: new mongoose.Types.ObjectId(userId), installSource: 'direct', status: 'paused', + })).rejects.toMatchObject({ code: 11000 }); + }); + it('revokes Slack’s envelope secret when the installation is uninstalled', async () => { const { userId, podId } = ids(); const installed = await install({ installableId: 'slack', installedBy: userId, podId }); @@ -545,6 +608,40 @@ describe('installable connector projection', () => { expect(parent.components.every((component) => component.status === 'stale')).toBe(true); }); + it('reconciles pause projections and prunes gates when the owner leaves a pod', async () => { + const { userId, podId } = ids(); + await Pod.create({ _id: podId, name: 'Current pod', type: 'team', createdBy: userId, members: [] }); + const installed = await install({ installableId: 'telegram', installedBy: userId, podId }); + const stalePodId = new mongoose.Types.ObjectId().toString(); + const pausedAt = new Date(); + await InstallableInstallation.updateOne( + { _id: installed.installation._id }, + { $set: { status: 'paused', errorMessage: 'Reviewing', adminPause: { reason: 'Reviewing', at: pausedAt, adminId: 'admin' } } }, + ); + await Integration.updateOne( + { _id: installed.integration._id }, + { + $unset: { 'config.adminPause': 1 }, + $set: { [`config.gates.${stalePodId}`]: { enabled: true, since: new Date() } }, + }, + ); + + const pausedSweep = await sweep(new Date()); + const pausedProjection = await Integration.findById(installed.integration._id); + expect(pausedSweep.restampedPauses).toBe(1); + expect(pausedSweep.prunedGates).toBe(1); + expect(pausedProjection.config.adminPause).toMatchObject({ reason: 'Reviewing', adminId: 'admin' }); + expect(pausedProjection.config.gates.get(stalePodId)).toBeUndefined(); + + await InstallableInstallation.updateOne( + { _id: installed.installation._id }, + { $set: { status: 'active' }, $unset: { adminPause: 1 } }, + ); + const resumedSweep = await sweep(new Date()); + expect(resumedSweep.clearedPauseProjections).toBe(1); + expect((await Integration.findById(installed.integration._id)).config.adminPause).toBeUndefined(); + }); + it('expires a pending Slack bind, revokes its encrypted token, and leaves a retryable active card', async () => { const { userId, podId } = ids(); const installed = await install({ installableId: 'slack', installedBy: userId, podId }); diff --git a/backend/__tests__/unit/services/slackBridgeService.test.js b/backend/__tests__/unit/services/slackBridgeService.test.js index ae19a96e0d..6e38f04504 100644 --- a/backend/__tests__/unit/services/slackBridgeService.test.js +++ b/backend/__tests__/unit/services/slackBridgeService.test.js @@ -7,7 +7,9 @@ const Integration = require('../../../models/Integration'); const Pod = require('../../../models/Pod'); const connectorSecrets = require('../../../services/connectorSecrets'); const SlackApi = require('../../../services/slackApi'); -const { relayAgentMessageToSlack, routeSlackReplyContent } = require('../../../services/slackBridgeService'); +const { + relayAgentMessageToSlack, relaySlackMessageToPod, routeSlackReplyContent, isRelayableIntegration, +} = require('../../../services/slackBridgeService'); const integration = { _id: 'integration-1', @@ -65,4 +67,26 @@ describe('Slack installable bridge', () => { expect(connectorSecrets.get).not.toHaveBeenCalled(); expect(SlackApi).not.toHaveBeenCalled(); }); + + test('uses an enabled user gate instead of the selected pod only', () => { + expect(isRelayableIntegration({ + ...integration, + scope: 'user', + config: { ...integration.config, gates: { 'second-pod': { enabled: true } } }, + }, 'second-pod')).toBe(true); + }); + + test('drops inbound rather than stringifying a user connector without an active pod', async () => { + const result = await relaySlackMessageToPod({ + integration: { + ...integration, + scope: 'user', + podId: undefined, + config: { ...integration.config, gates: { 'pod-1': { enabled: true } } }, + }, + event: { text: 'hello', user: 'U1' }, + }); + + expect(result).toEqual({ relayed: false }); + }); }); diff --git a/backend/__tests__/unit/services/telegramBridgeService.test.js b/backend/__tests__/unit/services/telegramBridgeService.test.js index 2277976215..4a341732e3 100644 --- a/backend/__tests__/unit/services/telegramBridgeService.test.js +++ b/backend/__tests__/unit/services/telegramBridgeService.test.js @@ -4,7 +4,9 @@ jest.mock('../../../models/Integration', () => ({ })); jest.mock('../../../services/telegramService', () => ({ sendMessage: jest.fn() })); -const { shouldEscalate, routeReplyContent } = require('../../../services/telegramBridgeService'); +const { + shouldEscalate, routeReplyContent, isRelayableIntegration, relayTelegramMessageToPod, +} = require('../../../services/telegramBridgeService'); describe('telegramBridgeService — escalation gate', () => { const integration = (config = {}) => ({ _id: 'i1', podId: 'p1', config }); @@ -50,6 +52,26 @@ describe('telegramBridgeService — escalation gate', () => { integration: integration({ relayAllAgentMessages: true }), })).toBe(true); }); + + it('overlays a gate mode and lead on the connector defaults for that pod', () => { + const gated = integration({ + relayAllAgentMessages: true, + leadAgentUsername: 'default-lead', + gates: { + 'pod-a': { enabled: true, mode: 'attention', lead: 'pod-lead' }, + 'pod-b': { enabled: true, mode: 'mirror' }, + }, + }); + expect(shouldEscalate({ + content: 'ordinary progress', agentUsername: 'worker', integration: gated, podId: 'pod-a', + })).toBe(false); + expect(shouldEscalate({ + content: 'ordinary progress', agentUsername: 'pod-lead', integration: gated, podId: 'pod-a', + })).toBe(true); + expect(shouldEscalate({ + content: 'ordinary progress', agentUsername: 'worker', integration: gated, podId: 'pod-b', + })).toBe(true); + }); }); describe('telegramBridgeService — quote-reply routing', () => { @@ -94,3 +116,40 @@ describe('telegramBridgeService — quote-reply routing', () => { expect(out.content).toBe('plain message'); }); }); + +describe('telegramBridgeService — user-scope outbound gate', () => { + const userScoped = { + _id: 'i1', + scope: 'user', + podId: 'active-pod', + type: 'telegram', + isActive: true, + config: { + liveRelay: true, + chatType: 'private', + chatId: 'chat-1', + gates: { 'other-pod': { enabled: true } }, + }, + }; + + it('allows an enabled gate rather than only the selected pod', () => { + expect(isRelayableIntegration(userScoped, 'other-pod')).toBe(true); + }); + + it('refuses a disabled or absent gate', () => { + expect(isRelayableIntegration({ + ...userScoped, + config: { ...userScoped.config, gates: { 'other-pod': { enabled: false } } }, + }, 'other-pod')).toBe(false); + expect(isRelayableIntegration(userScoped, 'missing-pod')).toBe(false); + }); + + it('drops inbound rather than stringifying a user connector without an active pod', async () => { + const result = await relayTelegramMessageToPod({ + integration: { ...userScoped, podId: undefined }, + telegramMessage: { text: 'hello from the phone' }, + }); + + expect(result).toEqual({ relayed: false }); + }); +}); diff --git a/backend/models/InstallableInstallation.ts b/backend/models/InstallableInstallation.ts index b8e3efcc8f..c069f1be4e 100644 --- a/backend/models/InstallableInstallation.ts +++ b/backend/models/InstallableInstallation.ts @@ -85,6 +85,21 @@ export interface IComponentInstallation { updatedAt: Date; } +export interface IInstallationAdminPause { + reason: string; + at: Date; + adminId: string; +} + +export interface IInstallationPauseAudit { + action: 'pause' | 'resume'; + installationId: string; + ownerId: string; + adminId: string; + reason: string; + at: Date; +} + export interface IInstallableInstallation extends Document { // Which Installable is installed installableId: string; // FK (logical) → Installable.installableId @@ -121,6 +136,9 @@ export interface IInstallableInstallation extends Document { claimedAt?: Date; errorMessage?: string; staleSince?: Date; + /** Canonical pause state; Integration.config.adminPause is its projection. */ + adminPause?: IInstallationAdminPause; + pauseAudit: IInstallationPauseAudit[]; createdAt: Date; updatedAt: Date; @@ -213,6 +231,22 @@ const InstallableInstallationSchema = new Schema( claimedAt: { type: Date }, errorMessage: { type: String }, staleSince: { type: Date }, + adminPause: { + reason: { type: String }, + at: { type: Date }, + adminId: { type: String }, + }, + pauseAudit: { + type: [new Schema({ + action: { type: String, enum: ['pause', 'resume'], required: true }, + installationId: { type: String, required: true }, + ownerId: { type: String, required: true }, + adminId: { type: String, required: true }, + reason: { type: String, required: true }, + at: { type: Date, required: true }, + }, { _id: false })], + default: [], + }, }, { timestamps: true }, ); @@ -232,7 +266,7 @@ InstallableInstallationSchema.index( unique: true, name: 'installable_live_target_unique', partialFilterExpression: { - status: { $in: ['installing', 'activating', 'uninstalling', 'active', 'error'] }, + status: { $in: ['installing', 'activating', 'uninstalling', 'active', 'paused', 'error'] }, }, }, ); diff --git a/backend/models/Integration.ts b/backend/models/Integration.ts index 917d5d07ca..7491257631 100644 --- a/backend/models/Integration.ts +++ b/backend/models/Integration.ts @@ -11,6 +11,21 @@ export type IntegrationType = | 'instagram'; export type IntegrationStatus = 'connected' | 'disconnected' | 'error' | 'pending'; +export type IntegrationScope = 'pod' | 'user'; + +export interface IIntegrationGate { + enabled: boolean; + mode?: 'attention' | 'mirror'; + lead?: string; + since: Date; +} + +/** Server-owned projection of an administrator's parent-level pause. */ +export interface IIntegrationAdminPause { + reason: string; + at: Date; + adminId: string; +} export interface IIngestToken { tokenHash: string; @@ -36,7 +51,13 @@ export interface IIntegration extends Document { installationClaimId?: string; /** Terminal tombstone: a revoked projection can never be activated again. */ revokedAt?: Date; - podId: Types.ObjectId; + /** + * Legacy pod-scoped rows require a pod. User-scoped connector rows retain + * their currently selected pod here for bare-message routing, but their + * outbound subscriptions live in config.gates. + */ + podId?: Types.ObjectId; + scope: IntegrationScope; type: IntegrationType; status: IntegrationStatus; config: { @@ -97,6 +118,8 @@ export interface IIntegration extends Document { leadAgentUsername?: string; relayAllAgentMessages?: boolean; relayMutedUntil?: Date; + gates?: Record; + adminPause?: IIntegrationAdminPause; /** Opaque ConnectorSecret id; credentials never live on the Integration. */ botTokenRef?: string; /** Slack's workspace identity and the bound one-to-one DM. */ @@ -136,7 +159,19 @@ const IntegrationSchema = new Schema( installationId: { type: String, unique: true, sparse: true }, installationClaimId: { type: String }, revokedAt: { type: Date }, - podId: { type: Schema.Types.ObjectId, ref: 'Pod', required: true }, + podId: { + type: Schema.Types.ObjectId, + ref: 'Pod', + required(this: IIntegration) { + return this.scope === 'pod'; + }, + }, + scope: { + type: String, + enum: ['pod', 'user'], + default: 'pod', + required: true, + }, type: { type: String, required: true, @@ -211,6 +246,23 @@ const IntegrationSchema = new Schema( leadAgentUsername: String, relayAllAgentMessages: { type: Boolean, default: false }, relayMutedUntil: Date, + gates: { + type: Map, + of: new Schema({ + enabled: { type: Boolean, required: true }, + mode: { type: String, enum: ['attention', 'mirror'] }, + lead: String, + since: { type: Date, required: true }, + }, { _id: false }), + }, + adminPause: { + type: new Schema({ + reason: { type: String, required: true }, + at: { type: Date, required: true }, + adminId: { type: String, required: true }, + }, { _id: false }), + default: undefined, + }, // Connector secrets live in ConnectorSecret; routes must never accept // this reference from clients (see SERVER_OWNED_CONFIG_KEYS). botTokenRef: String, diff --git a/backend/routes/admin/installables.ts b/backend/routes/admin/installables.ts new file mode 100644 index 0000000000..e7c4cc3bac --- /dev/null +++ b/backend/routes/admin/installables.ts @@ -0,0 +1,143 @@ +// Administrator stop/resume lever for user-scoped connector installations. +// It intentionally does not expose owner configuration: an admin can stop a +// private relay and say why, but cannot author into or reconfigure its DM. + +// eslint-disable-next-line @typescript-eslint/no-require-imports, global-require +const express = require('express'); +// eslint-disable-next-line @typescript-eslint/no-require-imports, global-require +const auth = require('../../middleware/auth'); +// eslint-disable-next-line @typescript-eslint/no-require-imports, global-require +const adminAuth = require('../../middleware/adminAuth'); +// eslint-disable-next-line @typescript-eslint/no-require-imports, global-require +const InstallableInstallation = require('../../models/InstallableInstallation'); +// eslint-disable-next-line @typescript-eslint/no-require-imports, global-require +const Integration = require('../../models/Integration'); +import { Types } from 'mongoose'; + +interface AuthReq { + user?: { id?: string }; + params?: Record; + body?: Record; +} + +const router: ReturnType = express.Router(); + +const requiredReason = (body: unknown): string | null => { + const reason = (body as { reason?: unknown } | null)?.reason; + if (typeof reason !== 'string' || !reason.trim()) return null; + return reason.trim(); +}; + +const installationFilter = (req: AuthReq, status: 'active' | 'paused') => { + const installationId = String(req.params?.installationId || ''); + if (!Types.ObjectId.isValid(installationId)) return null; + return { + _id: installationId, + installableId: String(req.params?.installableId || '').toLowerCase(), + targetType: 'user', + scope: 'user', + status, + }; +}; + +const auditEntry = (action: 'pause' | 'resume', installation: { + _id: unknown; + installedBy?: unknown; +}, adminId: string, reason: string, at: Date) => ({ + action, + installationId: String(installation._id), + ownerId: String(installation.installedBy || ''), + adminId, + reason, + at, +}); + +/** + * Pause is parent-first: a crash after the first write has already stopped + * outbound dispatch. The reconciler repairs a child which was not stamped. + */ +router.post('/:installableId/installations/:installationId/pause', auth, adminAuth, async (req: AuthReq, res: any) => { + const reason = requiredReason(req.body); + if (!reason) return res.status(400).json({ error: 'reason is required' }); + const filter = installationFilter(req, 'active'); + if (!filter) return res.status(400).json({ error: 'installationId must be a valid ObjectId' }); + const adminId = String(req.user?.id || ''); + const at = new Date(); + + try { + const existing = await InstallableInstallation.findOne(filter); + if (!existing) return res.status(404).json({ error: 'Active installation not found' }); + const installation = await InstallableInstallation.findOneAndUpdate( + filter, + { + $set: { + status: 'paused', + errorMessage: reason, + adminPause: { reason, at, adminId }, + }, + $push: { pauseAudit: auditEntry('pause', existing, adminId, reason, at) }, + }, + { new: true }, + ); + if (!installation) return res.status(404).json({ error: 'Active installation not found' }); + + try { + await Integration.updateMany( + { installationId: String(installation._id) }, + { $set: { 'config.adminPause': { reason, at, adminId } } }, + ); + } catch (error) { + console.error('[admin/installables] pause projection failed:', (error as Error).message); + return res.status(202).json({ status: 'paused', projected: false, installation }); + } + return res.json({ status: 'paused', projected: true, installation }); + } catch (error) { + console.error('[admin/installables] pause failed:', (error as Error).message); + return res.status(500).json({ error: 'Could not pause installation' }); + } +}); + +/** + * Resume is children-first. If parent completion fails, the parent remains + * paused and the reconciler re-stamps children on its next pass. + */ +router.post('/:installableId/installations/:installationId/resume', auth, adminAuth, async (req: AuthReq, res: any) => { + const reason = requiredReason(req.body); + if (!reason) return res.status(400).json({ error: 'reason is required' }); + const filter = installationFilter(req, 'paused'); + if (!filter) return res.status(400).json({ error: 'installationId must be a valid ObjectId' }); + const adminId = String(req.user?.id || ''); + const at = new Date(); + + try { + const existing = await InstallableInstallation.findOne(filter); + if (!existing) return res.status(404).json({ error: 'Paused installation not found' }); + try { + await Integration.updateMany( + { installationId: String(existing._id) }, + { $unset: { 'config.adminPause': 1 } }, + ); + } catch (error) { + console.error('[admin/installables] resume projection failed:', (error as Error).message); + return res.status(202).json({ status: 'paused', projected: false, installation: existing }); + } + const installation = await InstallableInstallation.findOneAndUpdate( + filter, + { + $set: { status: 'active', errorMessage: null }, + $unset: { adminPause: 1 }, + $push: { pauseAudit: auditEntry('resume', existing, adminId, reason, at) }, + }, + { new: true }, + ); + if (!installation) { + return res.status(202).json({ status: 'paused', projected: false, installation: existing }); + } + return res.json({ status: 'active', projected: true, installation }); + } catch (error) { + console.error('[admin/installables] resume failed:', (error as Error).message); + return res.status(500).json({ error: 'Could not resume installation' }); + } +}); + +module.exports = router; diff --git a/backend/routes/installables.ts b/backend/routes/installables.ts index 4c4fcd4608..dc137c4509 100644 --- a/backend/routes/installables.ts +++ b/backend/routes/installables.ts @@ -36,6 +36,7 @@ const { InstallableNotFoundError, InstallableProjectionError, InstallInProgressError, + InstallationPausedError, install, uninstall, } = require('../services/installable/installableInstallationService'); @@ -461,6 +462,15 @@ const sendInstallError = (error: Error, res: Res): void => { }); return; } + if (error instanceof InstallationPausedError) { + const { reason } = error as Error & { reason?: string }; + res.status(409).json({ + code: 'installation_paused', + error: error.message, + reason, + }); + return; + } if (error instanceof InstallableAlreadyInstalledError) { const { boundPodId } = error as Error & { boundPodId: string }; res.status(409).json({ diff --git a/backend/routes/integrations.ts b/backend/routes/integrations.ts index b94207bd1b..a558b05635 100644 --- a/backend/routes/integrations.ts +++ b/backend/routes/integrations.ts @@ -47,6 +47,9 @@ const SERVER_OWNED_CONFIG_KEYS = [ // OAuth callback and connectorSecrets own Slack identity and its opaque // credential reference. Accepting either from a browser body defeats D6. 'botTokenRef', 'teamId', 'teamName', 'slackUserId', 'slackUserName', 'pendingBind', + // An administrator's pause is projected from the parent installation. An + // owner's normal config write must never lift that stop. + 'adminPause', ]; const stripServerOwnedConfig = (config: Record): Record => { const next = { ...config }; @@ -63,6 +66,11 @@ const stripServerOwnedConfig = (config: Record): Record /^[a-f\d]{24}$/i.test(value); const readRelayFlags = (config: Record): { next: Record; invalid?: string } => { const next = { ...config }; for (const k of RELAY_FLAG_KEYS) { @@ -485,11 +493,57 @@ router.patch('/:id', auth, async (req: AuthReq, res: Res) => { try { const { id } = req.params || {}; const { config, status, isActive } = (req.body || {}) as { config?: Record; status?: string; isActive?: boolean }; - const integration = await Integration.findById(id) as { type?: string; createdBy?: { toString: () => string }; podId?: unknown; config?: { toObject?: () => Record } } | null; + const integration = await Integration.findById(id) as { + type?: string; + scope?: string; + createdBy?: { toString: () => string }; + podId?: unknown; + config?: { toObject?: () => Record }; + } | null; if (!integration) return res.status(404).json({ message: 'Integration not found' }); - const canUpdate = await canDeleteIntegration(integration, req.user?.id || ''); - if (!canUpdate) return res.status(403).json({ message: 'Access denied' }); const currentConfig = integration.config?.toObject ? integration.config.toObject() : (integration.config || {}) as Record; + const requesterId = req.user?.id || ''; + if (integration.scope === 'user') { + // A user-scoped connector is a private attention surface. A pod creator + // or instance admin must not turn on a gate (or live relay) for someone + // else, so its entire PATCH surface belongs to its linked owner alone. + if ( + integration.createdBy?.toString() !== requesterId + || String(currentConfig.linkedUserId || '') !== requesterId + ) { + return res.status(403).json({ message: 'Access denied' }); + } + if (config && Object.prototype.hasOwnProperty.call(config, 'adminPause')) { + return res.status(400).json({ message: 'adminPause is managed by an administrator' }); + } + const requestedGates = config?.gates; + if (requestedGates !== undefined) { + if (!requestedGates || Array.isArray(requestedGates) || typeof requestedGates !== 'object') { + return res.status(400).json({ message: 'gates must be an object keyed by pod id' }); + } + const gatePodIds = Object.keys(requestedGates as Record); + if (gatePodIds.length > MAX_GATES_PER_WRITE) { + return res.status(400).json({ message: `gates may contain at most ${MAX_GATES_PER_WRITE} pod ids` }); + } + // Validate every key before it is interpolated into `config.gates.*` + // or reaches the membership lookup. A Mongo key is syntax, not data: + // dots and dollar prefixes have different semantics from a bad value. + if (gatePodIds.some((gatePodId) => !isObjectIdKey(gatePodId))) { + return res.status(400).json({ message: 'every gate key must be a valid pod id' }); + } + // Check every requested key before constructing the update. This keeps + // a mixed valid/invalid PATCH atomic: no valid gate is written first. + for (const gatePodId of gatePodIds) { + const gatePod = await Pod.findById(gatePodId); + if (!gatePod || !isPodMember(gatePod, requesterId)) { + return res.status(403).json({ message: 'Access denied' }); + } + } + } + } else { + const canUpdate = await canDeleteIntegration(integration, requesterId); + if (!canUpdate) return res.status(403).json({ message: 'Access denied' }); + } // Bridge attribution guard: config.linkedUserId is the identity every // inbound live-relay message is AUTHORED as (pod row, socket payload, // agent wake). It is derived from the authenticated caller when liveRelay diff --git a/backend/routes/webhooks/slack.ts b/backend/routes/webhooks/slack.ts index f4530bbf29..3f189ff988 100644 --- a/backend/routes/webhooks/slack.ts +++ b/backend/routes/webhooks/slack.ts @@ -73,6 +73,7 @@ const finishEvent = async (eventId: string, teamId: string, event: any): Promise 'config.chatId': channelId, 'config.chatType': 'im', 'config.liveRelay': true, + 'config.adminPause': { $exists: false }, status: { $ne: 'error' }, }).lean(); if (integration) await relaySlackMessageToPod({ integration, event }); @@ -133,6 +134,7 @@ router.post('/commands', slackWebhookRateLimit, signed, async (req: any, res: an 'config.chatId': channelId, 'config.chatType': 'im', 'config.slackUserId': slackUserId, + 'config.adminPause': { $exists: false }, status: { $ne: 'error' }, }); if (!integration) { diff --git a/backend/routes/webhooks/telegram.ts b/backend/routes/webhooks/telegram.ts index 997afe4731..b7dabf2667 100644 --- a/backend/routes/webhooks/telegram.ts +++ b/backend/routes/webhooks/telegram.ts @@ -181,6 +181,10 @@ const handleSummaryCommand = async (chat: any, integration: any) => { ); return; } + if (!integration.podId) { + await telegramService.sendMessage(botToken, chatId, 'This connector has no active pod. Choose one in Commonly first.'); + return; + } const latest = await Integration.findById(integration._id).lean(); const buffer = latest?.config?.messageBuffer || []; @@ -252,6 +256,10 @@ const handlePodSummaryCommand = async (chat: any, integration: any) => { ); return; } + if (!integration.podId) { + await telegramService.sendMessage(botToken, chatId, 'This connector has no active pod. Choose one in Commonly first.'); + return; + } const latestSummary = await Summary.findOne({ type: 'chats', @@ -321,6 +329,7 @@ const handleModeCommand = async (chat: any, integration: any, arg?: string) => { const handleStatusCommand = async (chat: any, integration: any) => { const chatId = chat?.id?.toString(); if (!integration) return sendToChat(chatId, NOT_LINKED); + if (!integration.podId) return sendToChat(chatId, 'This connector has no active pod. Choose one in Commonly first.'); const pod = await Pod.findById(integration.podId).lean(); const mode = integration.config?.relayAllAgentMessages === true ? 'mirror' : 'attention'; const mutedUntil = integration.config?.relayMutedUntil; @@ -425,6 +434,7 @@ router.post('/', async (req: any, res: any) => { type: 'telegram', isActive: true, 'config.chatId': chatId, + 'config.adminPause': { $exists: false }, }); if (command === SUMMARY_COMMAND) { diff --git a/backend/scripts/migrate-connector-gates.ts b/backend/scripts/migrate-connector-gates.ts new file mode 100644 index 0000000000..355f2197c0 --- /dev/null +++ b/backend/scripts/migrate-connector-gates.ts @@ -0,0 +1,97 @@ +#!/usr/bin/env node +/* + * D8 Phase 2 connector-gate migration. + * + * Converts only installable-backed connector projections to user scope. A + * legacy integration has no installationId and remains pod scoped. The CAS + * filter makes a second run a no-op, including if a live writer creates a + * gate between this script's read and write. + * + * Usage: + * ts-node backend/scripts/migrate-connector-gates.ts --dry + * ts-node backend/scripts/migrate-connector-gates.ts + */ + +import mongoose from 'mongoose'; +import Integration from '../models/Integration'; +import InstallableInstallation from '../models/InstallableInstallation'; + +export interface ConnectorGateMigrationResult { + scanned: number; + migrated: number; + skipped: number; +} + +export async function migrateConnectorGates( + options: { dryRun?: boolean } = {}, +): Promise { + const dryRun = options.dryRun === true; + const result: ConnectorGateMigrationResult = { scanned: 0, migrated: 0, skipped: 0 }; + const cursor = Integration.find({ + installationId: { $exists: true, $type: 'string' }, + podId: { $exists: true, $ne: null }, + 'config.gates': { $exists: false }, + }).select('_id podId createdAt').cursor(); + + for await (const integration of cursor) { + result.scanned += 1; + const podId = String(integration.podId || ''); + if (!podId) { + result.skipped += 1; + continue; + } + if (dryRun) { + result.migrated += 1; + continue; + } + const updated = await Integration.updateOne( + { + _id: integration._id, + installationId: { $exists: true, $type: 'string' }, + podId: integration.podId, + 'config.gates': { $exists: false }, + }, + { + $set: { + scope: 'user', + [`config.gates.${podId}`]: { + enabled: true, + since: integration.createdAt || new Date(), + }, + }, + }, + ); + if (updated.modifiedCount) result.migrated += 1; + else result.skipped += 1; + } + return result; +} + +async function main(): Promise { + const dryRun = process.argv.includes('--dry'); + const mongoUri = process.env.MONGO_URI; + if (!mongoUri) { + console.error('MONGO_URI is required'); + process.exit(1); + } + await mongoose.connect(mongoUri); + try { + // The partial unique index's live set gained paused in this schema flip. + // A dry run remains fully read-only. + if (!dryRun) await InstallableInstallation.syncIndexes(); + const result = await migrateConnectorGates({ dryRun }); + console.log( + `[migrate-connector-gates] ${dryRun ? 'DRY-RUN ' : ''}` + + `scanned=${result.scanned} migrated=${result.migrated} skipped=${result.skipped}`, + ); + } finally { + await mongoose.disconnect(); + } +} + +if (require.main === module) { + main().catch((error) => { + console.error(error); + process.exit(1); + }); +} diff --git a/backend/server.ts b/backend/server.ts index c16e992434..bd2e9348f1 100644 --- a/backend/server.ts +++ b/backend/server.ts @@ -51,6 +51,7 @@ const agentAutonomyAdminRoutes = require('./routes/admin/agentAutonomy'); const agentEventsAdminRoutes = require('./routes/admin/agentEvents'); const adminUsersRoutes = require('./routes/admin/users'); const adminAnalyticsRoutes = require('./routes/admin/analytics'); +const adminInstallableRoutes = require('./routes/admin/installables'); // Conditionally load PostgreSQL routes and models let pgMessageRoutes: any; let pgStatusRoutes: any; @@ -250,6 +251,7 @@ app.use('/api/admin/agents/autonomy', agentAutonomyAdminRoutes); // Admin manual app.use('/api/admin/agents/events', agentEventsAdminRoutes); // Admin agent event debug/queue visibility app.use('/api/admin/users', adminUsersRoutes); // Admin user + invitation management app.use('/api/admin/analytics', adminAnalyticsRoutes); // Admin activation-funnel analytics (GH#661) +app.use('/api/admin/installables', adminInstallableRoutes); // Owner-private connector pause/resume app.use('/api/dev', devRoutes); // Dev tooling (LLM status, etc.) app.use('/api/health', healthRoutes); // Health check endpoints app.use('/api/stats', statsRoutes); // Public stats (no auth) diff --git a/backend/services/connectorRelayPolicy.ts b/backend/services/connectorRelayPolicy.ts index df22b28535..5e597b47e6 100644 --- a/backend/services/connectorRelayPolicy.ts +++ b/backend/services/connectorRelayPolicy.ts @@ -6,6 +6,11 @@ export interface RelayPolicyIntegration { config?: { relayAllAgentMessages?: boolean; leadAgentUsername?: string; + gates?: Record; }; } @@ -16,12 +21,22 @@ export const shouldEscalate = (opts: { content: string; agentUsername: string; integration: RelayPolicyIntegration; + podId?: string; }): boolean => { - const { content, agentUsername, integration } = opts; + const { + content, agentUsername, integration, podId, + } = opts; const cfg = integration.config || {}; - if (cfg.relayAllAgentMessages) return true; - if (cfg.leadAgentUsername - && agentUsername.toLowerCase() === String(cfg.leadAgentUsername).toLowerCase()) { + const gate = podId ? cfg.gates?.[String(podId)] : undefined; + const relayAllAgentMessages = gate?.mode === 'mirror' + ? true + : gate?.mode === 'attention' + ? false + : cfg.relayAllAgentMessages; + const leadAgentUsername = gate?.lead ?? cfg.leadAgentUsername; + if (relayAllAgentMessages) return true; + if (leadAgentUsername + && agentUsername.toLowerCase() === String(leadAgentUsername).toLowerCase()) { return true; } return ESCALATION_MARKERS.test(content) || QUESTION_AT_HUMAN.test(content); diff --git a/backend/services/installable/eventHandlers.ts b/backend/services/installable/eventHandlers.ts index b360fad0a2..690f6a0721 100644 --- a/backend/services/installable/eventHandlers.ts +++ b/backend/services/installable/eventHandlers.ts @@ -4,6 +4,8 @@ import { Types } from 'mongoose'; // eslint-disable-next-line @typescript-eslint/no-require-imports, global-require const Integration = require('../../models/Integration'); // eslint-disable-next-line @typescript-eslint/no-require-imports, global-require +const Pod = require('../../models/Pod'); +// eslint-disable-next-line @typescript-eslint/no-require-imports, global-require const telegramBridgeService = require('../telegramBridgeService'); // eslint-disable-next-line @typescript-eslint/no-require-imports, global-require const slackBridgeService = require('../slackBridgeService'); @@ -37,18 +39,41 @@ const activeHandlersForPod = async (podId: string): Promise> => { - // Selection lives here, before any handler invocation. The $lookup keeps the - // event path O(1) and proves the target pod's connector is the only one - // eligible; individual bridges remain defensive, not authoritative. + // Selection starts from the pod's current membership. A user-scoped + // connector is private to its owner, so a stale gate can never relay after + // that owner leaves even before the reconciler prunes the key. Legacy rows + // retain their pod-scoped selection until they are explicitly migrated. if (!Types.ObjectId.isValid(podId)) return []; + const pod = await Pod.findById(podId).select('createdBy members').lean() as { + createdBy?: unknown; + members?: unknown[]; + } | null; + if (!pod) return []; + const memberIds = Array.from(new Set([ + pod.createdBy, + ...(pod.members || []), + ].filter(Boolean).map((id) => String(id)))).map((id) => new Types.ObjectId(id)); + if (!memberIds.length) return []; + const gateEnabledPath = `config.gates.${String(podId)}.enabled`; return Integration.aggregate([ { $match: { type: { $in: ['telegram', 'slack'] }, isActive: true, status: { $ne: 'error' }, - podId: new Types.ObjectId(podId), 'config.liveRelay': true, + 'config.adminPause': { $exists: false }, + $or: [ + { + scope: { $ne: 'user' }, + podId: new Types.ObjectId(podId), + }, + { + scope: 'user', + createdBy: { $in: memberIds }, + [gateEnabledPath]: true, + }, + ], }, }, { diff --git a/backend/services/installable/installableInstallationService.ts b/backend/services/installable/installableInstallationService.ts index 6bd04d04f6..c58632a6c6 100644 --- a/backend/services/installable/installableInstallationService.ts +++ b/backend/services/installable/installableInstallationService.ts @@ -60,6 +60,19 @@ export class InstallableAlreadyInstalledError extends Error { } } +/** A paused parent is a hard administrative stop, never a retryable lock. */ +export class InstallationPausedError extends Error { + code = 'installation_paused'; + reason: string; + + constructor(reason?: string) { + const resolvedReason = reason || 'Paused by an administrator.'; + super(resolvedReason); + this.name = 'InstallationPausedError'; + this.reason = resolvedReason; + } +} + export class InstallableNotFoundError extends Error { code = 'installable_not_found'; @@ -80,7 +93,7 @@ export interface InstallResult { interface ClaimResult { installation: IInstallableInstallation; ownsClaim: boolean; - claimedState: 'installing' | 'activating' | 'uninstalling' | 'active'; + claimedState: 'installing' | 'activating' | 'uninstalling' | 'active' | 'paused'; } interface InstallArgs { @@ -100,6 +113,7 @@ const liveClaimStatuses: InstallationStatus[] = [ 'activating', 'uninstalling', 'active', + 'paused', 'error', ]; @@ -114,6 +128,7 @@ const statusForClaim = ( if (status === 'active') return 'active'; if (status === 'activating') return 'activating'; if (status === 'uninstalling') return 'uninstalling'; + if (status === 'paused') return 'paused'; return 'installing'; }; @@ -169,6 +184,12 @@ const resultForExisting = async ( installation: IInstallableInstallation, requestedPodId: Types.ObjectId, ): Promise => { + // A duplicate-key loser reaches this helper with ownsClaim:false. Paused is + // deliberately part of the live unique index, so this is the one place an + // owner install must become the explicit 409 rather than a fake transient. + if (installation.status === 'paused') { + throw new InstallationPausedError(installation.errorMessage); + } const integration = await integrationFor(installation); if (installation.status === 'active') { // A live Integration is a real channel binding. Every non-active state is @@ -543,6 +564,9 @@ const installAttempt = async ({ if (!claim.ownsClaim) return resultForExisting(claim.installation, targetPodId); let installation = claim.installation; + if (claim.claimedState === 'paused') { + throw new InstallationPausedError(installation.errorMessage); + } const claimId = installation.claimId; if (!claimId) throw new InstallLockLostError(); @@ -637,7 +661,15 @@ export const uninstall = async ({ if (!installation) throw new InstallableNotFoundError(); const claim = await claimUninstall(installation); - if (!claim.ownsClaim) return claim.installation; + if (!claim.ownsClaim) { + // claimUninstall returns a paused winner unchanged because its claim + // filter intentionally admits only active/error and stale transients. + // Returning it as 200 would make an administrative stop a silent no-op. + if (claim.installation.status === 'paused') { + throw new InstallationPausedError(claim.installation.errorMessage); + } + return claim.installation; + } const claimId = claim.installation.claimId; if (!claimId) throw new InstallLockLostError(); @@ -652,6 +684,7 @@ module.exports = { InstallableProjectionError, InstallInProgressError, InstallableAlreadyInstalledError, + InstallationPausedError, InstallableNotFoundError, install, uninstall, diff --git a/backend/services/installable/installableReconciler.ts b/backend/services/installable/installableReconciler.ts index cbeb93d079..b877b74edf 100644 --- a/backend/services/installable/installableReconciler.ts +++ b/backend/services/installable/installableReconciler.ts @@ -6,9 +6,13 @@ const InstallableInstallation = require('../../models/InstallableInstallation'); // eslint-disable-next-line @typescript-eslint/no-require-imports, global-require const Integration = require('../../models/Integration'); // eslint-disable-next-line @typescript-eslint/no-require-imports, global-require +const Pod = require('../../models/Pod'); +// eslint-disable-next-line @typescript-eslint/no-require-imports, global-require const ConnectorSecret = require('../../models/ConnectorSecret'); // eslint-disable-next-line @typescript-eslint/no-require-imports, global-require const connectorSecrets = require('../connectorSecrets'); +// eslint-disable-next-line @typescript-eslint/no-require-imports, global-require +const isPodMember = require('../../utils/isPodMember'); const ORPHAN_SECRET_GRACE_MS = 10 * 60_000; @@ -55,10 +59,22 @@ const sweepStaleLocks = async (now: Date): Promise<{ completed: number; errored: return { completed, errored }; }; -const sweepActiveInstallations = async (): Promise => { +const sweepActiveInstallations = async (): Promise<{ staleComponents: number; clearedPauseProjections: number }> => { const active = await InstallableInstallation.find({ status: 'active' }).lean() as IInstallableInstallation[]; let staleComponents = 0; + let clearedPauseProjections = 0; for (const installation of active) { + // Resume clears children before it flips the parent active. If the second + // write crashes, active is never reached; this branch only clears stale + // projection flags left by an otherwise completed resume. + const cleared = await Integration.updateMany( + { + installationId: installationIdFor(installation), + 'config.adminPause': { $exists: true }, + }, + { $unset: { 'config.adminPause': 1 } }, + ); + clearedPauseProjections += cleared.modifiedCount || 0; const integration = await Integration.findOne({ installationId: installationIdFor(installation) }).lean(); if (integration) continue; const result = await InstallableInstallation.updateOne( @@ -67,7 +83,65 @@ const sweepActiveInstallations = async (): Promise => { ); staleComponents += result.modifiedCount || 0; } - return staleComponents; + return { staleComponents, clearedPauseProjections }; +}; + +const pauseProjection = (installation: IInstallableInstallation) => ({ + reason: installation.adminPause?.reason || installation.errorMessage || 'Paused by an administrator.', + at: installation.adminPause?.at || installation.updatedAt, + adminId: installation.adminPause?.adminId || 'system', +}); + +// Pause writes the parent first and the child second. Re-apply the parent +// projection until each child agrees, so a failed child write heals toward the +// stopped state instead of accidentally restoring an inbound relay. +const sweepPausedInstallations = async (): Promise => { + const paused = await InstallableInstallation.find({ status: 'paused' }).lean() as IInstallableInstallation[]; + let restamped = 0; + for (const installation of paused) { + const pause = pauseProjection(installation); + const result = await Integration.updateMany( + { + installationId: installationIdFor(installation), + $or: [ + { 'config.adminPause': { $exists: false } }, + { 'config.adminPause.reason': { $ne: pause.reason } }, + { 'config.adminPause.at': { $ne: pause.at } }, + { 'config.adminPause.adminId': { $ne: pause.adminId } }, + ], + }, + { $set: { 'config.adminPause': pause } }, + ); + restamped += result.modifiedCount || 0; + } + return restamped; +}; + +// Membership is authoritative on outbound selection, but prune obsolete keys +// here as well so the owner's gate list does not promise a pod they left. +const sweepOrphanedGates = async (): Promise => { + const rows = await Integration.find({ + scope: 'user', + 'config.gates': { $exists: true }, + }).select('_id createdBy config.gates').lean() as Array<{ + _id: unknown; + createdBy?: unknown; + config?: { gates?: Record }; + }>; + let pruned = 0; + for (const row of rows) { + const gates = row.config?.gates; + if (!gates || typeof gates !== 'object') continue; + const unset: Record = {}; + for (const podId of Object.keys(gates)) { + const pod = await Pod.findById(podId).select('createdBy members').lean(); + if (!pod || !isPodMember(pod, row.createdBy)) unset[`config.gates.${podId}`] = 1; + } + if (!Object.keys(unset).length) continue; + const result = await Integration.updateOne({ _id: row._id }, { $unset: unset }); + pruned += result.modifiedCount || 0; + } + return pruned; }; const sweepUninstalledInstallations = async (): Promise => { @@ -217,6 +291,9 @@ export const sweep = async (now: Date = new Date()): Promise<{ completed: number; errored: number; staleComponents: number; + clearedPauseProjections: number; + restampedPauses: number; + prunedGates: number; uninstallsCompleted: number; deactivated: number; expiredSlackBinds: number; @@ -224,7 +301,9 @@ export const sweep = async (now: Date = new Date()): Promise<{ unavailableSlackSecretKeys: number; }> => { const locks = await sweepStaleLocks(now); - const staleComponents = await sweepActiveInstallations(); + const active = await sweepActiveInstallations(); + const restampedPauses = await sweepPausedInstallations(); + const prunedGates = await sweepOrphanedGates(); const uninstallsCompleted = await sweepStaleUninstalls(now); const deactivated = await sweepUninstalledInstallations(); const expiredSlackBinds = await sweepExpiredSlackBinds(now); @@ -232,7 +311,10 @@ export const sweep = async (now: Date = new Date()): Promise<{ const unavailableSlackSecretKeys = await sweepUnavailableSlackSecretKeys(); const result = { ...locks, - staleComponents, + staleComponents: active.staleComponents, + clearedPauseProjections: active.clearedPauseProjections, + restampedPauses, + prunedGates, uninstallsCompleted, deactivated, expiredSlackBinds, diff --git a/backend/services/installable/projectors/webhookProjector.ts b/backend/services/installable/projectors/webhookProjector.ts index 78f919bbfa..916b7bfc49 100644 --- a/backend/services/installable/projectors/webhookProjector.ts +++ b/backend/services/installable/projectors/webhookProjector.ts @@ -32,6 +32,11 @@ export const webhookProjector: ComponentProjector = { if (!context.podId) { throw new Error('Webhook projection requires a pod target'); } + const now = new Date(); + const initialGate = { + enabled: true, + since: now, + }; // Projection never mints a code. A partially projected row must remain // impossible for the unauthenticated enable route to redeem. @@ -41,6 +46,7 @@ export const webhookProjector: ComponentProjector = { $setOnInsert: { installationId, podId: context.podId, + scope: 'user', type: provider, status: 'pending', createdBy: context.installedBy, @@ -49,6 +55,7 @@ export const webhookProjector: ComponentProjector = { liveRelay: true, relayAllAgentMessages: true, linkedUserId: String(context.installedBy), + gates: { [String(context.podId)]: initialGate }, }, }, }, diff --git a/backend/services/slackBridgeService.ts b/backend/services/slackBridgeService.ts index 907195ca0f..841da5824a 100644 --- a/backend/services/slackBridgeService.ts +++ b/backend/services/slackBridgeService.ts @@ -16,6 +16,7 @@ interface SlackIntegrationDoc { _id: unknown; installationId?: string; podId: unknown; + scope?: 'pod' | 'user'; type?: string; isActive?: boolean; status?: string; @@ -28,8 +29,10 @@ interface SlackIntegrationDoc { slackUserId?: string; liveRelay?: boolean; relayAllAgentMessages?: boolean; + gates?: Record; leadAgentUsername?: string; relayMutedUntil?: Date | string; + adminPause?: { reason?: string; at?: Date | string; adminId?: string }; relayMap?: Array<{ externalMessageId?: string; tgMessageId?: string; @@ -41,12 +44,15 @@ interface SlackIntegrationDoc { } const isRelayableIntegration = (integration: SlackIntegrationDoc, podId: string): boolean => ( - String(integration.podId) === String(podId) + (integration.scope === 'user' + ? integration.config?.gates?.[String(podId)]?.enabled === true + : String(integration.podId) === String(podId)) && integration.type === 'slack' && integration.isActive === true && integration.status !== 'error' && integration.config?.liveRelay === true && integration.config?.chatType === 'im' + && !integration.config?.adminPause && Boolean(integration.config?.teamId) && Boolean(integration.config?.chatId) && Boolean(integration.config?.botTokenRef) @@ -63,6 +69,7 @@ const findLiveIntegration = async (podId: string): Promise new Date()) return; - if (!shouldEscalate({ content, agentUsername, integration })) return; + if (!shouldEscalate({ content, agentUsername, integration, podId })) return; const [pod, token] = await Promise.all([ Pod.findById(podId).select('name').lean(), @@ -154,6 +161,13 @@ export const relaySlackMessageToPod = async (opts: { const { integration, event } = opts; const rawText = String(event.text || '').trim(); if (!rawText || rawText.startsWith('/')) return { relayed: false }; + if (!integration.podId) { + // A user-scoped connector may have gates without an active inbound + // destination. Fail closed rather than querying/authoring under + // `String(undefined)`. + console.warn('[slack-bridge] inbound dropped — connector has no active pod'); + return { relayed: false }; + } const config = integration.config || {}; if ( !isRelayableIntegration(integration, String(integration.podId)) diff --git a/backend/services/telegramBridgeService.ts b/backend/services/telegramBridgeService.ts index b9fb924f77..86bc91abdf 100644 --- a/backend/services/telegramBridgeService.ts +++ b/backend/services/telegramBridgeService.ts @@ -37,6 +37,7 @@ export interface RelayMapEntry { interface TelegramIntegrationDoc { _id: unknown; podId: unknown; + scope?: 'pod' | 'user'; type?: string; isActive?: boolean; config?: { @@ -47,6 +48,8 @@ interface TelegramIntegrationDoc { leadAgentUsername?: string; relayMap?: RelayMapEntry[]; relayAllAgentMessages?: boolean; + gates?: Record; + adminPause?: { reason?: string; at?: Date | string; adminId?: string }; }; } @@ -90,6 +93,7 @@ const findLiveIntegration = async (podId: unknown): Promise ( - String(integration.podId) === String(podId) + (integration.scope === 'user' + ? integration.config?.gates?.[String(podId)]?.enabled === true + : String(integration.podId) === String(podId)) && (integration.type === undefined || integration.type === 'telegram') && integration.isActive !== false && integration.config?.liveRelay === true && integration.config?.chatType === 'private' + && !integration.config?.adminPause && Boolean(integration.config?.chatId) ); @@ -137,7 +144,7 @@ export const relayAgentMessageToTelegram = async (opts: { // mute means mute; /status shows it, and it self-expires. const mutedUntil = (integration.config as { relayMutedUntil?: Date | string })?.relayMutedUntil; if (mutedUntil && new Date(mutedUntil) > new Date()) return; - if (!shouldEscalate({ content, agentUsername, integration })) return; + if (!shouldEscalate({ content, agentUsername, integration, podId })) return; const botToken = process.env.TELEGRAM_BOT_TOKEN; const chatId = integration.config?.chatId; @@ -184,6 +191,13 @@ export const relayTelegramMessageToPod = async (opts: { const rawText = (telegramMessage.text || telegramMessage.caption || '').trim(); if (!rawText) return { relayed: false }; if (rawText.startsWith('/')) return { relayed: false }; // commands keep legacy handling + if (!integration.podId) { + // `podId` is optional for user-scoped rows, but inbound has one active + // destination. Never stringify null into a query or author into a pod the + // connector cannot name. + console.warn('[tg-bridge] inbound dropped — connector has no active pod'); + return { relayed: false }; + } const podId = String(integration.podId); const linkedUserId = integration.config?.linkedUserId; @@ -347,6 +361,7 @@ export const relayTelegramMessageToPod = async (opts: { module.exports = { shouldEscalate, routeReplyContent, + isRelayableIntegration, relayAgentMessageToTelegram, relayTelegramMessageToPod, }; From ba23a11dce2062e4b21d9e77615ebd86f975508b Mon Sep 17 00:00:00 2001 From: Lily Shen <115414357+lilyshen0722@users.noreply.github.com> Date: Fri, 4 Sep 2026 22:09:47 -0700 Subject: [PATCH 02/11] fix(connectors): keep inbound on active pod --- .../installableInstallationService.test.js | 42 ++++++++++++++++ .../unit/services/slackBridgeService.test.js | 40 ++++++++++++++- .../telegramBridgeService.attribution.test.js | 17 ++++++- ...telegramBridgeService.inboundRetry.test.js | 4 +- .../services/telegramBridgeService.test.js | 37 +++++++++++--- .../installable/installableReconciler.ts | 8 ++- backend/services/slackBridgeService.ts | 49 ++++++++++++++++--- backend/services/telegramBridgeService.ts | 49 ++++++++++++++++--- 8 files changed, 219 insertions(+), 27 deletions(-) diff --git a/backend/__tests__/unit/services/installableInstallationService.test.js b/backend/__tests__/unit/services/installableInstallationService.test.js index c0e5937475..19553244b7 100644 --- a/backend/__tests__/unit/services/installableInstallationService.test.js +++ b/backend/__tests__/unit/services/installableInstallationService.test.js @@ -22,6 +22,8 @@ const { const { sweep } = require('../../../services/installable/installableReconciler'); const { TELEGRAM_CONNECTOR, SLACK_CONNECTOR } = require('../../../scripts/seed-builtin-connectors'); const { put } = require('../../../services/connectorSecrets'); +const telegramService = require('../../../services/telegramService'); +const { relayTelegramMessageToPod } = require('../../../services/telegramBridgeService'); const { setupMongoDb, closeMongoDb, @@ -642,6 +644,46 @@ describe('installable connector projection', () => { expect((await Integration.findById(installed.integration._id)).config.adminPause).toBeUndefined(); }); + it('clears the active pod with its orphaned gate and tells the linked chat why', async () => { + const { userId, podId } = ids(); + await Pod.create({ + _id: podId, + name: 'Departed pod', + type: 'team', + createdBy: new mongoose.Types.ObjectId(), + members: [userId], + }); + const installed = await install({ installableId: 'telegram', installedBy: userId, podId }); + await Integration.updateOne( + { _id: installed.integration._id }, + { $set: { 'config.chatId': 'chat-1', 'config.chatType': 'private' } }, + ); + await Pod.updateOne({ _id: podId }, { $pull: { members: userId } }); + + const reconciled = await sweep(new Date()); + const projection = await Integration.findById(installed.integration._id).lean(); + expect(reconciled.prunedGates).toBe(1); + expect(projection.config.gates?.[podId]).toBeUndefined(); + expect(projection.podId).toBeUndefined(); + + const previousToken = process.env.TELEGRAM_BOT_TOKEN; + process.env.TELEGRAM_BOT_TOKEN = 'bot-token'; + const send = jest.spyOn(telegramService, 'sendMessage').mockResolvedValue({ messageId: 'reply-1' }); + try { + await expect(relayTelegramMessageToPod({ + integration: projection, + telegramMessage: { text: 'where did the pod go?' }, + })).resolves.toEqual({ relayed: false }); + expect(send).toHaveBeenCalledWith( + 'bot-token', 'chat-1', 'This connector has no active pod. Choose one in Commonly first.', + ); + } finally { + send.mockRestore(); + if (previousToken === undefined) delete process.env.TELEGRAM_BOT_TOKEN; + else process.env.TELEGRAM_BOT_TOKEN = previousToken; + } + }); + it('expires a pending Slack bind, revokes its encrypted token, and leaves a retryable active card', async () => { const { userId, podId } = ids(); const installed = await install({ installableId: 'slack', installedBy: userId, podId }); diff --git a/backend/__tests__/unit/services/slackBridgeService.test.js b/backend/__tests__/unit/services/slackBridgeService.test.js index 6e38f04504..6e09cb9cae 100644 --- a/backend/__tests__/unit/services/slackBridgeService.test.js +++ b/backend/__tests__/unit/services/slackBridgeService.test.js @@ -1,14 +1,20 @@ jest.mock('../../../models/Integration', () => ({ findOne: jest.fn(), findByIdAndUpdate: jest.fn() })); jest.mock('../../../models/Pod', () => ({ findById: jest.fn() })); +jest.mock('../../../models/User', () => ({ findById: jest.fn() })); jest.mock('../../../services/connectorSecrets', () => ({ get: jest.fn() })); jest.mock('../../../services/slackApi', () => jest.fn().mockImplementation(() => ({ postMessage: jest.fn() }))); const Integration = require('../../../models/Integration'); const Pod = require('../../../models/Pod'); +const User = require('../../../models/User'); const connectorSecrets = require('../../../services/connectorSecrets'); const SlackApi = require('../../../services/slackApi'); const { - relayAgentMessageToSlack, relaySlackMessageToPod, routeSlackReplyContent, isRelayableIntegration, + relayAgentMessageToSlack, + relaySlackMessageToPod, + routeSlackReplyContent, + isRelayableIntegration, + isInboundRelayableIntegration, } = require('../../../services/slackBridgeService'); const integration = { @@ -76,7 +82,15 @@ describe('Slack installable bridge', () => { }, 'second-pod')).toBe(true); }); - test('drops inbound rather than stringifying a user connector without an active pod', async () => { + test('keeps inbound on the active pod when its outbound gate is disabled', () => { + expect(isInboundRelayableIntegration({ + ...integration, + scope: 'user', + config: { ...integration.config, gates: { 'pod-1': { enabled: false } } }, + }, 'pod-1')).toBe(true); + }); + + test('answers once in Slack when a user connector has no active pod', async () => { const result = await relaySlackMessageToPod({ integration: { ...integration, @@ -88,5 +102,27 @@ describe('Slack installable bridge', () => { }); expect(result).toEqual({ relayed: false }); + expect(connectorSecrets.get).toHaveBeenCalledWith('secret-ref'); + const api = SlackApi.mock.results[0].value; + expect(api.postMessage).toHaveBeenCalledWith( + 'D1', 'This connector has no active pod. Choose one in Commonly first.', + ); + }); + + test('refuses inbound authorship when the linked user left the active pod', async () => { + Pod.findById.mockReturnValue({ + select: jest.fn().mockReturnValue({ lean: jest.fn().mockResolvedValue({ type: 'team', members: [] }) }), + }); + + await expect(relaySlackMessageToPod({ + integration: { + ...integration, + scope: 'user', + config: { ...integration.config, linkedUserId: 'user-1', slackUserId: 'U1' }, + }, + event: { text: 'hello', user: 'U1' }, + })).resolves.toEqual({ relayed: false }); + + expect(User.findById).not.toHaveBeenCalled(); }); }); diff --git a/backend/__tests__/unit/services/telegramBridgeService.attribution.test.js b/backend/__tests__/unit/services/telegramBridgeService.attribution.test.js index 8083c192d8..c8a437fb54 100644 --- a/backend/__tests__/unit/services/telegramBridgeService.attribution.test.js +++ b/backend/__tests__/unit/services/telegramBridgeService.attribution.test.js @@ -42,7 +42,9 @@ describe('relayTelegramMessageToPod — who the pod row is attributed to', () => beforeEach(() => { jest.clearAllMocks(); User.findById.mockReturnValue({ select: () => ({ lean: () => Promise.resolve({ username: 'sam', profilePicture: null }) }) }); - Pod.findById.mockReturnValue({ select: () => ({ lean: () => Promise.resolve({ type: 'chat' }) }) }); + Pod.findById.mockReturnValue({ + select: () => ({ lean: () => Promise.resolve({ type: 'chat', members: ['user-1'] }) }), + }); PGPod.findById.mockResolvedValue({ id: 'pod-1' }); PGMessage.create.mockResolvedValue({ id: 'm1', content: 'x' }); PGMessage.findById.mockResolvedValue({ id: 'm1', content: 'x' }); @@ -99,6 +101,19 @@ describe('relayTelegramMessageToPod — who the pod row is attributed to', () => expect(PGPod.findById).not.toHaveBeenCalled(); expect(User.findById).not.toHaveBeenCalled(); }); + + it('refuses an active pod after the linked user leaves it', async () => { + Pod.findById.mockReturnValue({ + select: () => ({ lean: () => Promise.resolve({ type: 'chat', members: [] }) }), + }); + + await expect(relayTelegramMessageToPod({ + integration: withChatType('private'), telegramMessage: messageFromSomeoneElse, + })).resolves.toEqual({ relayed: false }); + + expect(User.findById).not.toHaveBeenCalled(); + expect(PGMessage.create).not.toHaveBeenCalled(); + }); }); // Outbound mirror of the inbound gate (connector-verify F2, 2026-08-26): a diff --git a/backend/__tests__/unit/services/telegramBridgeService.inboundRetry.test.js b/backend/__tests__/unit/services/telegramBridgeService.inboundRetry.test.js index 7ffb6bc347..2204929df0 100644 --- a/backend/__tests__/unit/services/telegramBridgeService.inboundRetry.test.js +++ b/backend/__tests__/unit/services/telegramBridgeService.inboundRetry.test.js @@ -30,7 +30,9 @@ describe('relayTelegramMessageToPod — failures either side of the pod write', beforeEach(() => { jest.clearAllMocks(); User.findById.mockReturnValue({ select: () => ({ lean: () => Promise.resolve({ username: 'sam', profilePicture: null }) }) }); - Pod.findById.mockReturnValue({ select: () => ({ lean: () => Promise.resolve({ type: 'chat' }) }) }); + Pod.findById.mockReturnValue({ + select: () => ({ lean: () => Promise.resolve({ type: 'chat', members: ['user-1'] }) }), + }); PGPod.findById.mockResolvedValue({ id: 'pod-1' }); PGMessage.create.mockResolvedValue({ id: 'm1', content: 'x' }); PGMessage.findById.mockResolvedValue({ id: 'm1', content: 'x' }); diff --git a/backend/__tests__/unit/services/telegramBridgeService.test.js b/backend/__tests__/unit/services/telegramBridgeService.test.js index 4a341732e3..f24751ab3c 100644 --- a/backend/__tests__/unit/services/telegramBridgeService.test.js +++ b/backend/__tests__/unit/services/telegramBridgeService.test.js @@ -4,8 +4,13 @@ jest.mock('../../../models/Integration', () => ({ })); jest.mock('../../../services/telegramService', () => ({ sendMessage: jest.fn() })); +const telegramSend = require('../../../services/telegramService'); const { - shouldEscalate, routeReplyContent, isRelayableIntegration, relayTelegramMessageToPod, + shouldEscalate, + routeReplyContent, + isRelayableIntegration, + isInboundRelayableIntegration, + relayTelegramMessageToPod, } = require('../../../services/telegramBridgeService'); describe('telegramBridgeService — escalation gate', () => { @@ -144,12 +149,30 @@ describe('telegramBridgeService — user-scope outbound gate', () => { expect(isRelayableIntegration(userScoped, 'missing-pod')).toBe(false); }); - it('drops inbound rather than stringifying a user connector without an active pod', async () => { - const result = await relayTelegramMessageToPod({ - integration: { ...userScoped, podId: undefined }, - telegramMessage: { text: 'hello from the phone' }, - }); + it('keeps inbound on the active pod when its outbound gate is disabled', () => { + expect(isInboundRelayableIntegration({ + ...userScoped, + config: { ...userScoped.config, gates: { 'active-pod': { enabled: false } } }, + }, 'active-pod')).toBe(true); + }); - expect(result).toEqual({ relayed: false }); + it('answers once in the linked chat when a user connector has no active pod', async () => { + const originalToken = process.env.TELEGRAM_BOT_TOKEN; + process.env.TELEGRAM_BOT_TOKEN = 'bot-token'; + try { + const result = await relayTelegramMessageToPod({ + integration: { ...userScoped, podId: undefined }, + telegramMessage: { text: 'hello from the phone' }, + }); + + expect(result).toEqual({ relayed: false }); + expect(telegramSend.sendMessage).toHaveBeenCalledTimes(1); + expect(telegramSend.sendMessage).toHaveBeenCalledWith( + 'bot-token', 'chat-1', 'This connector has no active pod. Choose one in Commonly first.', + ); + } finally { + if (originalToken === undefined) delete process.env.TELEGRAM_BOT_TOKEN; + else process.env.TELEGRAM_BOT_TOKEN = originalToken; + } }); }); diff --git a/backend/services/installable/installableReconciler.ts b/backend/services/installable/installableReconciler.ts index b877b74edf..c304278613 100644 --- a/backend/services/installable/installableReconciler.ts +++ b/backend/services/installable/installableReconciler.ts @@ -123,9 +123,10 @@ const sweepOrphanedGates = async (): Promise => { const rows = await Integration.find({ scope: 'user', 'config.gates': { $exists: true }, - }).select('_id createdBy config.gates').lean() as Array<{ + }).select('_id createdBy podId config.gates').lean() as Array<{ _id: unknown; createdBy?: unknown; + podId?: unknown; config?: { gates?: Record }; }>; let pruned = 0; @@ -135,7 +136,10 @@ const sweepOrphanedGates = async (): Promise => { const unset: Record = {}; for (const podId of Object.keys(gates)) { const pod = await Pod.findById(podId).select('createdBy members').lean(); - if (!pod || !isPodMember(pod, row.createdBy)) unset[`config.gates.${podId}`] = 1; + if (!pod || !isPodMember(pod, row.createdBy)) { + unset[`config.gates.${podId}`] = 1; + if (String(row.podId) === podId) unset.podId = 1; + } } if (!Object.keys(unset).length) continue; const result = await Integration.updateOne({ _id: row._id }, { $unset: unset }); diff --git a/backend/services/slackBridgeService.ts b/backend/services/slackBridgeService.ts index 841da5824a..100dcd5da3 100644 --- a/backend/services/slackBridgeService.ts +++ b/backend/services/slackBridgeService.ts @@ -5,6 +5,8 @@ const Integration = require('../models/Integration'); // eslint-disable-next-line @typescript-eslint/no-require-imports, global-require const Pod = require('../models/Pod'); // eslint-disable-next-line @typescript-eslint/no-require-imports, global-require +const isPodMember = require('../utils/isPodMember'); +// eslint-disable-next-line @typescript-eslint/no-require-imports, global-require const connectorSecrets = require('./connectorSecrets'); // eslint-disable-next-line @typescript-eslint/no-require-imports, global-require const { shouldEscalate } = require('./connectorRelayPolicy'); @@ -58,6 +60,35 @@ const isRelayableIntegration = (integration: SlackIntegrationDoc, podId: string) && Boolean(integration.config?.botTokenRef) ); +// The enabled gate is only the pod → connector subscription. Inbound follows +// the owner's active pod selection and validates membership at receive time. +const isInboundRelayableIntegration = (integration: SlackIntegrationDoc, podId: string): boolean => ( + String(integration.podId) === String(podId) + && integration.type === 'slack' + && integration.isActive === true + && integration.status !== 'error' + && integration.config?.liveRelay === true + && integration.config?.chatType === 'im' + && !integration.config?.adminPause + && Boolean(integration.config?.teamId) + && Boolean(integration.config?.chatId) + && Boolean(integration.config?.botTokenRef) +); + +const NO_ACTIVE_POD_REPLY = 'This connector has no active pod. Choose one in Commonly first.'; + +const replyNoActivePod = async (integration: SlackIntegrationDoc): Promise => { + const chatId = integration.config?.chatId; + const botTokenRef = integration.config?.botTokenRef; + if (!chatId || !botTokenRef) return; + try { + const token = await connectorSecrets.get(String(botTokenRef)); + await new SlackApi(token).postMessage(String(chatId), NO_ACTIVE_POD_REPLY); + } catch (error) { + console.warn('[slack-bridge] could not send no-active-pod reply:', (error as Error).message); + } +}; + const findLiveIntegration = async (podId: string): Promise => ( Integration.findOne({ type: 'slack', @@ -166,11 +197,12 @@ export const relaySlackMessageToPod = async (opts: { // destination. Fail closed rather than querying/authoring under // `String(undefined)`. console.warn('[slack-bridge] inbound dropped — connector has no active pod'); + await replyNoActivePod(integration); return { relayed: false }; } const config = integration.config || {}; if ( - !isRelayableIntegration(integration, String(integration.podId)) + !isInboundRelayableIntegration(integration, String(integration.podId)) || !config.linkedUserId || !config.slackUserId || event.user !== config.slackUserId @@ -204,12 +236,14 @@ export const relaySlackMessageToPod = async (opts: { // eslint-disable-next-line @typescript-eslint/no-require-imports, global-require const socketConfig = require('../config/socket'); - const [linkedUser, pod] = await Promise.all([ - User.findById(linkedUserId).select('username profilePicture').lean(), - PodModel.findById(podId).select('type').lean(), - ]); - if (!linkedUser || !pod) { - console.warn('[slack-bridge] inbound dropped — linked user or pod missing'); + const pod = await PodModel.findById(podId).select('type createdBy members').lean(); + if (!pod || !isPodMember(pod, linkedUserId)) { + console.warn('[slack-bridge] inbound dropped — linked user is no longer a pod member'); + return { relayed: false }; + } + const linkedUser = await User.findById(linkedUserId).select('username profilePicture').lean(); + if (!linkedUser) { + console.warn('[slack-bridge] inbound dropped — linked user missing'); return { relayed: false }; } try { @@ -275,4 +309,5 @@ module.exports = { relaySlackMessageToPod, routeSlackReplyContent, isRelayableIntegration, + isInboundRelayableIntegration, }; diff --git a/backend/services/telegramBridgeService.ts b/backend/services/telegramBridgeService.ts index 86bc91abdf..dcdf872f95 100644 --- a/backend/services/telegramBridgeService.ts +++ b/backend/services/telegramBridgeService.ts @@ -21,6 +21,8 @@ // eslint-disable-next-line @typescript-eslint/no-require-imports, global-require const IntegrationModel = require('../models/Integration'); // eslint-disable-next-line @typescript-eslint/no-require-imports, global-require +const isPodMember = require('../utils/isPodMember'); +// eslint-disable-next-line @typescript-eslint/no-require-imports, global-require const telegramSend = require('./telegramService'); // eslint-disable-next-line @typescript-eslint/no-require-imports, global-require const { shouldEscalate } = require('./connectorRelayPolicy'); @@ -121,6 +123,35 @@ const isRelayableIntegration = ( && Boolean(integration.config?.chatId) ); +// Gates select outbound subscriptions. A connector has one active inbound +// destination, held by podId; do not let a disabled or pruned outbound gate +// silently disconnect the owner's messages from that active pod. +const isInboundRelayableIntegration = ( + integration: TelegramIntegrationDoc, + podId: string, +): boolean => ( + String(integration.podId) === String(podId) + && (integration.type === undefined || integration.type === 'telegram') + && integration.isActive !== false + && integration.config?.liveRelay === true + && integration.config?.chatType === 'private' + && !integration.config?.adminPause + && Boolean(integration.config?.chatId) +); + +const NO_ACTIVE_POD_REPLY = 'This connector has no active pod. Choose one in Commonly first.'; + +const replyNoActivePod = async (integration: TelegramIntegrationDoc): Promise => { + const botToken = process.env.TELEGRAM_BOT_TOKEN; + const chatId = integration.config?.chatId; + if (!botToken || !chatId) return; + try { + await telegramSend.sendMessage(botToken, chatId, NO_ACTIVE_POD_REPLY); + } catch (error) { + console.warn('[tg-bridge] could not send no-active-pod reply:', (error as Error).message); + } +}; + // Outbound: agent message → Telegram, attributed, with a deep link back into // the pod. Fire-and-forget from AgentMessageService.postMessage — a bridge // failure must never fail the post itself. @@ -196,12 +227,13 @@ export const relayTelegramMessageToPod = async (opts: { // destination. Never stringify null into a query or author into a pod the // connector cannot name. console.warn('[tg-bridge] inbound dropped — connector has no active pod'); + await replyNoActivePod(integration); return { relayed: false }; } const podId = String(integration.podId); const linkedUserId = integration.config?.linkedUserId; - if (!linkedUserId) { + if (!linkedUserId || !isInboundRelayableIntegration(integration, podId)) { console.warn('[tg-bridge] live relay without linkedUserId — inbound dropped'); return { relayed: false }; } @@ -276,12 +308,14 @@ export const relayTelegramMessageToPod = async (opts: { // eslint-disable-next-line @typescript-eslint/no-require-imports, global-require const socketConfig = require('../config/socket'); - const [linkedUser, pod] = await Promise.all([ - User.findById(linkedUserId).select('username profilePicture').lean(), - Pod.findById(podId).select('type').lean(), - ]); - if (!linkedUser || !pod) { - console.warn('[tg-bridge] inbound dropped — linked user or pod missing'); + const pod = await Pod.findById(podId).select('type createdBy members').lean(); + if (!pod || !isPodMember(pod, linkedUserId)) { + console.warn('[tg-bridge] inbound dropped — linked user is no longer a pod member'); + return { relayed: false }; + } + const linkedUser = await User.findById(linkedUserId).select('username profilePicture').lean(); + if (!linkedUser) { + console.warn('[tg-bridge] inbound dropped — linked user missing'); return { relayed: false }; } @@ -362,6 +396,7 @@ module.exports = { shouldEscalate, routeReplyContent, isRelayableIntegration, + isInboundRelayableIntegration, relayAgentMessageToTelegram, relayTelegramMessageToPod, }; From 196089161d981b86257861135f956a16429a51a1 Mon Sep 17 00:00:00 2001 From: Lily Shen <115414357+lilyshen0722@users.noreply.github.com> Date: Fri, 4 Sep 2026 22:13:02 -0700 Subject: [PATCH 03/11] fix(connectors): restore active pod routing --- .../routes/integrations.linkedUserId.test.js | 45 +++++++++++++++++++ .../unit/services/slackBridgeService.test.js | 5 +++ .../telegramBridgeService.attribution.test.js | 21 ++++++--- backend/routes/integrations.ts | 26 ++++++++++- backend/services/slackBridgeService.ts | 1 + backend/services/telegramBridgeService.ts | 1 + 6 files changed, 93 insertions(+), 6 deletions(-) diff --git a/backend/__tests__/unit/routes/integrations.linkedUserId.test.js b/backend/__tests__/unit/routes/integrations.linkedUserId.test.js index 5748fc6d01..9e036f1a5a 100644 --- a/backend/__tests__/unit/routes/integrations.linkedUserId.test.js +++ b/backend/__tests__/unit/routes/integrations.linkedUserId.test.js @@ -127,6 +127,15 @@ describe('PATCH /api/integrations/:id — linkedUserId guard', () => { expect(update.config.liveRelay).toBe(false); expect(update.config.linkedUserId).toBeUndefined(); }); + + it('does not let a pod-scoped connector move its installation-owned pod', async () => { + const res = await request(app) + .patch('/api/integrations/integration-1') + .send({ podId: '64b64c1f7e5b8f0a12345671' }); + + expect(res.status).toBe(400); + expect(Integration.findByIdAndUpdate).not.toHaveBeenCalled(); + }); }); describe('PATCH /api/integrations/:id — user-scoped connector gates', () => { @@ -205,6 +214,42 @@ describe('PATCH /api/integrations/:id — user-scoped connector gates', () => { expect(update.config.gates[allowedPodId]).toMatchObject({ enabled: true }); }); + it('allows the linked owner to select a member pod as the active inbound destination', async () => { + Pod.findById.mockResolvedValue({ _id: allowedPodId, createdBy: 'user-1', members: [] }); + + const res = await request(app) + .patch('/api/integrations/integration-user') + .send({ podId: allowedPodId }); + + expect(res.status).toBe(200); + const [, update] = Integration.findByIdAndUpdate.mock.calls[0]; + expect(update.podId).toBe(allowedPodId); + }); + + it('refuses selecting an active pod the linked owner is no longer a member of', async () => { + Pod.findById.mockResolvedValue({ _id: forbiddenPodId, createdBy: 'someone-else', members: [] }); + + const res = await request(app) + .patch('/api/integrations/integration-user') + .send({ podId: forbiddenPodId }); + + expect(res.status).toBe(403); + expect(Integration.findByIdAndUpdate).not.toHaveBeenCalled(); + }); + + it.each(['not-an-object-id', '64b64c1f7e5b8f0a1234567.', '$ne'])( + 'refuses an unsafe active pod id %p before membership lookup or write', + async (podId) => { + const res = await request(app) + .patch('/api/integrations/integration-user') + .send({ podId }); + + expect(res.status).toBe(400); + expect(Pod.findById).not.toHaveBeenCalled(); + expect(Integration.findByIdAndUpdate).not.toHaveBeenCalled(); + }, + ); + it.each(['a.b', '$ne'])('refuses an unsafe gate key before any membership lookup or write', async (key) => { const res = await request(app) .patch('/api/integrations/integration-user') diff --git a/backend/__tests__/unit/services/slackBridgeService.test.js b/backend/__tests__/unit/services/slackBridgeService.test.js index 6e09cb9cae..dd8f7d5ec9 100644 --- a/backend/__tests__/unit/services/slackBridgeService.test.js +++ b/backend/__tests__/unit/services/slackBridgeService.test.js @@ -124,5 +124,10 @@ describe('Slack installable bridge', () => { })).resolves.toEqual({ relayed: false }); expect(User.findById).not.toHaveBeenCalled(); + expect(connectorSecrets.get).toHaveBeenCalledWith('secret-ref'); + const api = SlackApi.mock.results[0].value; + expect(api.postMessage).toHaveBeenCalledWith( + 'D1', 'This connector has no active pod. Choose one in Commonly first.', + ); }); }); diff --git a/backend/__tests__/unit/services/telegramBridgeService.attribution.test.js b/backend/__tests__/unit/services/telegramBridgeService.attribution.test.js index c8a437fb54..1507204ae5 100644 --- a/backend/__tests__/unit/services/telegramBridgeService.attribution.test.js +++ b/backend/__tests__/unit/services/telegramBridgeService.attribution.test.js @@ -15,6 +15,7 @@ jest.mock('../../../config/socket', () => ({ getIO: jest.fn(() => null) })); const User = require('../../../models/User'); const Pod = require('../../../models/Pod'); const PGMessage = require('../../../models/pg/Message'); +const telegramSend = require('../../../services/telegramService'); const PGPod = require('../../../models/pg/Pod'); const { deliverMessageToAgents } = require('../../../services/messageAgentDeliveryService'); const { relayTelegramMessageToPod } = require('../../../services/telegramBridgeService'); @@ -103,16 +104,26 @@ describe('relayTelegramMessageToPod — who the pod row is attributed to', () => }); it('refuses an active pod after the linked user leaves it', async () => { + const originalToken = process.env.TELEGRAM_BOT_TOKEN; + process.env.TELEGRAM_BOT_TOKEN = 'bot-token'; Pod.findById.mockReturnValue({ select: () => ({ lean: () => Promise.resolve({ type: 'chat', members: [] }) }), }); - await expect(relayTelegramMessageToPod({ - integration: withChatType('private'), telegramMessage: messageFromSomeoneElse, - })).resolves.toEqual({ relayed: false }); + try { + await expect(relayTelegramMessageToPod({ + integration: withChatType('private'), telegramMessage: messageFromSomeoneElse, + })).resolves.toEqual({ relayed: false }); - expect(User.findById).not.toHaveBeenCalled(); - expect(PGMessage.create).not.toHaveBeenCalled(); + expect(User.findById).not.toHaveBeenCalled(); + expect(PGMessage.create).not.toHaveBeenCalled(); + expect(telegramSend.sendMessage).toHaveBeenCalledWith( + 'bot-token', '42', 'This connector has no active pod. Choose one in Commonly first.', + ); + } finally { + if (originalToken === undefined) delete process.env.TELEGRAM_BOT_TOKEN; + else process.env.TELEGRAM_BOT_TOKEN = originalToken; + } }); }); diff --git a/backend/routes/integrations.ts b/backend/routes/integrations.ts index a558b05635..6d3b756535 100644 --- a/backend/routes/integrations.ts +++ b/backend/routes/integrations.ts @@ -492,7 +492,12 @@ router.post('/:id/connect-code', writeIntegrationsRateLimit, auth, async (req: A router.patch('/:id', auth, async (req: AuthReq, res: Res) => { try { const { id } = req.params || {}; - const { config, status, isActive } = (req.body || {}) as { config?: Record; status?: string; isActive?: boolean }; + const { config, status, isActive, podId } = (req.body || {}) as { + config?: Record; + status?: string; + isActive?: boolean; + podId?: unknown; + }; const integration = await Integration.findById(id) as { type?: string; scope?: string; @@ -516,6 +521,19 @@ router.patch('/:id', auth, async (req: AuthReq, res: Res) => { if (config && Object.prototype.hasOwnProperty.call(config, 'adminPause')) { return res.status(400).json({ message: 'adminPause is managed by an administrator' }); } + // A user connector has one active inbound destination. Selecting it is + // an owner action, and it must be a pod that owner can still write to; + // otherwise a browser could redirect private inbound messages into a + // pod it merely knows the id of. + if (podId !== undefined) { + if (typeof podId !== 'string' || !isObjectIdKey(podId)) { + return res.status(400).json({ message: 'podId must be a valid pod id' }); + } + const activePod = await Pod.findById(podId); + if (!activePod || !isPodMember(activePod, requesterId)) { + return res.status(403).json({ message: 'Access denied' }); + } + } const requestedGates = config?.gates; if (requestedGates !== undefined) { if (!requestedGates || Array.isArray(requestedGates) || typeof requestedGates !== 'object') { @@ -541,6 +559,11 @@ router.patch('/:id', auth, async (req: AuthReq, res: Res) => { } } } else { + // A pod-scoped integration is anchored by its installation. Only a + // user-scoped connector exposes a movable personal inbound destination. + if (podId !== undefined) { + return res.status(400).json({ message: 'podId is only managed on user-scoped connectors' }); + } const canUpdate = await canDeleteIntegration(integration, requesterId); if (!canUpdate) return res.status(403).json({ message: 'Access denied' }); } @@ -569,6 +592,7 @@ router.patch('/:id', auth, async (req: AuthReq, res: Res) => { validateManifestIfComplete(integration.type || '', nextConfig); const update: Record = {}; if (config) update.config = nextConfig; + if (podId !== undefined) update.podId = podId; if (typeof status === 'string') update.status = status; if (typeof isActive === 'boolean') update.isActive = isActive; const autoStatusTypes = ['groupme', 'telegram', 'slack', 'x', 'instagram']; diff --git a/backend/services/slackBridgeService.ts b/backend/services/slackBridgeService.ts index 100dcd5da3..f0829aeb0c 100644 --- a/backend/services/slackBridgeService.ts +++ b/backend/services/slackBridgeService.ts @@ -239,6 +239,7 @@ export const relaySlackMessageToPod = async (opts: { const pod = await PodModel.findById(podId).select('type createdBy members').lean(); if (!pod || !isPodMember(pod, linkedUserId)) { console.warn('[slack-bridge] inbound dropped — linked user is no longer a pod member'); + await replyNoActivePod(integration); return { relayed: false }; } const linkedUser = await User.findById(linkedUserId).select('username profilePicture').lean(); diff --git a/backend/services/telegramBridgeService.ts b/backend/services/telegramBridgeService.ts index dcdf872f95..2dd47ebfe5 100644 --- a/backend/services/telegramBridgeService.ts +++ b/backend/services/telegramBridgeService.ts @@ -311,6 +311,7 @@ export const relayTelegramMessageToPod = async (opts: { const pod = await Pod.findById(podId).select('type createdBy members').lean(); if (!pod || !isPodMember(pod, linkedUserId)) { console.warn('[tg-bridge] inbound dropped — linked user is no longer a pod member'); + await replyNoActivePod(integration); return { relayed: false }; } const linkedUser = await User.findById(linkedUserId).select('username profilePicture').lean(); From f608825811b8ab3c5502c2510642a0b0f5be4b29 Mon Sep 17 00:00:00 2001 From: Lily Shen <115414357+lilyshen0722@users.noreply.github.com> Date: Fri, 4 Sep 2026 22:18:19 -0700 Subject: [PATCH 04/11] fix(connectors): harden control-plane routes --- .../routes/integrations.linkedUserId.test.js | 61 ++++++++++++------- backend/routes/admin/installables.ts | 38 ++++++++++-- backend/routes/integrations.ts | 6 +- 3 files changed, 78 insertions(+), 27 deletions(-) diff --git a/backend/__tests__/unit/routes/integrations.linkedUserId.test.js b/backend/__tests__/unit/routes/integrations.linkedUserId.test.js index 9e036f1a5a..3327291dfe 100644 --- a/backend/__tests__/unit/routes/integrations.linkedUserId.test.js +++ b/backend/__tests__/unit/routes/integrations.linkedUserId.test.js @@ -12,6 +12,10 @@ jest.mock('../../../middleware/auth', () => (req, res, next) => { next(); }); jest.mock('../../../middleware/adminAuth', () => (req, res, next) => next()); +jest.mock('../../../middleware/integrationRateLimit', () => ({ + writeIntegrationsRateLimit: (_req, _res, next) => next(), + listIntegrationsRateLimit: (_req, _res, next) => next(), +})); jest.mock('../../../models/Pod', () => ({ findById: jest.fn() })); jest.mock('../../../models/User', () => ({ findById: jest.fn() })); jest.mock('../../../models/DiscordIntegration', () => function DiscordIntegration(data) { @@ -40,6 +44,9 @@ const app = express(); app.use(express.json()); app.use('/api/integrations', integrationRoutes); +const integrationId = '64b64c1f7e5b8f0a12345674'; +const userScopedIntegrationId = '64b64c1f7e5b8f0a12345675'; + const telegramIntegration = () => ({ _id: 'integration-1', type: 'telegram', @@ -62,9 +69,19 @@ describe('PATCH /api/integrations/:id — linkedUserId guard', () => { Integration.findByIdAndUpdate.mockResolvedValue({ _id: 'integration-1' }); }); + it('rejects an invalid integration id before any database access', async () => { + const res = await request(app) + .patch('/api/integrations/$ne') + .send({ config: { liveRelay: true } }); + + expect(res.status).toBe(400); + expect(Integration.findById).not.toHaveBeenCalled(); + expect(Integration.findByIdAndUpdate).not.toHaveBeenCalled(); + }); + it('rejects a client-supplied linkedUserId naming someone else', async () => { const res = await request(app) - .patch('/api/integrations/integration-1') + .patch(`/api/integrations/${integrationId}`) .send({ config: { liveRelay: true, linkedUserId: 'VICTIM-USER-ID' } }); expect(res.status).toBe(400); @@ -73,7 +90,7 @@ describe('PATCH /api/integrations/:id — linkedUserId guard', () => { it('derives linkedUserId from the caller when liveRelay flips on', async () => { const res = await request(app) - .patch('/api/integrations/integration-1') + .patch(`/api/integrations/${integrationId}`) .send({ config: { liveRelay: true } }); expect(res.status).toBe(200); @@ -84,7 +101,7 @@ describe('PATCH /api/integrations/:id — linkedUserId guard', () => { it("derives linkedUserId when liveRelay arrives as the string 'true' (#1293)", async () => { const res = await request(app) - .patch('/api/integrations/integration-1') + .patch(`/api/integrations/${integrationId}`) .send({ config: { liveRelay: 'true' } }); expect(res.status).toBe(200); @@ -101,7 +118,7 @@ describe('PATCH /api/integrations/:id — linkedUserId guard', () => { 'refuses liveRelay %p with 400 instead of letting Mongoose cast it', async (value) => { const res = await request(app) - .patch('/api/integrations/integration-1') + .patch(`/api/integrations/${integrationId}`) .send({ config: { liveRelay: value } }); expect(res.status).toBe(400); expect(res.body.message).toMatch(/liveRelay must be true or false/); @@ -111,7 +128,7 @@ describe('PATCH /api/integrations/:id — linkedUserId guard', () => { it.each([1, '1', 'yes'])('refuses relayAllAgentMessages %p the same way', async (value) => { const res = await request(app) - .patch('/api/integrations/integration-1') + .patch(`/api/integrations/${integrationId}`) .send({ config: { relayAllAgentMessages: value } }); expect(res.status).toBe(400); expect(Integration.findByIdAndUpdate).not.toHaveBeenCalled(); @@ -119,7 +136,7 @@ describe('PATCH /api/integrations/:id — linkedUserId guard', () => { it('does not stamp linkedUserId when liveRelay is switched off', async () => { const res = await request(app) - .patch('/api/integrations/integration-1') + .patch(`/api/integrations/${integrationId}`) .send({ config: { liveRelay: false } }); expect(res.status).toBe(200); @@ -130,7 +147,7 @@ describe('PATCH /api/integrations/:id — linkedUserId guard', () => { it('does not let a pod-scoped connector move its installation-owned pod', async () => { const res = await request(app) - .patch('/api/integrations/integration-1') + .patch(`/api/integrations/${integrationId}`) .send({ podId: '64b64c1f7e5b8f0a12345671' }); expect(res.status).toBe(400); @@ -169,13 +186,13 @@ describe('PATCH /api/integrations/:id — user-scoped connector gates', () => { it('refuses a pod creator or admin who is not the linked owner', async () => { const podCreator = await request(app) - .patch('/api/integrations/integration-user') + .patch(`/api/integrations/${userScopedIntegrationId}`) .set('x-test-user', 'pod-creator') .send({ config: { liveRelay: false } }); expect(podCreator.status).toBe(403); const admin = await request(app) - .patch('/api/integrations/integration-user') + .patch(`/api/integrations/${userScopedIntegrationId}`) .set('x-test-user', 'instance-admin') .send({ config: { gates: {} } }); expect(admin.status).toBe(403); @@ -188,7 +205,7 @@ describe('PATCH /api/integrations/:id — user-scoped connector gates', () => { .mockResolvedValueOnce({ _id: forbiddenPodId, createdBy: 'someone-else', members: [] }); const res = await request(app) - .patch('/api/integrations/integration-user') + .patch(`/api/integrations/${userScopedIntegrationId}`) .send({ config: { gates: { @@ -206,7 +223,7 @@ describe('PATCH /api/integrations/:id — user-scoped connector gates', () => { Pod.findById.mockResolvedValue({ _id: allowedPodId, createdBy: 'user-1', members: [] }); const res = await request(app) - .patch('/api/integrations/integration-user') + .patch(`/api/integrations/${userScopedIntegrationId}`) .send({ config: { gates: { [allowedPodId]: { enabled: true, since: new Date().toISOString() } } } }); expect(res.status).toBe(200); @@ -218,7 +235,7 @@ describe('PATCH /api/integrations/:id — user-scoped connector gates', () => { Pod.findById.mockResolvedValue({ _id: allowedPodId, createdBy: 'user-1', members: [] }); const res = await request(app) - .patch('/api/integrations/integration-user') + .patch(`/api/integrations/${userScopedIntegrationId}`) .send({ podId: allowedPodId }); expect(res.status).toBe(200); @@ -230,7 +247,7 @@ describe('PATCH /api/integrations/:id — user-scoped connector gates', () => { Pod.findById.mockResolvedValue({ _id: forbiddenPodId, createdBy: 'someone-else', members: [] }); const res = await request(app) - .patch('/api/integrations/integration-user') + .patch(`/api/integrations/${userScopedIntegrationId}`) .send({ podId: forbiddenPodId }); expect(res.status).toBe(403); @@ -241,7 +258,7 @@ describe('PATCH /api/integrations/:id — user-scoped connector gates', () => { 'refuses an unsafe active pod id %p before membership lookup or write', async (podId) => { const res = await request(app) - .patch('/api/integrations/integration-user') + .patch(`/api/integrations/${userScopedIntegrationId}`) .send({ podId }); expect(res.status).toBe(400); @@ -252,7 +269,7 @@ describe('PATCH /api/integrations/:id — user-scoped connector gates', () => { it.each(['a.b', '$ne'])('refuses an unsafe gate key before any membership lookup or write', async (key) => { const res = await request(app) - .patch('/api/integrations/integration-user') + .patch(`/api/integrations/${userScopedIntegrationId}`) .send({ config: { gates: { [key]: { enabled: true } } } }); expect(res.status).toBe(400); @@ -265,7 +282,7 @@ describe('PATCH /api/integrations/:id — user-scoped connector gates', () => { index.toString(16).padStart(24, '0'), { enabled: true }, ])); const res = await request(app) - .patch('/api/integrations/integration-user') + .patch(`/api/integrations/${userScopedIntegrationId}`) .send({ config: { gates } }); expect(res.status).toBe(400); @@ -277,7 +294,7 @@ describe('PATCH /api/integrations/:id — user-scoped connector gates', () => { Pod.findById.mockResolvedValue(null); const missingPodId = '64b64c1f7e5b8f0a12345673'; const res = await request(app) - .patch('/api/integrations/integration-user') + .patch(`/api/integrations/${userScopedIntegrationId}`) .send({ config: { gates: { [missingPodId]: { enabled: true } } } }); expect(res.status).toBe(403); @@ -286,7 +303,7 @@ describe('PATCH /api/integrations/:id — user-scoped connector gates', () => { it('refuses client writes to the administrator pause projection', async () => { const res = await request(app) - .patch('/api/integrations/integration-user') + .patch(`/api/integrations/${userScopedIntegrationId}`) .send({ config: { adminPause: { reason: 'not owner controlled' } } }); expect(res.status).toBe(400); @@ -363,7 +380,7 @@ describe('PATCH /api/integrations/:id — live relay on a group chat', () => { it('refuses to flip liveRelay on when the bound chat is not private', async () => { const res = await request(app) - .patch('/api/integrations/integration-1') + .patch(`/api/integrations/${integrationId}`) .send({ config: { liveRelay: true } }); expect(res.status).toBe(400); expect(Integration.findByIdAndUpdate).not.toHaveBeenCalled(); @@ -371,7 +388,7 @@ describe('PATCH /api/integrations/:id — live relay on a group chat', () => { it("refuses the same flip when liveRelay arrives as the string 'true' (#1293)", async () => { const res = await request(app) - .patch('/api/integrations/integration-1') + .patch(`/api/integrations/${integrationId}`) .send({ config: { liveRelay: 'true' } }); expect(res.status).toBe(400); expect(Integration.findByIdAndUpdate).not.toHaveBeenCalled(); @@ -379,7 +396,7 @@ describe('PATCH /api/integrations/:id — live relay on a group chat', () => { it.each([1, '1', 'yes'])('refuses the flip when liveRelay is %p on a group', async (value) => { const res = await request(app) - .patch('/api/integrations/integration-1') + .patch(`/api/integrations/${integrationId}`) .send({ config: { liveRelay: value } }); expect(res.status).toBe(400); expect(Integration.findByIdAndUpdate).not.toHaveBeenCalled(); @@ -387,7 +404,7 @@ describe('PATCH /api/integrations/:id — live relay on a group chat', () => { it('ignores client-supplied chatId on PATCH', async () => { const res = await request(app) - .patch('/api/integrations/integration-1') + .patch(`/api/integrations/${integrationId}`) .send({ config: { chatId: '777', leadAgentUsername: 'theo' } }); expect(res.status).toBe(200); const [, update] = Integration.findByIdAndUpdate.mock.calls[0]; diff --git a/backend/routes/admin/installables.ts b/backend/routes/admin/installables.ts index e7c4cc3bac..f4cd5e7432 100644 --- a/backend/routes/admin/installables.ts +++ b/backend/routes/admin/installables.ts @@ -5,6 +5,8 @@ // eslint-disable-next-line @typescript-eslint/no-require-imports, global-require const express = require('express'); // eslint-disable-next-line @typescript-eslint/no-require-imports, global-require +const rateLimit = require('express-rate-limit'); +// eslint-disable-next-line @typescript-eslint/no-require-imports, global-require const auth = require('../../middleware/auth'); // eslint-disable-next-line @typescript-eslint/no-require-imports, global-require const adminAuth = require('../../middleware/adminAuth'); @@ -13,6 +15,8 @@ const InstallableInstallation = require('../../models/InstallableInstallation'); // eslint-disable-next-line @typescript-eslint/no-require-imports, global-require const Integration = require('../../models/Integration'); import { Types } from 'mongoose'; +// eslint-disable-next-line @typescript-eslint/no-require-imports, global-require +const { cloudflareIpRateLimitKeyGenerator } = require('../../middleware/ipRateLimit'); interface AuthReq { user?: { id?: string }; @@ -22,6 +26,20 @@ interface AuthReq { const router: ReturnType = express.Router(); +// Admin pause/resume is a mutating control plane. It gets its own bucket so +// an operator's connector setup cannot consume the protection on this path. +const adminInstallablesRateLimit = rateLimit({ + windowMs: 15 * 60_000, + max: 60, + standardHeaders: true, + legacyHeaders: false, + skip: () => process.env.NODE_ENV === 'test', + keyGenerator: cloudflareIpRateLimitKeyGenerator, + handler: (_req: unknown, res: { status: (code: number) => { json: (body: unknown) => void } }) => { + res.status(429).json({ error: 'rate_limited' }); + }, +}); + const requiredReason = (body: unknown): string | null => { const reason = (body as { reason?: unknown } | null)?.reason; if (typeof reason !== 'string' || !reason.trim()) return null; @@ -56,7 +74,12 @@ const auditEntry = (action: 'pause' | 'resume', installation: { * Pause is parent-first: a crash after the first write has already stopped * outbound dispatch. The reconciler repairs a child which was not stamped. */ -router.post('/:installableId/installations/:installationId/pause', auth, adminAuth, async (req: AuthReq, res: any) => { +router.post( + '/:installableId/installations/:installationId/pause', + adminInstallablesRateLimit, + auth, + adminAuth, + async (req: AuthReq, res: any) => { const reason = requiredReason(req.body); if (!reason) return res.status(400).json({ error: 'reason is required' }); const filter = installationFilter(req, 'active'); @@ -95,13 +118,19 @@ router.post('/:installableId/installations/:installationId/pause', auth, adminAu console.error('[admin/installables] pause failed:', (error as Error).message); return res.status(500).json({ error: 'Could not pause installation' }); } -}); + }, +); /** * Resume is children-first. If parent completion fails, the parent remains * paused and the reconciler re-stamps children on its next pass. */ -router.post('/:installableId/installations/:installationId/resume', auth, adminAuth, async (req: AuthReq, res: any) => { +router.post( + '/:installableId/installations/:installationId/resume', + adminInstallablesRateLimit, + auth, + adminAuth, + async (req: AuthReq, res: any) => { const reason = requiredReason(req.body); if (!reason) return res.status(400).json({ error: 'reason is required' }); const filter = installationFilter(req, 'paused'); @@ -138,6 +167,7 @@ router.post('/:installableId/installations/:installationId/resume', auth, adminA console.error('[admin/installables] resume failed:', (error as Error).message); return res.status(500).json({ error: 'Could not resume installation' }); } -}); + }, +); module.exports = router; diff --git a/backend/routes/integrations.ts b/backend/routes/integrations.ts index 6d3b756535..66483386ef 100644 --- a/backend/routes/integrations.ts +++ b/backend/routes/integrations.ts @@ -30,6 +30,7 @@ const { hash, randomSecret } = require('../utils/secret'); const { mintConnectCode } = require('../services/telegramConnectCode'); // eslint-disable-next-line global-require const isPodMember = require('../utils/isPodMember'); +import { Types } from 'mongoose'; // Keep this as an ESM import: static analysis recognizes the rate limiter at // the route sink, while the middleware owns the shared token/IP bucket. import { @@ -489,9 +490,12 @@ router.post('/:id/connect-code', writeIntegrationsRateLimit, auth, async (req: A } }); -router.patch('/:id', auth, async (req: AuthReq, res: Res) => { +router.patch('/:id', writeIntegrationsRateLimit, auth, async (req: AuthReq, res: Res) => { try { const { id } = req.params || {}; + if (!Types.ObjectId.isValid(id)) { + return res.status(400).json({ message: 'Invalid integration id' }); + } const { config, status, isActive, podId } = (req.body || {}) as { config?: Record; status?: string; From 10bb7d18abae5fa662eb68db3c7e4aac2e5a28f1 Mon Sep 17 00:00:00 2001 From: Lily Shen <115414357+lilyshen0722@users.noreply.github.com> Date: Fri, 4 Sep 2026 22:20:41 -0700 Subject: [PATCH 05/11] fix(connectors): anchor updates to authorized rows --- backend/routes/integrations.ts | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/backend/routes/integrations.ts b/backend/routes/integrations.ts index 66483386ef..86702f04be 100644 --- a/backend/routes/integrations.ts +++ b/backend/routes/integrations.ts @@ -503,6 +503,7 @@ router.patch('/:id', writeIntegrationsRateLimit, auth, async (req: AuthReq, res: podId?: unknown; }; const integration = await Integration.findById(id) as { + _id?: unknown; type?: string; scope?: string; createdBy?: { toString: () => string }; @@ -603,7 +604,10 @@ router.patch('/:id', writeIntegrationsRateLimit, auth, async (req: AuthReq, res: if (autoStatusTypes.includes(integration.type || '') && config) { update.status = (update.status as string | undefined) || (isManifestComplete(integration.type || '', nextConfig) ? 'connected' : 'pending'); } - const updated = await Integration.findByIdAndUpdate(id, update, { new: true }); + // Update the identity that Mongo returned after the access check, not the + // route parameter. This keeps the write anchored to the document we just + // authorised and avoids carrying a request value into the query object. + const updated = await Integration.findByIdAndUpdate(integration._id, update, { new: true }); return res.json(updated); } catch (error) { console.error('Error updating integration:', error); From e6138d31adbdf6b45490262fc0ac906054c6fba6 Mon Sep 17 00:00:00 2001 From: Lily Shen <115414357+lilyshen0722@users.noreply.github.com> Date: Fri, 4 Sep 2026 22:22:57 -0700 Subject: [PATCH 06/11] fix: canonicalize integration update id --- backend/routes/integrations.ts | 15 +++++++++------ 1 file changed, 9 insertions(+), 6 deletions(-) diff --git a/backend/routes/integrations.ts b/backend/routes/integrations.ts index 86702f04be..9eda0b201b 100644 --- a/backend/routes/integrations.ts +++ b/backend/routes/integrations.ts @@ -493,16 +493,20 @@ router.post('/:id/connect-code', writeIntegrationsRateLimit, auth, async (req: A router.patch('/:id', writeIntegrationsRateLimit, auth, async (req: AuthReq, res: Res) => { try { const { id } = req.params || {}; - if (!Types.ObjectId.isValid(id)) { + if (typeof id !== 'string' || !Types.ObjectId.isValid(id)) { return res.status(400).json({ message: 'Invalid integration id' }); } + // Parse the route segment exactly once. Subsequent reads and writes use + // this canonical BSON value rather than carrying the request string into + // a database selector. + const integrationId = new Types.ObjectId(id); const { config, status, isActive, podId } = (req.body || {}) as { config?: Record; status?: string; isActive?: boolean; podId?: unknown; }; - const integration = await Integration.findById(id) as { + const integration = await Integration.findById(integrationId) as { _id?: unknown; type?: string; scope?: string; @@ -604,10 +608,9 @@ router.patch('/:id', writeIntegrationsRateLimit, auth, async (req: AuthReq, res: if (autoStatusTypes.includes(integration.type || '') && config) { update.status = (update.status as string | undefined) || (isManifestComplete(integration.type || '', nextConfig) ? 'connected' : 'pending'); } - // Update the identity that Mongo returned after the access check, not the - // route parameter. This keeps the write anchored to the document we just - // authorised and avoids carrying a request value into the query object. - const updated = await Integration.findByIdAndUpdate(integration._id, update, { new: true }); + // The selector is the canonical ObjectId parsed at the route boundary; + // the access check above authorises that exact document before this write. + const updated = await Integration.findByIdAndUpdate(integrationId, update, { new: true }); return res.json(updated); } catch (error) { console.error('Error updating integration:', error); From b2e7eefbbe3437fbc8252a08eb08c590a245e7d8 Mon Sep 17 00:00:00 2001 From: Lily Shen <115414357+lilyshen0722@users.noreply.github.com> Date: Fri, 4 Sep 2026 22:26:16 -0700 Subject: [PATCH 07/11] fix: persist authorised connector document --- .../routes/integrations.linkedUserId.test.js | 21 +++++++++++++++++-- backend/routes/integrations.ts | 9 +++++--- 2 files changed, 25 insertions(+), 5 deletions(-) diff --git a/backend/__tests__/unit/routes/integrations.linkedUserId.test.js b/backend/__tests__/unit/routes/integrations.linkedUserId.test.js index 3327291dfe..126051d54a 100644 --- a/backend/__tests__/unit/routes/integrations.linkedUserId.test.js +++ b/backend/__tests__/unit/routes/integrations.linkedUserId.test.js @@ -47,7 +47,24 @@ app.use('/api/integrations', integrationRoutes); const integrationId = '64b64c1f7e5b8f0a12345674'; const userScopedIntegrationId = '64b64c1f7e5b8f0a12345675'; -const telegramIntegration = () => ({ +const persistedIntegration = (document) => ({ + ...document, + set(update) { Object.assign(this, update); }, + save: jest.fn().mockImplementation(function save() { + // PATCH now persists the previously-authorised document. Keep the legacy + // spy as an observable snapshot so the route assertions stay focused on + // the exact fields that would be persisted. + Integration.findByIdAndUpdate(this._id, { + config: this.config, + podId: this.podId, + status: this.status, + isActive: this.isActive, + }, { new: true }); + return Promise.resolve(this); + }), +}); + +const telegramIntegration = () => persistedIntegration({ _id: 'integration-1', type: 'telegram', podId: 'pod-1', @@ -160,7 +177,7 @@ describe('PATCH /api/integrations/:id — user-scoped connector gates', () => { const allowedPodId = '64b64c1f7e5b8f0a12345671'; const forbiddenPodId = '64b64c1f7e5b8f0a12345672'; - const userScopedIntegration = () => ({ + const userScopedIntegration = () => persistedIntegration({ _id: 'integration-user', scope: 'user', type: 'telegram', diff --git a/backend/routes/integrations.ts b/backend/routes/integrations.ts index 9eda0b201b..e877bc56e9 100644 --- a/backend/routes/integrations.ts +++ b/backend/routes/integrations.ts @@ -513,6 +513,8 @@ router.patch('/:id', writeIntegrationsRateLimit, auth, async (req: AuthReq, res: createdBy?: { toString: () => string }; podId?: unknown; config?: { toObject?: () => Record }; + set: (update: Record) => void; + save: () => Promise; } | null; if (!integration) return res.status(404).json({ message: 'Integration not found' }); const currentConfig = integration.config?.toObject ? integration.config.toObject() : (integration.config || {}) as Record; @@ -608,9 +610,10 @@ router.patch('/:id', writeIntegrationsRateLimit, auth, async (req: AuthReq, res: if (autoStatusTypes.includes(integration.type || '') && config) { update.status = (update.status as string | undefined) || (isManifestComplete(integration.type || '', nextConfig) ? 'connected' : 'pending'); } - // The selector is the canonical ObjectId parsed at the route boundary; - // the access check above authorises that exact document before this write. - const updated = await Integration.findByIdAndUpdate(integrationId, update, { new: true }); + // Persist the document we just authorised. This deliberately avoids a + // second route-derived database selector at the write boundary. + integration.set(update); + const updated = await integration.save(); return res.json(updated); } catch (error) { console.error('Error updating integration:', error); From 89187e6c8162af01f761d348f3f42f937461b9b7 Mon Sep 17 00:00:00 2001 From: Lily Shen <115414357+lilyshen0722@users.noreply.github.com> Date: Fri, 4 Sep 2026 22:31:33 -0700 Subject: [PATCH 08/11] revert: preserve connector patch compatibility --- .../routes/integrations.linkedUserId.test.js | 21 ++----------------- backend/routes/integrations.ts | 9 +++----- 2 files changed, 5 insertions(+), 25 deletions(-) diff --git a/backend/__tests__/unit/routes/integrations.linkedUserId.test.js b/backend/__tests__/unit/routes/integrations.linkedUserId.test.js index 126051d54a..3327291dfe 100644 --- a/backend/__tests__/unit/routes/integrations.linkedUserId.test.js +++ b/backend/__tests__/unit/routes/integrations.linkedUserId.test.js @@ -47,24 +47,7 @@ app.use('/api/integrations', integrationRoutes); const integrationId = '64b64c1f7e5b8f0a12345674'; const userScopedIntegrationId = '64b64c1f7e5b8f0a12345675'; -const persistedIntegration = (document) => ({ - ...document, - set(update) { Object.assign(this, update); }, - save: jest.fn().mockImplementation(function save() { - // PATCH now persists the previously-authorised document. Keep the legacy - // spy as an observable snapshot so the route assertions stay focused on - // the exact fields that would be persisted. - Integration.findByIdAndUpdate(this._id, { - config: this.config, - podId: this.podId, - status: this.status, - isActive: this.isActive, - }, { new: true }); - return Promise.resolve(this); - }), -}); - -const telegramIntegration = () => persistedIntegration({ +const telegramIntegration = () => ({ _id: 'integration-1', type: 'telegram', podId: 'pod-1', @@ -177,7 +160,7 @@ describe('PATCH /api/integrations/:id — user-scoped connector gates', () => { const allowedPodId = '64b64c1f7e5b8f0a12345671'; const forbiddenPodId = '64b64c1f7e5b8f0a12345672'; - const userScopedIntegration = () => persistedIntegration({ + const userScopedIntegration = () => ({ _id: 'integration-user', scope: 'user', type: 'telegram', diff --git a/backend/routes/integrations.ts b/backend/routes/integrations.ts index e877bc56e9..9eda0b201b 100644 --- a/backend/routes/integrations.ts +++ b/backend/routes/integrations.ts @@ -513,8 +513,6 @@ router.patch('/:id', writeIntegrationsRateLimit, auth, async (req: AuthReq, res: createdBy?: { toString: () => string }; podId?: unknown; config?: { toObject?: () => Record }; - set: (update: Record) => void; - save: () => Promise; } | null; if (!integration) return res.status(404).json({ message: 'Integration not found' }); const currentConfig = integration.config?.toObject ? integration.config.toObject() : (integration.config || {}) as Record; @@ -610,10 +608,9 @@ router.patch('/:id', writeIntegrationsRateLimit, auth, async (req: AuthReq, res: if (autoStatusTypes.includes(integration.type || '') && config) { update.status = (update.status as string | undefined) || (isManifestComplete(integration.type || '', nextConfig) ? 'connected' : 'pending'); } - // Persist the document we just authorised. This deliberately avoids a - // second route-derived database selector at the write boundary. - integration.set(update); - const updated = await integration.save(); + // The selector is the canonical ObjectId parsed at the route boundary; + // the access check above authorises that exact document before this write. + const updated = await Integration.findByIdAndUpdate(integrationId, update, { new: true }); return res.json(updated); } catch (error) { console.error('Error updating integration:', error); From 7efb94f72481b08745c76ad7c45654e2819f82ad Mon Sep 17 00:00:00 2001 From: Lily Shen <115414357+lilyshen0722@users.noreply.github.com> Date: Sat, 5 Sep 2026 00:27:26 -0700 Subject: [PATCH 09/11] test: use valid connector ids in validation routes --- .../routes/integrations.validation.test.js | 21 ++++++++++--------- 1 file changed, 11 insertions(+), 10 deletions(-) diff --git a/backend/__tests__/unit/routes/integrations.validation.test.js b/backend/__tests__/unit/routes/integrations.validation.test.js index 230093f1b5..4c23183df2 100644 --- a/backend/__tests__/unit/routes/integrations.validation.test.js +++ b/backend/__tests__/unit/routes/integrations.validation.test.js @@ -52,6 +52,8 @@ const User = require('../../../models/User'); const Integration = require('../../../models/Integration'); const integrationRoutes = require('../../../routes/integrations'); +const INTEGRATION_ID = '64b64c7f8a9e2f0012345678'; + const app = express(); app.use(express.json()); app.use('/api/integrations', integrationRoutes); @@ -88,7 +90,7 @@ describe('integration manifest validation', () => { it('rejects setting groupme to connected when required fields are missing', async () => { Integration.findById.mockResolvedValue({ - _id: 'integration-1', + _id: INTEGRATION_ID, type: 'groupme', podId: 'pod-1', createdBy: { toString: () => 'user-1' }, @@ -101,7 +103,7 @@ describe('integration manifest validation', () => { }); const res = await request(app) - .patch('/api/integrations/integration-1') + .patch(`/api/integrations/${INTEGRATION_ID}`) .send({ status: 'connected', config: { webhookListenerEnabled: true } }); if (res.status !== 400) { @@ -115,7 +117,7 @@ describe('integration manifest validation', () => { it('marks groupme connected when manifest requirements are satisfied', async () => { Integration.findById.mockResolvedValue({ - _id: 'integration-1', + _id: INTEGRATION_ID, type: 'groupme', podId: 'pod-1', createdBy: { toString: () => 'user-1' }, @@ -128,14 +130,14 @@ describe('integration manifest validation', () => { }); Integration.findByIdAndUpdate.mockResolvedValue({ - _id: 'integration-1', + _id: INTEGRATION_ID, type: 'groupme', status: 'connected', config: { botId: 'bot-1', groupId: 'group-1' }, }); const res = await request(app) - .patch('/api/integrations/integration-1') + .patch(`/api/integrations/${INTEGRATION_ID}`) .send({ config: { botId: 'bot-1', groupId: 'group-1' }, }); @@ -145,11 +147,10 @@ describe('integration manifest validation', () => { throw new Error(`unexpected status ${res.status}: ${JSON.stringify(res.body)}; logged=${loggedError?.message}`); } expect(res.status).toBe(200); - expect(Integration.findByIdAndUpdate).toHaveBeenCalledWith( - 'integration-1', - expect.objectContaining({ status: 'connected' }), - { new: true }, - ); + const [selector, update, options] = Integration.findByIdAndUpdate.mock.calls[0]; + expect(selector.toHexString()).toBe(INTEGRATION_ID); + expect(update).toEqual(expect.objectContaining({ status: 'connected' })); + expect(options).toEqual({ new: true }); }); it('rejects discord creation when required fields are missing', async () => { From 9da950aaaeeacfbb2210f21f5d8e4c6320aa4b59 Mon Sep 17 00:00:00 2001 From: Lily Shen <115414357+lilyshen0722@users.noreply.github.com> Date: Sat, 5 Sep 2026 01:49:44 -0700 Subject: [PATCH 10/11] fix: redact connector pause administrators --- .../__tests__/unit/routes/installables.test.js | 18 ++++++++++++++++-- backend/routes/installables.ts | 5 +++++ 2 files changed, 21 insertions(+), 2 deletions(-) diff --git a/backend/__tests__/unit/routes/installables.test.js b/backend/__tests__/unit/routes/installables.test.js index cf3d6a348d..2dc76f6734 100644 --- a/backend/__tests__/unit/routes/installables.test.js +++ b/backend/__tests__/unit/routes/installables.test.js @@ -76,7 +76,7 @@ describe('installable connector routes', () => { expect(installationService.install).not.toHaveBeenCalled(); }); - it('derives the install target from auth and accepts only the selected pod', async () => { + it('derives the install target from auth and redacts pause administrators', async () => { Pod.findById.mockResolvedValue({ _id: podId, createdBy: { toString: () => 'someone-else' }, @@ -86,7 +86,16 @@ describe('installable connector routes', () => { httpStatus: 200, state: 'active', installation: { _id: 'install-1' }, - integration: { _id: 'integration-1' }, + integration: { + _id: 'integration-1', + config: { + adminPause: { + reason: 'Safety review in progress.', + at: '2026-09-05T08:48:00.000Z', + adminId: 'admin-private-id', + }, + }, + }, }); const res = await request(app) @@ -100,6 +109,11 @@ describe('installable connector routes', () => { installedBy: '64b64c48c4f37a6b2f34c111', podId, }); + expect(res.body.integration.config.adminPause).toEqual({ + reason: 'Safety review in progress.', + at: '2026-09-05T08:48:00.000Z', + }); + expect(JSON.stringify(res.body)).not.toContain('admin-private-id'); }); it('returns typed 409 when activation loses its projection', async () => { diff --git a/backend/routes/installables.ts b/backend/routes/installables.ts index dc137c4509..ae00573140 100644 --- a/backend/routes/installables.ts +++ b/backend/routes/installables.ts @@ -142,6 +142,11 @@ const publicIntegration = (integration: unknown): unknown => { delete result.config.oauthStateNonce; const pending = result.config.pendingBind; if (pending && typeof pending === 'object') delete (pending as Record).botTokenRef; + const adminPause = result.config.adminPause; + if (adminPause && typeof adminPause === 'object') { + const { reason, at } = adminPause as { reason?: unknown; at?: unknown }; + result.config.adminPause = { reason, at }; + } return result; }; From f5b582d1d72c9f3f1440af22dbedf3976c6d215f Mon Sep 17 00:00:00 2001 From: Lily Shen <115414357+lilyshen0722@users.noreply.github.com> Date: Sat, 5 Sep 2026 01:52:55 -0700 Subject: [PATCH 11/11] fix: redact pause administrators in integration JSON --- .../models/Integration.serialization.test.js | 27 +++++++++++++++++++ backend/models/Integration.ts | 5 ++++ 2 files changed, 32 insertions(+) create mode 100644 backend/__tests__/unit/models/Integration.serialization.test.js diff --git a/backend/__tests__/unit/models/Integration.serialization.test.js b/backend/__tests__/unit/models/Integration.serialization.test.js new file mode 100644 index 0000000000..c9533b4c7c --- /dev/null +++ b/backend/__tests__/unit/models/Integration.serialization.test.js @@ -0,0 +1,27 @@ +const mongoose = require('mongoose'); +const Integration = require('../../../models/Integration'); + +describe('Integration JSON serialization', () => { + it('exposes a pause reason and time without the moderator identity', () => { + const pausedAt = new Date('2026-09-05T08:48:00.000Z'); + const integration = new Integration({ + type: 'telegram', + createdBy: new mongoose.Types.ObjectId(), + config: { + adminPause: { + reason: 'Safety review in progress.', + at: pausedAt, + adminId: 'admin-private-id', + }, + }, + }); + + const body = integration.toJSON(); + + expect(body.config.adminPause).toEqual({ + reason: 'Safety review in progress.', + at: pausedAt, + }); + expect(JSON.stringify(body)).not.toMatch(/adminId|admin-private-id/); + }); +}); diff --git a/backend/models/Integration.ts b/backend/models/Integration.ts index 7491257631..57f19180d5 100644 --- a/backend/models/Integration.ts +++ b/backend/models/Integration.ts @@ -346,6 +346,11 @@ IntegrationSchema.set('toJSON', { if (pending && typeof pending === 'object') { delete (pending as Record).botTokenRef; } + const adminPause = returned.config.adminPause; + if (adminPause && typeof adminPause === 'object') { + const { reason, at } = adminPause as { reason?: unknown; at?: unknown }; + returned.config.adminPause = { reason, at }; + } return returned; }, });