Django Ajax 实现Web命令行执行

Ajax序列化实现简单命令工具: 我们通过定制Django,配合命令行执行功能可以实现远程命令执行页面.

<!--name: index.html-->
{% extends "admin/base_site.html" %}
{% load i18n static %}
{% block content %}
<link rel="stylesheet" href="https://lyshark.com/cdn/xterm.css" />
<link rel="stylesheet" href="https://lyshark.com/cdn/bootstrap3.css" />
<script src="https://lyshark.com/cdn/xterm.js"></script>
<script src="https://lyshark.com/cdn/jquery.js"></script>

<div class="panel panel-primary">
<div class="panel-heading">
<h3 class="panel-title">命令执行CMD</h3>
</div>
<div class="panel-body">
<div id="terminal"></div>
</div>
<div class="panel-footer">
<input type="text" id="address" placeholder="主机地址" style="width:200px;height:40px"/>
<input type="text" id="command" placeholder="执行命令" style="width:400px;height:40px"/>
<input type="button" value="执行命令" onclick="show()">
</div>
</div>
<script type="text/javascript">
var window_width = $(window).width()-200;
var window_height = $(window).height()-300;
var term = new Terminal(
{
cols: Math.floor(window_width/9),
rows: Math.floor(window_height/20),
useStyle:false,
convertEol: true,
cursorBlink:false,
cursorStyle:null,
rendererType: "canvas",
}
);
term.open(document.getElementById('terminal'));
function show(){
var address = $("#address").val();
var command = $("#command").val();
console.log(command);
$.ajax({
url:"/term/",
type:"POST",
contentType:"application/json;",
data: JSON.stringify({"address":address,"command":command}),
success:function (res) {
term.clear();
term.writeln(res);
term.writeln("\x1B[1;3;32m 执行时间: \x1B[0m" + myDate.toLocaleString() +
"\x1B[1;3;33m IP地址: \x1B[0m" + address + "\x1B[1;3;34m 命令: \x1B[0m" +
command );
}
});
}
</script>
{% endblock %}

视图层

# name: views.py
from django.shortcuts import render,HttpResponse
import paramiko,json

ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())

def ssh_shell(address,username,password,port,command):
ssh.connect(address,port=port,username=username,password=password)
stdin, stdout, stderr = ssh.exec_command(command)
result = stdout.read()
if not result:
result=stderr.read()
ssh.close()
return result.decode()

def term(request):
if request.method == "POST":
data = request.body.decode("utf-8")
json_data = json.loads(data)
address = json_data.get("address")
command = json_data.get("command")
if len(data) >=2:
ret = ssh_shell(address,"root","123","22",command)
return HttpResponse(ret)
else:
return HttpResponse("None")
return render(request, "index.html")

路由层

# name: urls.py
from MyWeb import views
urlpatterns = [
path('admin/', admin.site.urls),
path('term/',views.term)
]

Ajax实现非交互Shell: 虽然Ajax是非交互的,但利用循环交互同样可以实现远程命令执行功能.

首先是前端​​index.html​​代码中我们使用了xterm这个前端命令行库,该库可实现一个WebShell环境.

<link rel="stylesheet" href="https://cdn.lyshark.com/xterm/xterm.css" />
<script src="https://cdn.lyshark.com/xterm/xterm.js"></script>
<script src="https://cdn.lyshark.com/jquery/3.5.1/jquery.min.js"></script>

<div id="terminal"></div>
<script type="text/javascript">
var window_width = $(window).width()-200;
var window_height = $(window).height()-300;
var term = new Terminal(
{
cols: Math.floor(window_width/9),
rows: Math.floor(window_height/20),
useStyle:false,
convertEol: true,
cursorBlink:true,
cursorStyle:null,
rendererType: "canvas",
}
);
term.open(document.getElementById('terminal'));
term.writeln("welcome to lyshark web terminal!");
term.write("[shell] # ");

let input = '';
term.on('key', (key, ev) => {
let code = key.charCodeAt(0);
console.log(code);
// 如果按下回车,则发送命令,并等待输出结果
if(code == 13)
{
term.write("\r\n");
$.ajax({
url:"/_ajax/",
type:"POST",
contentType:"application/json;",
data: JSON.stringify({"command":input}),
success:function (res) {
term.write(res);
}
});
input ='';
}
else
{
input += key
term.write(key);
}
});
</script>

