Merge pull request #106 from MISP/subzero

Pulling from several 0MQ feeds + screens + diagnostic tool

config/config.cfg.default

@@ -42,10 +42,21 @@ update_filename=updates.log
 [RedisGlobal]
 host=localhost
 port=6250
-#misp_web_url = http://192.168.56.50
-misp_web_url = http://localhost
-#zmq_url=tcp://192.168.56.50:50000
-zmq_url=tcp://localhost:50000
+misp_web_url = http://0.0.0.0
+misp_instances = [{
+    "name": "misp1",
+    "url": "http://localhost",
+    "zmq": "tcp://localhost:50000"}] 
+
+#misp_instances = [{
+#    "name": "misp1",
+#    "url": "http://localhost",
+#    "zmq": "tcp://localhost:50000"},
+#    {
+#    "name": "misp2",
+#    "url": "http://10.0.2.4",
+#    "zmq": "tcp://10.0.2.4:50000"}
+#    ] 
 
 [RedisLIST]
 db=3

diagnostic.py

@@ -171,25 +171,41 @@ def check_redis(spinner):
 def check_zmq(spinner):
     timeout = 15
     context = zmq.Context()
-    socket = context.socket(zmq.SUB)
-    socket.connect(configuration_file.get('RedisGlobal', 'zmq_url'))
-    socket.setsockopt_string(zmq.SUBSCRIBE, '')
-    poller = zmq.Poller()
+    misp_instances = json.loads(configuration_file.get('RedisGlobal', 'misp_instances'))
+    instances_status = {}
+    for misp_instance in misp_instances:
+        socket = context.socket(zmq.SUB)
+        socket.connect(misp_instance.get('zmq'))
+        socket.setsockopt_string(zmq.SUBSCRIBE, '')
+        poller = zmq.Poller()
 
-    start_time = time.time()
-    poller.register(socket, zmq.POLLIN)
-    for t in range(1, timeout+1):
-        socks = dict(poller.poll(timeout=1*1000))
-        if len(socks) > 0:
-            if socket in socks and socks[socket] == zmq.POLLIN:
-                rcv_string = socket.recv()
-                if rcv_string.startswith(b'misp_json'):
-                    return (True, '')
-        else:
-            pass
-            spinner.text = f'checking zmq - elapsed time: {int(time.time() - start_time)}s'
+        flag_skip = False
+        start_time = time.time()
+        poller.register(socket, zmq.POLLIN)
+        for t in range(1, timeout+1):
+            socks = dict(poller.poll(timeout=1*1000))
+            if len(socks) > 0:
+                if socket in socks and socks[socket] == zmq.POLLIN:
+                    rcv_string = socket.recv()
+                    if rcv_string.startswith(b'misp_json'):
+                        instances_status[misp_instance.get('name')] = True
+                        flag_skip = True
+                        break
+            else:
+                spinner.text = f'checking zmq of {misp_instance.get("name")} - elapsed time: {int(time.time() - start_time)}s'
+        if not flag_skip:
+            instances_status[misp_instance.get('name')] = False
+
+    results = [s for n, s in instances_status.items()]
+    if all(results):
+        return (True, '')
+    elif any(results):
+        return_text = 'Connection to ZMQ stream(s) failed.\n'
+        for name, status in instances_status.items():
+            return_text += f'\t➥ {name}: {"success" if status else "failed"}\n'
+        return (True, return_text)
     else:
