作者：从零开始的程序员生活来源：
前言最近工作中需要开发前端操作远程虚拟机的功能，简称 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(/) 就是一个很好的实例，但过于简单……
思路# asgi.py import osfrom django.core.asgi import get_asgi_applicationfrom websocket_app.websocket import websocket_applicationos.environ.setdefault('DJANGO_SETTINGS_MODULE', 'websocket_app.settings')django_application = get_asgi_application()async def application(scope, receive, send):if scope['type'] == '/) 基本可以复用了。
其中最核心的实现部分我放下面：
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 @propertydef headers(self):return Headers(self._scope) @propertydef scheme(self):return self._scope["scheme"] @propertydef path(self):return self._scope["path"] @propertydef query_params(self):return QueryParams(self._scope["query_string"].decode()) @propertydef query_string(self) -> str:return self._scope["query_string"] @propertydef scope(self):return self._scopeasync 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.DISCONNECTEDelse: self._app_state = State.CONNECTEDelif 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.DISCONNECTEDawait 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.CONNECTEDelif 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.DISCONNECTEDreturn message缝合怪做为合格的代码搬运工，为了提高搬运效率还是要造点轮子填点坑的，如何将上面的 WebSocket 类与 paramiko 结合起来，实现从前端接受字符传递给远程主机，并同时接受返回呢？
import asyncioimport tracebackimport paramikofrom webshell.ssh import Base, RemoteSSHfrom webshell.connection import WebSocketclass 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_sessiondef 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_sessionasync 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:returnawait 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:returnawait 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')returnif not self.ws_session.status:returnawait 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):continueawait 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 !
