83603332-15e8-4ee3-acad-c8c42f47f458

Build # 83603332-15e8-4ee3-acad-c8c42f47f458

Build Type

circleci

Committed by Robert Helmer

Commit Message

MNTOR-1452 - use UPDATE FOR to lock table and prevent race condition when adding email

Pull Request Pull Request #2982: MNTOR-1452 - use SELECT FOR UPDATE to lock rows and prevent race condition when adding email

Run Details

282 of 1679 branches covered (16.8%)

Branch coverage included in aggregate %.

3 of 3 new or added lines in 1 file covered. (100.0%)

959 of 4548 relevant lines covered (21.09%)

3.52 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/src/db/tables/emailAddresses.js

/* This Source Code Form is subject to the terms of the Mozilla Public
 * License, v. 2.0. If a copy of the MPL was not distributed with this
 * file, You can obtain one at http://mozilla.org/MPL/2.0/. */

import { v4 as uuidv4 } from 'uuid'
import Knex from 'knex'
import knexConfig from '../knexfile.js'
import mozlog from '../../utils/log.js'
import { fluentError } from '../../utils/fluent.js'
import { subscribeHash } from '../../utils/hibp.js'
import { getSha1 } from '../../utils/fxa.js'
import { getSubscriberByEmail, updateFxAData } from './subscribers.js'
const knex = Knex(knexConfig)
const log = mozlog('DB.email_addresses')

async function getEmailByToken (token) {
  const res = await knex('email_addresses')
    .where('verification_token', '=', token)

  return res[0]
}

async function getEmailById (emailAddressId) {
  const res = await knex('email_addresses')
    .where('id', '=', emailAddressId)

  return res[0]
}

async function getEmailAddressRecordByEmail (email) {
  const emailAddresses = await knex('email_addresses').where({
    email, verified: true
  })
  if (!emailAddresses) {
    return null
  }
  if (emailAddresses.length > 1) {
    // TODO: handle multiple emails in separate(?) subscriber accounts?
    log.warn('getEmailAddressRecordByEmail', { msg: 'found the same email multiple times' })
  }
  return emailAddresses[0]
}

async function addSubscriberUnverifiedEmailHash (user, email) {
  const res = await knex.transaction(async trx => {
    return knex('email_addresses')
      .transacting(trx)
      .forUpdate()
      .select({
        subscriber_id: user.id
      })
      .insert({
        subscriber_id: user.id,
        email,
        sha1: getSha1(email),
        verification_token: uuidv4(),
        verified: false
      }).returning('*')
  })
  return await res[0]
}

async function resetUnverifiedEmailAddress (emailAddressId) {
  const newVerificationToken = uuidv4()

  // Time in ms to require between verification reset.
  const verificationWait = 5 * 60 * 1000 // 5 minutes

  const verificationRecentlyUpdated = await knex('email_addresses')
    .select('id')
    .whereRaw('"updated_at" > NOW() - INTERVAL \'1 MILLISECOND\' * ?', verificationWait)
    .andWhere('id', emailAddressId)
    .first()

  if (verificationRecentlyUpdated?.id === parseInt(emailAddressId)) {
    throw fluentError('error-email-validation-pending')
  }

  const res = await knex('email_addresses')
    .update({
      verification_token: newVerificationToken,
      updated_at: knex.fn.now()
    })
    .where('id', emailAddressId)
    .returning('*')
  return res[0]
}

async function verifyEmailHash (token) {
  const unverifiedEmail = await getEmailByToken(token)
  if (!unverifiedEmail) {
    throw fluentError('Error message for this verification email timed out or something went wrong.')
  }
  const verifiedEmail = await _verifyNewEmail(unverifiedEmail)
  return verifiedEmail[0]
}

// TODO: refactor into an upsert? https://jaketrent.com/post/upsert-knexjs/
// Used internally, ideally should not be called by consumers.
async function _getSha1EntryAndDo (sha1, aFoundCallback, aNotFoundCallback) {
  const existingEntries = await knex('subscribers')
    .where('primary_sha1', sha1)

  if (existingEntries.length && aFoundCallback) {
    return await aFoundCallback(existingEntries[0])
  }

  if (!existingEntries.length && aNotFoundCallback) {
    return await aNotFoundCallback()
  }
}

