From eae4e15e2b0fb33ac2865d791f00e121bd605fbf Mon Sep 17 00:00:00 2001 From: Nirmal Khedkar Date: Mon, 26 Oct 2026 14:16:54 -0700 Subject: [PATCH 1/3] Enforce account ownership for private conversations Signed-off-by: Nirmal Khedkar --- CONFORMANCE.md | 12 ++ backend/__tests__/conversation.test.js | 4 +- backend/__tests__/conversationAccess.test.js | 171 ++++++++++++++++++ .../controllers/conversation_controller.js | 15 +- backend/docs/conversation-ownership.md | 90 +++++++++ backend/middleware/attachmentOwner.js | 19 ++ backend/middleware/chatAuth.js | 19 ++ backend/middleware/conversationOwner.js | 8 + backend/models/conversation.js | 56 ++++-- backend/models/message.js | 19 +- backend/routes/communicationRoutes.js | 7 +- backend/routes/conversationRoutes.js | 10 +- .../scripts/check-conversation-ownership.js | 56 ++++++ backend/server.js | 11 +- frontend/chat-ui/accountClient.test.js | 48 +++++ frontend/chat-ui/basic.test.js | 8 + frontend/chat-ui/src/accountClient.mjs | 53 ++++++ frontend/chat-ui/src/api.js | 40 ++-- frontend/chat-ui/src/chatReducer.mjs | 1 + .../chat-ui/src/components/MessageBubble.jsx | 14 +- frontend/chat-ui/src/store.jsx | 10 +- 21 files changed, 611 insertions(+), 60 deletions(-) create mode 100644 backend/__tests__/conversationAccess.test.js create mode 100644 backend/docs/conversation-ownership.md create mode 100644 backend/middleware/attachmentOwner.js create mode 100644 backend/middleware/chatAuth.js create mode 100644 backend/middleware/conversationOwner.js create mode 100644 backend/scripts/check-conversation-ownership.js create mode 100644 frontend/chat-ui/accountClient.test.js create mode 100644 frontend/chat-ui/src/accountClient.mjs diff --git a/CONFORMANCE.md b/CONFORMANCE.md index f3b9f0a1..a14c2a0b 100644 --- a/CONFORMANCE.md +++ b/CONFORMANCE.md @@ -43,3 +43,15 @@ Adopts the covenant scale (LBTAS, v2 floor) and carries its own record instance. ## Product-specific notes (last, per the ordering rule) The reference economy. Its operator/federation model is the named reference for the substrate's frontend-as-operator design target. + +Private application chat uses the verified backend account `userId` as an +immutable conversation owner. Conversation/message routes and attachment +downloads enforce that boundary; shared integration keys supply no participant +identity. Legacy ownerless rows remain inaccessible pending explicit operator +assignment. Bindings: `backend/middleware/chatAuth.js`, +`conversationOwner.js`, `attachmentOwner.js`, the conditional conversation +model writes, and `backend/__tests__/conversationAccess.test.js`. This does not +promote private chat to a sealed commons dialog or change the in-review ledger +claims above. SSE requires the companion authenticated-stream change before +rollout. See `backend/docs/conversation-ownership.md` for migration and client +integration limits. diff --git a/backend/__tests__/conversation.test.js b/backend/__tests__/conversation.test.js index da0bcaa7..2e168e0d 100644 --- a/backend/__tests__/conversation.test.js +++ b/backend/__tests__/conversation.test.js @@ -36,10 +36,10 @@ test('removing a conversation deletes its messages', async () => { Conversation.setDocClient(mockClient); Message.setDocClient(mockClient); - await mockClient.put({ TableName: 'Conversations', Item: { id: '1', title: 'seed', pinned: false, createdAt: new Date().toISOString() } }).promise(); + await mockClient.put({ TableName: 'Conversations', Item: { id: '1', ownerId: 'owner', title: 'seed', pinned: false, createdAt: new Date().toISOString() } }).promise(); await mockClient.put({ TableName: 'Messages', Item: { id: 'm1', conversationId: '1', content: 'hi' } }).promise(); - await Conversation.remove('1'); + await Conversation.remove('1', 'owner'); const msgs = await Message.listMessages('1'); assert.equal(msgs.length, 0); }); diff --git a/backend/__tests__/conversationAccess.test.js b/backend/__tests__/conversationAccess.test.js new file mode 100644 index 00000000..c1ba2644 --- /dev/null +++ b/backend/__tests__/conversationAccess.test.js @@ -0,0 +1,171 @@ +const request = require('supertest'); +const jwt = require('jsonwebtoken'); +const fs = require('fs'); +const path = require('path'); + +const mockTables = { Conversations: new Map(), Messages: new Map() }; +let mockMessageWrites = 0; +let mockMessageReads = 0; +function mockConditionalFailure() { + return Object.assign(new Error('condition failed'), { code: 'ConditionalCheckFailedException' }); +} +jest.mock('../lib/dynamodbClient', () => ({ + put: ({ TableName, Item }) => ({ promise: async () => { + if (TableName === 'Messages') mockMessageWrites++; + mockTables[TableName].set(Item.id, { ...Item }); + return {}; + } }), + get: ({ TableName, Key }) => ({ promise: async () => ({ Item: mockTables[TableName].get(Key.id) }) }), + scan: ({ TableName, ExpressionAttributeValues, ExclusiveStartKey }) => ({ promise: async () => { + let Items = [...mockTables[TableName].values()]; + // Deliberately include a continuation page with no matching items first. + if (ExclusiveStartKey) Items = Items.slice(1); + else if (Items.length > 1) Items = Items.slice(0, 1); + const LastEvaluatedKey = !ExclusiveStartKey && mockTables[TableName].size > 1 ? { id: Items[0].id } : undefined; + if (ExpressionAttributeValues?.[':ownerId']) { + Items = Items.filter(item => item.ownerId === ExpressionAttributeValues[':ownerId']); + } + if (ExpressionAttributeValues?.[':filePath']) { + Items = Items.filter(item => item.file?.path === ExpressionAttributeValues[':filePath']); + } + return { Items, LastEvaluatedKey }; + } }), + update: ({ TableName, Key, ExpressionAttributeNames, ExpressionAttributeValues, ConditionExpression }) => ({ promise: async () => { + const item = mockTables[TableName].get(Key.id); + if (!item || (ConditionExpression.includes('ownerId') && item.ownerId !== ExpressionAttributeValues[':ownerId'])) throw mockConditionalFailure(); + item[ExpressionAttributeNames['#field']] = ExpressionAttributeValues[':value']; + return { Attributes: { ...item } }; + } }), + delete: ({ TableName, Key, ConditionExpression, ExpressionAttributeValues }) => ({ promise: async () => { + const item = mockTables[TableName].get(Key.id); + if (ConditionExpression && (!item || item.ownerId !== ExpressionAttributeValues[':ownerId'])) throw mockConditionalFailure(); + mockTables[TableName].delete(Key.id); + return { Attributes: item }; + } }), + query: ({ ExpressionAttributeValues }) => ({ promise: async () => { + mockMessageReads++; + return { Items: [...mockTables.Messages.values()].filter(item => item.conversationId === ExpressionAttributeValues[':cid']) }; + } }), + batchWrite: ({ RequestItems }) => ({ promise: async () => { + for (const { DeleteRequest: { Key } } of RequestItems.Messages || []) mockTables.Messages.delete(Key.id); + return {}; + } }), +})); +jest.mock('../bull/pingJobs', () => ({ addPingJob: jest.fn() })); +jest.mock('../bull/smsQueue', () => ({ enqueueSms: jest.fn() })); +jest.mock('../federation/federationSyncJob', () => jest.fn()); +jest.mock('../services/openAIResponder', () => ({ generateResponse: async () => ({ from: 'ai', content: 'fixture reply' }) })); + +describe('private conversation HTTP boundary', () => { + let app; + const originalEnv = { ...process.env }; + const globals = Object.fromEntries(['broadcast', 'emitToken', 'emitMessage'].map(key => [key, global[key]])); + const filename = 'ownership-fixture.txt'; + const attachment = `/uploads/${filename}`; + const authorization = (userId, extra = {}) => `Bearer ${jwt.sign({ userId, ...extra }, process.env.JWT_SECRET, { expiresIn: '1h' })}`; + + beforeAll(() => { + process.env.MINIMAL_SERVER = 'false'; + process.env.API_KEY = 'conversation-test-key'; + process.env.JWT_SECRET = 'conversation-test-secret'; + ({ app } = require('../server')); + fs.writeFileSync(path.join(__dirname, '../uploads', filename), 'private attachment'); + }); + beforeEach(() => { + mockTables.Conversations.clear(); + mockTables.Messages.clear(); + mockMessageWrites = 0; + mockMessageReads = 0; + mockTables.Conversations.set('foreign', { id: 'foreign', ownerId: 'bob', title: 'Bobs private chat' }); + mockTables.Conversations.set('mine', { id: 'mine', ownerId: 'alice', title: 'My chat', pinned: false }); + mockTables.Conversations.set('legacy', { id: 'legacy', title: 'Unknown owner' }); + mockTables.Messages.set('secret', { id: 'secret', conversationId: 'foreign', content: 'private', file: { path: attachment } }); + }); + afterAll(() => { + fs.unlinkSync(path.join(__dirname, '../uploads', filename)); + process.env = originalEnv; + for (const [key, value] of Object.entries(globals)) { + if (value === undefined) delete global[key]; else global[key] = value; + } + }); + + test('creation derives immutable ownership from the verified account, ignoring supplied owner fields', async () => { + const result = await request(app).post('/conversations').set('Authorization', authorization('alice')) + .send({ title: 'New chat', ownerId: 'bob', userId: 'bob' }); + expect(result.status).toBe(201); + expect(result.body.ownerId).toBe('alice'); + expect(mockTables.Conversations.get(result.body.id).ownerId).toBe('alice'); + }); + test('listing traverses empty filtered pages without exposing foreign or legacy chats', async () => { + const result = await request(app).get('/conversations').set('Authorization', authorization('alice')); + expect(result.status).toBe(200); + expect(result.body.map(item => item.id)).toEqual(['mine']); + }); + test.each(['foreign', 'legacy', 'missing'])('all ID routes conceal %s and have no message side effects', async id => { + for (const [method, url, body] of [ + ['get', `/messages/${id}`], ['get', `/conversations/${id}`], ['put', `/conversations/${id}`, { title: 'stolen' }], + ['post', `/conversations/${id}/pin`, { pinned: true }], ['delete', `/conversations/${id}`], + ['get', `/messages/${id}`], ['post', `/messages/${id}`, { from: 'user', content: 'private AI trigger' }], + ]) { + const result = await request(app)[method](url).set('Authorization', authorization('alice')).send(body); + expect(result.status).toBe(404); + expect(result.body).toEqual({ error: 'Not found' }); + } + expect(mockMessageReads).toBe(0); + expect(mockMessageWrites).toBe(0); + expect(mockTables.Conversations.get('foreign').title).toBe('Bobs private chat'); + expect(mockTables.Messages.get('secret').content).toBe('private'); + }); + test.each([undefined, '', ' ', 7, ['alice']])('rejects a signed token without a valid userId (%p)', async userId => { + const result = await request(app).get('/conversations').set('Authorization', authorization(userId)); + expect(result.status).toBe(401); + }); + test('a shared API key cannot replace or bypass individual bearer authentication', async () => { + for (const bearer of [undefined, 'Bearer invalid']) { + let call = request(app).get('/conversations').set('x-api-key', process.env.API_KEY); + if (bearer) call = call.set('Authorization', bearer); + expect((await call).status).toBe(401); + } + const result = await request(app).get('/conversations').set('x-api-key', process.env.API_KEY) + .set('Authorization', authorization('alice')); + expect(result.status).toBe(200); + expect(result.body.map(item => item.id)).toEqual(['mine']); + }); + test('the rightful owner can read, rename, pin, send, list and delete', async () => { + const auth = authorization('alice'); + expect((await request(app).get('/conversations/mine').set('Authorization', auth)).status).toBe(200); + expect((await request(app).put('/conversations/mine').set('Authorization', auth).send({ title: 'Renamed', ownerId: 'bob' })).body.title).toBe('Renamed'); + expect((await request(app).post('/conversations/mine/pin').set('Authorization', auth).send({ pinned: true })).body.pinned).toBe(true); + const sent = await request(app).post('/messages/mine').set('Authorization', auth).send({ from: 'alice', content: 'hello', type: 'text' }); + expect(sent.status).toBe(201); + const messages = await request(app).get('/messages/mine').set('Authorization', auth); + expect(messages.body.map(item => item.content)).toEqual(['hello']); + expect((await request(app).delete('/conversations/mine').set('Authorization', auth)).status).toBe(204); + expect(mockTables.Conversations.has('mine')).toBe(false); + expect([...mockTables.Messages.values()].some(item => item.conversationId === 'mine')).toBe(false); + expect(mockTables.Conversations.get('foreign').ownerId).toBe('bob'); + }); + test('attachment paths enforce the same owner boundary, including encoded paths', async () => { + for (const url of [attachment, '/uploads/%6fwnership-fixture.txt']) { + const foreign = await request(app).get(url).set('Authorization', authorization('alice')); + expect(foreign.status).toBe(404); + const owner = await request(app).get(url).set('Authorization', authorization('bob')); + expect(owner.status).toBe(200); + expect(owner.body.toString()).toBe('private attachment'); + expect(owner.headers['cache-control']).toContain('no-store'); + expect(owner.headers['content-disposition']).toBe('attachment'); + } + expect((await request(app).get(attachment).set('x-api-key', process.env.API_KEY)).status).toBe(401); + }); + + test('conditional model writes cannot rename, pin or delete another owner or a legacy row', async () => { + const Conversation = require('../models/conversation'); + for (const id of ['foreign', 'legacy', 'missing']) { + expect(await Conversation.rename(id, 'stolen', 'alice')).toBeNull(); + expect(await Conversation.pin(id, true, 'alice')).toBeNull(); + expect(await Conversation.remove(id, 'alice')).toBe(false); + } + expect(mockTables.Messages.get('secret').content).toBe('private'); + expect(mockTables.Conversations.get('foreign').title).toBe('Bobs private chat'); + }); +}); diff --git a/backend/controllers/conversation_controller.js b/backend/controllers/conversation_controller.js index e4b0d94f..373aa894 100644 --- a/backend/controllers/conversation_controller.js +++ b/backend/controllers/conversation_controller.js @@ -2,33 +2,36 @@ const Conversation = require('../models/conversation'); exports.create = async (req, res) => { const { title } = req.body; - const convo = await Conversation.create(title); + const convo = await Conversation.create(title, req.user.userId); res.status(201).json(convo); }; -exports.list = async (_req, res) => { - const convos = await Conversation.list(); +exports.list = async (req, res) => { + const convos = await Conversation.list(req.user.userId); res.json(convos); }; +exports.get = (req, res) => res.json(req.conversation); + exports.rename = async (req, res) => { const { id } = req.params; const { title } = req.body; - const convo = await Conversation.rename(id, title); + const convo = await Conversation.rename(id, title, req.user.userId); if (!convo) return res.status(404).json({ error: 'Not found' }); res.json(convo); }; exports.remove = async (req, res) => { const { id } = req.params; - await Conversation.remove(id); + const removed = await Conversation.remove(id, req.user.userId); + if (!removed) return res.status(404).json({ error: 'Not found' }); res.status(204).end(); }; exports.pin = async (req, res) => { const { id } = req.params; const { pinned } = req.body; - const convo = await Conversation.pin(id, pinned); + const convo = await Conversation.pin(id, pinned, req.user.userId); if (!convo) return res.status(404).json({ error: 'Not found' }); res.json(convo); }; diff --git a/backend/docs/conversation-ownership.md b/backend/docs/conversation-ownership.md new file mode 100644 index 00000000..bdb7cf5b --- /dev/null +++ b/backend/docs/conversation-ownership.md @@ -0,0 +1,90 @@ +# Private conversation ownership + +## Boundary + +Private chat uses the existing backend login JWT's `userId`, verified as an +HS256 account token. A shared `API_KEY`/`WIX_API_KEY` identifies an integration, +not a participant, and grants no conversation, message or attachment access. +Supplying a valid integration key alongside a bearer never bypasses bearer +verification. No Cognito `sub`, caller-supplied owner, message sender label or +administrative role is treated as a substitute account identity. + +New Conversations rows carry immutable `ownerId`. Listing filters on that +owner across DynamoDB scan pages. Detail reads, rename, pin, delete, message +reads and sends check ownership. Rename/pin/delete also carry a conditional +owner check in DynamoDB. Foreign, missing and ownerless IDs return the same +404. A denied send reaches neither file writes, messages, responders nor +broadcasts. Attachments require a message reference belonging to an owned +conversation; responses disable caching and download as inert attachments. + +This is application-local private chat. It does not seal a JFA dialog or +modify Mycelium commitments, exchange records, reputation, escrow or federation. +The separate status of those mechanisms in CONFORMANCE.md remains accurate. + +## Existing data and deployment + +Back up Conversations and Messages before rollout. Existing rows without an +owner remain stored but inaccessible; the application never grants them to +the first caller or guesses an owner from `from`. An operator must establish +the rightful existing account through independent evidence before making an +explicit, conditional administrative backfill. This PR supplies no public +claim/transfer endpoint and performs no automatic data migration or deletion. +If ownership cannot be established, preserve the inaccessible row for the +operator's data-retention process. Ownership transfer is not supported here. + +The existing Messages deletion helper still has its previously documented +large-history batching/retry limitations. This change establishes who may +invoke deletion; it does not claim complete erasure of large histories or +physical attachment bytes. Stale attachment files become inaccessible when +their owned message/conversation reference no longer exists. + +The deployment needs DynamoDB GetItem/Scan/PutItem/UpdateItem/DeleteItem access +to the configured conversation table and the existing message access, now +including Scan for attachment-reference lookup. Scans require no new index +or table. They prioritize correct isolation; an indexed attachment lookup is +a possible measured follow-up, not supplied by this change. + +**Roll out together with the authenticated-stream companion PR (A2).** A1 +closes REST and attachment access; unauthenticated SSE remains a separate +existing path until A2 lands. Other open hardening work, including admin, +user/federation authorization and attachment write containment, remains +separate and must be assessed before exposing public signup. + +## Browser integration + +The chat UI previously sent only a shared build-time API key and has no +working account-login screen. The host application's account flow must obtain +the existing backend `/api/auth/login` response and pass its token to the +exported `setAccountToken(token)` in `frontend/chat-ui/src/api.js`. Call +`setAccountToken(null)` on logout. Tokens stay in memory; the adapter uses +Authorization headers, never query strings or browser storage. Account changes +clear chat state and discard in-flight responses from the old account. + +Do not substitute an Amplify/Cognito token without an approved backend identity +binding. PR #183 separately makes backend login/registration publicly +reachable; until it lands, login still needs the server's existing integration +authentication. This PR does not claim completed browser onboarding. + +Attachment buttons download through the authenticated adapter rather than +opening uncredentialed URLs. Malformed, external and non-upload download paths +are rejected before forwarding credentials. Redirects are rejected. + +## Verification + +`cd backend && npm test` includes production-server HTTP tests for two users, +legacy records, missing IDs, conditional mutation ownership, malformed subjects, +shared-key bypass attempts and protected attachments. AI provider calls and +background queues are isolated; account verification, mounted routes and the +authorization/model logic remain real. `cd frontend/chat-ui && npm test` +checks credential placement, account changes, attachment destinations and the +existing chat reducer. Database-backed verification uses isolated DynamoDB +Local tables, never live records. + +Run the storage check with `DYNAMODB_ENDPOINT=http://127.0.0.1:18001 node +scripts/check-conversation-ownership.js` from `backend`, after starting +DynamoDB Local on that loopback port. It creates uniquely named disposable +tables and removes them afterward. The check refuses non-loopback endpoints. + +Scope follows the [Fruitful communications overview](https://ntari.atlassian.net/wiki/spaces/Fruitful/pages/44761089) +and [Slack's instruction to use Janus and bylaws](https://ntari.slack.com/archives/C0AENHR5KG8/p1791982812690669). +Current repository invariants take precedence over older architecture summaries. diff --git a/backend/middleware/attachmentOwner.js b/backend/middleware/attachmentOwner.js new file mode 100644 index 00000000..bde65905 --- /dev/null +++ b/backend/middleware/attachmentOwner.js @@ -0,0 +1,19 @@ +const Message = require('../models/message'); +const Conversation = require('../models/conversation'); + +module.exports = async (req, res, next) => { + let filename; + try { filename = decodeURIComponent(req.path).slice(1); } catch { /* invalid URL */ } + if (!filename || filename.includes('/') || filename.includes('\\') || filename.includes('\0') || filename === '.' || filename === '..') { + return res.status(404).json({ error: 'Not found' }); + } + const messages = await Message.findByAttachmentPath(`/uploads/${filename}`); + for (const message of messages) { + if (await Conversation.getOwned(message.conversationId, req.user.userId)) { + // Private response must not be cached across accounts after authorization. + res.set('Cache-Control', 'private, no-store'); + return next(); + } + } + return res.status(404).json({ error: 'Not found' }); +}; diff --git a/backend/middleware/chatAuth.js b/backend/middleware/chatAuth.js new file mode 100644 index 00000000..7dbe46c7 --- /dev/null +++ b/backend/middleware/chatAuth.js @@ -0,0 +1,19 @@ +const jwt = require('jsonwebtoken'); + +// A shared integration key authenticates no individual participant. Verify the +// bearer independently even when the general API middleware accepted that key. +module.exports = (req, res, next) => { + const header = req.headers.authorization; + if (!header || !header.startsWith('Bearer ')) { + return res.status(401).json({ error: 'Account bearer token required' }); + } + try { + const payload = jwt.verify(header.slice(7), process.env.JWT_SECRET, { algorithms: ['HS256'] }); + if (typeof payload.userId !== 'string' || !payload.userId.trim()) throw new Error('Missing account'); + req.user = payload; + res.set('Cache-Control', 'private, no-store'); + return next(); + } catch { + return res.status(401).json({ error: 'Unauthorized: Invalid Token' }); + } +}; diff --git a/backend/middleware/conversationOwner.js b/backend/middleware/conversationOwner.js new file mode 100644 index 00000000..5338198b --- /dev/null +++ b/backend/middleware/conversationOwner.js @@ -0,0 +1,8 @@ +const Conversation = require('../models/conversation'); + +module.exports = async (req, res, next) => { + const conversation = await Conversation.getOwned(req.params.conversationId || req.params.id, req.user.userId); + if (!conversation) return res.status(404).json({ error: 'Not found' }); + req.conversation = conversation; + return next(); +}; diff --git a/backend/models/conversation.js b/backend/models/conversation.js index 79d5080c..c0a28ef0 100644 --- a/backend/models/conversation.js +++ b/backend/models/conversation.js @@ -4,9 +4,11 @@ const Message = require('./message'); const CONVERSATION_TABLE_NAME = process.env.CONVERSATION_TABLE_NAME || 'Conversations'; -async function create(title = 'New Chat') { +async function create(title = 'New Chat', ownerId) { + if (typeof ownerId !== 'string' || !ownerId.trim()) throw new Error('Conversation owner required'); const convo = { id: randomUUID(), + ownerId, title, pinned: false, createdAt: new Date().toISOString(), @@ -15,12 +17,31 @@ async function create(title = 'New Chat') { return convo; } -async function list() { - const res = await docClient.scan({ TableName: CONVERSATION_TABLE_NAME }).promise(); - return res.Items || []; +async function list(ownerId) { + if (typeof ownerId !== 'string' || !ownerId.trim()) return []; + const items = []; + let cursor; + do { + const res = await docClient.scan({ + TableName: CONVERSATION_TABLE_NAME, + FilterExpression: 'ownerId = :ownerId', + ExpressionAttributeValues: { ':ownerId': ownerId }, + ...(cursor && { ExclusiveStartKey: cursor }), + }).promise(); + items.push(...(res.Items || [])); + cursor = res.LastEvaluatedKey; + } while (cursor); + return items; } -async function update(id, field, value) { +async function getOwned(id, ownerId) { + if (typeof ownerId !== 'string' || !ownerId.trim()) return null; + const { Item } = await docClient.get({ TableName: CONVERSATION_TABLE_NAME, Key: { id }, ConsistentRead: true }).promise(); + return Item && Item.ownerId === ownerId ? Item : null; +} + +async function update(id, field, value, ownerId) { + if (typeof ownerId !== 'string' || !ownerId.trim()) return null; try { const res = await docClient .update({ @@ -28,9 +49,9 @@ async function update(id, field, value) { Key: { id }, UpdateExpression: 'SET #field = :value', ExpressionAttributeNames: { '#field': field }, - ExpressionAttributeValues: { ':value': value }, + ExpressionAttributeValues: { ':value': value, ':ownerId': ownerId }, ReturnValues: 'ALL_NEW', - ConditionExpression: 'attribute_exists(id)', + ConditionExpression: 'attribute_exists(id) AND ownerId = :ownerId', }) .promise(); return res.Attributes; @@ -40,17 +61,28 @@ async function update(id, field, value) { } } -function rename(id, title) { return update(id, 'title', title); } +function rename(id, title, ownerId) { return update(id, 'title', title, ownerId); } -async function remove(id) { - await docClient.delete({ TableName: CONVERSATION_TABLE_NAME, Key: { id } }).promise(); +async function remove(id, ownerId) { + if (typeof ownerId !== 'string' || !ownerId.trim()) return false; + try { + await docClient.delete({ + TableName: CONVERSATION_TABLE_NAME, Key: { id }, + ConditionExpression: 'attribute_exists(id) AND ownerId = :ownerId', + ExpressionAttributeValues: { ':ownerId': ownerId }, + }).promise(); + } catch (err) { + if (err.code === 'ConditionalCheckFailedException') return false; + throw err; + } await Message.removeByConversation(id); + return true; } -function pin(id, pinned) { return update(id, 'pinned', pinned); } +function pin(id, pinned, ownerId) { return update(id, 'pinned', pinned, ownerId); } function setDocClient(client) { docClient = client; } -module.exports = { create, list, rename, remove, pin, setDocClient }; +module.exports = { create, list, getOwned, rename, remove, pin, setDocClient }; diff --git a/backend/models/message.js b/backend/models/message.js index 3f0f0353..f37aafa5 100644 --- a/backend/models/message.js +++ b/backend/models/message.js @@ -66,4 +66,21 @@ function setDocClient(client) { docClient = client; } -module.exports = { sendMessage, listMessages, removeByConversation, setDocClient }; \ No newline at end of file +async function findByAttachmentPath(filePath) { + const items = []; + let cursor; + do { + const page = await docClient.scan({ + TableName: MESSAGE_TABLE_NAME, + FilterExpression: '#file.#path = :filePath', + ExpressionAttributeNames: { '#file': 'file', '#path': 'path' }, + ExpressionAttributeValues: { ':filePath': filePath }, + ...(cursor && { ExclusiveStartKey: cursor }), + }).promise(); + items.push(...(page.Items || [])); + cursor = page.LastEvaluatedKey; + } while (cursor); + return items; +} + +module.exports = { sendMessage, listMessages, removeByConversation, findByAttachmentPath, setDocClient }; diff --git a/backend/routes/communicationRoutes.js b/backend/routes/communicationRoutes.js index 3a0367c2..675bce23 100644 --- a/backend/routes/communicationRoutes.js +++ b/backend/routes/communicationRoutes.js @@ -1,11 +1,12 @@ const express = require('express'); const router = express.Router(); const commController = require('../controllers/communication_controller'); -const auth = require('../middleware/authMiddleware'); +const auth = require('../middleware/chatAuth'); +const owner = require('../middleware/conversationOwner'); const asyncHandler = require('../utils/asyncHandler'); router.use(auth); -router.post('/:conversationId', asyncHandler(commController.sendMessage)); -router.get('/:conversationId', asyncHandler(commController.listMessages)); +router.post('/:conversationId', asyncHandler(owner), asyncHandler(commController.sendMessage)); +router.get('/:conversationId', asyncHandler(owner), asyncHandler(commController.listMessages)); module.exports = router; diff --git a/backend/routes/conversationRoutes.js b/backend/routes/conversationRoutes.js index 49739001..d2880d73 100644 --- a/backend/routes/conversationRoutes.js +++ b/backend/routes/conversationRoutes.js @@ -1,14 +1,16 @@ const express = require('express'); const router = express.Router(); -const auth = require('../middleware/authMiddleware'); +const auth = require('../middleware/chatAuth'); +const owner = require('../middleware/conversationOwner'); const ctrl = require('../controllers/conversation_controller'); const asyncHandler = require('../utils/asyncHandler'); router.use(auth); router.post('/', asyncHandler(ctrl.create)); router.get('/', asyncHandler(ctrl.list)); -router.put('/:id', asyncHandler(ctrl.rename)); -router.delete('/:id', asyncHandler(ctrl.remove)); -router.post('/:id/pin', asyncHandler(ctrl.pin)); +router.get('/:id', asyncHandler(owner), ctrl.get); +router.put('/:id', asyncHandler(owner), asyncHandler(ctrl.rename)); +router.delete('/:id', asyncHandler(owner), asyncHandler(ctrl.remove)); +router.post('/:id/pin', asyncHandler(owner), asyncHandler(ctrl.pin)); module.exports = router; diff --git a/backend/scripts/check-conversation-ownership.js b/backend/scripts/check-conversation-ownership.js new file mode 100644 index 00000000..61ba60b9 --- /dev/null +++ b/backend/scripts/check-conversation-ownership.js @@ -0,0 +1,56 @@ +// Explicit integration check: creates and removes isolated DynamoDB Local tables. +const assert = require('node:assert/strict'); +const { randomUUID } = require('node:crypto'); +const { DynamoDBClient, CreateTableCommand, DeleteTableCommand } = require('@aws-sdk/client-dynamodb'); + +async function main() { + const endpoint = process.env.DYNAMODB_ENDPOINT; + const parsed = new URL(endpoint); + if (!['127.0.0.1', 'localhost', '[::1]'].includes(parsed.hostname)) throw new Error('DynamoDB Local loopback endpoint required'); + process.env.AWS_ACCESS_KEY_ID = 'localcheck'; + process.env.AWS_SECRET_ACCESS_KEY = 'localcheck'; + process.env.AWS_REGION = 'us-east-1'; + const suffix = randomUUID().replaceAll('-', ''); + process.env.CONVERSATION_TABLE_NAME = `ChatOwnership_${suffix}`; + process.env.MESSAGE_TABLE_NAME = `ChatMessages_${suffix}`; + const admin = new DynamoDBClient({ endpoint, region: 'us-east-1', credentials: { accessKeyId: 'localcheck', secretAccessKey: 'localcheck' } }); + const tables = []; + try { + for (const [TableName, keys] of [ + [process.env.CONVERSATION_TABLE_NAME, [{ AttributeName: 'id', KeyType: 'HASH' }]], + [process.env.MESSAGE_TABLE_NAME, [{ AttributeName: 'conversationId', KeyType: 'HASH' }, { AttributeName: 'id', KeyType: 'RANGE' }]], + ]) { + await admin.send(new CreateTableCommand({ + TableName, KeySchema: keys, AttributeDefinitions: keys.map(({ AttributeName }) => ({ AttributeName, AttributeType: 'S' })), BillingMode: 'PAY_PER_REQUEST', + })); + tables.push(TableName); + } + const Conversation = require('../models/conversation'); + const Message = require('../models/message'); + const db = require('../lib/dynamodbClient'); + const mine = await Conversation.create('private', 'alice'); + await Conversation.create('foreign', 'bob'); + await db.put({ TableName: process.env.CONVERSATION_TABLE_NAME, Item: { id: 'legacy', title: 'unknown owner' } }).promise(); + assert.equal(await Conversation.getOwned(mine.id, 'bob'), null); + assert.equal(await Conversation.getOwned('legacy', 'alice'), null); + assert.equal(await Conversation.rename(mine.id, 'stolen', 'bob'), null); + assert.equal(await Conversation.pin(mine.id, true, 'bob'), null); + assert.equal(await Conversation.remove(mine.id, 'bob'), false); + assert.equal((await Conversation.rename(mine.id, 'renamed', 'alice')).title, 'renamed'); + assert.equal((await Conversation.pin(mine.id, true, 'alice')).pinned, true); + for (let i = 0; i < 6; i++) await Conversation.create('x'.repeat(230000), i % 2 ? 'alice' : 'bob'); + assert.equal((await Conversation.list('alice')).length, 4); + const message = await Message.sendMessage(mine.id, { from: 'alice', content: 'private content' }); + assert.equal((await Message.listMessages(mine.id))[0].id, message.id); + await db.put({ TableName: process.env.MESSAGE_TABLE_NAME, Item: { id: 'attachment', conversationId: mine.id, file: { path: '/uploads/example.txt' } } }).promise(); + assert.equal((await Message.findByAttachmentPath('/uploads/example.txt'))[0].id, 'attachment'); + assert.equal(await Conversation.remove(mine.id, 'alice'), true); + assert.deepEqual(await Message.listMessages(mine.id), []); + assert.equal(await Conversation.getOwned(mine.id, 'alice'), null); + console.log('PASS: real DynamoDB owner conditions, legacy isolation, multi-page owner listing, message lookup and deletion'); + } finally { + for (const TableName of tables) await admin.send(new DeleteTableCommand({ TableName })); + admin.destroy(); + } +} +main().catch(error => { console.error(error); process.exitCode = 1; }); diff --git a/backend/server.js b/backend/server.js index 6623eac1..f127e30e 100644 --- a/backend/server.js +++ b/backend/server.js @@ -14,6 +14,8 @@ try { } const path = require("path"); const authMiddleware = require("./middleware/authMiddleware"); +const chatAuth = require('./middleware/chatAuth'); +const asyncHandler = require('./utils/asyncHandler'); // Load environment variables dotenv.config(); @@ -130,7 +132,14 @@ global.emitMessage = emitMessage; // Middleware app.use(authMiddleware); -app.use('/uploads', express.static(path.join(__dirname, 'uploads'))); +app.use('/uploads', chatAuth, asyncHandler(require('./middleware/attachmentOwner')), express.static(path.join(__dirname, 'uploads'), { + dotfiles: 'deny', index: false, + setHeaders(res) { + res.setHeader('Content-Disposition', 'attachment'); + res.setHeader('Content-Type', 'application/octet-stream'); + res.setHeader('X-Content-Type-Options', 'nosniff'); + }, +})); // Routes let runFederationSync; diff --git a/frontend/chat-ui/accountClient.test.js b/frontend/chat-ui/accountClient.test.js new file mode 100644 index 00000000..59197fd0 --- /dev/null +++ b/frontend/chat-ui/accountClient.test.js @@ -0,0 +1,48 @@ +const { test } = require('node:test'); +const assert = require('node:assert/strict'); + +test('account requests carry bearer headers without putting credentials in URLs', async () => { + const { createAccountClient } = await import('./src/accountClient.mjs'); + let received; + const client = createAccountClient({ baseUrl: 'https://backend.example', fetchImpl: async (url, options) => { + received = { url, options }; + return new Response('{"id":"owned"}', { status: 201 }); + } }); + client.setToken('account-token'); + assert.deepEqual(await client.request('POST', '/conversations', { title: 'mine' }), { id: 'owned' }); + assert.equal(received.url, 'https://backend.example/conversations'); + assert.equal(received.options.headers.Authorization, 'Bearer account-token'); + assert.equal(received.options.redirect, 'error'); + assert.equal(received.options.body, '{"title":"mine"}'); +}); + +test('clearing or changing accounts discards in-flight account responses', async () => { + const { createAccountClient } = await import('./src/accountClient.mjs'); + let finish; + const client = createAccountClient({ fetchImpl: () => new Promise(resolve => { finish = resolve; }) }); + client.setToken('alice-token'); + const pending = client.request('GET', '/messages/private'); + client.setToken('bob-token'); + finish(new Response('[{"content":"alice secret"}]')); + await assert.rejects(pending, /Account changed/); + client.setToken(null); + assert.equal(client.getToken(), null); +}); + +test('attachment downloads use the same account and reject credential destinations outside uploads', async () => { + const { createAccountClient } = await import('./src/accountClient.mjs'); + let received; + const client = createAccountClient({ fetchImpl: async (url, options) => { + received = { url, options }; + return new Response('private file'); + } }); + client.setToken('owner-token'); + const blob = await client.attachment('/uploads/file.txt'); + assert.equal(await blob.text(), 'private file'); + assert.equal(received.options.headers.Authorization, 'Bearer owner-token'); + for (const url of ['https://attacker.example/x', '//attacker.example/x', '/uploads/../users', '/uploads/%2e%2e/users']) { + await assert.rejects(client.attachment(url), /Invalid attachment path/); + } + client.setToken(null); + await assert.rejects(client.attachment('/uploads/file.txt'), /Sign in/); +}); diff --git a/frontend/chat-ui/basic.test.js b/frontend/chat-ui/basic.test.js index 85b3efc2..3784f216 100644 --- a/frontend/chat-ui/basic.test.js +++ b/frontend/chat-ui/basic.test.js @@ -1,6 +1,14 @@ const { test } = require('node:test'); const assert = require('node:assert/strict'); +test('account changes clear private chat state while retaining display preferences', async () => { + const { reducer } = await import('./src/chatReducer.mjs'); + const state = { messages: [{ content: 'private' }], conversations: [{ id: 'old' }], currentConversation: { id: 'old' }, aiThinking: true, theme: 'dark' }; + assert.deepEqual(reducer(state, { type: 'RESET_ACCOUNT' }), { + messages: [], conversations: [], currentConversation: null, aiThinking: false, theme: 'dark', accountVersion: 1, + }); +}); + test('chat streaming appends content and upsert preserves message metadata', async () => { const { reducer } = await import('./src/chatReducer.mjs'); const initial = { messages: [{ id: 'm1', content: 'Hello', from: 'ai' }] }; diff --git a/frontend/chat-ui/src/accountClient.mjs b/frontend/chat-ui/src/accountClient.mjs new file mode 100644 index 00000000..0c638415 --- /dev/null +++ b/frontend/chat-ui/src/accountClient.mjs @@ -0,0 +1,53 @@ +export function createAccountClient({ baseUrl = '', apiKey = '', fetchImpl = (...args) => fetch(...args) } = {}) { + let token = null; + let version = 0; + const changes = new EventTarget(); + function setToken(value) { + token = typeof value === 'string' && value ? value : null; + version++; + changes.dispatchEvent(new Event('change')); + } + function url(path) { + if (!path.startsWith('/') || path.startsWith('//')) throw new Error('Invalid API path'); + return `${baseUrl.replace(/\/$/, '')}${path}`; + } + function headers() { + const result = {}; + if (token) result.Authorization = `Bearer ${token}`; + if (apiKey) result['x-api-key'] = apiKey; + return result; + } + async function read(method, path, body, binary = false) { + const started = version; + const options = { method, headers: headers(), redirect: 'error' }; + if (body !== undefined) { + options.headers['Content-Type'] = 'application/json'; + options.body = JSON.stringify(body); + } + const response = await fetchImpl(url(path), options); + let data; + if (binary && response.ok) data = await response.blob(); + else { + const text = await response.text(); + try { data = text ? JSON.parse(text) : null; } catch { data = text; } + } + if (version !== started) throw new Error('Account changed during request'); + if (!response.ok) { + const error = new Error(data?.error || data?.message || `Request failed with status ${response.status}`); + error.status = response.status; + error.data = data; + throw error; + } + return data; + } + async function attachment(path) { + if (!token) throw new Error('Sign in to download attachments'); + // Never forward a bearer to a path supplied by another origin or endpoint. + if (!/^\/uploads\/[^/?#\\]+$/.test(path)) throw new Error('Invalid attachment path'); + let name; + try { name = decodeURIComponent(path.slice('/uploads/'.length)); } catch { /* invalid encoding */ } + if (!name || name === '.' || name === '..' || /[/\\\0]/.test(name)) throw new Error('Invalid attachment path'); + return read('GET', path, undefined, true); + } + return { setToken, getToken: () => token, changes, url, headers, request: read, attachment }; +} diff --git a/frontend/chat-ui/src/api.js b/frontend/chat-ui/src/api.js index 76673fd0..b4c0af30 100644 --- a/frontend/chat-ui/src/api.js +++ b/frontend/chat-ui/src/api.js @@ -1,5 +1,9 @@ +import { createAccountClient } from './accountClient.mjs'; + export const API_BASE_URL = import.meta.env.VITE_API_BASE_URL || ''; const API_KEY = import.meta.env.VITE_API_KEY || ''; +export const accountClient = createAccountClient({ baseUrl: API_BASE_URL, apiKey: API_KEY }); +export const setAccountToken = accountClient.setToken; function buildUrl(path) { const clean = path.startsWith('/') ? path : `/${path}`; @@ -7,33 +11,17 @@ function buildUrl(path) { } async function request(method, path, body) { - const url = buildUrl(path); - const headers = {}; - if (API_KEY) headers['x-api-key'] = API_KEY; - const options = { method, headers }; - if (body !== undefined) { - headers['Content-Type'] = 'application/json'; - options.body = JSON.stringify(body); - } + return accountClient.request(method, path.startsWith('/') ? path : `/${path}`, body); +} - const res = await fetch(url, options); - const text = await res.text(); - let data = null; - if (text) { - try { - data = JSON.parse(text); - } catch (err) { - // non-JSON response; preserve text in data - data = text; - } - } - if (!res.ok) { - const error = new Error(data?.message || `Request failed with status ${res.status}`); - error.status = res.status; - error.data = data; - throw error; - } - return data; +export async function downloadAttachment(file) { + const blob = await accountClient.attachment(file.path); + const objectUrl = URL.createObjectURL(blob); + const link = document.createElement('a'); + link.href = objectUrl; + link.download = file.originalname || 'attachment'; + link.click(); + setTimeout(() => URL.revokeObjectURL(objectUrl), 0); } export const get = (path) => request('GET', path); diff --git a/frontend/chat-ui/src/chatReducer.mjs b/frontend/chat-ui/src/chatReducer.mjs index 163bc39b..4c30cae3 100644 --- a/frontend/chat-ui/src/chatReducer.mjs +++ b/frontend/chat-ui/src/chatReducer.mjs @@ -1,4 +1,5 @@ const handlers = { + RESET_ACCOUNT: (state) => ({ ...state, messages: [], conversations: [], currentConversation: null, aiThinking: false, accountVersion: (state.accountVersion || 0) + 1 }), ADD_MESSAGE: (state, action) => ({ ...state, messages: [...state.messages, action.message] }), UPSERT_MESSAGE(state, action) { const index = state.messages.findIndex((message) => message.id === action.message.id); diff --git a/frontend/chat-ui/src/components/MessageBubble.jsx b/frontend/chat-ui/src/components/MessageBubble.jsx index 5a629303..8d46792f 100644 --- a/frontend/chat-ui/src/components/MessageBubble.jsx +++ b/frontend/chat-ui/src/components/MessageBubble.jsx @@ -1,5 +1,6 @@ -import React from 'react'; +import React, { useState } from 'react'; import MessageActions from './MessageActions.jsx'; +import { downloadAttachment } from '../api'; /** @typedef {import('../types').Message} Message */ @@ -8,6 +9,12 @@ import MessageActions from './MessageActions.jsx'; * @param {{message: Message}} props */ export default function MessageBubble({ message, index = 0 }) { + const [downloadError, setDownloadError] = useState(''); + const download = async () => { + setDownloadError(''); + try { await downloadAttachment(message.file); } + catch (error) { setDownloadError(error.message); } + }; const isUser = message.role === 'user' || message.sender === 'user' || message.from === 'user'; const alignment = isUser ? 'items-end text-right' : 'items-start text-left'; const bubbleClass = `message-bubble ${isUser ? 'user' : 'assistant'}`; @@ -18,9 +25,9 @@ export default function MessageBubble({ message, index = 0 }) { let content; if (message.type === 'file' && message.file) { content = ( - + ); } else { const text = message.content || message.message || ''; @@ -32,6 +39,7 @@ export default function MessageBubble({ message, index = 0 }) { {!isUser && }
{content} + {downloadError &&

{downloadError}

}
); diff --git a/frontend/chat-ui/src/store.jsx b/frontend/chat-ui/src/store.jsx index 66c55e23..22ed9b26 100644 --- a/frontend/chat-ui/src/store.jsx +++ b/frontend/chat-ui/src/store.jsx @@ -1,5 +1,6 @@ import { reducer } from './chatReducer.mjs'; -import React, { createContext, useContext, useReducer } from 'react'; +import React, { createContext, useContext, useEffect, useReducer } from 'react'; +import { accountClient } from './api'; /** * @typedef {import('./types').Message} Message @@ -53,8 +54,13 @@ const StoreContext = createContext(); */ export function StoreProvider({ children }) { const [state, dispatch] = useReducer(reducer, initialState); + useEffect(() => { + const reset = () => dispatch({ type: 'RESET_ACCOUNT' }); + accountClient.changes.addEventListener('change', reset); + return () => accountClient.changes.removeEventListener('change', reset); + }, []); return ( - + {children} ); From 5154bdf7b0082d51f0598278f42c3a20cedcea13 Mon Sep 17 00:00:00 2001 From: Nirmal Khedkar Date: Mon, 26 Oct 2026 14:24:48 -0700 Subject: [PATCH 2/3] Authenticate private conversation streams Signed-off-by: Nirmal Khedkar --- CONFORMANCE.md | 8 +- README.md | 8 + backend/__tests__/conversationStreams.test.js | 139 ++++++++++++++++++ backend/docs/authenticated-streams.md | 64 ++++++++ backend/docs/conversation-ownership.md | 4 +- backend/lib/conversationStreams.js | 77 ++++++++++ backend/models/conversation.js | 2 + backend/server.js | 72 ++------- frontend/chat-ui/ARCHITECTURE.md | 2 +- frontend/chat-ui/authenticatedStream.test.js | 63 ++++++++ frontend/chat-ui/src/api.js | 16 +- frontend/chat-ui/src/authenticatedStream.mjs | 78 ++++++++++ .../chat-ui/src/components/ChatWindow.jsx | 2 +- frontend/chat-ui/src/components/Sidebar.jsx | 2 +- 14 files changed, 455 insertions(+), 82 deletions(-) create mode 100644 backend/__tests__/conversationStreams.test.js create mode 100644 backend/docs/authenticated-streams.md create mode 100644 backend/lib/conversationStreams.js create mode 100644 frontend/chat-ui/authenticatedStream.test.js create mode 100644 frontend/chat-ui/src/authenticatedStream.mjs diff --git a/CONFORMANCE.md b/CONFORMANCE.md index a14c2a0b..865d9791 100644 --- a/CONFORMANCE.md +++ b/CONFORMANCE.md @@ -52,6 +52,10 @@ assignment. Bindings: `backend/middleware/chatAuth.js`, `conversationOwner.js`, `attachmentOwner.js`, the conditional conversation model writes, and `backend/__tests__/conversationAccess.test.js`. This does not promote private chat to a sealed commons dialog or change the in-review ledger -claims above. SSE requires the companion authenticated-stream change before -rollout. See `backend/docs/conversation-ownership.md` for migration and client +claims above. SSE uses the same verified owner identity before stream headers, +implemented in `backend/lib/conversationStreams.js` and checked by +`backend/__tests__/conversationStreams.test.js`. The unrestricted global feed +is retired. Streams expire with the account token and close on local parent +deletion; cross-replica invalidation remains a named residual. See +`backend/docs/conversation-ownership.md` for migration and client integration limits. diff --git a/README.md b/README.md index f481f283..b0422090 100644 --- a/README.md +++ b/README.md @@ -125,6 +125,14 @@ Additional chat interface endpoints (e.g., `/conversations`, `/messages/:id`, `/ structures are documented inline in the frontend components [`Sidebar.jsx`](frontend/chat-ui/src/components/Sidebar.jsx) and [`ChatWindow.jsx`](frontend/chat-ui/src/components/ChatWindow.jsx). +Private chat requires the existing backend account JWT, including for SSE and +attachments. Shared API keys do not identify a chat owner. Legacy ownerless +chats remain inaccessible pending explicit operator assignment. The host +application supplies its login token to `setAccountToken`; the chat UI does +not provide a complete onboarding flow. See +[conversation ownership](backend/docs/conversation-ownership.md) and +[authenticated streams](backend/docs/authenticated-streams.md). The global +`/events` feed is retired; use an authorized `/stream/:conversationId` request. ### User Registration #### POST /userRegistration ``` diff --git a/backend/__tests__/conversationStreams.test.js b/backend/__tests__/conversationStreams.test.js new file mode 100644 index 00000000..41804960 --- /dev/null +++ b/backend/__tests__/conversationStreams.test.js @@ -0,0 +1,139 @@ +const jwt = require('jsonwebtoken'); +const mockConversations = new Map(); +let mockCleanupFailure = false; +let mockHoldRead; +jest.mock('../lib/dynamodbClient', () => ({ + get: ({ Key }) => ({ promise: async () => { + const Item = mockConversations.get(Key.id); + const hold = mockHoldRead; + mockHoldRead = undefined; + if (hold) { hold.started(); await hold.wait; } + return { Item }; + } }), + delete: ({ Key }) => ({ promise: async () => { mockConversations.delete(Key.id); return {}; } }), + query: () => ({ promise: async () => { + if (mockCleanupFailure) throw new Error('fixture cleanup failure'); + return { Items: [] }; + } }), +})); +jest.mock('../bull/pingJobs', () => ({ addPingJob: jest.fn() })); +jest.mock('../bull/smsQueue', () => ({ enqueueSms: jest.fn() })); +jest.mock('../federation/federationSyncJob', () => jest.fn()); + +describe('production conversation SSE boundary', () => { + const originalEnv = { ...process.env }; + const originalGlobals = Object.fromEntries(['broadcast', 'emitToken', 'emitMessage'].map(key => [key, global[key]])); + let server; + let base; + const controllers = []; + const bearer = (userId, seconds = 60) => `Bearer ${jwt.sign({ userId }, process.env.JWT_SECRET, { expiresIn: seconds })}`; + async function open(path, headers = {}) { + const controller = new AbortController(); + controllers.push(controller); + return fetch(base + path, { headers, signal: AbortSignal.any([controller.signal, AbortSignal.timeout(1500)]) }); + } + beforeAll(async () => { + process.env.MINIMAL_SERVER = 'false'; + process.env.JWT_SECRET = 'stream-fixture-secret'; + process.env.API_KEY = 'stream-fixture-key'; + ({ server } = require('../server')); + await new Promise(resolve => server.listen(0, '127.0.0.1', resolve)); + base = `http://127.0.0.1:${server.address().port}`; + }); + beforeEach(() => { + mockCleanupFailure = false; + mockHoldRead = undefined; + mockConversations.clear(); + mockConversations.set('mine', { id: 'mine', ownerId: 'alice' }); + mockConversations.set('foreign', { id: 'foreign', ownerId: 'bob' }); + mockConversations.set('legacy', { id: 'legacy' }); + }); + afterEach(() => { for (const controller of controllers.splice(0)) controller.abort(); }); + afterAll(async () => { + server.closeAllConnections(); + await new Promise(resolve => server.close(resolve)); + process.env = originalEnv; + for (const [key, value] of Object.entries(originalGlobals)) { + if (value === undefined) delete global[key]; else global[key] = value; + } + }); + + test.each([ + ['anonymous', {}, ''], + ['shared key', { 'x-api-key': 'stream-fixture-key' }, ''], + ['invalid bearer and shared key', { Authorization: 'Bearer invalid', 'x-api-key': 'stream-fixture-key' }, ''], + ['query key', {}, '?x-api-key=stream-fixture-key'], + ])('rejects %s before opening stream headers', async (_name, headers, query) => { + let response; + try { response = await open('/stream/mine' + query, headers); } catch { response = { status: 'open stream', headers: new Headers() }; } + expect(response.status).toBe(401); + expect(response.headers.get('content-type')).not.toContain('text/event-stream'); + }); + test.each(['foreign', 'legacy', 'missing'])('conceals %s from an authenticated non-owner', async id => { + let response; + try { response = await open(`/stream/${id}`, { Authorization: bearer('alice') }); } catch { response = { status: 'open stream' }; } + expect(response.status).toBe(404); + expect(await response.json()).toEqual({ error: 'Not found' }); + }); + test('query bearer is not a substitute for an Authorization header', async () => { + const token = bearer('alice').slice(7); + let response; + try { response = await open(`/stream/mine?access_token=${token}`); } catch { response = { status: 'open stream' }; } + expect(response.status).toBe(401); + }); + test('retires the unrestricted global feed rather than exposing transaction events to every account', async () => { + const response = await open('/events', { Authorization: bearer('alice') }); + expect(response.status).toBe(410); + expect(response.headers.get('content-type')).not.toContain('text/event-stream'); + }); + test('owner receives only conversation-scoped events and deletion ends the stream', async () => { + const response = await open('/stream/mine', { Authorization: bearer('alice') }); + expect(response.status).toBe(200); + expect(response.headers.get('cache-control')).toContain('no-store'); + const reader = response.body.getReader(); + global.emitMessage('foreign', { content: 'Bobs secret' }); + global.emitMessage('mine', { content: 'Alice answer' }); + const chunk = new TextDecoder().decode((await reader.read()).value); + expect(chunk).toContain('Alice answer'); + expect(chunk).not.toContain('Bobs secret'); + const deletion = await fetch(base + '/conversations/mine', { method: 'DELETE', headers: { Authorization: bearer('alice') } }); + expect(deletion.status).toBe(204); + expect((await reader.read()).done).toBe(true); + }); + test('a connection closes when its account token expires', async () => { + const response = await open('/stream/mine', { Authorization: bearer('alice', 1) }); + expect(response.status).toBe(200); + const read = await response.body.getReader().read(); + expect(read.done).toBe(true); + expect((await open('/stream/mine', { Authorization: bearer('alice', -1) })).status).toBe(401); + }); + + test('parent deletion closes streams even when message cleanup fails', async () => { + const response = await open('/stream/mine', { Authorization: bearer('alice') }); + const reader = response.body.getReader(); + mockCleanupFailure = true; + const expectedError = jest.spyOn(console, 'error').mockImplementation(() => {}); + const deletion = await fetch(base + '/conversations/mine', { method: 'DELETE', headers: { Authorization: bearer('alice') } }); + expectedError.mockRestore(); + expect(deletion.status).toBe(500); + expect(mockConversations.has('mine')).toBe(false); + global.emitMessage('mine', { content: 'late answer' }); + const next = await reader.read(); + expect(next.done).toBe(true); + }); + + test('deletion revokes a stream whose ownership lookup is still in flight', async () => { + let release; + let started; + const reading = new Promise(resolve => { started = resolve; }); + mockHoldRead = { started, wait: new Promise(resolve => { release = resolve; }) }; + const pending = open('/stream/mine', { Authorization: bearer('alice') }); + await reading; + const deletion = await fetch(base + '/conversations/mine', { method: 'DELETE', headers: { Authorization: bearer('alice') } }); + expect(deletion.status).toBe(204); + release(); + const response = await pending; + expect(response.status).toBe(404); + expect(response.headers.get('content-type')).not.toContain('text/event-stream'); + }); +}); diff --git a/backend/docs/authenticated-streams.md b/backend/docs/authenticated-streams.md new file mode 100644 index 00000000..c4dab46a --- /dev/null +++ b/backend/docs/authenticated-streams.md @@ -0,0 +1,64 @@ +# Authenticated conversation streams + +This change depends on conversation ownership (A1). Deploy both changes +together. The existing backend account JWT must be supplied as +`Authorization: Bearer `; its verified `userId` must own the requested +conversation. Tokens must have a finite future expiry, as backend login tokens +already do. API keys and credentials in query strings grant no stream access. + +`GET /stream/:conversationId` authorizes before sending SSE headers or +subscribing to live events. Foreign, missing and ownerless conversations have +the same 404 response. Successful streams preserve the existing `token` and +`message` payloads, disable caching, and request that reverse proxies disable +buffering. Authorization errors are JSON responses, never open SSE connections. + +The unrestricted `GET /events` feed is retired: authenticated requests receive +410 and unauthenticated requests fail account authentication. No current chat +component uses it. Generic transaction broadcast producers remain callable, +but their events are no longer exposed to arbitrary subscribers. Reintroducing +transaction events needs a specific recipient-authorization contract; this PR +does not infer one from a shared key or broad account membership. + +Streams stop on socket closure/error and token expiry. The emitter rechecks +expiry before each write so delayed timers do not send post-expiry frames. +Parent conversation deletion revokes ready and pending connections on the +serving process immediately after the conditional database delete, even if +later message cleanup fails. Pending registration precedes its strongly +consistent owner lookup; a concurrent delete therefore cannot be missed by +a late ownership-read response. This is a process-local stream registry, not +a distributed invalidation system: another replica's deletion or manual store +changes are not pushed to existing sockets here. Tokens remain the existing +signed bearer capability until expiry; no server-wide JWT revocation system +is introduced. + +## Browser transport + +Native EventSource cannot attach Authorization headers. The chat adapter now +uses streaming fetch and keeps the subset ChatWindow uses: add/remove event +listeners and `close()`. It parses UTF-8 SSE token/message frames across chunk +and CRLF boundaries. It retries transient network disconnects after one second, +with fresh account authorization, and stops on failed HTTP authorization or an +invalid content type. There is no replay guarantee or Last-Event-ID contract. +Buffering has a defensive 1 MiB text limit; oversized responses are cancelled. + +Call A1's `setAccountToken(token)` after the host application's existing login, +and `setAccountToken(null)` on logout. Every account change closes existing +streams immediately and remounts private chat state. Old frames are not +dispatched after closure. Credentials stay in headers and memory, never URLs, +browser storage or build-time account-token configuration. A shared +`VITE_API_KEY` alone is insufficient. Full browser onboarding remains outside +these two changes; the integration limits in `conversation-ownership.md` apply. + +## Verification + +`cd backend && npm test` exercises the production server using real HTTP +connections, verified account JWTs and isolated persistence/queue fixtures. +Coverage includes anonymous/shared-key/query-key/invalid-token denial, foreign +and legacy IDs, owner-only payloads, token expiry, committed deletion with +failed child cleanup, and deletion racing a pending ownership lookup. + +`cd frontend/chat-ui && npm test` checks header credentials, fragmented SSE +frames, cancellation, account changes and terminal authorization errors. +`npm run build` verifies the browser bundle. These are private application +events; no Mycelium, covenant, escrow, protocol-wire or federation semantics +are changed, and the existing conformance stand-ins remain labeled. diff --git a/backend/docs/conversation-ownership.md b/backend/docs/conversation-ownership.md index bdb7cf5b..0df16788 100644 --- a/backend/docs/conversation-ownership.md +++ b/backend/docs/conversation-ownership.md @@ -45,8 +45,8 @@ or table. They prioritize correct isolation; an indexed attachment lookup is a possible measured follow-up, not supplied by this change. **Roll out together with the authenticated-stream companion PR (A2).** A1 -closes REST and attachment access; unauthenticated SSE remains a separate -existing path until A2 lands. Other open hardening work, including admin, +closes REST and attachment access; A2 supplies account-authorized SSE. See +`authenticated-streams.md` for that boundary and lifecycle. Other open hardening work, including admin, user/federation authorization and attachment write containment, remains separate and must be assessed before exposing public signup. diff --git a/backend/lib/conversationStreams.js b/backend/lib/conversationStreams.js new file mode 100644 index 00000000..8453fa90 --- /dev/null +++ b/backend/lib/conversationStreams.js @@ -0,0 +1,77 @@ +const streams = new Map(); +const MAX_TIMER_MS = 2 ** 31 - 1; + +async function open(req, res) { + const expiresAt = req.user.exp * 1000; + if (!Number.isFinite(expiresAt) || expiresAt <= Date.now()) { + return res.status(401).json({ error: 'Expiring account token required' }); + } + const id = req.params.conversationId; + if (!streams.has(id)) streams.set(id, new Set()); + const clients = streams.get(id); + let timer; + const client = { res, expiresAt, ready: false, revoked: false }; + const cleanup = () => { + clearTimeout(timer); + clients.delete(client); + if (!clients.size && streams.get(id) === clients) streams.delete(id); + }; + const expire = () => { + const remaining = expiresAt - Date.now(); + if (remaining <= 0) { cleanup(); res.end(); } + else timer = setTimeout(expire, Math.min(remaining, MAX_TIMER_MS)); + }; + res.on('close', cleanup); + res.on('error', cleanup); + clients.add(client); + // Register the pending connection BEFORE reading ownership, so a concurrent + // committed delete can revoke it even if the database response is in flight. + let conversation; + try { conversation = await require('../models/conversation').getOwned(id, req.user.userId); } + catch (error) { cleanup(); throw error; } + if (!conversation || client.revoked) { + cleanup(); + return res.status(404).json({ error: 'Not found' }); + } + if (res.destroyed) { cleanup(); return; } + if (expiresAt <= Date.now()) { + cleanup(); + return res.status(401).json({ error: 'Expiring account token required' }); + } + client.ready = true; + res.status(200).set({ + 'Content-Type': 'text/event-stream', + 'Cache-Control': 'private, no-store', + Connection: 'keep-alive', + 'X-Accel-Buffering': 'no', + }); + res.flushHeaders(); + expire(); +} + +function send(id, event, data) { + const clients = streams.get(id); + if (!clients) return; + const payload = `event: ${event}\ndata: ${JSON.stringify(data)}\n\n`; + for (const client of clients) { + if (!client.ready || client.revoked) continue; + // Check at emission too: an overdue timer must not leak post-expiry data. + if (client.expiresAt <= Date.now() || client.res.destroyed || client.res.writableEnded) { + client.res.end(); + continue; + } + client.res.write(payload); + } +} + +function close(id) { + const clients = streams.get(id); + if (!clients) return; + for (const client of clients) { + client.revoked = true; + if (client.ready) client.res.end(); + } + streams.delete(id); +} + +module.exports = { open, send, close }; diff --git a/backend/models/conversation.js b/backend/models/conversation.js index c0a28ef0..02cd2482 100644 --- a/backend/models/conversation.js +++ b/backend/models/conversation.js @@ -75,6 +75,8 @@ async function remove(id, ownerId) { if (err.code === 'ConditionalCheckFailedException') return false; throw err; } + // Closing streams must not depend on later narrative cleanup succeeding. + require('../lib/conversationStreams').close(id); await Message.removeByConversation(id); return true; } diff --git a/backend/server.js b/backend/server.js index f127e30e..953fee8b 100644 --- a/backend/server.js +++ b/backend/server.js @@ -62,73 +62,21 @@ app.get('/health', (req, res) => { // Server & SSE event stream const server = http.createServer(app); -// Simple Server-Sent Events implementation -const sseClients = new Set(); -// Map of conversationId -> Set of response objects -const conversationStreams = new Map(); - -app.get('/events', (req, res) => { - res.writeHead(200, { - 'Content-Type': 'text/event-stream', - 'Cache-Control': 'no-cache', - Connection: 'keep-alive' - }); - sseClients.add(res); - req.on('close', () => sseClients.delete(res)); -}); - -// Unified conversation-scoped Server-Sent Events endpoint -app.get('/stream/:conversationId', (req, res) => { - const { conversationId } = req.params; - res.writeHead(200, { - 'Content-Type': 'text/event-stream', - 'Cache-Control': 'no-cache', - Connection: 'keep-alive' - }); - - if (!conversationStreams.has(conversationId)) { - conversationStreams.set(conversationId, new Set()); - } - - const clients = conversationStreams.get(conversationId); - clients.add(res); - - req.on('close', () => { - clients.delete(res); - if (clients.size === 0) { - conversationStreams.delete(conversationId); - } - }); +// Account-scoped streams share the same owner check as message reads/sends. +const conversationStreams = require('./lib/conversationStreams'); +app.get('/events', chatAuth, (_req, res) => { + res.status(410).json({ error: 'Global event feed retired; use an owned conversation stream' }); }); - -function sendConversationEvent(conversationId, event, data) { - const clients = conversationStreams.get(conversationId); - if (!clients) return; - const payload = `event: ${event}\ndata: ${JSON.stringify(data)}\n\n`; - clients.forEach(res => res.write(payload)); -} +app.get('/stream/:conversationId', chatAuth, asyncHandler(conversationStreams.open)); function broadcast(event, data, conversationId) { - if (conversationId) { - sendConversationEvent(conversationId, event, data); - return; - } - const payload = `event: ${event}\ndata: ${JSON.stringify(data)}\n\n`; - sseClients.forEach(res => res.write(payload)); + if (conversationId) conversationStreams.send(conversationId, event, data); + // Generic transaction broadcasts have no recipient authorization contract. + // Keep producers callable, but never expose them through a global feed. } global.broadcast = broadcast; - -// Helpers to emit token and message events -function emitToken(conversationId, id, token) { - sendConversationEvent(conversationId, 'token', { id, token }); -} - -function emitMessage(conversationId, message) { - sendConversationEvent(conversationId, 'message', { message }); -} - -global.emitToken = emitToken; -global.emitMessage = emitMessage; +global.emitToken = (conversationId, id, token) => conversationStreams.send(conversationId, 'token', { id, token }); +global.emitMessage = (conversationId, message) => conversationStreams.send(conversationId, 'message', { message }); // Middleware app.use(authMiddleware); diff --git a/frontend/chat-ui/ARCHITECTURE.md b/frontend/chat-ui/ARCHITECTURE.md index d7119f75..f43ed345 100644 --- a/frontend/chat-ui/ARCHITECTURE.md +++ b/frontend/chat-ui/ARCHITECTURE.md @@ -30,7 +30,7 @@ This document describes the component-level architecture and user-experience flo ## 4. Sidebar (Left Panel) - **Component**: `Sidebar` with nested `ChatList` entries. -- **API notes**: Comments in [`src/components/Sidebar.jsx`](src/components/Sidebar.jsx) outline conversation/message endpoints, expected JSON, and `x-api-key` usage. +- **API notes**: Comments in [`src/components/Sidebar.jsx`](src/components/Sidebar.jsx) outline conversation/message endpoints and expected JSON. Private chat uses a backend account bearer supplied through `setAccountToken`; shared keys grant no private-chat access. SSE uses streaming fetch with Authorization headers, not native EventSource query credentials. See [`authenticated-streams.md`](../../backend/docs/authenticated-streams.md) for lifecycle and integration limits. - **New Chat**: Clears `messages[]` and establishes a new conversation identifier. - **History**: Loaded from `/conversations` and displayed as `ConversationCard` items with lazy‑loaded pagination. - **Pinned Chats**: Items marked with `pinned: true` render in a dedicated section. diff --git a/frontend/chat-ui/authenticatedStream.test.js b/frontend/chat-ui/authenticatedStream.test.js new file mode 100644 index 00000000..490dfbb7 --- /dev/null +++ b/frontend/chat-ui/authenticatedStream.test.js @@ -0,0 +1,63 @@ +const { test } = require('node:test'); +const assert = require('node:assert/strict'); +const { once } = require('node:events'); + +test('stream authenticates by header and parses fragmented CRLF token/message events', async () => { + const { createAccountClient } = await import('./src/accountClient.mjs'); + const { authenticatedStream } = await import('./src/authenticatedStream.mjs'); + const client = createAccountClient({ baseUrl: 'https://backend.example' }); + client.setToken('alice-token'); + let headers; + let destination; + const encoder = new TextEncoder(); + const stream = authenticatedStream(client, '/stream/mine', { fetchImpl: async (url, options) => { + headers = options.headers; + destination = url; + return new Response(new ReadableStream({ start(controller) { + for (const chunk of ['event: to', 'ken\r', '\ndata: {"token":"hello"}\r\n\r', '\nevent: message\ndata: {"message":{"id":"answer"}}\n\n']) controller.enqueue(encoder.encode(chunk)); + } }), { headers: { 'Content-Type': 'text/event-stream' } }); + } }); + const tokens = []; + stream.addEventListener('token', event => tokens.push(JSON.parse(event.data).token)); + const [message] = await once(stream, 'message'); + assert.deepEqual(tokens, ['hello']); + assert.equal(JSON.parse(message.data).message.id, 'answer'); + assert.equal(headers.Authorization, 'Bearer alice-token'); + assert.equal(destination, 'https://backend.example/stream/mine'); + stream.close(); +}); + +test('account change aborts and closes the old stream before later frames dispatch', async () => { + const { createAccountClient } = await import('./src/accountClient.mjs'); + const { authenticatedStream } = await import('./src/authenticatedStream.mjs'); + const client = createAccountClient(); + client.setToken('alice-token'); + let signal; + let cancelled = false; + const stream = authenticatedStream(client, '/stream/mine', { fetchImpl: async (_url, options) => { + signal = options.signal; + return new Response(new ReadableStream({ cancel() { cancelled = true; } }), { headers: { 'Content-Type': 'text/event-stream' } }); + } }); + await once(stream, 'open'); + client.setToken('bob-token'); + await new Promise(resolve => setImmediate(resolve)); + assert.equal(signal.aborted, true); + assert.equal(cancelled, true); + assert.equal(stream.readyState, 2); +}); + +test('authorization failures stop reconnecting and no token means no request', async () => { + const { createAccountClient } = await import('./src/accountClient.mjs'); + const { authenticatedStream } = await import('./src/authenticatedStream.mjs'); + const client = createAccountClient(); + let requests = 0; + const fetchImpl = async () => { requests++; return new Response('denied', { status: 401 }); }; + const anonymous = authenticatedStream(client, '/stream/mine', { fetchImpl }); + await once(anonymous, 'error'); + assert.equal(requests, 0); + client.setToken('account-token'); + const denied = authenticatedStream(client, '/stream/mine', { fetchImpl }); + await once(denied, 'error'); + assert.equal(denied.readyState, 2); + assert.equal(requests, 1); +}); diff --git a/frontend/chat-ui/src/api.js b/frontend/chat-ui/src/api.js index b4c0af30..6a0bca2c 100644 --- a/frontend/chat-ui/src/api.js +++ b/frontend/chat-ui/src/api.js @@ -1,15 +1,11 @@ import { createAccountClient } from './accountClient.mjs'; +import { authenticatedStream } from './authenticatedStream.mjs'; export const API_BASE_URL = import.meta.env.VITE_API_BASE_URL || ''; const API_KEY = import.meta.env.VITE_API_KEY || ''; export const accountClient = createAccountClient({ baseUrl: API_BASE_URL, apiKey: API_KEY }); export const setAccountToken = accountClient.setToken; -function buildUrl(path) { - const clean = path.startsWith('/') ? path : `/${path}`; - return API_BASE_URL ? `${API_BASE_URL.replace(/\/$/, '')}${clean}` : clean; -} - async function request(method, path, body) { return accountClient.request(method, path.startsWith('/') ? path : `/${path}`, body); } @@ -29,15 +25,9 @@ export const post = (path, body) => request('POST', path, body); export const put = (path, body) => request('PUT', path, body); export const del = (path) => request('DELETE', path); -// Create an EventSource for SSE. Many servers don't accept custom headers for SSE, -// so we append the API key as a query parameter named `x-api-key` when present. +// Native EventSource cannot carry Authorization. Use streaming fetch instead. export function stream(path) { - let url = buildUrl(path); - if (API_KEY) { - const hasQuery = url.includes('?'); - url = `${url}${hasQuery ? '&' : '?'}x-api-key=${encodeURIComponent(API_KEY)}`; - } - return new EventSource(url); + return authenticatedStream(accountClient, path); } const api = { get, post, put, del, stream }; diff --git a/frontend/chat-ui/src/authenticatedStream.mjs b/frontend/chat-ui/src/authenticatedStream.mjs new file mode 100644 index 00000000..187addf9 --- /dev/null +++ b/frontend/chat-ui/src/authenticatedStream.mjs @@ -0,0 +1,78 @@ +// EventSource-shaped subset used by ChatWindow, with bearer headers and an +// abortable lifetime. No credential is ever appended to a stream URL. +export function authenticatedStream(client, path, { fetchImpl = (...args) => fetch(...args), retryMs = 1000 } = {}) { + const events = new EventTarget(); + const controller = new AbortController(); + let reader; + let retry; + let closed = false; + events.readyState = 0; + events.close = () => { + closed = true; + events.readyState = 2; + clearTimeout(retry); + controller.abort(); + if (reader) void reader.cancel().catch(() => {}); + client.changes.removeEventListener('change', events.close); + }; + client.changes.addEventListener('change', events.close); + + async function connect() { + if (closed) return; + try { + if (!client.getToken()) { + events.close(); + events.dispatchEvent(new Event('error')); + return; + } + const response = await fetchImpl(client.url(path), { + headers: { ...client.headers(), Accept: 'text/event-stream' }, + signal: controller.signal, + redirect: 'error', + }); + if (closed) { await response.body?.cancel(); return; } + if (!response.ok || !response.headers.get('content-type')?.startsWith('text/event-stream')) { + await response.body?.cancel(); + events.close(); + events.dispatchEvent(new Event('error')); + return; + } + reader = response.body.getReader(); + events.readyState = 1; + events.dispatchEvent(new Event('open')); + const decoder = new TextDecoder(); + let buffer = ''; + let type = 'message'; + let data = []; + let frameSize = 0; + while (!closed) { + const result = await reader.read(); + if (closed || result.done) break; + buffer += decoder.decode(result.value, { stream: true }); + if (buffer.length + frameSize > 1024 * 1024) throw new Error('Stream frame too large'); + while (!closed) { + const end = buffer.search(/[\r\n]/); + if (end < 0 || (buffer[end] === '\r' && end === buffer.length - 1)) break; + const line = buffer.slice(0, end); + const width = buffer[end] === '\r' && buffer[end + 1] === '\n' ? 2 : 1; + buffer = buffer.slice(end + width); + if (!line) { + if (data.length) events.dispatchEvent(new MessageEvent(type, { data: data.join('\n') })); + type = 'message'; data = []; frameSize = 0; + } else if (line.startsWith('event:')) type = line.slice(6).replace(/^ /, '') || 'message'; + else if (line.startsWith('data:')) { data.push(line.slice(5).replace(/^ /, '')); frameSize += line.length; } + } + } + } catch { + if (!closed) events.dispatchEvent(new Event('error')); + } finally { + if (reader) { await reader.cancel().catch(() => {}); reader = null; } + if (!closed) { + events.readyState = 0; + retry = setTimeout(connect, retryMs); + } + } + } + queueMicrotask(connect); + return events; +} diff --git a/frontend/chat-ui/src/components/ChatWindow.jsx b/frontend/chat-ui/src/components/ChatWindow.jsx index 81efe4c1..2e3c5429 100644 --- a/frontend/chat-ui/src/components/ChatWindow.jsx +++ b/frontend/chat-ui/src/components/ChatWindow.jsx @@ -5,7 +5,7 @@ * token -> { "id": string, "token": string } * message -> { "message": { id, role, content } } * - * Assumes server-side auth (no 'x-api-key' header sent). + * Streaming fetch carries the current backend account bearer in a header. * Errors while parsing events are logged to the console. */ import React, { useEffect, useRef } from 'react'; diff --git a/frontend/chat-ui/src/components/Sidebar.jsx b/frontend/chat-ui/src/components/Sidebar.jsx index 3c7488c3..34b32f10 100644 --- a/frontend/chat-ui/src/components/Sidebar.jsx +++ b/frontend/chat-ui/src/components/Sidebar.jsx @@ -5,7 +5,7 @@ * GET /messages/{conversationId} -> [{ id, role, content }] * POST /conversations/{id}/pin -> body: { pinned: boolean } * - * All requests include an 'x-api-key' header and expect JSON responses. + * Private chat requests use the current backend account bearer and expect JSON. * Errors are caught and logged to the console without user-facing feedback. */ import React, { useEffect, useState } from 'react'; From fc1c498cb41bbd050060f71ba81dd1bec5e41184 Mon Sep 17 00:00:00 2001 From: Nirmal Khedkar Date: Fri, 30 Oct 2026 21:44:46 -0700 Subject: [PATCH 3/3] fix(chat): address authenticated stream review Apply account request budgets and cap pending plus active streams at five per account. Release capacity on expiry, disconnect and deletion, and return 410 for the retired global feed without authentication. Carry shared ownership and attachment fixes. Stop oversized-frame retries while preserving valid frames coalesced in one network chunk. Add regression coverage and update deployment documentation. --- backend/__tests__/attachmentStorage.test.js | 42 +++++++ backend/__tests__/chatRateLimit.test.js | 44 +++++++ backend/__tests__/conversationAccess.test.js | 24 ++++ backend/__tests__/conversationStreams.test.js | 21 +++- backend/__tests__/streamCapacity.test.js | 98 ++++++++++++++++ backend/docs/authenticated-streams.md | 21 +++- backend/docs/conversation-ownership.md | 49 ++++++-- backend/lib/conversationStreams.js | 68 ++++++++--- backend/middleware/chatRateLimit.js | 18 +++ backend/middleware/conversationAttachment.js | 21 ++++ backend/models/message.js | 32 +++++- backend/package-lock.json | 52 +++++++++ backend/package.json | 1 + backend/routes/communicationRoutes.js | 2 +- backend/routes/conversationRoutes.js | 3 +- .../scripts/check-conversation-ownership.js | 2 + backend/server.js | 8 +- frontend/chat-ui/accountClient.test.js | 36 ++++++ frontend/chat-ui/authenticatedStream.test.js | 18 +++ frontend/chat-ui/signIn.test.js | 22 ++++ frontend/chat-ui/src/accountClient.mjs | 83 +++++++++----- frontend/chat-ui/src/api.js | 4 +- frontend/chat-ui/src/authenticatedStream.mjs | 107 ++++++++---------- .../chat-ui/src/components/ChatWindow.jsx | 8 +- .../chat-ui/src/components/MessageBubble.jsx | 2 +- frontend/chat-ui/src/components/Sidebar.jsx | 7 +- frontend/chat-ui/src/sseFrames.mjs | 43 +++++++ frontend/chat-ui/sseFrames.test.js | 31 +++++ 28 files changed, 729 insertions(+), 138 deletions(-) create mode 100644 backend/__tests__/attachmentStorage.test.js create mode 100644 backend/__tests__/chatRateLimit.test.js create mode 100644 backend/__tests__/streamCapacity.test.js create mode 100644 backend/middleware/chatRateLimit.js create mode 100644 backend/middleware/conversationAttachment.js create mode 100644 frontend/chat-ui/signIn.test.js create mode 100644 frontend/chat-ui/src/sseFrames.mjs create mode 100644 frontend/chat-ui/sseFrames.test.js diff --git a/backend/__tests__/attachmentStorage.test.js b/backend/__tests__/attachmentStorage.test.js new file mode 100644 index 00000000..110ab305 --- /dev/null +++ b/backend/__tests__/attachmentStorage.test.js @@ -0,0 +1,42 @@ +const { test } = require('node:test'); +const assert = require('node:assert/strict'); +const fs = require('node:fs'); +const path = require('node:path'); +const Message = require('../models/message'); + +// Same-name writes in the same millisecond must never alias another message. +test('same-name attachments retain distinct bytes and server-generated names', async () => { + Message.setDocClient({ put: () => ({ promise: async () => ({}) }) }); + const now = Date.now; + const files = []; + Date.now = () => 123456789; + try { + for (const content of ['first', 'second']) { + const msg = await Message.sendMessage('mine', { file: { name: 'same.txt', data: Buffer.from(content).toString('base64') } }); + files.push(msg.file); + } + assert.notEqual(files[0].path, files[1].path); + for (const [index, file] of files.entries()) { + assert.match(file.path, /^\/uploads\/[0-9a-f-]{36}\.txt$/); + assert.equal(file.originalname, 'same.txt'); + assert.equal(fs.readFileSync(path.join(__dirname, '..', file.path), 'utf8'), ['first', 'second'][index]); + } + } finally { + Date.now = now; + for (const file of files) fs.rmSync(path.join(__dirname, '..', file.path), { force: true }); + } +}); + +test('scoped attachment lookup follows empty filtered pages within its conversation partition', async () => { + let reads = 0; + Message.setDocClient({ query: params => ({ promise: async () => { + assert.equal(params.KeyConditionExpression, 'conversationId = :cid'); + assert.deepEqual(params.ExpressionAttributeValues, { ':cid': 'mine', ':filePath': '/uploads/100%.txt' }); + reads++; + if (!params.ExclusiveStartKey) return { Items: [], LastEvaluatedKey: { conversationId: 'mine', id: 'first' } }; + assert.deepEqual(params.ExclusiveStartKey, { conversationId: 'mine', id: 'first' }); + return { Items: [{ file: { path: '/uploads/100%.txt' } }] }; + } }) }); + assert.equal(await Message.findConversationAttachment('mine', '/uploads/100%.txt'), true); + assert.equal(reads, 2); +}); diff --git a/backend/__tests__/chatRateLimit.test.js b/backend/__tests__/chatRateLimit.test.js new file mode 100644 index 00000000..b9691d33 --- /dev/null +++ b/backend/__tests__/chatRateLimit.test.js @@ -0,0 +1,44 @@ +const request = require('supertest'); +const jwt = require('jsonwebtoken'); +jest.mock('../lib/dynamodbClient', () => ({ + scan: () => ({ promise: async () => ({ Items: [] }) }), + get: () => ({ promise: async () => ({}) }), +})); +jest.mock('../bull/pingJobs', () => ({ addPingJob: jest.fn() })); +jest.mock('../bull/smsQueue', () => ({ enqueueSms: jest.fn() })); +jest.mock('../federation/federationSyncJob', () => jest.fn()); + +describe('account-keyed rate limits on private routes', () => { + let app; + const originalEnv = { ...process.env }; + const globals = Object.fromEntries(['broadcast', 'emitToken', 'emitMessage'].map(key => [key, global[key]])); + const bearer = userId => `Bearer ${jwt.sign({ userId }, process.env.JWT_SECRET, { expiresIn: '1h' })}`; + beforeAll(() => { + process.env.MINIMAL_SERVER = 'false'; + process.env.JWT_SECRET = 'rate-test-secret'; + ({ app } = require('../server')); + }); + afterAll(() => { + process.env = originalEnv; + for (const [key, value] of Object.entries(globals)) { + if (value === undefined) delete global[key]; else global[key] = value; + } + }); + test.each(['/conversations', '/messages/missing', '/uploads/missing.txt'])('%s rejects excess requests per verified account', async path => { + expect((await request(app).get(path)).status).toBe(403); + const account = 'limit-' + path; + const first = await request(app).get(path).set('Authorization', bearer(account)); + expect(first.status).not.toBe(429); + expect(first.headers['ratelimit-limit']).toBeDefined(); + const limit = Number(first.headers['ratelimit-limit']); + expect(limit).toBeGreaterThan(0); + for (let i = 1; i < limit; i++) { + const result = await request(app).get(path).set('Authorization', bearer(account)); + expect(result.status).not.toBe(429); + } + const blocked = await request(app).get(path).set('Authorization', bearer(account)); + expect(blocked.status).toBe(429); + expect(Number(blocked.headers['retry-after'])).toBeGreaterThan(0); + expect((await request(app).get(path).set('Authorization', bearer('another-' + path))).status).not.toBe(429); + }); +}); diff --git a/backend/__tests__/conversationAccess.test.js b/backend/__tests__/conversationAccess.test.js index c1ba2644..23edc18b 100644 --- a/backend/__tests__/conversationAccess.test.js +++ b/backend/__tests__/conversationAccess.test.js @@ -168,4 +168,28 @@ describe('private conversation HTTP boundary', () => { expect(mockTables.Messages.get('secret').content).toBe('private'); expect(mockTables.Conversations.get('foreign').title).toBe('Bobs private chat'); }); + test.each(['unreferenced', 'deleted'])('conceals an %s attachment reference', async state => { + if (state === 'unreferenced') mockTables.Messages.clear(); + else mockTables.Conversations.delete('foreign'); + const response = await request(app).get(attachment).set('Authorization', authorization('bob')); + expect(response.status).toBe(404); + }); + test('scoped attachment reads query only the owned conversation', async () => { + const response = await request(app).get('/conversations/foreign/attachments/' + filename).set('Authorization', authorization('bob')); + expect(response.status).toBe(200); + expect(response.body.toString()).toBe('private attachment'); + expect(mockMessageReads).toBe(1); + const denied = await request(app).get('/conversations/foreign/attachments/' + filename).set('Authorization', authorization('alice')); + expect(denied.status).toBe(404); + expect(mockMessageReads).toBe(1); + }); + test('trusted chat preflight permits the integration header and denies unknown origins', async () => { + const response = await request(app).options('/conversations').set('Origin', 'https://www.ntari.org') + .set('Access-Control-Request-Method', 'GET').set('Access-Control-Request-Headers', 'authorization,x-api-key,idempotency-key'); + expect(response.status).toBe(204); + expect(response.headers['access-control-allow-headers'].toLowerCase()).toContain('x-api-key'); + const denied = await request(app).options('/conversations').set('Origin', 'https://evil.example'); + expect(denied.headers['access-control-allow-origin']).toBeUndefined(); + }); + }); diff --git a/backend/__tests__/conversationStreams.test.js b/backend/__tests__/conversationStreams.test.js index 41804960..d0c0749b 100644 --- a/backend/__tests__/conversationStreams.test.js +++ b/backend/__tests__/conversationStreams.test.js @@ -30,7 +30,7 @@ describe('production conversation SSE boundary', () => { async function open(path, headers = {}) { const controller = new AbortController(); controllers.push(controller); - return fetch(base + path, { headers, signal: AbortSignal.any([controller.signal, AbortSignal.timeout(1500)]) }); + return fetch(base + path, { headers, signal: AbortSignal.any([controller.signal, AbortSignal.timeout(4000)]) }); } beforeAll(async () => { process.env.MINIMAL_SERVER = 'false'; @@ -101,7 +101,7 @@ describe('production conversation SSE boundary', () => { expect((await reader.read()).done).toBe(true); }); test('a connection closes when its account token expires', async () => { - const response = await open('/stream/mine', { Authorization: bearer('alice', 1) }); + const response = await open('/stream/mine', { Authorization: `Bearer ${jwt.sign({ userId: 'alice', exp: Math.ceil(Date.now() / 1000) + 1 }, process.env.JWT_SECRET)}` }); expect(response.status).toBe(200); const read = await response.body.getReader().read(); expect(read.done).toBe(true); @@ -136,4 +136,21 @@ describe('production conversation SSE boundary', () => { expect(response.status).toBe(404); expect(response.headers.get('content-type')).not.toContain('text/event-stream'); }); + test('retired global feed returns 410 without account credentials', async () => { + const response = await open('/events'); + expect(response.status).toBe(410); + }); + test('stream opening requests have an account budget before ownership reads', async () => { + const auth = { Authorization: bearer('stream-rate-fixture') }; + const first = await open('/stream/missing', auth); + expect(first.status).toBe(404); + expect(first.headers.get('ratelimit-limit')).not.toBeNull(); + const limit = Number(first.headers.get('ratelimit-limit')); + for (let i = 1; i < limit; i++) expect((await open('/stream/missing', auth)).status).toBe(404); + const blocked = await open('/stream/missing', auth); + expect(blocked.status).toBe(429); + expect(Number(blocked.headers.get('retry-after'))).toBeGreaterThan(0); + expect((await open('/stream/missing', { Authorization: bearer('other-stream-user') })).status).toBe(404); + }); + }); diff --git a/backend/__tests__/streamCapacity.test.js b/backend/__tests__/streamCapacity.test.js new file mode 100644 index 00000000..054ab63b --- /dev/null +++ b/backend/__tests__/streamCapacity.test.js @@ -0,0 +1,98 @@ +const { test } = require('node:test'); +const assert = require('node:assert/strict'); +const { EventEmitter } = require('node:events'); +const Conversation = require('../models/conversation'); +const streams = require('../lib/conversationStreams'); + +function response() { + return Object.assign(new EventEmitter(), { + statusCode: 200, + status(code) { this.statusCode = code; return this; }, + json(body) { this.body = body; this.end(); return this; }, + set() { return this; }, flushHeaders() {}, write() {}, + end() { this.writableEnded = true; this.emit('close'); }, + }); +} +function request(id, userId = 'alice') { + return { params: { conversationId: id }, user: { userId, exp: Math.ceil(Date.now() / 1000) + 60 } }; +} + +test('stream capacity is shared across conversations and renewed tokens, isolated by account, and reusable after close', async () => { + Conversation.setDocClient({ get: ({ Key }) => ({ promise: async () => ({ Item: { id: Key.id, ownerId: Key.id.startsWith('bob') ? 'bob' : 'alice' } }) }) }); + const connections = []; + try { + for (let i = 0; i < 5; i++) { + const res = response(); connections.push(res); + await streams.open(request(`alice-${i}`), res); + assert.equal(res.statusCode, 200); + } + const denied = response(); connections.push(denied); + await streams.open(request('sixth'), denied); + assert.equal(denied.statusCode, 429); + const bob = response(); connections.push(bob); + await streams.open(request('bob-1', 'bob'), bob); + assert.equal(bob.statusCode, 200); + connections[0].end(); + connections[0].emit('error', new Error('duplicate cleanup')); + const replacement = response(); connections.push(replacement); + await streams.open(request('replacement'), replacement); + assert.equal(replacement.statusCode, 200); + const stillFull = response(); connections.push(stillFull); + await streams.open(request('still-full'), stillFull); + assert.equal(stillFull.statusCode, 429); + } finally { for (const res of connections) res.end(); } +}); + +test('pending ownership reads reserve capacity and deletion releases it before the database returns', async () => { + let release; + const wait = new Promise(resolve => { release = resolve; }); + Conversation.setDocClient({ get: ({ Key }) => ({ promise: async () => { await wait; return { Item: { id: Key.id, ownerId: 'alice' } }; } }) }); + const connections = Array.from({ length: 5 }, response); + const pending = connections.map((res, index) => streams.open(request(`pending-${index}`), res)); + try { + const denied = response(); connections.push(denied); + await Promise.race([streams.open(request('pending-sixth'), denied), new Promise(resolve => setTimeout(resolve, 100))]); + assert.equal(denied.statusCode, 429); + streams.close('pending-0'); + const replacement = response(); connections.push(replacement); + const next = streams.open(request('replacement'), replacement); + release(); + await Promise.all([...pending, next]); + assert.equal(connections[0].statusCode, 404); + assert.equal(replacement.statusCode, 200); + } finally { release(); await Promise.allSettled(pending); for (const res of connections) res.end(); } +}); + +test('failed ownership checks never consume account stream capacity', async () => { + let fail = true; + Conversation.setDocClient({ get: () => ({ promise: async () => { if (fail) throw new Error('database unavailable'); return {}; } }) }); + for (let i = 0; i < 6; i++) await assert.rejects(streams.open(request('failed'), response()), /database unavailable/); + fail = false; + for (let i = 0; i < 6; i++) { + const res = response(); + await streams.open(request('missing'), res); + assert.equal(res.statusCode, 404); + } +}); + +test('token expiry closes pending reads and frees slots for a renewed token before database recovery', async () => { + let release; + const wait = new Promise(resolve => { release = resolve; }); + Conversation.setDocClient({ get: ({ Key }) => ({ promise: async () => { + if (Key.id !== 'renewed') await wait; + return { Item: { id: Key.id, ownerId: 'alice' } }; + } }) }); + const exp = Math.ceil(Date.now() / 1000) + 1; + const connections = Array.from({ length: 5 }, response); + const pending = connections.map((res, index) => streams.open({ ...request(`expiring-${index}`), user: { userId: 'alice', exp } }, res)); + try { + await new Promise(resolve => setTimeout(resolve, exp * 1000 - Date.now() + 30)); + assert.ok(connections.every(res => res.writableEnded && res.statusCode === 401)); + const renewed = response(); connections.push(renewed); + await streams.open(request('renewed'), renewed); + assert.equal(renewed.statusCode, 200); + release(); + await Promise.all(pending); + assert.ok(connections.slice(0, 5).every(res => res.statusCode === 401)); + } finally { release(); await Promise.allSettled(pending); for (const res of connections) res.end(); } +}); diff --git a/backend/docs/authenticated-streams.md b/backend/docs/authenticated-streams.md index c4dab46a..77380585 100644 --- a/backend/docs/authenticated-streams.md +++ b/backend/docs/authenticated-streams.md @@ -1,6 +1,6 @@ # Authenticated conversation streams -This change depends on conversation ownership (A1). Deploy both changes +This change depends on conversation ownership (#191). Deploy both changes together. The existing backend account JWT must be supplied as `Authorization: Bearer `; its verified `userId` must own the requested conversation. Tokens must have a finite future expiry, as backend login tokens @@ -12,8 +12,8 @@ the same 404 response. Successful streams preserve the existing `token` and `message` payloads, disable caching, and request that reverse proxies disable buffering. Authorization errors are JSON responses, never open SSE connections. -The unrestricted `GET /events` feed is retired: authenticated requests receive -410 and unauthenticated requests fail account authentication. No current chat +The unrestricted `GET /events` feed is retired: all requests receive 410 +without account authentication. No current chat component uses it. Generic transaction broadcast producers remain callable, but their events are no longer exposed to arbitrary subscribers. Reintroducing transaction events needs a specific recipient-authorization contract; this PR @@ -31,6 +31,16 @@ changes are not pushed to existing sockets here. Tokens remain the existing signed bearer capability until expiry; no server-wide JWT revocation system is introduced. +## Resource limits + +Stream opening requests allow 60 attempts per minute per verified account, +before the ownership lookup. Open streams are separately capped at five per +account across all conversations, including pending ownership reads. Renewing a +token for the same account does not reset either budget. Excess attempts return +429. Idempotent cleanup releases capacity on disconnect, errors, expiry, +ownership failure, and deletion (including pending connections). Counters are +process-local; deployments with multiple replicas need shared enforcement. + ## Browser transport Native EventSource cannot attach Authorization headers. The chat adapter now @@ -39,9 +49,10 @@ listeners and `close()`. It parses UTF-8 SSE token/message frames across chunk and CRLF boundaries. It retries transient network disconnects after one second, with fresh account authorization, and stops on failed HTTP authorization or an invalid content type. There is no replay guarantee or Last-Event-ID contract. -Buffering has a defensive 1 MiB text limit; oversized responses are cancelled. +Buffering has a defensive 1 MiB text limit; oversized responses are cancelled +and stop permanently rather than entering a reconnect loop. -Call A1's `setAccountToken(token)` after the host application's existing login, +Call #191's `setAccountToken(token)` after the host application's existing login, and `setAccountToken(null)` on logout. Every account change closes existing streams immediately and remounts private chat state. Old frames are not dispatched after closure. Credentials stay in headers and memory, never URLs, diff --git a/backend/docs/conversation-ownership.md b/backend/docs/conversation-ownership.md index 0df16788..93d7e82f 100644 --- a/backend/docs/conversation-ownership.md +++ b/backend/docs/conversation-ownership.md @@ -23,7 +23,8 @@ The separate status of those mechanisms in CONFORMANCE.md remains accurate. ## Existing data and deployment -Back up Conversations and Messages before rollout. Existing rows without an +Back up Conversations and Messages before rollout. Legacy conversations were a +shared pool, not account-owned records. Existing rows without an owner remain stored but inaccessible; the application never grants them to the first caller or guesses an owner from `from`. An operator must establish the rightful existing account through independent evidence before making an @@ -40,16 +41,41 @@ their owned message/conversation reference no longer exists. The deployment needs DynamoDB GetItem/Scan/PutItem/UpdateItem/DeleteItem access to the configured conversation table and the existing message access, now -including Scan for attachment-reference lookup. Scans require no new index -or table. They prioritize correct isolation; an indexed attachment lookup is -a possible measured follow-up, not supplied by this change. - -**Roll out together with the authenticated-stream companion PR (A2).** A1 -closes REST and attachment access; A2 supplies account-authorized SSE. See -`authenticated-streams.md` for that boundary and lifecycle. Other open hardening work, including admin, -user/federation authorization and attachment write containment, remains +including Query for conversation-scoped attachment lookup and Scan for legacy +`/uploads/:name` compatibility. New downloads use +`/conversations/:id/attachments/:name`, checking ownership before querying only +that conversation partition, including continuation pages. No new index or +table is required. New files use UUID names, safe extensions and exclusive +writes; original names remain in message metadata. + +**Roll out #191 together with the authenticated-stream companion PR #192.** +#191 closes REST and attachment access; #192 supplies account-authorized SSE. +See `authenticated-streams.md` for that boundary and lifecycle. Other open hardening work, including admin, +user/federation authorization, remains separate and must be assessed before exposing public signup. +Operators can count ownerless conversations without changing records: + +```sh +aws dynamodb scan --table-name "$CONVERSATION_TABLE_NAME" --select COUNT \ + --filter-expression 'attribute_not_exists(ownerId)' --consistent-read +``` + +Use the deployment's table name and normal read credentials. Include every +page when counting through an SDK; a filtered page can be empty and still have +a continuation key. This count does not establish ownership or migrate data. + +## Request limits + +Authenticated conversation and message routers each allow 120 requests per +minute per account. Legacy uploads allow 60 requests per minute. Limits run +after account verification and before database work; a renewed token for the +same user does not reset the budget. IP subnet keys are the fallback when no +account identity is available. Rejected requests return 429 and Retry-After. +These budgets are configured at route registration through `chatRateLimit`. +The in-memory counters are local to one server process; multi-process deployments +need a shared express-rate-limit store to enforce a deployment-wide budget. + ## Browser integration The chat UI previously sent only a shared build-time API key and has no @@ -58,7 +84,10 @@ the existing backend `/api/auth/login` response and pass its token to the exported `setAccountToken(token)` in `frontend/chat-ui/src/api.js`. Call `setAccountToken(null)` on logout. Tokens stay in memory; the adapter uses Authorization headers, never query strings or browser storage. Account changes -clear chat state and discard in-flight responses from the old account. +clear chat state and discard in-flight responses from the old account. Setting +the same token again preserves state. Missing or rejected account credentials +show “Sign in to use chat”; a 401 clears the expired token. Trusted-origin +preflights permit Authorization, x-api-key and Idempotency-Key. Do not substitute an Amplify/Cognito token without an approved backend identity binding. PR #183 separately makes backend login/registration publicly diff --git a/backend/lib/conversationStreams.js b/backend/lib/conversationStreams.js index 8453fa90..cdbbe760 100644 --- a/backend/lib/conversationStreams.js +++ b/backend/lib/conversationStreams.js @@ -1,39 +1,70 @@ const streams = new Map(); +const accounts = new Map(); +const MAX_STREAMS_PER_ACCOUNT = 5; const MAX_TIMER_MS = 2 ** 31 - 1; -async function open(req, res) { - const expiresAt = req.user.exp * 1000; +function reserve(user) { + const expiresAt = user.exp * 1000; if (!Number.isFinite(expiresAt) || expiresAt <= Date.now()) { - return res.status(401).json({ error: 'Expiring account token required' }); + return { status: 401, error: 'Expiring account token required' }; } - const id = req.params.conversationId; + const count = accounts.get(user.userId) || 0; + if (count >= MAX_STREAMS_PER_ACCOUNT) { + return { status: 429, error: 'Too many open streams' }; + } + accounts.set(user.userId, count + 1); +} + +function register(id, ownerId, expiresAt, res) { if (!streams.has(id)) streams.set(id, new Set()); const clients = streams.get(id); let timer; - const client = { res, expiresAt, ready: false, revoked: false }; + const client = { res, expiresAt, ready: false, revoked: false, cleaned: false }; const cleanup = () => { - clearTimeout(timer); - clients.delete(client); - if (!clients.size && streams.get(id) === clients) streams.delete(id); + if (!client.cleaned) { + client.cleaned = true; + const remaining = accounts.get(ownerId) - 1; + if (remaining) accounts.set(ownerId, remaining); + else accounts.delete(ownerId); + clearTimeout(timer); + clients.delete(client); + if (!clients.size && streams.get(id) === clients) streams.delete(id); + } }; + client.cleanup = cleanup; const expire = () => { const remaining = expiresAt - Date.now(); - if (remaining <= 0) { cleanup(); res.end(); } + if (remaining <= 0) { + cleanup(); + if (client.ready) res.end(); + else res.status(401).json({ error: 'Expiring account token required' }); + } else timer = setTimeout(expire, Math.min(remaining, MAX_TIMER_MS)); }; res.on('close', cleanup); res.on('error', cleanup); clients.add(client); + expire(); + return client; +} + +async function open(req, res) { + const denial = reserve(req.user); + if (denial) return res.status(denial.status).json({ error: denial.error }); + const expiresAt = req.user.exp * 1000; + const ownerId = req.user.userId; + const id = req.params.conversationId; + const client = register(id, ownerId, expiresAt, res); + const cleanup = client.cleanup; // Register the pending connection BEFORE reading ownership, so a concurrent // committed delete can revoke it even if the database response is in flight. - let conversation; - try { conversation = await require('../models/conversation').getOwned(id, req.user.userId); } - catch (error) { cleanup(); throw error; } + const conversation = await readOwnership(client, id, ownerId); + if (res.destroyed || res.writableEnded) { cleanup(); return; } if (!conversation || client.revoked) { cleanup(); return res.status(404).json({ error: 'Not found' }); } - if (res.destroyed) { cleanup(); return; } + if (client.cleaned) return; if (expiresAt <= Date.now()) { cleanup(); return res.status(401).json({ error: 'Expiring account token required' }); @@ -46,7 +77,15 @@ async function open(req, res) { 'X-Accel-Buffering': 'no', }); res.flushHeaders(); - expire(); +} + +async function readOwnership(client, id, ownerId) { + try { return await require('../models/conversation').getOwned(id, ownerId); } + catch (error) { + const ended = client.cleaned; + client.cleanup(); + if (!ended) throw error; + } } function send(id, event, data) { @@ -69,6 +108,7 @@ function close(id) { if (!clients) return; for (const client of clients) { client.revoked = true; + client.cleanup(); if (client.ready) client.res.end(); } streams.delete(id); diff --git a/backend/middleware/chatRateLimit.js b/backend/middleware/chatRateLimit.js new file mode 100644 index 00000000..ca24c4cf --- /dev/null +++ b/backend/middleware/chatRateLimit.js @@ -0,0 +1,18 @@ +const { rateLimit, ipKeyGenerator } = require('express-rate-limit'); + +function accountKey(req) { + return req.user?.userId ? `account:${req.user.userId}` : `ip:${ipKeyGenerator(req.ip)}`; +} + +function chatRateLimit(limit) { + return rateLimit({ + windowMs: 60 * 1000, + limit, + keyGenerator: accountKey, + standardHeaders: 'draft-6', + legacyHeaders: false, + message: { error: 'Too many requests, please try again later' }, + }); +} + +module.exports = chatRateLimit; diff --git a/backend/middleware/conversationAttachment.js b/backend/middleware/conversationAttachment.js new file mode 100644 index 00000000..84ccf31c --- /dev/null +++ b/backend/middleware/conversationAttachment.js @@ -0,0 +1,21 @@ +const path = require('node:path'); +const Message = require('../models/message'); + +module.exports = async (req, res) => { + const { name } = req.params; + if (!name || name === '.' || name === '..' || /[/\\\0]/.test(name)) { + return res.status(404).json({ error: 'Not found' }); + } + if (!await Message.findConversationAttachment(req.params.id, '/uploads/' + name)) { + return res.status(404).json({ error: 'Not found' }); + } + res.sendFile(name, { + root: path.join(__dirname, '../uploads'), dotfiles: 'deny', + headers: { + 'Cache-Control': 'private, no-store', + 'Content-Disposition': 'attachment', + 'Content-Type': 'application/octet-stream', + 'X-Content-Type-Options': 'nosniff', + }, + }); +}; diff --git a/backend/models/message.js b/backend/models/message.js index f37aafa5..1d68c40d 100644 --- a/backend/models/message.js +++ b/backend/models/message.js @@ -26,9 +26,10 @@ async function sendMessage( }; if (file && file.data) { - const filename = Date.now() + '_' + file.name; + const extension = path.extname(file.name || ''); + const filename = randomUUID() + (/^\.[a-zA-Z0-9]{1,10}$/.test(extension) ? extension : ''); const filePath = path.join(UPLOAD_DIR, filename); - fs.writeFileSync(filePath, Buffer.from(file.data, 'base64')); + fs.writeFileSync(filePath, Buffer.from(file.data, 'base64'), { flag: 'wx' }); msg.file = { path: '/uploads/' + filename, originalname: file.name, @@ -51,6 +52,13 @@ async function listMessages(conversationId) { return res.Items || []; } +async function getMessage(conversationId, id) { + const { Item } = await docClient.get({ + TableName: MESSAGE_TABLE_NAME, Key: { conversationId, id }, ConsistentRead: true, + }).promise(); + return Item || null; +} + async function removeByConversation(conversationId) { const messages = await listMessages(conversationId); if (!messages.length) return; @@ -83,4 +91,22 @@ async function findByAttachmentPath(filePath) { return items; } -module.exports = { sendMessage, listMessages, removeByConversation, findByAttachmentPath, setDocClient }; +async function findConversationAttachment(conversationId, filePath) { + let cursor; + do { + const page = await docClient.query({ + TableName: MESSAGE_TABLE_NAME, + KeyConditionExpression: 'conversationId = :cid', + FilterExpression: '#file.#path = :filePath', + ExpressionAttributeNames: { '#file': 'file', '#path': 'path' }, + ExpressionAttributeValues: { ':cid': conversationId, ':filePath': filePath }, + Limit: 100, + ...(cursor && { ExclusiveStartKey: cursor }), + }).promise(); + if (page.Items?.some(message => message.file?.path === filePath)) return true; + cursor = page.LastEvaluatedKey; + } while (cursor); + return false; +} + +module.exports = { sendMessage, listMessages, getMessage, removeByConversation, findByAttachmentPath, findConversationAttachment, setDocClient }; diff --git a/backend/package-lock.json b/backend/package-lock.json index f066c735..71cdb57a 100644 --- a/backend/package-lock.json +++ b/backend/package-lock.json @@ -17,6 +17,7 @@ "cors": "^2.8.5", "dotenv": "^16.4.7", "express": "^4.21.2", + "express-rate-limit": "^8.7.2", "franc": "^6.1.0", "jsonwebtoken": "^9.0.2", "openai": "^4.76.0", @@ -3662,6 +3663,48 @@ "url": "https://opencollective.com/express" } }, + "node_modules/express-rate-limit": { + "version": "8.7.2", + "resolved": "https://registry.npmjs.org/express-rate-limit/-/express-rate-limit-8.7.2.tgz", + "integrity": "sha512-U1ScocxDD9WI09LaBVypvNkOtNR5yWAmZLkuKJdAzIchA4YKFYF6GT5IKER7epG58oXtjvGlV0/6ii0F83EyZg==", + "license": "MIT", + "dependencies": { + "debug": "^4.4.3", + "ip-address": "^10.2.0" + }, + "engines": { + "node": ">= 16" + }, + "funding": { + "url": "https://github.com/sponsors/express-rate-limit" + }, + "peerDependencies": { + "express": ">= 4.11" + } + }, + "node_modules/express-rate-limit/node_modules/debug": { + "version": "4.4.3", + "resolved": "https://registry.npmjs.org/debug/-/debug-4.4.3.tgz", + "integrity": "sha512-RGwwWnwQvkVfavKVt22FGLw+xYSdzARwm0ru6DhTVA3umU5hZc28V3kO4stgYryrTlLpuvgI9GiijltAjNbcqA==", + "license": "MIT", + "dependencies": { + "ms": "^2.1.3" + }, + "engines": { + "node": ">=6.0" + }, + "peerDependenciesMeta": { + "supports-color": { + "optional": true + } + } + }, + "node_modules/express-rate-limit/node_modules/ms": { + "version": "2.1.3", + "resolved": "https://registry.npmjs.org/ms/-/ms-2.1.3.tgz", + "integrity": "sha512-6FlzubTLZG3J2a/NVCAleEhjzq5oxgHyaCU9yYXvcLsvoVaHJq/s5xXI6/XXP6tz7R9xAOtHnSO/tXtF3WRTlA==", + "license": "MIT" + }, "node_modules/fast-json-stable-stringify": { "version": "2.1.0", "resolved": "https://registry.npmjs.org/fast-json-stable-stringify/-/fast-json-stable-stringify-2.1.0.tgz", @@ -4211,6 +4254,15 @@ "integrity": "sha512-6FlzubTLZG3J2a/NVCAleEhjzq5oxgHyaCU9yYXvcLsvoVaHJq/s5xXI6/XXP6tz7R9xAOtHnSO/tXtF3WRTlA==", "license": "MIT" }, + "node_modules/ip-address": { + "version": "10.7.5", + "resolved": "https://registry.npmjs.org/ip-address/-/ip-address-10.7.5.tgz", + "integrity": "sha512-t2v0rkILYv9WXmSL7TpAaU3HXgaFYOPUcH3QHgwQ++gpJuH3bqyMy4Fvt/DJCTKwzH3Cf+bAp5W/mFL39KPLzA==", + "license": "MIT", + "engines": { + "node": ">= 12" + } + }, "node_modules/ipaddr.js": { "version": "1.9.1", "resolved": "https://registry.npmjs.org/ipaddr.js/-/ipaddr.js-1.9.1.tgz", diff --git a/backend/package.json b/backend/package.json index 0e1551e8..ba7c7356 100644 --- a/backend/package.json +++ b/backend/package.json @@ -16,6 +16,7 @@ "cors": "^2.8.5", "dotenv": "^16.4.7", "express": "^4.21.2", + "express-rate-limit": "^8.7.2", "franc": "^6.1.0", "jsonwebtoken": "^9.0.2", "openai": "^4.76.0", diff --git a/backend/routes/communicationRoutes.js b/backend/routes/communicationRoutes.js index 675bce23..c3c5128a 100644 --- a/backend/routes/communicationRoutes.js +++ b/backend/routes/communicationRoutes.js @@ -5,7 +5,7 @@ const auth = require('../middleware/chatAuth'); const owner = require('../middleware/conversationOwner'); const asyncHandler = require('../utils/asyncHandler'); -router.use(auth); +router.use(auth, require('../middleware/chatRateLimit')(120)); router.post('/:conversationId', asyncHandler(owner), asyncHandler(commController.sendMessage)); router.get('/:conversationId', asyncHandler(owner), asyncHandler(commController.listMessages)); diff --git a/backend/routes/conversationRoutes.js b/backend/routes/conversationRoutes.js index d2880d73..07cf673a 100644 --- a/backend/routes/conversationRoutes.js +++ b/backend/routes/conversationRoutes.js @@ -5,9 +5,10 @@ const owner = require('../middleware/conversationOwner'); const ctrl = require('../controllers/conversation_controller'); const asyncHandler = require('../utils/asyncHandler'); -router.use(auth); +router.use(auth, require('../middleware/chatRateLimit')(120)); router.post('/', asyncHandler(ctrl.create)); router.get('/', asyncHandler(ctrl.list)); +router.get('/:id/attachments/:name', asyncHandler(owner), asyncHandler(require('../middleware/conversationAttachment'))); router.get('/:id', asyncHandler(owner), ctrl.get); router.put('/:id', asyncHandler(owner), asyncHandler(ctrl.rename)); router.delete('/:id', asyncHandler(owner), asyncHandler(ctrl.remove)); diff --git a/backend/scripts/check-conversation-ownership.js b/backend/scripts/check-conversation-ownership.js index 61ba60b9..cd1cedfe 100644 --- a/backend/scripts/check-conversation-ownership.js +++ b/backend/scripts/check-conversation-ownership.js @@ -44,6 +44,8 @@ async function main() { assert.equal((await Message.listMessages(mine.id))[0].id, message.id); await db.put({ TableName: process.env.MESSAGE_TABLE_NAME, Item: { id: 'attachment', conversationId: mine.id, file: { path: '/uploads/example.txt' } } }).promise(); assert.equal((await Message.findByAttachmentPath('/uploads/example.txt'))[0].id, 'attachment'); + assert.equal(await Message.findConversationAttachment(mine.id, '/uploads/example.txt'), true); + assert.equal(await Message.findConversationAttachment('foreign', '/uploads/example.txt'), false); assert.equal(await Conversation.remove(mine.id, 'alice'), true); assert.deepEqual(await Message.listMessages(mine.id), []); assert.equal(await Conversation.getOwned(mine.id, 'alice'), null); diff --git a/backend/server.js b/backend/server.js index 953fee8b..83e805d7 100644 --- a/backend/server.js +++ b/backend/server.js @@ -36,7 +36,7 @@ app.use(cors({ }, credentials: true, methods: ['GET', 'POST', 'PUT', 'DELETE', 'OPTIONS'], - allowedHeaders: ['Content-Type', 'Authorization'] + allowedHeaders: ['Content-Type', 'Authorization', 'Idempotency-Key', 'x-api-key'] })); // ----------------------------------------- @@ -64,10 +64,10 @@ const server = http.createServer(app); // Account-scoped streams share the same owner check as message reads/sends. const conversationStreams = require('./lib/conversationStreams'); -app.get('/events', chatAuth, (_req, res) => { +app.get('/events', (_req, res) => { res.status(410).json({ error: 'Global event feed retired; use an owned conversation stream' }); }); -app.get('/stream/:conversationId', chatAuth, asyncHandler(conversationStreams.open)); +app.get('/stream/:conversationId', chatAuth, require('./middleware/chatRateLimit')(60), asyncHandler(conversationStreams.open)); function broadcast(event, data, conversationId) { if (conversationId) conversationStreams.send(conversationId, event, data); @@ -80,7 +80,7 @@ global.emitMessage = (conversationId, message) => conversationStreams.send(conve // Middleware app.use(authMiddleware); -app.use('/uploads', chatAuth, asyncHandler(require('./middleware/attachmentOwner')), express.static(path.join(__dirname, 'uploads'), { +app.use('/uploads', chatAuth, require('./middleware/chatRateLimit')(60), asyncHandler(require('./middleware/attachmentOwner')), express.static(path.join(__dirname, 'uploads'), { dotfiles: 'deny', index: false, setHeaders(res) { res.setHeader('Content-Disposition', 'attachment'); diff --git a/frontend/chat-ui/accountClient.test.js b/frontend/chat-ui/accountClient.test.js index 59197fd0..d5f3569d 100644 --- a/frontend/chat-ui/accountClient.test.js +++ b/frontend/chat-ui/accountClient.test.js @@ -46,3 +46,39 @@ test('attachment downloads use the same account and reject credential destinatio client.setToken(null); await assert.rejects(client.attachment('/uploads/file.txt'), /Sign in/); }); + +test('setting the same token preserves pending requests and account state', async () => { + const { createAccountClient } = await import('./src/accountClient.mjs'); + let finish; + const client = createAccountClient({ fetchImpl: () => new Promise(resolve => { finish = resolve; }) }); + client.setToken('alice'); + let changes = 0; + client.changes.addEventListener('change', () => changes++); + const pending = client.request('GET', '/conversations'); + client.setToken('alice'); + finish(new Response('[]')); + assert.deepEqual(await pending, []); + assert.equal(changes, 0); +}); + +test('raw attachment names encode URL punctuation without decoding percent sequences', async () => { + const { createAccountClient } = await import('./src/accountClient.mjs'); + const urls = []; + const client = createAccountClient({ fetchImpl: async url => { urls.push(url); return new Response('file'); } }); + client.setToken('alice'); + for (const name of ['100%.txt', 'a#b?.txt', '%2e%2e.txt']) await client.attachment('/uploads/' + name, 'mine'); + assert.deepEqual(urls, ['/conversations/mine/attachments/100%25.txt', '/conversations/mine/attachments/a%23b%3F.txt', '/conversations/mine/attachments/%252e%252e.txt']); + for (const name of ['..', '..\\x', '../x', '\0']) await assert.rejects(client.attachment('/uploads/' + name, 'mine'), /Invalid attachment path/); + await assert.rejects(client.request('GET', '//evil.example/x'), /Invalid API path/); +}); + +test('401 clears expired account credentials and notifies the UI', async () => { + const { createAccountClient } = await import('./src/accountClient.mjs'); + const client = createAccountClient({ fetchImpl: async () => new Response('{"error":"expired"}', { status: 401 }) }); + client.setToken('expired'); + let changes = 0; + client.changes.addEventListener('change', () => changes++); + await assert.rejects(client.request('GET', '/conversations'), error => error.status === 401); + assert.equal(client.getToken(), null); + assert.equal(changes, 1); +}); diff --git a/frontend/chat-ui/authenticatedStream.test.js b/frontend/chat-ui/authenticatedStream.test.js index 490dfbb7..9cfcbff0 100644 --- a/frontend/chat-ui/authenticatedStream.test.js +++ b/frontend/chat-ui/authenticatedStream.test.js @@ -61,3 +61,21 @@ test('authorization failures stop reconnecting and no token means no request', a assert.equal(denied.readyState, 2); assert.equal(requests, 1); }); + +test('oversized frames close permanently instead of repeatedly reconnecting', async () => { + const { createAccountClient } = await import('./src/accountClient.mjs'); + const { authenticatedStream } = await import('./src/authenticatedStream.mjs'); + const client = createAccountClient(); + client.setToken('alice'); + let requests = 0; + const stream = authenticatedStream(client, '/stream/mine', { retryMs: 1, fetchImpl: async () => { + requests++; + return new Response('data: ' + 'x'.repeat(1024 * 1024), { headers: { 'Content-Type': 'text/event-stream' } }); + } }); + try { + await once(stream, 'error'); + await new Promise(resolve => setTimeout(resolve, 20)); + assert.equal(stream.readyState, 2); + assert.equal(requests, 1); + } finally { stream.close(); } +}); diff --git a/frontend/chat-ui/signIn.test.js b/frontend/chat-ui/signIn.test.js new file mode 100644 index 00000000..2b7a8b19 --- /dev/null +++ b/frontend/chat-ui/signIn.test.js @@ -0,0 +1,22 @@ +const { test } = require('node:test'); +const assert = require('node:assert/strict'); +const React = require('react'); +const { renderToString } = require('react-dom/server'); + +test('sidebar and chat show sign-in guidance until an account token is supplied', async () => { + const { createServer } = await import('vite'); + const server = await createServer({ server: { middlewareMode: true, hmr: false, ws: false }, optimizeDeps: { noDiscovery: true, include: [] }, appType: 'custom' }); + try { + const { StoreProvider } = await server.ssrLoadModule('/src/store.jsx'); + const { accountClient } = await server.ssrLoadModule('/src/api.js'); + for (const file of ['Sidebar', 'ChatWindow']) { + const { default: Component } = await server.ssrLoadModule(`/src/components/${file}.jsx`); + const render = () => renderToString(React.createElement(StoreProvider, null, React.createElement(Component))); + accountClient.setToken(null); + assert.match(render(), /Sign in to use chat/); + accountClient.setToken('account-token'); + assert.doesNotMatch(render(), /Sign in to use chat/); + } + accountClient.setToken(null); + } finally { await server.close(); } +}); diff --git a/frontend/chat-ui/src/accountClient.mjs b/frontend/chat-ui/src/accountClient.mjs index 0c638415..d4d20435 100644 --- a/frontend/chat-ui/src/accountClient.mjs +++ b/frontend/chat-ui/src/accountClient.mjs @@ -1,22 +1,56 @@ +function normalizeToken(value) { + return typeof value === 'string' && value ? value : null; +} + +function apiUrl(baseUrl, path) { + if (!path.startsWith('/') || path.startsWith('//')) throw new Error('Invalid API path'); + return `${baseUrl.replace(/\/$/, '')}${path}`; +} + +function requestHeaders(token, apiKey) { + const result = {}; + if (token) result.Authorization = `Bearer ${token}`; + if (apiKey) result['x-api-key'] = apiKey; + return result; +} + +function requestError(response, data) { + const error = new Error(data?.error || data?.message || `Request failed with status ${response.status}`); + error.status = response.status; + error.data = data; + return error; +} + +async function responseData(response, binary) { + if (binary && response.ok) return response.blob(); + const text = await response.text(); + try { return text ? JSON.parse(text) : null; } catch { return text; } +} + +function attachmentDestination(path, conversationId) { + const prefix = '/uploads/'; + if (!path.startsWith(prefix)) throw new Error('Invalid attachment path'); + const name = path.slice(prefix.length); + if (!name || name === '.' || name === '..' || /[/\\\0]/.test(name)) throw new Error('Invalid attachment path'); + return conversationId + ? `/conversations/${encodeURIComponent(conversationId)}/attachments/${encodeURIComponent(name)}` + : prefix + encodeURIComponent(name); +} + export function createAccountClient({ baseUrl = '', apiKey = '', fetchImpl = (...args) => fetch(...args) } = {}) { let token = null; let version = 0; const changes = new EventTarget(); function setToken(value) { - token = typeof value === 'string' && value ? value : null; - version++; - changes.dispatchEvent(new Event('change')); - } - function url(path) { - if (!path.startsWith('/') || path.startsWith('//')) throw new Error('Invalid API path'); - return `${baseUrl.replace(/\/$/, '')}${path}`; - } - function headers() { - const result = {}; - if (token) result.Authorization = `Bearer ${token}`; - if (apiKey) result['x-api-key'] = apiKey; - return result; + const next = normalizeToken(value); + if (next !== token) { + token = next; + version++; + changes.dispatchEvent(new Event('change')); + } } + const url = path => apiUrl(baseUrl, path); + const headers = () => requestHeaders(token, apiKey); async function read(method, path, body, binary = false) { const started = version; const options = { method, headers: headers(), redirect: 'error' }; @@ -25,29 +59,18 @@ export function createAccountClient({ baseUrl = '', apiKey = '', fetchImpl = (.. options.body = JSON.stringify(body); } const response = await fetchImpl(url(path), options); - let data; - if (binary && response.ok) data = await response.blob(); - else { - const text = await response.text(); - try { data = text ? JSON.parse(text) : null; } catch { data = text; } - } + const data = await responseData(response, binary); if (version !== started) throw new Error('Account changed during request'); if (!response.ok) { - const error = new Error(data?.error || data?.message || `Request failed with status ${response.status}`); - error.status = response.status; - error.data = data; - throw error; + if (response.status === 401) setToken(null); + throw requestError(response, data); } return data; } - async function attachment(path) { + async function attachment(path, conversationId) { if (!token) throw new Error('Sign in to download attachments'); - // Never forward a bearer to a path supplied by another origin or endpoint. - if (!/^\/uploads\/[^/?#\\]+$/.test(path)) throw new Error('Invalid attachment path'); - let name; - try { name = decodeURIComponent(path.slice('/uploads/'.length)); } catch { /* invalid encoding */ } - if (!name || name === '.' || name === '..' || /[/\\\0]/.test(name)) throw new Error('Invalid attachment path'); - return read('GET', path, undefined, true); + return read('GET', attachmentDestination(path, conversationId), undefined, true); } + return { setToken, getToken: () => token, changes, url, headers, request: read, attachment }; } diff --git a/frontend/chat-ui/src/api.js b/frontend/chat-ui/src/api.js index 6a0bca2c..40c83cd3 100644 --- a/frontend/chat-ui/src/api.js +++ b/frontend/chat-ui/src/api.js @@ -10,8 +10,8 @@ async function request(method, path, body) { return accountClient.request(method, path.startsWith('/') ? path : `/${path}`, body); } -export async function downloadAttachment(file) { - const blob = await accountClient.attachment(file.path); +export async function downloadAttachment(file, conversationId) { + const blob = await accountClient.attachment(file.path, conversationId); const objectUrl = URL.createObjectURL(blob); const link = document.createElement('a'); link.href = objectUrl; diff --git a/frontend/chat-ui/src/authenticatedStream.mjs b/frontend/chat-ui/src/authenticatedStream.mjs index 187addf9..b5a29f4d 100644 --- a/frontend/chat-ui/src/authenticatedStream.mjs +++ b/frontend/chat-ui/src/authenticatedStream.mjs @@ -1,78 +1,61 @@ +import { readFrames } from './sseFrames.mjs'; + // EventSource-shaped subset used by ChatWindow, with bearer headers and an // abortable lifetime. No credential is ever appended to a stream URL. export function authenticatedStream(client, path, { fetchImpl = (...args) => fetch(...args), retryMs = 1000 } = {}) { const events = new EventTarget(); const controller = new AbortController(); - let reader; - let retry; - let closed = false; + const state = { reader: null, retry: null, closed: false }; events.readyState = 0; events.close = () => { - closed = true; + state.closed = true; events.readyState = 2; - clearTimeout(retry); + clearTimeout(state.retry); controller.abort(); - if (reader) void reader.cancel().catch(() => {}); + if (state.reader) void state.reader.cancel().catch(() => {}); client.changes.removeEventListener('change', events.close); }; client.changes.addEventListener('change', events.close); - async function connect() { - if (closed) return; - try { - if (!client.getToken()) { - events.close(); - events.dispatchEvent(new Event('error')); - return; - } - const response = await fetchImpl(client.url(path), { - headers: { ...client.headers(), Accept: 'text/event-stream' }, - signal: controller.signal, - redirect: 'error', - }); - if (closed) { await response.body?.cancel(); return; } - if (!response.ok || !response.headers.get('content-type')?.startsWith('text/event-stream')) { - await response.body?.cancel(); - events.close(); - events.dispatchEvent(new Event('error')); - return; - } - reader = response.body.getReader(); - events.readyState = 1; - events.dispatchEvent(new Event('open')); - const decoder = new TextDecoder(); - let buffer = ''; - let type = 'message'; - let data = []; - let frameSize = 0; - while (!closed) { - const result = await reader.read(); - if (closed || result.done) break; - buffer += decoder.decode(result.value, { stream: true }); - if (buffer.length + frameSize > 1024 * 1024) throw new Error('Stream frame too large'); - while (!closed) { - const end = buffer.search(/[\r\n]/); - if (end < 0 || (buffer[end] === '\r' && end === buffer.length - 1)) break; - const line = buffer.slice(0, end); - const width = buffer[end] === '\r' && buffer[end + 1] === '\n' ? 2 : 1; - buffer = buffer.slice(end + width); - if (!line) { - if (data.length) events.dispatchEvent(new MessageEvent(type, { data: data.join('\n') })); - type = 'message'; data = []; frameSize = 0; - } else if (line.startsWith('event:')) type = line.slice(6).replace(/^ /, '') || 'message'; - else if (line.startsWith('data:')) { data.push(line.slice(5).replace(/^ /, '')); frameSize += line.length; } - } - } - } catch { - if (!closed) events.dispatchEvent(new Event('error')); - } finally { - if (reader) { await reader.cancel().catch(() => {}); reader = null; } - if (!closed) { - events.readyState = 0; - retry = setTimeout(connect, retryMs); - } + const connection = { client, path, fetchImpl, retryMs, events, controller, state }; + queueMicrotask(() => connect(connection)); + return events; +} + +async function connect(connection) { + const { client, path, fetchImpl, retryMs, events, controller, state } = connection; + if (state.closed) return; + try { + if (!client.getToken()) { + events.close(); + events.dispatchEvent(new Event('error')); + return; + } + const response = await fetchImpl(client.url(path), { + headers: { ...client.headers(), Accept: 'text/event-stream' }, + signal: controller.signal, + redirect: 'error', + }); + if (state.closed) { await response.body?.cancel(); return; } + if (!response.ok || !response.headers.get('content-type')?.startsWith('text/event-stream')) { + await response.body?.cancel(); + if (response.status === 401) client.setToken(null); + events.close(); + events.dispatchEvent(new Event('error')); + return; + } + state.reader = response.body.getReader(); + events.readyState = 1; + events.dispatchEvent(new Event('open')); + await readFrames(state.reader, events, () => state.closed); + } catch (error) { + if (error instanceof RangeError) { events.close(); events.dispatchEvent(new Event('error')); } + else if (!state.closed) events.dispatchEvent(new Event('error')); + } finally { + if (state.reader) { await state.reader.cancel().catch(() => {}); state.reader = null; } + if (!state.closed) { + events.readyState = 0; + state.retry = setTimeout(() => connect(connection), retryMs); } } - queueMicrotask(connect); - return events; } diff --git a/frontend/chat-ui/src/components/ChatWindow.jsx b/frontend/chat-ui/src/components/ChatWindow.jsx index 2e3c5429..18b10f29 100644 --- a/frontend/chat-ui/src/components/ChatWindow.jsx +++ b/frontend/chat-ui/src/components/ChatWindow.jsx @@ -11,7 +11,7 @@ import React, { useEffect, useRef } from 'react'; import { useStore } from '../store'; import MessageBubble from './MessageBubble.jsx'; -import { stream } from '../api'; +import { accountClient, stream } from '../api'; import ThinkingTicker from './ThinkingTicker.jsx'; export default function ChatWindow() { @@ -29,7 +29,7 @@ export default function ChatWindow() { }, [state.currentConversation, dispatch]); useEffect(() => { - if (!state.currentConversation) return; + if (!accountClient.getToken() || !state.currentConversation) return; const events = stream(`/stream/${state.currentConversation.id}`); const handleToken = (e) => { @@ -63,6 +63,10 @@ export default function ChatWindow() { }; }, [state.currentConversation, dispatch]); + if (!accountClient.getToken()) { + return
Sign in to use chat.
; + } + return (
{state.messages.map((m, index) => ( diff --git a/frontend/chat-ui/src/components/MessageBubble.jsx b/frontend/chat-ui/src/components/MessageBubble.jsx index 8d46792f..73139cd1 100644 --- a/frontend/chat-ui/src/components/MessageBubble.jsx +++ b/frontend/chat-ui/src/components/MessageBubble.jsx @@ -12,7 +12,7 @@ export default function MessageBubble({ message, index = 0 }) { const [downloadError, setDownloadError] = useState(''); const download = async () => { setDownloadError(''); - try { await downloadAttachment(message.file); } + try { await downloadAttachment(message.file, message.conversationId); } catch (error) { setDownloadError(error.message); } }; const isUser = message.role === 'user' || message.sender === 'user' || message.from === 'user'; diff --git a/frontend/chat-ui/src/components/Sidebar.jsx b/frontend/chat-ui/src/components/Sidebar.jsx index 34b32f10..313c65d8 100644 --- a/frontend/chat-ui/src/components/Sidebar.jsx +++ b/frontend/chat-ui/src/components/Sidebar.jsx @@ -16,7 +16,7 @@ import { useStore } from '../store'; * @typedef {import('../types').Message} Message */ -import { get, post } from '../api'; +import { accountClient, get, post } from '../api'; /** * Sidebar containing conversation list and controls. @@ -27,6 +27,7 @@ export default function Sidebar() { const [loading, setLoading] = useState(true); useEffect(() => { + if (!accountClient.getToken()) { setLoading(false); return; } const load = async () => { try { const data = await get('/conversations'); @@ -90,6 +91,10 @@ export default function Sidebar() { /** @type {Conversation[]} */ const others = state.conversations.filter((c) => !c.pinned); + if (!accountClient.getToken()) { + return ; + } + return (