Merge branch 'pipeio'

Rate limit magic

.drone.yml

@@ -1,6 +1,6 @@
 pipeline:
   build:
-    image: eu.gcr.io/avatao-public/docker:el7
+    image: eu.gcr.io/avatao-challengestore/challenge-builder
     volumes:
       - /etc/docker:/etc/docker:ro
       - /root/.docker:/root/.docker:ro

Dockerfile

@@ -10,6 +10,7 @@ RUN curl -sL https://deb.nodesource.com/setup_8.x | sudo -E bash -
         supervisor                             \
         libzmq5                                \
         nginx                                  \
+        jq                                     \
         gettext-base                                                                                   &&\
     rm -rf /var/lib/apt/lists/*                                                                        &&\
     ln -sf /bin/bash /bin/sh
@@ -34,10 +35,11 @@ ENV PYTHONPATH="/usr/local/lib"                                            \
     TFW_NGINX_CONF="/etc/nginx/nginx.conf"                                 \
     TFW_NGINX_DEFAULT="/etc/nginx/sites-enabled/default"                   \
     TFW_NGINX_COMPONENTS="/etc/nginx/components"                           \
-    TFW_LIB_DIR="/usr/local/lib/"                                          \
+    TFW_LIB_DIR="/usr/local/lib"                                           \
     TFW_TERMINADO_DIR="/tmp/terminado_server"                              \
     TFW_FRONTEND_DIR="/srv/frontend"                                       \
     TFW_SERVER_DIR="/srv/.tfw"                                             \
+    TFW_AUTH_KEY="/tmp/tfw-auth.key"                                       \
     TFW_HISTFILE="/home/${AVATAO_USER}/.bash_history"                      \
     PROMPT_COMMAND="history -a"
 
@@ -50,7 +52,7 @@ COPY supervisor/components/ ${TFW_SUPERVISORD_COMPONENTS}
 COPY nginx/nginx.conf ${TFW_NGINX_CONF}
 COPY nginx/default.conf ${TFW_NGINX_DEFAULT}
 COPY nginx/components/ ${TFW_NGINX_COMPONENTS}
-COPY lib LICENSE ${TFW_LIB_DIR}
+COPY lib LICENSE ${TFW_LIB_DIR}/
 COPY supervisor/tfw_server.py ${TFW_SERVER_DIR}/
 
 RUN for dir in "${TFW_LIB_DIR}"/{tfw,tao,envvars} "/etc/nginx" "/etc/supervisor"; do \
@@ -66,7 +68,7 @@ ONBUILD COPY ${BUILD_CONTEXT}/supervisor/ ${TFW_SUPERVISORD_COMPONENTS}
 ONBUILD RUN for f in "${TFW_NGINX_DEFAULT}" ${TFW_NGINX_COMPONENTS}/*.conf; do                               \
                 envsubst "$(printenv | cut -d= -f1 | grep TFW_ | sed -e 's/^/$/g')" < $f > $f~ && mv $f~ $f ;\
             done
-ONBUILD VOLUME ["/etc/nginx", "/var/lib/nginx", "/var/log/nginx"]
+ONBUILD VOLUME ["/etc/nginx", "/var/lib/nginx", "/var/log/nginx", "${TFW_LIB_DIR}/envvars", "${TFW_LIB_DIR}/tfw"]
 
 ONBUILD COPY ${BUILD_CONTEXT}/frontend /data/
 ONBUILD RUN test -z "${NOFRONTEND}" && cd /data && yarn install --frozen-lockfile || :

README.md

@@ -244,7 +244,7 @@ To set the tail length of logs (the monitor will send back the last `value` char
 
 ### FSMManagingEventHandler
 
-To attempt executing a trigger on the FSM use:
+To attempt executing a trigger on the FSM use (this will also generate an FSM update message):
 ```
 {
     "key": "fsm",
@@ -266,3 +266,16 @@ To force the broadcasting of an FSM update you can use this message:
     }
 }
 ```
+
+This event handler broadcasts FSM update messages after handling commands in the following format:
+```
+{
+    "key": "fsm_update",
+    "data" :
+    {
+        "current_state": ...string...,
+        "valid_transitions": ...array of {"trigger": ...string...} objects...
+    }
+}
+```
+

lib/tfw/__init__.py

@@ -1,7 +1,7 @@
 # Copyright (C) 2018 Avatao.com Innovative Learning Kft.
 # All Rights Reserved. See LICENSE file for details.
 
-from .event_handler_base import EventHandlerBase, TriggeredEventHandler, BroadcastingEventHandler
+from .event_handler_base import EventHandlerBase, FSMAwareEventHandler, BroadcastingEventHandler
 from .fsm_base import FSMBase
 from .linear_fsm import LinearFSM
 from .yaml_fsm import YamlFSM

lib/tfw/components/__init__.py

@@ -9,3 +9,6 @@ from .history_monitor import HistoryMonitor, BashMonitor, GDBMonitor
 from .terminal_commands import TerminalCommands
 from .log_monitoring_event_handler import LogMonitoringEventHandler
 from .fsm_managing_event_handler import FSMManagingEventHandler
+from .snapshot_provider import SnapshotProvider
+from .pipe_io_event_handler import PipeIOEventHandlerBase, PipeIOEventHandler, PipeIOServer
+from .pipe_io_event_handler import TransformerPipeIOEventHandler, CommandEventHandler

lib/tfw/components/fsm_managing_event_handler.py

@@ -2,16 +2,19 @@
 # All Rights Reserved. See LICENSE file for details.
 
 from tfw import EventHandlerBase
+from tfw.crypto import KeyManager, sign_message, verify_message
 from tfw.config.logs import logging
 
 LOG = logging.getLogger(__name__)
 
 
 class FSMManagingEventHandler(EventHandlerBase):
-    def __init__(self, key, fsm_type):
+    def __init__(self, key, fsm_type, require_signature=False):
         super().__init__(key)
         self.fsm = fsm_type()
         self._fsm_updater = FSMUpdater(self.fsm)
+        self.auth_key = KeyManager().auth_key
+        self._require_signature = require_signature
 
         self.command_handlers = {
             'trigger': self.handle_trigger,
@@ -20,19 +23,29 @@ class FSMManagingEventHandler(EventHandlerBase):
 
     def handle_event(self, message):
         try:
-            data = message['data']
+            message = self.command_handlers[message['data']['command']](message)
-            message['data'] = self.command_handlers[data['command']](data)
+            if message:
-            self.server_connector.broadcast(self._fsm_updater.generate_fsm_update())
+                fsm_update_message = self._fsm_updater.generate_fsm_update()
+                sign_message(self.auth_key, message)
+                sign_message(self.auth_key, fsm_update_message)
+                self.server_connector.broadcast(fsm_update_message)
             return message
         except KeyError:
             LOG.error('IGNORING MESSAGE: Invalid message received: %s', message)
 
-    def handle_trigger(self, data):
+    def handle_trigger(self, message):
-        self.fsm.step(data['value'])
+        trigger = message['data']['value']
-        return data
+        if self._require_signature:
+            if not verify_message(self.auth_key, message):
+                LOG.error('Ignoring unsigned trigger command: %s', message)
+                return None
+        if self.fsm.step(trigger):
+            return message
+        return None
 
-    def handle_update(self, data):
+    def handle_update(self, message):
-        return data
+        # pylint: disable=no-self-use
+        return message
 
 
 class FSMUpdater:
@@ -51,7 +64,11 @@ class FSMUpdater:
             {'trigger': trigger}
             for trigger in self.fsm.get_triggers(self.fsm.state)
         ]
+        last_trigger = self.fsm.trigger_history[-1] if self.fsm.trigger_history else None
+        in_accepted_state = state in self.fsm.accepted_states
         return {
             'current_state': state,
-            'valid_transitions': valid_transitions
+            'valid_transitions': valid_transitions,
+            'last_trigger': last_trigger,
+            'in_accepted_state': in_accepted_state
         }

lib/tfw/components/ide_event_handler.py

@@ -4,7 +4,7 @@
 from os.path import isfile, join, relpath, exists, isdir, realpath
 from glob import glob
 from fnmatch import fnmatchcase
-from collections import Iterable
+from typing import Iterable
 
 from tfw import EventHandlerBase
 from tfw.mixins import MonitorManagerMixin
@@ -103,7 +103,7 @@ class FileManager:  # pylint: disable=too-many-instance-attributes
 
 
 class IdeEventHandler(EventHandlerBase, MonitorManagerMixin):
-    # pylint: disable=too-many-arguments
+    # pylint: disable=too-many-arguments,anomalous-backslash-in-string
     """
     Event handler implementing the backend of our browser based IDE.
     By default all files in the directory specified in __init__ are displayed

lib/tfw/components/pipe_io_event_handler.py

@@ -0,0 +1,143 @@
+from abc import abstractmethod
+from json import loads, dumps
+from subprocess import run, PIPE, Popen
+from functools import partial
+from os import getpgid, killpg
+from os.path import join
+from signal import SIGTERM
+from secrets import token_urlsafe
+from threading import Thread
+from contextlib import suppress
+
+from tfw import EventHandlerBase
+from tfw.config.logs import logging
+
+from .pipe_io_server import PipeIOServer, terminate_process_on_failure
+
+LOG = logging.getLogger(__name__)
+DEFAULT_PERMISSIONS = 0o600
+
+
+class PipeIOEventHandlerBase(EventHandlerBase):
+    def __init__(self, key, in_pipe_path, out_pipe_path, permissions=DEFAULT_PERMISSIONS):
+        super().__init__(key)
+        self.pipe_io = CallbackPipeIOServer(
+            in_pipe_path,
+            out_pipe_path,
+            self.handle_pipe_event,
+            permissions
+        )
+        self.pipe_io.start()
+
+    @abstractmethod
+    def handle_pipe_event(self, message_bytes):
+        raise NotImplementedError()
+
+    def cleanup(self):
+        self.pipe_io.stop()
+
+
+class CallbackPipeIOServer(PipeIOServer):
+    def __init__(self, in_pipe_path, out_pipe_path, callback, permissions):
+        super().__init__(in_pipe_path, out_pipe_path, permissions)
+        self.callback = callback
+
+    def handle_message(self, message):
+        try:
+            self.callback(message)
+        except:  # pylint: disable=bare-except
+            LOG.exception('Failed to handle message %s from pipe %s!', message, self.in_pipe)
+
+
+class PipeIOEventHandler(PipeIOEventHandlerBase):
+    def handle_event(self, message):
+        json_bytes = dumps(message).encode()
+        self.pipe_io.send_message(json_bytes)
+
+    def handle_pipe_event(self, message_bytes):
+        json = loads(message_bytes)
+        self.server_connector.send(json)
+
+
+class TransformerPipeIOEventHandler(PipeIOEventHandlerBase):
+    # pylint: disable=too-many-arguments
+    def __init__(
+            self, key, in_pipe_path, out_pipe_path,
+            transform_in_cmd, transform_out_cmd,
+            permissions=DEFAULT_PERMISSIONS
+    ):
+        self._transform_in = partial(self._transform_message, transform_in_cmd)
+        self._transform_out = partial(self._transform_message, transform_out_cmd)
+        super().__init__(key, in_pipe_path, out_pipe_path, permissions)
+
+    @staticmethod
+    def _transform_message(transform_cmd, message):
+        proc = run(
+            transform_cmd,
+            input=message,
+            stdout=PIPE,
+            stderr=PIPE,
+            shell=True
+        )
+        if proc.returncode == 0:
+            return proc.stdout
+        raise ValueError(f'Transforming message {message} failed!')
+
+    def handle_event(self, message):
+        json_bytes = dumps(message).encode()
+        transformed_bytes = self._transform_out(json_bytes)
+        if transformed_bytes:
+            self.pipe_io.send_message(transformed_bytes)
+
+    def handle_pipe_event(self, message_bytes):
+        transformed_bytes = self._transform_in(message_bytes)
+        if transformed_bytes:
+            json_message = loads(transformed_bytes)
+            self.server_connector.send(json_message)
+
+
+class CommandEventHandler(PipeIOEventHandler):
+    def __init__(self, key, command, permissions=DEFAULT_PERMISSIONS):
+        super().__init__(
+            key,
+            self._generate_tempfilename(),
+            self._generate_tempfilename(),
+            permissions
+        )
+
+        self._proc_stdin = open(self.pipe_io.out_pipe, 'rb')
+        self._proc_stdout = open(self.pipe_io.in_pipe, 'wb')
+        self._proc = Popen(
+            command, shell=True, executable='/bin/bash',
+            stdin=self._proc_stdin, stdout=self._proc_stdout, stderr=PIPE,
+            start_new_session=True
+        )
+        self._monitor_proc_thread = self._start_monitor_proc()
+
+    def _generate_tempfilename(self):
+        # pylint: disable=no-self-use
+        random_filename = partial(token_urlsafe, 10)
+        return join('/tmp', f'{type(self).__name__}.{random_filename()}')
+
+    def _start_monitor_proc(self):
+        thread = Thread(target=self._monitor_proc, daemon=True)
+        thread.start()
+        return thread
+
+    @terminate_process_on_failure
+    def _monitor_proc(self):
+        return_code = self._proc.wait()
+        if return_code == -int(SIGTERM):
+            # supervisord asked the program to terminate, this is fine
+            return
+        if return_code != 0:
+            _, stderr = self._proc.communicate()
+            raise RuntimeError(f'Subprocess failed ({return_code})! Stderr:\n{stderr.decode()}')
+
+    def cleanup(self):
+        with suppress(ProcessLookupError):
+            process_group_id = getpgid(self._proc.pid)
+            killpg(process_group_id, SIGTERM)
+            self._proc_stdin.close()
+            self._proc_stdout.close()
+        super().cleanup()

lib/tfw/components/pipe_io_server/__init__.py

@@ -0,0 +1,2 @@
+from .pipe_io_server import PipeIOServer
+from .terminate_process_on_failure import terminate_process_on_failure

lib/tfw/components/pipe_io_server/deque.py

@@ -0,0 +1,27 @@
+from collections import deque
+from threading import Lock, Condition
+
+
+class Deque:
+    def __init__(self):
+        self._queue = deque()
+
+        self._mutex = Lock()
+        self._not_empty = Condition(self._mutex)
+
+    def pop(self):
+        with self._mutex:
+            while not self._queue:
+                self._not_empty.wait()
+            return self._queue.pop()
+
+    def push(self, item):
+        self._push(item, self._queue.appendleft)
+
+    def push_front(self, item):
+        self._push(item, self._queue.append)
+
+    def _push(self, item, put_method):
+        with self._mutex:
+            put_method(item)
+            self._not_empty.notify()

lib/tfw/components/pipe_io_server/pipe.py

@@ -0,0 +1,16 @@
+from os import mkfifo, remove, chmod
+from os.path import exists
+
+
+class Pipe:
+    def __init__(self, path):
+        self.path = path
+
+    def recreate(self, permissions):
+        self.remove()
+        mkfifo(self.path)
+        chmod(self.path, permissions)  # use chmod to ignore umask
+
+    def remove(self):
+        if exists(self.path):
+            remove(self.path)

lib/tfw/components/pipe_io_server/pipe_io_server.py

@@ -0,0 +1,73 @@
+from abc import ABC, abstractmethod
+from threading import Thread, Event
+from typing import Callable
+
+from .pipe_reader_thread import PipeReaderThread
+from .pipe_writer_thread import PipeWriterThread
+from .pipe import Pipe
+from .terminate_process_on_failure import terminate_process_on_failure
+
+
+class PipeIOServer(ABC, Thread):
+    def __init__(self, in_pipe=None, out_pipe=None, permissions=0o600):
+        super().__init__(daemon=True)
+        self._in_pipe, self._out_pipe = in_pipe, out_pipe
+        self._create_pipes(permissions)
+        self._stop_event = Event()
+        self._reader_thread, self._writer_thread = self._create_io_threads()
+        self._io_threads = (self._reader_thread, self._writer_thread)
+        self._on_stop = lambda: None
+
+    def _create_pipes(self, permissions):
+        Pipe(self.in_pipe).recreate(permissions)
+        Pipe(self.out_pipe).recreate(permissions)
+
+    @property
+    def in_pipe(self):
+        return self._in_pipe
+
+    @property
+    def out_pipe(self):
+        return self._out_pipe
+
+    def _create_io_threads(self):
+        reader_thread = PipeReaderThread(self.in_pipe, self._stop_event, self.handle_message)
+        writer_thread = PipeWriterThread(self.out_pipe, self._stop_event)
+        return reader_thread, writer_thread
+
+    @abstractmethod
+    def handle_message(self, message):
+        raise NotImplementedError()
+
+    def send_message(self, message):
+        self._writer_thread.write(message)
+
+    @terminate_process_on_failure
+    def run(self):
+        for thread in self._io_threads:
+            thread.start()
+        self._stop_event.wait()
+        self._stop_threads()
+
+    def stop(self):
+        self._stop_event.set()
+        if self.is_alive():
+            self.join()
+
+    def _stop_threads(self):
+        for thread in self._io_threads:
+            if thread.is_alive():
+                thread.stop()
+        Pipe(self.in_pipe).remove()
+        Pipe(self.out_pipe).remove()
+        self._on_stop()
+
+    def _set_on_stop(self, value):
+        if not isinstance(value, Callable):
+            raise ValueError("Supplied object is not callable!")
+        self._on_stop = value
+
+    on_stop = property(fset=_set_on_stop)
+
+    def wait(self):
+        self._stop_event.wait()

lib/tfw/components/pipe_io_server/pipe_reader_thread.py

@@ -0,0 +1,44 @@
+from contextlib import suppress
+from os import open as osopen
+from os import write, close, O_WRONLY, O_NONBLOCK
+from threading import Thread
+
+from .terminate_process_on_failure import terminate_process_on_failure
+
+
+class PipeReaderThread(Thread):
+    eof = b''
+    stop_sequence = b'stop_reading\n'
+
+    def __init__(self, pipe_path, stop_event, message_handler):
+        super().__init__(daemon=True)
+        self._message_handler = message_handler
+        self._pipe_path = pipe_path
+        self._stop_event = stop_event
+
+    @terminate_process_on_failure
+    def run(self):
+        with self._open() as pipe:
+            while True:
+                message = pipe.readline()
+                if message == self.stop_sequence:
+                    self._stop_event.set()
+                    break
+                if message == self.eof:
+                    self._open().close()
+                    continue
+                self._message_handler(message[:-1])
+
+    def _open(self):
+        return open(self._pipe_path, 'rb')
+
+    def stop(self):
+        while self.is_alive():
+            self._unblock()
+        self.join()
+
+    def _unblock(self):
+        with suppress(OSError):
+            fd = osopen(self._pipe_path, O_WRONLY | O_NONBLOCK)
+            write(fd, self.stop_sequence)
+            close(fd)

lib/tfw/components/pipe_io_server/pipe_writer_thread.py

@@ -0,0 +1,50 @@
+from contextlib import suppress
+from os import O_NONBLOCK, O_RDONLY, close
+from os import open as osopen
+from threading import Thread
+
+from .terminate_process_on_failure import terminate_process_on_failure
+from .deque import Deque
+
+
+class PipeWriterThread(Thread):
+    def __init__(self, pipe_path, stop_event):
+        super().__init__(daemon=True)
+        self._pipe_path = pipe_path
+        self._stop_event = stop_event
+        self._write_queue = Deque()
+
+    def write(self, message):
+        self._write_queue.push(message)
+
+    @terminate_process_on_failure
+    def run(self):
+        with self._open() as pipe:
+            while True:
+                message = self._write_queue.pop()
+                if message is None:
+                    self._stop_event.set()
+                    break
+                try:
+                    pipe.write(message + b'\n')
+                    pipe.flush()
+                except BrokenPipeError:
+                    try:  # pipe was reopened, close() flushed the message
+                        pipe.close()
+                    except BrokenPipeError:  # close() discarded the message
+                        self._write_queue.push_front(message)
+                    pipe = self._open()
+
+    def _open(self):
+        return open(self._pipe_path, 'wb')
+
+    def stop(self):
+        while self.is_alive():
+            self._unblock()
+        self.join()
+
+    def _unblock(self):
+        with suppress(OSError):
+            fd = osopen(self._pipe_path, O_RDONLY | O_NONBLOCK)
+            self._write_queue.push_front(None)
+            close(fd)

lib/tfw/components/pipe_io_server/terminate_process_on_failure.py

@@ -0,0 +1,15 @@
+from functools import wraps
+from os import kill, getpid
+from signal import SIGTERM
+from traceback import print_exc
+
+
+def terminate_process_on_failure(fun):
+    @wraps(fun)
+    def wrapper(*args, **kwargs):
+        try:
+            return fun(*args, **kwargs)
+        except:  # pylint: disable=bare-except
+            print_exc()
+            kill(getpid(), SIGTERM)
+    return wrapper

lib/tfw/components/process_managing_event_handler.py

@@ -38,7 +38,6 @@ class ProcessManagingEventHandler(EventHandlerBase):
     """
     def __init__(self, key, dirmonitor=None, log_tail=0):
         super().__init__(key)
-        self.key = key
         self.monitor = dirmonitor
         self.processmanager = ProcessManager()
         self.log_tail = log_tail

lib/tfw/components/snapshot_provider.py

@@ -0,0 +1,175 @@
+# Copyright (C) 2018 Avatao.com Innovative Learning Kft.
+# All Rights Reserved. See LICENSE file for details.
+
+import re
+from subprocess import run, CalledProcessError
+from getpass import getuser
+from os.path import isdir
+from datetime import datetime
+from uuid import uuid4
+
+
+class SnapshotProvider:
+    def __init__(self, directory, git_dir):
+        self._classname = self.__class__.__name__
+        author = f'{getuser()} via TFW {self._classname}'
+        self.gitenv = {
+            'GIT_DIR': git_dir,
+            'GIT_WORK_TREE': directory,
+            'GIT_AUTHOR_NAME': author,
+            'GIT_AUTHOR_EMAIL': '',
+            'GIT_COMMITTER_NAME': author,
+            'GIT_COMMITTER_EMAIL': '',
+            'GIT_PAGER': 'cat'
+        }
+
+        self._init_repo()
+        self.__last_valid_branch = self._branch
+
+    def _init_repo(self):
+        self._check_environment()
+
+        if not self._repo_is_initialized:
+            self._run(('git', 'init'))
+
+        if self._number_of_commits == 0:
+            try:
+                self._snapshot()
+            except CalledProcessError:
+                raise EnvironmentError(f'{self._classname} cannot init on empty directories!')
+
+        self._check_head_not_detached()
+
+    def _check_environment(self):
+        if not isdir(self.gitenv['GIT_DIR']) or not isdir(self.gitenv['GIT_WORK_TREE']):
+            raise EnvironmentError(f'{self._classname}: "directory" and "git_dir" must exist!')
+
+    @property
+    def _repo_is_initialized(self):
+        return self._run(
+            ('git', 'status'),
+            check=False
+        ).returncode == 0
+
+    @property
+    def _number_of_commits(self):
+        return int(
+            self._get_stdout((
+                'git', 'rev-list',
+                '--all',
+                '--count'
+            ))
+        )
+
+    def _snapshot(self):
+        self._run((
+            'git', 'add',
+            '-A'
+        ))
+        self._run((
+            'git', 'commit',
+            '-m', 'Snapshot'
+        ))
+
+    def _check_head_not_detached(self):
+        if self._head_detached:
+            raise EnvironmentError(f'{self._classname} cannot init from detached HEAD state!')
+
+    @property
+    def _head_detached(self):
+        return self._branch == 'HEAD'
+
+    @property
+    def _branch(self):
+        return self._get_stdout((
+            'git', 'rev-parse',
+            '--abbrev-ref', 'HEAD'
+        ))
+
+    def _get_stdout(self, *args, **kwargs):
+        kwargs['capture_output'] = True
+        stdout_bytes = self._run(*args, **kwargs).stdout
+        return stdout_bytes.decode().rstrip('\n')
+
+    def _run(self, *args, **kwargs):
+        if 'check' not in kwargs:
+            kwargs['check'] = True
+        if 'env' not in kwargs:
+            kwargs['env'] = self.gitenv
+        return run(*args, **kwargs)
+
+    def take_snapshot(self):
+        if self._head_detached:
+            self._checkout_new_branch_from_head()
+        self._snapshot()
+
+    def _checkout_new_branch_from_head(self):
+        branch_name = uuid4()
+        self._run((
+            'git', 'branch',
+            branch_name
+        ))
+        self._checkout(branch_name)
+
+    def _checkout(self, what):
+        self._run((
+            'git', 'checkout',
+            what
+        ))
+
+    def restore_snapshot(self, date):
+        commit = self._get_commit_from_timestamp(date)
+        self._checkout(commit)
+
+    def _get_commit_from_timestamp(self, date):
+        return self._get_stdout((
+            'git', 'rev-list',
+            '--date=iso',
+            '-n', '1',
+            f'--before="{date.isoformat()}"',
+            self._last_valid_branch
+        ))
+
+    @property
+    def _last_valid_branch(self):
+        if not self._head_detached:
+            self.__last_valid_branch = self._branch
+        return self.__last_valid_branch
+
+    @property
+    def all_timelines(self):
+        return self._branches
+
+    @property
+    def _branches(self):
+        git_branch_output = self._get_stdout(('git', 'branch'))
+        regex_pattern = re.compile(r'(?:[^\S\n]|[*])')  # matches '*' and non-newline whitespace chars
+        return re.sub(regex_pattern, '', git_branch_output).splitlines()
+
+    @property
+    def timeline(self):
+        return self._last_valid_branch
+
+    @timeline.setter
+    def timeline(self, value):
+        self._checkout(value)
+
+    @property
+    def snapshots(self):
+        return self._pretty_log_branch()
+
+    def _pretty_log_branch(self):
+        git_log_output = self._get_stdout((
+            'git', 'log',
+            '--pretty=%H@%aI'
+        ))
+
+        commits = []
+        for line in git_log_output.splitlines():
+            commit_hash, timestamp = line.split('@')
+            commits.append({
+                'hash': commit_hash,
+                'timestamp': datetime.fromisoformat(timestamp)
+            })
+
+        return commits

lib/tfw/components/terminal_commands.py

@@ -11,6 +11,7 @@ LOG = logging.getLogger(__name__)
 
 
 class TerminalCommands(ABC):
+    # pylint: disable=anomalous-backslash-in-string
     """
     A class you can use to define hooks for terminal commands. This means that you can
     have python code executed when the user enters a specific command to the terminal on

lib/tfw/components/terminal_event_handler.py

@@ -68,6 +68,7 @@ class TerminalEventHandler(EventHandlerBase):
                      (command to be written to the pty)
         """
         self.terminado_server.pty.write(data['value'])
+        return data
 
     def read(self, data):
         """

lib/tfw/crypto.py

@@ -0,0 +1,107 @@
+# Copyright (C) 2018 Avatao.com Innovative Learning Kft.
+# All Rights Reserved. See LICENSE file for details.
+
+from functools import wraps
+from base64 import b64encode, b64decode
+from copy import deepcopy
+from hashlib import md5
+from os import urandom, chmod
+from os.path import exists
+from stat import S_IRUSR, S_IWUSR, S_IXUSR
+
+from cryptography.hazmat.backends import default_backend
+from cryptography.hazmat.primitives.hashes import SHA256
+from cryptography.hazmat.primitives.hmac import HMAC as _HMAC
+from cryptography.exceptions import InvalidSignature
+
+from tfw.networking import message_bytes
+from tfw.decorators import lazy_property
+from tfw.config import TFWENV
+
+
+def message_checksum(message):
+    return md5(message_bytes(message)).hexdigest()
+
+
+def sign_message(key, message):
+    signature = message_signature(key, message)
+    message['signature'] = b64encode(signature).decode()
+
+
+def message_signature(key, message):
+    return HMAC(key, message_bytes(message)).signature
+
+
+def verify_message(key, message):
+    message = deepcopy(message)
+    try:
+        signature_b64 = message.pop('signature')
+        signature = b64decode(signature_b64)
+        actual_signature = message_signature(key, message)
+        return signature == actual_signature
+    except KeyError:
+        return False
+
+
+class KeyManager:
+    def __init__(self):
+        self.keyfile = TFWENV.AUTH_KEY
+        if not exists(self.keyfile):
+            self._init_auth_key()
+
+    @lazy_property
+    def auth_key(self):
+        with open(self.keyfile, 'rb') as ifile:
+            return ifile.read()
+
+    def _init_auth_key(self):
+        key = self.generate_key()
+        with open(self.keyfile, 'wb') as ofile:
+            ofile.write(key)
+        self._chmod_700_keyfile()
+        return key
+
+    @staticmethod
+    def generate_key():
+        return urandom(32)
+
+    def _chmod_700_keyfile(self):
+        chmod(self.keyfile, S_IRUSR | S_IWUSR | S_IXUSR)
+
+
+class HMAC:
+    def __init__(self, key, message):
+        self.key = key
+        self.message = message
+        self._hmac = _HMAC(
+            key=key,
+            algorithm=SHA256(),
+            backend=default_backend()
+        )
+
+    def _reload_if_finalized(f):
+        # pylint: disable=no-self-argument,not-callable
+        @wraps(f)
+        def wrapped(instance, *args, **kwargs):
+            if getattr(instance, '_finalized', False):
+                instance.__init__(instance.key, instance.message)
+            ret_val = f(instance, *args, **kwargs)
+            setattr(instance, '_finalized', True)
+            return ret_val
+        return wrapped
+
+    @property
+    @_reload_if_finalized
+    def signature(self):
+        self._hmac.update(self.message)
+        signature = self._hmac.finalize()
+        return signature
+
+    @_reload_if_finalized
+    def verify(self, signature):
+        self._hmac.update(self.message)
+        try:
+            self._hmac.verify(signature)
+            return True
+        except InvalidSignature:
+            return False

lib/tfw/decorators/rate_limiter.py

@@ -1,25 +1,100 @@
 # Copyright (C) 2018 Avatao.com Innovative Learning Kft.
 # All Rights Reserved. See LICENSE file for details.
 
-from functools import wraps
+from functools import wraps, partial
 from time import time, sleep
 
+from tfw.decorators.lazy_property import lazy_property
+
 
 class RateLimiter:
+    """
+    Decorator class for rate limiting, blocking.
+
+    When applied to a function this decorator will apply rate limiting
+    if the function is invoked more frequently than rate_per_seconds.
+
+    By default rate limiting means sleeping until the next invocation time
+    as per __init__ parameter rate_per_seconds.
+
+    Note that this decorator BLOCKS THE THREAD it is being executed on,
+    so it is only acceptable for stuff running on a separate thread.
+
+    If this is no good for you please refer to AsyncRateLimiter in this module,
+    which is designed not to block and use the IOLoop it is being called from.
+    """
     def __init__(self, rate_per_second):
+        """
+        :param rate_per_second: max frequency the decorated method should be
+                                invoked with
+        """
         self.min_interval = 1 / float(rate_per_second)
+        self.fun = None
         self.last_call = time()
 
+    def action(self, seconds_to_next_call):
+        if seconds_to_next_call:
+            sleep(seconds_to_next_call)
+        self.fun()
+
     def __call__(self, fun):
         @wraps(fun)
         def wrapper(*args, **kwargs):
-            self._limit_rate()
+            self.fun = partial(fun, *args, **kwargs)
-            fun(*args, **kwargs)
+            limit_seconds = self._limit_rate()
+            self.action(limit_seconds)
         return wrapper
 
     def _limit_rate(self):
-        since_last_call = time() - self.last_call
+        seconds_since_last_call = time() - self.last_call
-        to_next_call = self.min_interval - since_last_call
+        seconds_to_next_call = self.min_interval - seconds_since_last_call
+
+        if seconds_to_next_call > 0:
+            return seconds_to_next_call
         self.last_call = time()
-        if to_next_call > 0:
+        return 0
-            sleep(to_next_call)
+
+
+class AsyncRateLimiter(RateLimiter):
+    """
+    Decorator class for rate limiting, non-blocking.
+
+    The semantics of the rate limiting:
+        - unlike RateLimiter this decorator never blocks, instead it adds an async
+          callback version of the decorated function to the IOLoop
+          (to be executed after the rate limiting has expired).
+        - the timing works similarly to RateLimiter
+    """
+    def __init__(self, rate_per_second, ioloop_factory):
+        """
+        :param rate_per_second: max frequency the decorated method should be
+                                invoked with
+        :param ioloop_factory: callable that should return an instance of the
+                               IOLoop of the application
+        """
+        self._ioloop_factory = ioloop_factory
+        self._ioloop = None
+        self._last_callback = None
+
+        self._make_action_thread_safe()
+        super().__init__(rate_per_second=rate_per_second)
+
+    def _make_action_thread_safe(self):
+        self.action = partial(self.ioloop.add_callback, self.action)
+
+    @lazy_property
+    def ioloop(self):
+        return self._ioloop_factory()
+
+    def action(self, seconds_to_next_call):
+        if self._last_callback:
+            self.ioloop.remove_timeout(self._last_callback)
+
+        self._last_callback = self.ioloop.call_later(
+            seconds_to_next_call,
+            self.fun_with_debounce
+        )
+
+    def fun_with_debounce(self):
+        self.last_call = time()
+        self.fun()

lib/tfw/event_handler_base.py

@@ -2,10 +2,11 @@
 # All Rights Reserved. See LICENSE file for details.
 
 from abc import ABC, abstractmethod
-from json import dumps
+from inspect import currentframe
-from hashlib import md5
+from typing import Iterable
 
 from tfw.networking.event_handlers import ServerConnector
+from tfw.crypto import message_checksum, KeyManager, verify_message
 from tfw.config.logs import logging
 
 LOG = logging.getLogger(__name__)
@@ -20,10 +21,22 @@ class EventHandlerBase(ABC):
     """
     def __init__(self, key):
         self.server_connector = ServerConnector()
-        self.key = key
+        self.keys = []
-        self.subscribe(self.key, 'reset')
+        if isinstance(key, str):
+            self.keys.append(key)
+        elif isinstance(key, Iterable):
+            self.keys = list(key)
+
+        self.subscribe(*self.keys)
         self.server_connector.register_callback(self.event_handler_callback)
 
+    @property
+    def key(self):
+        """
+        Returns the oldest key this EventHandler was subscribed to.
+        """
+        return self.keys[0]
+
     def event_handler_callback(self, message):
         """
         Callback that is invoked when receiving a message.
@@ -48,7 +61,9 @@ class EventHandlerBase(ABC):
         subscribed to 'fsm' will receive 'fsm_update'
         messages as well.
         """
-        return self.key == message['key']
+        if '' in self.keys:
+            return True
+        return message['key'] in self.keys
 
     def dispatch_handling(self, message):
         """
@@ -57,9 +72,7 @@ class EventHandlerBase(ABC):
         :param message: the message received
         :returns: the message to send back
         """
-        if message['key'] != 'reset':
+        return self.handle_event(message)
-            return self.handle_event(message)
-        return self.handle_reset(message)
 
     @abstractmethod
     def handle_event(self, message):
@@ -71,27 +84,18 @@ class EventHandlerBase(ABC):
         """
         raise NotImplementedError
 
-    def handle_reset(self, message):
-        # pylint: disable=unused-argument,no-self-use
-        """
-        Usually 'reset' events receive some sort of special treatment.
-
-        :param message: the message received
-        :returns: the message to send back
-        """
-        return None
-
     def subscribe(self, *keys):
         """
         Subscribe this EventHandler to receive events for given keys.
         Note that you can subscribe to the same key several times in which
         case you will need to unsubscribe multiple times in order to stop
         receiving events.
-        
+
         :param keys: list of keys to subscribe to
         """
         for key in keys:
             self.server_connector.subscribe(key)
+            self.keys.append(key)
 
     def unsubscribe(self, *keys):
         """
@@ -101,6 +105,7 @@ class EventHandlerBase(ABC):
         """
         for key in keys:
             self.server_connector.unsubscribe(key)
+            self.keys.remove(key)
 
     def cleanup(self):
         """
@@ -109,21 +114,55 @@ class EventHandlerBase(ABC):
         """
         pass
 
+    @classmethod
+    def get_local_instances(cls):
+        frame = currentframe()
+        if frame is None:
+            raise EnvironmentError('inspect.currentframe() is not supported!')
 
-class TriggeredEventHandler(EventHandlerBase, ABC):
+        locals_values = frame.f_back.f_locals.values()
+        return {
+            instance for instance in locals_values
+            if isinstance(instance, cls)
+        }
+
+
+class FSMAwareEventHandler(EventHandlerBase, ABC):
     # pylint: disable=abstract-method
     """
-    Abstract base class for EventHandlers which are only triggered in case
+    Abstract base class for EventHandlers which automatically
-    TFWServer has successfully triggered an FSM step defined in __init__.
+    keep track of the state of the TFW FSM.
     """
-    def __init__(self, key, trigger):
+    def __init__(self, key):
         super().__init__(key)
-        self.trigger = trigger
+        self.subscribe('fsm_update')
+        self.fsm_state = None
+        self.in_accepted_state = False
+        self._auth_key = KeyManager().auth_key
 
     def dispatch_handling(self, message):
-        if message.get('trigger') == self.trigger:
+        if message['key'] == 'fsm_update':
-            return super().dispatch_handling(message)
+            if verify_message(self._auth_key, message):
-        return None
+                self._handle_fsm_update(message)
+            return None
+        return super().dispatch_handling(message)
+
+    def _handle_fsm_update(self, message):
+        try:
+            new_state = message['data']['current_state']
+            trigger = message['data']['last_trigger']
+            if self.fsm_state != new_state:
+                self.handle_fsm_step(self.fsm_state, new_state, trigger)
+            self.fsm_state = new_state
+            self.in_accepted_state = message['data']['in_accepted_state']
+        except KeyError:
+            LOG.error('Invalid fsm_update message received!')
+
+    def handle_fsm_step(self, from_state, to_state, trigger):
+        """
+        Called in case the TFW FSM has stepped.
+        """
+        pass
 
 
 class BroadcastingEventHandler(EventHandlerBase, ABC):
@@ -137,7 +176,7 @@ class BroadcastingEventHandler(EventHandlerBase, ABC):
         self.own_message_hashes = []
 
     def event_handler_callback(self, message):
-        message_hash = self.hash_message(message)
+        message_hash = message_checksum(message)
 
         if message_hash in self.own_message_hashes:
             self.own_message_hashes.remove(message_hash)
@@ -145,10 +184,5 @@ class BroadcastingEventHandler(EventHandlerBase, ABC):
 
         response = self.dispatch_handling(message)
         if response:
-            self.own_message_hashes.append(self.hash_message(response))
+            self.own_message_hashes.append(message_checksum(response))
             self.server_connector.broadcast(response)
-
-    @staticmethod
-    def hash_message(message):
-        message_bytes = dumps(message, sort_keys=True).encode()
-        return md5(message_bytes).hexdigest()

lib/tfw/fsm_base.py

@@ -22,8 +22,9 @@ class FSMBase(Machine, CallbackMixin):
     states, transitions = [], []
 
     def __init__(self, initial=None, accepted_states=None):
-        self.accepted_states = accepted_states or [self.states[-1]]
+        self.accepted_states = accepted_states or [self.states[-1].name]
         self.trigger_predicates = defaultdict(list)
+        self.trigger_history = []
 
         Machine.__init__(
             self,
@@ -57,9 +58,11 @@ class FSMBase(Machine, CallbackMixin):
             for predicate in self.trigger_predicates[trigger]
         )
 
-        # TODO: think about what could we do when this prevents triggering
         if all(predicate_results):
             try:
                 self.trigger(trigger)
+                self.trigger_history.append(trigger)
+                return True
             except (AttributeError, MachineError):
                 LOG.debug('FSM failed to execute nonexistent trigger: "%s"', trigger)
+        return False

lib/tfw/linear_fsm.py

@@ -1,6 +1,8 @@
 # Copyright (C) 2018 Avatao.com Innovative Learning Kft.
 # All Rights Reserved. See LICENSE file for details.
 
+from transitions import State
+
 from .fsm_base import FSMBase
 
 
@@ -13,17 +15,17 @@ class LinearFSM(FSMBase):
     (0) --  step_1  --> (1) --  step_2  --> (2) --  step_3  --> (3) ... and so on
     """
     def __init__(self, number_of_steps):
-        self.states = list(map(str, range(number_of_steps)))
+        self.states = [State(name=str(index)) for index in range(number_of_steps)]
         self.transitions = []
-        for index in self.states[:-1]:
+        for state in self.states[:-1]:
             self.transitions.append({
-                'trigger': f'step_{int(index)+1}',
+                'trigger': f'step_{int(state.name)+1}',
-                'source': index,
+                'source': state.name,
-                'dest': str(int(index)+1)
+                'dest': str(int(state.name)+1)
             })
             self.transitions.append({
                 'trigger': 'step_next',
-                'source': index,
+                'source': state.name,
-                'dest': str(int(index)+1)
+                'dest': str(int(state.name)+1)
             })
         super(LinearFSM, self).__init__()

lib/tfw/mixins/callback_mixin.py

@@ -9,6 +9,7 @@ from tfw.decorators import lazy_property
 class CallbackMixin:
     @lazy_property
     def _callbacks(self):
+        # pylint: disable=no-self-use
         return []
 
     def subscribe_callback(self, callback, *args, **kwargs):

lib/tfw/mixins/observer_mixin.py

@@ -9,6 +9,7 @@ from tfw.decorators import lazy_property
 class ObserverMixin:
     @lazy_property
     def observer(self):
+        # pylint: disable=no-self-use
         return Observer()
 
     def watch(self):

lib/tfw/mixins/supervisor_mixin.py

@@ -13,6 +13,7 @@ from tfw.config import TFWENV
 class SupervisorBaseMixin:
     @lazy_property
     def supervisor(self):
+        # pylint: disable=no-self-use
         return xmlrpc.client.ServerProxy(TFWENV.SUPERVISOR_HTTP_URI).supervisor
 
 

lib/tfw/networking/__init__.py

@@ -1,9 +1,9 @@
 # Copyright (C) 2018 Avatao.com Innovative Learning Kft.
 # All Rights Reserved. See LICENSE file for details.
 
-from .serialization import serialize_tfw_msg, deserialize_tfw_msg, with_deserialize_tfw_msg
+from .serialization import serialize_tfw_msg, deserialize_tfw_msg
+from .serialization import with_deserialize_tfw_msg, message_bytes
 from .zmq_connector_base import ZMQConnectorBase
-# from .controller_connector import ControllerConnector # TODO: readd once controller stuff is resolved
 from .message_sender import MessageSender
 from .event_handlers.server_connector import ServerUplinkConnector as TFWServerConnector
 from .server.tfw_server import TFWServer

lib/tfw/networking/controller_connector.py

@@ -1,18 +0,0 @@
-# Copyright (C) 2018 Avatao.com Innovative Learning Kft.
-# All Rights Reserved. See LICENSE file for details.
-
-import zmq
-from zmq.eventloop.zmqstream import ZMQStream
-
-from tfw.config import TFWENV
-from tfw.networking import ZMQConnectorBase
-
-
-class ControllerConnector(ZMQConnectorBase):
-    def __init__(self, zmq_context=None):
-        super(ControllerConnector, self).__init__(zmq_context)
-        self._zmq_rep_socket = self._zmq_context.socket(zmq.REP)
-        self._zmq_rep_socket.connect(f'tcp://localhost:{TFWENV.CONTROLLER_PORT}')
-        self._zmq_rep_stream = ZMQStream(self._zmq_rep_socket)
-
-        self.register_callback = self._zmq_rep_stream.on_recv_stream

lib/tfw/networking/message_sender.py

@@ -44,3 +44,11 @@ class MessageSender:
             'key': self.queue_key,
             'data': data
         })
+
+    @staticmethod
+    def generate_messages_from_queue(queue_message):
+        for message in queue_message['data']['messages']:
+            yield {
+                'key': 'message',
+                'data': message
+            }

lib/tfw/networking/serialization.py

@@ -67,10 +67,14 @@ def _serialize_single(data):
     (serialize input if it is JSON)
     """
     if not isinstance(data, str):
-        data = json.dumps(data)
+        data = message_bytes(data)
     return _encode_if_needed(data)
 
 
+def message_bytes(message):
+    return json.dumps(message, sort_keys=True).encode()
+
+
 def _deserialize_single(data):
     """
     Try parsing input as JSON, return it as

lib/tfw/networking/server/__init__.py

@@ -3,4 +3,3 @@
 
 from .event_handler_connector import EventHandlerConnector, EventHandlerUplinkConnector, EventHandlerDownlinkConnector
 from .tfw_server import TFWServer
-# from .controller_responder import ControllerResponder  # TODO: readd once controller stuff is resolved

lib/tfw/networking/server/controller_responder.py

@@ -1,38 +0,0 @@
-# Copyright (C) 2018 Avatao.com Innovative Learning Kft.
-# All Rights Reserved. See LICENSE file for details.
-
-from tfw.networking import deserialize_all, serialize_all, ControllerConnector
-
-
-class ControllerResponder:
-    def __init__(self, fsm):
-        self.fsm = fsm
-        self.token = None
-        self.controller_connector = ControllerConnector()
-        self.controller_connector.register_callback(self.handle_controller_request)
-        self.controller_request_handlers = {
-            'solution_check': self.handle_solution_check_request,
-            'test': self.handle_test_request,
-            'token': self.handle_token_request
-        }
-
-    def handle_controller_request(self, stream, msg_parts):
-        key, data = deserialize_all(*msg_parts)
-        response = self.controller_request_handlers[key](data)
-        stream.send_multipart(serialize_all(self.token, response))
-
-    def handle_test_request(self, data):
-        # pylint: disable=unused-argument,no-self-use
-        return 'OK'
-
-    def handle_token_request(self, data):
-        if self.token is None:
-            self.token = data
-        return {'token': self.token}
-
-    def handle_solution_check_request(self, data):
-        # pylint: disable=unused-argument
-        return {
-            'solved': self.fsm.is_solved(),
-            'message': 'solved' if self.fsm.is_solved() else 'not solved'
-        }

lib/tfw/networking/server/tfw_server.py

@@ -1,10 +1,15 @@
 # Copyright (C) 2018 Avatao.com Innovative Learning Kft.
 # All Rights Reserved. See LICENSE file for details.
 
+from abc import ABC, abstractmethod
+from contextlib import suppress
+
 from tornado.web import Application
 
 from tfw.networking.event_handlers import ServerUplinkConnector
 from tfw.networking.server import EventHandlerConnector
+from tfw.networking import MessageSender
+from tfw.crypto import KeyManager, verify_message, sign_message
 from tfw.config.logs import logging
 from .zmq_websocket_proxy import ZMQWebSocketProxy
 
@@ -20,24 +25,78 @@ class TFWServer:
     def __init__(self):
         self._event_handler_connector = EventHandlerConnector()
         self._uplink_connector = ServerUplinkConnector()
+        self._auth_key = KeyManager().auth_key
 
         self.application = Application([(
-            r'/ws', ZMQWebSocketProxy,{
+            r'/ws', ZMQWebSocketProxy, {
                 'event_handler_connector': self._event_handler_connector,
-                'message_handlers': [self.handle_trigger]
+                'message_handlers': [self.handle_trigger, self.handle_recover],
-            })]
+                'frontend_message_handlers': [self.save_frontend_messages]
-        )
+            })])
+
+        self._frontend_messages = FrontendMessageStorage()
 
     def handle_trigger(self, message):
         if 'trigger' in message:
             LOG.debug('Executing handler for trigger "%s"', message.get('trigger', ''))
-            self._uplink_connector.send_to_eventhandler({
+            fsm_eh_command = {
                 'key': 'fsm',
                 'data': {
                     'command': 'trigger',
-                    'value': message.get('trigger', '')
+                    'value': message['trigger']
                 }
-            })
+            }
+            if verify_message(self._auth_key, message):
+                sign_message(self._auth_key, fsm_eh_command)
+            self._uplink_connector.send_to_eventhandler(fsm_eh_command)
+
+    def handle_recover(self, message):
+        if message['key'] == 'recover':
+            self._frontend_messages.replay_messages(self._uplink_connector)
+            self._frontend_messages.clear()
+
+    def save_frontend_messages(self, message):
+        self._frontend_messages.save_message(message)
 
     def listen(self, port):
         self.application.listen(port)
+
+
+class MessageStorage(ABC):
+    def __init__(self):
+        self.saved_messages = []
+
+    def save_message(self, message):
+        with suppress(KeyError, AttributeError):
+            if self.filter_message(message):
+                self.saved_messages.extend(self.transform_message(message))
+
+    @abstractmethod
+    def filter_message(self, message):
+        raise NotImplementedError
+
+    def transform_message(self, message):   # pylint: disable=no-self-use
+        yield message
+
+    def clear(self):
+        self.saved_messages.clear()
+
+
+class FrontendMessageStorage(MessageStorage):
+    def filter_message(self, message):
+        key = message['key']
+        command = message.get('data', {}).get('command')
+        return (
+            key in ('message', 'dashboard', 'queueMessages')
+            or key == 'ide' and command in ('select', 'read')
+        )
+
+    def transform_message(self, message):
+        if message['key'] == 'queueMessages':
+            yield from MessageSender.generate_messages_from_queue(message)
+        else:
+            yield message
+
+    def replay_messages(self, connector):
+        for message in self.saved_messages:
+            connector.send(message)

lib/tfw/networking/server/zmq_websocket_proxy.py

@@ -12,11 +12,15 @@ LOG = logging.getLogger(__name__)
 
 
 class ZMQWebSocketProxy(WebSocketHandler):
+    # pylint: disable=abstract-method
     instances = set()
 
     def initialize(self, **kwargs): # pylint: disable=arguments-differ
         self._event_handler_connector = kwargs['event_handler_connector']
+
         self._message_handlers = kwargs.get('message_handlers', [])
+        self._frontend_message_handlers = kwargs.get('frontend_message_handlers', [])
+        self._eventhandler_message_handlers = kwargs.get('eventhandler_message_handlers', [])
         self._proxy_filters = kwargs.get('proxy_filters', [])
 
         self.proxy_eventhandler_to_websocket = TFWProxy(
@@ -28,10 +32,18 @@ class ZMQWebSocketProxy(WebSocketHandler):
             self.send_eventhandler_message
         )
 
-        proxies = (self.proxy_eventhandler_to_websocket, self.proxy_websocket_to_eventhandler)
+        self.subscribe_proxy_callbacks()
-        for proxy in proxies:
+
-            proxy.proxy_filters.subscribe_callbacks(*self._proxy_filters)
+    def subscribe_proxy_callbacks(self):
-            proxy.proxy_callbacks.subscribe_callbacks(*self._message_handlers)
+        self.proxy_websocket_to_eventhandler.subscribe_proxy_callbacks_and_filters(
+            self._eventhandler_message_handlers + self._message_handlers,
+            self._proxy_filters
+        )
+
+        self.proxy_eventhandler_to_websocket.subscribe_proxy_callbacks_and_filters(
+            self._frontend_message_handlers + self._message_handlers,
+            self._proxy_filters
+        )
 
     def prepare(self):
         ZMQWebSocketProxy.instances.add(self)
@@ -72,6 +84,7 @@ class ZMQWebSocketProxy(WebSocketHandler):
 
 
 class TFWProxy:
+    # pylint: disable=protected-access
     def __init__(self, to_source, to_destination):
         self.to_source = to_source
         self.to_destination = to_destination
@@ -119,3 +132,7 @@ class TFWProxy:
         LOG.debug('Broadcasting message: %s', message)
         self.to_source(message)
         self.to_destination(message)
+
+    def subscribe_proxy_callbacks_and_filters(self, proxy_callbacks, proxy_filters):
+        self.proxy_callbacks.subscribe_callbacks(*proxy_callbacks)
+        self.proxy_filters.subscribe_callbacks(*proxy_filters)

lib/tfw/yaml_fsm.py

@@ -40,19 +40,19 @@ class YamlFSM(FSMBase):
     def subscribe_and_remove_predicates(self, json_obj):
         if 'predicates' in json_obj:
             for predicate in json_obj['predicates']:
-                    self.subscribe_predicate(
+                self.subscribe_predicate(
-                        json_obj['trigger'],
+                    json_obj['trigger'],
-                        partial(
+                    partial(
-                            command_statuscode_is_zero,
+                        command_statuscode_is_zero,
-                            predicate
+                        predicate
                         )
-                    )
+                )
-                
+
         with suppress(KeyError):
             json_obj.pop('predicates')
 
 
-def run_command_async(command, event):
+def run_command_async(command, _):
     Popen(command, shell=True)
 
 
@@ -62,7 +62,7 @@ def command_statuscode_is_zero(command):
 
 class ConfigParser:
     def __init__(self, config_file, jinja2_variables):
-        self.read_variables = singledispatch(self.read_variables)
+        self.read_variables = singledispatch(self._read_variables)
         self.read_variables.register(dict, self._read_variables_dict)
         self.read_variables.register(str, self._read_variables_str)
 
@@ -82,16 +82,14 @@ class ConfigParser:
             return ifile.read()
 
     @staticmethod
-    def read_variables(variables):
+    def _read_variables(variables):
         raise TypeError(f'Invalid variables type {type(variables)}')
 
     @staticmethod
     def _read_variables_str(variables):
-        if isinstance(variables, str):
+        with open(variables, 'r') as ifile:
-            with open(variables, 'r') as ifile:
+            return yaml.safe_load(ifile)
-                return yaml.safe_load(ifile)
 
     @staticmethod
     def _read_variables_dict(variables):
-            return variables
+        return variables
-

requirements.txt

@@ -5,3 +5,4 @@ terminado==0.8.1
 watchdog==0.8.3
 PyYAML==3.12
 Jinja2==2.10
+cryptography==2.2.2