泛指一种技术可以在网页上实现一个 终端。从而无需 之类的模拟终端工具进行 连接,将 这一比较低层的操作也从 架构扭成了 架构 这样的架构常用在运维制作开发一些堡垒机等系统中,或是目前比较新型的在线教育方式,通过向学生提供一个可以直接使用浏览器进行相关 操作或代码编写的学习方式 主要是建立客户端与服务端的即时通信
此种 实现方式,将通过结合 以及后端的 来进行实现,所需要的技术 栈如下
# 前端
vue
websocket
xterm.js
# 后端
django
dwebsocket (channels)
paramiko
threading
xterm
前端通过xterm插件进行shell黑窗口环境的搭建,这个插件会自动解析由后台paramiko返回的带有标记样式的命令结果,并渲染到浏览器中,非常酷炫
websocket
这里通过websocket进行浏览器与django的数据交通桥梁
paramiko
paramiko此时的角色用来承担django与linux环境的交互,将前端发来的命令发送给后台,将 后台发来的命令结果返回到前端的xterm组件中
vue发送websocket请求
@click="send" value="发送"> @click="close\_socket" value="关闭">
ws.vue
安装xterm
cnpm install xterm@3.1.0 --save //指定版本安装
在vue框架中引入xterm的样式文件
// The Vue build version to load with the `import` command
// (runtime-only or standalone) has been set in webpack.base.conf with an alias. import Vue from 'vue'
import App from './App'
import router from './router'
import 'xterm/dist/xterm.css' // 看这里,添加xterm css文件样式
Vue.config.productionTip = false
/* eslint-disable no-new */
new Vue({
el: '#app',
router,
components: { App },
template: '
})
main.js
使用xterm和websocket来实时发送命令
webssh.vue
安装channels
pip install channels
在setting的同级目录下创建routing.py
#routing.py
from channels.routing import ProtocolTypeRouter
application = ProtocolTypeRouter({
# 暂时为空
})
配置setting
INSTALLED_APPS = [
'channels'
]
ASGI_APPLICATION = "项目名.routing.application"
启动带有ASGI的django项目
带有ASGI的项目
平常项目
在app-chats中创建一个wsserver.py文件夹来保存关于websocket的处理视图
from channels.generic.websocket import WebsocketConsumer
class ChatService(WebsocketConsumer):
# 当Websocket创建连接时
def connect(self):
#websocket保持连接
self.accept()
pass
# 当Websocket接收到消息时
def receive(self, text\_data=None, bytes\_data=None):
pass
# 当Websocket发生断开连接时
def disconnect(self, code):
pass
wsserver.py
配置对应的路由
from django.urls import path
from chats.chatService import ChatService
websocket_url = [
path("ws/",ChatService)
]
url.py
在routing.py里增加关于websocket的非http请求的url
from channels.routing import ProtocolTypeRouter,URLRouter
from chats.urls import websocket_url
application = ProtocolTypeRouter({
"websocket":URLRouter(
websocket_url
)
})
routing.py
安装paramiko
pip install paramiko
使用paramiko
from django.test import TestCase
import paramiko
class WebSsh(object):
def client_ssh(self):
sh = paramiko.SSHClient() # 1 创建SSH对象
sh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) # 2 允许连接不在know_hosts文件中的主机
sh.connect("10.211.55.17", username="parallels", password="beijing") # 3 连接服务器
stdin, stdout, stderr = sh.exec_command('ls')
right_info = stdout.read()
err_info = stderr.read()
if right\_info:
print(right\_info.decode("utf-8"))
elif err\_info:
print(err\_info.decode("utf-8"))
else:
print("命令执行成功")
if __name__ == '__main__':
a = WebSsh()
a.client_ssh()
INSTALLED_APPS=[
'channels',
'chats',
]
ASGI_APPLICATION = "shiyanloupro.routing.application"
shiyanloupro/setting.py
from channels.routing import ProtocolTypeRouter,URLRouter
from chats.urls import websocket_url
application = ProtocolTypeRouter({
"websocket":URLRouter(
websocket_url
)
})
shiyanloupro/routing.py
from django.urls import path
from chats.chatservice import ChatService,WebSSHService
websocket_url = [
path("ws/",ChatService),
path("web/",WebSSHService),
]
chats/urls.py
from channels.generic.websocket import WebsocketConsumer
import paramiko
socket_list = []
class ChatService(WebsocketConsumer):
# 当Websocket创建连接时
def connect(self):
self.accept()
socket_list.append(self)
# 当Websocket接收到消息时
def receive(self, text\_data=None, bytes\_data=None):
print(text\_data) # 打印收到的数据
for ws in socket\_list: # 遍历所有的WebsocketConsumer对象
ws.send(text\_data) # 对每一个WebsocketConsumer对象发送数据
# 当Websocket发生断开连接时
def disconnect(self, code):
print(f'sorry{self},你被女朋友抛弃了')
socket\_list.remove(self)
class WebSSHService(WebsocketConsumer):
def connect(self):
self.accept()
self.sh = paramiko.SSHClient() # 1 创建SSH对象
self.sh.set\_missing\_host\_key\_policy(paramiko.AutoAddPolicy()) # 2 允许连接不在know\_hosts文件中的主机
self.sh.connect("10.211.55.17", username="parallels", password="beijing") # 3 连接服务器
print("连接成功")
def receive(self, text\_data=None, bytes\_data=None):
print(str(text\_data)) # 打印收到的数据
print(type(text\_data))
stdin, stdout, stderr = self.sh.exec\_command(text\_data)
right\_info = stdout.read()
err\_info = stderr.read()
print(right\_info)
if right\_info:
new\_data = right\_info.decode("utf-8").replace("\\n","\\r\\n")
print(new\_data)
self.send(new\_data)
elif err\_info:
new\_data = err\_info.decode("utf-8").replace("\\n", "\\r\\n")
print(new\_data)
self.send(new\_data)
else:
print(self.send("命令执行成功"))
def disconnect(self, code):
print(f'sorry{self},你被女朋友抛弃了')
chats/chatservice.py
手机扫一扫
移动阅读更方便
你可能感兴趣的文章