User Profile Access Controller

packages/backend/package.json

@@ -22,11 +22,11 @@
     "lint": "npm run lint:no-fix -- --fix",
     "lint-ci": "npm run lint:no-fix",
     "lint-staged": "lint-staged --no-stash",
-    "test-nest": "cross-env NODE_OPTIONS=\"--experimental-vm-modules\" DEBUG=ipfs:*,backend:* node_modules/jest/bin/jest.js --detectOpenHandles --forceExit   ./src/nest/**/*.spec.ts",
-    "test": "cross-env NODE_OPTIONS=\"--experimental-vm-modules\" DEBUG=ipfs:*,backend:* jest --runInBand --verbose --testPathIgnorePatterns=\".src/(!?nodeTest*)|(.node_modules*)\" --detectOpenHandles --forceExit",
-    "test-ci": "cross-env NODE_OPTIONS=\"--experimental-vm-modules\" jest --runInBand --colors --ci --silent --verbose --detectOpenHandles --forceExit --testPathIgnorePatterns \"/node_modules/\" --testPathIgnorePatterns \"/dist/\" --testPathIgnorePatterns \"src/nest/.*\\.tor\\.spec\\.(t|j)s$\" --testPathIgnorePatterns \"src/nest/ipfs-file-manager/big-files\\.long\\.spec\\.ts$\"",
-    "test-ci-tor": "cross-env NODE_OPTIONS=\"--experimental-vm-modules\" jest --runInBand --colors --ci --verbose --detectOpenHandles --forceExit ./src/nest/**/*.tor.spec.ts",
-    "test-ci-long-running": "cross-env DEBUG=backend:* NODE_OPTIONS=\"--experimental-vm-modules\" jest --colors --ci --verbose --detectOpenHandles --forceExit ./src/nest/**/*.long.spec.ts",
+    "test-nest": "cross-env NODE_OPTIONS=\"--experimental-vm-modules\" DEBUG=ipfs:*,backend:* node_modules/jest/bin/jest.js --forceExit   ./src/nest/**/*.spec.ts",
+    "test": "cross-env NODE_OPTIONS=\"--experimental-vm-modules\" DEBUG=ipfs:*,backend:* jest --runInBand --verbose --testPathIgnorePatterns=\".src/(!?nodeTest*)|(.node_modules*)\" --forceExit",
+    "test-ci": "cross-env NODE_OPTIONS=\"--experimental-vm-modules\" jest --runInBand --colors --ci --silent --verbose --forceExit --testPathIgnorePatterns \"/node_modules/\" --testPathIgnorePatterns \"/dist/\" --testPathIgnorePatterns \"src/nest/.*\\.tor\\.spec\\.(t|j)s$\" --testPathIgnorePatterns \"src/nest/ipfs-file-manager/big-files\\.long\\.spec\\.ts$\"",
+    "test-ci-tor": "cross-env NODE_OPTIONS=\"--experimental-vm-modules\" jest --runInBand --colors --ci --verbose --forceExit ./src/nest/**/*.tor.spec.ts",
+    "test-ci-long-running": "cross-env DEBUG=backend:* NODE_OPTIONS=\"--experimental-vm-modules\" jest --colors --ci --verbose --forceExit ./src/nest/**/*.long.spec.ts",
     "test-connect": "cross-env NODE_OPTIONS=\"--experimental-vm-modules\" DEBUG='libp2p:websockets*' jest ./src/nodeTest/* --verbose",
     "test-connect-ci": "cross-env NODE_OPTIONS=\"--experimental-vm-modules\" jest ./src/nodeTest/* --colors --ci --silent --verbose",
     "test-replication-no-tor": "cross-env NODE_OPTIONS=\"--experimental-vm-modules\" ts-node -v && cross-env DEBUG='backend:dbSnap*,backend:localTest*' ts-node src/nodeTest/testReplicate.ts --nodesCount 1 --timeThreshold 200 --entriesCount 1000 --no-useTor",

packages/backend/src/nest/libp2p/integration-tests/userProfile-sync.spec.ts

@@ -1,7 +1,8 @@
 import { jest } from '@jest/globals'
 import { headsAreEqual, Hash } from '@localfirst/crdx'
