Initial implementation

Author: zubairov

README.md

@@ -38,9 +38,10 @@ $ git push elasticio master
 This component is using the internal RabbitMQ instance that is available as part of elastic.io
 infrastructure, it expects following environment variables to be present when started:
 * ``ELASTICIO_AMQP_URI`` something like ``amqp://foo:bar@server``
-* ``ELASTICIO_MESSAGE_CRYPTO_IV`` key for symmetric encryption
+* ``ELASTICIO_MESSAGE_CRYPTO_IV`` vector for symmetric encryption
 * ``ELASTICIO_MESSAGE_CRYPTO_PASSWORD`` password for symmetric encryption
-* ``ELASTICIO_PUBLISH_MESSAGES_TO`` AMQP exchange name, like ``56a63248661ec20500000014_user``
+* ``ELASTICIO_TASK_ID`` ID of the currently running task, used to construct name for exchange
+* ``ELASTICIO_USER_ID`` ID of the current user, used to construct name of the exchange
 
 
 ## Known issues

lib/actions/publish.js

@@ -1,7 +1,10 @@
 'use strict';
 const eioUtils = require('elasticio-node').messages;
 const co = require('co');
-const rp = require('request-promise');
+const amqp = require('amqplib');
+const encryptor = require('../encryptor.js');
+
+var connection, channel;
 
 module.exports.process = processAction;
 
@@ -13,23 +16,31 @@ module.exports.process = processAction;
  * @param snapshot - current values from the snapshot
  */
 function processAction(msg, cfg, snapshot) {
-  console.log('Action started, snapshot=%j', snapshot);
-
+  console.log('Action started');
+  const amqpURI = process.env.ELASTICIO_AMQP_URI;
+  const amqpExchange = `pubsub_${process.env.ELASTICIO_TASK_ID}_${process.env.ELASTICIO_USER_ID}`;
   co(function*() {
-    console.log('Creating new request bin');
-
-    const bin = yield rp({
-      method: 'POST',
-      uri: 'http://requestb.in/api/v1/bins',
-      json: true
+    if (!connection) {
+      console.log('Connecting to amqpURI=%s', amqpURI);
+      connection = yield amqp.connect(amqpURI);
+    }
+    if (!channel) {
+      console.log('Creating a confirm channel');
+      channel = yield conn.createConfirmChannel();
+      console.log('Asserting topic exchange exchange=%s', amqpExchange);
+      yield channel.assertExchange(amqpExchange, 'topic');
+    }
+    console.log('Publishing message id=%s', msg.id);
+    let encryptedData = encryptor.encryptMessageContent({
+      id: msg.id,
+      body: msg.body,
+      attachments: msg.attachments
     });
-
-    console.log('New request bin created bin=%j', bin);
-
-    this.emit('data', eioUtils.newMessageWithBody(bin));
-
-    console.log('Processing completed');
-
+    channel.publish(amqpExchange, 'foo', encryptedData);
+    console.log('Message published id=%s', msg.id);
+    yield channel.waitForConfirms();
+    console.log('Message publishing confirmed id=%s', msg.id);
+    this.emit('data', msg);
     this.emit('end');
   }.bind(this)).catch(err => {
     console.log('Error occurred', err.stack || err);

lib/cipher.js

@@ -0,0 +1,56 @@
+var _ = require('lodash');
+var crypto = require('crypto');
+var debug = require('debug')('sailor:cipher');
+
+var ALGORYTHM = 'aes-256-cbc';
+var PASSWORD = process.env.ELASTICIO_MESSAGE_CRYPTO_PASSWORD;
+var VECTOR = process.env.ELASTICIO_MESSAGE_CRYPTO_IV;
+
+exports.id = 1;
+exports.encrypt = encryptIV;
+exports.decrypt = decryptIV;
+
+function encryptIV(rawData) {
+    debug('About to encrypt:', rawData);
+
+    if (!_.isString(rawData)) {
+        throw new Error('RabbitMQ message cipher.encryptIV() accepts only string as parameter.');
+    }
+
+    if (!PASSWORD) {
+        return rawData;
+    }
+
+    if (!VECTOR) {
+        throw new Error('process.env.ELASTICIO_MESSAGE_CRYPTO_IV is not set');
+    }
+
+    var encodeKey = crypto.createHash('sha256').update(PASSWORD, 'utf-8').digest();
+    var cipher = crypto.createCipheriv(ALGORYTHM, encodeKey, VECTOR);
+    return cipher.update(rawData, 'utf-8', 'base64') + cipher.final('base64');
+}
+
+function decryptIV(encData, options) {
+    debug('About to decrypt:', encData);
+
+    options = options || {};
+
+    if (!_.isString(encData)) {
+        throw new Error('RabbitMQ message cipher.decryptIV() accepts only string as parameter.');
+    }
+
+    if (!PASSWORD) {
+        return encData;
+    }
+
+    if (!VECTOR) {
+        throw new Error('process.env.ELASTICIO_MESSAGE_CRYPTO_IV is not set');
+    }
+
+    var decodeKey = crypto.createHash('sha256').update(PASSWORD, 'utf-8').digest();
+    var cipher = crypto.createDecipheriv(ALGORYTHM, decodeKey, VECTOR);
+
+    var result = cipher.update(encData, 'base64', 'utf-8') + cipher.final('utf-8');
+
+    return result;
+}

lib/encryptor.js

@@ -0,0 +1,20 @@
+var cipher = require('./cipher.js');
+
+exports.encryptMessageContent = encryptMessageContent;
+exports.decryptMessageContent = decryptMessageContent;
+
+function encryptMessageContent(messagePayload) {
+    return cipher.encrypt(JSON.stringify(messagePayload));
+}
+
+function decryptMessageContent(messagePayload, messageHeaders) {
+    if (!messagePayload || messagePayload.toString().length === 0) {
+        return null;
+    }
+    try {
+        return JSON.parse(cipher.decrypt(messagePayload.toString(), messageHeaders));
+    } catch (err) {
+        console.error(err.stack);
+        throw Error('Failed to decrypt message: ' + err.message);
+    }
+}

package.json

@@ -22,12 +22,14 @@
     "elasticio-component"
   ],
   "dependencies": {
-    "co": "^4.6.0",
+    "amqplib": "^0.5.1",
     "bluebird": "^3.4.6",
-    "request": "^2.75.0",
-    "request-promise": "^4.1.1",
+    "co": "^4.6.0",
+    "debug": "^2.4.4",
+    "elasticio-node": "0.0.5",
     "elasticio-sailor-nodejs": "1.3.0",
-    "elasticio-node": "0.0.5"
+    "request": "^2.75.0",
+    "request-promise": "^4.1.1"
   },
   "devDependencies": {
     "eslint": "^2.1.0",