<filename>webterminal/interactive.py
# -*- coding: utf-8 -*-
import socket
from paramiko.py3compat import u
from django.utils.encoding import smart_unicode
import os

try:
    import termios
    import tty
    has_termios = True
except ImportError:
    has_termios = False
    raise Exception('This project does\'t support windows system!')
try:
    import simplejson as json
except ImportError:
    import json
import sys
import time
import codecs
import errno
from django.utils import timezone
from cmdb.models import Log
from adminset.settings import MEDIA_ROOT
import threading
import ast
import traceback

def get_redis_instance():
    from webterminal.asgi import channel_layer
    return channel_layer._connection_list[0]#获取backends数据,即是redis

def mkdir_p(path):
    """
    Pythonic version of "mkdir -p".  Example equivalents::

        >>> mkdir_p('/tmp/test/testing') # Does the same thing as...
        >>> from subprocess import call
        >>> call('mkdir -p /tmp/test/testing')

    .. note:: This doesn't actually call any external commands.
    """
    try:
        os.makedirs(path)
    except OSError as exc:
        if exc.errno == errno.EEXIST:
            pass
        else:
            raise # The original exception
        
def interactive_shell(chan,channel,log_name=None,width=90,height=40):
    if has_termios:
        posix_shell(chan,channel,log_name=log_name,width=width,height=height)
    else:
        sys.exit(1)
       
class CustomeFloatEncoder(json.JSONEncoder):
    def encode(self, obj):
        if isinstance(obj, float):
            return format(obj, '.6f')
        return json.JSONEncoder.encode(self, obj)

def posix_shell(chan,channel,log_name=None,width=90,height=40):
    """实现shell的内容发送，和发送给监控组,记录日志和关闭shell
        chan:ssh的shell
    """
    from webterminal.asgi import channel_layer
    stdout = list()
    begin_time = time.time()
    last_write_time = {'last_activity_time':begin_time}    
    try:
        chan.settimeout(0.0)
        while True:
            try:               
                x = u(chan.recv(1024))
                if len(x) == 0:#没内容就断开
                    channel_layer.send(channel, {'text': json.dumps(['disconnect',smart_unicode('\r\n*** EOF\r\n')]) })
                    break             
                now = time.time()
                delay = now - last_write_time['last_activity_time']
                last_write_time['last_activity_time'] = now                
                if x == "exit\r\n" or x == "logout\r\n" or x == 'logout':#用户输入这几个退出
                    chan.close()
                else:
                    if isinstance(x,unicode):
                        stdout.append([delay,x])#添加unicode格式的内容进列表
                    else:
                        stdout.append([delay,codecs.getincrementaldecoder('UTF-8')('replace').decode(x)])
                if isinstance(x,unicode):
                    channel_layer.send(channel, {'text': json.dumps(['stdout',x]) })
                else:
                    channel_layer.send(channel, {'text': json.dumps(['stdout',smart_unicode(x)]) })#中文转换成unicode
                #send message to monitor group
                if log_name:#{0}-{1}-{2}.json
                    channel_layer.send_group(u'monitor-{0}'.format(log_name.rsplit('/')[1].rsplit('.json')[0]), {'text': json.dumps(['stdout',smart_unicode(x)]) })
            except socket.timeout:
                pass
            except Exception,e:
                print traceback.print_exc()
                channel_layer.send(channel, {'text': json.dumps(['stdout','A bug find,You can report it to me' + smart_unicode(e)]) })

    finally:
        attrs = {
            "version": 1,
            "width": width,#int(subprocess.check_output(['tput', 'cols'])),
            "height": height,#int(subprocess.check_output(['tput', 'lines'])),
            "duration": round(time.time()- begin_time,6),
            "command": os.environ.get('SHELL',None),
            'title':None,
            "env": {
                "TERM": os.environ.get('TERM'),
                "SHELL": os.environ.get('SHELL','sh')
                },
            'stdout':list(map(lambda frame: [round(frame[0], 6), frame[1]], stdout))
            }
        # print log_name,"finaly"
        mkdir_p('/'.join(os.path.join(MEDIA_ROOT,log_name).rsplit('/')[0:-1]))#创建log目录
        with open(os.path.join(MEDIA_ROOT,log_name), "a") as f:
            f.write(json.dumps(attrs, ensure_ascii=True,cls=CustomeFloatEncoder,indent=2))#操作写进本地目录
        
        audit_log=Log.objects.get(channel=channel,log=log_name.rsplit('/')[-1].rsplit('.json')[0])
        audit_log.is_finished = True
        audit_log.end_time = timezone.now()
        audit_log.save()
        #hand ssh terminal exit
        queue = get_redis_instance()
        redis_channel = queue.pubsub()
        queue.publish(channel, json.dumps(['close']))#发布关闭

