Django结合Websocket进行WebSSH的实现
阅读原文时间:2023年07月10日阅读:1

  泛指一种技术可以在网页上实现一个 终端。从而无需 之类的模拟终端工具进行 连接,将 这一比较低层的操作也从 架构扭成了 架构 这样的架构常用在运维制作开发一些堡垒机等系统中,或是目前比较新型的在线教育方式,通过向学生提供一个可以直接使用浏览器进行相关 操作或代码编写的学习方式 主要是建立客户端与服务端的即时通信

此种 实现方式,将通过结合 以及后端的 来进行实现,所需要的技术 栈如下

# 前端
vue
websocket
xterm.js

# 后端
django
dwebsocket (channels)
paramiko
threading

  xterm

    前端通过xterm插件进行shell黑窗口环境的搭建,这个插件会自动解析由后台paramiko返回的带有标记样式的命令结果,并渲染到浏览器中,非常酷炫

  websocket

    这里通过websocket进行浏览器与django的数据交通桥梁

  paramiko

    paramiko此时的角色用来承担django与linux环境的交互,将前端发来的命令发送给后台,将 后台发来的命令结果返回到前端的xterm组件中

vue发送websocket请求

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实现websocket

安装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

Create your tests here.

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

手机扫一扫

移动阅读更方便

阿里云服务器
腾讯云服务器
七牛云服务器

你可能感兴趣的文章