// Used internally.
async function _addEmailHash (sha1, email, signupLanguage, verified = false) {
  log.debug('_addEmailHash', { sha1, email, signupLanguage, verified })
  try {
    return await _getSha1EntryAndDo(sha1, async aEntry => {
      // Entry existed, patch the email value if supplied.
      if (email) {
        const res = await knex('subscribers')
          .update({
            primary_email: email,
            primary_sha1: getSha1(email.toLowerCase()),
            primary_verified: verified,
            updated_at: knex.fn.now()
          })
          .where('id', '=', aEntry.id)
          .returning('*')
        return res[0]
      }

      return aEntry
    }, async () => {
      // Always add a verification_token value
      const verificationToken = uuidv4()
      const res = await knex('subscribers')
        .insert({ primary_sha1: getSha1(email.toLowerCase()), primary_email: email, signup_language: signupLanguage, primary_verification_token: verificationToken, primary_verified: verified })
        .returning('*')
      return res[0]
    })
  } catch (e) {
    log.error(e)
    throw fluentError('error-could-not-add-email')
  }
}

/**
 * Add a subscriber:
 * 1. Add a record to subscribers
 * 2. Immediately call _verifySubscriber
 * 3. For FxA subscriber, add refresh token and profile data
 *
 * @param {string} email to add
 * @param {string} signupLanguage from Accept-Language
 * @param {string} fxaAccessToken from Firefox Account Oauth
 * @param {string} fxaRefreshToken from Firefox Account Oauth
 * @param {string} fxaProfileData from Firefox Account
 * @returns {object} subscriber knex object added to DB
 */
async function addSubscriber (email, signupLanguage, fxaAccessToken = null, fxaRefreshToken = null, fxaProfileData = null) {
  console.log({ email })
  console.log({ signupLanguage })
  const emailHash = await _addEmailHash(getSha1(email), email, signupLanguage, true)
  const verified = await _verifySubscriber(emailHash)
  const verifiedSubscriber = Array.isArray(verified) ? verified[0] : null
  if (fxaRefreshToken || fxaProfileData) {
    return updateFxAData(verifiedSubscriber, fxaAccessToken, fxaRefreshToken, fxaProfileData)
  }
  return verifiedSubscriber
}

/**
 * When an email is verified, convert it into a subscriber:
 * 1. Subscribe the hash to HIBP
 * 2. Update our subscribers record to verified
 * 3. (if opted in) Subscribe the email to Fx newsletter
 *
 * @param {object} emailHash knex object in DB
 * @returns {object} verified subscriber knex object in DB
 */
async function _verifySubscriber (emailHash) {
  await subscribeHash(emailHash.primary_sha1)

  const verifiedSubscriber = await knex('subscribers')
    .where('primary_email', '=', emailHash.primary_email)
    .update({
      primary_verified: true,
      updated_at: knex.fn.now()
    })
    .returning('*')

  return verifiedSubscriber
}

// Verifies new emails added by existing users
async function _verifyNewEmail (emailHash) {
  await subscribeHash(emailHash.sha1)

  const verifiedEmail = await knex('email_addresses')
    .where('id', '=', emailHash.id)
    .update({
      verified: true
    })
    .returning('*')

  return verifiedEmail
}

/**
 * @typedef {object} EmailRow Email data, as returned from the database table `email_addresses`
 * @property {string} email
 * @property {string} sha1
 * @property {boolean} verified
 */

/**
 * @param {number} userId
 * @returns {Promise<EmailRow[]>}
 */
async function getUserEmails (userId) {
  const userEmails = await knex('email_addresses')
    .where('subscriber_id', '=', userId)
    .returning('*')

  return userEmails
}