class SshTerminalThread(threading.Thread):
    """Thread class with a stop() method. The thread itself has to check
    regularly for the stopped() condition."""
    
    def __init__(self,message,chan):
        super(SshTerminalThread, self).__init__()
        self._stop_event = threading.Event()
        self.message = message
        self.queue = self.redis_queue()#订阅的队列
        self.chan = chan #ssh交互式shell
        
    def stop(self):
        self._stop_event.set()#将“Flag”设置为True

    def stopped(self):
        return self._stop_event.is_set()#返回真假
    
    def redis_queue(self):
        redis_instance = get_redis_instance()
        redis_sub = redis_instance.pubsub()#查看订阅发布
        redis_sub.subscribe(self.message.reply_channel.name)#订阅
        return redis_sub
            
    def run(self):
        #fix the first login 1 bug
        first_flag = True
        while (not self._stop_event.is_set()):
            text = self.queue.get_message()
            if text:
                #deserialize data
                if isinstance(text['data'],(str,basestring,unicode)):
                    try:
                        data = ast.literal_eval(text['data'])
                    except Exception,e:
                        data = text['data']
                else:
                    data = text['data']
                
                if isinstance(data,(list,tuple)):
                    if data[0] == 'close':
                        print 'close threading'
                        self.chan.close()#关闭shell
                        self.stop()#关闭线程
                    elif data[0] == 'set_size':
                        self.chan.resize_pty(width=data[3], height=data[4])#设置终端大小
                        break
                    elif data[0] in ['stdin','stdout']:
                        self.chan.send(data[1])
                        
                elif isinstance(data,(int,long)):
                    if data == 1 and first_flag:#第一次发送数据
                        first_flag = False
                    else:
                        self.chan.send(str(data))
                else:
                    try:
                        self.chan.send(str(data))
                    except socket.error:
                        print 'close threading error'
                        self.stop()

class InterActiveShellThread(threading.Thread):
    """初始化"""
    def __init__(self,chan,channel,log_name=None,width=90,height=40):
        super(InterActiveShellThread, self).__init__()
        self.chan = chan
        self.channel = channel
        self.log_name = log_name
        self.width = width
        self.height = height
    
    def run(self):
        interactive_shell(self.chan, self.channel,log_name=self.log_name,width=self.width,height=self.height)
        
<filename>webterminal/interactive.py
# -*- coding: utf-8 -*-
import socket
from paramiko.py3compat import u
from django.utils.encoding import smart_unicode
import os

try:
    import termios
    import tty
    has_termios = True
except ImportError:
    has_termios = False
    raise Exception('This project does\'t support windows system!')
try:
    import simplejson as json
except ImportError:
    import json
import sys
import time
import codecs
import errno
from django.utils import timezone
from cmdb.models import Log
from adminset.settings import MEDIA_ROOT
import threading
import ast
import traceback

def get_redis_instance():
    from webterminal.asgi import channel_layer
    return channel_layer._connection_list[0]#获取backends数据,即是redis

def mkdir_p(path):
    """
    Pythonic version of "mkdir -p".  Example equivalents::

        >>> mkdir_p('/tmp/test/testing') # Does the same thing as...
        >>> from subprocess import call
        >>> call('mkdir -p /tmp/test/testing')

    .. note:: This doesn't actually call any external commands.
    """
    try:
        os.makedirs(path)
    except OSError as exc:
        if exc.errno == errno.EEXIST:
            pass
        else:
            raise # The original exception
        
def interactive_shell(chan,channel,log_name=None,width=90,height=40):
    if has_termios:
        posix_shell(chan,channel,log_name=log_name,width=width,height=height)
    else:
        sys.exit(1)
       
class CustomeFloatEncoder(json.JSONEncoder):
    def encode(self, obj):
        if isinstance(obj, float):
            return format(obj, '.6f')
        return json.JSONEncoder.encode(self, obj)