-        return (False, '''Can\'t connect to the ZMQ stream.
+        return (False, '''Can\'t connect to the ZMQ stream(s).
 \t➥ Make sure the MISP ZMQ is running: `/servers/serverSettings/diagnostics`
 \t➥ Make sure your network infrastucture allows you to connect to the ZMQ''')
 
@@ -202,11 +218,8 @@ def check_processes_status(spinner):
         universal_newlines=True
     )
     for line in response.splitlines():
-        lines = line.split(' ')
-        if len(lines) == 2:
-            pid, p_name = lines
-        elif len(lines) ==3:
-            pid, _, p_name = lines
+        lines = line.split(' ', maxsplit=1)
+        pid, p_name = lines
 
         if 'zmq_subscriber.py' in p_name:
             pgrep_subscriber_output = line

install_dependencies.sh

@@ -6,6 +6,15 @@
 ## Debug mode
 #set -x
 
+sudo chmod -R g+w . 
+
+if ! id zmqs >/dev/null 2>&1; then
+	# Create zmq user
+	sudo useradd -U -G www-data -m -s /bin/bash  zmqs
+	# Adds right to www-data to run ./start-zmq as zmq
+	sudo echo "www-data ALL=(zmqs)	NOPASSWD:/bin/bash	/var/www/misp-dashboard/start_zmq.sh" > /etc/sudoers.d/www-data
+fi
+
 sudo apt-get install python3-virtualenv virtualenv screen redis-server unzip -y
 
 if [ -z "$VIRTUAL_ENV" ]; then

start_all.sh

@@ -39,8 +39,6 @@ fi
 
 netstat -an |grep LISTEN |grep 6250 |grep -v tcp6 ; check_redis_port=$?
 netstat -an |grep LISTEN |grep 8001 |grep -v tcp6 ; check_dashboard_port=$?
-ps auxw |grep zmq_subscriber.py |grep -v grep ; check_zmq_subscriber=$?
-ps auxw |grep zmq_dispatcher.py |grep -v grep ; check_zmq_dispatcher=$?
 
 # Configure accordingly, remember: 0.0.0.0 exposes to every active IP interface, play safe and bind it to something you trust and know
 export FLASK_APP=server.py
@@ -62,22 +60,6 @@ else
     echo -e $RED"\t* NOT starting Redis server, made a very unrealiable check on port 6250, and something seems to be there… please double check if this is good!"$DEFAULT
 fi
 
-sleep 0.1
-if [ "${check_zmq_subscriber}" == "1" ]; then
-    echo -e $GREEN"\t* Launching zmq subscriber"$DEFAULT
-    ${ENV_PY} ./zmq_subscriber.py &
-else
-    echo -e $RED"\t* NOT starting zmq subscriber, made a rather unrealiable ps -auxw | grep for zmq_subscriber.py, and something seems to be there… please double check if this is good!"$DEFAULT
-fi
-
-sleep 0.1
-if [ "${check_zmq_dispatcher}" == "1" ]; then
-    echo -e $GREEN"\t* Launching zmq dispatcher"$DEFAULT
-    ${ENV_PY} ./zmq_dispatcher.py &
-else
-    echo -e $RED"\t* NOT starting zmq dispatcher, made a rather unrealiable ps -auxw | grep for zmq_dispatcher.py, and something seems to be there… please double check if this is good!"$DEFAULT
-fi
-
 sleep 0.1
 if [ "${check_dashboard_port}" == "1" ]; then
     echo -e $GREEN"\t* Launching flask server"$DEFAULT
@@ -85,3 +67,6 @@ if [ "${check_dashboard_port}" == "1" ]; then
 else
     echo -e $RED"\t* NOT starting flask server, made a very unrealiable check on port 8001, and something seems to be there… please double check if this is good!"$DEFAULT
 fi
+
+sleep 0.1
+sudo -u zmqs /bin/bash /var/www/misp-dashboard/start_zmq.sh &

start_zmq.sh

@@ -0,0 +1,42 @@
+#!/usr/bin/env bash
+
+#set -x
+
+GREEN="\\033[1;32m"
+DEFAULT="\\033[0;39m"
+RED="\\033[1;31m"
+
+# Getting CWD where bash script resides
+DIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )" && pwd )"
+DASH_HOME="${DIR}"
+
+cd ${DASH_HOME}
+
+if [ -e "${DIR}/DASHENV/bin/python" ]; then
+    echo "dashboard virtualenv seems to exist, good"
+    ENV_PY="${DIR}/DASHENV/bin/python"
+else
+    echo "Please make sure you have a dashboard environment, au revoir"
+    exit 1
+fi
+
+ps auxw |grep zmq_subscriber.py |grep -v grep ; check_zmq_subscriber=$?
+ps auxw |grep zmq_dispatcher.py |grep -v grep ; check_zmq_dispatcher=$?
+
+screen -dmS "Misp_Dashboard"
+
+sleep 0.1
+if [ "${check_zmq_subscriber}" == "1" ]; then
+    echo -e $GREEN"\t* Launching zmq subscribers"$DEFAULT
+    screen -S "Misp_Dashboard" -X screen -t "zmq-subscribers" bash -c ${ENV_PY}' ./zmq_subscribers.py; read x'
+else
+    echo -e $RED"\t* NOT starting zmq subscribers, made a rather unrealiable ps -auxw | grep for zmq_subscriber.py, and something seems to be there… please double check if this is good!"$DEFAULT
+fi
+
+sleep 0.1
+if [ "${check_zmq_dispatcher}" == "1" ]; then
+    echo -e $GREEN"\t* Launching zmq dispatcher"$DEFAULT
+    screen -S "Misp_Dashboard" -X screen -t "zmq-dispacher" bash -c ${ENV_PY}' ./zmq_dispatcher.py; read x'
+else
+    echo -e $RED"\t* NOT starting zmq dispatcher, made a rather unrealiable ps -auxw | grep for zmq_dispatcher.py, and something seems to be there… please double check if this is good!"$DEFAULT
+fi

zmq_subscriber.py

@@ -28,7 +28,6 @@ except PermissionError as error:
     sys.exit(126)
 logger = logging.getLogger('zmq_subscriber')
 
-ZMQ_URL = cfg.get('RedisGlobal', 'zmq_url')
 CHANNEL = cfg.get('RedisLog', 'channel')
 LISTNAME = cfg.get('RedisLIST', 'listName')
 
@@ -49,16 +48,17 @@ def put_in_redis_list(zmq_name, content):
     serv_list.lpush(LISTNAME, json.dumps(to_add))
     logger.debug('Pushed: {}'.format(json.dumps(to_add)))
 
-def main(zmqName):
+def main(zmqName, zmqurl):
     context = zmq.Context()
     socket = context.socket(zmq.SUB)
-    socket.connect(ZMQ_URL)
+    socket.connect(zmqurl)
     socket.setsockopt_string(zmq.SUBSCRIBE, '')
 
     while True:
         try:
             content = socket.recv()
             put_in_redis_list(zmqName, content)
+            print(zmqName, content)
         except KeyboardInterrupt:
             return
         except Exception as e:
@@ -69,10 +69,10 @@ if __name__ == "__main__":
 
     parser = argparse.ArgumentParser(description='A zmq subscriber. It subscribes to a ZMQ then redispatch it to the misp-dashboard')
     parser.add_argument('-n', '--name', required=False, dest='zmqname', help='The ZMQ feed name', default="MISP Standard ZMQ")
-    parser.add_argument('-u', '--url', required=False, dest='zmqurl', help='The URL to connect to', default=ZMQ_URL)
+    parser.add_argument('-u', '--url', required=False, dest='zmqurl', help='The URL to connect to', default="tcp://localhost:50000")
     args = parser.parse_args()
 
     try:
-        main(args.zmqname)
+        main(args.zmqname, args.zmqurl)
     except redis.exceptions.ResponseError as error:
         print(error)

zmq_subscribers.py

@@ -0,0 +1,74 @@
+#!/usr/bin/env python3
+
+import time, datetime
+import logging
+import redis
+import configparser
+import argparse
+import os
+import subprocess
+import sys
+import json
+import atexit
+import signal
+import shlex
+import pty
+import threading
+
+configfile = os.path.join(os.path.dirname(os.path.realpath(__file__)), 'config/config.cfg')
+cfg = configparser.ConfigParser()
+cfg.read(configfile)
+logDir = cfg.get('Log', 'directory')
+logfilename = cfg.get('Log', 'subscriber_filename')
+logPath = os.path.join(logDir, logfilename)
+if not os.path.exists(logDir):
+    os.makedirs(logDir)
+logging.basicConfig(filename=logPath, filemode='a', level=logging.INFO)
+logger = logging.getLogger('zmq_subscriber')
+
+CHANNEL = cfg.get('RedisLog', 'channel')
+LISTNAME = cfg.get('RedisLIST', 'listName')
+
+serv_list = redis.StrictRedis(
+        host=cfg.get('RedisGlobal', 'host'),
+        port=cfg.getint('RedisGlobal', 'port'),
+        db=cfg.getint('RedisLIST', 'db'))
+
+children = []
+
+def signal_handler(signal, frame):
+    for child in children:
+        # We don't resume as we are already attached
+        cmd = "screen -p"+child+" -X {arg}"
+        argsc = shlex.split(cmd.format(arg = "kill"))
+        print("\n\033[1;31m [-] Terminating {child}\033[0;39m".format(child=child))
+        logger.info('Terminate: {child}'.format(child=child))
+        subprocess.call(argsc) # kill window
+    sys.exit(0)
+
+###############
+## MAIN LOOP ##
+###############
+
+def main():
+    print("\033[1;31m [+] I am the subscriber's master - kill me to kill'em'all \033[0;39m")
+    # screen needs a shell and I an no fan of shell=True
+    (master, slave) = pty.openpty()
+    try:
+        for item in json.loads(cfg.get('RedisGlobal', 'misp_instances')):
+            name = shlex.quote(item.get("name"))
+            zmq = shlex.quote(item.get("zmq"))
+            print("\033[1;32m [+] Subscribing to "+zmq+"\033[0;39m")
+            logger.info('Launching: {child}'.format(child=name))
+            children.append(name)
+            subprocess.Popen(["screen", "-r", "Misp_Dashboard", "-X", "screen", "-t", name ,sys.executable, "./zmq_subscriber.py", "-n", name, "-u", zmq], close_fds=True, shell=False, stdin=slave, stdout=slave, stderr=slave)
+    except ValueError as error:
+        print("\033[1;31m [!] Fatal exception: {error} \033[0;39m".format(error=error))
+        logger.error("JSON error: %s", error)
+        sys.exit(1)
+    signal.signal(signal.SIGINT, signal_handler)
+    forever = threading.Event()
+    forever.wait()  # Wait for SIGINT
+
+if __name__ == "__main__":
+    main()