正文
Django结合Websocket进行WebSSH的实现
提示：扫一扫查出行【扫一扫了解最新限行尾号】
复制提示
什么是webssh?泛指一种技术可以在网页上实现一个 终端。从而无需 之类的模拟终端工具进行 连接，将 这一比较低层的操作也从 架构扭成了 架构 这样的架构常用在运维制作开发一些堡垒机等系统中，或是目前比较新型的在线教育方式，通过向学生提供一个可以直接使用浏览器进行相关 操作或代码编写的学习方式 主要是建立客户端与服务端的即时通信
模型此种 实现方式，将通过结合 以及后端的 来进行实现，所需要的技术 栈如下
# 前端
vue
websocket
xterm.js
# 后端
django
dwebsocket (channels)
paramiko
threading
技术介绍xterm
前端通过xterm插件进行shell黑窗口环境的搭建，这个插件会自动解析由后台paramiko返回的带有标记样式的命令结果，并渲染到浏览器中，非常酷炫
websocket
这里通过websocket进行浏览器与django的数据交通桥梁
paramiko
paramiko此时的角色用来承担django与linux环境的交互，将前端发来的命令发送给后台，将 后台发来的命令结果返回到前端的xterm组件中
前端实现
vue发送websocket请求
<template>
<div>
<input type="text" v-model="message">
<p><input type="button" @click="send" value="发送"></p>
<p><input type="button" @click="close_socket" value="关闭"></p>
</div>
</template><script>
export default {
name:'ws',
data() {
return {
message:'',
testsocket:''
}
},
methods:{
send(){ // send 发送信息
// close 关闭连接 this.testsocket.send(this.message)
this.testsocket.onmessage = (res) => {
console.log("WS的返回结果",res.data); } },
close_socket(){
this.testsocket.close()
} },
mounted(){
this.testsocket = new WebSocket("ws://127.0.0.1:8000/ws/") // onopen 定义打开时的函数
// onclose 定义关闭时的函数
// onmessage 定义接收数据时候的函数
this.testsocket.onopen = function(){
console.log("开始连接socket")
},
this.testsocket.onclose = function(){
console.log("socket连接已经关闭")
}
}
}
</script>
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: '<App/>'
})
main.js
使用xterm和websocket来实时发送命令
<template>
<div class="console" id="terminal"></div>
</template>
<script>
import { Terminal } from 'xterm'
import * as attach from 'xterm/lib/addons/attach/attach'
import * as fit from 'xterm/lib/addons/fit/fit'
export default {
name: 'webssh',
data () {
return { term: null,
terminalSocket: null,
order:''
}
},
methods: { },
mounted () { //实例化一个websocket，用于和django江湖
this.terminalSocket = new WebSocket("ws://127.0.0.1:8000/web/");
//获取到后端传回的信息
this.terminalSocket.onmessage = (res) => {
console.log(res.data);
// var message = JSON.parse(res.data);
//将传回来的数据显示在xterm里
this.term.writeln("\r\n"+res.data);
//重置要发送的信息
this.order = ""
//换行，显示下一个开头
this.term.write("\r\n$ ");
}
//ws连接的时候
// this.terminalSocket.onopen = function(){
// console.log('websocket is Connected...')
// }
//ws关闭的时候
// this.terminalSocket.onclose = function(){
// console.log('websocket is Closed...')
// }
//ws错误的时候
// this.terminalSocket.onerror = function(){
// console.log('damn Websocket is broken!')
// }
// this.term.attach(this.terminalSocket)
// 绑定xterm到ws流中 }, let terminalContainer = document.getElementById('terminal')
//创建xterm实例
this.term = new Terminal({
cursorBlink: true, // 显示光标
cursorStyle: "underline" // 光标样式
}) // 创建一个新的Terminal对象 this.term.open(terminalContainer) // 将term挂载到dom节点上 //在xterm上显示命令行提示
this.term.write('$ ')
//监听xterm的键盘事件
this.term.on('key', (key, ev)=>{
// key是输入的字符 ev是键盘按键事件
console.log("key==========", ev.keyCode);
this.term.write(key) // 将输入的字符打印到黑板中
if (ev.keyCode == 13) { // 输入回车
// console.log("输入回车")
// this.term.write('$ ')
// console.log(this.order) //使用webscoket将数据发送到django
this.terminalSocket.send(this.order)
// this.order=''
console.log("里面的order",this.order)
}else if(ev.keyCode == 8){//删除按钮
//截取字符串[0,lenth-1]
this.order = this.order.substr(0,this.order.length-1) //清空当前一条的命令
this.term.write("\x1b[2K\r")
//简化当前的新的命令显示上
this.term.write("$ "+this.order) console.log("截取的字符串"+this.order)
typeof this.order
}else{// 将每次输入的字符拼凑起来
this.order += key
console.log("外面的order",this.order)} })
}, }
</script>
webssh.vue
后端实现基于channels实现websocket
安装channels
pip install channels
在setting的同级目录下创建routing.py
#routing.py
from channels.routing import ProtocolTypeRouterapplication = ProtocolTypeRouter({
# 暂时为空
})
配置setting
INSTALLED_APPS = [
'channels'
]ASGI_APPLICATION = "项目名.routing.application"
启动带有ASGI的django项目
带有ASGI的项目
平常项目
在app-chats中创建一个wsserver.py文件夹来保存关于websocket的处理视图
from channels.generic.websocket import WebsocketConsumerclass 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_urlapplication = ProtocolTypeRouter({
"websocket":URLRouter(
websocket_url
)
})
routing.py
Paramiko的使用
安装paramiko
pip install paramiko
使用paramiko
from django.test import TestCase# Create your tests here.
import paramikoclass 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()
webssh的后端实现INSTALLED_APPS=[
'channels',
'chats',
]ASGI_APPLICATION = "shiyanloupro.routing.application"
shiyanloupro/setting.py
from channels.routing import ProtocolTypeRouter,URLRouter
from chats.urls import websocket_urlapplication = 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 paramikosocket_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
泛指一种技术可以在网页上实现一个 终端。从而无需 之类的模拟终端工具进行 连接，将 这一比较低层的操作也从 架构扭成了 架构 这样的架构常用在运维制作开发一些堡垒机等系统中，或是目前比较新型的在线教育方式，通过向学生提供一个可以直接使用浏览器进行相关 操作或代码编写的学习方式 主要是建立客户端与服务端的即时通信
模型此种 实现方式，将通过结合 以及后端的 来进行实现，所需要的技术 栈如下
# 前端
vue
websocket
xterm.js
# 后端
django
dwebsocket (channels)
paramiko
threading
技术介绍xterm
前端通过xterm插件进行shell黑窗口环境的搭建，这个插件会自动解析由后台paramiko返回的带有标记样式的命令结果，并渲染到浏览器中，非常酷炫
websocket
这里通过websocket进行浏览器与django的数据交通桥梁
paramiko
paramiko此时的角色用来承担django与linux环境的交互，将前端发来的命令发送给后台，将 后台发来的命令结果返回到前端的xterm组件中
前端实现
vue发送websocket请求
<template>
<div>
<input type="text" v-model="message">
<p><input type="button" @click="send" value="发送"></p>
<p><input type="button" @click="close_socket" value="关闭"></p>
</div>
</template><script>
export default {
name:'ws',
data() {
return {
message:'',
testsocket:''
}
},
methods:{
send(){ // send 发送信息
// close 关闭连接 this.testsocket.send(this.message)
this.testsocket.onmessage = (res) => {
console.log("WS的返回结果",res.data); } },
close_socket(){
this.testsocket.close()
} },
mounted(){
this.testsocket = new WebSocket("ws://127.0.0.1:8000/ws/") // onopen 定义打开时的函数
// onclose 定义关闭时的函数
// onmessage 定义接收数据时候的函数
this.testsocket.onopen = function(){
console.log("开始连接socket")
},
this.testsocket.onclose = function(){
console.log("socket连接已经关闭")
}
}
}
</script>
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: '<App/>'
})
main.js
使用xterm和websocket来实时发送命令
<template>
<div class="console" id="terminal"></div>
</template>
<script>
import { Terminal } from 'xterm'
import * as attach from 'xterm/lib/addons/attach/attach'
import * as fit from 'xterm/lib/addons/fit/fit'
export default {
name: 'webssh',
data () {
return { term: null,
terminalSocket: null,
order:''
}
},
methods: { },
mounted () { //实例化一个websocket，用于和django江湖
this.terminalSocket = new WebSocket("ws://127.0.0.1:8000/web/");
//获取到后端传回的信息
this.terminalSocket.onmessage = (res) => {
console.log(res.data);
// var message = JSON.parse(res.data);
//将传回来的数据显示在xterm里
this.term.writeln("\r\n"+res.data);
//重置要发送的信息
this.order = ""
//换行，显示下一个开头
this.term.write("\r\n$ ");
}
//ws连接的时候
// this.terminalSocket.onopen = function(){
// console.log('websocket is Connected...')
// }
//ws关闭的时候
// this.terminalSocket.onclose = function(){
// console.log('websocket is Closed...')
// }
//ws错误的时候
// this.terminalSocket.onerror = function(){
// console.log('damn Websocket is broken!')
// }
// this.term.attach(this.terminalSocket)
// 绑定xterm到ws流中 }, let terminalContainer = document.getElementById('terminal')
//创建xterm实例
this.term = new Terminal({
cursorBlink: true, // 显示光标
cursorStyle: "underline" // 光标样式
}) // 创建一个新的Terminal对象 this.term.open(terminalContainer) // 将term挂载到dom节点上 //在xterm上显示命令行提示
this.term.write('$ ')
//监听xterm的键盘事件
this.term.on('key', (key, ev)=>{
// key是输入的字符 ev是键盘按键事件
console.log("key==========", ev.keyCode);
this.term.write(key) // 将输入的字符打印到黑板中
if (ev.keyCode == 13) { // 输入回车
// console.log("输入回车")
// this.term.write('$ ')
// console.log(this.order) //使用webscoket将数据发送到django
this.terminalSocket.send(this.order)
// this.order=''
console.log("里面的order",this.order)
}else if(ev.keyCode == 8){//删除按钮
//截取字符串[0,lenth-1]
this.order = this.order.substr(0,this.order.length-1) //清空当前一条的命令
this.term.write("\x1b[2K\r")
//简化当前的新的命令显示上
this.term.write("$ "+this.order) console.log("截取的字符串"+this.order)
typeof this.order
}else{// 将每次输入的字符拼凑起来
this.order += key
console.log("外面的order",this.order)} })
}, }
</script>
webssh.vue
后端实现基于channels实现websocket
安装channels
pip install channels
在setting的同级目录下创建routing.py
#routing.py
from channels.routing import ProtocolTypeRouterapplication = ProtocolTypeRouter({
# 暂时为空
})
配置setting
INSTALLED_APPS = [
'channels'
]ASGI_APPLICATION = "项目名.routing.application"
启动带有ASGI的django项目
带有ASGI的项目
平常项目
在app-chats中创建一个wsserver.py文件夹来保存关于websocket的处理视图
from channels.generic.websocket import WebsocketConsumerclass 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_urlapplication = ProtocolTypeRouter({
"websocket":URLRouter(
websocket_url
)
})
routing.py
Paramiko的使用
安装paramiko
pip install paramiko
使用paramiko
from django.test import TestCase# Create your tests here.
import paramikoclass 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()
webssh的后端实现INSTALLED_APPS=[
'channels',
'chats',
]ASGI_APPLICATION = "shiyanloupro.routing.application"
shiyanloupro/setting.py
from channels.routing import ProtocolTypeRouter,URLRouter
from chats.urls import websocket_urlapplication = 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 paramikosocket_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
此种 实现方式，将通过结合 以及后端的 来进行实现，所需要的技术 栈如下
# 前端
vue
websocket
xterm.js
# 后端
django
dwebsocket (channels)
paramiko
threading
技术介绍xterm
xterm
前端通过xterm插件进行shell黑窗口环境的搭建，这个插件会自动解析由后台paramiko返回的带有标记样式的命令结果，并渲染到浏览器中，非常酷炫
websocket
这里通过websocket进行浏览器与django的数据交通桥梁
paramiko
paramiko此时的角色用来承担django与linux环境的交互，将前端发来的命令发送给后台，将 后台发来的命令结果返回到前端的xterm组件中
前端实现
vue发送websocket请求
<template>
<div>
<input type="text" v-model="message">
<p><input type="button" @click="send" value="发送"></p>
<p><input type="button" @click="close_socket" value="关闭"></p>
</div>
</template><script>
export default {
name:'ws',
data() {
return {
message:'',
testsocket:''
}
},
methods:{
send(){ // send 发送信息
// close 关闭连接 this.testsocket.send(this.message)
this.testsocket.onmessage = (res) => {
console.log("WS的返回结果",res.data); } },
close_socket(){
this.testsocket.close()
} },
mounted(){
this.testsocket = new WebSocket("ws://127.0.0.1:8000/ws/") // onopen 定义打开时的函数
// onclose 定义关闭时的函数
// onmessage 定义接收数据时候的函数
this.testsocket.onopen = function(){
console.log("开始连接socket")
},
this.testsocket.onclose = function(){
console.log("socket连接已经关闭")
}
}
}
</script>
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: '<App/>'
})
main.js
使用xterm和websocket来实时发送命令
<template>
<div class="console" id="terminal"></div>
</template>
<script>
import { Terminal } from 'xterm'
import * as attach from 'xterm/lib/addons/attach/attach'
import * as fit from 'xterm/lib/addons/fit/fit'
export default {
name: 'webssh',
data () {
return { term: null,
terminalSocket: null,
order:''
}
},
methods: { },
mounted () { //实例化一个websocket，用于和django江湖
this.terminalSocket = new WebSocket("ws://127.0.0.1:8000/web/");
//获取到后端传回的信息
this.terminalSocket.onmessage = (res) => {
console.log(res.data);
// var message = JSON.parse(res.data);
//将传回来的数据显示在xterm里
this.term.writeln("\r\n"+res.data);
//重置要发送的信息
this.order = ""
//换行，显示下一个开头
this.term.write("\r\n$ ");
}
//ws连接的时候
// this.terminalSocket.onopen = function(){
// console.log('websocket is Connected...')
// }
//ws关闭的时候
// this.terminalSocket.onclose = function(){
// console.log('websocket is Closed...')
// }
//ws错误的时候
// this.terminalSocket.onerror = function(){
// console.log('damn Websocket is broken!')
// }
// this.term.attach(this.terminalSocket)
// 绑定xterm到ws流中 }, let terminalContainer = document.getElementById('terminal')
//创建xterm实例
this.term = new Terminal({
cursorBlink: true, // 显示光标
cursorStyle: "underline" // 光标样式
}) // 创建一个新的Terminal对象 this.term.open(terminalContainer) // 将term挂载到dom节点上 //在xterm上显示命令行提示
this.term.write('$ ')
//监听xterm的键盘事件
this.term.on('key', (key, ev)=>{
// key是输入的字符 ev是键盘按键事件
console.log("key==========", ev.keyCode);
this.term.write(key) // 将输入的字符打印到黑板中
if (ev.keyCode == 13) { // 输入回车
// console.log("输入回车")
// this.term.write('$ ')
// console.log(this.order) //使用webscoket将数据发送到django
this.terminalSocket.send(this.order)
// this.order=''
console.log("里面的order",this.order)
}else if(ev.keyCode == 8){//删除按钮
//截取字符串[0,lenth-1]
this.order = this.order.substr(0,this.order.length-1) //清空当前一条的命令
this.term.write("\x1b[2K\r")
//简化当前的新的命令显示上
this.term.write("$ "+this.order) console.log("截取的字符串"+this.order)
typeof this.order
}else{// 将每次输入的字符拼凑起来
this.order += key
console.log("外面的order",this.order)} })
}, }
</script>
webssh.vue
后端实现基于channels实现websocket
安装channels
pip install channels
在setting的同级目录下创建routing.py
#routing.py
from channels.routing import ProtocolTypeRouterapplication = ProtocolTypeRouter({
# 暂时为空
})
配置setting
INSTALLED_APPS = [
'channels'
]ASGI_APPLICATION = "项目名.routing.application"
启动带有ASGI的django项目
带有ASGI的项目
平常项目
在app-chats中创建一个wsserver.py文件夹来保存关于websocket的处理视图
from channels.generic.websocket import WebsocketConsumerclass 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_urlapplication = ProtocolTypeRouter({
"websocket":URLRouter(
websocket_url
)
})
routing.py
Paramiko的使用
安装paramiko
pip install paramiko
使用paramiko
from django.test import TestCase# Create your tests here.
import paramikoclass 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()
webssh的后端实现INSTALLED_APPS=[
'channels',
'chats',
]ASGI_APPLICATION = "shiyanloupro.routing.application"
shiyanloupro/setting.py
from channels.routing import ProtocolTypeRouter,URLRouter
from chats.urls import websocket_urlapplication = 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 paramikosocket_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
INSTALLED_APPS=[
'channels',
'chats',
]ASGI_APPLICATION = "shiyanloupro.routing.application"
shiyanloupro/setting.py
from channels.routing import ProtocolTypeRouter,URLRouter
from chats.urls import websocket_urlapplication = 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 paramikosocket_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
