diff --git a/docs/SYNC-merge-vectors.json b/docs/SYNC-merge-vectors.json index fecd4bd..51cf297 100644 --- a/docs/SYNC-merge-vectors.json +++ b/docs/SYNC-merge-vectors.json @@ -1325,6 +1325,769 @@ } } ], + "keys": [ + { + "what": "a pending re-key held here is kept", + "base": { + "channels": [ + { + "messageStreamId": "0x00000000000000000000000000000000000000a1/c0ffee-1", + "name": "Channel", + "type": "public", + "createdAt": 1000, + "joinedAt": 1000, + "storageDays": 30, + "accessSnapshot": [] + } + ], + "epochKeys": { + "0x00000000000000000000000000000000000000a1/c0ffee-1": { + "epochs": {}, + "announces": {}, + "currentEpoch": 1, + "pubKey": { + "keyId": "p1", + "keyHex": "0xbb", + "rev": 1 + }, + "pubAnnounce": { + "keyId": "p1", + "rev": 1, + "timestamp": 100 + }, + "intKey": { + "keyId": "i1", + "keyHex": "0xcc", + "rev": 1 + }, + "intAnnounce": { + "keyId": "i1", + "rev": 1, + "timestamp": 100 + }, + "intKeyPending": { + "keyId": "i2", + "keyHex": "0x22", + "rev": 2, + "oldAddress": "0x0000000000000000000000000000000000000015", + "mintedAt": 200 + } + } + } + }, + "incoming": { + "channels": [ + { + "messageStreamId": "0x00000000000000000000000000000000000000a1/c0ffee-1", + "name": "Channel", + "type": "public", + "createdAt": 1000, + "joinedAt": 1000, + "storageDays": 30, + "accessSnapshot": [] + } + ], + "epochKeys": { + "0x00000000000000000000000000000000000000a1/c0ffee-1": { + "epochs": {}, + "announces": {}, + "currentEpoch": 1, + "pubKey": { + "keyId": "p1", + "keyHex": "0xbb", + "rev": 1 + }, + "pubAnnounce": { + "keyId": "p1", + "rev": 1, + "timestamp": 100 + }, + "intKey": { + "keyId": "i1", + "keyHex": "0xcc", + "rev": 1 + }, + "intAnnounce": { + "keyId": "i1", + "rev": 1, + "timestamp": 100 + } + } + } + }, + "expected": { + "epochKeys": { + "0x00000000000000000000000000000000000000a1/c0ffee-1": { + "epochs": {}, + "announces": {}, + "currentEpoch": 1, + "pendingRequests": {}, + "helloEpochs": [], + "pubKey": { + "keyId": "p1", + "keyHex": "0xbb", + "rev": 1 + }, + "pubAnnounce": { + "keyId": "p1", + "rev": 1, + "timestamp": 100 + }, + "intKey": { + "keyId": "i1", + "keyHex": "0xcc", + "rev": 1 + }, + "intAnnounce": { + "keyId": "i1", + "rev": 1, + "timestamp": 100 + }, + "intKeyPending": { + "keyId": "i2", + "keyHex": "0x22", + "rev": 2, + "oldAddress": "0x0000000000000000000000000000000000000015", + "mintedAt": 200 + } + } + } + } + }, + { + "what": "a pending re-key from another device arrives", + "base": { + "channels": [ + { + "messageStreamId": "0x00000000000000000000000000000000000000a1/c0ffee-1", + "name": "Channel", + "type": "public", + "createdAt": 1000, + "joinedAt": 1000, + "storageDays": 30, + "accessSnapshot": [] + } + ], + "epochKeys": { + "0x00000000000000000000000000000000000000a1/c0ffee-1": { + "epochs": {}, + "announces": {}, + "currentEpoch": 1, + "pubKey": { + "keyId": "p1", + "keyHex": "0xbb", + "rev": 1 + }, + "pubAnnounce": { + "keyId": "p1", + "rev": 1, + "timestamp": 100 + }, + "intKey": { + "keyId": "i1", + "keyHex": "0xcc", + "rev": 1 + }, + "intAnnounce": { + "keyId": "i1", + "rev": 1, + "timestamp": 100 + } + } + } + }, + "incoming": { + "channels": [ + { + "messageStreamId": "0x00000000000000000000000000000000000000a1/c0ffee-1", + "name": "Channel", + "type": "public", + "createdAt": 1000, + "joinedAt": 1000, + "storageDays": 30, + "accessSnapshot": [] + } + ], + "epochKeys": { + "0x00000000000000000000000000000000000000a1/c0ffee-1": { + "epochs": {}, + "announces": {}, + "currentEpoch": 1, + "pubKey": { + "keyId": "p1", + "keyHex": "0xbb", + "rev": 1 + }, + "pubAnnounce": { + "keyId": "p1", + "rev": 1, + "timestamp": 100 + }, + "intKey": { + "keyId": "i1", + "keyHex": "0xcc", + "rev": 1 + }, + "intAnnounce": { + "keyId": "i1", + "rev": 1, + "timestamp": 100 + }, + "intKeyPending": { + "keyId": "i2", + "keyHex": "0x22", + "rev": 2, + "oldAddress": "0x0000000000000000000000000000000000000015", + "mintedAt": 200 + } + } + } + }, + "expected": { + "epochKeys": { + "0x00000000000000000000000000000000000000a1/c0ffee-1": { + "epochs": {}, + "announces": {}, + "currentEpoch": 1, + "pendingRequests": {}, + "helloEpochs": [], + "pubKey": { + "keyId": "p1", + "keyHex": "0xbb", + "rev": 1 + }, + "pubAnnounce": { + "keyId": "p1", + "rev": 1, + "timestamp": 100 + }, + "intKey": { + "keyId": "i1", + "keyHex": "0xcc", + "rev": 1 + }, + "intAnnounce": { + "keyId": "i1", + "rev": 1, + "timestamp": 100 + }, + "intKeyPending": { + "keyId": "i2", + "keyHex": "0x22", + "rev": 2, + "oldAddress": "0x0000000000000000000000000000000000000015", + "mintedAt": 200 + } + } + } + } + }, + { + "what": "the later of two pending re-keys wins", + "base": { + "channels": [ + { + "messageStreamId": "0x00000000000000000000000000000000000000a1/c0ffee-1", + "name": "Channel", + "type": "public", + "createdAt": 1000, + "joinedAt": 1000, + "storageDays": 30, + "accessSnapshot": [] + } + ], + "epochKeys": { + "0x00000000000000000000000000000000000000a1/c0ffee-1": { + "epochs": {}, + "announces": {}, + "currentEpoch": 1, + "pubKey": { + "keyId": "p1", + "keyHex": "0xbb", + "rev": 1 + }, + "pubAnnounce": { + "keyId": "p1", + "rev": 1, + "timestamp": 100 + }, + "intKey": { + "keyId": "i1", + "keyHex": "0xcc", + "rev": 1 + }, + "intAnnounce": { + "keyId": "i1", + "rev": 1, + "timestamp": 100 + }, + "intKeyPending": { + "keyId": "i2", + "keyHex": "0x22", + "rev": 2, + "oldAddress": "0x0000000000000000000000000000000000000015", + "mintedAt": 200 + } + } + } + }, + "incoming": { + "channels": [ + { + "messageStreamId": "0x00000000000000000000000000000000000000a1/c0ffee-1", + "name": "Channel", + "type": "public", + "createdAt": 1000, + "joinedAt": 1000, + "storageDays": 30, + "accessSnapshot": [] + } + ], + "epochKeys": { + "0x00000000000000000000000000000000000000a1/c0ffee-1": { + "epochs": {}, + "announces": {}, + "currentEpoch": 1, + "pubKey": { + "keyId": "p1", + "keyHex": "0xbb", + "rev": 1 + }, + "pubAnnounce": { + "keyId": "p1", + "rev": 1, + "timestamp": 100 + }, + "intKey": { + "keyId": "i1", + "keyHex": "0xcc", + "rev": 1 + }, + "intAnnounce": { + "keyId": "i1", + "rev": 1, + "timestamp": 100 + }, + "intKeyPending": { + "keyId": "i3", + "keyHex": "0x33", + "rev": 3, + "oldAddress": "0x0000000000000000000000000000000000000015", + "mintedAt": 300 + } + } + } + }, + "expected": { + "epochKeys": { + "0x00000000000000000000000000000000000000a1/c0ffee-1": { + "epochs": {}, + "announces": {}, + "currentEpoch": 1, + "pendingRequests": {}, + "helloEpochs": [], + "pubKey": { + "keyId": "p1", + "keyHex": "0xbb", + "rev": 1 + }, + "pubAnnounce": { + "keyId": "p1", + "rev": 1, + "timestamp": 100 + }, + "intKey": { + "keyId": "i1", + "keyHex": "0xcc", + "rev": 1 + }, + "intAnnounce": { + "keyId": "i1", + "rev": 1, + "timestamp": 100 + }, + "intKeyPending": { + "keyId": "i3", + "keyHex": "0x33", + "rev": 3, + "oldAddress": "0x0000000000000000000000000000000000000015", + "mintedAt": 300 + } + } + } + } + }, + { + "what": "a pending re-key another device promoted is dropped", + "base": { + "channels": [ + { + "messageStreamId": "0x00000000000000000000000000000000000000a1/c0ffee-1", + "name": "Channel", + "type": "public", + "createdAt": 1000, + "joinedAt": 1000, + "storageDays": 30, + "accessSnapshot": [] + } + ], + "epochKeys": { + "0x00000000000000000000000000000000000000a1/c0ffee-1": { + "epochs": {}, + "announces": {}, + "currentEpoch": 1, + "pubKey": { + "keyId": "p1", + "keyHex": "0xbb", + "rev": 1 + }, + "pubAnnounce": { + "keyId": "p1", + "rev": 1, + "timestamp": 100 + }, + "intKey": { + "keyId": "i1", + "keyHex": "0xcc", + "rev": 1 + }, + "intAnnounce": { + "keyId": "i1", + "rev": 1, + "timestamp": 100 + }, + "intKeyPending": { + "keyId": "i2", + "keyHex": "0x22", + "rev": 2, + "oldAddress": "0x0000000000000000000000000000000000000015", + "mintedAt": 200 + } + } + } + }, + "incoming": { + "channels": [ + { + "messageStreamId": "0x00000000000000000000000000000000000000a1/c0ffee-1", + "name": "Channel", + "type": "public", + "createdAt": 1000, + "joinedAt": 1000, + "storageDays": 30, + "accessSnapshot": [] + } + ], + "epochKeys": { + "0x00000000000000000000000000000000000000a1/c0ffee-1": { + "epochs": {}, + "announces": {}, + "currentEpoch": 1, + "pubKey": { + "keyId": "p1", + "keyHex": "0xbb", + "rev": 1 + }, + "pubAnnounce": { + "keyId": "p1", + "rev": 1, + "timestamp": 100 + }, + "intKey": { + "keyId": "i2", + "keyHex": "0x22", + "rev": 2 + }, + "intAnnounce": { + "keyId": "i2", + "rev": 2, + "timestamp": 200 + } + } + } + }, + "expected": { + "epochKeys": { + "0x00000000000000000000000000000000000000a1/c0ffee-1": { + "epochs": {}, + "announces": {}, + "currentEpoch": 1, + "pendingRequests": {}, + "helloEpochs": [], + "pubKey": { + "keyId": "p1", + "keyHex": "0xbb", + "rev": 1 + }, + "pubAnnounce": { + "keyId": "p1", + "rev": 1, + "timestamp": 100 + }, + "intKey": { + "keyId": "i2", + "keyHex": "0x22", + "rev": 2 + }, + "intAnnounce": { + "keyId": "i2", + "rev": 2, + "timestamp": 200 + } + } + } + } + }, + { + "what": "a pending re-key a later announce superseded is dropped", + "base": { + "channels": [ + { + "messageStreamId": "0x00000000000000000000000000000000000000a1/c0ffee-1", + "name": "Channel", + "type": "public", + "createdAt": 1000, + "joinedAt": 1000, + "storageDays": 30, + "accessSnapshot": [] + } + ], + "epochKeys": { + "0x00000000000000000000000000000000000000a1/c0ffee-1": { + "epochs": {}, + "announces": {}, + "currentEpoch": 1, + "pubKey": { + "keyId": "p1", + "keyHex": "0xbb", + "rev": 1 + }, + "pubAnnounce": { + "keyId": "p1", + "rev": 1, + "timestamp": 100 + }, + "intKey": { + "keyId": "i1", + "keyHex": "0xcc", + "rev": 1 + }, + "intAnnounce": { + "keyId": "i1", + "rev": 1, + "timestamp": 100 + }, + "intKeyPending": { + "keyId": "i2", + "keyHex": "0x22", + "rev": 2, + "oldAddress": "0x0000000000000000000000000000000000000015", + "mintedAt": 200 + } + } + } + }, + "incoming": { + "channels": [ + { + "messageStreamId": "0x00000000000000000000000000000000000000a1/c0ffee-1", + "name": "Channel", + "type": "public", + "createdAt": 1000, + "joinedAt": 1000, + "storageDays": 30, + "accessSnapshot": [] + } + ], + "epochKeys": { + "0x00000000000000000000000000000000000000a1/c0ffee-1": { + "epochs": {}, + "announces": {}, + "currentEpoch": 1, + "pubKey": { + "keyId": "p1", + "keyHex": "0xbb", + "rev": 1 + }, + "pubAnnounce": { + "keyId": "p1", + "rev": 1, + "timestamp": 100 + }, + "intKey": { + "keyId": "i1", + "keyHex": "0xcc", + "rev": 1 + }, + "intAnnounce": { + "keyId": "i3", + "rev": 3, + "timestamp": 300 + } + } + } + }, + "expected": { + "epochKeys": { + "0x00000000000000000000000000000000000000a1/c0ffee-1": { + "epochs": {}, + "announces": {}, + "currentEpoch": 1, + "pendingRequests": {}, + "helloEpochs": [], + "pubKey": { + "keyId": "p1", + "keyHex": "0xbb", + "rev": 1 + }, + "pubAnnounce": { + "keyId": "p1", + "rev": 1, + "timestamp": 100 + }, + "intKey": { + "keyId": "i1", + "keyHex": "0xcc", + "rev": 1 + }, + "intAnnounce": { + "keyId": "i3", + "rev": 3, + "timestamp": 300 + } + } + } + } + }, + { + "what": "a pending publish key travels like the interactions one", + "base": { + "channels": [ + { + "messageStreamId": "0x00000000000000000000000000000000000000a1/c0ffee-1", + "name": "Channel", + "type": "public", + "createdAt": 1000, + "joinedAt": 1000, + "storageDays": 30, + "accessSnapshot": [] + } + ], + "epochKeys": { + "0x00000000000000000000000000000000000000a1/c0ffee-1": { + "epochs": {}, + "announces": {}, + "currentEpoch": 1, + "pubKey": { + "keyId": "p1", + "keyHex": "0xbb", + "rev": 1 + }, + "pubAnnounce": { + "keyId": "p1", + "rev": 1, + "timestamp": 100 + }, + "intKey": { + "keyId": "i1", + "keyHex": "0xcc", + "rev": 1 + }, + "intAnnounce": { + "keyId": "i1", + "rev": 1, + "timestamp": 100 + } + } + } + }, + "incoming": { + "channels": [ + { + "messageStreamId": "0x00000000000000000000000000000000000000a1/c0ffee-1", + "name": "Channel", + "type": "public", + "createdAt": 1000, + "joinedAt": 1000, + "storageDays": 30, + "accessSnapshot": [] + } + ], + "epochKeys": { + "0x00000000000000000000000000000000000000a1/c0ffee-1": { + "epochs": {}, + "announces": {}, + "currentEpoch": 1, + "pubKey": { + "keyId": "p1", + "keyHex": "0xbb", + "rev": 1 + }, + "pubAnnounce": { + "keyId": "p1", + "rev": 1, + "timestamp": 100 + }, + "intKey": { + "keyId": "i1", + "keyHex": "0xcc", + "rev": 1 + }, + "intAnnounce": { + "keyId": "i1", + "rev": 1, + "timestamp": 100 + }, + "pubKeyPending": { + "keyId": "p2", + "keyHex": "0x22", + "rev": 2, + "oldAddress": "0x0000000000000000000000000000000000000015", + "mintedAt": 200 + } + } + } + }, + "expected": { + "epochKeys": { + "0x00000000000000000000000000000000000000a1/c0ffee-1": { + "epochs": {}, + "announces": {}, + "currentEpoch": 1, + "pendingRequests": {}, + "helloEpochs": [], + "pubKey": { + "keyId": "p1", + "keyHex": "0xbb", + "rev": 1 + }, + "pubAnnounce": { + "keyId": "p1", + "rev": 1, + "timestamp": 100 + }, + "intKey": { + "keyId": "i1", + "keyHex": "0xcc", + "rev": 1 + }, + "intAnnounce": { + "keyId": "i1", + "rev": 1, + "timestamp": 100 + }, + "pubKeyPending": { + "keyId": "p2", + "keyHex": "0x22", + "rev": 2, + "oldAddress": "0x0000000000000000000000000000000000000015", + "mintedAt": 200 + } + } + } + } + } + ], "publish": { "base": { "sentMessages": { @@ -1455,6 +2218,20 @@ "keyId": "i1", "rev": 1, "timestamp": 100 + }, + "pubKeyPending": { + "keyId": "p2", + "keyHex": "0xdd", + "rev": 2, + "oldAddress": "0x0000000000000000000000000000000000000013", + "mintedAt": 100 + }, + "intKeyPending": { + "keyId": "i2", + "keyHex": "0xee", + "rev": 2, + "oldAddress": "0x0000000000000000000000000000000000000014", + "mintedAt": 100 } } }, @@ -2312,6 +3089,52 @@ }, "same": false }, + { + "what": "a change in a channel's epoch keys pubKeyPending is news", + "patch": { + "set": [ + [ + [ + "epochKeys", + "0x00000000000000000000000000000000000000a1/c0ffee-1", + "pubKeyPending" + ], + { + "keyId": "p2", + "keyHex": "0xdd", + "rev": 2, + "oldAddress": "0x0000000000000000000000000000000000000013", + "mintedAt": 100, + "x": 1 + } + ] + ] + }, + "same": false + }, + { + "what": "a change in a channel's epoch keys intKeyPending is news", + "patch": { + "set": [ + [ + [ + "epochKeys", + "0x00000000000000000000000000000000000000a1/c0ffee-1", + "intKeyPending" + ], + { + "keyId": "i2", + "keyHex": "0xee", + "rev": 2, + "oldAddress": "0x0000000000000000000000000000000000000014", + "mintedAt": 100, + "x": 1 + } + ] + ] + }, + "same": false + }, { "what": "a change in a channel's epoch keys announces is not news", "patch": { diff --git a/src/js/backupImport.js b/src/js/backupImport.js index fd7777d..e20a603 100644 --- a/src/js/backupImport.js +++ b/src/js/backupImport.js @@ -17,7 +17,7 @@ */ import { Logger } from './logger.js'; -import { mergeChannels, mergeSentDeletedAt, withoutDeleted } from './syncMerge.js'; +import { dropSettledPending, mergeChannels, mergeSentDeletedAt, withoutDeleted } from './syncMerge.js'; /** * Apply backup state to the unlocked secure storage and the live managers. @@ -188,6 +188,14 @@ export async function importBackupData(data, { secureStorage, channelManager, id local.currentEpoch = incoming.currentEpoch; summary.changed = true; } + // Shared keys and pending re-keys: the higher rev wins, as in the sync merge. + for (const field of ['pubKey', 'pubAnnounce', 'intKey', 'intAnnounce', 'pubKeyPending', 'intKeyPending']) { + if (incoming[field] && (incoming[field].rev || 0) > (local[field]?.rev || 0)) { + local[field] = incoming[field]; + summary.changed = true; + } + } + dropSettledPending(local); } } diff --git a/src/js/epochKeyManager.js b/src/js/epochKeyManager.js index dbb59de..e226733 100644 --- a/src/js/epochKeyManager.js +++ b/src/js/epochKeyManager.js @@ -125,6 +125,22 @@ const ROTATION_RETRY_MS = 60 * 60 * 1000; /** Wraps held for an announce that has not arrived; memory only. */ const PARKED_WRAPS_PER_EPOCH = 4; const PARKED_EPOCHS = 8; +// A re-key whose grant is nowhere on chain after this long never landed. It +// outlasts the write's own retries, so a transaction still pending is not +// mistaken for one that was never sent. +const REKEY_PENDING_MAX_AGE_MS = 60 * 60 * 1000; +const REKEY_SLOTS = { + pub: { + key: 'pubKey', announce: 'pubAnnounce', pending: 'pubKeyPending', + freshness: 'pubAnnounceFreshness', grants: 'rekeySharedPublishGrants', label: 'publish', + unsettledWarning: 'The publish key reset has not finished. It retries when you open the channel.' + }, + int: { + key: 'intKey', announce: 'intAnnounce', pending: 'intKeyPending', + freshness: 'intAnnounceFreshness', grants: 'rekeyInteractionsGrants', label: 'interactions', + unsettledWarning: 'The interactions key reset has not finished. It retries when you open the channel.' + } +}; class EpochKeyManager { constructor() { @@ -184,6 +200,12 @@ class EpochKeyManager { // Newest pub_announce timestamp seen in storage — drives the // TTL re-announce, exactly like announceFreshness. pubAnnounceFreshness: 0, + // A re-key between mint and on-chain confirmation: + // { keyId, keyHex, address, rev, oldAddress, mintedAt }. + // Persisted BEFORE the grant is sent: a grant that lands while + // the call fails has already revoked the old key. + pubKeyPending: null, + intKeyPending: null, // Session pseudonym for our own publishes in this channel: // { privateKey, publicKey, bindProof } — MEMORY ONLY. Members // resolve the account from the bind proof, so a fresh @@ -270,6 +292,10 @@ class EpochKeyManager { if (persisted.intAnnounce && (persisted.intAnnounce.rev || 0) > (s.intAnnounce?.rev || 0)) { s.intAnnounce = { ...persisted.intAnnounce }; } + for (const slot of ['pubKeyPending', 'intKeyPending']) { + const pending = persisted[slot]; + if (pending && (pending.rev || 0) > (s[slot]?.rev || 0)) s[slot] = { ...pending }; + } } async _persist(messageStreamId, s) { @@ -304,6 +330,8 @@ class EpochKeyManager { } } : {}), ...(s.intKey ? { intKey: { ...s.intKey } } : {}), + ...(s.pubKeyPending ? { pubKeyPending: { ...s.pubKeyPending } } : {}), + ...(s.intKeyPending ? { intKeyPending: { ...s.intKeyPending } } : {}), ...(s.intAnnounce ? { intAnnounce: { keyId: s.intAnnounce.keyId, keyHash: s.intAnnounce.keyHash, @@ -440,37 +468,7 @@ class EpochKeyManager { if (!this.isOwnAdmin(channel)) { throw new Error('rekeyPublishKey: only the channel admin can re-key'); } - const s = this._getState(channel.messageStreamId); - if (!s.loaded) { - this._loadPersisted(channel.messageStreamId, s); - s.loaded = true; - } - const oldAddress = s.pubKey?.address || s.pubAnnounce?.address || null; - const rev = Math.max(s.pubKey?.rev || 0, s.pubAnnounce?.rev || 0) + 1; - const pubKey = this.mintPublishKey(rev); - - // Chain first: a published announce for a key the network rejects - // would strand every member on an unusable key. - await streamrController.rekeySharedPublishGrants(channel, pubKey.address, oldAddress); - - s.pubKey = { ...pubKey }; - const announce = { - t: KEYS_MSG_TYPE.PUB_ANNOUNCE, - keyId: pubKey.keyId, - keyHash: await epochKeyCrypto.computeKeyHash(pubKey.keyHex), - addr: pubKey.address, - rev - }; - await streamrController.publishKeysMessage(channel.keysStreamId, announce); - this._applyPubAnnounce(channel, s, announce, authManager.getAddress(), Date.now()); - s.pubAnnounceFreshness = Date.now(); - await this._persist(channel.messageStreamId, s); - this.onKeysAdopted?.(channel.messageStreamId, pubKey.keyId); - Logger.info(`epochKeys: publish key re-keyed to rev ${rev} on`, channel.keysStreamId.slice(-30)); - // Members cannot write until this announce is readable from storage — - // verify retention exactly like a fresh epoch announce. - this._ensureAnnounceRetained(channel, announce).catch(() => {}); - return rev; + return this._rekeyShared(channel, 'pub'); } /** @@ -484,36 +482,148 @@ class EpochKeyManager { if (!this.isOwnAdmin(channel)) { throw new Error('rekeyInteractionsKey: only the channel admin can re-key'); } + return this._rekeyShared(channel, 'int'); + } + + async _rekeyShared(channel, kind) { + const slots = REKEY_SLOTS[kind]; const s = this._getState(channel.messageStreamId); if (!s.loaded) { this._loadPersisted(channel.messageStreamId, s); s.loaded = true; } - const oldAddress = s.intKey?.address || s.intAnnounce?.address || null; - const rev = Math.max(s.intKey?.rev || 0, s.intAnnounce?.rev || 0) + 1; - const intKey = this.mintInteractionsKey(rev); + // Minting over a pending key the chain cannot settle drops the only + // record of a key whose grant may have landed. + if (s[slots.pending] && await this._reconcileRekey(channel, s, kind) === 'unreadable') { + throw new Error('the previous reset has not settled and the chain cannot be read right now'); + } + const unsettled = s[slots.pending]; + const oldAddress = s[slots.key]?.address || s[slots.announce]?.address || null; + const rev = Math.max(s[slots.key]?.rev || 0, s[slots.announce]?.rev || 0, unsettled?.rev || 0) + 1; + const key = kind === 'int' ? this.mintInteractionsKey(rev) : this.mintPublishKey(rev); + s[slots.pending] = { ...key, oldAddress, mintedAt: Date.now() }; + await this._persist(channel.messageStreamId, s); + this.onKeysAdopted?.(channel.messageStreamId, key.keyId); + // An unsettled earlier re-key may still land: revoking its key too + // leaves the new one as the only holder whichever lands last. + const revoke = [...new Set([oldAddress, unsettled?.address] + .filter(Boolean).map((address) => address.toLowerCase()))]; // Chain first: a published announce for a key the network rejects - // would leave every member reacting into the void. - await streamrController.rekeyInteractionsGrants(channel, intKey.address, oldAddress); + // would strand every member on an unusable key. + try { + await streamrController[slots.grants](channel, key.address, revoke); + } catch (e) { + if (await this._reconcileRekey(channel, s, kind) !== 'promoted') throw e; + return rev; + } + await this._promoteRekey(channel, s, kind, key.keyId); + return rev; + } + + /** The pending key becomes the held one, reaches sync, and is announced. */ + async _promoteRekey(channel, s, kind, expectedKeyId) { + const slots = REKEY_SLOTS[kind]; + const pending = s[slots.pending]; + if (pending?.keyId !== expectedKeyId) return false; + const { keyId, keyHex, address, rev } = pending; + s[slots.key] = { keyId, keyHex, address, rev }; + s[slots.pending] = null; + await this._persist(channel.messageStreamId, s); + this.onKeysAdopted?.(channel.messageStreamId, keyId); - s.intKey = { ...intKey }; const announce = { t: KEYS_MSG_TYPE.PUB_ANNOUNCE, - k: 'i', - keyId: intKey.keyId, - keyHash: await epochKeyCrypto.computeKeyHash(intKey.keyHex), - addr: intKey.address, + ...(kind === 'int' ? { k: 'i' } : {}), + keyId, + keyHash: await epochKeyCrypto.computeKeyHash(keyHex), + addr: address, rev }; await streamrController.publishKeysMessage(channel.keysStreamId, announce); this._applyPubAnnounce(channel, s, announce, authManager.getAddress(), Date.now()); - s.intAnnounceFreshness = Date.now(); + s[slots.freshness] = Date.now(); await this._persist(channel.messageStreamId, s); - this.onKeysAdopted?.(channel.messageStreamId, intKey.keyId); - Logger.info(`epochKeys: interactions key re-keyed to rev ${rev} on`, channel.keysStreamId.slice(-30)); + Logger.info(`epochKeys: ${slots.label} key re-keyed to rev ${rev} on`, channel.keysStreamId.slice(-30)); + // Members cannot write until this announce is readable from storage — + // verify retention exactly like a fresh epoch announce. this._ensureAnnounceRetained(channel, announce).catch(() => {}); - return rev; + return true; + } + + /** + * Settle a pending re-key from what the chain holds. The old key is let go + * only once the new one holds its grant on every stream. + * @returns {Promise<'none'|'promoted'|'dropped'|'kept'|'partial'|'unreadable'|'inconsistent'>} + */ + async _reconcileRekey(channel, s, kind) { + const slots = REKEY_SLOTS[kind]; + const pending = s[slots.pending]; + if (!pending) return 'none'; + const held = s[slots.key]; + const announced = s[slots.announce]; + + if (announced?.keyId === pending.keyId) { + // Another device of the account confirmed and announced it. + const { keyId, keyHex, address, rev } = pending; + s[slots.key] = { keyId, keyHex, address, rev }; + s[slots.pending] = null; + await this._persist(channel.messageStreamId, s); + this.onKeysAdopted?.(channel.messageStreamId, keyId); + return 'promoted'; + } + if (pending.rev <= Math.max(held?.rev || 0, announced?.rev || 0)) { + s[slots.pending] = null; + await this._persist(channel.messageStreamId, s); + return 'dropped'; + } + + let grants; + try { + grants = await streamrController.rekeyGrantsState(channel, kind, pending.address, pending.oldAddress); + } catch (e) { + Logger.warn(`epochKeys: ${slots.label} re-key unsettled, chain unreadable:`, e?.message); + return 'unreadable'; + } + if (grants.next.every(Boolean)) { + return await this._promoteRekey(channel, s, kind, pending.keyId) ? 'promoted' : 'none'; + } + if (grants.next.some(Boolean)) { + // Only the publish key's grants are one transaction per stream: + // finish the reset the owner already started on the rest. + if (kind !== 'pub') return 'partial'; + const missing = grants.streamIds.filter((_, index) => !grants.next[index]); + try { + await streamrController.rekeySharedPublishGrants(channel, pending.address, + pending.oldAddress ? [pending.oldAddress] : [], missing); + } catch (e) { + Logger.warn(`epochKeys: ${slots.label} re-key still partial:`, e?.message); + return 'partial'; + } + return await this._promoteRekey(channel, s, kind, pending.keyId) ? 'promoted' : 'none'; + } + if (pending.oldAddress && !grants.old.every(Boolean)) return 'inconsistent'; + if (Date.now() - (pending.mintedAt || 0) < REKEY_PENDING_MAX_AGE_MS) return 'kept'; + if (s[slots.pending]?.keyId !== pending.keyId) return 'none'; + s[slots.pending] = null; + await this._persist(channel.messageStreamId, s); + Logger.info(`epochKeys: ${slots.label} re-key never landed, pending key dropped on`, + channel.keysStreamId.slice(-30)); + return 'dropped'; + } + + /** Channel-open settle of both shared keys; never fails the open. */ + async _settleRekeys(channel, s) { + for (const kind of ['pub', 'int']) { + try { + const outcome = await this._reconcileRekey(channel, s, kind); + if (outcome === 'unreadable' || outcome === 'inconsistent' || outcome === 'partial') { + this._gateWarningHandler?.(channel.messageStreamId, REKEY_SLOTS[kind].unsettledWarning); + } + } catch (e) { + Logger.warn(`epochKeys: ${REKEY_SLOTS[kind].label} re-key settle failed:`, e?.message); + } + } } /** @@ -707,6 +817,7 @@ class EpochKeyManager { if (this.isOwnAdmin(channel)) { await this._bootstrapFirstEpoch(channel, s); this._armScheduledRotation(channel, s); + await this._settleRekeys(channel, s); await this._maybeAnnouncePub(channel, s); await this._maybeAnnounceInteractions(channel, s); } else { @@ -726,6 +837,7 @@ class EpochKeyManager { if (this.isOwnAdmin(channel)) { await this._maybeReannounceAging(channel, s); this._armScheduledRotation(channel, s); + await this._settleRekeys(channel, s); await this._maybeAnnouncePub(channel, s); await this._maybeAnnounceInteractions(channel, s); } @@ -790,9 +902,12 @@ class EpochKeyManager { // is on-chain work. if (!usesSharedPublish(channel) || !s.intKey) return; if (s.intAnnounce && s.intAnnounce.rev > s.intKey.rev) return; + // A fresh announce of an OLDER rev says nothing about the held key. + const unannounced = (s.intAnnounce?.rev || 0) < s.intKey.rev; const retentionMs = keysRetentionDays(channel) * 86_400_000; const freshest = s.intAnnounceFreshness || 0; - if (freshest && Date.now() - freshest < retentionMs * CONFIG.storage.ttlRepublishAgeFraction) return; + if (!unannounced && freshest + && Date.now() - freshest < retentionMs * CONFIG.storage.ttlRepublishAgeFraction) return; const announce = { t: KEYS_MSG_TYPE.PUB_ANNOUNCE, @@ -813,9 +928,11 @@ class EpochKeyManager { async _maybeAnnouncePub(channel, s) { if (!usesSharedPublish(channel) || !s.pubKey) return; if (s.pubAnnounce && s.pubAnnounce.rev > s.pubKey.rev) return; // we hold the superseded key + const unannounced = (s.pubAnnounce?.rev || 0) < s.pubKey.rev; const retentionMs = keysRetentionDays(channel) * 86_400_000; const freshest = s.pubAnnounceFreshness || 0; - if (freshest && Date.now() - freshest < retentionMs * CONFIG.storage.ttlRepublishAgeFraction) return; + if (!unannounced && freshest + && Date.now() - freshest < retentionMs * CONFIG.storage.ttlRepublishAgeFraction) return; const announce = { t: KEYS_MSG_TYPE.PUB_ANNOUNCE, diff --git a/src/js/streamr.js b/src/js/streamr.js index e782732..c83f87b 100644 --- a/src/js/streamr.js +++ b/src/js/streamr.js @@ -2837,20 +2837,27 @@ class StreamrController { return password ? cryptoManager.encryptedLength(bytes) + 2 : bytes; } + /** The streams a shared key ('pub' or 'int') publishes to. */ + _rekeyStreamIds(channel, kind) { + return (kind === 'int' + ? [channel.interactionsStreamId || deriveInteractionsId(channel.messageStreamId), channel.ephemeralStreamId] + : [channel.messageStreamId, channel.ephemeralStreamId] + ).filter(Boolean); + } + /** * Re-key a Sealed channel's shared publish grants: the new key's - * address gains publish+subscribe on -1/-2 and the old one loses - * everything — one setPermissions tx per stream (an assignment with an - * empty permission list clears that user). The admin escape valve + * address gains publish+subscribe on -1/-2 and every address in `revoke` + * loses everything — one setPermissions tx per stream (an assignment with + * an empty permission list clears that user). The admin escape valve * against ex-key-holder abuse; exceptional, never routine. */ - async rekeySharedPublishGrants(channel, newAddress, oldAddress) { - for (const streamId of [channel.messageStreamId, channel.ephemeralStreamId]) { - if (!streamId) continue; + async rekeySharedPublishGrants(channel, newAddress, revoke = [], streamIds = this._rekeyStreamIds(channel, 'pub')) { + for (const streamId of streamIds) { const assignments = [ // PUBLISH alone: the shared key writes, the clone reads. { userId: newAddress.toLowerCase(), permissions: ['publish'] }, - ...(oldAddress ? [{ userId: oldAddress.toLowerCase(), permissions: [] }] : []) + ...revoke.map((address) => ({ userId: address.toLowerCase(), permissions: [] })) ]; await executeWithRetryAndVerify(`rekeySharedPublishGrants(${streamId.slice(-20)})`, async () => { await this.client.setPermissions({ streamId, assignments }); @@ -2861,17 +2868,15 @@ class StreamrController { /** * Re-key a Sealed channel's interactions grants on -5 and -2 in ONE - * transaction: the new key's address gains publish, the old one loses it. + * transaction: the new key's address gains publish, every address in + * `revoke` loses it. */ - async rekeyInteractionsGrants(channel, newAddress, oldAddress) { + async rekeyInteractionsGrants(channel, newAddress, revoke = []) { const assignments = [ { userId: newAddress.toLowerCase(), permissions: ['publish'] }, - ...(oldAddress ? [{ userId: oldAddress.toLowerCase(), permissions: [] }] : []) + ...revoke.map((address) => ({ userId: address.toLowerCase(), permissions: [] })) ]; - const items = [ - channel.interactionsStreamId || deriveInteractionsId(channel.messageStreamId), - channel.ephemeralStreamId - ].filter(Boolean).map((streamId) => ({ streamId, assignments })); + const items = this._rekeyStreamIds(channel, 'int').map((streamId) => ({ streamId, assignments })); await executeWithRetryAndVerify('rekeyInteractionsGrants', async () => { await this.client.setPermissions(...items); return true; @@ -2883,6 +2888,24 @@ class StreamrController { }); } + /** + * Which side of a re-key holds PUBLISH on chain, stream by stream. Reads + * the registry; throws when the chain cannot be read. + * @returns {Promise<{streamIds: string[], next: boolean[], old: boolean[]}>} + */ + async rekeyGrantsState(channel, kind, newAddress, oldAddress) { + const holds = (streamId, address) => this.client.hasPermission( + { streamId, userId: address.toLowerCase(), permission: 'publish', allowPublic: false }); + const streamIds = this._rekeyStreamIds(channel, kind); + const next = []; + const old = []; + for (const streamId of streamIds) { + next.push(await holds(streamId, newAddress)); + old.push(oldAddress ? await holds(streamId, oldAddress) : false); + } + return { streamIds, next, old }; + } + /** * Identity for the channel's SHARED publish key, cached per keyId — a * re-key changes the keyId and naturally mints the replacement. diff --git a/src/js/syncMerge.js b/src/js/syncMerge.js index 292f7ae..8b0b444 100644 --- a/src/js/syncMerge.js +++ b/src/js/syncMerge.js @@ -294,18 +294,41 @@ export function mergeEpochKeys(base, incoming, keepIds = null) { const pubAnnounce = higherRev(b.pubAnnounce, i.pubAnnounce); const intKey = higherRev(b.intKey, i.intKey); const intAnnounce = higherRev(b.intAnnounce, i.intAnnounce); - result[streamId] = { + const pubKeyPending = higherRev(b.pubKeyPending, i.pubKeyPending); + const intKeyPending = higherRev(b.intKeyPending, i.intKeyPending); + result[streamId] = dropSettledPending({ epochs, announces, currentEpoch, pendingRequests, helloEpochs, ...(pubKey ? { pubKey } : {}), ...(pubAnnounce ? { pubAnnounce } : {}), ...(intKey ? { intKey } : {}), - ...(intAnnounce ? { intAnnounce } : {}) - }; + ...(intAnnounce ? { intAnnounce } : {}), + ...(pubKeyPending ? { pubKeyPending } : {}), + ...(intKeyPending ? { intKeyPending } : {}) + }); } return result; } +/** + * A pending re-key travels with the shared keys; a key or announce at its rev + * or above means some device already settled it, so it is dropped. + * @param {Object} record - One channel's epoch-key record, changed in place + * @returns {Object} the same record + */ +export function dropSettledPending(record) { + for (const [pending, key, announce] of [ + ['pubKeyPending', 'pubKey', 'pubAnnounce'], + ['intKeyPending', 'intKey', 'intAnnounce'] + ]) { + if (record[pending] && (record[pending].rev || 0) + <= Math.max(record[key]?.rev || 0, record[announce]?.rev || 0)) { + delete record[pending]; + } + } + return record; +} + /** * Defaults for the timestamped latest-wins slices. */ diff --git a/tests/unit/backupImport.test.js b/tests/unit/backupImport.test.js index 803f56b..2372ad0 100644 --- a/tests/unit/backupImport.test.js +++ b/tests/unit/backupImport.test.js @@ -283,6 +283,43 @@ describe('importBackupData', () => { expect(merged.currentEpoch).toBe(2); // never regresses }); + it('restores the shared keys by rev when the channel already has local keys', async () => { + const key = (prefix, rev) => ({ keyId: `${prefix}${rev}`, keyHex: `0x${rev}`, rev }); + const { secureStorage, channelManager, cache } = makeDeps({ + epochKeys: { + 'ch-1': { epochs: {}, currentEpoch: 1, intKey: key('i', 1), pubKey: key('p', 3) }, + 'ch-2': { epochs: {}, currentEpoch: 1, intKey: key('i', 1) } + } + }); + const data = { + epochKeys: { + 'ch-1': { epochs: {}, currentEpoch: 1, intKey: key('i', 2), pubKey: key('p', 2) }, + 'ch-2': { epochs: {}, currentEpoch: 1, intKeyPending: key('i', 2) } + } + }; + + const summary = await importBackupData(data, { secureStorage, channelManager }); + + expect(summary.changed).toBe(true); + expect(cache.epochKeys['ch-1'].intKey.rev).toBe(2); // newer from the backup + expect(cache.epochKeys['ch-1'].pubKey.rev).toBe(3); // newer here, kept + expect(cache.epochKeys['ch-2'].intKeyPending.keyId).toBe('i2'); + }); + + it('drops a pending re-key from the backup that the local keys already settled', async () => { + const { secureStorage, channelManager, cache } = makeDeps({ + epochKeys: { 'ch-1': { epochs: {}, currentEpoch: 1, intKey: { keyId: 'i2', keyHex: '0x2', rev: 2 } } } + }); + const data = { + epochKeys: { 'ch-1': { epochs: {}, currentEpoch: 1, intKeyPending: { keyId: 'i2', keyHex: '0x2', rev: 2 } } } + }; + + await importBackupData(data, { secureStorage, channelManager }); + + expect(cache.epochKeys['ch-1']).not.toHaveProperty('intKeyPending'); + expect(cache.epochKeys['ch-1'].intKey.keyId).toBe('i2'); + }); + it('returns an empty summary when data is null or storage is locked', async () => { const { secureStorage, channelManager } = makeDeps(); diff --git a/tests/unit/epochKeyManager.rekeyInteractions.test.js b/tests/unit/epochKeyManager.rekeyInteractions.test.js index 5e04442..170dff2 100644 --- a/tests/unit/epochKeyManager.rekeyInteractions.test.js +++ b/tests/unit/epochKeyManager.rekeyInteractions.test.js @@ -54,7 +54,7 @@ describe('resetting the interactions key', () => { const s = epochKeyManager.state.get(STREAM); expect(rev).toBe(2); - expect(calls[0].grants).toEqual({ next: s.intKey.address, old: OLD.address }); + expect(calls[0].grants).toEqual({ next: s.intKey.address, old: [OLD.address] }); expect(calls[1].published).toMatchObject({ t: 'pub_announce', k: 'i', keyId: s.intKey.keyId, rev: 2 }); expect(s.intKey.address).not.toBe(OLD.address); expect(s.intAnnounce.keyId).toBe(s.intKey.keyId); @@ -62,6 +62,7 @@ describe('resetting the interactions key', () => { it('announces nothing when the grants did not move', async () => { streamrController.rekeyInteractionsGrants.mockRejectedValueOnce(new Error('tx reverted')); + vi.spyOn(streamrController, 'rekeyGrantsState').mockResolvedValue({ next: [false, false], old: [true, true] }); await expect(epochKeyManager.rekeyInteractionsKey(channel)).rejects.toThrow('tx reverted'); @@ -83,7 +84,7 @@ describe('the interactions grants', () => { const previous = streamrController.client; streamrController.client = { setPermissions }; try { - await streamrController.rekeyInteractionsGrants(channel, '0xNEW', '0xOLD'); + await streamrController.rekeyInteractionsGrants(channel, '0xNEW', ['0xOLD']); } finally { streamrController.client = previous; } diff --git a/tests/unit/epochKeyManager.rekeyRecovery.test.js b/tests/unit/epochKeyManager.rekeyRecovery.test.js new file mode 100644 index 0000000..4a9cd76 --- /dev/null +++ b/tests/unit/epochKeyManager.rekeyRecovery.test.js @@ -0,0 +1,340 @@ +/** + * A re-key whose grant lands while the call fails must not cost the owner the + * key: the new key is written down before the grant is sent, and the chain + * decides which of the two keys survives, in the call or on the next open. + */ + +import { describe, it, expect, beforeEach, afterEach, vi } from 'vitest'; +import { ethers } from 'ethers'; + +globalThis.ethers = ethers; + +const { epochKeyManager } = await import('../../src/js/epochKeyManager.js'); +const { streamrController } = await import('../../src/js/streamr.js'); +const { authManager } = await import('../../src/js/auth.js'); +const { secureStorage } = await import('../../src/js/secureStorage.js'); + +const ADMIN = '0x' + 'aa'.repeat(20); +const GATE = '0x' + 'ab'.repeat(20); +const STREAM = `${ADMIN}/sealed-1`; +const channel = { + messageStreamId: STREAM, keysStreamId: `${ADMIN}/sealed-4`, + ephemeralStreamId: `${ADMIN}/sealed-2`, interactionsStreamId: `${ADMIN}/sealed-5`, + type: 'gated', gate: { address: GATE }, wireIdentity: 'sealed', members: [] +}; +const OLD_INT = { keyId: 'i1.k', keyHex: '0x' + '33'.repeat(32), address: '0x' + '33'.repeat(20), rev: 1 }; +const OLD_PUB = { keyId: 'p1.k', keyHex: '0x' + '44'.repeat(32), address: '0x' + '44'.repeat(20), rev: 1 }; +const PENDING_INT = { + keyId: 'i2.k', keyHex: '0x' + '55'.repeat(32), address: '0x' + '55'.repeat(20), rev: 2, + oldAddress: OLD_INT.address +}; +const HOUR = 60 * 60 * 1000; +const UNSETTLED = 'The interactions key reset has not finished. It retries when you open the channel.'; + +const announced = (key) => ({ + keyId: key.keyId, keyHash: '0xh', address: key.address, rev: key.rev, + publisher: ADMIN, timestamp: Date.now() +}); + +describe('recovering a re-key', () => { + let store; + let published; + let warnings; + + const record = () => store.get(STREAM); + const state = () => epochKeyManager.state.get(STREAM); + const restart = () => { + epochKeyManager.state.clear(); + epochKeyManager.loadPersistedState(STREAM); + return state(); + }; + const seed = (extra = {}) => { + store.set(STREAM, JSON.parse(JSON.stringify({ + epochs: {}, announces: {}, currentEpoch: 0, + intKey: OLD_INT, intAnnounce: announced(OLD_INT), + pubKey: OLD_PUB, pubAnnounce: announced(OLD_PUB), + ...extra + }))); + }; + + beforeEach(() => { + epochKeyManager.state.clear(); + store = new Map(); + published = []; + warnings = []; + vi.spyOn(secureStorage, 'getEpochKeys').mockImplementation((id) => store.get(id) || null); + vi.spyOn(secureStorage, 'setEpochKeys').mockImplementation(async (id, data) => { + store.set(id, JSON.parse(JSON.stringify(data))); + }); + vi.spyOn(authManager, 'getAddress').mockReturnValue(ADMIN); + vi.spyOn(epochKeyManager, '_ensureAnnounceRetained').mockResolvedValue(undefined); + vi.spyOn(streamrController, 'publishKeysMessage').mockImplementation(async (_, msg) => { + published.push(msg); + }); + epochKeyManager.setGateWarningHandler((_, warning) => warnings.push(warning)); + seed(); + }); + + afterEach(() => { + vi.restoreAllMocks(); + epochKeyManager.setGateWarningHandler(null); + epochKeyManager.state.clear(); + }); + + describe('in the reset call', () => { + it('writes the new key down, and hands it to sync, before the grant is sent', async () => { + let seenAtGrant = null; + const synced = []; + epochKeyManager.onKeysAdopted = (_, keyId) => synced.push(keyId); + vi.spyOn(streamrController, 'rekeyInteractionsGrants').mockImplementation(async (_, next) => { + seenAtGrant = { next, record: record(), synced: [...synced] }; + }); + + try { + await epochKeyManager.rekeyInteractionsKey(channel); + } finally { + epochKeyManager.onKeysAdopted = undefined; + } + + expect(seenAtGrant.synced).toEqual([seenAtGrant.record.intKeyPending.keyId]); + + expect(seenAtGrant.record.intKeyPending).toMatchObject({ + address: seenAtGrant.next, rev: 2, oldAddress: OLD_INT.address + }); + expect(seenAtGrant.record.intKey.keyId).toBe(OLD_INT.keyId); + expect(record().intKeyPending).toBeUndefined(); + expect(record().intKey.address).toBe(seenAtGrant.next); + }); + + it('keeps the new key when the grant landed but the call failed', async () => { + vi.spyOn(streamrController, 'rekeyInteractionsGrants').mockRejectedValue(new Error('receipt 503')); + vi.spyOn(streamrController, 'rekeyGrantsState').mockResolvedValue({ next: [true, true], old: [false, false] }); + + const rev = await epochKeyManager.rekeyInteractionsKey(channel); + + expect(rev).toBe(2); + expect(state().intKey.rev).toBe(2); + expect(published).toEqual([expect.objectContaining({ t: 'pub_announce', k: 'i', rev: 2, keyId: state().intKey.keyId })]); + expect(record().intKey.keyId).toBe(state().intKey.keyId); + expect(record().intKeyPending).toBeUndefined(); + }); + + it('keeps the old key, and the new one pending, when the grant never landed', async () => { + vi.spyOn(streamrController, 'rekeyInteractionsGrants').mockRejectedValue(new Error('tx reverted')); + vi.spyOn(streamrController, 'rekeyGrantsState').mockResolvedValue({ next: [false, false], old: [true, true] }); + + await expect(epochKeyManager.rekeyInteractionsKey(channel)).rejects.toThrow('tx reverted'); + + expect(state().intKey).toEqual(OLD_INT); + expect(record().intKeyPending.rev).toBe(2); + expect(published).toEqual([]); + }); + + it('keeps the new key pending when the chain cannot be read', async () => { + vi.spyOn(streamrController, 'rekeyInteractionsGrants').mockRejectedValue(new Error('receipt 503')); + vi.spyOn(streamrController, 'rekeyGrantsState').mockRejectedValue(new Error('rpc down')); + + await expect(epochKeyManager.rekeyInteractionsKey(channel)).rejects.toThrow('receipt 503'); + + expect(state().intKey).toEqual(OLD_INT); + expect(record().intKeyPending.rev).toBe(2); + }); + + it('revokes an unsettled earlier key too, and goes above its rev', async () => { + seed({ intKeyPending: { ...PENDING_INT, mintedAt: Date.now() } }); + restart(); + vi.spyOn(streamrController, 'rekeyGrantsState').mockResolvedValue({ next: [false, false], old: [true, true] }); + const grants = vi.spyOn(streamrController, 'rekeyInteractionsGrants').mockResolvedValue(undefined); + + const rev = await epochKeyManager.rekeyInteractionsKey(channel); + + expect(rev).toBe(3); + expect(grants.mock.calls[0][2]).toEqual([OLD_INT.address, PENDING_INT.address]); + }); + + it('refuses to reset over an unsettled key while the chain cannot be read', async () => { + seed({ intKeyPending: { ...PENDING_INT, mintedAt: Date.now() } }); + restart(); + vi.spyOn(streamrController, 'rekeyGrantsState').mockRejectedValue(new Error('rpc down')); + const grants = vi.spyOn(streamrController, 'rekeyInteractionsGrants'); + + await expect(epochKeyManager.rekeyInteractionsKey(channel)).rejects.toThrow('chain cannot be read'); + + expect(grants).not.toHaveBeenCalled(); + expect(record().intKeyPending.keyId).toBe(PENDING_INT.keyId); + }); + + it('recovers the publish key the same way', async () => { + vi.spyOn(streamrController, 'rekeySharedPublishGrants').mockRejectedValue(new Error('receipt 503')); + const read = vi.spyOn(streamrController, 'rekeyGrantsState').mockResolvedValue({ next: [true, true], old: [false, false] }); + + const rev = await epochKeyManager.rekeyPublishKey(channel); + + expect(rev).toBe(2); + expect(read.mock.calls[0][1]).toBe('pub'); + expect(state().pubKey.rev).toBe(2); + expect(published[0]).not.toHaveProperty('k'); + }); + }); + + describe('after a restart with a key pending', () => { + it('promotes it when the chain holds its grant', async () => { + seed({ intKeyPending: { ...PENDING_INT, mintedAt: Date.now() - 5 * HOUR } }); + const s = restart(); + vi.spyOn(streamrController, 'rekeyGrantsState').mockResolvedValue({ next: [true, true], old: [false, false] }); + + await epochKeyManager._settleRekeys(channel, s); + + expect(s.intKey.keyId).toBe(PENDING_INT.keyId); + expect(published).toEqual([expect.objectContaining({ k: 'i', keyId: PENDING_INT.keyId, rev: 2 })]); + expect(record().intKeyPending).toBeUndefined(); + expect(warnings).toEqual([]); + }); + + it('keeps it while the grant may still land', async () => { + seed({ intKeyPending: { ...PENDING_INT, mintedAt: Date.now() - HOUR / 2 } }); + const s = restart(); + vi.spyOn(streamrController, 'rekeyGrantsState').mockResolvedValue({ next: [false, false], old: [true, true] }); + + await epochKeyManager._settleRekeys(channel, s); + + expect(s.intKey.keyId).toBe(OLD_INT.keyId); + expect(record().intKeyPending.keyId).toBe(PENDING_INT.keyId); + expect(warnings).toEqual([]); + }); + + it('drops it once the grant clearly never landed', async () => { + seed({ intKeyPending: { ...PENDING_INT, mintedAt: Date.now() - 2 * HOUR } }); + const s = restart(); + vi.spyOn(streamrController, 'rekeyGrantsState').mockResolvedValue({ next: [false, false], old: [true, true] }); + + await epochKeyManager._settleRekeys(channel, s); + + expect(s.intKey.keyId).toBe(OLD_INT.keyId); + expect(record().intKeyPending).toBeUndefined(); + expect(published).toEqual([]); + }); + + it('keeps it and warns when the chain cannot be read', async () => { + seed({ intKeyPending: { ...PENDING_INT, mintedAt: Date.now() - 2 * HOUR } }); + const s = restart(); + vi.spyOn(streamrController, 'rekeyGrantsState').mockRejectedValue(new Error('rpc down')); + + await epochKeyManager._settleRekeys(channel, s); + + expect(record().intKeyPending.keyId).toBe(PENDING_INT.keyId); + expect(warnings).toEqual([UNSETTLED]); + }); + + it('keeps it and warns when neither key holds the grant', async () => { + seed({ intKeyPending: { ...PENDING_INT, mintedAt: Date.now() - 2 * HOUR } }); + const s = restart(); + vi.spyOn(streamrController, 'rekeyGrantsState').mockResolvedValue({ next: [false, false], old: [false, false] }); + + await epochKeyManager._settleRekeys(channel, s); + + expect(record().intKeyPending.keyId).toBe(PENDING_INT.keyId); + expect(warnings).toEqual([UNSETTLED]); + }); + + it('finishes a publish re-key whose grant landed on one stream only', async () => { + const PENDING_PUB = { keyId: 'p2.k', keyHex: '0x' + '99'.repeat(32), address: '0x' + '99'.repeat(20), rev: 2 }; + seed({ pubKeyPending: { ...PENDING_PUB, oldAddress: OLD_PUB.address, mintedAt: Date.now() } }); + const s = restart(); + vi.spyOn(streamrController, 'rekeyGrantsState').mockResolvedValue({ + streamIds: [channel.messageStreamId, channel.ephemeralStreamId], next: [true, false], old: [false, true] + }); + const grants = vi.spyOn(streamrController, 'rekeySharedPublishGrants').mockResolvedValue(undefined); + + await epochKeyManager._settleRekeys(channel, s); + + expect(grants).toHaveBeenCalledWith(channel, PENDING_PUB.address, [OLD_PUB.address], [channel.ephemeralStreamId]); + expect(s.pubKey.keyId).toBe(PENDING_PUB.keyId); + expect(record().pubKeyPending).toBeUndefined(); + expect(warnings).toEqual([]); + }); + + it('keeps a half-landed publish re-key pending, and warns, when the missing grant fails again', async () => { + const PENDING_PUB = { keyId: 'p2.k', keyHex: '0x' + '99'.repeat(32), address: '0x' + '99'.repeat(20), rev: 2 }; + seed({ pubKeyPending: { ...PENDING_PUB, oldAddress: OLD_PUB.address, mintedAt: Date.now() } }); + const s = restart(); + vi.spyOn(streamrController, 'rekeyGrantsState').mockResolvedValue({ + streamIds: [channel.messageStreamId, channel.ephemeralStreamId], next: [true, false], old: [false, true] + }); + vi.spyOn(streamrController, 'rekeySharedPublishGrants').mockRejectedValue(new Error('rpc down')); + + await epochKeyManager._settleRekeys(channel, s); + + expect(s.pubKey.keyId).toBe(OLD_PUB.keyId); + expect(record().pubKeyPending.keyId).toBe(PENDING_PUB.keyId); + expect(warnings).toEqual(['The publish key reset has not finished. It retries when you open the channel.']); + }); + + it('adopts it without the chain when another device already announced it', async () => { + seed({ + intKeyPending: { ...PENDING_INT, mintedAt: Date.now() }, + intAnnounce: announced(PENDING_INT) + }); + const s = restart(); + const read = vi.spyOn(streamrController, 'rekeyGrantsState'); + + await epochKeyManager._settleRekeys(channel, s); + + expect(read).not.toHaveBeenCalled(); + expect(s.intKey.keyId).toBe(PENDING_INT.keyId); + expect(published).toEqual([]); + }); + + it('drops it without the chain when a later re-key superseded it', async () => { + const later = { keyId: 'i3.x', address: '0x' + '66'.repeat(20), rev: 3 }; + seed({ intKeyPending: { ...PENDING_INT, mintedAt: Date.now() }, intAnnounce: announced(later) }); + const s = restart(); + const read = vi.spyOn(streamrController, 'rekeyGrantsState'); + + await epochKeyManager._settleRekeys(channel, s); + + expect(read).not.toHaveBeenCalled(); + expect(record().intKeyPending).toBeUndefined(); + }); + + it('settles it when the owner opens the channel', async () => { + seed({ + epochs: { 'e1.k': { keyHex: '0x' + '77'.repeat(32), keyHash: '0xe', epoch: 1 } }, + announces: { 1: { keyId: 'e1.k', keyHash: '0xe', publisher: ADMIN, timestamp: Date.now(), validFrom: Date.now() } }, + currentEpoch: 1, + intKeyPending: { ...PENDING_INT, mintedAt: Date.now() } + }); + vi.spyOn(streamrController, 'resendKeysMessages').mockResolvedValue([]); + vi.spyOn(epochKeyManager, '_maybeReannounceAging').mockResolvedValue(undefined); + vi.spyOn(epochKeyManager, '_armScheduledRotation').mockImplementation(() => {}); + vi.spyOn(streamrController, 'rekeyGrantsState').mockResolvedValue({ next: [true, true], old: [false, false] }); + + await epochKeyManager.ensureChannelKeys(channel); + + expect(state().intKey.keyId).toBe(PENDING_INT.keyId); + expect(published.filter((m) => m.k === 'i')).toEqual([expect.objectContaining({ keyId: PENDING_INT.keyId, rev: 2 })]); + }); + }); + +}); + +describe('reading a re-key off the chain', () => { + it('asks for PUBLISH of both keys on each stream the shared key writes to', async () => { + const hasPermission = vi.fn(async ({ streamId, userId }) => + userId === '0xnew' && streamId.endsWith('-5')); + const previous = streamrController.client; + streamrController.client = { hasPermission }; + try { + const grants = await streamrController.rekeyGrantsState(channel, 'int', '0xNEW', '0xOLD'); + expect(grants).toEqual({ + streamIds: [channel.interactionsStreamId, channel.ephemeralStreamId], + next: [true, false], old: [false, false] + }); + } finally { + streamrController.client = previous; + } + expect(hasPermission.mock.calls.map(([q]) => [q.streamId.slice(-2), q.userId, q.permission])) + .toEqual([['-5', '0xnew', 'publish'], ['-5', '0xold', 'publish'], + ['-2', '0xnew', 'publish'], ['-2', '0xold', 'publish']]); + }); +}); diff --git a/tests/unit/epochKeyManager.unannouncedSharedKey.test.js b/tests/unit/epochKeyManager.unannouncedSharedKey.test.js new file mode 100644 index 0000000..5f038e5 --- /dev/null +++ b/tests/unit/epochKeyManager.unannouncedSharedKey.test.js @@ -0,0 +1,78 @@ +/** + * A held shared key whose rev no announce carries yet must be announced on the + * next open, however fresh the announce of the older rev is. + */ + +import { describe, it, expect, beforeEach, afterEach, vi } from 'vitest'; +import { ethers } from 'ethers'; + +globalThis.ethers = ethers; + +const { epochKeyManager } = await import('../../src/js/epochKeyManager.js'); +const { streamrController } = await import('../../src/js/streamr.js'); +const { authManager } = await import('../../src/js/auth.js'); + +const ADMIN = '0x' + 'aa'.repeat(20); +const STREAM = `${ADMIN}/sealed-1`; +const channel = { + messageStreamId: STREAM, keysStreamId: `${ADMIN}/sealed-4`, + ephemeralStreamId: `${ADMIN}/sealed-2`, interactionsStreamId: `${ADMIN}/sealed-5`, + type: 'gated', gate: { address: '0x' + 'ab'.repeat(20) }, wireIdentity: 'sealed' +}; +const key = (prefix, rev, byte) => ({ + keyId: `${prefix}${rev}.k`, keyHex: '0x' + byte.repeat(32), address: '0x' + byte.repeat(20), rev +}); +const announced = (k) => ({ keyId: k.keyId, keyHash: '0xh', address: k.address, rev: k.rev, publisher: ADMIN, timestamp: Date.now() }); + +describe('the shared-key announce self-heal', () => { + let published; + let s; + + beforeEach(() => { + epochKeyManager.state.clear(); + published = []; + vi.spyOn(authManager, 'getAddress').mockReturnValue(ADMIN); + vi.spyOn(epochKeyManager, '_persist').mockResolvedValue(undefined); + vi.spyOn(epochKeyManager, '_ensureAnnounceRetained').mockResolvedValue(undefined); + vi.spyOn(streamrController, 'publishKeysMessage').mockImplementation(async (_, msg) => { + published.push(msg); + }); + s = epochKeyManager._getState(STREAM); + s.loaded = true; + s.intAnnounce = announced(key('i', 1, '33')); + s.pubAnnounce = announced(key('p', 1, '44')); + s.intAnnounceFreshness = Date.now(); + s.pubAnnounceFreshness = Date.now(); + }); + + afterEach(() => { + vi.restoreAllMocks(); + epochKeyManager.state.clear(); + }); + + it('announces an interactions key above the announced rev', async () => { + s.intKey = key('i', 2, '55'); + + await epochKeyManager._maybeAnnounceInteractions(channel, s); + + expect(published).toEqual([expect.objectContaining({ k: 'i', keyId: 'i2.k', rev: 2 })]); + }); + + it('announces a publish key above the announced rev', async () => { + s.pubKey = key('p', 2, '88'); + + await epochKeyManager._maybeAnnouncePub(channel, s); + + expect(published).toEqual([expect.objectContaining({ keyId: 'p2.k', rev: 2 })]); + }); + + it('stays quiet while the held key is the one freshly announced', async () => { + s.intKey = key('i', 1, '33'); + s.pubKey = key('p', 1, '44'); + + await epochKeyManager._maybeAnnounceInteractions(channel, s); + await epochKeyManager._maybeAnnouncePub(channel, s); + + expect(published).toEqual([]); + }); +}); diff --git a/tests/unit/syncMerge.vectors.test.js b/tests/unit/syncMerge.vectors.test.js index c8323b5..36dafe2 100644 --- a/tests/unit/syncMerge.vectors.test.js +++ b/tests/unit/syncMerge.vectors.test.js @@ -50,6 +50,12 @@ describe('sync merge parity vectors', () => { }); } + for (const v of vectors.keys) { + it(v.what, () => { + expect(mergeState(v.base, v.incoming).epochKeys).toEqual(v.expected.epochKeys); + }); + } + for (const v of vectors.publish.cases) { it(v.what, () => { const from = applyPatch(vectors.publish.base, v.basePatch); diff --git a/tests/vectors/gen_sync_merge_vectors.mjs b/tests/vectors/gen_sync_merge_vectors.mjs index 1322913..8e0eaad 100644 --- a/tests/vectors/gen_sync_merge_vectors.mjs +++ b/tests/vectors/gen_sync_merge_vectors.mjs @@ -9,8 +9,8 @@ // The vectors fix the slice outcome of one merge step (base = this device, // incoming = a remote snapshot), the stamping of unstamped values before a // state leaves the device, how sent DMs carry deletions and edits, how two -// copies of a channel record merge, and which changes to a state are news -// worth a push. +// copies of a channel record merge, how shared keys and pending re-keys +// merge, and which changes to a state are news worth a push. import { mergeState, stampedSliceTs } from '../../src/js/syncMerge.js'; import { syncStateKey } from '../../src/js/syncStateKey.js'; @@ -83,7 +83,9 @@ const PUBLISH_BASE = { pendingRequests: { r1: { fromEpoch: 1, sentAt: 100 } }, helloEpochs: [1], helloName: 'Bob', helloTs: 100, seenRequesters: [A(1)], pubKey: { keyId: 'p1', keyHex: '0xbb', rev: 1 }, pubAnnounce: { keyId: 'p1', rev: 1, timestamp: 100 }, - intKey: { keyId: 'i1', keyHex: '0xcc', rev: 1 }, intAnnounce: { keyId: 'i1', rev: 1, timestamp: 100 } + intKey: { keyId: 'i1', keyHex: '0xcc', rev: 1 }, intAnnounce: { keyId: 'i1', rev: 1, timestamp: 100 }, + pubKeyPending: { keyId: 'p2', keyHex: '0xdd', rev: 2, oldAddress: A(13), mintedAt: 100 }, + intKeyPending: { keyId: 'i2', keyHex: '0xee', rev: 2, oldAddress: A(14), mintedAt: 100 } } }, blockedPeers: [A(6)], @@ -104,7 +106,7 @@ const NEWS = { channel: ['name', 'type', 'createdAt', 'joinedAt', 'storageDays', 'accessSnapshot', 'gate', 'wireIdentity', 'createdBy', 'password', 'members', 'rotatedForNoAccess', 'knownBanned', 'storageEnabled', 'adminStorageDays', 'keysStorageDays', 'interactionsStorageDays', 'exposure', 'description', 'language', 'category', 'metaUpdatedAt', 'readOnly', 'writeOnly', 'classification', 'peerAddress', 'fieldTs'], - epochKeys: ['epochs', 'currentEpoch', 'pubKey', 'intKey'] + epochKeys: ['epochs', 'currentEpoch', 'pubKey', 'intKey', 'pubKeyPending', 'intKeyPending'] }; const mutate = (value) => { if (typeof value === 'string') return `${value}x`; @@ -179,6 +181,25 @@ const sent = (what, base, incoming) => { return { what, base, incoming, expected: { sentMessages: merged.sentMessages, sentDeletedAt: merged.sentDeletedAt } }; }; +// Shared keys: the higher rev wins. A pending re-key travels the same way and +// is dropped once a key or announce at its rev or above shows it was settled. +const HELD = { + epochs: {}, announces: {}, currentEpoch: 1, + pubKey: { keyId: 'p1', keyHex: '0xbb', rev: 1 }, pubAnnounce: { keyId: 'p1', rev: 1, timestamp: 100 }, + intKey: { keyId: 'i1', keyHex: '0xcc', rev: 1 }, intAnnounce: { keyId: 'i1', rev: 1, timestamp: 100 } +}; +const pending = (prefix, rev) => ({ keyId: `${prefix}${rev}`, keyHex: `0x${rev}${rev}`, rev, oldAddress: A(15), mintedAt: 100 * rev }); +const keys = (what, base, incoming, survivors) => { + const state = (slice) => ({ channels: [record()], epochKeys: { [CH]: slice } }); + const epochKeys = mergeState(state(base), state(incoming)).epochKeys; + for (const slot of ['pubKeyPending', 'intKeyPending']) { + if ((epochKeys[CH][slot]?.keyId ?? null) !== (survivors[slot] ?? null)) { + throw new Error(`keys vector "${what}": expected ${slot} ${survivors[slot] ?? 'gone'}`); + } + } + return { what, base: state(base), incoming: state(incoming), expected: { epochKeys } }; +}; + console.log(JSON.stringify({ merge: [ merge('an unstamped empty snapshot never erases unstamped values', RESTORED, EMPTY), @@ -271,5 +292,19 @@ console.log(JSON.stringify({ { channels: [record({ joinedAt: 7000 })] }, { channels: [], channelsLeftAt: { [CH]: 5000 } }) ], + keys: [ + keys('a pending re-key held here is kept', { ...HELD, intKeyPending: pending('i', 2) }, HELD, + { intKeyPending: 'i2' }), + keys('a pending re-key from another device arrives', HELD, { ...HELD, intKeyPending: pending('i', 2) }, + { intKeyPending: 'i2' }), + keys('the later of two pending re-keys wins', { ...HELD, intKeyPending: pending('i', 2) }, + { ...HELD, intKeyPending: pending('i', 3) }, { intKeyPending: 'i3' }), + keys('a pending re-key another device promoted is dropped', { ...HELD, intKeyPending: pending('i', 2) }, + { ...HELD, intKey: { keyId: 'i2', keyHex: '0x22', rev: 2 }, intAnnounce: { keyId: 'i2', rev: 2, timestamp: 200 } }, {}), + keys('a pending re-key a later announce superseded is dropped', { ...HELD, intKeyPending: pending('i', 2) }, + { ...HELD, intAnnounce: { keyId: 'i3', rev: 3, timestamp: 300 } }, {}), + keys('a pending publish key travels like the interactions one', HELD, { ...HELD, pubKeyPending: pending('p', 2) }, + { pubKeyPending: 'p2' }) + ], publish: { base: PUBLISH_BASE, cases: publishCases } }, null, 2));