verify action IDs

lib/agent/actions.js

@@ -17,18 +17,18 @@ var load_action = function(type, name) {
 }
 
 // type can be 'action' or 'trigger'
-var action_running = function(type, action, name, opts, emitter) {
-  logger.info('Running: ' + name)
-  running[name] = action;
+var action_running = function(type, id, action, name, opts, emitter) {
+  logger.info(`Running: ${name} ${id}`);
+  running[id] = {name: name, action: action};
 
-  emitter.once('end', function(err, out) {
+  emitter.once('end', function(id, err, out) {
     if (err) hooks.trigger('error', err);
 
-    logger.info('Stopped: ' + name);
-    action_stopped(name);
+    logger.info(`Stopped: ${name}`);
+    action_stopped(id);
 
     setTimeout(function() {
-      hooks.trigger(type, 'stopped', name, opts, err, out);
+      hooks.trigger(type, 'stopped', id, name, opts, err, out);
     }, 1000);
 
     if (action.events)
@@ -48,25 +48,34 @@ var watch_action_events = function(events, emitter) {
   })
 }
 
-var action_stopped = function(name) {
-  delete running[name];
+var action_stopped = function(id) {
+  delete running[id];
 }
 
-var start = function(type, name, opts, cb){
-  if (running[name]) {
-    var err = new Error('Already running: ' + name);
+var start = function(type, id, name, opts, cb){
+  // Action with same id is not executed
+  if (running[id]) {
+    var err = new Error(`Already running: ${running[id].name} with id: ${id}`);
+    hooks.emit('error', err);
+
+    return cb && cb(err);
+  }
+
+  if (Object.values(running).filter(x => x.name == name).length > 0) {
+    var err = new Error(`Already running: ${name}`);
     hooks.emit('error', err);
     return cb && cb(err);
   }
 
   logger.info('Starting ' + type + ': '+ name);
+
   var action = load_action(type, name);
   if (!action) return; // load_action will emit trigger
 
   action.type    = type;
   action.options = typeof opts == 'function' ? {} : opts;
 
-  action.start(opts, function(err, emitter) {
+  action.start(id, opts, function(err, emitter) {
     cb && cb(err);
 
     if (err) {
@@ -80,11 +89,13 @@ var start = function(type, name, opts, cb){
 
     if (!opts) opts = {};
 
-    if (type == 'action')
-      hooks.trigger(type, 'started', name, opts);
-
+    if (type == 'action') {
+      setTimeout(function() {
+        hooks.trigger(type, 'started', id, name, opts);
+      }, 500);
+    }
     if (emitter) {
-      action_running(type, action, name, opts, emitter);
+      action_running(type, id, action, name, opts, emitter);
     } else {
 
       // no emitter was returned, so we have no way of knowing when
@@ -93,50 +104,42 @@ var start = function(type, name, opts, cb){
       // trigger this event after 10 seconds.
 
       setTimeout(function() {
-        hooks.trigger(type, 'stopped', name, opts);
+        hooks.trigger(type, 'stopped', id, name, opts);
       }, 10000);
 
     }
   })
 
 };
 
-actions.start = function(name, opts) {
-  start('action', name, opts);
+actions.start = function(id, name, opts, cb) {
+  start('action', id, name, opts, cb);
 }
 
-actions.start_trigger = function(name, opts, cb) {
-  start('trigger', name, opts, cb);
-}
+// TODO: agregar lo del opts para el response
+actions.stop = function(id, name, opts) {
+  if (!running[id]) {
+    hooks.trigger('action', 'stopped', id, name, opts);
+  }
 
-actions.stop = function(name, opts) {
-  logger.info('Stopping: ' + name);
-  var action = running[name];
+  var action = running[id].action;
 
   if (!action) {
     logger.warn('Action not running!');
-    hooks.trigger('action', 'stopped', name, opts);
+    hooks.trigger('action', 'stopped', id, name, opts);
   } else if (!action.stop) {
     logger.warn('Action not stoppable!');
   } else {
+    logger.info(`Stopping: ${running[id].name}, with id: ${id}`);
     action.stop();
+    action_stopped(id);
   }
 }
 
 actions.stop_all = function() {
-  for (var name in running)
-    this.stop(name);
-}
-
-actions.running = function() {
-  var list = [];
-  for (var key in running) {
-    if (running[key].type == 'action') {
-      var obj = { name: key, options: running[key].options || {} }
-      list.push(obj);
-    }
+  for (var id in running) {
+    this.stop(id);
   }
-  return list;
 }
 
 module.exports = actions;

lib/agent/actions/alarm/index.js

@@ -19,7 +19,7 @@ var child,
     emitter,
     raise_interval;
 
-exports.start = function(options, cb) {
+exports.start = function(id, options, cb) {
 
   var error,
       self     = this,
@@ -30,10 +30,10 @@ exports.start = function(options, cb) {
       file    = type + '.mp3',
       loops   = options.loops ? parseInt(options.loops) : 1;
 
-  var done = function(err) {
+  var done = function(id, err) {
     clearInterval(raise_interval);
     if (returned++) returned;
-    if (emitter) emitter.emit('end', err);
+    if (emitter) emitter.emit('end', id, err);
     emitter = null;
   }
 
@@ -45,15 +45,15 @@ exports.start = function(options, cb) {
     loops--;
 
     system.spawn_as_logged_user(commands.play, [ file ], function(err, alarm) {
-      if (err) return done(err);
+      if (err) return done(id, err);
 
       alarm.on('error', done);
 
       alarm.once('exit', function(code) {
         child = null;
 
         if (loops === 0)
-          return done();
+          return done(id);
 
         play_queue();
       })

lib/agent/actions/alert/index.js

@@ -17,7 +17,7 @@ var child,
     binary,
     user_input_str = 'User input: ';
 
-exports.start = function(opts, cb) {
+exports.start = function(id, opts, cb) {
   var opts    = opts || {},
       message = opts.message || opts.alert_message,
       title   = opts.title,
@@ -68,7 +68,7 @@ exports.start = function(opts, cb) {
     if (returned++) return;
 
     if (emitter)
-      emitter.emit('end', err, reply);
+      emitter.emit('end', id, err, reply);
 
     emitter = null;
   }

lib/agent/actions/diskencryption/index.js

@@ -54,7 +54,7 @@ var process_options = (options) => {
   return opts;
 }
 
-exports.start = function(opts, cb) {
+exports.start = function(id, opts, cb) {
   var os_name = os.platform().replace('darwin', 'mac').replace('win32', 'windows');
 
   if (os_name != 'windows')
@@ -85,7 +85,7 @@ exports.start = function(opts, cb) {
       commands.perform({command: 'get', target: 'encryption_keys'})
 
       output = {}
-      if (!out) return emitter.emit('end');
+      if (!out) return emitter.emit('end', id);
 
       out.forEach(disk => {
         var data = disk.disk.slice(0, -1);
@@ -99,7 +99,7 @@ exports.start = function(opts, cb) {
     }
 
     if (!emitter) return;
-    return emitter.emit('end', err, output);
+    return emitter.emit('end', id, err, output);
   }
 
   emitter = new Emitter;

lib/agent/actions/fileretrieval/index.js

@@ -11,7 +11,7 @@ var fs          = require('fs'),
     path        = require('path'),
     needle      = require('needle'),
     common      = require('./../../common'),
-    files       = require('./storage'),
+    storage     = require('./../../utils/storage'),
     Emitter     = require('events').EventEmitter;
 
 var system      = common.system,
@@ -26,6 +26,7 @@ var config   = common.config,
     url      = protocol + '://' + host;
 
 var UPLOAD_SERVER = url + '/upload/upload';
+const { v4: uuidv4 } = require('uuid');
 
 var em,
     cp;
@@ -36,6 +37,33 @@ var path_arg,
 // check_pending_files is used to resume any files that might been pending. It's called from
 // filesagent/providers/network.
 
+var run_stored = (host, cb) => {
+  if (host != 'solid.preyproject.com') return;
+
+  storage.do('all', { type: 'files' }, (err, files) => {
+    if (err) return cb(new Error('Error retrieving file from local database'))
+
+    var count = Object.keys(files).length;
+    if (count <= 0)
+      return;
+    logger.warn('Re-uploading ' + count + ' pending files.');
+
+    files.forEach(file => {
+      var opts = {
+        path:      file.path,
+        user:      file.user,
+        name:      file.name,
+        size:      file.size,
+        file_id:   file.id,
+        attempts:  file.attempts,
+        resumable: file.resumable
+      }
+      let id = uuidv4();
+      exports.start(id, opts, cb);
+    });
+  })
+}
+
 var retrieve_file_as_user = function(options, cb) {
   if (os_name == 'windows') {
     path_arg = path.resolve(options.path);
@@ -61,11 +89,13 @@ var retrieve_file_as_user = function(options, cb) {
     }
     logger.info("Ran as user: " + out);
     if (out.indexOf("File succesfuly uploaded") != -1) {
-      files.del(options.file_id);
+      logger.debug('Removing file_id from DB: ' + options.file_id);
+      storage.do('del', { type: 'files', id: options.file_id });
       return;
     }
     if (out.includes("EPIPE") || out.includes("EACCES")) {
-      files.update(options.file_id, options.path, options.size, options.user, options.name, options.resumable, function(err) {
+      let resumable_value = options.resumable == 1 ? 0 : 1;
+      storage.do('update', { type: 'files', id: file_id, columns: 'resumable', values: resumable_value }, (err) => { // resumable
         if (err) logger.error("Database update error");
         logger.info("Resume file option activated for ID: " + options.file_id);
       });
@@ -74,10 +104,12 @@ var retrieve_file_as_user = function(options, cb) {
 }
 
 exports.check_pending_files = function() {
-  files.run_stored(host);
+  run_stored(host, (err) => {
+    if (err) logger.error(err.message)
+  });
 }
 
-exports.start = function(options, cb) {
+exports.start = function(id, options, cb) {
 
   var url = UPLOAD_SERVER + '?uploadID=' + options.file_id;
   // Make a call to get the last byte processed by the upload server
@@ -88,25 +120,44 @@ exports.start = function(options, cb) {
       return;
     }
     if (res.statusCode == 404) {
-      files.del(options.file_id);
+      logger.debug('Removing file_id from DB: ' + options.file_id);
+      storage.do('del', { type: 'files', id: options.file_id });
       return;
     }
     var data = JSON.parse(res.body);
     var file_status = JSON.parse(res.body).Status
     options.total = data.Total;
 
     if (file_status == 0 || file_status == 4) { // File in progress(0) or Pending(4)  
-      files.exist(options.file_id, function(err, exist) {
-        if (!exist) {
-          options.resumable = false;
+      storage.do('query', {type: 'files', column: "id", data: options.file_id,}, (err, files) => {
+        if (files.length == 0) {
+          options.resumable = 0;
           options.total = 0;
-          files.store(options.file_id, options.path, options.size, options.user, options.name, options.resumable);
-          retrieve_file_as_user(options);
+
+          var data = {
+            name: options.name,
+            path: options.path,
+            size: options.size,
+            user: options.user,
+            attempts: 0,
+            resumable: options.resumable
+          }
+
+          logger.debug('Storing file_id in DB: ' + options.file_id);
+          storage.do('set', {type: 'files', id: options.file_id, data: data}, () => {
+            retrieve_file_as_user(options);
+          });
 
         } else {
           setTimeout(function() {
-            if (options.resumable) {
-              files.update(options.file_id, options.path, options.size, options.user, options.name, options.resumable, function(err) {
+            var file_attempts = files[0].attempts;
+            if (file_attempts >= 3) {
+              logger.info('File ' + options.file_id + 'has no more attempts, deleting...');
+              return;
+            }
+
+            if (options.resumable == 1) {
+              storage.do('update', { type: 'files', id: file_id, columns: ['attempts', 'resumable'], values: [file_attempts + 1, 0] }, (err) => { // resumable
                 if (err) logger.error("Database update error");
                 logger.info("Resume file option deactivated for ID: " + options.file_id);
                 retrieve_file_as_user(options);
@@ -121,15 +172,17 @@ exports.start = function(options, cb) {
         logger.debug("File already uploaded, deleting from db...");
       else 
         logger.debug("File cancelled or with an error, deleting from db...");
-      files.del(options.file_id);
+
+      logger.info('Removing file_id from DB: ' + options.file_id);
+      storage.do('del', { type: 'files', id: options.file_id });
       return;
     }
   })
 
   em = em || new Emitter();
 
   if (cb) cb(null, em);
-  em.emit('end');
+  em.emit('end', id);
 }
 
 exports.stop = function() {