详解基于django实现的webssh简单例子

站长资源 2024/11/26 佚名

2 0 1

本文介绍了详解基于django实现的webssh简单例子，分享给大家，具体如下：

说明

新建一个 django 程序，本文为 chain。

以下仅为简单例子，实际应用可根据自己平台情况进行修改。

打开首页后，需要输入1，后台去登录主机，然后返回登录结果。

正常项目可以post 主机和登录账户，进行权限判断，然后去后台读取账户密码，进行登录。

djang后台

需要安装以下模块

安装后会有一个版本号报错，不影响

channels==2.0.2
channels-redis==2.1.0
amqp==1.4.9
anyjson==0.3.3
asgi-redis==1.4.3
asgiref==2.3.0
async-timeout==2.0.0
attrs==17.4.0

cd /tmp/
wget https://files.pythonhosted.org/packages/12/2a/e9e4fb2e6b2f7a75577e0614926819a472934b0b85f205ba5d5d2add54d0/Twisted-18.4.0.tar.bz2
tar xf Twisted-18.4.0.tar.bz2
cd Twisted-18.4.0
python3 setup.py install

启动redis

chain/
    chain/
       settings.py
       asgi.py
       consumers.py
       routing.py
  templates/
      index.html

settings.py

# django-channels配置
CHANNEL_LAYERS = {
  "default": {
    "BACKEND": "channels_redis.core.RedisChannelLayer",
    "CONFIG": {
      "hosts": [("127.0.0.1", 6379)],
    },
  },
}

# 配置ASGI
ASGI_APPLICATION = "chain.routing.application"

consumers.py

from asgiref.sync import async_to_sync
from channels.generic.websocket import WebsocketConsumer

import paramiko
import threading
import time

from channels.layers import get_channel_layer
channel_layer = get_channel_layer()

class MyThread(threading.Thread):
  def __init__(self, id, chan):
    threading.Thread.__init__(self)
    self.chan = chan

  def run(self):
    while not self.chan.chan.exit_status_ready():
      time.sleep(0.1)
      try:
        data = self.chan.chan.recv(1024)
        async_to_sync(self.chan.channel_layer.group_send)(
          self.chan.scope['user'].username,
          {
            "type": "user.message",
            "text": bytes.decode(data)
          },
        )
      except Exception as ex:
        print(str(ex))
    self.chan.sshclient.close()
    return False

class EchoConsumer(WebsocketConsumer):

  def connect(self):
    # 创建channels group， 命名为：用户名，并使用channel_layer写入到redis
    async_to_sync(self.channel_layer.group_add)(self.scope['user'].username, self.channel_name)
    # 返回给receive方法处理
    self.accept()

  def receive(self, text_data):

    if text_data == '1':
      self.sshclient = paramiko.SSHClient()
      self.sshclient.load_system_host_keys()
      self.sshclient.set_missing_host_key_policy(paramiko.AutoAddPolicy())
      self.sshclient.connect('47.104.140.38', 22, 'root', '123456')
      self.chan = self.sshclient.invoke_shell(term='xterm')
      self.chan.settimeout(0)
      t1 = MyThread(999, self)
      t1.setDaemon(True)
      t1.start()
    else:
      try:
        self.chan.send(text_data)
      except Exception as ex:
        print(str(ex))

  def user_message(self, event):
    # 消费
    self.send(text_data=event["text"])

  def disconnect(self, close_code):
    async_to_sync(self.channel_layer.group_discard)(self.scope['user'].username, self.channel_name)

asgi.py

import os
import django
from channels.routing import get_default_application

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "chain.settings")
django.setup()
application = get_default_application()
routing.py
from channels.auth import AuthMiddlewareStack
from channels.routing import URLRouter, ProtocolTypeRouter
from django.urls import path

from .consumers import EchoConsumer

application = ProtocolTypeRouter({
  "websocket": AuthMiddlewareStack(
    URLRouter([
      path(r"ws/", EchoConsumer),
      # path(r"stats/", StatsConsumer),
    ])
  )
})

网页设置：

index.html

<!DOCTYPE html>
<html lang="en">
<head>
  <meta charset="UTF-8">
  <title>django webssh 例子</title>

  <link href="/static/css/plugins/ztree/awesomeStyle/awesome.css" rel="external nofollow" rel="stylesheet">

  <link href="/static/webssh_static/css/xterm.min.css" rel="external nofollow" rel="stylesheet" type="text/css"/>
  <style>
    body {
      padding-bottom: 30px;
    }

    .terminal {
      border: #000 solid 5px;
      font-family: cursive;
    {#        font-family: Arial, Helvetica, Tahoma ,"Monaco", "DejaVu Sans Mono", "Liberation Mono", sans-serif;#}{#        font-family: Tahoma, Helvetica, Arial, sans-serif;#}{#        font-family: "\5B8B\4F53","","Monaco", "DejaVu Sans Mono", "Liberation Mono", "Microsoft YaHei", monospace;#} font-size: 15px;
    {#        color: #f0f0f0;#} background: #000;
    {#        width: 893px;#}{#        height: 550px;#} box-shadow: rgba(0, 0, 0, 0.8) 2px 2px 20px;
    }

    .reverse-video {
      color: #000;
      background: #f0f0f0;
    }
  </style>
</head>
<body>

<div id="terms"></div>
</body>

<script src="/UploadFiles/2021-04-08/xterm.min.js">

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持。

django实现webssh,django,webssh

广告合作：本站广告合作请联系QQ：858582 申请时备注：广告合作（否则不回）
免责声明：本站资源来自互联网收集,仅供用于学习和交流,请遵循相关法律法规,本站一切资源不代表本站立场,如有侵权、后门、不妥请联系本站删除！

评论“详解基于django实现的webssh简单例子”

暂无评论...

www.wwsws.com 伏龙阁资源网

39,976影音资源

44,792技术资源

21,817软件资源

651,128站长资源

稳了！魔兽国服回归的3条重磅消息！官宣时间再确认！

昨天有一位朋友在大神群里分享，自己亚服账号被封号之后居然弹出了国服的封号信息对话框。

这里面让他访问的是一个国服的战网网址，com.cn和后面的zh都非常明白地表明这就是国服战网。
而他在复制这个网址并且进行登录之后，确实是网易的网址，也就是我们熟悉的停服之后国服发布的暴雪游戏产品运营到期开放退款的说明。这是一件比较奇怪的事情，因为以前都没有出现这样的情况，现在突然提示跳转到国服战网的网址，是不是说明了简体中文客户端已经开始进行更新了呢？

更新日志

2024年11月26日

详解基于django实现的webssh简单例子

python中使用print输出中文的方法

django用户登录和注销的实现方法

评论“详解基于django实现的webssh简单例子”

稳了！魔兽国服回归的3条重磅消息！官宣时间再确认！

更新日志

友情链接