Add an SQS-based API for blockIp and blockEmail (fix #27)

Gruntfile.js

@@ -18,7 +18,7 @@ module.exports = function (grunt) {
         },
         src: [
           '*.js',
-          '{bin/,scripts/,test/}*'
+          '{bans/,bin/,scripts/,test/}*'
         ]
       },
       tests: {
@@ -37,7 +37,7 @@ module.exports = function (grunt) {
         reporter: require('jshint-stylish')
       },
       app: [
-        '{,bin/,scripts/,test/}*.js'
+        '{,bans/,bin/,scripts/,test/}*.js'
       ]
     }
   })

bans/handler.js

@@ -0,0 +1,80 @@
+/* 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/. */
+
+var config = require('../config')
+
+var LIFETIME = config.recordLifetimeSeconds
+var BLOCK_INTERVAL_MS = config.blockIntervalSeconds * 1000
+var RATE_LIMIT_INTERVAL_MS = config.rateLimitIntervalSeconds * 1000
+
+var EmailRecord = require('../email_record')(RATE_LIMIT_INTERVAL_MS, BLOCK_INTERVAL_MS, config.maxEmails)
+var IpRecord = require('../ip_record')(BLOCK_INTERVAL_MS)
+
+module.exports = function (mc, log) {
+
+  function blockIp(ip, cb) {
+    mc.get(ip,
+      function (err, data) {
+        if (err) {
+          log.error({ op: 'handleBan.blockIp', ip: ip, err: err })
+          return cb(err)
+        }
+
+        log.info({ op: 'handleBan.blockIp', ip: ip })
+        var ir = IpRecord.parse(data)
+        ir.block()
+        mc.set(ip, ir, LIFETIME,
+          function (err) {
+            if (err) {
+              log.error({ op: 'memcachedError', err: err })
+              return cb(err)
+            }
+            mc.end()
+            cb(null)
+          }
+        )
+      }
+    )
+  }
+
+  function blockEmail(email, cb) {
+    mc.get(email,
+      function (err, data) {
+        if (err) {
+          log.error({ op: 'handleBan.blockEmail', email: email, err: err })
+          return cb(err)
+        }
+
+        log.info({ op: 'handleBan.blockEmail', email: email })
+        var er = EmailRecord.parse(data)
+        er.block()
+        mc.set(email, er, LIFETIME,
+          function (err) {
+            if (err) {
+              log.error({ op: 'memcachedError', err: err })
+              return cb(err)
+            }
+            mc.end()
+            cb(null)
+          }
+        )
+      }
+    )
+  }
+
+  function handleBan(message, cb) {
+    if (message.ban && message.ban.ip) {
+      blockIp(message.ban.ip, cb)
+    }
+    else if (message.ban && message.ban.email) {
+      blockEmail(message.ban.email, cb)
+    }
+    else {
+      log.error({ op: 'handleBan', ban: !!message.ban })
+      cb('invalid message')
+    }
+  }
+
+  return handleBan
+}

bans/index.js

@@ -0,0 +1,17 @@
+/* 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/. */
+
+var handleBans = require('./handler')
+
+module.exports = function (log) {
+
+  var BanQueue = require('./sqs')(log)
+
+  return function start(config, mc) {
+    var banQueue = new BanQueue(config)
+    banQueue.on('data', handleBans(mc, log))
+    banQueue.start()
+    return banQueue
+  }
+}

bans/sqs.js

@@ -0,0 +1,72 @@
+/* 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/. */
+
+var AWS = require('aws-sdk')
+var inherits = require('util').inherits
+var EventEmitter = require('events').EventEmitter
+
+module.exports = function (log) {
+
+  function SQSBanQueue(config) {
+    this.sqs = new AWS.SQS({ region : config.region })
+    this.queueUrl = config.queueUrl
+    EventEmitter.call(this)
+  }
+  inherits(SQSBanQueue, EventEmitter)
+
+  function checkDeleteError(err) {
+    if (err) {
+      log.error({ op: 'sqs.deleteMessage', err: err })
+    }
+  }
+
+  SQSBanQueue.prototype.fetch = function (url) {
+    var errTimer = null
+    this.sqs.receiveMessage(
+      {
+        QueueUrl: url,
+        AttributeNames: [],
+        MaxNumberOfMessages: 10,
+        WaitTimeSeconds: 20
+      },
+      function (err, data) {
+        if (err) {
+          log.error({ op: 'sqs.fetch', url: url, err: err })
+          if (!errTimer) {
+            // unacceptable! this aws lib will call the callback
+            // more than once with different errors.
+            errTimer = setTimeout(this.fetch.bind(this, url), 2000)
+          }
+          return
+        }
+        data.Messages = data.Messages || []
+        for (var i = 0; i < data.Messages.length; i++) {
+          var msg = data.Messages[i]
+          this.sqs.deleteMessage(
+            {
+              QueueUrl: url,
+              ReceiptHandle: msg.ReceiptHandle
+            },
+            checkDeleteError
+          )
+          try {
+            var body = JSON.parse(msg.Body)
+            var message = JSON.parse(body.Message)
+            this.emit('data', message)
+          }
+          catch (e) {
+            log.error({ op: 'sqs.fetch', message: message, err: e })
+          }
+        }
+        this.fetch(url)
+      }.bind(this)
+    )
+  }
+
+  SQSBanQueue.prototype.start = function () {
+    this.fetch(this.queueUrl)
+  }
+
+  return SQSBanQueue
+}

bin/customs_server.js

@@ -39,6 +39,15 @@ var mc = new Memcached(
   }
 )
 
