更新,最近在學unix環境編程,了解一下進程的創建過程,用最原始的方式實現了一個ssh命令的執行。
#coding=utf8 '''用python實現了一個簡單的shell,了解進程創建類unix 環境下 fork和exec 兩個系統調用完成進程的創建''' import sys, os def myspawn(cmdline): argv = cmdline.split() if len(argv) == 0: return program_file = argv[0] pid = os.fork() if pid < 0: sys.stderr.write("fork error") elif pid == 0: # child os.execvp(program_file, argv) sys.stderr.write("cannot exec: "+ cmdline) sys.exit(127) # parent pid, status = os.waitpid(pid, 0) ret = status >> 8 # 返回值是一個16位的二進制數字,高8位為退出狀態碼,低8位為程序結束系統信號的編號 signal_num = status & 0x0F sys.stdout.write("ret: %s, signal: %s/n" % (ret, signal_num)) return ret def ssh(host, user, port=22, password=None): if password: sys.stdout.write("password is: '%s' , plz paste it into ssh/n" % (password)) cmdline = "ssh %s@%s -p %s " % (user, host, port) ret = myspawn(cmdline) if __name__ == "__main__": host = '' user = '' password = '' ssh(host, user, password=password)最近在做一個項目,需要在客戶端集成一個交互式ssh功能,大概就是客戶端跟服務器申請個可用的機器,服務端返回個ip,端口,密碼, 然后客戶端就可以直接登錄到機器上操做了。該程序基于paramiko模塊。
經查找,從paramiko的源碼包demos目錄下,可以看到交互式shell的實現,就是那個demo.py。但是用起來有些bug,于是我給修改了一下interactive.py(我把windows的代碼刪掉了,剩下的只能在linux下用)。代碼如下:
#coding=utf-8import socketimport sysimport osimport termiosimport ttyimport fcntlimport signalimport structimport select now_channel = None def interactive_shell(chan): posix_shell(chan) def ioctl_GWINSZ(fd): try: cr = struct.unpack('hh', fcntl.ioctl(fd, termios.TIOCGWINSZ,'aaaa')) except: return return cr def getTerminalSize(): cr = ioctl_GWINSZ(0) or ioctl_GWINSZ(1) or ioctl_GWINSZ(2) return int(cr[1]), int(cr[0]) def resize_pty(signum=0, frame=0): width, height = getTerminalSize() if now_channel is not None: now_channel.resize_pty(width=width, height=height) def posix_shell(chan): global now_channel now_channel = chan resize_pty() signal.signal(signal.SIGWINCH, resize_pty) # 終端大小改變時,修改pty終端大小 stdin = os.fdopen(sys.stdin.fileno(), 'r', 0) # stdin buff置為空,否則粘貼多字節或者按方向鍵的時候顯示不正確 fd = stdin.fileno() oldtty = termios.tcgetattr(fd) newtty = termios.tcgetattr(fd) newtty[3] = newtty[3] | termios.ICANON try: termios.tcsetattr(fd, termios.TCSANOW, newtty) tty.setraw(fd) tty.setcbreak(fd) chan.settimeout(0.0) while True: try: r, w, e = select.select([chan, stdin], [], []) except: # 解決SIGWINCH信號將休眠的select系統調用喚醒引發的系統中斷,忽略中斷重新調用解決。 continue if chan in r: try: x = chan.recv(1024) if len(x) == 0: print 'rn*** EOFrn', break sys.stdout.write(x) sys.stdout.flush() except socket.timeout: pass if stdin in r: x = stdin.read(1) if len(x) == 0: break chan.send(x) finally: termios.tcsetattr(sys.stdin, termios.TCSADRAIN, oldtty)
新聞熱點
疑難解答