// This is used by SES callbacks to remove email addresses when recipients
// perma-bounce or mark our emails as spam
// Removes from either subscribers or email_addresses as necessary
async function removeEmail (email) {
  const subscriber = await getSubscriberByEmail(email)
  if (!subscriber) {
    const emailAddress = await getEmailAddressRecordByEmail(email)
    if (!emailAddress) {
      log.warn('removed-subscriber-not-found')
      return
    }
    await knex('email_addresses')
      .where({
        email,
        verified: true
      })
      .del()
    return
  }
  // This can fail if a subscriber has more email_addresses and marks
  // a primary email as spam, but we should let it fail so we can see it
  // in the logs
  await knex('subscribers')
    .where({
      primary_verification_token: subscriber.primary_verification_token,
      primary_sha1: subscriber.primary_sha1
    })
    .del()
}

async function removeOneSecondaryEmail (emailId) {
  await knex('email_addresses')
    .where({
      id: emailId
    })
    .del()
}

async function getEmailAddressesByHashes (hashes) {
  return await knex('email_addresses')
    .join('subscribers', 'email_addresses.subscriber_id', '=', 'subscribers.id')
    .whereIn('email_addresses.sha1', hashes)
    .andWhere('email_addresses.verified', '=', true)
}

async function deleteEmailAddressesByUid (uid) {
  await knex('email_addresses').where({ subscriber_id: uid }).del()
}

export {
  getEmailByToken,
  getEmailById,
  getEmailAddressRecordByEmail,
  addSubscriberUnverifiedEmailHash,
  resetUnverifiedEmailAddress,
  verifyEmailHash,
  addSubscriber,
  getUserEmails,
  removeEmail,
  removeOneSecondaryEmail,
  getEmailAddressesByHashes,
  deleteEmailAddressesByUid
}

