Django3如何使用WebSocket实现WebShell
前言最近工作中需要开发前端操作远程虚拟机的功能，简称 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'] == 'http': await django_application(scope, receive, send) elif scope['type'] == 'websocket': await websocket_application(scope, receive, send) else: raise notimplementederror(funknown scope type {scope['type']})# websocket.pyasync def websocket_application(scope, receive, send): pass # websocket.pyasync 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!' }) 实现 上面的代码提供了思路
其中最核心的实现部分我放下面：
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 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_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(' { // 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( + cols + , + rows) }); window.addeventlistener('resize', this.onresize); } componentwillunmount() { window.removeeventlistener('resize', this.onresize); } onresize = () => { this.fitaddon.fit(); } render() { return this.terminal = ref as htmldivelement}>
; }} 原文链接：https://www.cnblogs.com/lgjbky/p/15186188.html
水蒸气透过率测试仪的产品特点及其配置的介绍
基于PLC控制的卷绕机如何实现远程监控和程序上下载
L4970A构成的多输出DC DC变换器电路
交付上海浦东机场，菲亚特动力科技再次助力机场地面设备
种子低温低湿储藏柜是什么，它的作用有哪些
Django3如何使用WebSocket实现WebShell
嵌入式开发之java常用开发工具介绍
联想在CES 2020上推出拯救者显卡坞，内置标准的ATX电源
XQ58-A Valkyrie无人机僚机未来与战斗机协同作战!
中美贸易摩擦升级,戴乐格半导体CEO力挺中国,
单片机内部为什么没有晶振？
华为2021年目标：搭载鸿蒙的设备至少3亿台
霍尔式转速传感器原理 磁电式转速传感器原理
如何为偏置电流提供直流回路
谊美吉斯LED智慧大屏赋能城市商业发展
怎么设计一个阶梯阶梯波发生器
飞凌年度展会盘点，来看看最受欢迎的产品有哪些？
搭建做场外大宗合约C2C交易所源码APP系统
从手机制造者到5G中坚力量，对于vivo你到底了解多少？
中国造柔性屏预计2020年市场空间将突破25%
大致看了下觉得这不够有趣，翻了翻 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'] == 'http': await django_application(scope, receive, send) elif scope['type'] == 'websocket': await websocket_application(scope, receive, send) else: raise notimplementederror(funknown scope type {scope['type']})# websocket.pyasync def websocket_application(scope, receive, send): pass # websocket.pyasync 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!' }) 实现 上面的代码提供了思路
其中最核心的实现部分我放下面：
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 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_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(' { // 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( + cols + , + rows) }); window.addeventlistener('resize', this.onresize); } componentwillunmount() { window.removeeventlistener('resize', this.onresize); } onresize = () => { this.fitaddon.fit(); } render() { return this.terminal = ref as htmldivelement}>
; }} 原文链接：https://www.cnblogs.com/lgjbky/p/15186188.html
水蒸气透过率测试仪的产品特点及其配置的介绍
基于PLC控制的卷绕机如何实现远程监控和程序上下载
L4970A构成的多输出DC DC变换器电路
交付上海浦东机场，菲亚特动力科技再次助力机场地面设备
种子低温低湿储藏柜是什么，它的作用有哪些
Django3如何使用WebSocket实现WebShell
嵌入式开发之java常用开发工具介绍
联想在CES 2020上推出拯救者显卡坞，内置标准的ATX电源
XQ58-A Valkyrie无人机僚机未来与战斗机协同作战!
中美贸易摩擦升级,戴乐格半导体CEO力挺中国,
单片机内部为什么没有晶振？
华为2021年目标：搭载鸿蒙的设备至少3亿台
霍尔式转速传感器原理 磁电式转速传感器原理
如何为偏置电流提供直流回路
谊美吉斯LED智慧大屏赋能城市商业发展
怎么设计一个阶梯阶梯波发生器
飞凌年度展会盘点，来看看最受欢迎的产品有哪些？
搭建做场外大宗合约C2C交易所源码APP系统
从手机制造者到5G中坚力量，对于vivo你到底了解多少？
中国造柔性屏预计2020年市场空间将突破25%