Added debug environment to config.json and environment switch to

spamscope_topology cli.

Fixed dependencies.

Splitted requirements in options and not.

Removed sleep in files_mails to avoid timeout.

.gitignore

@@ -11,4 +11,4 @@ _resources
 build/
 dist/
 logs
-virtualenvs
+venv/

conf/spamscope.example.yml

@@ -6,9 +6,6 @@ files-mails:
     # Reload new mails after reload.mails analyzed
     reload.mails: 1000
 
-    # Waiting new mails, sleep seconds
-    waiting.sleep: 1
-
     # Post processing
     post_processing:
 

config.json

@@ -13,12 +13,29 @@
             "log": {
                 "path": "/var/log/spamscope",
                 "max_bytes": 5000000,
-                "backup_count": 10,
+                "backup_count": 3,
                 "level": "info"
             },
             "use_virtualenv": false,
             "use_ssh_for_nimbus": false,
             "virtualenv_root": ""
+        },
+        "debug": {
+            "user": "fedelemantuano",
+            "nimbus": "localhost",
+            "workers": [
+                "localhost"
+            ],
+            "log": {
+                "path": "/tmp/logs/",
+                "max_bytes": 5000000,
+                "backup_count": 1,
+                "level": "debug"
+            },
+            "use_virtualenv": true,
+            "install_virtualenv": false,
+            "use_ssh_for_nimbus": false,
+            "virtualenv_root": "venv/"
         }
     }
 }

project.clj

@@ -3,8 +3,8 @@
   :target-path "_build"
   :min-lein-version "2.0.0"
   :jvm-opts ["-client"]
-  :dependencies  [[org.apache.storm/storm-core "1.1.0"]
-                  [org.apache.storm/flux-core "1.1.0"]]
+  :dependencies  [[org.apache.storm/storm-core "1.1.1"]
+                  [org.apache.storm/flux-core "1.1.1"]]
   :jar-exclusions     [#"log4j\.properties" #"org\.apache\.storm\.(?!flux)" #"trident" #"META-INF" #"meta-inf" #"\.yaml"]
   :uberjar-exclusions [#"log4j\.properties" #"org\.apache\.storm\.(?!flux)" #"trident" #"META-INF" #"meta-inf" #"\.yaml"]
   )

requirements.txt

@@ -2,17 +2,12 @@ PyYAML
 astropy>=1.3.3
 backports.functools-lru-cache>=1.3
 chainmap
-elasticsearch>=6.0.0, <7
 lxml
 mail-parser>=3
 patool
 pyparsing
 python-magic
-redis>=2.10.5, <3
-shodan
 simplejson
 six
 ssdeep
 streamparse==3.12.0
-tika-app
-virustotal-api

requirements_optional.txt

@@ -0,0 +1,5 @@
+elasticsearch>=6.0.0, <7
+redis>=2.10.5, <3
+shodan
+tika-app
+virustotal-api

src/bolts/__init__.py

@@ -18,8 +18,6 @@
 from .json_maker import JsonMaker
 from .network import Network
 from .output_debug import OutputDebug
-from .output_elasticsearch import OutputElasticsearch
-from .output_redis import OutputRedis
 from .phishing import Phishing
 from .raw_mail import RawMail
 from .tokenizer import Tokenizer

src/bolts/output_redis.py

@@ -19,7 +19,8 @@
 
 from __future__ import absolute_import, print_function, unicode_literals
 from collections import Counter
-from modules import AbstractBolt, reformat_output, Redis
+from modules import AbstractBolt, reformat_output
+from modules.redis_client import Redis
 
 try:
     import simplejson as json

src/cli/spamscope_topology.py

@@ -63,6 +63,13 @@ def get_args():
         help="SpamScope topology.",
         dest="topology")
 
+    submit.add_argument(
+        "-e",
+        "--environment",
+        default="prod",
+        help="The environment to use for the command.",
+        dest="environment")
+
     submit.add_argument(
         "-w",
         "--workers",
@@ -98,7 +105,7 @@ def get_args():
     submit.add_argument(
         "-t",
         "--timeout",
-        default=720,
+        default=120,
         type=int,
         help=("How long (in s) between heartbeats until supervisor considers "
               "that worker dead."),
@@ -111,17 +118,18 @@ def create_jar():
     pass
 
 
-def submit_topology(topology, nr_worker, tick, max_pending,
-                    spout_sleep, timeout):
+def submit_topology(topology, environment, nr_worker, tick,
+                    max_pending, spout_sleep, timeout):
     command_line = (
-        "sparse submit -f -n {topology} -w {nr_worker} "
+        "sparse submit -f -n {topology} -w {nr_worker} -e {environment} "
         "-o topology.tick.tuple.freq.secs={tick} "
         "-o topology.max.spout.pending={max_pending} "
         "-o topology.sleep.spout.wait.strategy.time.ms={spout_sleep} "
         "-o supervisor.worker.timeout.secs={timeout} "
         "-o topology.message.timeout.secs={timeout}".format(
-            topology=topology, nr_worker=nr_worker, tick=tick,
-            max_pending=max_pending, spout_sleep=spout_sleep, timeout=timeout))
+            topology=topology, environment=environment, nr_worker=nr_worker,
+            tick=tick, max_pending=max_pending, spout_sleep=spout_sleep,
+            timeout=timeout))
 
     args = shlex.split(command_line)
     proc = Popen(args, stderr=STDOUT)