1	/* This Source Code Form is subject to the terms of the Mozilla Public
2	* License, v. 2.0. If a copy of the MPL was not distributed with this
3	* file, You can obtain one at http://mozilla.org/MPL/2.0/. */
4
5	import { v4 as uuidv4 } from 'uuid'
6	import Knex from 'knex'
7	import knexConfig from '../knexfile.js'
8	import mozlog from '../../utils/log.js'
9	import { fluentError } from '../../utils/fluent.js'
10	import { subscribeHash } from '../../utils/hibp.js'
11	import { getSha1 } from '../../utils/fxa.js'
12	import { getSubscriberByEmail, updateFxAData } from './subscribers.js'
13	const knex = Knex(knexConfig)	×
14	const log = mozlog('DB.email_addresses')	×
15
16	async function getEmailByToken (token) {
17	const res = await knex('email_addresses')	×
18	.where('verification_token', '=', token)
19
20	return res[0]	×
21	}
22
23	async function getEmailById (emailAddressId) {
24	const res = await knex('email_addresses')	×
25	.where('id', '=', emailAddressId)
26
27	return res[0]	×
28	}
29
30	async function getEmailAddressRecordByEmail (email) {
31	const emailAddresses = await knex('email_addresses').where({	×
32	email, verified: true
33	})
34	if (!emailAddresses) {	×
35	return null	×
36	}
37	if (emailAddresses.length > 1) {	×
38	// TODO: handle multiple emails in separate(?) subscriber accounts?
39	log.warn('getEmailAddressRecordByEmail', { msg: 'found the same email multiple times' })	×
40	}
41	return emailAddresses[0]	×
42	}
43
44	async function addSubscriberUnverifiedEmailHash (user, email) {
45	const res = await knex.transaction(async trx => {	×
46	return knex('email_addresses')	×
47	.transacting(trx)
48	.forUpdate()
49	.select({
50	subscriber_id: user.id
51	})
52	.insert({
53	subscriber_id: user.id,
54	email,
55	sha1: getSha1(email),
56	verification_token: uuidv4(),
57	verified: false
58	}).returning('*')
59	})
60	return await res[0]	×
61	}
62
63	async function resetUnverifiedEmailAddress (emailAddressId) {
64	const newVerificationToken = uuidv4()	×
65
66	// Time in ms to require between verification reset.
67	const verificationWait = 5 * 60 * 1000 // 5 minutes	×
68
69	const verificationRecentlyUpdated = await knex('email_addresses')	×
70	.select('id')
71	.whereRaw('"updated_at" > NOW() - INTERVAL \'1 MILLISECOND\' * ?', verificationWait)
72	.andWhere('id', emailAddressId)
73	.first()
74
75	if (verificationRecentlyUpdated?.id === parseInt(emailAddressId)) {	×
76	throw fluentError('error-email-validation-pending')	×
77	}
78
79	const res = await knex('email_addresses')	×
80	.update({
81	verification_token: newVerificationToken,
82	updated_at: knex.fn.now()
83	})
84	.where('id', emailAddressId)
85	.returning('*')
86	return res[0]	×
87	}
88
89	async function verifyEmailHash (token) {
90	const unverifiedEmail = await getEmailByToken(token)	×
91	if (!unverifiedEmail) {	×
92	throw fluentError('Error message for this verification email timed out or something went wrong.')	×
93	}
94	const verifiedEmail = await _verifyNewEmail(unverifiedEmail)	×
95	return verifiedEmail[0]	×
96	}
97
98	// TODO: refactor into an upsert? https://jaketrent.com/post/upsert-knexjs/
99	// Used internally, ideally should not be called by consumers.
100	async function _getSha1EntryAndDo (sha1, aFoundCallback, aNotFoundCallback) {
101	const existingEntries = await knex('subscribers')	×
102	.where('primary_sha1', sha1)
103
104	if (existingEntries.length && aFoundCallback) {	×
105	return await aFoundCallback(existingEntries[0])	×
106	}
107
108	if (!existingEntries.length && aNotFoundCallback) {	×
109	return await aNotFoundCallback()	×
110	}
111	}
112
113	// Used internally.
114	async function _addEmailHash (sha1, email, signupLanguage, verified = false) {	×
115	log.debug('_addEmailHash', { sha1, email, signupLanguage, verified })	×
116	try {	×
117	return await _getSha1EntryAndDo(sha1, async aEntry => {	×
118	// Entry existed, patch the email value if supplied.
119	if (email) {	×
120	const res = await knex('subscribers')	×
121	.update({
122	primary_email: email,
123	primary_sha1: getSha1(email.toLowerCase()),
124	primary_verified: verified,
125	updated_at: knex.fn.now()
126	})
127	.where('id', '=', aEntry.id)
128	.returning('*')
129	return res[0]	×
130	}
131
132	return aEntry	×
133	}, async () => {
134	// Always add a verification_token value
135	const verificationToken = uuidv4()	×
136	const res = await knex('subscribers')	×
137	.insert({ primary_sha1: getSha1(email.toLowerCase()), primary_email: email, signup_language: signupLanguage, primary_verification_token: verificationToken, primary_verified: verified })
138	.returning('*')
139	return res[0]	×
140	})
141	} catch (e) {
142	log.error(e)	×
143	throw fluentError('error-could-not-add-email')	×
144	}
145	}
146
147	/**
148	* Add a subscriber:
149	* 1. Add a record to subscribers
150	* 2. Immediately call _verifySubscriber
151	* 3. For FxA subscriber, add refresh token and profile data
152	*
153	* @param {string} email to add
154	* @param {string} signupLanguage from Accept-Language
155	* @param {string} fxaAccessToken from Firefox Account Oauth
156	* @param {string} fxaRefreshToken from Firefox Account Oauth
157	* @param {string} fxaProfileData from Firefox Account
158	* @returns {object} subscriber knex object added to DB
159	*/
160	async function addSubscriber (email, signupLanguage, fxaAccessToken = null, fxaRefreshToken = null, fxaProfileData = null) {	×
161	console.log({ email })	×
162	console.log({ signupLanguage })	×
163	const emailHash = await _addEmailHash(getSha1(email), email, signupLanguage, true)	×
164	const verified = await _verifySubscriber(emailHash)	×
165	const verifiedSubscriber = Array.isArray(verified) ? verified[0] : null	×
166	if (fxaRefreshToken \|\| fxaProfileData) {	×
167	return updateFxAData(verifiedSubscriber, fxaAccessToken, fxaRefreshToken, fxaProfileData)	×
168	}
169	return verifiedSubscriber	×
170	}
171
172	/**
173	* When an email is verified, convert it into a subscriber:
174	* 1. Subscribe the hash to HIBP
175	* 2. Update our subscribers record to verified
176	* 3. (if opted in) Subscribe the email to Fx newsletter
177	*
178	* @param {object} emailHash knex object in DB
179	* @returns {object} verified subscriber knex object in DB
180	*/
181	async function _verifySubscriber (emailHash) {
182	await subscribeHash(emailHash.primary_sha1)	×
183
184	const verifiedSubscriber = await knex('subscribers')	×
185	.where('primary_email', '=', emailHash.primary_email)
186	.update({
187	primary_verified: true,
188	updated_at: knex.fn.now()
189	})
190	.returning('*')
191
192	return verifiedSubscriber	×
193	}
194
195	// Verifies new emails added by existing users
196	async function _verifyNewEmail (emailHash) {
197	await subscribeHash(emailHash.sha1)	×
198
199	const verifiedEmail = await knex('email_addresses')	×
200	.where('id', '=', emailHash.id)
201	.update({
202	verified: true
203	})
204	.returning('*')
205
206	return verifiedEmail	×
207	}
208
209	/**
210	* @typedef {object} EmailRow Email data, as returned from the database table `email_addresses`
211	* @property {string} email
212	* @property {string} sha1
213	* @property {boolean} verified
214	*/
215
216	/**
217	* @param {number} userId
218	* @returns {Promise<EmailRow[]>}
219	*/
220	async function getUserEmails (userId) {
221	const userEmails = await knex('email_addresses')	×
222	.where('subscriber_id', '=', userId)
223	.returning('*')
224
225	return userEmails	×
226	}
227
228	// This is used by SES callbacks to remove email addresses when recipients
229	// perma-bounce or mark our emails as spam
230	// Removes from either subscribers or email_addresses as necessary
231	async function removeEmail (email) {
232	const subscriber = await getSubscriberByEmail(email)	×
233	if (!subscriber) {	×
234	const emailAddress = await getEmailAddressRecordByEmail(email)	×
235	if (!emailAddress) {	×
236	log.warn('removed-subscriber-not-found')	×
237	return	×
238	}
239	await knex('email_addresses')	×
240	.where({
241	email,
242	verified: true
243	})
244	.del()
245	return	×
246	}
247	// This can fail if a subscriber has more email_addresses and marks
248	// a primary email as spam, but we should let it fail so we can see it
249	// in the logs
250	await knex('subscribers')	×
251	.where({
252	primary_verification_token: subscriber.primary_verification_token,
253	primary_sha1: subscriber.primary_sha1
254	})
255	.del()
256	}
257
258	async function removeOneSecondaryEmail (emailId) {
259	await knex('email_addresses')	×
260	.where({
261	id: emailId
262	})
263	.del()
264	}
265
266	async function getEmailAddressesByHashes (hashes) {
267	return await knex('email_addresses')	×
268	.join('subscribers', 'email_addresses.subscriber_id', '=', 'subscribers.id')
269	.whereIn('email_addresses.sha1', hashes)
270	.andWhere('email_addresses.verified', '=', true)
271	}
272
273	async function deleteEmailAddressesByUid (uid) {
274	await knex('email_addresses').where({ subscriber_id: uid }).del()	×
275	}
276
277	export {
278	getEmailByToken,
279	getEmailById,
280	getEmailAddressRecordByEmail,
281	addSubscriberUnverifiedEmailHash,
282	resetUnverifiedEmailAddress,
283	verifyEmailHash,
284	addSubscriber,
285	getUserEmails,
286	removeEmail,
287	removeOneSecondaryEmail,
288	getEmailAddressesByHashes,
289	deleteEmailAddressesByUid
290	}

mozilla / blurts-server / 83603332-15e8-4ee3-acad-c8c42f47f458

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous