前言

最近工作中需要開發前端操作遠程虛擬機的功能，簡稱 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>;
      }
  }