[Update] 移除app依赖

coco/app.py

@@ -50,6 +50,7 @@ class Coco:
         'HEARTBEAT_INTERVAL': 5,
         'MAX_CONNECTIONS': 500,
         'ADMINS': '',
+        'WORKERS': 4,
         'COMMAND_STORAGE': {'TYPE': 'server'},   # server
         'REPLAY_STORAGE': {'TYPE': 'server'},
     }
@@ -92,13 +93,13 @@ class Coco:
     @property
     def httpd(self):
         if self._httpd is None:
-            self._httpd = HttpServer(self)
+            self._httpd = HttpServer()
         return self._httpd
 
     @property
     def task_handler(self):
         if self._task_handler is None:
-            self._task_handler = TaskHandler(self)
+            self._task_handler = TaskHandler()
         return self._task_handler
 
     def make_logger(self):
@@ -116,11 +117,10 @@ class Coco:
         self.command_recorder_class = get_command_recorder_class(self.config)
 
     def new_command_recorder(self):
-        recorder = self.command_recorder_class(self)
-        return recorder
+        return self.command_recorder_class()
 
     def new_replay_recorder(self):
-        return self.replay_recorder_class(self)
+        return self.replay_recorder_class()
 
     def bootstrap(self):
         self.make_logger()

coco/connection.py

 # -*- coding: utf-8 -*-
 #
 
-import weakref
 import os
 import socket
 
 import paramiko
 from paramiko.ssh_exception import SSHException
 
+from .ctx import app_service
 from .utils import get_logger, get_private_key_fingerprint
 
 logger = get_logger(__file__)
@@ -15,20 +15,13 @@ TIMEOUT = 10
 
 
 class SSHConnection:
-    def __init__(self, app):
-        self._app = weakref.ref(app)
-
-    @property
-    def app(self):
-        return self._app()
-
     def get_system_user_auth(self, system_user):
         """
         获取系统用户的认证信息，密码或秘钥
         :return: system user have full info
         """
         password, private_key = \
-            self.app.service.get_system_user_auth_info(system_user)
+            app_service.get_system_user_auth_info(system_user)
         system_user.password = password
         system_user.private_key = private_key
 
@@ -97,7 +90,7 @@ class SSHConnection:
 
     def get_proxy_sock(self, asset):
         sock = None