而对应的后台文件​​views.py​​视图中只需要接受参数并判断执行即可.

from django.shortcuts import render,HttpResponse
import paramiko,json

ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())

def ssh_shell(address,username,password,port,command):
ssh.connect(address,port=port,username=username,password=password)
stdin, stdout, stderr = ssh.exec_command(command)
result = stdout.read()
if not result:
result=stderr.read()
ssh.close()
return result.decode()

def ajax(request):
if request.method == "POST":
data = request.body.decode("utf-8")
json_data = json.loads(data)
command = json_data.get("command")

if len(data) == 0:
return HttpResponse("[shell] # ")
else:
ref = ssh_shell("192.168.1.112","root","1233","22",command)
return HttpResponse(ref + "\n[shell] # ")
return render(request, "index.html")

Web命令执行工具: 实现一个简单的Web版命令行执行工具.

{% extends "admin/base_site.html" %}
{% load i18n static %}
{% block content %}
<link rel="stylesheet" href="https://lyshark.com/cdn/xterm.css" />
<link rel="stylesheet" href="https://lyshark.com/cdn/bootstrap3.css" />
<script src="https://lyshark.com/cdn/xterm.js"></script>
<script src="https://lyshark.com/cdn/jquery.js"></script>

<div class="panel panel-primary">
<div class="panel-heading">
<h3 class="panel-title">批量命令执行CMD工具</h3>
</div>
<div class="panel-body">
<div id="terminal"></div>
</div>
<div class="panel-footer">
<input type="text" id="address" placeholder="主机范围" style="width:200px;height:40px"/>
<input type="text" id="command" placeholder="执行命令" style="width:400px;height:40px"/>
<input type="button" value="执行命令" onclick="show()">
</div>
</div>
<script type="text/javascript">
var window_width = $(window).width()-200;
var window_height = $(window).height()-300;
var term = new Terminal(
{
cols: Math.floor(window_width/9),
rows: Math.floor(window_height/20),
useStyle:false,
convertEol: true,
cursorBlink:true,
cursorStyle:null,
rendererType: "canvas",
}
);
term.open(document.getElementById('terminal'));
function show(){
var address = $("#address").val();
var command = $("#command").val();
console.log(command);
$.ajax({
url:"/term/",
type:"POST",
contentType:"application/json;",
data: JSON.stringify({"address":address,"command":command}),
success:function (res) {
term.writeln(res);
}
});
}
</script>
{% endblock %}

视图层

from django.shortcuts import render,HttpResponse
import paramiko,json,time

ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())

def ssh_shell(address,username,password,port,command):
try:
ssh.connect(address,port=port,username=username,password=password)
stdin, stdout, stderr = ssh.exec_command(command)
result = stdout.read()
if not result:
result=stderr.read()
ssh.close()
return result.decode()
except Exception:
ssh.close()
def term(request):
if request.method == "POST":
data = request.body.decode("utf-8")
json_data = json.loads(data)
address = json_data.get("address")
command = json_data.get("command")
if len(address) >=2 and len(command) >=2:
ret = ssh_shell(address,"root","123","22",command)
if ret !=None:
times = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
times = "\x1B[1;3;25m [执行成功] \x1B[0m ---> \x1B[1;3;32m 执行时间:[{}] \x1B[0m".format(times)
address = "\x1B[1;3;33m 地址:[ {} ] \x1B[0m".format(address)
command = "\x1B[1;3;35m 命令:[ {} ] \x1B[0m".format(command)
retn = times + address + command
return HttpResponse(retn)
else:
times = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
times = "\x1B[1;3;31m [执行失败] \x1B[0m ---> \x1B[1;3;32m 执行时间:[{}] \x1B[0m".format(times)
address = "\x1B[1;3;33m 地址:[ {} ] \x1B[0m".format(address)
command = "\x1B[1;3;35m 命令:[ {} ] \x1B[0m".format(command)
retn = times + address + command
return HttpResponse(retn)
else:
return HttpResponse("Error")
return render(request, "index.html")

