diff --git a/src/js/channels.js b/src/js/channels.js index 0edb585..e6b823b 100644 --- a/src/js/channels.js +++ b/src/js/channels.js @@ -2252,20 +2252,23 @@ class ChannelManager { */ async startEpochKeys(channel) { if (!channel?.gate?.address) return; + let sweepDeferred = false; try { - await this._setupEpochKeys(channel); + sweepDeferred = await this._setupEpochKeys(channel); channel._epochSetupRetry = 0; } catch (e) { Logger.warn('Epoch key setup failed (messages will wait for key):', e.message); this._scheduleEpochSetupRetry(channel); } - this._rotateForLostAccess(channel).catch(() => {}); + if (!sweepDeferred) this._rotateForLostAccess(channel).catch(() => {}); } /** * Wire a gated channel into the epoch-key protocol: live -4 * subscription, refresh-on-adopt listener, and initial key state * (bootstrap as admin, or request as member). Idempotent per channel. + * @returns {Promise} true when the -4 history is read in the + * background, and the owner's lost-access sweep runs after it */ async _setupEpochKeys(channel) { const keysStreamId = channel.keysStreamId || deriveKeysId(channel.messageStreamId); @@ -2301,9 +2304,11 @@ class ChannelManager { // cold path. Logger.warn('Background epoch reconcile failed (will retry):', e.message); this._scheduleEpochSetupRetry(channel); - }); + }) + // Who holds the key in force comes from this -4 read. + .finally(() => this._rotateForLostAccess(channel).catch(() => {})); }, 8_000); - return; + return true; } await epochKeyManager.ensureChannelKeys(channel); @@ -2360,10 +2365,16 @@ class ChannelManager { (channel.rotatedForNoAccess || channel.rotatedForBanned || []) .map(lower).filter(a => !withAccess.has(a))); + // Whoever holds the key in force and lost access rotates even when + // covered: a moderator may have re-admitted and removed them between + // two owner opens, with no sweep seeing the regain. + const holders = new Set(epochKeyManager.getCurrentKeyHolders(channel.messageStreamId)); + const pending = [...new Set([ - ...[...noAccessNow].filter(a => previously.has(a)), - ...bannedNow - ])].filter(a => !covered.has(a)); + ...[...noAccessNow].filter(a => previously.has(a) && !covered.has(a)), + ...bannedNow.filter(a => !covered.has(a)), + ...[...noAccessNow].filter(a => holders.has(a)) + ])]; try { if (pending.length > 0) { diff --git a/src/js/channels/Membership.js b/src/js/channels/Membership.js index 9a2e283..77abb61 100644 --- a/src/js/channels/Membership.js +++ b/src/js/channels/Membership.js @@ -43,7 +43,8 @@ export class Membership { const normalizedAddress = address.toLowerCase(); - if (channel.members.map(m => m.toLowerCase()).includes(normalizedAddress)) { + if (channel.members.some(m => m.toLowerCase() === normalizedAddress) + && (await this._stillAllowed(channel, [normalizedAddress])).has(normalizedAddress)) { throw new Error('Address is already a member'); } @@ -75,8 +76,10 @@ export class Membership { if (!channel) throw new Error('Channel not found'); if (!channel.gate?.address) throw new Error('Can only add members to gated channels'); - const known = new Set(channel.members.map(m => m.toLowerCase())); - const fresh = [...new Set(addresses.map(a => a.toLowerCase()))].filter(a => !known.has(a)); + const unique = [...new Set(addresses.map(a => a.toLowerCase()))]; + const listed = new Set(channel.members.map(m => m.toLowerCase())); + const allowed = await this._stillAllowed(channel, unique.filter(a => listed.has(a))); + const fresh = unique.filter(a => !allowed.has(a)); if (fresh.length === 0) throw new Error('Every address is already a member'); try { @@ -93,6 +96,26 @@ export class Membership { } } + /** + * Of the addresses this device lists as members, the ones the gate still + * allows. The list goes stale when a moderator or another device revokes + * someone, so the entries the gate no longer allows are dropped here. + * @returns {Promise>} lowercase addresses still allowed + */ + async _stillAllowed(channel, addresses) { + if (addresses.length === 0) return new Set(); + const { gateManager } = await import('../gate.js'); + const asked = new Set(addresses); + const allowed = new Set((await gateManager.getGateMembers(channel.gate.address, addresses)) + .filter(m => asked.has(m.address) && m.allowed) + .map(m => m.address)); + channel.members = channel.members.filter(m => { + const addr = m.toLowerCase(); + return !asked.has(addr) || allowed.has(addr); + }); + return allowed; + } + /** * Remove a member from a gated channel * @param {string} messageStreamId - Message Stream ID (channel key) @@ -252,6 +275,9 @@ export class Membership { ...(channel.accessSnapshot || []), ...(channel.rotatedForNoAccess || channel.rotatedForBanned || []), ...epochKeyManager.getSeenRequesters(channel.messageStreamId), + // The requester pool stops growing at its cap; whoever holds + // the key in force must stay a candidate. + ...epochKeyManager.getCurrentKeyHolders(channel.messageStreamId), ...roster.map(m => m.account), ...onChain ]; diff --git a/src/js/epochKeyManager.js b/src/js/epochKeyManager.js index 0600cea..dbb59de 100644 --- a/src/js/epochKeyManager.js +++ b/src/js/epochKeyManager.js @@ -199,6 +199,9 @@ class EpochKeyManager { // enumerates members for TOKEN/NFT/PAID gates where holding // or pay() bypasses the owner — no indexer, no event scan. seenRequesters: new Set(), + // requestId → the account that sent that KEY_REQUEST; memory + // only, the -4 history refills it on every open + requestAuthors: new Map(), // epoch → wraps that arrived before the announce that // legitimises them. A responder answers a request as soon as // it hears it, so on a cold subscribe the wrap regularly @@ -668,6 +671,7 @@ class EpochKeyManager { } else if (data.t === KEYS_MSG_TYPE.KEY_REQUEST) { storedRequests.push({ data, publisherId, timestamp }); this._recordRequester(s, publisherId, channel.messageStreamId); + this._recordRequestAuthor(s, data.requestId, publisherId); } } if (changed) await this._persist(channel.messageStreamId, s); @@ -1436,6 +1440,7 @@ class EpochKeyManager { async _handleRequest(channel, s, data, publisherId) { this._recordRequester(s, publisherId, channel.messageStreamId); + this._recordRequestAuthor(s, data.requestId, publisherId); // Skip only what THIS session asked for, by requestId. Skipping // every request from our own account left a second device of the // same account unable to ever get the keys: nobody else answers a @@ -1681,6 +1686,31 @@ class EpochKeyManager { return s ? Array.from(s.seenRequesters) : []; } + _recordRequestAuthor(s, requestId, publisherId) { + const addr = (publisherId || '').toLowerCase(); + if (typeof requestId !== 'string' || !/^0x[0-9a-f]{40}$/.test(addr)) return; + if (!s.requestAuthors.has(requestId) && s.requestAuthors.size >= SEEN_WRAPS_MAX) { + s.requestAuthors.delete(s.requestAuthors.keys().next().value); + } + s.requestAuthors.set(requestId, addr); + } + + /** + * Accounts some responder handed the key in force to, whoever answered, + * as far as the -4 history this session read goes back. + */ + getCurrentKeyHolders(messageStreamId) { + const s = this.state.get(messageStreamId); + const keyId = s?.announces.get(s.currentEpoch)?.keyId; + if (!keyId) return []; + const holders = new Set(); + for (const [requestId, keyIds] of s.seenWraps) { + const author = s.requestAuthors.get(requestId); + if (author && keyIds.has(keyId)) holders.add(author); + } + return [...holders]; + } + _recordSeenWrap(s, requestId, keyId) { let set = s.seenWraps.get(requestId); if (!set) { diff --git a/tests/unit/channels.extended.test.js b/tests/unit/channels.extended.test.js index 75bf19b..808ac17 100644 --- a/tests/unit/channels.extended.test.js +++ b/tests/unit/channels.extended.test.js @@ -185,6 +185,7 @@ vi.mock('../../src/js/epochKeyManager.js', () => ({ epochKeyManager: { rotateEpoch: vi.fn().mockResolvedValue(undefined), getSeenRequesters: vi.fn().mockReturnValue([]), + getCurrentKeyHolders: vi.fn().mockReturnValue([]), getRosterMembers: vi.fn().mockResolvedValue([]), onKeyAdopted: vi.fn(), handleKeysMessage: vi.fn(), @@ -291,8 +292,46 @@ describe('ChannelManager Extended', () => { await expect(channelManager.addMember('public-ch', '0x1')).rejects.toThrow('gated channels'); }); - it('throws if already a member', async () => { + it('throws if the gate still allows someone already listed, with no transaction', async () => { + const { gateManager } = await import('../../src/js/gate.js'); + gateManager.getGateMembers.mockResolvedValueOnce([{ address: '0xmyaddress', allowed: true }]); await expect(channelManager.addMember(streamId, '0xMyAddress')).rejects.toThrow('already a member'); + expect(gateManager.allow).not.toHaveBeenCalled(); + }); + + it('re-adds someone this device still lists but the gate no longer allows', async () => { + const { gateManager } = await import('../../src/js/gate.js'); + channelManager.channels.get(streamId).members.push('0xremoved'); + gateManager.getGateMembers.mockResolvedValueOnce([{ address: '0xremoved', allowed: false }]); + + await channelManager.addMember(streamId, '0xRemoved'); + + expect(gateManager.allow).toHaveBeenCalledWith('0xgate', '0xRemoved'); + const listed = channelManager.channels.get(streamId).members.filter(m => m.toLowerCase() === '0xremoved'); + expect(listed).toHaveLength(1); + }); + + it('a batch skips only who the gate still allows', async () => { + const { gateManager } = await import('../../src/js/gate.js'); + gateManager.allowBatch.mockClear(); + channelManager.channels.get(streamId).members.push('0xremoved', '0xstill'); + gateManager.getGateMembers.mockResolvedValueOnce([ + { address: '0xremoved', allowed: false }, + { address: '0xstill', allowed: true } + ]); + + await channelManager.addMembers(streamId, ['0xremoved', '0xstill', '0xnew']); + + expect(gateManager.allowBatch).toHaveBeenCalledWith('0xgate', ['0xremoved', '0xnew']); + }); + + it('a batch of addresses the gate all still allows sends nothing', async () => { + const { gateManager } = await import('../../src/js/gate.js'); + gateManager.allowBatch.mockClear(); + gateManager.getGateMembers.mockResolvedValueOnce([{ address: '0xmyaddress', allowed: true }]); + + await expect(channelManager.addMembers(streamId, ['0xmyaddress'])).rejects.toThrow('already a member'); + expect(gateManager.allowBatch).not.toHaveBeenCalled(); }); it('throws chain error on gate failure', async () => { @@ -625,6 +664,69 @@ describe('ChannelManager Extended', () => { expect(channel.rotatedForNoAccess).toContain('0xmember1'); expect(channelManager.isRotationOwed(streamId)).toBe(false); }); + + describe("the owner's sweep and whoever holds the key in force", () => { + const gateSays = (access) => vi.spyOn(channelManager, 'getGateMemberFlags').mockResolvedValue([ + { address: '0xmyaddress', access: true, isOwner: true }, + { address: '0xmember3', access } + ]); + + beforeEach(() => { + channel.accessSnapshot = ['0xmyaddress', '0xmember1']; + epochKeyManager.rotateEpoch.mockResolvedValue(undefined); + }); + + afterEach(() => epochKeyManager.getCurrentKeyHolders.mockReturnValue([])); + + it('rotates for a re-admitted member who took the key in force and was removed, though covered', async () => { + channel.rotatedForNoAccess = ['0xmember3']; + epochKeyManager.getCurrentKeyHolders.mockReturnValue(['0xmember3']); + gateSays(false); + + await channelManager._rotateForLostAccess(channel); + + expect(epochKeyManager.rotateEpoch).toHaveBeenCalledTimes(1); + expect(channel.rotatedForNoAccess).toEqual(['0xmember3']); + }); + + it('leaves a covered member alone when they hold no key in force', async () => { + channel.rotatedForNoAccess = ['0xmember3']; + gateSays(false); + + await channelManager._rotateForLostAccess(channel); + + expect(epochKeyManager.rotateEpoch).not.toHaveBeenCalled(); + }); + + it('rotates for a member the owner never saw who took the key in force', async () => { + epochKeyManager.getCurrentKeyHolders.mockReturnValue(['0xmember3']); + gateSays(false); + + await channelManager._rotateForLostAccess(channel); + + expect(epochKeyManager.rotateEpoch).toHaveBeenCalledTimes(1); + expect(channel.rotatedForNoAccess).toContain('0xmember3'); + }); + + it('leaves a key holder the gate still admits alone', async () => { + epochKeyManager.getCurrentKeyHolders.mockReturnValue(['0xmember3']); + gateSays(true); + + await channelManager._rotateForLostAccess(channel); + + expect(epochKeyManager.rotateEpoch).not.toHaveBeenCalled(); + }); + + it('keeps key holders among the gate candidates', async () => { + const { gateManager } = await import('../../src/js/gate.js'); + epochKeyManager.getCurrentKeyHolders.mockReturnValue(['0xmember3']); + gateManager.getGateMembers.mockClear(); + + await channelManager.getGateMemberFlags(streamId); + + expect(gateManager.getGateMembers.mock.calls[0][1]).toContain('0xmember3'); + }); + }); }); // ==================== updateMemberPermissions ==================== @@ -1492,6 +1594,43 @@ describe('ChannelManager Extended', () => { expect(streamrController.subscribeToKeysStream).not.toHaveBeenCalled(); expect(epochKeyManager.onKeyAdopted).not.toHaveBeenCalled(); }); + + it('on a warm open, sweeps for lost access only after the -4 read that names the key holders', async () => { + const channel = gated(); + channelManager.channels.set(channel.messageStreamId, channel); + epochKeyManager.ensureChannelKeys.mockClear(); + + await channelManager.startEpochKeys(channel); + expect(channelManager._rotateForLostAccess).not.toHaveBeenCalled(); + + await vi.advanceTimersByTimeAsync(8_000); + + expect(channelManager._rotateForLostAccess).toHaveBeenCalledWith(channel); + expect(epochKeyManager.ensureChannelKeys.mock.invocationCallOrder[0]) + .toBeLessThan(channelManager._rotateForLostAccess.mock.invocationCallOrder[0]); + }); + + it('on a warm open whose -4 read fails, still sweeps with what it knows', async () => { + const channel = gated(); + channelManager.channels.set(channel.messageStreamId, channel); + epochKeyManager.ensureChannelKeys.mockRejectedValueOnce(new Error('P1 timed out')); + vi.spyOn(channelManager, '_scheduleEpochSetupRetry').mockImplementation(() => {}); + + await channelManager.startEpochKeys(channel); + await vi.advanceTimersByTimeAsync(8_000); + + expect(channelManager._rotateForLostAccess).toHaveBeenCalledWith(channel); + }); + + it('on a cold open, sweeps once the setup has read the -4, as before', async () => { + const channel = gated(); + channelManager.channels.set(channel.messageStreamId, channel); + epochKeyManager.hasCurrentKey.mockReturnValue(false); + + await channelManager.startEpochKeys(channel); + + expect(channelManager._rotateForLostAccess).toHaveBeenCalledWith(channel); + }); }); // ==================== handlePresenceMessage ==================== diff --git a/tests/unit/epochKeyManager.gatedPolicy.test.js b/tests/unit/epochKeyManager.gatedPolicy.test.js index 54ed6fe..3604385 100644 --- a/tests/unit/epochKeyManager.gatedPolicy.test.js +++ b/tests/unit/epochKeyManager.gatedPolicy.test.js @@ -149,3 +149,52 @@ describe('epochKeyManager seen requesters (N-D member candidates)', () => { expect(epochKeyManager.getSeenRequesters('0xaaa/none')).toEqual([]); }); }); + +describe('epochKeyManager holders of the key in force', () => { + const A = '0x' + 'a1'.repeat(20); + const RESPONDER = '0x' + 'b2'.repeat(20); + const streamId = '0xaaa/holders'; + const channel = { messageStreamId: streamId, keysStreamId: '0xaaa/holders-4', type: 'gated', gate: { address: GATE } }; + + const inForce = () => { + const s = epochKeyManager._getState(streamId); + s.announces = new Map([[1, { keyId: '1.old' }], [2, { keyId: '2.now' }]]); + s.currentEpoch = 2; + }; + const request = (requestId, from = A) => epochKeyManager.handleKeysMessage(channel, + { t: 'key_request', requestId, fromEpoch: 1 }, from, 1000); + const wrap = (requestId, keyId) => epochKeyManager.handleKeysMessage(channel, + { t: 'key_wrap', requestId, keyId, epoch: 2, tag: 't' }, RESPONDER, 1000); + + beforeEach(inForce); + afterEach(() => epochKeyManager.state.clear()); + + it('a wrap of the key in force from any responder makes the requester a holder', async () => { + await request('r1', A.toUpperCase().replace('0X', '0x')); + await wrap('r1', '2.now'); + expect(epochKeyManager.getCurrentKeyHolders(streamId)).toEqual([A]); + }); + + it('counts in either order', async () => { + await wrap('r1', '2.now'); + await request('r1'); + expect(epochKeyManager.getCurrentKeyHolders(streamId)).toEqual([A]); + }); + + it('a wrap only of an epoch already rotated away does not count', async () => { + await request('r1'); + await wrap('r1', '1.old'); + expect(epochKeyManager.getCurrentKeyHolders(streamId)).toEqual([]); + }); + + it('a request nobody answered does not count', async () => { + await request('r1'); + expect(epochKeyManager.getCurrentKeyHolders(streamId)).toEqual([]); + }); + + it('a wrap with no matching request in the window does not count', async () => { + await request('r1'); + await wrap('r2', '2.now'); + expect(epochKeyManager.getCurrentKeyHolders(streamId)).toEqual([]); + }); +});