Small fixes

data/config.yaml

@@ -25,12 +25,16 @@ services:
     package: lib.plugin.base.lib.Telegram
     service: TelegramFeed
     storage: pool
-    token: "358947514:"
+    token: "358947212354:"
     chats:
     - id: aiWeipeighah7vufoHa0ieToipooYe
       if:
         steps.ftp_apply_tpl: true
         data.filter: false
+    - id: ohl7AeGah5uo8cho4nae9Eemaeyae3
+      if:
+        steps.gopher_apply_tpl: true
+        data.filter: false
 
 storage:
   pool:

lib/exeq/Executor.py

@@ -22,14 +22,13 @@ class RQExecutor(Executor):
   def __run(self):
     redis_conn = Redis(host=self.lcnf.get('redis').get('host'))
     jobs = []
-    
+
     while self._running:
-      sleep(self.lcnf.get('delay', 0.2))
+      sleep(self.lcnf.get('delay', 0.5))
       try:
         for pn, pipeline in self.cnf.get("pipelines").items():
           if pn not in self.cnf.get('core').get('pipelines'):
             continue
-          source = Loader.by_id('storage', pipeline.get('source'))
           for step in pipeline['steps']:
             q = Queue(step.get('priority', 'normal'), connection=redis_conn)
             for job_id in jobs:
@@ -49,12 +48,15 @@ class RQExecutor(Executor):
             items = self._data.get(block=False, count=count, filter=filter)
             # obtain everything else from source
             if len(items) < count:
+              source = Loader.by_id('storage', pipeline.get('source'))
               new_items = source.get(block=False, count=(count - len(items)), filter=filter)
               items.extend(new_items)
               source.remove(new_items)
 
             if items:
-              self._data.update(items, {'$set': {'steps.%s' % step['task']: None}})
+              for i in items:
+                i['steps'][step['task']] = None
+              self._data.update(items)
               job = q.enqueue("lib.exeq.Task.run", step['task'], items)
               self._logger.info("%s|%s|%s|%s", job.id, step.get('priority', 'normal'), step['task'], len(items))
               jobs.append(job.id)

lib/plugin/iscan/tasks/common.py

@@ -38,7 +38,7 @@ class MasScanTask(Task):
 
     gi = GeoIP.open(cnf.get("geoip_dat", "/usr/share/GeoIP/GeoIPCity.dat"), GeoIP.GEOIP_INDEX_CACHE | GeoIP.GEOIP_CHECK_CACHE)
     ip_list = [i['data']['ip'] for i in items]
-    port_list = cnf.get("tasks").get(self._id).get("ports")
+    port_list = self.lcnf.get("ports")
     
     self._logger.debug("Starting scan, ip_list=%s, port_list=%s", ip_list, port_list)
     

lib/plugin/plugins/Gopher.py

@@ -1,25 +0,0 @@
-import netaddr
-
-from Config import cnf
-from lib.plugin.plugins import BasePlugin
-
-
-class Plugin(BasePlugin):
-  class TelegramMessage(BasePlugin.TelegramMessage):
-    def _init(self):
-      self._name = "Gopher"
-
-    def _generate(self):
-
-  class Pipeline(BasePlugin.Pipeline):
-    def _init(self):
-      self._name = "Gopher"
-
-    def run(self):
-      try:
-        self._find()
-        self._push()
-      except Exception as e:
-        self._logger.debug("Error occured: %s (%s)", e, self._host['ip'])
-      else:
-        self._logger.info("Succeeded for %s" % self._host['ip'])