def posix_shell(chan,channel,log_name=None,width=90,height=40):
    """实现shell的内容发送，和发送给监控组,记录日志和关闭shell
        chan:ssh的shell
    """
    from webterminal.asgi import channel_layer
    stdout = list()
    begin_time = time.time()
    last_write_time = {'last_activity_time':begin_time}    
    try:
        chan.settimeout(0.0)
        while True:
            try:               
                x = u(chan.recv(1024))
                if len(x) == 0:#没内容就断开
                    channel_layer.send(channel, {'text': json.dumps(['disconnect',smart_unicode('\r\n*** EOF\r\n')]) })
                    break             
                now = time.time()
                delay = now - last_write_time['last_activity_time']
                last_write_time['last_activity_time'] = now                
                if x == "exit\r\n" or x == "logout\r\n" or x == 'logout':#用户输入这几个退出
                    chan.close()
                else:
                    if isinstance(x,unicode):
                        stdout.append([delay,x])#添加unicode格式的内容进列表
                    else:
                        stdout.append([delay,codecs.getincrementaldecoder('UTF-8')('replace').decode(x)])
                if isinstance(x,unicode):
                    channel_layer.send(channel, {'text': json.dumps(['stdout',x]) })
                else:
                    channel_layer.send(channel, {'text': json.dumps(['stdout',smart_unicode(x)]) })#中文转换成unicode
                #send message to monitor group
                if log_name:#{0}-{1}-{2}.json
                    channel_layer.send_group(u'monitor-{0}'.format(log_name.rsplit('/')[1].rsplit('.json')[0]), {'text': json.dumps(['stdout',smart_unicode(x)]) })
            except socket.timeout:
                pass
            except Exception,e:
                print traceback.print_exc()
                channel_layer.send(channel, {'text': json.dumps(['stdout','A bug find,You can report it to me' + smart_unicode(e)]) })

    finally:
        attrs = {
            "version": 1,
            "width": width,#int(subprocess.check_output(['tput', 'cols'])),
            "height": height,#int(subprocess.check_output(['tput', 'lines'])),
            "duration": round(time.time()- begin_time,6),
            "command": os.environ.get('SHELL',None),
            'title':None,
            "env": {
                "TERM": os.environ.get('TERM'),
                "SHELL": os.environ.get('SHELL','sh')
                },
            'stdout':list(map(lambda frame: [round(frame[0], 6), frame[1]], stdout))
            }
        # print log_name,"finaly"
        mkdir_p('/'.join(os.path.join(MEDIA_ROOT,log_name).rsplit('/')[0:-1]))#创建log目录
        with open(os.path.join(MEDIA_ROOT,log_name), "a") as f:
            f.write(json.dumps(attrs, ensure_ascii=True,cls=CustomeFloatEncoder,indent=2))#操作写进本地目录
        
        audit_log=Log.objects.get(channel=channel,log=log_name.rsplit('/')[-1].rsplit('.json')[0])
        audit_log.is_finished = True
        audit_log.end_time = timezone.now()
        audit_log.save()
        #hand ssh terminal exit
        queue = get_redis_instance()
        redis_channel = queue.pubsub()
        queue.publish(channel, json.dumps(['close']))#发布关闭

class SshTerminalThread(threading.Thread):
    """Thread class with a stop() method. The thread itself has to check
    regularly for the stopped() condition."""
    
    def __init__(self,message,chan):
        super(SshTerminalThread, self).__init__()
        self._stop_event = threading.Event()
        self.message = message
        self.queue = self.redis_queue()#订阅的队列
        self.chan = chan #ssh交互式shell
        
    def stop(self):
        self._stop_event.set()#将“Flag”设置为True

    def stopped(self):
        return self._stop_event.is_set()#返回真假
    
    def redis_queue(self):
        redis_instance = get_redis_instance()
        redis_sub = redis_instance.pubsub()#查看订阅发布
        redis_sub.subscribe(self.message.reply_channel.name)#订阅
        return redis_sub
            
    def run(self):
        #fix the first login 1 bug
        first_flag = True
        while (not self._stop_event.is_set()):
            text = self.queue.get_message()
            if text:
                #deserialize data
                if isinstance(text['data'],(str,basestring,unicode)):
                    try:
                        data = ast.literal_eval(text['data'])
                    except Exception,e:
                        data = text['data']
                else:
                    data = text['data']
                
                if isinstance(data,(list,tuple)):
                    if data[0] == 'close':
                        print 'close threading'
                        self.chan.close()#关闭shell
                        self.stop()#关闭线程
                    elif data[0] == 'set_size':
                        self.chan.resize_pty(width=data[3], height=data[4])#设置终端大小
                        break
                    elif data[0] in ['stdin','stdout']:
                        self.chan.send(data[1])
                        
                elif isinstance(data,(int,long)):
                    if data == 1 and first_flag:#第一次发送数据
                        first_flag = False
                    else:
                        self.chan.send(str(data))
                else:
                    try:
                        self.chan.send(str(data))
                    except socket.error:
                        print 'close threading error'
                        self.stop()

class InterActiveShellThread(threading.Thread):
    """初始化"""
    def __init__(self,chan,channel,log_name=None,width=90,height=40):
        super(InterActiveShellThread, self).__init__()
        self.chan = chan
        self.channel = channel
        self.log_name = log_name
        self.width = width
        self.height = height
    
    def run(self):
        interactive_shell(self.chan, self.channel,log_name=self.log_name,width=self.width,height=self.height)
        
# -*- coding: utf-8 -*- #获取backends数据,即是redis Pythonic version of "mkdir -p".  Example equivalents::          >>> mkdir_p('/tmp/test/testing') # Does the same thing as...         >>> from subprocess import call         >>> call('mkdir -p /tmp/test/testing')      .. note:: This doesn't actually call any external commands. # The original exception 实现shell的内容发送，和发送给监控组,记录日志和关闭shell         chan:ssh的shell #没内容就断开 #用户输入这几个退出 #添加unicode格式的内容进列表 #中文转换成unicode #send message to monitor group #{0}-{1}-{2}.json #int(subprocess.check_output(['tput', 'cols'])), #int(subprocess.check_output(['tput', 'lines'])), # print log_name,"finaly" #创建log目录 #操作写进本地目录 #hand ssh terminal exit #发布关闭 Thread class with a stop() method. The thread itself has to check     regularly for the stopped() condition. #订阅的队列 #ssh交互式shell #将“Flag”设置为True #返回真假 #查看订阅发布 #订阅 #fix the first login 1 bug #deserialize data #关闭shell #关闭线程 #设置终端大小 #第一次发送数据 初始化