前言

最近工作中需要开发前端操作远程虚拟机的功能，简称 webshell。基于当前的技术栈为 react+django，调研了一会发现大部分的后端实现都是 django+channels 来实现 websocket 服务。

大致看了下觉得这不够有趣，翻了翻 django 的官方文档发现 django 原生是不支持 websocket 的，但 django3 之后支持了 asgi 协议可以自己实现 websocket 服务。

于是选定 gunicorn+uvicorn+asgi+websocket+django3.2+paramiko 来实现 webshell。

实现 websocket 服务

使用 django 自带的脚手架生成的项目会自动生成 asgi.py 和 wsgi.py 两个文件，普通应用大部分用的都是 wsgi.py 配合 nginx 部署线上服务。

这次主要使用 asgi.py 实现 websocket 服务的思路大致网上搜一下就能找到，主要就是实现
connect/send/receive/disconnect 这个几个动作的处理方法。

这里 how to add websockets to a django app without extra dependencies(
https://jaydenwindle.com/writing/django-websockets-zero-dependencies/) 就是一个很好的实例，但过于简单……

思路

  
  # asgi.py 
  import os
  
  from django.core.asgi import get_asgi_application
  from websocket_app.websocket import websocket_application
  
  os.environ.setdefault('django_settings_module', 'websocket_app.settings')
  
  django_application = get_asgi_application()
  
  
  async def application(scope, receive, send):
      if scope['type'] == 'http':
          await django_application(scope, receive, send)
      elif scope['type'] == 'websocket':
          await websocket_application(scope, receive, send)
      else:
          raise notimplementederror(f"unknown scope type {scope['type']}")
  
  
  # websocket.py
  async def websocket_application(scope, receive, send):
      pass

  
  # websocket.py
  async def websocket_application(scope, receive, send):
      while true:
          event = await receive()
  
          if event['type'] == 'websocket.connect':
              await send({
                  'type': 'websocket.accept'
              })
  
          if event['type'] == 'websocket.disconnect':
              break
  
          if event['type'] == 'websocket.receive':
              if event['text'] == 'ping':
                  await send({
                      'type': 'websocket.send',
                      'text': 'pong!'
                  })

实现

上面的代码提供了思路，比较完整的可以参考这里 websockets-in-django-3-1 (
https://aliashkevich.com/websockets-in-django-3-1/) 基本可以复用了。

其中最核心的实现部分我放下面：

  
  class websocket:
      def __init__(self, scope, receive, send):
          self._scope = scope
          self._receive = receive
          self._send = send
          self._client_state = state.connecting
          self._app_state = state.connecting
  
      @property
      def headers(self):
          return headers(self._scope)
  
      @property
      def scheme(self):
          return self._scope["scheme"]
  
      @property
      def path(self):
          return self._scope["path"]
  
      @property
      def query_params(self):
          return queryparams(self._scope["query_string"].decode())
  
      @property
      def query_string(self) -> str:
          return self._scope["query_string"]
  
      @property
      def scope(self):
          return self._scope
  
      async def accept(self, subprotocol: str = none):
          """accept connection.
          :param subprotocol: the subprotocol the server wishes to accept.
          :type subprotocol: str, optional
          """
          if self._client_state == state.connecting:
              await self.receive()
          await self.send({"type": sendevent.accept, "subprotocol": subprotocol})
  
      async def close(self, code: int = 1000):
          await self.send({"type": sendevent.close, "code": code})
  
      async def send(self, message: t.mapping):
          if self._app_state == state.disconnected:
              raise runtimeerror("websocket is disconnected.")
  
          if self._app_state == state.connecting:
              assert message["type"] in {sendevent.accept, sendevent.close}, (
                      'could not write event "%s" into socket in connecting state.'
                      % message["type"]
              )
              if message["type"] == sendevent.close:
                  self._app_state = state.disconnected
              else:
                  self._app_state = state.connected
  
          elif self._app_state == state.connected:
              assert message["type"] in {sendevent.send, sendevent.close}, (
                      'connected socket can send "%s" and "%s" events, not "%s"'
                      % (sendevent.send, sendevent.close, message["type"])
              )
              if message["type"] == sendevent.close:
                  self._app_state = state.disconnected
  
          await self._send(message)
  
      async def receive(self):
          if self._client_state == state.disconnected:
              raise runtimeerror("websocket is disconnected.")
  
          message = await self._receive()
  
          if self._client_state == state.connecting:
              assert message["type"] == receiveevent.connect, (
                      'websocket is in connecting state but received "%s" event'
                      % message["type"]
              )
              self._client_state = state.connected
  
          elif self._client_state == state.connected:
              assert message["type"] in {receiveevent.receive, receiveevent.disconnect}, (
                      'websocket is connected but received invalid event "%s".'
                      % message["type"]
              )
              if message["type"] == receiveevent.disconnect:
                  self._client_state = state.disconnected
  
          return message

缝合怪

做为合格的代码搬运工，为了提高搬运效率还是要造点轮子填点坑的，如何将上面的 websocket 类与 paramiko 结合起来，实现从前端接受字符传递给远程主机，并同时接受返回呢？

  
  import asyncio
  import traceback
  import paramiko
  from webshell.ssh import base, remotessh
  from webshell.connection import websocket
  
  
  class webshell:
      """整理 websocket 和 paramiko.channel,实现两者的数据互通"""
  
      def __init__(self, ws_session: websocket,
                   ssh_session: paramiko.sshclient = none,
                   chanel_session: paramiko.channel = none
                   ):
          self.ws_session = ws_session
          self.ssh_session = ssh_session
          self.chanel_session = chanel_session
  
      def init_ssh(self, host=none, port=22, user="admin", passwd="admin@123"):
          self.ssh_session, self.chanel_session = remotessh(host, port, user, passwd).session()
  
      def set_ssh(self, ssh_session, chanel_session):
          self.ssh_session = ssh_session
          self.chanel_session = chanel_session
  
      async def ready(self):
          await self.ws_session.accept()
  
      async def welcome(self):
          # 展示linux欢迎相关内容
          for i in range(2):
              if self.chanel_session.send_ready():
                  message = self.chanel_session.recv(2048).decode('utf-8')
                  if not message:
                      return
                  await self.ws_session.send_text(message)
  
      async def web_to_ssh(self):
          # print('--------web_to_ssh------->')
          while true:
              # print('--------------->')
              if not self.chanel_session.active or not self.ws_session.status:
                  return
              await asyncio.sleep(0.01)
              shell = await self.ws_session.receive_text()
              # print('-------shell-------->', shell)
              if self.chanel_session.active and self.chanel_session.send_ready():
                  self.chanel_session.send(bytes(shell, 'utf-8'))
              # print('--------------->', "end")
  
      async def ssh_to_web(self):
          # print('<--------ssh_to_web-----------')
          while true:
              # print('<-------------------')
              if not self.chanel_session.active:
                  await self.ws_session.send_text('ssh closed')
                  return
              if not self.ws_session.status:
                  return
              await asyncio.sleep(0.01)
              if self.chanel_session.recv_ready():
                  message = self.chanel_session.recv(2048).decode('utf-8')
                  # print('<---------message----------', message)
                  if not len(message):
                      continue
                  await self.ws_session.send_text(message)
              # print('<-------------------', "end")
  
      async def run(self):
          if not self.ssh_session:
              raise exception("ssh not init!")
          await self.ready()
          await asyncio.gather(
              self.web_to_ssh(),
              self.ssh_to_web()
          )
  
      def clear(self):
          try:
              self.ws_session.close()
          except exception:
              traceback.print_stack()
          try:
              self.ssh_session.close()
          except exception:
              traceback.print_stack()

前端

xterm.js 完全满足，搜索下找个看着简单的就行。

  
  export class term extends react.component {
      private terminal!: htmldivelement;
      private fitaddon = new fitaddon();
  
      componentdidmount() {
          const xterm = new terminal();
          xterm.loadaddon(this.fitaddon);
          xterm.loadaddon(new weblinksaddon());
  
          // using wss for https
          //         const socket = new websocket("ws://" + window.location.host + "/api/v1/ws");
          const socket = new websocket("ws://localhost:8000/webshell/");
          // socket.onclose = (event) => {
          //     this.props.onclose();
          // }
          socket.onopen = (event) => {
              xterm.loadaddon(new attachaddon(socket));
              this.fitaddon.fit();
              xterm.focus();
          }
  
          xterm.open(this.terminal);
          xterm.onresize(({ cols, rows }) => {
              socket.send("<resize>" + cols + "," + rows)
          });
  
          window.addeventlistener('resize', this.onresize);
      }
  
      componentwillunmount() {
          window.removeeventlistener('resize', this.onresize);
      }
  
      onresize = () => {
          this.fitaddon.fit();
      }
  
      render() {
          return <div classname="terminal" ref={(ref) => this.terminal = ref as htmldivelement}></div>;
      }
  }