これが状況です。リモートで通信しているウィンドウでサブプロセスをラップしています。
connection_process = subprocess.Popen(r"C:\Windows\System32\cmd.exe", shell=True, stdout=subprocess.PIPE,
stderr=subprocess.PIPE, stdin=subprocess.PIPE)
これは、stdout、stderr、および stdin への書き込みをポーリングする複数のスレッドで実行されます。リモート側は Linux から実行されており、ansi エスケープ シーケンスを含むすべてのデータを送信します。これらのエスケープ シーケンスの詳細については、http://en.wikipedia.org/wiki/ANSI_escape_sequencesをご覧ください。.. 問題は 2 つあります。まず、これらのエスケープ シーケンスをサブプロセスに送信する前に、ウィンドウが使用するものに変換する方法と、その逆の方法が必要です。Windowsでこれを行う良い方法はありますか? 私はコロラマを見てきましたが、それを行う方法の良い例はありますか? 2 番目に、各コマンド ラインの先頭にあるばかげた C:> を含む完全なコマンド プロンプトですべてが表示されるように、サブプロセスを開く必要があります。何らかの理由で、python の popen はそれをあなたに与えません。それは、99% の確率で、人々が望んでいるものであるそれをフィルタリングします... それをフィルタリングしないように popen に指示する方法は何ですか? 参考までに、私のコミュニケーター コードとその使用例を示します。何かを入力すると、C:> プロンプトが表示されます。私の質問を繰り返しますが、
'''
Created on Mar 2, 2013
@author: rweber
'''
import subprocess
import Queue
from Queue import Empty
import threading
class Process_Communicator():
def join(self):
self.te.join()
self.to.join()
self.running = False
self.aggregator.join()
self.ti.join()
def enqueue_in(self):
while self.running and self.p.stdin is not None:
while not self.stdin_queue.empty():
s = self.stdin_queue.get()
self.p.stdin.write(str(s) + '\n\r')
pass
def enqueue_output(self):
if not self.p.stdout or self.p.stdout.closed:
return
out = self.p.stdout
for line in iter(out.readline, b''):
self.qo.put(line)
def enqueue_err(self):
if not self.p.stderr or self.p.stderr.closed:
return
err = self.p.stderr
for line in iter(err.readline, b''):
self.qe.put(line)
def aggregate(self):
while (self.running):
self.update()
self.update()
def update(self):
line = ""
try:
while self.qe.not_empty:
line = self.qe.get_nowait() # or q.get(timeout=.1)
self.unbblocked_err += line
except Empty:
pass
line = ""
try:
while self.qo.not_empty:
line = self.qo.get_nowait() # or q.get(timeout=.1)
self.unbblocked_out += line
except Empty:
pass
while not self.stdin_queue.empty():
s = self.stdin_queue.get()
self.p.stdin.write(str(s) + '\n\r')
def get_stdout(self, clear=True):
ret = self.unbblocked_out
if clear:
self.unbblocked_out = ""
return ret
def has_stdout(self):
ret = self.get_stdout(False)
if ret == '':
return None
else:
return ret
def get_stderr(self, clear=True):
ret = self.unbblocked_out
if clear:
self.unbblocked_out = ""
return ret
def has_stderr(self):
ret = self.get_stdout(False)
if ret == '':
return None
else:
return ret
def __init__(self, subp):
'''This is a simple class that collects and aggregates the
output from a subprocess so that you can more reliably use
the class without having to block for subprocess.communicate.'''
self.p = subp
self.unbblocked_out = ""
self.unbblocked_err = ""
self.running = True
self.qo = Queue.Queue()
self.to = threading.Thread(name="out_read",
target=self.enqueue_output,
args=())
self.to.daemon = True # thread dies with the program
self.to.start()
self.qe = Queue.Queue()
self.te = threading.Thread(name="err_read",
target=self.enqueue_err,
args=())
self.te.daemon = True # thread dies with the program
self.te.start()
self.stdin_queue = Queue.Queue()
self.aggregator = threading.Thread(name="aggregate",
target=self.aggregate,
args=())
self.aggregator.daemon = True # thread dies with the program
self.aggregator.start()
pass
def write_stdin(p,c):
while p.poll() == None:
i = raw_input("send to process:")
if i is not None:
c.stdin_queue.put(i)
p = subprocess.Popen("cmd.exe", shell=True, stdout=subprocess.PIPE,
stderr=subprocess.PIPE, stdin=subprocess.PIPE)
c = Process_Communicator(p)
stdin = threading.Thread(name="write_stdin",
target=write_stdin,
args=(p,c))
stdin.daemon = True # thread dies with the program
stdin.start()
while p.poll() == None:
if c.has_stdout():
print c.get_stdout()
if c.has_stderr():
print c.get_stderr()
c.join()
print "Exit"
さらに良い..これは、私がこれまでに取り組んでいるsshサーバーのコードです..
#!/usr/bin/env python
# Copyright (C) 2003-2007 Robey Pointer
#
# This file is part of paramiko.
#
# Paramiko is free software; you can redistribute it and/or modify it under the
# terms of the GNU Lesser General Public License as published by the Free
# Software Foundation; either version 2.1 of the License, or (at your option)
# any later version.
#
# Paramiko is distrubuted in the hope that it will be useful, but WITHOUT ANY
# WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
# A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more
# details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with Paramiko; if not, write to the Free Software Foundation, Inc.,
# 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA.
import base64
from binascii import hexlify
import os
import socket
import sys
import threading
import traceback
import win32com
import win32net
import win32process
import win32security
import win32api
import paramiko
import ntsecuritycon
import Queue
import subprocess
try:
import colorama
colorama.init()
except:
try:
import tendo.ansiterm
except:
pass
#@todo max number of connections
#this is a map of user's and windows handles for
global user_handles
user_handles = {}
def main():
# setup logging
paramiko.util.log_to_file('demo_server.log')
# @todo: make the server so that it will automatically generate it's own keys
'''
if not os.path.isfile(os.path.join(os.getcwd() , 'test_rsa.key')):
from Crypto import Random
random_generator = Random.new().read
key = paramiko.rsakey.RSA.generate(2048, random_generator)
'''
#@todo: read key from a file first.. if file doesn't exist, make a new key and save it to the file.
host_key = paramiko.RSAKey(filename='test_rsa', password="password")
#host_key = paramiko.DSSKey(filename='test_dss.key')
print 'Read key: ' + hexlify(host_key.get_fingerprint())
run_server(host_key)
class Server (paramiko.ServerInterface):
# 'data' is the output of base64.encodestring(str(key))
# (using the "user_rsa_key" files)
data = 'AAAAB3NzaC1yc2EAAAABIwAAAIEAyO4it3fHlmGZWJaGrfeHOVY7RWO3P9M7hp' + \
'fAu7jJ2d7eothvfeuoRFtJwhUmZDluRdFyhFY/hFAh76PJKGAusIqIQKlkJxMC' + \
'KDqIexkgHAfID/6mqvmnSJf0b5W8v5h2pI/stOSwTQ+pxVhwJ9ctYDhRSlF0iT' + \
'UWT10hcuO4Ks8='
good_pub_key = paramiko.RSAKey(data=base64.decodestring(data))
def __init__(self):
self.event = threading.Event()
'''self.AdjustPrivilege( ntsecuritycon.SE_CHANGE_NOTIFY_NAME )
self.AdjustPrivilege( ntsecuritycon.SE_ASSIGNPRIMARYTOKEN_NAME )
self.AdjustPrivilege( ntsecuritycon.SE_TCB_NAME )
ntsecuritycon.se
def adjust_windows_privilages( self, priv):
flags = ntsecuritycon.TOKEN_ADJUST_PRIVILEGES | ntsecuritycon.TOKEN_QUERY
htoken = win32security.OpenProcessToken(win32api.GetCurrentProcess(), flags)
id = win32security.LookupPrivilegeValue(None, priv)
newPrivileges = [(id, ntsecuritycon.SE_PRIVILEGE_ENABLED)]
win32security.AdjustTokenPrivileges(htoken, 0, newPrivileges)'''
def check_channel_request(self, kind, chanid):
if kind == 'session':
return paramiko.OPEN_SUCCEEDED
return paramiko.OPEN_FAILED_ADMINISTRATIVELY_PROHIBITED
def check_auth_password(self, username, password):
import re
authorized = False
try:
domain = win32api.GetDomainName()
#@todo
except:
import traceback
traceback.print_exc()
print domain
if not authorized:
try:
hUser = win32security.LogonUser ( username,
domain,
password,
win32security.LOGON32_LOGON_INTERACTIVE ,
win32security.LOGON32_PROVIDER_DEFAULT)
print hUser
except win32security.error:
print "Failed"
import traceback
traceback.print_exc()
authorized = False
except:
import traceback
traceback.print_exc()
authorized = False
else:
print "Succeeded"
authorized = True
if not authorized:
try:
hUser = win32security.LogonUser ( username,
domain,
password,
win32security.LOGON32_LOGON_INTERACTIVE ,
win32security.LOGON32_PROVIDER_DEFAULT)
print hUser
except win32security.error:
print "Failed"
import traceback
traceback.print_exc()
authorized = False
except:
import traceback
traceback.print_exc()
authorized = False
else:
print "Succeeded"
authorized = True
if authorized and hUser is not None:
user_handles[username] = hUser
return paramiko.AUTH_SUCCESSFUL
else:
return paramiko.AUTH_FAILED
def check_auth_publickey(self, username, key):
print 'Auth attempt with key: ' + hexlify(key.get_fingerprint())
#todo check the list of users to make sure that there is a user with that credential on the domian
'''
try:
local_users = win32net.NetGroupGetUsers( "localhost", 'none', 0)
domain_controler = win32net.NetGetDCName()
domain_user = win32net.NetUseGetInfo(domain_controler, username, 1)
users = []
users.extend(local_users)
users.append(domain_user)
except:
import traceback
traceback.print_exc()
for windows_user in users:
windows_user_name = str(windows_user['name']).lower()
print windows_user_name
if (windows_user_name == str(username).lower()) and (key == self.good_pub_key):
print "Key was accepted"
return paramiko.AUTH_SUCCESSFUL
else:
print windows_user_name + " != " + str(username).lower()
return paramiko.AUTH_FAILED'''
if (key == self.good_pub_key):
return paramiko.AUTH_SUCCESSFUL
else:
return paramiko.AUTH_FAILED
def get_allowed_auths(self, username):
return 'password,publickey'
def check_channel_shell_request(self, channel):
self.event.set()
return True
def check_channel_pty_request(self, channel, term, width, height, pixelwidth,
pixelheight, modes):
return True
def run_server(host_key):
# now connect
while True:
try:
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
#address = socket.get
sock.bind(('', 22))
except Exception, e:
print '*** Bind failed: ' + str(e)
traceback.print_exc()
sys.exit(1)
try:
#@todo max number of connections
sock.listen(100)
print 'Listening for connection ...'
# always be listening for connections
connections = []
while True:
client, addr = sock.accept()
#@todo add this to a thread that will handle each connection.
new_connection = threading.Thread(name="connection_%s" % str(len(connections)),
target=handle_connection,
args=(client,addr,host_key))
new_connection.setDaemon(True) # thread dies with the program
new_connection.start()
connections.append(new_connection)
print 'Got a connection!'
except Exception, e:
print '*** Listen/accept failed: ' + str(e)
traceback.print_exc()
sys.exit(1)
print 'Got a connection!'
def handle_connection(client, addr, host_key):
try:
t = paramiko.Transport(client)
try:
t.load_server_moduli()
username = t.get_username()
except:
print '(Failed to load moduli -- gex will be unsupported.)'
raise
t.add_server_key(host_key)
server = Server()
try:
t.start_server(server=server)
except paramiko.SSHException, x:
print '*** SSH negotiation failed.'
sys.exit(1)
# wait for auth
chan = t.accept()
if chan is None:
print '*** No channel.'
sys.exit(1)
print 'Authenticated!'
#@todo: spawn a process and while the process is running communicate with it
#open up a channel for stdin
f = chan.makefile('rU')
'''import pywintypes
import win32pipe
import pywintypes
import win32process
import win32con
import win32api
import win32file
import string
import types
security_attributes = pywintypes.SECURITY_ATTRIBUTES()
security_attributes.bInheritHandle = 1
Pyh_stdout_read_end, Pyh_stdout_write_end = win32pipe.CreatePipe(security_attributes, 0)
Pyh_stderr_read_end, Pyh_stderr_write_end = win32pipe.CreatePipe(security_attributes, 0)
Pyh_stdin_read_end, Pyh_stdin_write_end = win32pipe.CreatePipe(security_attributes, 0)'''
connection_process = subprocess.Popen(r"C:\Windows\System32\cmd.exe", shell=True, stdout=subprocess.PIPE,
stderr=subprocess.PIPE, stdin=subprocess.PIPE)
'''creationFlags = win32process.CREATE_NO_WINDOW
startupInfo = win32process.STARTUPINFO()
startupInfo.dwFlags = win32process.STARTF_USESTDHANDLES
startupInfo.hStdInput = Pyh_stdout_read_end
startupInfo.hStdOutput = Pyh_stdout_write_end
startupInfo.hStdError = Pyh_stderr_write_end'''
#PyHANDLE, PyHANDLE, int, int = CreateProcessAsUser(hToken, appName ,
#commandLine , processAttributes , threadAttributes , bInheritHandles , dwCreationFlags , newEnvironment , currentDirectory , startupinfo )
#@todo: use the correct user's handle in creating the process so that it spawns as that user.
#hProcess, hThread, dwProcessId, dwThreadId = win32process.CreateProcessAsUser(user_handles[username], None, r"C:\Windows\System32\cmd.exe", None, None, 1, 0, None,
#hProcess, hThread, dwProcessId, dwThreadId = win32process.CreateProcess(None, r"C:\Windows\System32\cmd.exe", None, None, 1, 0, None,
# None, startupInfo)
#stdout = open(Pyh_stdout_read_end)
#temp = stdout.readline()
#print temp
communicator = Process_Communicator(connection_process)
#communicator = win32_Process_Communicator(Pyh_stdout_read_end,Pyh_stderr_read_end, Pyh_stdin_write_end)
communicator.stdin_queue.put("\r")
#username = f.readline().strip('\r\n')
while not connection_process.poll():
server.event.wait(10)
if not server.event.isSet():
print '*** Client never asked for a shell.'
sys.exit(1)
if communicator.has_stdout() and chan.send_ready():
chan.send(communicator.get_stdout())
if communicator.has_stderr() and chan.send_ready():
#chan.send_stderr(communicator.get_stderr())
chan.send(communicator.get_stderr())
stdin_string = ''
while f.channel.recv_ready():
stdin_string += f.read(1)
#str = f.readline()
translator = colorama.ansitowin32.AnsiToWin32()
translator.write_and_convert(stdin_string)
communicator.stdin_queue.put(stdin_string)
chan.close()
except Exception, e:
print '*** Caught exception: ' + str(e.__class__) + ': ' + str(e)
traceback.print_exc()
try:
t.close()
except:
pass
sys.exit(1)
class Process_Communicator():
def join(self):
self.te.join()
self.to.join()
self.running = False
self.aggregator.join()
self.ti.join()
def enqueue_in(self):
while self.running and self.p.stdin is not None:
while not self.stdin_queue.empty():
s = self.stdin_queue.get()
self.p.stdin.write(str(s) + '\n\r')
pass
def enqueue_output(self):
if not self.p.stdout or self.p.stdout.closed:
return
out = self.p.stdout
for line in iter(out.readline, b''):
self.qo.put(line)
def enqueue_err(self):
if not self.p.stderr or self.p.stderr.closed:
return
err = self.p.stderr
for line in iter(err.readline, b''):
self.qe.put(line)
def aggregate(self):
while (self.running):
self.update()
self.update()
def update(self):
line = ""
try:
while self.qe.not_empty:
line = self.qe.get_nowait() # or q.get(timeout=.1)
self.unbblocked_err += line
except Queue.Empty:
pass
line = ""
try:
while self.qo.not_empty:
line = self.qo.get_nowait() # or q.get(timeout=.1)
self.unbblocked_out += line
except Queue.Empty:
pass
while not self.stdin_queue.empty():
s = self.stdin_queue.get()
self.p.stdin.write(str(s))
def get_stdout(self, clear=True):
ret = self.unbblocked_out
if clear:
self.unbblocked_out = ""
return ret
def has_stdout(self):
ret = self.get_stdout(False)
if ret == '':
return None
else:
return ret
def get_stderr(self, clear=True):
ret = self.unbblocked_out
if clear:
self.unbblocked_out = ""
return ret
def has_stderr(self):
ret = self.get_stdout(False)
if ret == '':
return None
else:
return ret
def __init__(self, subp):
'''This is a simple class that collects and aggregates the
output from a subprocess so that you can more reliably use
the class without having to block for subprocess.communicate.'''
self.p = subp
self.unbblocked_out = ""
self.unbblocked_err = ""
self.running = True
self.qo = Queue.Queue()
self.to = threading.Thread(name="out_read",
target=self.enqueue_output,
args=())
self.to.daemon = True # thread dies with the program
self.to.start()
self.qe = Queue.Queue()
self.te = threading.Thread(name="err_read",
target=self.enqueue_err,
args=())
self.te.daemon = True # thread dies with the program
self.te.start()
self.stdin_queue = Queue.Queue()
self.aggregator = threading.Thread(name="aggregate",
target=self.aggregate,
args=())
self.aggregator.daemon = True # thread dies with the program
self.aggregator.start()
pass
class win32_Process_Communicator():
def __init__(self,read_end_of_stdout, read_end_of_stderr, write_end_of_stdin):
pass
main()