Authenticate conversation streams and retire the global feed

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

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
 ```

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);
+});

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);
+  });
+});

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);
 });

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();
+  });
+
+});