-import { Test, TestingModule } from '@nestjs/testing'
+import { TestingModule } from '@nestjs/testing'
 import waitForExpect from 'wait-for-expect'
+import { Entry, type LogEntry } from '@orbitdb/core'
 
 import { getBaseTypesFactory } from '@quiet/state-manager'
 import { FactoryGirl } from 'factory-girl'
@@ -15,6 +16,7 @@ import { spawnTestModules, spawnLibp2pInstancesInMemory } from '../../common/tes
 import { UserProfile } from '@quiet/types'
 import { createLogger } from '../../common/logger'
 import { Libp2pEvents } from '../libp2p.types'
+import { EncryptedAndSignedPayload } from '../../auth/services/crypto/types'
 
 const logger = createLogger('UserProfile-sync')
 const N_PEERS = 3
@@ -38,6 +40,25 @@ describe('UserProfileStore OrbitDB Sync', () => {
     await userProfileStores[i].init()
   }
 
+  const getLogEntries = async (store: UserProfileStore): Promise<Array<LogEntry<EncryptedAndSignedPayload>>> => {
+    const entries: Array<LogEntry<EncryptedAndSignedPayload>> = []
+    for await (const entry of store.getStore().log.traverse()) {
+      entries.push(entry as LogEntry<EncryptedAndSignedPayload>)
+    }
+    return entries
+  }
+
+  const expectAliceAndBobProfiles = async (aliceNickname: string, bobNickname: string) => {
+    const aliceProfiles = await userProfileStores[0].getUserProfiles()
+    const bobProfiles = await userProfileStores[1].getUserProfiles()
+    expect(aliceProfiles.length).toBe(2)
+    expect(bobProfiles.length).toBe(2)
+    expect(aliceProfiles.find(p => p.userId === userIds[0])?.nickname).toBe(aliceNickname)
+    expect(aliceProfiles.find(p => p.userId === userIds[1])?.nickname).toBe(bobNickname)
+    expect(bobProfiles.find(p => p.userId === userIds[0])?.nickname).toBe(aliceNickname)
+    expect(bobProfiles.find(p => p.userId === userIds[1])?.nickname).toBe(bobNickname)
+  }
+
   beforeAll(async () => {
     factory = await getBaseTypesFactory()
     modules = await spawnTestModules(N_PEERS)
@@ -84,11 +105,15 @@ describe('UserProfileStore OrbitDB Sync', () => {
 
   afterAll(async () => {
     for (let i = 0; i < N_PEERS; i++) {
-      await userProfileStores[i].close()
-      await orbitDbServices[i].stop()
-      await ipfsServices[i].stop()
-      await libp2pServices[i].close()
-      await localDbServices[i].close()
+      await userProfileStores[i]?.close()
+      await orbitDbServices[i]?.stop()
+      await libp2pServices[i]?.close(false)
+      await ipfsServices[i]?.stop()
+      await libp2pServices[i]?.closeDatastore()
+      await localDbServices[i]?.close()
+    }
+    for (const module of modules) {
+      await module.close()
     }
   })
 
@@ -175,42 +200,78 @@ describe('UserProfileStore OrbitDB Sync', () => {
     )
   })
 
-  it("peer cannot update the other peer's userProfile", async () => {
-    // Provide a valid base64 photo
-    // Bob tries to update Alice's profile
-    await userProfileStores[1].setEntry(userIds[0], bobProfile)
+  it("rejects adversarial attempts to overwrite another peer's userProfile", async () => {
+    const aliceStore = userProfileStores[0].getStore()
+    const bobStore = userProfileStores[1].getStore()
+    const aliceEncryptedProfile = (await userProfileStores[0].getEncryptedEntries([userIds[0]]))[userIds[0]]
+    if (aliceEncryptedProfile == null) {
+      throw new Error("Alice's encrypted profile was not stored before adversarial overwrite attempts")
+    }
 
-    await waitForExpect(
-      async () => {
-        // expect that both Alice and Bob have the maliciousProfile in their log (this is a bug)
-        const aliceAllEntries: any[] = []
-        for await (const entry of userProfileStores[0].getStore().log.traverse()) {
-          aliceAllEntries.push(entry)
-        }
-        const bobAllEntries: any[] = []
-        for await (const entry of userProfileStores[1].getStore().log.traverse()) {
-          bobAllEntries.push(entry)
-        }
-        // TODO: after implementating the access control and identities, this should be 2
-        expect(aliceAllEntries.length).toBe(3)
-        expect(bobAllEntries.length).toBe(3)
+    // Bob tries to write his own signed profile under Alice's key via the store API.
+    await userProfileStores[1].setEntry(userIds[0], {
+      ...bobProfile,
+      nickname: 'Bob as Alice',
+    })
+
+    // Bob signs profile contents that claim Alice's userId and tries to store them under Alice's key.
+    const bobSignedAliceProfile = await userProfileStores[1].encryptEntry({
+      ...aliceProfile,
+      userId: userIds[0],
+      nickname: 'Alice overwritten by Bob',
+    })
+    await bobStore.put(userIds[0], bobSignedAliceProfile)
+
+    // Bob replays Alice's legitimate encrypted payload from his own OrbitDB writer identity.
+    await bobStore.put(userIds[0], aliceEncryptedProfile)
+
+    // Bob serves a spoofed raw OrbitDB entry to Alice over the OrbitDB sync protocol.
+    const currentBobHeadHashes = (await bobStore.log.heads()).map(
+      (entry: LogEntry<EncryptedAndSignedPayload>) => entry.hash
+    )
+    const spoofedSyncEntry = await Entry.create<EncryptedAndSignedPayload>(
+      bobStore.identity,
+      bobStore.log.id,
+      {
+        op: 'PUT',
+        key: userIds[0],
+        value: aliceEncryptedProfile,
       },
-      5000,
-      100
+      undefined,
+      currentBobHeadHashes
     )
+    const aliceSyncErrors: Error[] = []
+    const onAliceSyncError = (error: Error) => {
+      aliceSyncErrors.push(error)
+    }
+    aliceStore.events.on('error', onAliceSyncError)
+    try {
+      await bobStore.sync.add(spoofedSyncEntry)
+      await waitForExpect(
+        async () => {
+          expect(
+            aliceSyncErrors.some(error => {
+              const isAccessDenied = error.message.includes('Could not append entry')
+              const isSpoofedWriter = error.message.includes(spoofedSyncEntry.identity)
+              return isAccessDenied && isSpoofedWriter
+            })
+          ).toBe(true)
+        },
+        5000,
+        100
+      )
+    } finally {
+      aliceStore.events.off('error', onAliceSyncError)
+    }
 
-    // Alice's profile remains unchanged in the index
     await waitForExpect(
       async () => {
-        const aliceProfiles = await userProfileStores[0].getUserProfiles()
-        const bobProfiles = await userProfileStores[1].getUserProfiles()
-        expect(aliceProfiles.length).toBe(2)
-        expect(bobProfiles.length).toBe(2)
-        expect(aliceProfiles.find(p => p.userId === userIds[1])?.nickname).toBe('Bob')
-        expect(bobProfiles.find(p => p.userId === userIds[0])?.nickname).toBe('Alice')
+        expect(await getLogEntries(userProfileStores[0])).toHaveLength(2)
+        expect(await getLogEntries(userProfileStores[1])).toHaveLength(2)
+        await expectAliceAndBobProfiles('Alice', 'Bob')
       },
       5000,
-      1000
+      100
     )
   })
 
@@ -224,18 +285,8 @@ describe('UserProfileStore OrbitDB Sync', () => {
     // Wait for sync
     await waitForExpect(
       async () => {
-        // expect that both Alice and Bob have the maliciousProfile in their log (this is a bug)
-        const aliceAllEntries: any[] = []
-        for await (const entry of userProfileStores[0].getStore().log.traverse()) {
-          aliceAllEntries.push(entry)
-        }
-        const bobAllEntries: any[] = []
-        for await (const entry of userProfileStores[1].getStore().log.traverse()) {
-          bobAllEntries.push(entry)
-        }
-        // TODO: after implementating the access control and identities, this should be 2
-        expect(aliceAllEntries.length).toBe(5) // 2 updates + 2 initial profiles + 1 malicious profile
-        expect(bobAllEntries.length).toBe(5)
+        expect(await getLogEntries(userProfileStores[0])).toHaveLength(4)
+        expect(await getLogEntries(userProfileStores[1])).toHaveLength(4)
       },
       5000,
       100

packages/backend/src/nest/storage/orbitDb/ipfsBlockStorage.ts

@@ -75,9 +75,11 @@ const IPFSBlockStorage = async ({ ipfs, pin, timeout }: IPFSBlockStorageParams =
     } else {
       const abortController = new AbortController()
       const timer = setTimeout(() => abortController.abort(), timeout ?? DefaultTimeout)
-      const block = await ipfs.blockstore.get(cid, abortController)
-      clearTimeout(timer)
-      return block
+      try {
+        return await ipfs.blockstore.get(cid, abortController)
+      } finally {
+        clearTimeout(timer)
+      }
     }
   }
 

packages/backend/src/nest/storage/orbitDb/orbitDb.service.spec.ts

@@ -253,4 +253,33 @@ describe('OrbitDbService', () => {
     await orbitDbService.create(ipfsService.ipfsInstance!)
     expect(OrbitDbService.events.listeners('update')).toContain(updateListener)
   })
+
+  it('ingests concurrent heads even when their Lamport clock times differ', async () => {
+    const firstHead = {
+      id: 'concurrent-log',
+      hash: 'zdpuAn4pdCWF7HEhYqaW45woXfTqWn4ccG24JBHbf3sDR1XHK',
+      bytes: new Uint8Array([1]),
+      next: [],
+      clock: { time: 1 },
+    } as unknown as LogEntry
+    const ancestor = {
+      id: 'concurrent-log',
+      hash: 'zdpuAkTACgJnmr667GAafc5YQQYkYZHohonEhG2U9N9Q7WzmU',
+      bytes: new Uint8Array([2]),
+      next: [],
+      clock: { time: 2 },
+    } as unknown as LogEntry
+    const secondHead = {
+      id: 'concurrent-log',
+      hash: 'zdpuB3HUfW7TszueRD7X5GsbRDp2Xk8hSfvyn5SoEhB1zNihi',
+      bytes: new Uint8Array([3]),
+      next: [ancestor.hash],
+      clock: { time: 3 },
+    } as unknown as LogEntry
+    const joinHeadsSpy = jest.spyOn(orbitDbService as any, 'joinHeads').mockResolvedValue(undefined)
+
+    await orbitDbService.ingestEntries([firstHead, ancestor, secondHead])
+
+    expect(joinHeadsSpy).toHaveBeenCalledWith('concurrent-log', [firstHead, secondHead])
+  })
 })

packages/backend/src/nest/storage/orbitDb/orbitDb.service.ts

@@ -8,6 +8,7 @@ import { createLogger } from '../../common/logger'
 import { logEntryToLogUpdate, posixJoin } from './util'
 import { MessagesAccessController } from '../channels/messages/orbitdb/MessagesAccessController'
 import { ChannelMetadataAccessController } from '../channels/orbitdb/ChannelMetadataAccessController'
+import { UserProfileAccessController } from '../userProfile/UserProfileAccessController'
 import {
   createOrbitDB,
   type OrbitDBType,
@@ -171,7 +172,8 @@ export class OrbitDbService {
     private readonly sigChainService: SigChainService,
     private readonly lfaIdentities: LFAIdentities,
     private readonly messagesAccessController: MessagesAccessController,
-    private readonly channelMetadataAccessController: ChannelMetadataAccessController
+    private readonly channelMetadataAccessController: ChannelMetadataAccessController,
+    private readonly userProfileAccessController: UserProfileAccessController
   ) {
     this.attachOrbitDbUpdateListener()
   }
@@ -220,6 +222,12 @@ export class OrbitDbService {
         sigchainService: this.sigChainService,
       }) as any
     )
+    orbitDbUseAccessController(
+      this.userProfileAccessController.createAccessControllerFunc({
+        write: ['*'],
+        sigchainService: this.sigChainService,
+      }) as any
+    )
 
     /**
      * This overrides the built-in identity system to use our custom LFA-based identity service
@@ -372,7 +380,7 @@ export class OrbitDbService {
     if (this.orbitDbInstance == undefined) {
       throw new Error('OrbitDB instance is not initialized. Call create() first.')
     }
-    const newHeads: Map<string, LogEntry[]> = new Map()
+    const entriesByLog: Map<string, LogEntry[]> = new Map()
     for (const entry of entries) {
       const cid = CID.parse(entry.hash, base58btc)
       await this.orbitDbInstance.ipfs.blockstore.put(cid, entry.bytes)
@@ -384,20 +392,21 @@ export class OrbitDbService {
         }
       }
 
-      if (!newHeads.has(entry.id)) {
-        newHeads.set(entry.id, [entry])
-        continue
-      }
-      const currentMaxHead = newHeads.get(entry.id)?.[0]
-      if (currentMaxHead?.clock && currentMaxHead.clock.time < entry.clock.time) {
-        newHeads.set(entry.id, [entry])
-      } else if (currentMaxHead?.clock && currentMaxHead.clock.time === entry.clock.time) {
-        newHeads.get(entry.id)?.push(entry)
-      }
+      const logEntries = entriesByLog.get(entry.id) ?? []
+      logEntries.push(entry)
+      entriesByLog.set(entry.id, logEntries)
     }
 
+    // Lamport clock order does not imply ancestry: concurrent branches can have
+    // different clock times. Keep every entry that is not referenced by another
+    // entry in this batch so independent heads are not dropped during QSS pulls.
+    const newHeads = Array.from(entriesByLog.entries()).map(([id, logEntries]) => {
+      const referencedHashes = new Set(logEntries.flatMap(entry => entry.next))
+      return [id, logEntries.filter(entry => !referencedHashes.has(entry.hash))] as const
+    })
+
     // For each id, try to join heads (async, using joinQueue)
-    const joinAll = Array.from(newHeads.entries()).map(([id, heads]) => this.joinHeads(id, heads))
+    const joinAll = newHeads.map(([id, heads]) => this.joinHeads(id, heads))
     await Promise.all(joinAll)
   }
 

packages/backend/src/nest/storage/orbitDb/orbitdb.module.ts

@@ -16,6 +16,7 @@ import { NotificationTokensStore } from '../notifications/notificationTokens.sto
 import { MessagesAccessController } from '../channels/messages/orbitdb/MessagesAccessController'
 import { PrivateMessagesAccessController } from '../channels/messages/orbitdb/PrivateMessagesAccessController'
 import { ChannelMetadataAccessController } from '../channels/orbitdb/ChannelMetadataAccessController'
+import { UserProfileAccessController } from '../userProfile/UserProfileAccessController'
 
 @Module({
   imports: [
@@ -38,6 +39,7 @@ import { ChannelMetadataAccessController } from '../channels/orbitdb/ChannelMeta
     MessagesAccessController,
     PrivateMessagesAccessController,
     ChannelMetadataAccessController,
+    UserProfileAccessController,
   ],
   exports: [
     OrbitDbService,
@@ -52,6 +54,7 @@ import { ChannelMetadataAccessController } from '../channels/orbitdb/ChannelMeta
     MessagesAccessController,
     PrivateMessagesAccessController,
     ChannelMetadataAccessController,
+    UserProfileAccessController,
   ],
 })
 export class OrbitDbModule {}

packages/backend/src/nest/storage/storage.module.ts

@@ -6,9 +6,18 @@ import { IpfsModule } from '../ipfs/ipfs.module'
 import { SigChainModule } from '../auth/sigchain.service.module'
 import { OrbitDbModule } from './orbitDb/orbitdb.module'
 import { CommonModule } from '../common/common.module'
+import { SocketModule } from '../socket/socket.module'
 
 @Module({
-  imports: [CommonModule, LocalDbModule, IpfsModule, IpfsFileManagerModule, SigChainModule, OrbitDbModule],
+  imports: [
+    CommonModule,
+    LocalDbModule,
+    IpfsModule,
+    IpfsFileManagerModule,
+    SigChainModule,
+    OrbitDbModule,
+    SocketModule,
+  ],
   providers: [StorageService],
   exports: [StorageService],
 })