Added object storage support

.eslintrc.js

@@ -145,6 +145,7 @@ module.exports = {
             }
         ],
         'mocha/no-skipped-tests': ERROR,
-        'mocha/no-exclusive-tests': ERROR
+        'mocha/no-exclusive-tests': ERROR,
+        'new-parens': ERROR
     }
 };

lib/amqp.js

@@ -1,6 +1,7 @@
 const log = require('./logging.js');
 const amqplib = require('amqplib');
 const encryptor = require('./encryptor.js');
+const ObjectStorage = require('./objectStorage.js');
 const co = require('co');
 const _ = require('lodash');
 const eventToPromise = require('event-to-promise');
@@ -11,6 +12,7 @@ const HEADER_ERROR_RESPONSE = 'x-eio-error-response';
 class Amqp {
     constructor(settings) {
         this.settings = settings;
+        this.objectStorage = new ObjectStorage(settings);
     }
 
     connect(uri) {
@@ -53,13 +55,38 @@ class Amqp {
         return Promise.resolve();
     }
 
+    async decryptMessage(message) {
+        const objectId = message.properties.headers.objectId;
+        if (!objectId) {
+            return encryptor.decryptMessageContent(message.content);
+        }
+
+        return await this.objectStorage.getObject(objectId);
+    }
+
+    async encryptMessage(data, properties) {
+        if (!this.settings.OBJECT_STORAGE_ENABLED) {
+            return encryptor.encryptMessageContent(data);
+        }
+
+        let message = '';
+        try {
+            properties.headers.objectId = await this.objectStorage.addObject(data);
+        } catch (e) {
+            log.error('Failed to add message to object storage: %s', e);
+            message = encryptor.encryptMessageContent(data);
+        }
+
+        return message;
+    }
+
     listenQueue(queueName, callback) {
         //eslint-disable-next-line consistent-this
         const self = this;
 
         this.subscribeChannel.prefetch(this.settings.RABBITMQ_PREFETCH_SAILOR);
 
-        return this.subscribeChannel.consume(queueName, function decryptMessage(message) {
+        return this.subscribeChannel.consume(queueName, async function onMessage(message) {
             log.trace('Message received: %j', message);
 
             if (message === null) {
@@ -69,10 +96,10 @@ class Amqp {
 
             let decryptedContent;
             try {
-                decryptedContent = encryptor.decryptMessageContent(message.content, message.properties.headers);
+                decryptedContent = await self.decryptMessage(message);
             } catch (err) {
                 console.error(
-                    'Error occurred while parsing message #%j payload (%s)',
+                    'Error occurred while getting message #%j payload (%s)',
                     message.fields.deliveryTag,
                     err.message
                 );
@@ -149,8 +176,7 @@ class Amqp {
     async prepareMessageAndSendToExchange(data, properties, routingKey, throttle) {
         const settings = this.settings;
         data.headers = filterMessageHeaders(data.headers);
-        const encryptedData = encryptor.encryptMessageContent(data);
-
+        const encryptedData = await this.encryptMessage(data, properties);
         return this.sendToExchange(settings.PUBLISH_MESSAGES_TO, routingKey, encryptedData, properties, throttle);
     }
 

lib/cipher.js

@@ -1,6 +1,7 @@
 var _ = require('lodash');
 var crypto = require('crypto');
 var debug = require('debug')('sailor:cipher');
+var PassThrough = require('stream').PassThrough;
 
 var ALGORYTHM = 'aes-256-cbc';
 var PASSWORD = process.env.ELASTICIO_MESSAGE_CRYPTO_PASSWORD;
@@ -9,6 +10,8 @@ var VECTOR = process.env.ELASTICIO_MESSAGE_CRYPTO_IV;
 exports.id = 1;
 exports.encrypt = encryptIV;
 exports.decrypt = decryptIV;
+exports.decryptStream = decryptStreamIV;
+exports.encryptStream = encryptStreamIV;
 
 function encryptIV(rawData) {
     debug('About to encrypt:', rawData);
@@ -30,11 +33,9 @@ function encryptIV(rawData) {
     return cipher.update(rawData, 'utf-8', 'base64') + cipher.final('base64');
 }
 
-function decryptIV(encData, options) {
+function decryptIV(encData) {
     debug('About to decrypt:', encData);
 
-    options = options || {};
-
     if (!_.isString(encData)) {
         throw new Error('RabbitMQ message cipher.decryptIV() accepts only string as parameter.');
     }
@@ -54,3 +55,33 @@ function decryptIV(encData, options) {
 
     return result;
 }
+
+function encryptStreamIV() {
+    debug('Creating encryption stream');
+
+    if (!PASSWORD) {
+        return new PassThrough();
+    }
+
+    if (!VECTOR) {
+        throw new Error('process.env.ELASTICIO_MESSAGE_CRYPTO_IV is not set');
+    }
+
+    var encodeKey = crypto.createHash('sha256').update(PASSWORD, 'utf-8').digest();
+    return crypto.createCipheriv(ALGORYTHM, encodeKey, VECTOR);
+}
+
+function decryptStreamIV() {
+    debug('Creating decryption stream');
+
+    if (!PASSWORD) {
+        return new PassThrough();
+    }
+
+    if (!VECTOR) {
+        throw new Error('process.env.ELASTICIO_MESSAGE_CRYPTO_IV is not set');
+    }
+
+    var decodeKey = crypto.createHash('sha256').update(PASSWORD, 'utf-8').digest();
+    return crypto.createDecipheriv(ALGORYTHM, decodeKey, VECTOR);
+}

lib/encryptor.js

@@ -1,20 +1,41 @@
 var cipher = require('./cipher.js');
+var Readable = require('stream').Readable;
+var zlib = require('zlib');
+var getStream = require('get-stream');
 
 exports.encryptMessageContent = encryptMessageContent;
 exports.decryptMessageContent = decryptMessageContent;
+exports.encryptMessageContentStream = encryptMessageContentStream;
+exports.decryptMessageContentStream = decryptMessageContentStream;
 
 function encryptMessageContent(messagePayload) {
     return cipher.encrypt(JSON.stringify(messagePayload));
 }
 
-function decryptMessageContent(messagePayload, messageHeaders) {
+function decryptMessageContent(messagePayload) {
     if (!messagePayload || messagePayload.toString().length === 0) {
         return null;
     }
     try {
-        return JSON.parse(cipher.decrypt(messagePayload.toString(), messageHeaders));
+        return JSON.parse(cipher.decrypt(messagePayload.toString()));
     } catch (err) {
         console.error(err.stack);
         throw Error('Failed to decrypt message: ' + err.message);
     }
 }
+
+function encryptMessageContentStream(data) {
+    const dataStream = new Readable();
+    dataStream.push(JSON.stringify(data));
+    dataStream.push(null);
+    return dataStream
+        .pipe(zlib.createGzip())
+        .pipe(cipher.encryptStream());
+}
+
+async function decryptMessageContentStream(stream) {
+    const s = stream.pipe(cipher.decryptStream()).pipe(zlib.createGunzip());
+    const content = await getStream(s);
+
+    return JSON.parse(content);
+}

lib/objectStorage.js

@@ -0,0 +1,81 @@
+const log = require('./logging.js');
+const encryptor = require('./encryptor.js');
+const uuid = require('uuid');
+const axios = require('axios');
+const http = require('http');
+const https = require('https');
+
+class ObjectStorage {
+    constructor(settings) {
+        this.api = axios.create({
+            baseURL: `${settings.OBJECT_STORAGE_URI}/`,
+            httpAgent: new http.Agent({ keepAlive: true }),
+            httpsAgent: new https.Agent({ keepAlive: true }),
+            headers: { Authorization: `Bearer ${settings.OBJECT_STORAGE_TOKEN}` },
+            validateStatus: null
+        });
+    }
+
+    async requestRetry({ maxAttempts, delay, request, onResponse }) {
+        let attempts = 0;
+        let res;
+        let err;
+        while (attempts < maxAttempts) {
+            err = null;
+            res = null;
+            attempts++;
+            try {
+                res = await request();
+            } catch (e) {
+                err = e;
+            }
+            if (onResponse && onResponse(err, res)) {
+                continue;
+            }
+            if (err || res.status >= 400) {
+                log.warn('Error during object get: %s', err ? err : `${res.status} (${res.statusText})`);
+                await new Promise((resolve) => setTimeout(resolve, delay));
+                continue;
+            }
+            break;
+        }
+        if (err || res.status >= 400) {
+            throw err || new Error(`HTTP error during object get: ${res.status} (${res.statusText})`);
+        }
+        return res;
+    }
+
+    async getObject(objectId) {
+        const res = await this.requestRetry({
+            maxAttempts: 3,
+            delay: 100,
+            request: () => this.api.get(`/objects/${objectId}`, { responseType: 'stream' })
+        });
+
+        return await encryptor.decryptMessageContentStream(res.data);
+    }
+
+    async addObject(data) {
+        let objectId = uuid.v4();
+        await this.requestRetry({
+            maxAttempts: 3,
+            delay: 100,
+            request: () => this.api.put(
+                `/objects/${objectId}`,
+                encryptor.encryptMessageContentStream(data),
+                { headers: { 'content-type': 'application/octet-stream' } }
+            ),
+            onResponse: (err, res) => {
+                if (!err && res.status === 409) {
+                    log.warn('Generated already existing UUID');
+                    objectId = uuid.v4();
+                    return true;
+                }
+            }
+        });
+
+        return objectId;
+    }
+}
+
+module.exports = ObjectStorage;

lib/settings.js

@@ -48,7 +48,10 @@ function readFrom(envVars) {
         RATE_INTERVAL: 100, // 100ms
         PROCESS_AMQP_DRAIN: true,
         AMQP_PUBLISH_RETRY_DELAY: 100, // 100ms
-        AMQP_PUBLISH_RETRY_ATTEMPTS: 10
+        AMQP_PUBLISH_RETRY_ATTEMPTS: 10,
+        OBJECT_STORAGE_URI: '',
+        OBJECT_STORAGE_TOKEN: '',
+        OBJECT_STORAGE_ENABLED: false
     };
 
     _.forEach(requiredAlways, function readRequired(key) {

mocha_spec/integration_helpers.js

@@ -5,6 +5,8 @@ const amqplib = require('amqplib');
 const { EventEmitter } = require('events');
 const PREFIX = 'sailor_nodejs_integration_test';
 const nock = require('nock');
+const getStream = require('get-stream');
+const encryptor = require('../lib/encryptor.js');
 
 const env = process.env;
 
@@ -131,11 +133,12 @@ class AmqpHelper extends EventEmitter {
 
         this.publishChannel.ack(message);
 
-        const emittedMessage = JSON.parse(message.content.toString());
+        const content = message.content.toString();
+        const emittedMessage = content ? JSON.parse(content) : content;
 
         const data = {
             properties: message.properties,
-            body: emittedMessage.body,
+            body: emittedMessage ? emittedMessage.body : null,
             emittedMessage
         };
         this.dataMessages.push(data);
@@ -178,7 +181,9 @@ function prepareEnv() {
 
     env.DEBUG = 'sailor:debug';
 
-
+    env.ELASTICIO_OBJECT_STORAGE_URI = 'http://ma.es.ter';
+    env.ELASTICIO_OBJECT_STORAGE_ENABLED = '';
+    env.ELASTICIO_OBJECT_STORAGE_TOKEN = 'jwt';
 }
 
 function mockApiTaskStepResponse(response) {
@@ -197,6 +202,11 @@ function mockApiTaskStepResponse(response) {
         .reply(200, Object.assign(defaultResponse, response));
 }
 
+async function encryptForObjectStorage(input) {
+    const stream = encryptor.encryptMessageContentStream(input);
+    return await getStream.buffer(stream);
+}
+
 exports.PREFIX = PREFIX;
 
 exports.amqp = function amqp() {
@@ -205,4 +215,4 @@ exports.amqp = function amqp() {
 
 exports.prepareEnv = prepareEnv;
 exports.mockApiTaskStepResponse = mockApiTaskStepResponse;
-
+exports.encryptForObjectStorage = encryptForObjectStorage;