路由层

from MyWeb import views
urlpatterns = [
path('admin/', admin.site.urls),
path('term/',views.term)
]

批量CMD执行工具: 利用DjangoAdmin与Socket通信实现的主机批量执行并回显.

<!--name:index.html-->
{% extends "admin/base_site.html" %}
{% load i18n static %}
{% block content %}
<link rel="stylesheet" href="https://lyshark.com/cdn/xterm.css" />
<link rel="stylesheet" href="https://lyshark.com/cdn/bootstrap3.css" />
<script src="https://lyshark.com/cdn/xterm.js"></script>
<script src="https://lyshark.com/cdn/jquery.js"></script>
<div class="panel panel-primary">
<div class="panel-heading">
<h3 class="panel-title">批量命令执行CMD工具</h3>
</div>
<div class="panel-body">
<div id="terminal"></div>
</div>
<div class="panel-footer">
<input type="text" id="address" placeholder="主机范围 127.0.0.1-100" style="width:200px;height:40px"/>
<input type="text" id="command" placeholder="执行命令 ls -lh " style="width:400px;height:40px"/>
<input type="button" id="send_message" value="批量执行">
</div>
</div>
<script type="text/javascript">
$(function(){
var window_width = $(window).width()-200;
var window_height = $(window).height()-300;
var term = new Terminal(
{
cols: Math.floor(window_width/9),
rows: Math.floor(window_height/20),
convertEol: true,
cursorBlink:false,
});
var sock = new WebSocket("ws://" + window.location.host + "/echo/");
sock.onopen = function () {
term.open(document.getElementById('terminal'));
console.log('WebSocket Open');
};
sock.onmessage = function (recv) {
if(recv.data.substring(0,7) == "[Suces]"){
term.writeln("\x1B[1;3;32m " + recv.data + "\x1B[0m");
}else{
term.writeln("\x1B[1;3;31m " + recv.data + "\x1B[0m");
}

};
$('#send_message').click(function () {
var message ={"address":null,"command":null};
message['address'] = $("#address").val();
message['command'] = $("#command").val();
var send_data = JSON.stringify(message);
window.sock.send(send_data);
});
window.sock = sock;
});
</script>
{% endblock %}

视图层

# name:views.py
from django.shortcuts import render
from dwebsocket.decorators import accept_websocket
import paramiko,time

ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())

def ssh_shell(address,username,password,port,command):
try:
ssh.connect(address,port=port,username=username,password=password,timeout=1)
stdin, stdout, stderr = ssh.exec_command(command)
result = stdout.read()
if not result:
result=stderr.read()
ssh.close()
return result.decode()
except Exception:
return 0

def CalculationIP(Addr_Count):
ret = []
try:
IP_Start = str(Addr_Count.split("-")[0]).split(".")
IP_Heads = str(IP_Start[0] + "." + IP_Start[1] + "." + IP_Start[2] +".")
IP_Start_Range = int(Addr_Count.split(".")[3].split("-")[0])
IP_End_Range = int(Addr_Count.split("-")[1])
for item in range(IP_Start_Range,IP_End_Range+1):
ret.append(IP_Heads+str(item))
return ret
except Exception:
return 0

@accept_websocket
def echo(request):
if not request.is_websocket():
return render(request, "index.html")
else:
for message in request.websocket:
data = eval(message)
Addr_list = CalculationIP(data['address'])
command = data['command']
for ip in Addr_list:
ret = ssh_shell(ip,"root","123","22",command)
if ret != 0:
times = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
retn = "[Suces] ---> " + str(times) + " " + command + " " + ip
request.websocket.send(retn)
else:
times = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
retn = "[Error] ---> " + str(times) + " " + command + " " + ip
request.websocket.send(retn)

路由层

# name:urls.py
from MyWeb import views

urlpatterns = [
path('admin/', admin.site.urls),
path("echo/",views.echo)
]

版权声明：本博客文章与代码均为学习时整理的笔记，文章 [均为原创] 作品，转载请 [添加出处] ，您添加出处是我创作的动力！