* Upgrade to sailor 2.6.18 (#5)

Upgrade to sailor 2.6.18

Author: denyshld

.circleci/config.yml

@@ -0,0 +1,28 @@
+version: 2
+jobs:
+  test:
+    docker:
+      - image: circleci/node:14-stretch
+    steps:
+      - checkout
+      - restore_cache:
+          key: dependency-cache-{{ checksum "package.json" }}
+      - run:
+          name: Audit Dependencies
+          command: npm audit --audit-level=high
+      - run:
+          name: Installing Dependencies
+          command: npm install
+      - save_cache:
+          key: dependency-cache-{{ checksum "package.json" }}
+          paths:
+            - node_modules
+      - run:
+          name: Running Integration Tests
+          command: npm run integration-test
+
+workflows:
+  version: 2
+  build_and_test:
+    jobs:
+      - test

.travis.yml

@@ -1,3 +1,9 @@
+## 1.3.2 (November 5, 2020)
+
+* Upgrade to sailor 2.6.18
+* Annual audit of the component code to check if it exposes a sensitive data in the logs
+* Annual npm vulnerabilities audit
+
 ## 1.3.1 (May 22, 2020)
 
 * Update sailor version to 2.6.7

CHANGELOG.md

@@ -1,6 +1,6 @@
-const co = require('co');
 const amqp = require('amqplib');
 const logger = require('@elastic.io/component-logger')();
+
 const encryptor = require('../encryptor.js');
 
 let channel;
@@ -10,18 +10,16 @@ let channel;
  *
  * @param cfg
  */
-function init(cfg) {
+async function init(cfg) {
   logger.info('Starting initialization');
   const { amqpURI } = cfg;
   const amqpExchange = cfg.topic;
-  return co(function* gen() {
-    logger.debug('Connecting to amqpURI=%s', amqpURI);
-    const conn = yield amqp.connect(amqpURI);
-    logger.debug('Creating a confirm channel');
-    channel = yield conn.createConfirmChannel();
-    logger.debug('Asserting topic exchange exchange=%s', amqpExchange);
-    yield channel.assertExchange(amqpExchange, 'topic');
-  });
+  logger.debug('Connecting to amqp...');
+  const conn = await amqp.connect(amqpURI);
+  logger.debug('Creating a confirm channel');
+  channel = await conn.createConfirmChannel();
+  logger.debug('Asserting topic exchange exchange...');
+  await channel.assertExchange(amqpExchange, 'topic');
 }
 
 /**
@@ -30,24 +28,23 @@ function init(cfg) {
  * @param msg incoming message object that contains ``body`` with payload
  * @param cfg configuration that is account information and configuration field values
  */
-function processAction(msg, cfg) {
+async function processAction(msg, cfg) {
   const self = this;
   const amqpExchange = cfg.topic;
-  return co(function* sendMessage() {
-    self.logger.info('Publishing message id=%s', msg.id);
-    const encryptedData = encryptor.encryptMessageContent(self, {
-      body: msg.body.payload || msg.body,
-      attachments: msg.attachments,
-    });
-    channel.publish(amqpExchange, msg.body.routingKey || '', encryptedData, {
-      contentType: 'application/octet-stream',
-      messageId: msg.id,
-    });
-    self.logger.info('Message published id=%s', msg.id);
-    yield channel.waitForConfirms();
-    self.logger.info('Message publishing confirmed id=%s', msg.id);
-    return msg;
+
+  self.logger.info('Publishing message...');
+  const encryptedData = encryptor.encryptMessageContent(self, {
+    body: msg.body.payload || msg.body,
+    attachments: msg.attachments,
+  });
+  channel.publish(amqpExchange, msg.body.routingKey || '', encryptedData, {
+    contentType: 'application/octet-stream',
+    messageId: msg.id,
   });
+  self.logger.info('Message published');
+  await channel.waitForConfirms();
+  self.logger.info('Message publishing confirmed');
+  return msg;
 }
 
 module.exports.process = processAction;

lib/actions/publish.js

@@ -1,5 +1,4 @@
 /* eslint-disable new-cap */
-
 const _ = require('lodash');
 const crypto = require('crypto');
 
@@ -8,7 +7,7 @@ const PASSWORD = process.env.ELASTICIO_MESSAGE_CRYPTO_PASSWORD;
 const VECTOR = process.env.ELASTICIO_MESSAGE_CRYPTO_IV;
 
 function encryptIV(self, rawData) {
-  self.logger.debug('About to encrypt:', rawData);
+  self.logger.debug('About to encrypt raw data');
 
   if (!_.isString(rawData)) {
     throw new Error('RabbitMQ message cipher.encryptIV() accepts only string as parameter.');
@@ -29,7 +28,7 @@ function encryptIV(self, rawData) {
 }
 
 function decryptIV(self, encData) {
-  self.logger.debug('About to decrypt:', encData);
+  self.logger.debug('About to decrypt encrypted data');
 
   if (!PASSWORD) {
     self.logger.info('Decryption will be skipped as ELASTICIO_MESSAGE_CRYPTO_PASSWORD env is empty');

lib/cipher.js

@@ -11,7 +11,7 @@ function decryptMessageContent(self, messagePayload) {
   try {
     return JSON.parse(cipher.decrypt(self, messagePayload));
   } catch (err) {
-    self.logger.error(err.stack);
+    self.logger.error('Error occurred while message decrypt!');
     throw Error(`Failed to decrypt message: ${err.message}`);
   }
 }

lib/encryptor.js

@@ -1,7 +1,7 @@
 const { messages } = require('elasticio-node');
-const co = require('co');
 const amqp = require('amqplib');
 const logger = require('@elastic.io/component-logger')();
+
 const encryptor = require('../encryptor.js');
 
 let channel;
@@ -13,35 +13,34 @@ let listening;
  *
  * @param cfg
  */
-function init(cfg) {
-  logger.info('Starting initialization, queueName=%s', queueName);
+async function init(cfg) {
+  logger.info('Starting initialization...');
   const { amqpURI } = cfg;
   const amqpExchange = cfg.topic;
   const keys = (cfg.bindingKeys || '#').split(',').map((s) => s.trim());
-  return co(function* initialize() {
-    logger.debug('Connecting to amqpURI=%s', amqpURI);
-    const conn = yield amqp.connect(amqpURI);
-
-    logger.debug('Creating a receiver channel');
-    channel = yield conn.createChannel();
+  logger.debug('Connecting to amqpURI...');
+  const conn = await amqp.connect(amqpURI);
 
-    logger.debug('Asserting topic exchange exchange=%s', amqpExchange);
-    yield channel.assertExchange(amqpExchange, 'topic');
+  logger.debug('Creating a receiver channel');
+  channel = await conn.createChannel();
 
-    logger.debug('Asserting queue');
-    yield channel.assertQueue(queueName, {
-      exclusive: false,
-      durable: false,
-      autoDelete: true,
-    });
+  logger.debug('Asserting topic exchange exchange...');
+  await channel.assertExchange(amqpExchange, 'topic');
 
-    // eslint-disable-next-line no-restricted-syntax
-    for (const key of keys) {
-      logger.debug(`Binding queue to exchange queue=${queueName} exchange=${amqpExchange} bindingKey=${key}`);
-      yield channel.bindQueue(queueName, amqpExchange, key);
-    }
-    logger.info('Initialization completed');
+  logger.debug('Asserting queue');
+  await channel.assertQueue(queueName, {
+    exclusive: false,
+    durable: false,
+    autoDelete: true,
   });
+
+  // eslint-disable-next-line no-restricted-syntax
+  for (const key of keys) {
+    logger.debug('Binding queue to exchange...');
+    // eslint-disable-next-line no-await-in-loop
+    await channel.bindQueue(queueName, amqpExchange, key);
+  }
+  logger.info('Initialization completed');
 }
 
 /**
@@ -50,8 +49,8 @@ function init(cfg) {
  * @param msg incoming message object that contains ``body`` with payload
  * @param cfg configuration that is account information and configuration field values
  */
-// eslint-disable-next-line no-unused-vars
-function processAction(msg, cfg) {
+// eslint-disable-next-line no-unused-vars,consistent-return
+async function processAction(msg, cfg) {
   const self = this;
   self.logger.info('Trigger started');
   if (listening) {
@@ -60,23 +59,21 @@ function processAction(msg, cfg) {
   }
   // eslint-disable-next-line no-shadow
   const consumer = (msg) => {
-    self.logger.debug('Have got message fields=%j properties=%j', msg.fields, msg.properties);
+    self.logger.debug('New message got');
     const decrypted = encryptor.decryptMessageContent(self, msg.content);
-    self.logger.debug('Decrypted message=%j', decrypted);
+    self.logger.debug('Message decrypted');
     const newMsg = messages.newMessageWithBody(decrypted.body || {});
     newMsg.id = msg.properties.messageId;
     newMsg.attachments = decrypted.attachments || {};
     self.emit('data', newMsg);
   };
-  return co(function* consume() {
-    self.logger.info('Starting consuming from %s', queueName);
-    yield channel.consume(queueName, consumer, {
-      noAck: true, // We can't really assert if message was consumed if we emit it yet
-      consumerTag: `consumer_${process.env.ELASTICIO_EXEC_ID}_${process.env.ELASTICIO_FLOW_ID}`,
-    });
-    self.logger.info('Consumption started');
-    listening = true;
+  self.logger.info('Starting consuming from %s', queueName);
+  await channel.consume(queueName, consumer, {
+    noAck: true, // We can't really assert if message was consumed if we emit it yet
+    consumerTag: `consumer_${process.env.ELASTICIO_EXEC_ID}_${process.env.ELASTICIO_FLOW_ID}`,
   });
+  self.logger.info('Consumption started');
+  listening = true;
 }
 
 module.exports.process = processAction;