add test, use native mask unmask module

packages/websub-hub/index.js

@@ -11,13 +11,13 @@ const Boom = require('boom')
 const Hoek = require('hoek')
 const Pino = require('pino')
 const Serializer = require('./lib/serializer')
-const safeEqual = require('./lib/safeEqual')
 const Crypto = require('crypto')
 const PEvent = require('p-event')
 const Stream = require('stream')
 const WebSocket = require('./plugins/websocket')
 const PQueue = require('p-queue')
 const Utils = require('./lib/utils')
+const WS = require('ws')
 
 module.exports = build
 
@@ -28,7 +28,7 @@ const wsCodes = {
   WSH_INTERNAL_ERROR: 'WSH_INTERNAL_ERROR',
   WSH_SUBSCRIPTION_NOT_EXISTS: 'WSH_SUBSCRIPTION_NOT_EXISTS'
 }
-const verifiedState = {
+const verificationState = {
   ACCEPTED: 'ACCEPTED',
   DECLINED: 'DECLINED',
   HTTP_ERROR: 'HTTP_ERROR'
@@ -117,7 +117,7 @@ function WebSubHub(options) {
           this.log.error(
             'cannot open ws connection because subscription does not exists'
           )
-          if (client.readyState === WebSocket.OPEN) {
+          if (client.readyState === WS.OPEN) {
             client.send(
               JSON.stringify({ code: wsCodes.WSH_SUBSCRIPTION_NOT_EXISTS })
             )
@@ -130,7 +130,7 @@ function WebSubHub(options) {
         this._wsClients.set(key, client)
       } catch (err) {
         this.log.error(err, 'connection could not be accepted')
-        if (client.readyState === WebSocket.OPEN) {
+        if (client.readyState === WS.OPEN) {
           client.send(JSON.stringify({ code: wsCodes.WSH_INTERNAL_ERROR }))
         }
         client.terminate()
@@ -223,19 +223,19 @@ WebSubHub.prototype._verifyIntent = async function(
       callbackUrl,
       err.statusCode
     )
-    return verifiedState.HTTP_ERROR
+    return verificationState.HTTP_ERROR
   }
 
   if (
     response.body &&
     response.body['hub.challenge'] &&
     challenge &&
-    safeEqual(response.body['hub.challenge'], challenge)
+    Utils.safeEqual(response.body['hub.challenge'], challenge)
   ) {
-    return verifiedState.ACCEPTED
+    return verificationState.ACCEPTED
   }
 
-  return verifiedState.DECLINED
+  return verificationState.DECLINED
 }
 
 WebSubHub.prototype._distributeContentHTTP = async function(
@@ -313,7 +313,7 @@ WebSubHub.prototype._sendContentWS = function(sub, data, headers) {
     } catch (err) {
       this.log.error(
         err,
-        `content could not be published over WS to '%s'`,
+        `content could not be send over WS to '%s'`,
         sub.callbackUrl
       )
     }
@@ -429,10 +429,10 @@ WebSubHub.prototype._handleSubscriptionRequest = async function(req, reply) {
 
   const intentResult = await this._verifyIntent(sub, challenge)
 
-  if (intentResult === verifiedState.DECLINED) {
+  if (intentResult === verificationState.DECLINED) {
     reply.send(Boom.forbidden('subscriber has declined'))
     return
-  } else if (intentResult === verifiedState.HTTP_ERROR) {
+  } else if (intentResult === verificationState.HTTP_ERROR) {
     reply.send(Boom.forbidden('subscriber could not be verified'))
     return
   }
@@ -483,13 +483,17 @@ WebSubHub.prototype._unsubscribe = async function(sub) {
   })
 
   if (this.options.ws) {
-    const key = this.server.websocketClientKey(sub.topic, sub.callbackUrl)
+    this._unsubscribeWS(sub)
+  }
+}
 
-    if (this._wsClients.has(key)) {
-      const ws = this._wsClients.get(key)
-      ws.terminate()
-      this._wsClients.delete(key)
-    }
+WebSubHub.prototype._unsubscribeWS = async function(sub) {
+  const key = this.server.websocketClientKey(sub.topic, sub.callbackUrl)
+
+  if (this._wsClients.has(key)) {
+    const ws = this._wsClients.get(key)
+    ws.terminate()
+    this._wsClients.delete(key)
   }
 }
 

packages/websub-hub/lib/utils.js

@@ -15,3 +15,25 @@ module.exports.normalizeUrl = function(url) {
     query: parsedUrl.query
   }
 }
+
+module.exports.safeEqual = function(a, b) {
+  if (typeof a !== 'string' || typeof b !== 'string') {
+    return a === b
+  }
+
+  var maxLength = Math.max(a.length, b.length)
+
+  // xor strings for security
+  var mismatch = 0
+  for (var i = 0; i < maxLength; ++i) {
+    mismatch |= a.charCodeAt(i) ^ b.charCodeAt(i)
+
+    // check after for perf, we don't want to
+    // re-enter the loop if we have a failure.
+    if (mismatch > 0) {
+      break
+    }
+  }
+
+  return !mismatch
+}

test/websocket-content-distribution.spec.js

@@ -121,4 +121,62 @@ describe('Websocket Content Distribution', function() {
           .catch(done)
       })
   })
+
+  it('Should return error when subscription was not created before', function(done) {
+    const callbackUrl = 'http://127.0.0.1:3002'
+    const createSubscriptionBody = {
+      'hub.callback': callbackUrl,
+      'hub.mode': 'subscribe',
+      'hub.topic': topic + '/feeds',
+      'hub.ws': true
+    }
+
+    const verifyIntentMock = Nock(callbackUrl)
+      .get('/')
+      .query(true)
+      .reply(uri => {
+        const query = parse(uri, true).query
+        return [
+          200,
+          { ...createSubscriptionBody, 'hub.challenge': query['hub.challenge'] }
+        ]
+      })
+
+    const topicContentMock = Nock(topic)
+      .get('/feeds')
+      .query(true)
+      .reply(200, blogFeeds)
+
+    const ws = new WebSocket(`ws://localhost:${PORT}/`, {
+      origin: `ws://localhost:${PORT}/`,
+      headers: {
+        'x-hub-topic': createSubscriptionBody['hub.topic'],
+        'x-hub-callback': createSubscriptionBody['hub.callback']
+      }
+    })
+
+    PEvent(ws, 'open')
+      .then(() => {
+        ws.once('message', function incoming(data) {
+          const msg = JSON.parse(data)
+          expect(msg.code).to.be.equals('WSH_SUBSCRIPTION_NOT_EXISTS')
+          done()
+        })
+      })
+      .then(() => {
+        return Got.post(`http://localhost:${PORT}/publish`, {
+          form: true,
+          body: {
+            'hub.mode': 'publish',
+            'hub.url': topic + '/feeds'
+          }
+        })
+      })
+      .then(response => {
+        expect(response.statusCode).to.be.equals(200)
+        verifyIntentMock.done()
+        topicContentMock.done()
+      })
+      .catch(done)
+  })
 })