@@ -138,6 +146,7 @@ def main():
     if args.subparser == "submit":
         submit_topology(
             topology=args.topology,
+            environment=args.environment,
             nr_worker=args.workers,
             tick=args.tick,
             max_pending=args.max_pending,

src/modules/__init__.py

@@ -16,4 +16,3 @@
 
 from .abstracts import AbstractBolt, AbstractSpout, MAIL_PATH, MAIL_STRING, MAIL_PATH_OUTLOOK
 from .utils import *
-from .redis_client import Redis, RedisConnectionFailed

src/spouts/files_mails.py

@@ -23,7 +23,6 @@
 import glob
 import os
 import shutil
-import time
 
 from modules import AbstractSpout, MailItem, MAIL_PATH, MAIL_PATH_OUTLOOK
 
@@ -40,7 +39,6 @@ def initialize(self, stormconf, context):
         self._queue_tail = set()
         self._count = 1
         self._what = self.conf["post_processing"]["what"].lower()
-        self._waiting_sleep = float(self.conf["waiting.sleep"])
         self._load_mails()
 
     def _check_conf(self):
@@ -95,7 +93,6 @@ def _load_mails(self):
     def next_tuple(self):
 
         # After reload.mails next_tuple reload spout config
-        # About reload.mails * waiting_sleep seconds
         if (self._count % self.conf["reload.mails"]):
             self._count += 1
 
@@ -122,13 +119,6 @@ def next_tuple(self):
                 mail.headers],  # 6
                 tup_id=mail.filename)
 
-        # If queue is empty
-        else:
-            self.log("Queue mails for {!r} is empty".format(
-                self.component_name), "debug")
-            time.sleep(self._waiting_sleep)
-            self._load_mails()
-
     def ack(self, tup_id):
         """Acknowledge tup_id, that is the path_mail. """
 

tests/test_redis.py

@@ -25,7 +25,7 @@
 root = os.path.join(base_path, '..')
 sys.path.append(root)
 
-from src.modules import Redis, RedisConnectionFailed
+from src.modules.redis_client import Redis, RedisConnectionFailed
 
 
 class TestRedis(unittest.TestCase):

topologies/spamscope_debug.py

@@ -37,7 +37,7 @@ class OutputDebugTopology(Topology):
     attachments = Attachments.spec(
         name="attachments",
         inputs={tokenizer['attachments']: Grouping.fields('sha256_random')},
-        par=2)
+        par=1)
 
     urls = Urls.spec(
         name="urls",
@@ -55,12 +55,12 @@ class OutputDebugTopology(Topology):
     network = Network.spec(
         name="network",
         inputs={tokenizer['network']: Grouping.fields('sha256_random')},
-        par=2)
+        par=1)
 
     raw_mail = RawMail.spec(
         name="raw_mail",
         inputs={tokenizer['raw_mail']: Grouping.fields('sha256_random')},
-        par=2)
+        par=1)
 
     json_maker = JsonMaker.spec(
         name="json_maker",

topologies/spamscope_elasticsearch.py

@@ -20,7 +20,8 @@
 
 from spouts import FilesMailSpout
 from bolts import (Attachments, JsonMaker, Phishing, Tokenizer,
-                   Urls, Network, RawMail, OutputElasticsearch)
+                   Urls, Network, RawMail)
+from bolts.output_elasticsearch import OutputElasticsearch
 from streamparse import Grouping, Topology
 
 

topologies/spamscope_redis.py

@@ -20,7 +20,8 @@
 
 from spouts import FilesMailSpout
 from bolts import (Attachments, JsonMaker, Phishing, Tokenizer,
-                   Urls, Network, RawMail, OutputRedis)
+                   Urls, Network, RawMail)
+from bolts.output_redis import OutputRedis
 from streamparse import Grouping, Topology