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

本文介绍了详解基于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="/static/webssh_static/js/xterm.min.js"></script>
<script>
  var socket = new WebSocket('ws://' + window.location.host + '/ws/');

  socket.onopen = function () {

    var term = new Terminal();
    term.open(document.getElementById('terms'));

    term.on('data', function (data) {
      console.log(data);
      socket.send(data);
    });

    socket.onmessage = function (msg) {
      console.log(msg);
      console.log(msg.data);
      term.write(msg.data);
    };
    socket.onerror = function (e) {
      console.log(e);
    };

    socket.onclose = function (e) {
      console.log(e);
      term.destroy();
    };
  };

</script>
</html>

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

时间: 2018-07-14

Python如何爬取实时变化的WebSocket数据的方法

一.前言 作为一名爬虫工程师,在工作中常常会遇到爬取实时数据的需求,比如体育赛事实时数据.股市实时数据或币圈实时变化的数据.如下图: Web 领域中,用于实现数据'实时'更新的手段有轮询和 WebSocket 这两种.轮询指的是客户端按照一定时间间隔(如 1 秒)访问服务端接口,从而达到 '实时' 的效果,虽然看起来数据像是实时更新的,但实际上它有一定的时间间隔,并不是真正的实时更新.轮询通常采用 拉 模式,由客户端主动从服务端拉取数据. WebSocket 采用的是 推 模式,由服务端主动将数

详解Django-channels 实现WebSocket实例

引入 先安装三个模块 pip install channels pip install channels_redis pip install pywin32 创建一个Django项目和一个app 项目名随意,app名随意.这里项目名为 django_websocket_demo ,app名 chat 把app文件夹下除了 views.py 和 __init__.py 的文件都删了,最终项目目录结构如下: django_websocket_demo/ manage.py django_websoc

Django通过dwebsocket实现websocket的例子

与django推荐的channel不同,dwebsocket使用更加方便简单 使用方法1: 只需views.py文件中,将对应的视图函数添加装饰器 accept_websocket--可以接受websocket请求和普通http请求 require_websocket----只接受websocket请求,拒绝普通http请求 from dwebsocket.decorators import accept_websocket,require_websocket @accept_websocket

python实现WebSocket服务端过程解析

一种类似Flask开发的WebSocket-Server服务端框架,适用python3.X 1.安装模块Pywss pip install pywss 2.搭建简易服务器 2.1 服务端代码 代码简介 route: 注册请求路径 example_1(request, data): request: socket句柄,能够发送和接收数据接.发送数据request.ws.send(data),收数据request.ws_recv(1024) data: 客户端发送的数据存于此处 from pywss

Python Websocket服务端通信的使用示例

1.唠唠叨叨 最近又回顾了下Websocket,发现已经忘的七七八八了.于是用js写了客户端,用python写了服务端,来复习一下这方面的知识. WebSocket 是一种标准协议,用于在客户端和服务端之间进行双向数据传输.但它跟 HTTP 没什么关系,它是基于 TCP 的一种独立实现. 以前客户端想知道服务端的处理进度,要不停地使用 Ajax 进行轮询,让浏览器隔个几秒就向服务器发一次请求,这对服务器压力较大.另外一种轮询就是采用 long poll 的方式,这就跟打电话差不多,没收到消息就一

python thrift搭建服务端和客户端测试程序

本文生动简洁介绍了如何通过python搭建一个服务端和客户端的简单测试程序. 一.简介 thrift是一个软件框架,用来进行可扩展且跨语言的服务的开发.它结合了功能强大的软件堆栈和代码生成引擎,以构建在 C++, Java, Python, PHP, Ruby, Erlang, Perl, Haskell, C#, Cocoa, JavaScript, Node.js, Smalltalk, and OCaml 这些编程语言间无缝结合的.高效的服务. 二.安装 1.下载地址 http://www

python打包成so文件过程解析

这篇文章主要介绍了python打包成so文件过程解析,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下 wget https://bootstrap.pypa.io/get-pip.py python get-pip.py pip install cython 编写setput.py文件: setup.py文件内容如下: from distutils.core import setup from distutils.extension import

python文字转语音实现过程解析

这篇文章主要介绍了python文字转语音实现过程解析,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下 使用百度接口 接口地址 https://ai.baidu.com/docs#/TTS-Online-Python-SDK/top 安装接口 pip install baidu-aip from aip import AipSpeech """ 你的 APPID AK SK """ APP_ID =

python使用rsa非对称加密过程解析

这篇文章主要介绍了python使用rsa非对称加密过程解析,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下 1.安装rsa 支持python 2.7 或者 python 3.5 以上版本 使用豆瓣pypi源来安装rsa pip install -i https://pypi.douban.com/simple rsa 2.加密解密 2.1.生成公私钥对 import rsa # 1.接收者(A)生成512位公私钥对 # a. lemon_pub为

Python测试线程应用程序过程解析

这篇文章主要介绍了Python测试线程应用程序过程解析,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下 在本章中,我们将学习线程应用程序的测试.我们还将了解测试的重要性. 为什么要测试? 在我们深入讨论测试的重要性之前,我们需要知道测试的内容.一般来说,测试是一种了解某些东西是如何运作的技术.另一方面,特别是如果我们谈论计算机程序或软件,那么测试就是访问软件程序功能的技术. 在本节中,我们将讨论软件测试的重要性.在软件开发中,必须在向客户端发布软

python全局变量引用与修改过程解析

这篇文章主要介绍了python全局变量引用与修改过程解析,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下 一.引用 使用到的全局变量只是作为引用,不在函数中修改它的值的话,不需要加global关键字.如: #! /usr/bin/python a = 1 b = [2, 3] def func(): if a == 1: print("a: %d" %a) for i in range(4): if i in b: print(&quo

Python namedtuple命名元组实现过程解析

这篇文章主要介绍了Python namedtuple命名元组实现过程解析,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下 命名元组(namedtuple)是一种带有属性的元组,它们是组合只读数据的很好的方式. 相比一般的元组,构造命名元组需要先导入namedtuple,因为它不在默认的命名空间里.然后通过名字和属性来定义一个命名元组.这会返回一个像类一样的对象,可以进行多次实例化. 命名元组可以被打包.解包以及做所有可以对普通元组做的事,并且还可

python自动化unittest yaml使用过程解析

这篇文章主要介绍了python自动化unittest yaml使用过程解析,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下 在编写unittest自动化用例时,一个请求需要编写多条用例,而涉及的参数基本相同,这时候就会用到配置文件,可以把参数配置项统一管理,避免重复代码,也方便后期维护 此处用到的是yaml,首先需要安装yaml库,pip install yaml 安装成功后,脚本导入语句,import yaml,具体语法可参照如上入门教程 举例