Merge pull request #52 from project-sunbird/cts-vul-fixes

Vulnerability & Code Quality Fixes

Author: pallakartheekreddy

.gitignore

@@ -2,6 +2,7 @@ node_modules
 .nyc_output
 coverage
 .audit.json
+*-audit.json
 telemetry-*.log
 mochawesome-report
 *.log

src/app.js

@@ -1,6 +1,7 @@
 const express = require('express'),
   cluster = require('express-cluster'),
   cookieParser = require('cookie-parser'),
+  helmet = require('helmet'),
   logger = require('morgan'),
   bodyParser = require('body-parser'),
   envVariables = require('./envVariables'),
@@ -9,12 +10,15 @@ const express = require('express'),
 
 const createAppServer = () => {
   const app = express();
+  app.use(helmet());
   app.use((req, res, next) => {
-    res.header('Access-Control-Allow-Origin', '*');
+    res.header('Access-Control-Allow-Origin', envVariables.allowedOrigins);
     res.header('Access-Control-Allow-Methods', 'GET,PUT,POST,PATCH,DELETE,OPTIONS');
-    res.header('Access-Control-Allow-Headers', 'Content-Type, Authorization,' + 'cid, user-id, x-auth, Cache-Control, X-Requested-With, datatype, *');
-    if (req.method === 'OPTIONS') res.sendStatus(200);
-    else next();
+    res.header('Access-Control-Allow-Headers', 'Content-Type, Authorization, cid, user-id, x-auth, Cache-Control, X-Requested-With, datatype, *');
+    if (req.method === 'OPTIONS') {
+      return res.status(200).end();
+    }
+    next();
   });
   app.use(bodyParser.json({ limit: '5mb' }));
   app.use(logger('dev'));

src/dispatcher/dispatcher.js

@@ -1,7 +1,4 @@
 const winston = require('winston');
-require('winston-daily-rotate-file');
-require('./kafka-dispatcher');
-require('./cassandra-dispatcher');
 
 const defaultFileOptions = {
   filename: 'dispatcher-%DATE%.log',
@@ -15,33 +12,55 @@ const defaultFileOptions = {
 class Dispatcher {
   constructor(options) {
     if (!options) throw new Error('Dispatcher options are required');
-    this.logger = new(winston.Logger)({level: 'info'});
     this.options = options;
-    if (this.options.dispatcher == 'kafka') {
-      this.logger.add(winston.transports.Kafka, this.options);
+    this.transport = null;
+    
+    if (this.options.dispatcher === 'kafka') {
+      const { KafkaDispatcher } = require('./kafka-dispatcher');
+      this.transport = new KafkaDispatcher(this.options);
       console.log('Kafka transport enabled !!!');
-    } else if (this.options.dispatcher == 'file') {
+    } else if (this.options.dispatcher === 'file') {
+      require('winston-daily-rotate-file');
       const config = Object.assign(defaultFileOptions, this.options);
-      this.logger.add(winston.transports.DailyRotateFile, config);
+      this.transport = new winston.transports.DailyRotateFile(config);
       console.log('File transport enabled !!!');
     } else if (this.options.dispatcher === 'cassandra') {
-      this.logger.add(winston.transports.Cassandra, this.options);
+      require('./cassandra-dispatcher');
+      this.transport = new winston.transports.Cassandra(this.options);
       console.log('Cassandra transport enabled !!!');
     } else { // Log to console
       this.options.dispatcher = 'console';
       const config = Object.assign({json: true,stringify: (obj) => JSON.stringify(obj)}, this.options);
-      this.logger.add(winston.transports.Console, config);
+      this.transport = new winston.transports.Console(config);
       console.log('Console transport enabled !!!');
     }
+    
+    this.logger = winston.createLogger({
+      level: 'info',
+      transports: [this.transport]
+    });
   }
 
   dispatch(mid, message, callback) {
-    this.logger.log('info', message, {mid: mid}, callback);
+    // Winston 3.x logger.log doesn't support callbacks, but individual transports do
+    // We call the transport's log method directly to get proper callback support
+    const info = {
+      level: 'info',
+      message: message,
+      mid: mid
+    };
+    
+    // Call the transport's log method directly with callback
+    this.transport.log(info, (err) => {
+      if (callback) {
+        callback(err);
+      }
+    });
   }
 
   health(callback) {
     if (this.options.dispatcher === 'kafka') {
-      this.logger.transports['kafka'].health(callback);
+      this.transport.health(callback);
     } else if (this.options.dispatcher === 'console') {
       callback(true);
     } else { // need to add health method for file/cassandra

src/dispatcher/kafka-dispatcher.js

@@ -1,42 +1,125 @@
-const winston = require('winston'),
-  kafka = require('kafka-node'),
+const Transport = require('winston-transport'),
   _ = require('lodash'),
-  HighLevelProducer = kafka.HighLevelProducer,
+  { Kafka, CompressionTypes } = require('kafkajs'),
+  config = require('../envVariables'),
   defaultOptions = {
     kafkaHost: 'localhost:9092',
     maxAsyncRequests: 100,
     topic: 'local.ingestion',
     compression_type: 'none'
   };
 
-class KafkaDispatcher extends winston.Transport {
+function mapCompressionAttr(attr) {
+  // kafka-node used numeric attributes: 0 = none, 1 = gzip, 2 = snappy
+  if (attr === 2) return CompressionTypes.Snappy;
+  if (attr === 1) return CompressionTypes.GZIP;
+  return CompressionTypes.None;
+}
+
+class KafkaDispatcher extends Transport {
   constructor(options) {
-    super();
+    super(options);
     this.name = 'kafka';
     this.options = _.assignInWith(defaultOptions, options, (objValue, srcValue) => srcValue ? srcValue : objValue);
-    if (this.options.compression_type == 'snappy') {
+    if (this.options.compression_type === 'snappy') {
       this.compression_attribute = 2;
-    } else if(this.options.compression_type == 'gzip') {
+    } else if(this.options.compression_type === 'gzip') {
       this.compression_attribute = 1;
     } else {
       this.compression_attribute = 0;
     }
-    this.client = new kafka.KafkaClient({
-      kafkaHost: this.options.kafkaHost,
-      maxAsyncRequests: this.options.maxAsyncRequests
-    });
-    this.producer = new HighLevelProducer(this.client);
-    this.producer.on('ready', () => console.log('kafka dispatcher is ready'));
-    this.producer.on('error', (err) => console.error('Unable to connect to kafka', err));
+
+    // kafkajs expects an array of broker strings
+    const brokers = (typeof this.options.kafkaHost === 'string') ? [this.options.kafkaHost] : this.options.kafkaHost;
+    this._kafka = new Kafka({ brokers });
+    this._producer = this._kafka.producer();
+    this._admin = this._kafka.admin();
+    this._producerConnected = false;
+
+    // Backwards-compatible lightweight wrappers so existing code/tests that
+    // expect producer.send(payloads, cb) and client.topicExists(topic, cb)
+    // continue to work.
+    this.producer = {
+      send: (payloads, cb) => {
+        // payloads is an array like [{ topic, key, messages, attributes, partition }]
+        const topicMessages = payloads.map(p => {
+          const msg = { key: p.key, value: p.messages };
+          if (p.hasOwnProperty('partition')) msg.partition = p.partition;
+          return {
+            topic: p.topic,
+            messages: [msg],
+            compression: mapCompressionAttr(p.attributes)
+          };
+        });
+
+        // ensure producer is connected, then send batch
+        const sendPromise = this._producerConnected 
+          ? Promise.resolve()
+          : this._producer.connect().then(() => { this._producerConnected = true; });
+
+        sendPromise
+          .then(() => this._producer.sendBatch({ topicMessages }))
+          .then(() => { if (cb) cb(); })
+          .catch(err => { if (cb) cb(err); });
+      }
+    };
+
+    this.client = {
+      topicExists: (topic, cb) => {
+        // kafkajs admin.fetchTopicMetadata throws if topic doesn't exist
+        this._admin.connect()
+          .then(() => this._admin.fetchTopicMetadata({ topics: [topic] }))
+          .then(() => this._admin.disconnect())
+          .then(() => cb && cb(null))
+          .catch(err => {
+            // ensure disconnect
+            this._admin.disconnect().catch(() => {});
+            cb && cb(err);
+          });
+      }
+    };
+
+    // log basic connection info asynchronously
+    this._producer.connect()
+      .then(() => {
+        this._producerConnected = true;
+        console.log('kafka dispatcher producer connected');
+      })
+      .catch(err => console.error('Unable to connect kafka producer', err));
+    this._admin.connect()
+      .then(() => this._admin.disconnect())
+      .catch(() => {});
   }
-  log(level, msg, meta, callback) {
+
+  log(info, callback) {
+    // Modern winston 3.x transport signature: log(info, callback)
+    // info contains: level, message, and other metadata
+    // msg/message is expected to be a JSON string. Inject a top-level dataset key
+    // from configuration if provided and not already present.
+    const msg = info.message;
+    let outgoing = msg;
+    try {
+      if (typeof msg === 'string') {
+        const parsed = JSON.parse(msg);
+        if (parsed && typeof parsed === 'object') {
+          if (config.dataset && !parsed.hasOwnProperty('dataset')) {
+            parsed.dataset = config.dataset;
+          }
+          outgoing = JSON.stringify(parsed);
+        }
+      }
+    } catch (e) {
+      // if parsing fails, leave the message as-is
+    }
+
     this.producer.send([{
       topic: this.options.topic,
-      key: meta.mid,
-      messages: msg,
+      key: info.mid,
+      messages: outgoing,
       attributes: this.compression_attribute
     }], callback);
   }
+
   health(callback) {
     this.client.topicExists(this.options.topic, (err) => {
       if (err) callback(false);
@@ -45,6 +128,4 @@ class KafkaDispatcher extends winston.Transport {
   }
 }
 
-winston.transports.Kafka = KafkaDispatcher;
-
 module.exports = { KafkaDispatcher };

src/envVariables.js

@@ -19,6 +19,10 @@ const envVariables = {
   contactPoints: (process.env.telemetry_cassandra_contactpoints || 'localhost').split(','),
   cassandraTtl: process.env.telemetry_cassandra_ttl,
   port: process.env.telemetry_service_port || 9001,
-  threads: process.env.telemetry_service_threads || os.cpus().length
+  threads: process.env.telemetry_service_threads || os.cpus().length,
+  // dataset to be injected into outgoing telemetry events
+  dataset: process.env.telemetry_dataset || 'sb-telemetry',
+  // CORS configuration
+  allowedOrigins: process.env.telemetry_allowed_origins || '*'
 };
 module.exports = envVariables;