feat(dataflow): Log if DataFlow declares a block that FxA does not have.

lib/dataflow.js

@@ -18,7 +18,7 @@
  *    rules from the content server and rely upon DataFlow.
  */
 
-module.exports = function (config, log) {
+module.exports = function (config, log, fetchRecords) {
   if (! config.dataflow.enabled) {
     // no-op if not enabled
     return
@@ -59,6 +59,21 @@ module.exports = function (config, log) {
   const messageHandler = message => {
     logMessage(message)
 
+    let metadata
+
+    try {
+      metadata = parseMessage(message)
+    } catch (err) {
+      log.error({
+        op: 'dataflow.message.invalid',
+        reason: err.message
+      })
+    }
+
+    if (metadata) {
+      checkForUnexpectedRateLimit(metadata)
+    }
+
     message.ack()
   }
 
@@ -74,4 +89,97 @@ module.exports = function (config, log) {
   }
 
   subscription.on('error', logError)
+
+  function logUnexpectedRateLimit(category) {
+    log.info({
+      op: 'dataflow.unexpected_block',
+      customs_category: category
+    })
+  }
+
+  async function checkForUnexpectedRateLimit(metadata) {
+    const {
+      customs_category: category,
+      accountid: email,
+      sourceaddress: ip,
+      uid
+    } = metadata
+
+    if (! category) {
+      log.error({
+        op: 'dataflow.customs_category.missing'
+      })
+      return
+    }
+
+    const {
+      ipRecord,
+      emailRecord,
+      ipEmailRecord,
+      uidRecord,
+     } = await fetchRecords({ ip, email, uid })
+
+    switch (category) {
+    case 'rl_login_failure_sourceaddress_accountid':
+      if (! (ipEmailRecord && ipEmailRecord.shouldBlock())) {
+        logUnexpectedRateLimit(category)
+      }
+      break
+    case 'rl_sms_sourceaddress':
+      if (! (ipRecord && ipRecord.shouldBlock())) {
+        logUnexpectedRateLimit(category)
+      }
+      break
+    case 'rl_sms_accountid':
+      if (! (emailRecord && emailRecord.shouldBlock())) {
+        logUnexpectedRateLimit(category)
+      }
+      break
+    case 'rl_email_recipient':
+      if (! (emailRecord && emailRecord.shouldBlock())) {
+        logUnexpectedRateLimit(category)
+      }
+      break
+    case 'rl_statuscheck':
+      if (! (ipRecord && ipRecord.shouldBlock())) {
+        logUnexpectedRateLimit(category)
+      }
+      break
+    case 'rl_verifycode_sourceaddress_accountid':
+      if (! (uidRecord && uidRecord.isRateLimited())) {
+        logUnexpectedRateLimit(category)
+      }
+      break
+    default:
+      log.error({
+        op: 'dataflow.customs_category.unknown',
+        customs_category: category
+      })
+    }
+  }
+}
+
+function parseMessage (message) {
+  if (! message || ! message.data) {
+    throw new Error('parse error')
+  }
+  // message.data is a Buffer
+  const body = message.data.toString()
+  const parsedBody = JSON.parse(body)
+
+  if (! parsedBody.metadata) {
+    throw new Error('metadata missing')
+  }
+
+  if (! parsedBody.metadata.forEach) {
+    throw new Error('invalid metadata')
+  }
+
+  const metadata = {}
+
+  parsedBody.metadata.forEach(entry => {
+    metadata[entry.key] = entry.value
+  })
+
+  return metadata
 }

lib/server.js

@@ -21,8 +21,6 @@ const dataflow = require('./dataflow')
 // from the given config.
 
 module.exports = function createServer(config, log) {
-  dataflow(config, log)
-
   var startupDefers = []
 
   // Setup blocklist manager
@@ -67,6 +65,8 @@ module.exports = function createServer(config, log) {
 
   const checkUserDefinedRateLimitRules = require('./user_defined_rules')(config, fetchRecord, setRecords)
 
+  dataflow(config, log, fetchRecords)
+
   if (config.updatePollIntervalSeconds) {
     [
       allowedEmailDomains,

test/local/dataflow_tests.js

@@ -51,7 +51,7 @@ test(
       }
     }
 
-    dataflow(config, log)
+    dataflow(config, log, async () => {})
 
     t.equal(PubSub.PubSub.called, false)
   }
@@ -68,7 +68,7 @@ test(
       }
     }
 
-    t.throws(() => dataflow(config, log))
+    t.throws(() => dataflow(config, log, async () => {}))
   }
 )
 
@@ -83,7 +83,7 @@ test(
       }
     }
 
-    t.throws(() => dataflow(config, log))
+    t.throws(() => dataflow(config, log, async () => {}))
   }
 )
 
@@ -99,7 +99,7 @@ test(
       }
     }
 
-    dataflow(config, log)
+    dataflow(config, log, async () => {})
   }
 )
 
@@ -120,19 +120,21 @@ test('subscription messages are listened for', async (t) => {
 test('subscription messages are acked and logged', async (t) => {
   const messageHandler = subscription.on.args[0][1]
 
+  const message1Data = JSON.stringify({ metadata: [{ key: 'customs_category', value: 'bar' }]})
   const message1Mock = {
     ack: sinon.spy(),
     id: 'message1',
-    data: Buffer.from('wibble'),
+    data: Buffer.from(message1Data),
     attributes: {
       quix: 'quux'
     }
   }
 
+  const message2Data = JSON.stringify({ metadata: [{ key: 'customs_category', value: 'womble' }]})
   const message2Mock = {
     ack: sinon.spy(),
     id: 'message2',
-    data: Buffer.from('wobble'),
+    data: Buffer.from(message2Data),
     attributes: {
       garply: 'waldo'
     }
@@ -144,7 +146,7 @@ test('subscription messages are acked and logged', async (t) => {
     op: 'fxa.customs.dataflow.message',
     count: 0,
     id: 'message1',
-    data: 'wibble',
+    data: message1Data,
     attributes: {
       quix: 'quux'
     }
@@ -159,7 +161,7 @@ test('subscription messages are acked and logged', async (t) => {
     op: 'fxa.customs.dataflow.message',
     count: 1,
     id: 'message2',
-    data: 'wobble',
+    data: message2Data,
     attributes: {
       garply: 'waldo'
     }