diff --git a/CHANGELOG.md b/CHANGELOG.md index d9213a14f1..95d77f053f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -18,6 +18,7 @@ # Refactorings: +* Refactor registration service, replace promise waiting mechanism around certificate requests and help prevent duplicate username registration * Removed SAVE_OWNER_CERTIFICATE event. * Removed registrar reminders and rename LAUNCH_REGISTRAR. * Removed unused SEND_USER_CERTIFICATE event. diff --git a/packages/backend/src/nest/connections-manager/connections-manager.service.ts b/packages/backend/src/nest/connections-manager/connections-manager.service.ts index d99fd25f30..eac7095a4f 100644 --- a/packages/backend/src/nest/connections-manager/connections-manager.service.ts +++ b/packages/backend/src/nest/connections-manager/connections-manager.service.ts @@ -41,7 +41,7 @@ import { PeerId as PeerIdType, SaveCSRPayload, CommunityMetadata, - PermsData, + type PermsData, type UserProfile, type UserProfilesLoadedEvent, } from '@quiet/types' @@ -177,8 +177,6 @@ export class ConnectionsManagerService extends EventEmitter implements OnModuleI if (this.storageService) { this.logger('Stopping orbitdb') await this.storageService?.stopOrbitDb() - this.logger('reset CsrReplicated map and id and certificate store values') - this.storageService.resetCsrAndCertsValues() } if (this.serverIoProvider?.io) { this.logger('Closing socket server') @@ -423,6 +421,12 @@ export class ConnectionsManagerService extends EventEmitter implements OnModuleI this.serverIoProvider.io.emit(SocketActionTypes.PEER_DISCONNECTED, payload) }) await this.storageService.init(_peerId) + // We can use Nest for dependency injection, but I think since the + // registration service depends on the storage service being + // initialized, this is helpful to manually inject the storage + // service for now. Both object construction and object + // initialization need to happen in order based on dependencies. + await this.registrationService.init(this.storageService) this.logger('storage initialized') this.serverIoProvider.io.emit( @@ -449,15 +453,6 @@ export class ConnectionsManagerService extends EventEmitter implements OnModuleI this.registrationService.on(RegistrationEvents.ERROR, payload => { emitError(this.serverIoProvider.io, payload) }) - this.registrationService.on(RegistrationEvents.NEW_USER, async payload => { - await this.storageService?.saveCertificate(payload) - }) - - this.registrationService.on(RegistrationEvents.FINISHED_ISSUING_CERTIFICATES_FOR_ID, payload => { - if (payload.id) { - this.storageService.resolveCsrReplicatedPromise(payload.id) - } - }) } private attachSocketServiceListeners() { @@ -473,7 +468,9 @@ export class ConnectionsManagerService extends EventEmitter implements OnModuleI SocketActionTypes.CONNECTED_PEERS, Array.from(this.libp2pService.connectedPeers.keys()) ) - await this.storageService?.loadAllCertificates() + this.serverIoProvider.io.emit(SocketActionTypes.RESPONSE_GET_CERTIFICATES, { + certificates: await this.storageService?.loadAllCertificates(), + }) await this.storageService?.loadAllChannels() } }) @@ -511,7 +508,7 @@ export class ConnectionsManagerService extends EventEmitter implements OnModuleI // when creating the community. this.socketService.on(SocketActionTypes.SEND_COMMUNITY_CA_DATA, async (payload: PermsData) => { this.logger(`socketService - ${SocketActionTypes.SEND_COMMUNITY_CA_DATA}`) - this.registrationService.permsData = payload + this.registrationService.setPermsData(payload) }) // Public Channels @@ -626,15 +623,12 @@ export class ConnectionsManagerService extends EventEmitter implements OnModuleI this.logger(`Storage - ${StorageEvents.CHANNEL_DELETION_RESPONSE}`) this.serverIoProvider.io.emit(SocketActionTypes.CHANNEL_DELETION_RESPONSE, payload) }) - this.storageService.on( - StorageEvents.REPLICATED_CSR, - async (payload: { csrs: string[]; certificates: string[]; id: string }) => { - this.logger(`Storage - ${StorageEvents.REPLICATED_CSR}`) - this.libp2pService.emit(Libp2pEvents.DIAL_PEERS, await getLibp2pAddressesFromCsrs(payload.csrs)) - this.serverIoProvider.io.emit(SocketActionTypes.RESPONSE_GET_CSRS, { csrs: payload.csrs }) - this.registrationService.emit(RegistrationEvents.REGISTER_USER_CERTIFICATE, payload) - } - ) + this.storageService.on(StorageEvents.REPLICATED_CSR, async (payload: { csrs: string[] }) => { + this.logger(`Storage - ${StorageEvents.REPLICATED_CSR}`) + this.libp2pService.emit(Libp2pEvents.DIAL_PEERS, await getLibp2pAddressesFromCsrs(payload.csrs)) + this.serverIoProvider.io.emit(SocketActionTypes.RESPONSE_GET_CSRS, payload) + this.registrationService.emit(RegistrationEvents.REGISTER_USER_CERTIFICATE, payload) + }) this.socketService.on(SocketActionTypes.SEND_COMMUNITY_METADATA, async (payload: CommunityMetadata) => { await this.storageService?.updateCommunityMetadata(payload) diff --git a/packages/backend/src/nest/registration/registration.service.spec.ts b/packages/backend/src/nest/registration/registration.service.spec.ts index 98abe55d4b..f1fb7e0a60 100644 --- a/packages/backend/src/nest/registration/registration.service.spec.ts +++ b/packages/backend/src/nest/registration/registration.service.spec.ts @@ -4,12 +4,19 @@ import { RegistrationModule } from './registration.module' import { RegistrationService } from './registration.service' import { configCrypto, createRootCA, createUserCsr, type RootCA, verifyUserCert, type UserCsr } from '@quiet/identity' import { type DirResult } from 'tmp' -import { type PermsData } from '@quiet/types' +import { type PermsData, type SaveCertificatePayload } from '@quiet/types' import { Time } from 'pkijs' import { issueCertificate, extractPendingCsrs } from './registration.functions' import { jest } from '@jest/globals' import { createTmpDir } from '../common/utils' import { RegistrationEvents } from './registration.types' +import { EventEmitter } from 'events' +import { CertificatesStore } from '../storage/certificates/certificates.store' +import { StorageService } from '../storage/storage.service' +import { StorageModule } from '../storage/storage.module' +import { OrbitDb } from '../storage/orbitDb/orbitDb.service' +import { create } from 'ipfs-core' +import PeerId from 'peer-id' describe('RegistrationService', () => { let module: TestingModule @@ -148,10 +155,34 @@ describe('RegistrationService', () => { expect(pendingCsrs[0]).toBe(userCsr.userCsr) }) - it('wait for all NEW_USER events until emitting FINISHED_ISSUING_CERTIFICATES_FOR_ID', async () => { - registrationService.permsData = permsData + it('only issues one group of certs at a time', async () => { + const storageModule = await Test.createTestingModule({ + imports: [TestModule, StorageModule], + }).compile() + const certificatesStore = await storageModule.resolve(CertificatesStore) + const orbitDb = await storageModule.resolve(OrbitDb) + const peerId = await PeerId.create() + const ipfs = await create() + const loadAllCertificates = async () => { + return await certificatesStore.loadAllCertificates() + } + const saveCertificate = async (payload: SaveCertificatePayload) => { + await certificatesStore.addCertificate(payload.certificate) + } - const eventSpy = jest.spyOn(registrationService, 'emit') + await orbitDb.create(peerId, ipfs) + await certificatesStore.init(new EventEmitter()) + certificatesStore.updateMetadata({ + id: '39F7485441861F4A2A1A512188F1E0AA', + rootCa: + 'MIIBUDCB+KADAgECAgEBMAoGCCqGSM49BAMCMBIxEDAOBgNVBAMTB3JvY2tldHMwHhcNMTAxMjI4MTAxMDEwWhcNMzAxMjI4MTAxMDEwWjASMRAwDgYDVQQDEwdyb2NrZXRzMFkwEwYHKoZIzj0CAQYIKoZIzj0DAQcDQgAE/ESHf6rXksyiuxSKpQgtiSAhVWNtx4vbFgW6knWfH7MR4dPyxiCNgSeCzRfreuhqVpVtv3U49tcwsqDGkoWHsKM/MD0wDwYDVR0TBAgwBgEB/wIBAzALBgNVHQ8EBAMCAIYwHQYDVR0lBBYwFAYIKwYBBQUHAwIGCCsGAQUFBwMBMAoGCCqGSM49BAMCA0cAMEQCIHrYMhgU/RluSsWoO205EjCQ8pE5MeBZ4Cp8PTgNkOW7AiA690+KIgobiObH6/1JDuS82R0NPO84Ttc8PY886AoKbA==', + ownerCertificate: + 'MIIDeTCCAx6gAwIBAgIGAYwVp42mMAoGCCqGSM49BAMCMBIxEDAOBgNVBAMTB3JvY2tldHMwHhcNMjMxMTI4MTExOTExWhcNMzAwMTMxMjMwMDAwWjBJMUcwRQYDVQQDEz5jYXJhaTJ0d2phem50aW56bndtcnlqdzNlNzVmdXF0Z2xrd2hsemo2d3RlcWx4ano2NnRsZnhpZC5vbmlvbjBZMBMGByqGSM49AgEGCCqGSM49AwEHA0IABNMUauWsTJiuDGt4zoj4lKGgHMkTH96M11fCxMwIInhan0RUB5sv+PtGKbfEfawGjhSQiUaTLdwUGjyIdMs3OMWjggInMIICIzAJBgNVHRMEAjAAMAsGA1UdDwQEAwIAgDAdBgNVHSUEFjAUBggrBgEFBQcDAgYIKwYBBQUHAwEwggFHBgkqhkiG9w0BCQwEggE4BIIBNBETZ2k8vszIRvkuOUk/cNtOb8JcGmw5yVhs45/+e7To4t51nwcdAODj5juVi6+SpLCcHCHhE+g7KswEkC1ScFrW6CRinSgrNBOAUIjOtvWZ/GvK6lI4WTMf7xAaRaJSCF6H0m4cFoUY3JpklJleHhzj0re+NmFZEJ/hNRKochGFy4Xq9Z7StvPpGBlfxhmR7X2t/+HtZaAAbLRLLgbHtCQ7fecg0Qb9Ej58uc+T4Gd2+8ptWvebtOQVU70VAL7uT6aLkFXaDibgSt3kDNvGrwn3AxWlESgROTh5+OWWbfYIbFxjf0PkPDdUSAIOKS9qbYZ+bSYfVq+/0JFyZAa0zhPtgW8wjj0gDCLVm5joyW5Hz2eZ36W7u3cxFME2qmT9G2Dh6NGLn7G19ulVzoTkVmP5/tGPMBUGCisGAQQBg4wbAgEEBxMFZGF2aWQwPQYJKwYBAgEPAwEBBDATLlFtZE5GVjc3dXZOcTJBaWlqUEY0dzY2OU1ucWdiYVdMR1VhZlh0WTdlZjNRRFMwSQYDVR0RBEIwQII+Y2FyYWkydHdqYXpudGluem53bXJ5anczZTc1ZnVxdGdsa3dobHpqNnd0ZXFseGp6NjZ0bGZ4aWQub25pb24wCgYIKoZIzj0EAwIDSQAwRgIhAOafgBe5T0EFjyy0tCRrTHJ1+5ri0W6kAUfc6eRKHIZAAiEA7rFEfPDU+D8MiOF+w0QOdp46dqaWsHFjrDHYPSYGxQA=', + }) + + registrationService.setPermsData(permsData) + registrationService.onModuleInit() + registrationService.init({ certificatesStore, saveCertificate, loadAllCertificates } as unknown as StorageService) const userCsr = await createUserCsr({ nickname: 'alice', @@ -161,8 +192,9 @@ describe('RegistrationService', () => { signAlg: configCrypto.signAlg, hashAlg: configCrypto.hashAlg, }) + const userCsr2 = await createUserCsr({ - nickname: 'karol', + nickname: 'alice', commonName: 'nnnnnnc4c77fb47lk52m5l57h4tcxceo7ymxekfn7yh5m66t4jv2olad.onion', peerId: 'QmffffffqLET9xtAtDzvAr5Pp3egK1H3C5iJAZm1SpLEp6', dmPublicKey: 'testdmPublicKey', @@ -170,14 +202,18 @@ describe('RegistrationService', () => { hashAlg: configCrypto.hashAlg, }) - const csrs: string[] = [userCsr.userCsr, userCsr2.userCsr] - // @ts-ignore - fn 'issueCertificates' is private - await registrationService.issueCertificates({ certificates: [], csrs, id: 1 }) + const certificates: string[] = [] - expect(eventSpy).toHaveBeenLastCalledWith(RegistrationEvents.FINISHED_ISSUING_CERTIFICATES_FOR_ID, { - id: 1, - }) + registrationService.emit(RegistrationEvents.REGISTER_USER_CERTIFICATE, { csrs: [userCsr.userCsr] }) + + registrationService.emit(RegistrationEvents.REGISTER_USER_CERTIFICATE, { csrs: [userCsr2.userCsr] }) + + await new Promise(r => setTimeout(r, 2000)) + + expect((await certificatesStore.loadAllCertificates()).length).toEqual(1) - expect(eventSpy).toHaveBeenCalledTimes(3) + await orbitDb.stop() + await ipfs.stop() + await certificatesStore.close() }) }) diff --git a/packages/backend/src/nest/registration/registration.service.ts b/packages/backend/src/nest/registration/registration.service.ts index 174cb286c2..2dc4bbf970 100644 --- a/packages/backend/src/nest/registration/registration.service.ts +++ b/packages/backend/src/nest/registration/registration.service.ts @@ -4,35 +4,82 @@ import { extractPendingCsrs, issueCertificate } from './registration.functions' import { ErrorCodes, ErrorMessages, PermsData, RegisterOwnerCertificatePayload, SocketActionTypes } from '@quiet/types' import { RegistrationEvents } from './registration.types' import Logger from '../common/logger' +import { StorageService } from '../storage/storage.service' @Injectable() export class RegistrationService extends EventEmitter implements OnModuleInit { private readonly logger = Logger(RegistrationService.name) - public certificates: string[] = [] - private _permsData: PermsData + private permsData: PermsData + private storageService: StorageService + private registrationEvents: { csrs: string[] }[] = [] + private registrationEventInProgress = false constructor() { super() } onModuleInit() { - this.on( - RegistrationEvents.REGISTER_USER_CERTIFICATE, - async (payload: { csrs: string[]; certificates: string[]; id: string }) => { - await this.issueCertificates(payload) + this.on(RegistrationEvents.REGISTER_USER_CERTIFICATE, async (payload: { csrs: string[] }) => { + // Save the registration event and then try to process it, but + // since we only process a single event at a time, it might not + // get processed until other events have been processed. + this.registrationEvents.push(payload) + await this.tryIssueCertificates() + }) + } + + public init(storageService: StorageService) { + this.storageService = storageService + } + + public setPermsData(permsData: PermsData) { + this.permsData = permsData + } + + public async tryIssueCertificates() { + this.logger('Trying to issue certificates', this.registrationEventInProgress, this.registrationEvents) + // Process only a single registration event at a time so that we + // do not register two certificates with the same name. + if (!this.registrationEventInProgress) { + // Get the next event. + const event = this.registrationEvents.shift() + if (event) { + this.logger('Issuing certificates', event) + // Event processing in progress + this.registrationEventInProgress = true + + // Await the processing function and make sure everything that + // needs to be done in order is awaited inside this function. + await this.issueCertificates({ + ...event, + certificates: (await this.storageService?.loadAllCertificates()) as string[], + }) + + this.logger('Finished issuing certificates') + // Event processing finished + this.registrationEventInProgress = false + + // Re-run this function if there are more events to process + if (this.registrationEvents.length > 0) { + setTimeout(this.tryIssueCertificates.bind(this), 0) + } } - ) + } } - private async issueCertificates(payload: { csrs: string[]; certificates: string[]; id?: string }) { + private async issueCertificates(payload: { csrs: string[]; certificates: string[] }) { // Lack of permsData means that we are not the owner of the // community in the official model of the app, however anyone can // modify the source code, put malicious permsData here, issue // false certificates and try to trick other users. To prevent // that, peers verify that anything that is written to the // certificate store is signed by the owner. - if (!this._permsData) { - if (payload.id) this.emit(RegistrationEvents.FINISHED_ISSUING_CERTIFICATES_FOR_ID, { id: payload.id }) + // + // NOTE: There may be a race condition here if we try to issue + // certs before permsData is set. We may want to refactor this or + // add the ability to retry. + if (!this.permsData) { + this.logger('Not issuing certificates due to missing perms data') return } @@ -44,20 +91,11 @@ export class RegistrationService extends EventEmitter implements OnModuleInit { await this.registerUserCertificate(csr) }) ) - - if (payload.id) this.emit(RegistrationEvents.FINISHED_ISSUING_CERTIFICATES_FOR_ID, { id: payload.id }) - } - - public set permsData(perms: PermsData) { - this._permsData = { - certificate: perms.certificate, - privKey: perms.privKey, - } } public async registerOwnerCertificate(payload: RegisterOwnerCertificatePayload): Promise { - this._permsData = payload.permsData - const result = await issueCertificate(payload.userCsr.userCsr, this._permsData) + this.permsData = payload.permsData + const result = await issueCertificate(payload.userCsr.userCsr, this.permsData) if (result?.cert) { this.emit(SocketActionTypes.SAVED_OWNER_CERTIFICATE, { communityId: payload.communityId, @@ -74,10 +112,15 @@ export class RegistrationService extends EventEmitter implements OnModuleInit { } public async registerUserCertificate(csr: string): Promise { - const result = await issueCertificate(csr, this._permsData) + const result = await issueCertificate(csr, this.permsData) this.logger('DuplicatedCertBug', { result }) if (result?.cert) { - this.emit(RegistrationEvents.NEW_USER, { certificate: result.cert }) + // Save certificate (awaited) so that we are sure that the certs + // are saved before processing the next round of CSRs. + // Otherwise, we could issue a duplicate certificate. + + // @ts-ignore + await this.storageService?.saveCertificate({ certificate: result.cert }) } } } diff --git a/packages/backend/src/nest/registration/registration.types.ts b/packages/backend/src/nest/registration/registration.types.ts index b9025ff870..02da49c0d6 100644 --- a/packages/backend/src/nest/registration/registration.types.ts +++ b/packages/backend/src/nest/registration/registration.types.ts @@ -2,5 +2,4 @@ export enum RegistrationEvents { ERROR = 'error', NEW_USER = 'newUser', REGISTER_USER_CERTIFICATE = 'registerUserCertificate', - FINISHED_ISSUING_CERTIFICATES_FOR_ID = 'FINISHED_ISSUING_CERTIFICATES_FOR_ID', } diff --git a/packages/backend/src/nest/socket/socket.service.ts b/packages/backend/src/nest/socket/socket.service.ts index 74d465e7ca..a2194921fa 100644 --- a/packages/backend/src/nest/socket/socket.service.ts +++ b/packages/backend/src/nest/socket/socket.service.ts @@ -15,6 +15,7 @@ import { DeleteFilesFromChannelSocketPayload, SaveCSRPayload, CommunityMetadata, + type PermsData, type UserProfile, } from '@quiet/types' import EventEmitter from 'events' @@ -153,10 +154,6 @@ export class SocketService extends EventEmitter implements OnModuleInit { }) // ====== Community ====== - socket.on(SocketActionTypes.SEND_COMMUNITY_METADATA, (payload: CommunityMetadata) => { - this.emit(SocketActionTypes.SEND_COMMUNITY_METADATA, payload) - }) - socket.on(SocketActionTypes.CREATE_COMMUNITY, async (payload: InitCommunityPayload) => { this.logger(`Creating community ${payload.id}`) this.emit(SocketActionTypes.CREATE_COMMUNITY, payload) @@ -177,11 +174,20 @@ export class SocketService extends EventEmitter implements OnModuleInit { this.logger('Leaving community') this.emit(SocketActionTypes.LEAVE_COMMUNITY) }) + socket.on(SocketActionTypes.LIBP2P_PSK_SAVED, payload => { this.logger('Saving PSK', payload) this.emit(SocketActionTypes.LIBP2P_PSK_SAVED, payload) }) + socket.on(SocketActionTypes.SEND_COMMUNITY_METADATA, (payload: CommunityMetadata) => { + this.emit(SocketActionTypes.SEND_COMMUNITY_METADATA, payload) + }) + + socket.on(SocketActionTypes.SEND_COMMUNITY_CA_DATA, (payload: PermsData) => { + this.emit(SocketActionTypes.SEND_COMMUNITY_CA_DATA, payload) + }) + // ====== Users ====== socket.on(SocketActionTypes.SAVE_USER_PROFILE, (profile: UserProfile) => { diff --git a/packages/backend/src/nest/storage/certifacteRequests/certificatesRequestsStore.spec.ts b/packages/backend/src/nest/storage/certifacteRequests/certificatesRequestsStore.spec.ts index c1040dc9d8..a35257b78e 100644 --- a/packages/backend/src/nest/storage/certifacteRequests/certificatesRequestsStore.spec.ts +++ b/packages/backend/src/nest/storage/certifacteRequests/certificatesRequestsStore.spec.ts @@ -111,61 +111,4 @@ describe('CertificatesRequestsStore', () => { expect(spy).toBeCalledTimes(1) }) - it('2 replicated events - first not resolved ', async () => { - const emitter = new EventEmitter() - await certificatesRequestsStore.init(emitter) - - const spy = jest.fn() - - emitter.on(StorageEvents.LOADED_USER_CSRS, spy) - await replicatedEvent(certificatesRequestsStore.store) - await replicatedEvent(certificatesRequestsStore.store) - - expect(spy).toBeCalledTimes(1) - }) - it('2 replicated events - first resolved ', async () => { - const emitter = new EventEmitter() - await certificatesRequestsStore.init(emitter) - - const spy = jest.fn() - - emitter.on(StorageEvents.LOADED_USER_CSRS, spy) - await replicatedEvent(certificatesRequestsStore.store) - await replicatedEvent(certificatesRequestsStore.store) - - certificatesRequestsStore.resolveCsrReplicatedPromise(1) - await new Promise(resolve => setTimeout(() => resolve(), 500)) - - expect(spy).toBeCalledTimes(2) - }) - it('3 replicated events - no resolved promises', async () => { - const emitter = new EventEmitter() - await certificatesRequestsStore.init(emitter) - - const spy = jest.fn() - - emitter.on(StorageEvents.LOADED_USER_CSRS, spy) - await replicatedEvent(certificatesRequestsStore.store) - await replicatedEvent(certificatesRequestsStore.store) - await replicatedEvent(certificatesRequestsStore.store) - - expect(spy).toBeCalledTimes(1) - }) - it('3 replicated events - two resolved promises ', async () => { - const emitter = new EventEmitter() - await certificatesRequestsStore.init(emitter) - - const spy = jest.fn() - emitter.on(StorageEvents.LOADED_USER_CSRS, spy) - - await replicatedEvent(certificatesRequestsStore.store) - await replicatedEvent(certificatesRequestsStore.store) - certificatesRequestsStore.resolveCsrReplicatedPromise(1) - await new Promise(resolve => setTimeout(() => resolve(), 500)) - await replicatedEvent(certificatesRequestsStore.store) - certificatesRequestsStore.resolveCsrReplicatedPromise(2) - await new Promise(resolve => setTimeout(() => resolve(), 500)) - - expect(spy).toBeCalledTimes(3) - }) }) diff --git a/packages/backend/src/nest/storage/certifacteRequests/certificatesRequestsStore.ts b/packages/backend/src/nest/storage/certifacteRequests/certificatesRequestsStore.ts index 2675542a1d..9b09fc4f9e 100644 --- a/packages/backend/src/nest/storage/certifacteRequests/certificatesRequestsStore.ts +++ b/packages/backend/src/nest/storage/certifacteRequests/certificatesRequestsStore.ts @@ -13,58 +13,42 @@ import Logger from '../../common/logger' @Injectable() export class CertificatesRequestsStore { public store: EventStore - public csrReplicatedPromiseMap: Map = new Map() - private csrReplicatedPromiseId: number = 0 + private emitter: EventEmitter private readonly logger = Logger(CertificatesRequestsStore.name) constructor(private readonly orbitDbService: OrbitDb) {} public async init(emitter: EventEmitter) { - this.logger('Initializing...') + this.logger('Initializing certificates requests store') + this.store = await this.orbitDbService.orbitDb.log('csrs', { replicate: false, accessController: { write: ['*'], }, }) + this.emitter = emitter this.store.events.on('write', async (_address, entry) => { this.logger('Added CSR to database') - emitter.emit(StorageEvents.LOADED_USER_CSRS, { - csrs: await this.getCsrs(), - id: this.csrReplicatedPromiseId, - }) + this.loadedCertificateRequests() }) this.store.events.on('replicated', async () => { - this.logger('Replicated CSRS') - - this.csrReplicatedPromiseId++ - - const filteredCsrs = await this.getCsrs() - this.createCsrReplicatedPromise(this.csrReplicatedPromiseId) - - // Lock replicated event until previous event is processed by registration service - if (this.csrReplicatedPromiseId > 1) { - const csrReplicatedPromiseMapId = this.csrReplicatedPromiseMap.get(this.csrReplicatedPromiseId - 1) - - if (csrReplicatedPromiseMapId?.promise) { - await csrReplicatedPromiseMapId.promise - } - } - - emitter.emit(StorageEvents.LOADED_USER_CSRS, { - csrs: filteredCsrs, - id: this.csrReplicatedPromiseId, - }) + this.logger('Replicated CSRs') + this.loadedCertificateRequests() }) - emitter.emit(StorageEvents.LOADED_USER_CSRS, { + // TODO: Load CSRs in case the owner closes the app before issuing + // certificates + this.logger('Initialized') + } + + public async loadedCertificateRequests() { + this.emitter.emit(StorageEvents.LOADED_USER_CSRS, { csrs: await this.getCsrs(), - id: this.csrReplicatedPromiseId, }) - this.logger('Initialized') } public async close() { @@ -77,30 +61,6 @@ export class CertificatesRequestsStore { return this.store?.address } - public resetCsrReplicatedMapAndId() { - this.csrReplicatedPromiseMap = new Map() - this.csrReplicatedPromiseId = 0 - } - - private createCsrReplicatedPromise(id: number) { - let resolveFunction - const promise = new Promise(resolve => { - resolveFunction = resolve - }) - this.csrReplicatedPromiseMap.set(id, { promise, resolveFunction }) - } - - public resolveCsrReplicatedPromise(id: number) { - const csrReplicatedPromiseMapId = this.csrReplicatedPromiseMap.get(id) - if (csrReplicatedPromiseMapId) { - csrReplicatedPromiseMapId?.resolveFunction(id) - this.csrReplicatedPromiseMap.delete(id) - } else { - this.logger.error(`No promise with ID ${id} found.`) - return - } - } - public async addUserCsr(csr: string) { await this.store.add(csr) return true @@ -164,4 +124,13 @@ export class CertificatesRequestsStore { this.logger('DuplicatedCertBug', '[...filteredCsrsMap.values()]', [...filteredCsrsMap.values()]) return [...filteredCsrsMap.values()] } + + public clean() { + // FIXME: Add correct typings on object fields. + + // @ts-ignore + this.store = undefined + // @ts-ignore + this.emitter = undefined + } } diff --git a/packages/backend/src/nest/storage/certificates/certificates.store.ts b/packages/backend/src/nest/storage/certificates/certificates.store.ts index 1a57c419a1..f04d487d1b 100644 --- a/packages/backend/src/nest/storage/certificates/certificates.store.ts +++ b/packages/backend/src/nest/storage/certificates/certificates.store.ts @@ -23,6 +23,7 @@ export class CertificatesStore { private metadata: CommunityMetadata | undefined private filteredCertificatesMapping: Map> private usernameMapping: Map + private emitter: EventEmitter private readonly logger = Logger(CertificatesStore.name) @@ -31,12 +32,6 @@ export class CertificatesStore { this.usernameMapping = new Map() } - public resetValues() { - this.metadata = undefined - this.filteredCertificatesMapping = new Map() - this.usernameMapping = new Map() - } - public async init(emitter: EventEmitter) { this.logger('Initializing certificates log store') @@ -46,6 +41,7 @@ export class CertificatesStore { write: ['*'], }, }) + this.emitter = emitter this.store.events.on('ready', async () => { this.logger('Loaded certificates to memory') @@ -54,27 +50,27 @@ export class CertificatesStore { this.store.events.on('write', async () => { this.logger('Saved certificate locally') - await loadedCertificates() + await this.loadedCertificates() }) this.store.events.on('replicated', async () => { this.logger('REPLICATED: Certificates') emitter.emit(SocketActionTypes.CONNECTION_PROCESS_INFO, ConnectionProcessInfo.CERTIFICATES_REPLICATED) - await loadedCertificates() + await this.loadedCertificates() }) - const loadedCertificates = async () => { - emitter.emit(StorageEvents.LOADED_CERTIFICATES, { - certificates: await this.getCertificates(), - }) - } - // @ts-expect-error - OrbitDB's type declaration of `load` lacks 'options' await this.store.load({ fetchEntryTimeout: 15000 }) this.logger('Initialized') } + public async loadedCertificates() { + this.emitter?.emit(StorageEvents.LOADED_CERTIFICATES, { + certificates: await this.getCertificates(), + }) + } + public async close() { await this.store?.close() } @@ -85,7 +81,7 @@ export class CertificatesStore { public async addCertificate(certificate: string) { this.logger('Adding user certificate') - await this.store.add(certificate) + await this.store?.add(certificate) return true } @@ -97,6 +93,17 @@ export class CertificatesStore { public updateMetadata(metadata: CommunityMetadata) { if (!metadata) return this.metadata = metadata + // FIXME: Community metadata is required for validating + // certificates, so we re-validate certificates once community + // metadata is set. Currently the certificates store receives the + // community metadata via an event. Is there a better way to + // organize this so that the dependencies are clearer? Having + // CertificateStore depend on CommunityMetadataStore directly? + // Storing community metadata in LevelDB? Only initializing + // certificate store after community metadata is available? + if (this.store) { + this.loadedCertificates() + } } private async validateCertificate(certificate: string) { @@ -117,15 +124,12 @@ export class CertificatesStore { throw new NoCryptoEngineError() } - const parsedCertificate = loadCertificate(certificate) - - let metadata = this.metadata - while (!metadata) { - await new Promise(res => setTimeout(res, 100)) - metadata = this.metadata + if (!this.metadata) { + throw new Error('Community metadata missing') } - const parsedRootCertificate = loadCertificate(metadata.rootCa) + const parsedRootCertificate = loadCertificate(this.metadata.rootCa) + const parsedCertificate = loadCertificate(certificate) const verification = await parsedCertificate.verify(parsedRootCertificate) return verification @@ -144,6 +148,10 @@ export class CertificatesStore { * https://github.com/TryQuiet/quiet/issues/1899 */ protected async getCertificates() { + if (!this.store) { + return [] + } + // @ts-expect-error - OrbitDB's type declaration of `load` lacks 'options' await this.store.load({ fetchEntryTimeout: 15000 }) const allCertificates = this.store @@ -196,4 +204,16 @@ export class CertificatesStore { // Return desired data from updated cache return this.usernameMapping.get(pubkey) } + + public clean() { + // FIXME: Add correct typings on object fields. + + // @ts-ignore + this.store = undefined + // @ts-ignore + this.emitter = undefined + this.metadata = undefined + this.filteredCertificatesMapping = new Map() + this.usernameMapping = new Map() + } } diff --git a/packages/backend/src/nest/storage/communityMetadata/communityMetadata.store.ts b/packages/backend/src/nest/storage/communityMetadata/communityMetadata.store.ts index 2d0bc2099c..7d04ac5f68 100644 --- a/packages/backend/src/nest/storage/communityMetadata/communityMetadata.store.ts +++ b/packages/backend/src/nest/storage/communityMetadata/communityMetadata.store.ts @@ -81,6 +81,10 @@ export class CommunityMetadataStore { // @ts-expect-error - OrbitDB's type declaration of `load` lacks 'options' await this.store.load({ fetchEntryTimeout: 15000 }) + const meta = this.getCommunityMetadata() + if (meta) { + emitter.emit(StorageEvents.COMMUNITY_METADATA_SAVED, meta) + } logger('Loaded community metadata to memory') } @@ -207,6 +211,13 @@ export class CommunityMetadataStore { return metadata[0] } } + + public clean() { + // FIXME: Add correct typings on object fields. + + // @ts-ignore + this.store = undefined + } } export class CommunityMetadataKeyValueIndex extends KeyValueIndex { diff --git a/packages/backend/src/nest/storage/storage.service.spec.ts b/packages/backend/src/nest/storage/storage.service.spec.ts index a1e1d52a2b..0546578233 100644 --- a/packages/backend/src/nest/storage/storage.service.spec.ts +++ b/packages/backend/src/nest/storage/storage.service.spec.ts @@ -609,63 +609,4 @@ describe('StorageService', () => { await expect(storageService.deleteFilesFromChannel(messages)).resolves.not.toThrowError() }) }) - - describe.skip('replicate certificatesRequests event', () => { - const replicatedEvent = async () => { - // @ts-ignore - Property 'certificates' is private - storageService.certificatesRequestsStore.events.emit('replicated') - await new Promise(resolve => setTimeout(() => resolve(), 2000)) - } - - it('replicated event ', async () => { - await storageService.init(peerId) - const spyOnUpdatePeersList = jest.spyOn(storageService, 'updatePeersList') - await replicatedEvent() - expect(spyOnUpdatePeersList).toBeCalledTimes(1) - }) - - it('2 replicated events - first not resolved ', async () => { - await storageService.init(peerId) - const spyOnUpdatePeersList = jest.spyOn(storageService, 'updatePeersList') - await replicatedEvent() - await replicatedEvent() - expect(spyOnUpdatePeersList).toBeCalledTimes(1) - }) - - it('2 replicated events - first resolved ', async () => { - await storageService.init(peerId) - const spyOnUpdatePeersList = jest.spyOn(storageService, 'updatePeersList') - await replicatedEvent() - await replicatedEvent() - storageService.resolveCsrReplicatedPromise(1) - await new Promise(resolve => setTimeout(() => resolve(), 500)) - expect(spyOnUpdatePeersList).toBeCalledTimes(2) - }) - - it('3 replicated events - no resolved promises', async () => { - await storageService.init(peerId) - const spyOnUpdatePeersList = jest.spyOn(storageService, 'updatePeersList') - - await replicatedEvent() - await replicatedEvent() - await replicatedEvent() - - expect(spyOnUpdatePeersList).toBeCalledTimes(1) - }) - - it('3 replicated events - two resolved promises ', async () => { - await storageService.init(peerId) - const spyOnUpdatePeersList = jest.spyOn(storageService, 'updatePeersList') - - await replicatedEvent() - await replicatedEvent() - storageService.resolveCsrReplicatedPromise(1) - await new Promise(resolve => setTimeout(() => resolve(), 500)) - await replicatedEvent() - storageService.resolveCsrReplicatedPromise(2) - await new Promise(resolve => setTimeout(() => resolve(), 500)) - - expect(spyOnUpdatePeersList).toBeCalledTimes(3) - }) - }) }) diff --git a/packages/backend/src/nest/storage/storage.service.ts b/packages/backend/src/nest/storage/storage.service.ts index 7bfd2dcbda..0cdd3126cd 100644 --- a/packages/backend/src/nest/storage/storage.service.ts +++ b/packages/backend/src/nest/storage/storage.service.ts @@ -186,6 +186,11 @@ export class StorageService extends EventEmitter { this.logger('1/3') console.time('Storage.initDatabases') + // FIXME: This is sort of messy how we are initializing things. + // Currently, the CommunityMetadataStore sends an event during + // initialization which is picked up by the CertificatesStore, but + // the CertificatesStore is not initialized yet. Perhaps we can + // attach `this` as an EventEmitter first and then load data. await this.communityMetadataStore.init(this) await this.certificatesStore.init(this) await this.certificatesRequestsStore.init(this) @@ -299,9 +304,7 @@ export class StorageService extends EventEmitter { public async loadAllCertificates() { this.logger('Loading all certificates') - this.emit(StorageEvents.REPLICATED_CERTIFICATES, { - certificates: await this.certificatesStore.loadAllCertificates(), - }) + return await this.certificatesStore.loadAllCertificates() } public async attachCertificatesStoreListeners() { @@ -312,18 +315,13 @@ export class StorageService extends EventEmitter { } public async attachCsrsStoreListeners() { - this.on(StorageEvents.LOADED_USER_CSRS, async payload => { - const allCertificates = this.getAllEventLogEntries(this.certificatesStore.store) - this.emit(StorageEvents.REPLICATED_CSR, { csrs: payload.csrs, certificates: allCertificates, id: payload.id }) + this.on(StorageEvents.LOADED_USER_CSRS, async (payload: { csrs: string[] }) => { + this.emit(StorageEvents.REPLICATED_CSR, payload) // TODO await this.updatePeersList() }) } - public resolveCsrReplicatedPromise(id: number) { - this.certificatesRequestsStore.resolveCsrReplicatedPromise(id) - } - public async loadAllChannels() { this.logger('Getting all channels') // @ts-expect-error - OrbitDB's type declaration of `load` lacks 'options' @@ -776,11 +774,6 @@ export class StorageService extends EventEmitter { }) } - public resetCsrAndCertsValues() { - this.certificatesRequestsStore.resetCsrReplicatedMapAndId() - this.certificatesStore.resetValues() - } - private clean() { // @ts-ignore this.channels = undefined @@ -796,6 +789,8 @@ export class StorageService extends EventEmitter { this.filesManager = null this.peerId = null - this.resetCsrAndCertsValues() + this.certificatesRequestsStore.clean() + this.certificatesStore.clean() + this.communityMetadataStore.clean() } } diff --git a/packages/state-manager/src/sagas/communities/communities.master.saga.ts b/packages/state-manager/src/sagas/communities/communities.master.saga.ts index e7fe4d7d42..7984c9ece8 100644 --- a/packages/state-manager/src/sagas/communities/communities.master.saga.ts +++ b/packages/state-manager/src/sagas/communities/communities.master.saga.ts @@ -8,6 +8,7 @@ import { createNetworkSaga } from './createNetwork/createNetwork.saga' import { responseCreateNetworkSaga } from './responseCreateNetwork/responseCreateNetwork.saga' import { saveCommunityMetadataSaga } from './saveCommunityMetadata/saveCommunityMetadata.saga' import { sendCommunityMetadataSaga } from './updateCommunityMetadata/updateCommunityMetadata.saga' +import { sendCommunityCaDataSaga } from './sendCommunityCaData/sendCommunityCaData.saga' export function* communitiesMasterSaga(socket: Socket): Generator { yield all([ @@ -18,5 +19,6 @@ export function* communitiesMasterSaga(socket: Socket): Generator { takeEvery(communitiesActions.launchCommunity.type, launchCommunitySaga, socket), takeEvery(communitiesActions.saveCommunityMetadata.type, saveCommunityMetadataSaga, socket), takeEvery(communitiesActions.sendCommunityMetadata.type, sendCommunityMetadataSaga, socket), + takeEvery(communitiesActions.sendCommunityCaData.type, sendCommunityCaDataSaga, socket), ]) }