From 9e05463bbaa9c90b0dc136d0d4144f526b21d712 Mon Sep 17 00:00:00 2001
From: Shane Tomlinson <stomlinson@mozilla.com>
Date: Wed, 20 Mar 2019 12:48:47 +0000
Subject: [PATCH] feat(dataflow): Log if DataFlow declares a block that FxA
 does not have.

---
 lib/dataflow.js              | 110 ++++++++++++++++++++++++++++++++++-
 lib/server.js                |   4 +-
 test/local/dataflow_tests.js |  18 +++---
 3 files changed, 121 insertions(+), 11 deletions(-)

diff --git a/lib/dataflow.js b/lib/dataflow.js
index d0a21ca..6913cd5 100644
--- a/lib/dataflow.js
+++ b/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
 }
diff --git a/lib/server.js b/lib/server.js
index f5eb484..6d4ec68 100755
--- a/lib/server.js
+++ b/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,
diff --git a/test/local/dataflow_tests.js b/test/local/dataflow_tests.js
index 7150016..f2d5a35 100644
--- a/test/local/dataflow_tests.js
+++ b/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'
     }