diff --git a/CONFORMANCE.md b/CONFORMANCE.md index a9843d48..e514de6d 100644 --- a/CONFORMANCE.md +++ b/CONFORMANCE.md @@ -44,6 +44,22 @@ Adopts the covenant scale (LBTAS, v2 floor) and carries its own record instance. 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 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. + ### Browser garden planner The Next.js `/garden-planner` page is a member-local planning aid, not a ledger 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__/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__/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..23edc18b --- /dev/null +++ b/backend/__tests__/conversationAccess.test.js @@ -0,0 +1,195 @@ +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'); + }); + 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 new file mode 100644 index 00000000..d0c0749b --- /dev/null +++ b/backend/__tests__/conversationStreams.test.js @@ -0,0 +1,156 @@ +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(4000)]) }); + } + 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 ${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); + 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'); + }); + 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/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/authenticated-streams.md b/backend/docs/authenticated-streams.md new file mode 100644 index 00000000..77380585 --- /dev/null +++ b/backend/docs/authenticated-streams.md @@ -0,0 +1,75 @@ +# Authenticated conversation streams + +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 +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: 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 +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. + +## 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 +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 +and stop permanently rather than entering a reconnect loop. + +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, +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 new file mode 100644 index 00000000..93d7e82f --- /dev/null +++ b/backend/docs/conversation-ownership.md @@ -0,0 +1,119 @@ +# 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. 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 +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 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 +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. 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 +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/lib/conversationStreams.js b/backend/lib/conversationStreams.js new file mode 100644 index 00000000..cdbbe760 --- /dev/null +++ b/backend/lib/conversationStreams.js @@ -0,0 +1,117 @@ +const streams = new Map(); +const accounts = new Map(); +const MAX_STREAMS_PER_ACCOUNT = 5; +const MAX_TIMER_MS = 2 ** 31 - 1; + +function reserve(user) { + const expiresAt = user.exp * 1000; + if (!Number.isFinite(expiresAt) || expiresAt <= Date.now()) { + return { status: 401, error: 'Expiring account token required' }; + } + 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, cleaned: false }; + const cleanup = () => { + 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(); + 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. + 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 (client.cleaned) 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(); +} + +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) { + 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; + client.cleanup(); + if (client.ready) client.res.end(); + } + streams.delete(id); +} + +module.exports = { open, send, close }; 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/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/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..02cd2482 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,30 @@ 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; + } + // Closing streams must not depend on later narrative cleanup succeeding. + require('../lib/conversationStreams').close(id); 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..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; @@ -66,4 +74,39 @@ 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; +} + +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 3a0367c2..c3c5128a 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.use(auth, require('../middleware/chatRateLimit')(120)); +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..07cf673a 100644 --- a/backend/routes/conversationRoutes.js +++ b/backend/routes/conversationRoutes.js @@ -1,14 +1,17 @@ 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.use(auth, require('../middleware/chatRateLimit')(120)); 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/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)); +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..cd1cedfe --- /dev/null +++ b/backend/scripts/check-conversation-ownership.js @@ -0,0 +1,58 @@ +// 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 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); + 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..83e805d7 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(); @@ -34,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'] })); // ----------------------------------------- @@ -60,77 +62,32 @@ 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)); +// Account-scoped streams share the same owner check as message reads/sends. +const conversationStreams = require('./lib/conversationStreams'); +app.get('/events', (_req, res) => { + res.status(410).json({ error: 'Global event feed retired; use an owned conversation stream' }); }); - -// 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); - } - }); -}); - -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, require('./middleware/chatRateLimit')(60), 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); -app.use('/uploads', 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'); + res.setHeader('Content-Type', 'application/octet-stream'); + res.setHeader('X-Content-Type-Options', 'nosniff'); + }, +})); // Routes let runFederationSync; 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/accountClient.test.js b/frontend/chat-ui/accountClient.test.js new file mode 100644 index 00000000..d5f3569d --- /dev/null +++ b/frontend/chat-ui/accountClient.test.js @@ -0,0 +1,84 @@ +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/); +}); + +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 new file mode 100644 index 00000000..9cfcbff0 --- /dev/null +++ b/frontend/chat-ui/authenticatedStream.test.js @@ -0,0 +1,81 @@ +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); +}); + +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/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/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 new file mode 100644 index 00000000..d4d20435 --- /dev/null +++ b/frontend/chat-ui/src/accountClient.mjs @@ -0,0 +1,76 @@ +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) { + 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' }; + if (body !== undefined) { + options.headers['Content-Type'] = 'application/json'; + options.body = JSON.stringify(body); + } + const response = await fetchImpl(url(path), options); + const data = await responseData(response, binary); + if (version !== started) throw new Error('Account changed during request'); + if (!response.ok) { + if (response.status === 401) setToken(null); + throw requestError(response, data); + } + return data; + } + async function attachment(path, conversationId) { + if (!token) throw new Error('Sign in to download attachments'); + 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 76673fd0..40c83cd3 100644 --- a/frontend/chat-ui/src/api.js +++ b/frontend/chat-ui/src/api.js @@ -1,39 +1,23 @@ +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 || ''; - -function buildUrl(path) { - const clean = path.startsWith('/') ? path : `/${path}`; - return API_BASE_URL ? `${API_BASE_URL.replace(/\/$/, '')}${clean}` : clean; -} +export const accountClient = createAccountClient({ baseUrl: API_BASE_URL, apiKey: API_KEY }); +export const setAccountToken = accountClient.setToken; 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, conversationId) { + const blob = await accountClient.attachment(file.path, conversationId); + 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); @@ -41,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..b5a29f4d --- /dev/null +++ b/frontend/chat-ui/src/authenticatedStream.mjs @@ -0,0 +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(); + const state = { reader: null, retry: null, closed: false }; + events.readyState = 0; + events.close = () => { + state.closed = true; + events.readyState = 2; + clearTimeout(state.retry); + controller.abort(); + if (state.reader) void state.reader.cancel().catch(() => {}); + client.changes.removeEventListener('change', events.close); + }; + client.changes.addEventListener('change', events.close); + + 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); + } + } +} 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/ChatWindow.jsx b/frontend/chat-ui/src/components/ChatWindow.jsx index 81efe4c1..18b10f29 100644 --- a/frontend/chat-ui/src/components/ChatWindow.jsx +++ b/frontend/chat-ui/src/components/ChatWindow.jsx @@ -5,13 +5,13 @@ * 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'; 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 5a629303..73139cd1 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, message.conversationId); } + 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/components/Sidebar.jsx b/frontend/chat-ui/src/components/Sidebar.jsx index 3c7488c3..313c65d8 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'; @@ -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 (