-        domain = self.app.service.get_domain_detail_with_gateway(
+        domain = app_service.get_domain_detail_with_gateway(
             asset.domain
         )
         if not domain.has_ssh_gateway():

coco/ctx.py

 # -*- coding: utf-8 -*-
 #
 
-current_app = []
-current_service = []
+from werkzeug.local import LocalProxy
+from functools import partial
+
+
+stack = {}
+
+
+def _find(name):
+    if stack.get(name):
+        return stack[name]
+    else:
+        raise ValueError("Not found in stack: {}".format(name))
+
+
+current_app = LocalProxy(partial(_find, 'app'))
+app_service = LocalProxy(partial(_find, 'service'))
+
+# current_app = []
+# current_service = []

coco/httpd.py

@@ -11,6 +11,7 @@ from flask import Flask, request, current_app, redirect
 from .models import Request, Client, WSProxy
 from .proxy import ProxyServer
 from .utils import get_logger
+from .ctx import current_app, app_service
 
 BASE_DIR = os.path.dirname(os.path.dirname(__file__))
 
@@ -21,11 +22,6 @@ class BaseNamespace(Namespace):
     clients = None
     current_user = None
 
-    @property
-    def app(self):
-        app = current_app.config['coco']
-        return app
-
     def on_connect(self):
         self.current_user = self.get_current_user()
         if self.current_user is None:
@@ -38,9 +34,9 @@ class BaseNamespace(Namespace):
         token = request.headers.get("Authorization")
         user = None
         if session_id and csrf_token:
-            user = self.app.service.check_user_cookie(session_id, csrf_token)
+            user = app_service.check_user_cookie(session_id, csrf_token)
         if token:
-            user = self.app.service.check_user_with_token(token)
+            user = app_service.check_user_with_token(token)
         return user
 
     def close(self):
@@ -143,8 +139,8 @@ class ProxyNamespace(BaseNamespace):
             # self.on_connect()
             return
 
-        asset = self.app.service.get_asset(asset_id)
-        system_user = self.app.service.get_system_user(user_id)
+        asset = app_service.get_asset(asset_id)
+        system_user = app_service.get_system_user(user_id)
 
         if not asset or not system_user:
             self.on_connect()
@@ -152,7 +148,7 @@ class ProxyNamespace(BaseNamespace):
 
         child, parent = socket.socketpair()
         client = Client(parent, room["request"])
-        forwarder = ProxyServer(self.app, client)
+        forwarder = ProxyServer(client)
         room["client"] = client
         room["forwarder"] = forwarder
         room["proxy"] = WSProxy(self, child, room["id"])
@@ -187,7 +183,7 @@ class ProxyNamespace(BaseNamespace):
             self.emit('disconnect')
             return None
 
-        info = self.app.service.get_token_asset(token)
+        info = app_service.get_token_asset(token)
         logger.debug(info)
         if not info:
             logger.debug("Token info is None")
@@ -197,7 +193,7 @@ class ProxyNamespace(BaseNamespace):
             return None
 
         user_id = info.get('user', None)
-        self.current_user = self.app.service.get_user_profile(user_id)
+        self.current_user = app_service.get_user_profile(user_id)
         room["request"].user = self.current_user
         logger.debug(self.current_user)
         self.on_host({
@@ -250,20 +246,20 @@ class ProxyNamespace(BaseNamespace):
 class HttpServer:
     # prepare may be rewrite it
     config = {
-        'SECRET_KEY': '',
+        'SECRET_KEY': 'someWOrkSD20KMS9330)&#',
         'coco': None,
         'LOGIN_URL': '/login'
     }
     init_kwargs = dict(
+        # async_mode="gevent",
         async_mode="threading",
         ping_timeout=20,
         ping_interval=10
     )
 
-    def __init__(self, coco):
-        config = coco.config
+    def __init__(self):
+        config = {k: v for k, v in current_app.config.items()}
         config.update(self.config)
-        config['coco'] = coco
         self.flask_app = Flask(__name__, template_folder='dist')
         self.flask_app.config.update(config)
         self.socket_io = SocketIO()

coco/interactive.py

@@ -4,16 +4,16 @@
 
 import socket
 import threading
-import weakref
 import os
 
 from jms.models import Asset, AssetGroup
 
 from . import char
 from .utils import wrap_with_line_feed as wr, wrap_with_title as title, \
-    wrap_with_primary as primary, wrap_with_warning as warning, \
-    is_obj_attr_has, is_obj_attr_eq, sort_assets, TtyIOParser, \
-    ugettext as _, get_logger
+    wrap_with_warning as warning, is_obj_attr_has, \
+    is_obj_attr_eq, sort_assets, TtyIOParser, \
+    ugettext as _, get_logger, net_input
+from .ctx import current_app, app_service
 from .proxy import ProxyServer
 
 logger = get_logger(__file__)
@@ -22,19 +22,14 @@ logger = get_logger(__file__)
 class InteractiveServer:
     _sentinel = object()
 
-    def __init__(self, app, client):
-        self._app = weakref.ref(app)
+    def __init__(self, client):
         self.client = client
         self.request = client.request
         self.assets = None
         self._search_result = None
         self.asset_groups = None
         self.get_user_assets_async()
-        self.get_user_asset_groups_async()
-
-    @property
-    def app(self):
-        return self._app()
+        self.get_user_nodes_async()
 
     @property
     def search_result(self):
@@ -50,7 +45,7 @@ class InteractiveServer:
 
     def display_banner(self):
         self.client.send(char.CLEAR_CHAR)
-        logo_path = os.path.join(self.app.root_path, "logo.txt")
+        logo_path = os.path.join(current_app.root_path, "logo.txt")
         if os.path.isfile(logo_path):
             with open(logo_path, 'rb') as f:
                 for i in f:
@@ -71,63 +66,6 @@ class InteractiveServer:
         )
         self.client.send(banner)
 
-    def get_option(self, prompt='Opt> '):
-        """实现了一个ssh input, 提示用户输入, 获取并返回
-
-        :return user input string
-        """
-        # Todo: 实现自动hostname或IP补全
-        input_data = []
-        parser = TtyIOParser()
-        self.client.send(wr(prompt, before=1, after=0))
-
-        while True:
-            data = self.client.recv(10)
-            if len(data) == 0:
-                self.app.remove_client(self.client)
-                break
-            # Client input backspace
-            if data in char.BACKSPACE_CHAR:
-                # If input words less than 0, should send 'BELL'
-                if len(input_data) > 0:
-                    data = char.BACKSPACE_CHAR[data]
-                    input_data.pop()
-                else:
-                    data = char.BELL_CHAR
-                self.client.send(data)
-                continue
-
-            if data.startswith(b'\x03'):
-                # Ctrl-C
-                self.client.send(b'^C\r\nOpt> ')
-                input_data = []
-                continue
-            elif data.startswith(b'\x04'):
-                # Ctrl-D
-                return 'q'
-
-            # Todo: Move x1b to char
-            if data.startswith(b'\x1b') or data in char.UNSUPPORTED_CHAR:
-                self.client.send(b'')
-                continue
-
-            # handle shell expect
-            multi_char_with_enter = False
-            if len(data) > 1 and data[-1] in char.ENTER_CHAR_ORDER:
-                self.client.send(data)
-                input_data.append(data[:-1])
-                multi_char_with_enter = True
-
-            # If user type ENTER we should get user input
-            if data in char.ENTER_CHAR or multi_char_with_enter:
-                self.client.send(wr(b'', after=2))
-                option = parser.parse_input(input_data)
-                del input_data[:]
-                return option.strip()
-            else:
-                self.client.send(data)
-                input_data.append(data)
-
     def dispatch(self, opt):
         if opt is None:
             return self._sentinel
@@ -179,7 +117,7 @@ class InteractiveServer:
 
     def display_asset_groups(self):
         if self.asset_groups is None:
-            self.get_user_asset_groups()
+            self.get_user_nodes()
 
         if len(self.asset_groups) == 0:
             self.client.send(warning(_("无")))
@@ -208,7 +146,7 @@ class InteractiveServer:
         self.display_search_result()
 
     def display_search_result(self):
-        self.search_result = sort_assets(self.search_result, self.app.config["ASSET_LIST_SORT_BY"])
+        self.search_result = sort_assets(self.search_result, current_app.config["ASSET_LIST_SORT_BY"])
         fake_asset = Asset(hostname=_("Hostname"), ip=_("IP"), _system_users_name_list=_("LoginAs"),
                            comment=_("Comment"))
         id_max_length = max(len(str(len(self.search_result))), 3)
@@ -231,11 +169,11 @@ class InteractiveServer:
         self.search_assets(q)
         self.display_search_result()
 
-    def get_user_asset_groups(self):
-        self.asset_groups = self.app.service.get_user_asset_groups(self.client.user)
+    def get_user_nodes(self):
+        self.asset_groups = app_service.get_user_asset_groups(self.client.user)
 
-    def get_user_asset_groups_async(self):
-        thread = threading.Thread(target=self.get_user_asset_groups)
+    def get_user_nodes_async(self):
+        thread = threading.Thread(target=self.get_user_nodes)
         thread.start()
 
     @staticmethod
@@ -248,7 +186,7 @@ class InteractiveServer:
         return assets
 
     def get_user_assets(self):
-        self.assets = self.app.service.get_user_assets(self.client.user)
+        self.assets = app_service.get_user_assets(self.client.user)
         logger.debug("Get user {} assets total: {}".format(self.client.user, len(self.assets)))
 
     def get_user_assets_async(self):
@@ -267,7 +205,7 @@ class InteractiveServer:
         while True:
             self.client.send(wr(_("选择一个登陆: "), after=1))
             self.display_system_users(system_users)
-            opt = self.get_option("ID> ")
+            opt = net_input(self.client, prompt="ID> ")
             if opt.isdigit() and len(system_users) > int(opt):
                 return system_users[int(opt)]
             elif opt in ['q', 'Q']:
@@ -297,14 +235,14 @@ class InteractiveServer:
         if system_user is None:
             self.client.send(_("没有系统用户"))
             return
-        forwarder = ProxyServer(self.app, self.client)
+        forwarder = ProxyServer(self.client)
         forwarder.proxy(asset, system_user)
 
     def interact(self):
         self.display_banner()
         while True:
             try:
-                opt = self.get_option()
+                opt = net_input(self.client)
                 rv = self.dispatch(opt)
                 if rv is self._sentinel:
                     break
@@ -318,7 +256,7 @@ class InteractiveServer:
         thread.start()
 
     def close(self):
-        self.app.remove_client(self.client)
+        current_app.remove_client(self.client)
 
     # def __del__(self):
     #     print("GC: Interactive class been gc")

coco/interface.py

@@ -7,6 +7,7 @@ import threading
 import weakref
 
 from .utils import get_logger
+from .ctx import current_app, app_service
 
 logger = get_logger(__file__)
 
@@ -19,18 +20,13 @@ class SSHInterface(paramiko.ServerInterface):
     https://github.com/paramiko/paramiko/blob/master/demos/demo_server.py
     """
 
-    def __init__(self, app, request):
-        self._app = weakref.ref(app)
+    def __init__(self, request):
         self._request = weakref.ref(request)
         self.event = threading.Event()
         self.auth_valid = False
         self.otp_auth = False
         self.info = None
 
-    @property
-    def app(self):
-        return self._app()
-
     @property
     def request(self):
         return self._request()
@@ -55,7 +51,7 @@ class SSHInterface(paramiko.ServerInterface):
         if not seed:
             return paramiko.AUTH_FAILED
 
-        is_valid = self.app.service.authenticate_otp(seed, otp_code)
+        is_valid = app_service.authenticate_otp(seed, otp_code)
         if is_valid:
             return paramiko.AUTH_SUCCESSFUL
         return paramiko.AUTH_FAILED
@@ -67,9 +63,9 @@ class SSHInterface(paramiko.ServerInterface):
         supported = []
         if self.otp_auth:
             return 'keyboard-interactive'
-        if self.app.config["PASSWORD_AUTH"]:
+        if current_app.config["PASSWORD_AUTH"]:
             supported.append("password")
-        if self.app.config["PUBLIC_KEY_AUTH"]:
+        if current_app.config["PUBLIC_KEY_AUTH"]:
             supported.append("publickey")
         return ",".join(supported)
 
@@ -100,7 +96,7 @@ class SSHInterface(paramiko.ServerInterface):
             return paramiko.AUTH_SUCCESSFUL
 
     def validate_auth(self, username, password="", public_key=""):
-        info = self.app.service.authenticate(
+        info = app_service.authenticate(
             username, password=password, public_key=public_key,
             remote_addr=self.request.remote_ip
         )

coco/models.py

@@ -251,6 +251,7 @@ class WSProxy:
             if len(data) == 0:
                 self.close()
             data = data.decode(errors="ignore")
+            print("Send data: {}".format(data))
             self.ws.emit("data", {'data': data, 'room': self.room_id},
                          room=self.room_id)
             if len(data) == BUF_SIZE:

coco/proxy.py

@@ -4,13 +4,13 @@
 
 import threading
 import time
-import weakref
 
 from paramiko.ssh_exception import SSHException
 
 from .session import Session
 from .models import Server
 from .connection import SSHConnection
+from .ctx import current_app, app_service
 from .utils import wrap_with_line_feed as wr, wrap_with_warning as warning, \
      get_logger, net_input
 
@@ -21,24 +21,19 @@ BUF_SIZE = 4096
 
 
 class ProxyServer:
-    def __init__(self, app, client):
-        self._app = weakref.ref(app)
+    def __init__(self, client):
         self.client = client
         self.server = None
         self.connecting = True
         self.stop_event = threading.Event()
 
-    @property
-    def app(self):
-        return self._app()
-
     def get_system_user_auth(self, system_user):
         """
         获取系统用户的认证信息，密码或秘钥
         :return: system user have full info
         """
         password, private_key = \
-            self.app.service.get_system_user_auth_info(system_user)
+            app_service.get_system_user_auth_info(system_user)
         if not password and not private_key:
             prompt = "{}'s password: ".format(system_user.username)
             password = net_input(self.client, prompt=prompt, sensitive=True)
@@ -51,26 +46,26 @@ class ProxyServer:
         self.server = self.get_server_conn(asset, system_user)
         if self.server is None:
             return
-        command_recorder = self.app.new_command_recorder()
-        replay_recorder = self.app.new_replay_recorder()
+        command_recorder = current_app.new_command_recorder()
+        replay_recorder = current_app.new_replay_recorder()
         session = Session(
             self.client, self.server,
             command_recorder=command_recorder,
             replay_recorder=replay_recorder,
         )
-        self.app.add_session(session)
+        current_app.add_session(session)
         self.watch_win_size_change_async()
         session.bridge()
         self.stop_event.set()
         self.end_watch_win_size_change()
-        self.app.remove_session(session)
+        current_app.remove_session(session)
 
     def validate_permission(self, asset, system_user):
         """
         验证用户是否有连接改资产的权限
         :return: True or False
         """
-        return self.app.service.validate_user_asset_permission(
+        return app_service.validate_user_asset_permission(
             self.client.user.id, asset.id, system_user.id
         )
 
@@ -90,11 +85,11 @@ class ProxyServer:
         pass
 
     def get_ssh_server_conn(self, asset, system_user):
-        ssh = SSHConnection(self.app)
         request = self.client.request
         term = request.meta.get('term', 'xterm')
         width = request.meta.get('width', 80)
         height = request.meta.get('height', 24)
+        ssh = SSHConnection()
         chan, msg = ssh.get_channel(asset, system_user, term=term,
                                     width=width, height=height)
         if not chan:

coco/recorder.py

@@ -13,14 +13,14 @@ import jms_storage
 
 from .utils import get_logger, Singleton
 from .alignment import MemoryQueue
+from .ctx import current_app, app_service
 
 logger = get_logger(__file__)
 BUF_SIZE = 1024
 
 
 class ReplayRecorder(metaclass=abc.ABCMeta):
-    def __init__(self, app, session=None):
-        self.app = app
+    def __init__(self, session=None):
         self.session = session
 
     @abc.abstractmethod
@@ -47,8 +47,7 @@ class ReplayRecorder(metaclass=abc.ABCMeta):
 
 
 class CommandRecorder:
-    def __init__(self, app, session=None):
-        self.app = app
+    def __init__(self, session=None):
         self.session = session
 
     def record(self, data):
@@ -78,8 +77,8 @@ class ServerReplayRecorder(ReplayRecorder):
     time_start = None
     storage = None
 
-    def __init__(self, app):
-        super().__init__(app)
+    def __init__(self):
+        super().__init__()
         self.file = None
         self.file_path = None
 
@@ -100,8 +99,8 @@ class ServerReplayRecorder(ReplayRecorder):
 
     def session_start(self, session_id):
         self.time_start = time.time()
-        filename = session_id+'.replay.gz'
-        self.file_path = os.path.join(self.app.config['LOG_DIR'], filename)
+        filename = session_id + '.replay.gz'
+        self.file_path = os.path.join(current_app.config['LOG_DIR'], filename)
         self.file = gzip.open(self.file_path, 'at')
         self.file.write('{')
 
@@ -114,11 +113,11 @@ class ServerReplayRecorder(ReplayRecorder):
             logger.error("Failed to push {}'s {}".format(session_id, "record"))
 
     def upload_replay(self, session_id):
-        configs = self.app.service.load_config_from_server()
+        configs = app_service.load_config_from_server()
         logger.debug("upload_replay print config: {}".format(configs))
         self.storage = jms_storage.init(configs["REPLAY_STORAGE"])
         if not self.storage:
-            self.storage = jms_storage.jms(self.app.service)
+            self.storage = jms_storage.jms(app_service)
         if self.push_file(3, session_id):
             os.unlink(self.file_path)
             return True
@@ -137,7 +136,7 @@ class ServerReplayRecorder(ReplayRecorder):
             else:
                 msg = "Failed push session {}'s replay log to storage".format(session_id)
                 logger.error(msg)
-                self.storage = jms_storage.jms(self.app.service)
+                self.storage = jms_storage.jms(app_service)
                 return self.push_file(3, session_id)
 
         if self.push_to_storage(session_id):
@@ -153,7 +152,7 @@ class ServerReplayRecorder(ReplayRecorder):
             logger.error("Failed finished session {}'s replay".format(session_id))
             return False
 
-        if self.app.service.finish_replay(session_id):
+        if app_service.finish_replay(session_id):
             logger.info("Success finish session {}'s replay ".format(session_id))
             return True
         else:
@@ -166,8 +165,8 @@ class ServerCommandRecorder(CommandRecorder, metaclass=Singleton):
     timeout = 5
     no = 0
 
-    def __init__(self, app):
-        super().__init__(app)
+    def __init__(self):
+        super().__init__()
         self.queue = MemoryQueue()
         self.stop_evt = threading.Event()
         self.push_to_server_async()
@@ -190,7 +189,7 @@ class ServerCommandRecorder(CommandRecorder, metaclass=Singleton):
                 if not data_set:
                     continue
                 logger.debug("Send {} commands to server".format(len(data_set)))
-                ok = self.app.service.push_session_command(data_set)
+                ok = app_service.push_session_command(data_set)
                 if not ok:
                     self.queue.mput(data_set)
 
@@ -214,13 +213,15 @@ class ESCommandRecorder(CommandRecorder, metaclass=Singleton):
     no = 0
     default_hosts = ["http://localhost"]
 
-    def __init__(self, app):
-        super().__init__(app)
+    def __init__(self):
+        super().__init__()
         self.queue = MemoryQueue()
         self.stop_evt = threading.Event()
         self.push_to_es_async()
         self.__class__.no += 1
-        self.store = jms_storage.ESStore(app.config["COMMAND_STORAGE"].get("HOSTS", self.default_hosts))
+        self.store = jms_storage.ESStore(
+            current_app.config["COMMAND_STORAGE"].get("HOSTS", self.default_hosts)
+        )
         if not self.store.ping():
             raise AssertionError("ESCommand storage init error")
 

coco/sshd.py

@@ -5,16 +5,14 @@
 import os
 import socket
 import threading
-import random
 import paramiko
-from multiprocessing.reduction import recv_handle, send_handle
-from multiprocessing import Process, Pipe
 
-from .utils import ssh_key_gen, get_logger, get_app
+from .utils import ssh_key_gen, get_logger
 from .interface import SSHInterface
 from .interactive import InteractiveServer
 from .models import Client, Request
 from .sftp import SFTPServer
+from .ctx import current_app
 
 logger = get_logger(__file__)
 BACKLOG = 5
@@ -24,64 +22,40 @@ class SSHServer:
 
     def __init__(self):
         self.stop_evt = threading.Event()
-        self.host_key_path = os.path.join(self.app.root_path, 'keys', 'host_rsa_key')
         self.workers = []
         self.pipe = None
 
-    @property
-    def app(self):
-        return get_app()
-
     @property
     def host_key(self):
-        if not os.path.isfile(self.host_key_path):
-            self.gen_host_key()
-        return paramiko.RSAKey(filename=self.host_key_path)
+        host_key_path = os.path.join(current_app.root_path, 'keys', 'host_rsa_key')
+        if not os.path.isfile(host_key_path):
+            self.gen_host_key(host_key_path)
+        return paramiko.RSAKey(filename=host_key_path)
 
-    def gen_host_key(self):
+    @staticmethod
+    def gen_host_key(key_path):
         ssh_key, _ = ssh_key_gen()
-        with open(self.host_key_path, 'w') as f:
+        with open(key_path, 'w') as f:
             f.write(ssh_key)
 
-    def start_worker(self, in_pipe, out_pipe):
-        print("APP: {}".format(self.app))
-        print("APP sessions: {}".format(self.app))
-        out_pipe.close()
-        while not self.stop_evt.is_set():
-            fd = recv_handle(in_pipe)
-            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM, fileno=fd)
-            addr = sock.getpeername()
-            thread = threading.Thread(target=self.handle_connection, args=(sock, addr))
-            thread.daemon = True
-            thread.start()
-
-    def start_server(self, in_pipe, out_pipe, workers):
-        in_pipe.close()
-        host = self.app.config["BIND_HOST"]
-        port = self.app.config["SSHD_PORT"]
+    def run(self):
+        host = current_app.config["BIND_HOST"]
+        port = current_app.config["SSHD_PORT"]
         print('Starting ssh server at {}:{}'.format(host, port))
         sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
         sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
         sock.bind((host, port))
         sock.listen(BACKLOG)
-
         while not self.stop_evt.is_set():
-            client, addr = sock.accept()
-            logger.info("Get ssh request from {}".format(addr))
-            send_handle(out_pipe, client.fileno(), random.choice(workers).pid)
-            client.close()
-
-    def run(self):
-        in_pipe, out_pipe = Pipe()
-        self.pipe = (in_pipe, out_pipe)
-        workers = []
-        for i in range(4):
-            worker = Process(target=self.start_worker, args=(in_pipe, out_pipe))
-            worker.start()
-            workers.append(worker)
-        self.start_server(in_pipe, out_pipe, workers)
-        in_pipe.close()
-        out_pipe.close()
+            try:
+                client, addr = sock.accept()
+                logger.info("Get ssh request from {}: {}".format(*addr))
+                thread = threading.Thread(target=self.handle_connection,
+                                          args=(client, addr))
+                thread.daemon = True
+                thread.start()
+            except IndexError as e:
+                logger.error("Start SSH server error: {}".format(e))
 
     def handle_connection(self, sock, addr):
         transport = paramiko.Transport(sock, gss_kex=False)
@@ -95,7 +69,7 @@ class SSHServer:
             'sftp', paramiko.SFTPServer, SFTPServer
         )
         request = Request(addr)
-        server = SSHInterface(self.app, request)
+        server = SSHInterface(request)
         try:
             transport.start_server(server=server)
         except paramiko.SSHException:
@@ -126,7 +100,7 @@ class SSHServer:
 
     def handle_chan(self, chan, request):
         client = Client(chan, request)
-        self.app.add_client(client)
+        current_app.add_client(client)
         self.dispatch(client)
 
     def dispatch(self, client):
@@ -134,7 +108,7 @@ class SSHServer:
         request_type = set(client.request.type)
         if supported & request_type:
             logger.info("Request type `pty`, dispatch to interactive mode")
-            InteractiveServer(self.app, client).interact()
+            InteractiveServer(client).interact()
         elif 'subsystem' in request_type:
             pass
         else:

coco/tasks.py

 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
 #
-import weakref
 
+from .ctx import current_app, app_service
 from .utils import get_logger
 
 logger = get_logger(__file__)
 
 
 class TaskHandler:
+    routes = None
 
-    def __init__(self, app):
-        self._app = weakref.ref(app)
+    def init(self):
+        self.routes = {
+            'kill_session': self.handle_kill_session
+        }
 
-    @property
-    def app(self):
-        return self._app()
-
-    def handle_kill_session(self, task):
+    @staticmethod
+    def handle_kill_session(task):
         logger.info("Handle kill session task: {}".format(task.args))
         session_id = task.args
         session = None
-        for s in self.app.sessions:
+        for s in current_app.sessions:
             if s.id == session_id:
                 session = s
                 break
 
         if session:
             session.terminate()
-        self.app.service.finish_task(task.id)
+        app_service.finish_task(task.id)
 
     def handle(self, task):
-        if task.name == "kill_session":
-            self.handle_kill_session(task)
-        else:
-            logger.error("No handler for this task: {}".format(task.name))
+        func = self.routes.get(task.name)
+        return func(task)

coco/utils.py

@@ -15,7 +15,7 @@ import paramiko
 import pyte
 
 from . import char
-from .ctx import current_app, current_service
+from .ctx import stack
 
 BASE_DIR = os.path.abspath(os.path.dirname(os.path.dirname(__file__)))
 
@@ -370,25 +370,18 @@ def net_input(client, prompt='Opt> ', sensitive=False):
 
 
 def register_app(app):
-    current_app.insert(0, app)
+    stack['app'] = app
 
 
 def register_service(service):
-    current_service.insert(0, service)
+    stack['service'] = service
 
 
 def get_app():
-    if current_app:
-        return current_app[0]
+    if stack.get("app"):
+        return stack["app"]
     else:
-        raise ValueError("App not found")
-
-
-def get_service():
-    if current_service:
-        return current_app[0]
-    else:
-        raise ValueError("Service not found")
+        return ValueError("No app found")
 
 
 ugettext = _gettext()