Storage Service: Write

ts/services/storage.ts

@@ -0,0 +1,707 @@
+/* tslint:disable no-backbone-get-set-outside-model */
+import _ from 'lodash';
+import pMap from 'p-map';
+
+import Crypto from '../textsecure/Crypto';
+import dataInterface from '../sql/Client';
+import {
+  arrayBufferToBase64,
+  base64ToArrayBuffer,
+  deriveStorageItemKey,
+  deriveStorageManifestKey,
+} from '../Crypto';
+import {
+  ManifestRecordClass,
+  ManifestRecordIdentifierClass,
+  StorageItemClass,
+  StorageManifestClass,
+  StorageRecordClass,
+} from '../textsecure.d';
+import { ConversationModelType } from '../model-types.d';
+import {
+  mergeAccountRecord,
+  mergeContactRecord,
+  mergeGroupV1Record,
+  toAccountRecord,
+  toContactRecord,
+  toGroupV1Record,
+} from './storageRecordOps';
+
+const {
+  eraseStorageServiceStateFromConversations,
+  updateConversation,
+} = dataInterface;
+
+let consecutiveConflicts = 0;
+
+type UnknownRecord = {
+  itemType: number;
+  storageID: string;
+};
+
+function createWriteOperation(
+  storageManifest: StorageManifestClass,
+  newItems: Array<StorageItemClass>,
+  deleteKeys: Array<ArrayBuffer>,
+  clearAll = false
+) {
+  const writeOperation = new window.textsecure.protobuf.WriteOperation();
+  writeOperation.manifest = storageManifest;
+  writeOperation.insertItem = newItems;
+  writeOperation.deleteKey = deleteKeys;
+  writeOperation.clearAll = clearAll;
+
+  return writeOperation;
+}
+
+async function encryptRecord(
+  storageID: string | undefined,
+  storageRecord: StorageRecordClass
+): Promise<StorageItemClass> {
+  const storageItem = new window.textsecure.protobuf.StorageItem();
+
+  const storageKeyBuffer = storageID
+    ? base64ToArrayBuffer(String(storageID))
+    : generateStorageID();
+
+  const storageKeyBase64 = window.storage.get('storageKey');
+  const storageKey = base64ToArrayBuffer(storageKeyBase64);
+  const storageItemKey = await deriveStorageItemKey(
+    storageKey,
+    arrayBufferToBase64(storageKeyBuffer)
+  );
+
+  const encryptedRecord = await Crypto.encryptProfile(
+    storageRecord.toArrayBuffer(),
+    storageItemKey
+  );
+
+  storageItem.key = storageKeyBuffer;
+  storageItem.value = encryptedRecord;
+
+  return storageItem;
+}
+
+function generateStorageID(): ArrayBuffer {
+  return Crypto.getRandomBytes(16);
+}
+
+type GeneratedManifestType = {
+  conversationsToUpdate: Array<{
+    conversation: ConversationModelType;
+    storageID: string | undefined;
+  }>;
+  deleteKeys: Array<ArrayBuffer>;
+  newItems: Set<StorageItemClass>;
+  storageManifest: StorageManifestClass;
+};
+
+/* tslint:disable-next-line max-func-body-length */
+async function generateManifest(
+  version: number,
+  isNewManifest = false
+): Promise<GeneratedManifestType> {
+  window.log.info(
+    `storageService.generateManifest: generating manifest for version ${version}. Is new? ${isNewManifest}`
+  );
+
+  const ITEM_TYPE = window.textsecure.protobuf.ManifestRecord.Identifier.Type;
+
+  const conversationsToUpdate = [];
+  const deleteKeys = [];
+  const manifestRecordKeys: Set<ManifestRecordIdentifierClass> = new Set();
+  const newItems: Set<StorageItemClass> = new Set();
+
+  const conversations = window.getConversations();
+  for (let i = 0; i < conversations.length; i += 1) {
+    const conversation = conversations.models[i];
+    const identifier = new window.textsecure.protobuf.ManifestRecord.Identifier();
+
+    let storageRecord;
+    if (conversation.isMe()) {
+      storageRecord = new window.textsecure.protobuf.StorageRecord();
+      // eslint-disable-next-line no-await-in-loop
+      storageRecord.account = await toAccountRecord(conversation);
+      identifier.type = ITEM_TYPE.ACCOUNT;
+    } else if (conversation.isPrivate()) {
+      storageRecord = new window.textsecure.protobuf.StorageRecord();
+      // eslint-disable-next-line no-await-in-loop
+      storageRecord.contact = await toContactRecord(conversation);
+      identifier.type = ITEM_TYPE.CONTACT;
+    } else {
+      storageRecord = new window.textsecure.protobuf.StorageRecord();
+      // eslint-disable-next-line no-await-in-loop
+      storageRecord.groupV1 = await toGroupV1Record(conversation);
+      identifier.type = ITEM_TYPE.GROUPV1;
+    }
+
+    if (storageRecord) {
+      const isNewItem =
+        isNewManifest || Boolean(conversation.get('needsStorageServiceSync'));
+
+      const storageID = isNewItem
+        ? arrayBufferToBase64(generateStorageID())
+        : conversation.get('storageID');
+
+      // eslint-disable-next-line no-await-in-loop
+      const storageItem = await encryptRecord(storageID, storageRecord);
+      identifier.raw = storageItem.key;
+
+      // When a client needs to update a given record it should create it
+      // under a new key and delete the existing key.
+      if (isNewItem) {
+        newItems.add(storageItem);
+
+        const oldStorageID = conversation.get('storageID');
+        if (oldStorageID) {
+          deleteKeys.push(base64ToArrayBuffer(oldStorageID));
+        }
+
+        conversationsToUpdate.push({
+          conversation,
+          storageID,
+        });
+      }
+
+      manifestRecordKeys.add(identifier);
+    }
+  }
+
+  const unknownRecordsArray =
+    window.storage.get('storage-service-unknown-records') || [];
+
+  window.log.info(
+    `storageService.generateManifest: adding ${unknownRecordsArray.length} unknown records`
+  );
+
+  // When updating the manifest, ensure all "unknown" keys are added to the
+  // new manifest, so we don't inadvertently delete something we don't understand
+  unknownRecordsArray.forEach((record: UnknownRecord) => {
+    const identifier = new window.textsecure.protobuf.ManifestRecord.Identifier();
+    identifier.type = record.itemType;
+    identifier.raw = base64ToArrayBuffer(record.storageID);
+
+    manifestRecordKeys.add(identifier);
+  });
+
+  const manifestRecord = new window.textsecure.protobuf.ManifestRecord();
+  manifestRecord.version = version;
+  manifestRecord.keys = Array.from(manifestRecordKeys);
+
+  const storageKeyBase64 = window.storage.get('storageKey');
+  const storageKey = base64ToArrayBuffer(storageKeyBase64);
+  const storageManifestKey = await deriveStorageManifestKey(
+    storageKey,
+    version
+  );
+  const encryptedManifest = await Crypto.encryptProfile(
+    manifestRecord.toArrayBuffer(),
+    storageManifestKey
+  );
+
+  const storageManifest = new window.textsecure.protobuf.StorageManifest();
+  storageManifest.version = version;
+  storageManifest.value = encryptedManifest;
+
+  return {
+    conversationsToUpdate,
+    deleteKeys,
+    newItems,
+    storageManifest,
+  };
+}
+
+async function uploadManifest(
+  version: number,
+  {
+    conversationsToUpdate,
+    deleteKeys,
+    newItems,
+    storageManifest,
+  }: GeneratedManifestType
+): Promise<void> {
+  if (!window.textsecure.messaging) {
+    throw new Error('storageService.uploadManifest: We are offline!');
+  }
+
+  const credentials = window.storage.get('storageCredentials');
+  try {
+    window.log.info(
+      `storageService.uploadManifest: inserting ${newItems.size} items, deleting ${deleteKeys.length} keys`
+    );
+
+    const writeOperation = createWriteOperation(
+      storageManifest,
+      Array.from(newItems),
+      deleteKeys
+    );
+
+    window.log.info('storageService.uploadManifest: uploading...');
+    await window.textsecure.messaging.modifyStorageRecords(
+      writeOperation.toArrayBuffer(),
+      {
+        credentials,
+      }
+    );
+
+    window.log.info(
+      `storageService.uploadManifest: upload done, updating ${conversationsToUpdate.length} conversation(s) with new storageIDs`
+    );
+
+    // update conversations with the new storageID
+    conversationsToUpdate.forEach(({ conversation, storageID }) => {
+      conversation.set({
+        needsStorageServiceSync: false,
+        storageID,
+      });
+      updateConversation(conversation.attributes);
+    });
+  } catch (err) {
+    window.log.error(
+      `storageService.uploadManifest: failed! ${
+        err && err.stack ? err.stack : String(err)
+      }`
+    );
+
+    if (err.code === 409) {
+      window.log.info(
+        `storageService.uploadManifest: Conflict found, running sync job times(${consecutiveConflicts})`
+      );
+
+      if (consecutiveConflicts > 3) {
+        window.log.error(
+          'storageService.uploadManifest: Exceeded maximum consecutive conflicts'
+        );
+        return;
+      }
+      consecutiveConflicts += 1;
+
+      throw err;
+    }
+
+    throw err;
+  }
+
+  window.log.info(
+    'storageService.uploadManifest: setting new manifestVersion',
+    version
+  );
+  window.storage.put('manifestVersion', version);
+  consecutiveConflicts = 0;
+  await window.textsecure.messaging.sendFetchManifestSyncMessage();
+}
+
+async function stopStorageServiceSync() {
+  window.log.info('storageService.stopStorageServiceSync');
+
+  await window.storage.remove('storageKey');
+
+  if (!window.textsecure.messaging) {
+    throw new Error('storageService.stopStorageServiceSync: We are offline!');
+  }
+
+  await window.textsecure.messaging.sendRequestKeySyncMessage();
+}
+
+async function createNewManifest() {
+  window.log.info('storageService.createNewManifest: creating new manifest');
+
+  const version = window.storage.get('manifestVersion') || 0;
+
+  const {
+    conversationsToUpdate,
+    newItems,
+    storageManifest,
+  } = await generateManifest(version, true);
+
+  await uploadManifest(version, {
+    conversationsToUpdate,
+    // we have created a new manifest, there should be no keys to delete
+    deleteKeys: [],
+    newItems,
+    storageManifest,
+  });
+}
+
+async function decryptManifest(
+  encryptedManifest: StorageManifestClass
+): Promise<ManifestRecordClass> {
+  const { version, value } = encryptedManifest;
+
+  const storageKeyBase64 = window.storage.get('storageKey');
+  const storageKey = base64ToArrayBuffer(storageKeyBase64);
+  const storageManifestKey = await deriveStorageManifestKey(
+    storageKey,
+    typeof version === 'number' ? version : version.toNumber()
+  );
+
+  const decryptedManifest = await Crypto.decryptProfile(
+    typeof value.toArrayBuffer === 'function' ? value.toArrayBuffer() : value,
+    storageManifestKey
+  );
+
+  return window.textsecure.protobuf.ManifestRecord.decode(decryptedManifest);
+}
+
+async function fetchManifest(
+  manifestVersion: string
+): Promise<ManifestRecordClass | undefined> {
+  window.log.info('storageService.fetchManifest');
+
+  if (!window.textsecure.messaging) {
+    throw new Error('storageService.fetchManifest: We are offline!');
+  }
+
+  try {
+    const credentials = await window.textsecure.messaging.getStorageCredentials();
+    window.storage.put('storageCredentials', credentials);
+
+    const manifestBinary = await window.textsecure.messaging.getStorageManifest(
+      {
+        credentials,
+        greaterThanVersion: manifestVersion,
+      }
+    );
+    const encryptedManifest = window.textsecure.protobuf.StorageManifest.decode(
+      manifestBinary
+    );
+
+    // if we don't get a value we're assuming that there's no newer manifest
+    if (!encryptedManifest.value || !encryptedManifest.version) {
+      window.log.info('storageService.fetchManifest: nothing changed');
+      return;
+    }
+
+    try {
+      // eslint-disable-next-line consistent-return
+      return decryptManifest(encryptedManifest);
+    } catch (err) {
+      await stopStorageServiceSync();
+      return;
+    }
+  } catch (err) {
+    window.log.error(
+      `storageService.fetchManifest: failed! ${
+        err && err.stack ? err.stack : String(err)
+      }`
+    );
+
+    if (err.code === 404) {
+      await createNewManifest();
+      return;
+    } else if (err.code === 204) {
+      // noNewerManifest we're ok
+      return;
+    }
+
+    throw err;
+  }
+}
+
+type MergeableItemType = {
+  itemType: number;
+  storageID: string;
+  storageRecord: StorageRecordClass;
+};
+
+type MergedRecordType = UnknownRecord & {
+  hasConflict: boolean;
+  isUnsupported: boolean;
+};
+
+async function mergeRecord(
+  itemToMerge: MergeableItemType
+): Promise<MergedRecordType> {
+  const { itemType, storageID, storageRecord } = itemToMerge;
+
+  const ITEM_TYPE = window.textsecure.protobuf.ManifestRecord.Identifier.Type;
+
+  let hasConflict = false;
+  let isUnsupported = false;
+
+  try {
+    if (itemType === ITEM_TYPE.UNKNOWN) {
+      window.log.info(
+        'storageService.mergeRecord: Unknown item type',
+        storageID
+      );
+    } else if (itemType === ITEM_TYPE.CONTACT && storageRecord.contact) {
+      hasConflict = await mergeContactRecord(storageID, storageRecord.contact);
+    } else if (itemType === ITEM_TYPE.GROUPV1 && storageRecord.groupV1) {
+      hasConflict = await mergeGroupV1Record(storageID, storageRecord.groupV1);
+    } else if (itemType === ITEM_TYPE.ACCOUNT && storageRecord.account) {
+      hasConflict = await mergeAccountRecord(storageID, storageRecord.account);
+    } else {
+      isUnsupported = true;
+      window.log.info(
+        `storageService.mergeRecord: Unknown record: ${itemType}::${storageID}`
+      );
+    }
+  } catch (err) {
+    window.log.error(
+      `storageService.mergeRecord: merging record failed ${storageID}`
+    );
+  }
+
+  return {
+    hasConflict,
+    isUnsupported,
+    itemType,
+    storageID,
+  };
+}
+
+/* tslint:disable-next-line max-func-body-length */
+async function processManifest(
+  manifest: ManifestRecordClass
+): Promise<boolean> {
+  const storageKeyBase64 = window.storage.get('storageKey');
+  const storageKey = base64ToArrayBuffer(storageKeyBase64);
+
+  if (!window.textsecure.messaging) {
+    throw new Error('storageService.processManifest: We are offline!');
+  }
+
+  const remoteKeysTypeMap = new Map();
+  manifest.keys.forEach((identifier: ManifestRecordIdentifierClass) => {
+    remoteKeysTypeMap.set(
+      arrayBufferToBase64(identifier.raw.toArrayBuffer()),
+      identifier.type
+    );
+  });
+
+  const localKeys = window
+    .getConversations()
+    .map((conversation: ConversationModelType) => conversation.get('storageID'))
+    .filter(Boolean);
+
+  const unknownRecordsArray =
+    window.storage.get('storage-service-unknown-records') || [];
+
+  unknownRecordsArray.forEach((record: UnknownRecord) => {
+    localKeys.push(record.storageID);
+  });
+
+  window.log.info(
+    `storageService.processManifest: localKeys.length ${localKeys.length}`
+  );
+
+  const remoteKeys = Array.from(remoteKeysTypeMap.keys());
+
+  const remoteOnly = remoteKeys.filter(
+    (key: string) => !localKeys.includes(key)
+  );
+
+  window.log.info(
+    `storageService.processManifest: remoteOnly.length ${remoteOnly.length}`
+  );
+
+  const readOperation = new window.textsecure.protobuf.ReadOperation();
+  readOperation.readKey = remoteOnly.map(base64ToArrayBuffer);
+
+  const credentials = window.storage.get('storageCredentials');
+  const storageItemsBuffer = await window.textsecure.messaging.getStorageRecords(
+    readOperation.toArrayBuffer(),
+    {
+      credentials,
+    }
+  );
+
+  const storageItems = window.textsecure.protobuf.StorageItems.decode(
+    storageItemsBuffer
+  );
+
+  if (!storageItems.items) {
+    window.log.info(
+      'storageService.processManifest: No storage items retrieved'
+    );
+    return false;
+  }
+
+  const decryptedStorageItems = await pMap(
+    storageItems.items,
+    async (storageRecordWrapper: StorageItemClass) => {
+      const { key, value: storageItemCiphertext } = storageRecordWrapper;
+
+      if (!key || !storageItemCiphertext) {
+        await stopStorageServiceSync();
+        throw new Error(
+          'storageService.processManifest: Missing key and/or Ciphertext'
+        );
+      }
+
+      const base64ItemID = arrayBufferToBase64(key.toArrayBuffer());
+
+      const storageItemKey = await deriveStorageItemKey(
+        storageKey,
+        base64ItemID
+      );
+
+      let storageItemPlaintext;
+      try {
+        storageItemPlaintext = await Crypto.decryptProfile(
+          storageItemCiphertext.toArrayBuffer(),
+          storageItemKey
+        );
+      } catch (err) {
+        await stopStorageServiceSync();
+        throw err;
+      }
+
+      const storageRecord = window.textsecure.protobuf.StorageRecord.decode(
+        storageItemPlaintext
+      );
+
+      return {
+        itemType: remoteKeysTypeMap.get(base64ItemID),
+        storageID: base64ItemID,
+        storageRecord,
+      };
+    },
+    { concurrency: 50 }
+  );
+
+  try {
+    const mergedRecords = await pMap(decryptedStorageItems, mergeRecord, {
+      concurrency: 5,
+    });
+
+    const unknownRecords: Map<string, UnknownRecord> = new Map();
+    unknownRecordsArray.forEach((record: UnknownRecord) => {
+      unknownRecords.set(record.storageID, record);
+    });
+
+    const hasConflict = mergedRecords.some((mergedRecord: MergedRecordType) => {
+      if (mergedRecord.isUnsupported) {
+        unknownRecords.set(mergedRecord.storageID, {
+          itemType: mergedRecord.itemType,
+          storageID: mergedRecord.storageID,
+        });
+      }
+      return mergedRecord.hasConflict;
+    });
+
+    window.storage.put(
+      'storage-service-unknown-records',
+      Array.from(unknownRecords.values())
+    );
+
+    if (hasConflict) {
+      window.log.info(
+        'storageService.processManifest: Conflict found, uploading changes'
+      );
+
+      return true;
+    } else {
+      consecutiveConflicts = 0;
+    }
+  } catch (err) {
+    window.log.error(
+      `storageService.processManifest: failed! ${
+        err && err.stack ? err.stack : String(err)
+      }`
+    );
+  }
+
+  return false;
+}
+
+// Exported functions
+
+export async function runStorageServiceSyncJob(): Promise<void> {
+  if (!window.storage.get('storageKey')) {
+    throw new Error(
+      'storageService.runStorageServiceSyncJob: Cannot start; no storage key!'
+    );
+  }
+
+  window.log.info('storageService.runStorageServiceSyncJob: starting...');
+
+  try {
+    const localManifestVersion = window.storage.get('manifestVersion') || 0;
+    const manifest = await fetchManifest(localManifestVersion);
+
+    // Guarding against no manifests being returned, everything should be ok
+    if (!manifest) {
+      window.log.info(
+        'storageService.runStorageServiceSyncJob: no manifest, returning early'
+      );
+      return;
+    }
+
+    const version = manifest.version.toNumber();
+
+    window.log.info(
+      `storageService.runStorageServiceSyncJob: manifest versions - previous: ${localManifestVersion}, current: ${version}`
+    );
+
+    const hasConflicts = await processManifest(manifest);
+    if (hasConflicts) {
+      await storageServiceUploadJob();
+    }
+
+    window.storage.put('manifestVersion', version);
+  } catch (err) {
+    window.log.error(
+      `storageService.runStorageServiceSyncJob: error processing manifest ${
+        err && err.stack ? err.stack : String(err)
+      }`
+    );
+  }
+
+  window.log.info('storageService.runStorageServiceSyncJob: complete');
+}
+
+// Note: this function must be called at startup once we handle unknown records
+// of a certain type. This way once the runStorageServiceSyncJob function runs
+// it'll pick up the new storage IDs and process them accordingly.
+export function handleUnknownRecords(itemType: number): void {
+  const unknownRecordsArray =
+    window.storage.get('storage-service-unknown-records') || [];
+  const newUnknownRecords = unknownRecordsArray.filter(
+    (record: UnknownRecord) => record.itemType !== itemType
+  );
+  window.storage.put('storage-service-unknown-records', newUnknownRecords);
+}
+
+// Note: this function is meant to be called before ConversationController is hydrated.
+//   It goes directly to the database, so in-memory conversations will be out of date.
+export async function eraseAllStorageServiceState(): Promise<void> {
+  window.log.info('storageService.eraseAllStorageServiceState: starting...');
+  await Promise.all([
+    window.storage.remove('manifestVersion'),
+    window.storage.remove('storage-service-unknown-records'),
+    window.storage.remove('storageCredentials'),
+  ]);
+  await eraseStorageServiceStateFromConversations();
+  window.log.info('storageService.eraseAllStorageServiceState: complete');
+}
+
+async function nondebouncedStorageServiceUploadJob(): Promise<void> {
+  if (!window.storage.get('storageKey')) {
+    throw new Error(
+      'storageService.storageServiceUploadJob: Cannot start; no storage key!'
+    );
+  }
+
+  const localManifestVersion = window.storage.get('manifestVersion') || 0;
+  const version = Number(localManifestVersion) + 1;
+
+  window.log.info(
+    'storageService.storageServiceUploadJob: will update to manifest version',
+    version
+  );
+
+  try {
+    await uploadManifest(version, await generateManifest(version));
+  } catch (err) {
+    if (err.code === 409) {
+      await runStorageServiceSyncJob();
+    }
+  }
+}
+
+export const storageServiceUploadJob = _.debounce(
+  nondebouncedStorageServiceUploadJob,
+  500
+);