+var handleBan = P.promisify(require('../bans/handler')(mc, log))
+
+// optional SQS-based IP/email banning API
+if (config.bans.region && config.bans.queueUrl) {
+  var bans = require('../bans')(log)
+  bans(config.bans, mc)
+  log.info({ op: 'listening', sqsRegion: config.bans.region, sqsQueueUrl: config.bans.queueUrl })
+}
+
 var api = restify.createServer()
 api.use(restify.bodyParser())
 
@@ -201,19 +210,14 @@ api.post(
       next()
     }
 
-    mc.getAsync(email)
-      .then(EmailRecord.parse, EmailRecord.parse)
-      .then(
-        function (emailRecord) {
-          emailRecord.block()
-          return mc.setAsync(email, emailRecord, LIFETIME).catch(ignore)
-        }
-      )
+    handleBan({ ban: { email: email } })
       .then(
         function () {
           log.info({ op: 'request.blockEmail', email: email })
           res.send({})
-        },
+        }
+      )
+      .catch(
         function (err) {
           log.error({ op: 'request.blockEmail', email: email, err: err })
           res.send(500, err)
@@ -234,19 +238,14 @@ api.post(
       next()
     }
 
-    mc.getAsync(ip)
-      .then(IpRecord.parse, IpRecord.parse)
-      .then(
-        function (ipRecord) {
-          ipRecord.block()
-          return mc.setAsync(ip, ipRecord, LIFETIME).catch(ignore)
-        }
-      )
+    handleBan({ ban: { ip: ip } })
       .then(
         function () {
           log.info({ op: 'request.blockIp', ip: ip })
           res.send({})
-        },
+        }
+      )
+      .catch(
         function (err) {
           log.error({ op: 'request.blockIp', ip: ip, err: err })
           res.send(500, err)

config.js

@@ -12,6 +12,10 @@ module.exports = require('rc')(
     blockIntervalSeconds: 60 * 60 * 24, // duration of a manual ban
     rateLimitIntervalSeconds: 60 * 15,  // duration of automatic throttling
     maxEmails: 3,   // number of emails sent within rateLimitIntervalSeconds before throttling
-    maxBadLogins: 2 // number failed login attempts within rateLimitIntervalSeconds before throttling
-  }
+    maxBadLogins: 2, // number failed login attempts within rateLimitIntervalSeconds before throttling
+    bans: {
+      region: '',
+      queueUrl: ''
+    }
+ }
 )

package.json

@@ -16,6 +16,7 @@
     "test": "scripts/test-local.sh"
   },
   "dependencies": {
+    "aws-sdk": "2.0.0-rc.19",
     "bluebird": "1.2.2",
     "bunyan": "0.22.3",
     "memcached": "0.2.8",

test/local/ban_tests.js

@@ -0,0 +1,141 @@
+/* Any copyright is dedicated to the Public Domain.
+ * http://creativecommons.org/publicdomain/zero/1.0/ */
+
+require('ass')
+var test = require('tap').test
+var banHandler = require('../../bans/handler')
+var mcHelper = require('../memcache-helper')
+
+var log = {
+  info: function () {},
+  error: function () {}
+}
+
+var config = {
+  blockIntervalSeconds: 1
+}
+
+var TEST_IP = '192.0.2.1'
+var TEST_EMAIL = 'test@example.com'
+
+test(
+  'clear everything',
+  function (t) {
+    mcHelper.clearEverything(
+      function (err) {
+        t.notOk(err, 'no errors were returned')
+        t.end()
+      }
+    )
+  }
+)
+
+test(
+  'well-formed ip blocking request',
+  function (t) {
+    var message = {
+      ban: {
+        ip: TEST_IP
+      }
+    }
+    banHandler(mcHelper.mc, log)(message,
+      function (err) {
+        t.notOk(err, 'no errors were returned')
+
+        mcHelper.blockedIpCheck(
+          function (isBlocked) {
+            t.equal(isBlocked, true, 'ip is blocked')
+            t.end()
+          }
+        )
+      }
+    )
+  }
+)
+
+test(
+  'ip block has expired',
+  function (t) {
+    setTimeout(
+      function () {
+        mcHelper.blockedIpCheck(
+          function (isBlocked) {
+            t.equal(isBlocked, false, 'ip is not blocked')
+            t.end()
+          }
+        )
+      },
+      config.blockIntervalSeconds * 1000
+    )
+  }
+)
+
+test(
+  'well-formed email blocking request',
+  function (t) {
+    var message = {
+      ban: {
+        email: TEST_EMAIL
+      }
+    }
+    banHandler(mcHelper.mc, log)(message,
+      function (err) {
+        t.notOk(err, 'no errors were returned')
+
+        mcHelper.blockedEmailCheck(
+          function (isBlocked) {
+            t.equal(isBlocked, true, 'email is blocked')
+            t.end()
+          }
+        )
+      }
+    )
+  }
+)
+
+test(
+  'email block has expired',
+  function (t) {
+    setTimeout(
+      function () {
+        mcHelper.blockedEmailCheck(
+          function (isBlocked) {
+            t.equal(isBlocked, false, 'email is not blocked')
+            t.end()
+          }
+        )
+      },
+      config.blockIntervalSeconds * 1000
+    )
+  }
+)
+
+test(
+  'missing ip and email',
+  function (t) {
+    var message = {
+      ban: {
+      }
+    }
+    banHandler(mcHelper.mc, log)(message,
+      function (err) {
+        t.equal(err, 'invalid message')
+        t.end()
+      }
+    )
+  }
+)
+
+test(
+  'missing ban',
+  function (t) {
+    var message = {
+    }
+    banHandler(mcHelper.mc, log)(message,
+      function (err) {
+        t.equal(err, 'invalid message')
+        t.end()
+      }
+    )
+  }
+)