Preface
Pexpect 是一個用來啟動子程序並對其進行自動控制的純Python 模塊。Pexpect 可以用來和像ssh、ftp、passwd、telnet 等命令行程序進行自動交互。繼第一部分《探索Pexpect,第1 部分:剖析Pexpect 》介紹了Pexpect 的基礎和如何使用後,本文將結合具體實例入手,詳細介紹Pexpect 的用法和在實際應用中的注意點。
Pexpect 是個純Python語言實現的模塊,使用其可以輕鬆方便的實現與 ssh、ftp、passwd 和 telnet 等程序的自動交互,但是讀者的理解還可能只是停留在理論基礎上,本文將從實際例子入手具體介紹 Pexpect 的使用場景和使用心得體驗,實例中的代碼讀者都可以直接拿來使用,相信會對大家產生比較大的幫助。
例1:ftp 的使用
本例實現瞭如下功能:ftp登錄到develperWorks.ibm.com主機上,並用二進制傳輸模式下載一個名叫rmall的文件:
- 清單1. ftp 的例子代碼
- #!/usr/bin/env python
- import pexpect
- # 即將ftp 所要登錄的遠程主機的域名
- ipAddress = 'develperWorks.ibm.com'
- # 登錄用戶名
- loginName = 'root'
- # 用戶名密碼
- loginPassword = 'passw0rd'
- # 拼湊ftp 命令
- cmd = 'ftp ' + ipAddress
- # 利用ftp 命令作為spawn 類構造函數的參數,生成一個spawn 類的對象
- child = pexpect.spawn(cmd)
- # 期望具有提示輸入用戶名的字符出現
- index = child.expect(["(?i)name", "(?i)Unknown host", pexpect.EOF, pexpect.TIMEOUT])
- # 匹配到了"(?i)name",表明接下來要輸入用戶名
- if ( index == 0 ):
- # 發送登錄用戶名+ 換行符給子程序.
- child.sendline(loginName)
- # 期望"(?i)password" 具有提示輸入密碼的字符出現.
- index = child.expect(["(?i)password", pexpect.EOF, pexpect.TIMEOUT])
- # 匹配到了pexpect.EOF 或pexpect.TIMEOUT,表示超時或者EOF,程序打印提示信息並退出.
- if (index != 0):
- print "ftp login failed"
- child.close(force=True)
- # 匹配到了密碼提示符,發送密碼+ 換行符給子程序.
- child.sendline(loginPassword)
- # 期望登錄成功後,提示符"ftp>" 字符出現.
- index = child.expect( ['ftp>', 'Login incorrect', 'Service not available',
- pexpect.EOF, pexpect.TIMEOUT])
- # 匹配到了'ftp>',登錄成功.
- if (index == 0):
- print 'Congratulations! ftp login correct!'
- # 發送'bin'+ 換行符給子程序,表示接下來使用二進制模式來傳輸文件.
- child.sendline("bin")
- print 'getting a file...'
- # 向子程序發送下載文件rmall 的命令.
- child.sendline("get rmall")
- # 期望下載成功後,出現'Transfer complete.*ftp>',其實下載成功後,
- # 會出現以下類似於以下的提示信息:
- # 200 PORT command successful.
- # 150 Opening data connection for rmall (548 bytes).
- # 226 Transfer complete.
- # 548 bytes received in 0.00019 seconds (2.8e+03 Kbytes/s)
- # 所以直接用正則表達式'.*' 將'Transfer complete' 和提示符'ftp>' 之間的字符全省去.
- index = child.expect( ['Transfer complete.*ftp>', pexpect.EOF, pexpect.TIMEOUT] )
- # 匹配到了pexpect.EOF 或pexpect.TIMEOUT,表示超時或者EOF,程序打印提示信息並退出.
- if (index != 0):
- print "failed to get the file"
- child.close(force=True)
- # 匹配到了'Transfer complete.*ftp>',表明下載文件成功,打印成功信息,並輸入'bye',結束ftp session.
- print 'successfully received the file'
- child.sendline("bye")
- # 用戶名或密碼不對,會先出現'Login incorrect',然後仍會出現'ftp>',但是pexpect 是最小匹配,不是貪婪匹配,
- # 所以如果用戶名或密碼不對,會匹配到'Login incorrect',而不是'ftp>',然後程序打印提示信息並退出.
- elif (index == 1):
- print "You entered an invalid login name or password. Program quits!"
- child.close(force=True)
- # 匹配到了'Service not available',一般表明421 Service not available, remote server has
- # closed connection,程序打印提示信息並退出.
- # 匹配到了pexpect.EOF 或pexpect.TIMEOUT,表示超時或者EOF,程序打印提示信息並退出.
- else:
- print "ftp login failed! index = " + index
- child.close(force=True)
- # 匹配到了"(?i)Unknown host",表示server 地址不對,程序打印提示信息並退出
- elif index == 1 :
- print "ftp login failed, due to unknown host"
- child.close(force=True)
- # 匹配到了pexpect.EOF 或pexpect.TIMEOUT,表示超時或者EOF,程序打印提示信息並退出
- else:
- print "ftp login failed, due to TIMEOUT or EOF"
- child.close(force=True)
Note.
例2:記錄log
本例實現瞭如下功能:運行一個命令,並將該命令的運行輸出結果記錄到 log 文件中 ./command.py [-a] [-c command] {logfilename}. -c 後接的是要運行的命令的名字,默認是 “ls -l”;logfilename 是記錄命令運行結果的 log 文件名,默認是 “command.log”;指定 -a 表示命令的輸出結果會附加在 logfilename 後,如果 logfilename 之前已經存在的話。
- 清單2. 記錄log 的例子代碼
- #!/usr/bin/env python
- """
- This run a user specified command and log its result.
- ./command.py [-a] [-c command] {logfilename}
- logfilename : This is the name of the log file. Default is command.log.
- -a : Append to log file. Default is to overwrite log file.
- -c : spawn command. Default is the command 'ls -l'.
- Example:
- This will execute the command 'pwd' and append to the log named my_session.log:
- ./command.py -a -c 'pwd' my_session.log
- """
- import os, sys, getopt
- import traceback
- import pexpect
- # 如果程序中間出錯,打印提示信息後退出
- def exit_with_usage():
- print globals()['__doc__']
- os._exit(1)
- def main():
- ################################################## ####################
- # Parse the options, arguments, get ready, etc.
- ################################################## ####################
- try:
- optlist, args = getopt.getopt(sys.argv[1:], 'h?ac:', ['help','h','?'])
- # 如果指定的參數不是' -a ' , ' -h ' , ' -c ' , ' -? ' , ' --help ' ,
- #' --h '或' --? '時,會拋出exception,
- # 這裡catch 住,然後打印出exception 的信息,並輸出usage 提示信息.
- except Exception, e:
- print str(e)
- exit_with_usage()
- options = dict(optlist)
- # 最多只能指定一個logfile,否則出錯.
- if len(args) > 1:
- exit_with_usage()
- # 如果指定的是'-h','--h','-?','--?' 或'--help',只輸出usage 提示信息.
- if [elem for elem in options if elem in ['-h','--h','-?','--?','--help']]:
- print "Help:"
- exit_with_usage()
- # 獲取logfile 的名字.
- if len(args) == 1:
- script_filename = args[0]
- else:
- # 如果用戶沒指定,默認logfile 的名字是command.log
- script_filename = "command.log"
- # 如果用戶指定了參數-a,如果之前該logfile 存在,那麼接下來的內容會附加在原先內容之後,
- # 如果之前沒有該logfile,新建一個文件,並且接下來將內容寫入到該文件中.
- if '-a' in options:
- fout = open (script_filename, "ab")
- else:
- # 如果用戶沒指定參數-a,默認按照用戶指定logfile 文件名新建一個文件,然後將接下來將內容寫入到該文件中.
- fout = open (script_filename, "wb")
- # 如果用戶指定了-c 參數,那麼運行用戶指定的命令.
- if '-c' in options:
- command = options['-c']
- # 如果用戶沒有指定-c 參數,那麼默認運行命令'ls – l'
- else:
- command = "ls -l"
- # logfile 文件的title
- fout.write ('==========Log Tile: IBM developerWorks China==========\n')
- # 為接下來的運行命令生成一個pexpect 的spawn 類子程序的對象.
- p = pexpect.spawn(command)
- # 將之前open 的file 對象指定為spawn 類子程序對象的log 文件.
- p.logfile = fout
- # 命令運行完後,expect EOF 出現,這時會將spawn 類子程序對象的輸出寫入到log 文件.
- p.expect(pexpect.EOF)
- #open 完文件,使用完畢後,需關閉該文件.
- fout.close()
- return 0
- if __name__ == "__main__":
- try:
- main()
- except SystemExit, e:
- raise e
- except Exception, e:
- print "ERROR"
- print str(e)
- traceback.print_exc()
- os._exit(1)
* logfile
* logfile_read 和 logfile_send
Note.
- 清單3. log 內容為空的例子代碼
- import pexpect
- p = pexpect.spawn( ' ls -l ' )
- fout = open ('log.txt', "w")
- p.logfile = fout
- fout.close()
例3:ssh 的使用
本例實現瞭如下功能:ssh 登錄到某個用戶指定的主機上,運行某個用戶指定的命令,並輸出該命令的結果.
- 清單4. ssh 的例子代碼
- #!/usr/bin/env python
- """
- This runs a command on a remote host using SSH. At the prompts enter hostname,
- user, password and the command.
- """
- import pexpect
- import getpass, os
- #user: ssh 主機的用戶名
- #host:ssh 主機的域名
- #password:ssh 主機的密碼
- #command:即將在遠端ssh 主機上運行的命令
- def ssh_command (user, host, password, command):
- """
- This runs a command on the remote host. This could also be done with the
- pxssh class, but this demonstrates what that class does at a simpler level.
- This returns a pexpect.spawn object. This handles the case when you try to
- connect to a new host and ssh asks you if you want to accept the public key
- fingerprint and continue connecting.
- """
- ssh_newkey = 'Are you sure you want to continue connecting'
- # 為ssh 命令生成一個spawn 類的子程序對象.
- child = pexpect.spawn('ssh -l %s %s %s'%(user, host, command))
- i = child.expect([pexpect.TIMEOUT, ssh_newkey, 'password: '])
- # 如果登錄超時,打印出錯信息,並退出.
- if i == 0: # Timeout
- print 'ERROR!'
- print 'SSH could not login. Here is what SSH said:'
- print child.before, child.after
- return None
- # 如果ssh 沒有public key,接受它.
- if i == 1: # SSH does not have the public key. Just accept it.
- child.sendline ('yes')
- child.expect ('password: ')
- i = child.expect([pexpect.TIMEOUT, 'password: '])
- if i == 0: # Timeout
- print 'ERROR!'
- print 'SSH could not login. Here is what SSH said:'
- print child.before, child.after
- return None
- # 輸入密碼.
- child.sendline(password)
- return child
- def main ():
- # 獲得用戶指定ssh 主機域名.
- host = raw_input('Hostname: ')
- # 獲得用戶指定ssh 主機用戶名.
- user = raw_input('User: ')
- # 獲得用戶指定ssh 主機密碼.
- password = getpass.getpass()
- # 獲得用戶指定ssh 主機上即將運行的命令.
- command = raw_input('Enter the command: ')
- child = ssh_command (user, host, password, command)
- # 匹配pexpect.EOF
- child.expect(pexpect.EOF)
- # 輸出命令結果.
- print child.before
- if __name__ == '__main__':
- try:
- main()
- except Exception, e:
- print str(e)
- traceback.print_exc()
- os._exit(1)
Note.
例4:pxssh 的使用
本例實現瞭如下功能:使用pexpect 自帶的pxssh 模塊實現ssh 登錄到某個用戶指定的主機上,運行命令' uptime '和' ls -l ',並輸出該命令的結果。
- 清單5. 使用pxssh 的例子代碼
- #!/usr/bin/env python
- import pxssh
- import getpass
- try:
- # 調用構造函數,創建一個pxssh 類的對象.
- s = pxssh.pxssh()
- # 獲得用戶指定ssh 主機域名.
- hostname = raw_input('hostname: ')
- # 獲得用戶指定ssh 主機用戶名.
- username = raw_input('username: ')
- # 獲得用戶指定ssh 主機密碼.
- password = getpass.getpass('password: ')
- # 利用pxssh 類的login 方法進行ssh 登錄,原始prompt 為'$' , '#'或'>'
- s.login (hostname, username, password, original_prompt='[$#>]')
- # 發送命令'uptime'
- s.sendline ('uptime')
- # 匹配prompt
- s.prompt()
- # 將prompt 前所有內容打印出,即命令'uptime' 的執行結果.
- print s.before
- # 發送命令' ls -l '
- s.sendline ('ls -l')
- # 匹配prompt
- s.prompt()
- # 將prompt 前所有內容打印出,即命令' ls -l ' 的執行結果.
- print s.before
- # 退出ssh session
- s.logout()
- except pxssh.ExceptionPxssh, e:
- print "pxssh failed on login."
- print str(e)
pxssh 是pexpect 中spawn 類的子類,增加了login, logout 和prompt 幾個方法,使用其可以輕鬆實現ssh 連接,而不用自己調用相對複雜的pexpect 的方法來實現。pxssh 做了很多tricky 的東西來處理ssh login 過程中所可能遇到的各種情況。比如:如果這個session 是第一次login,pxssh 會自動接受遠程整數remote certificate ;如果你已經設置了公鑰認證,pxssh 將不會再等待password 的提示符。(更多ssh 相關知識,請參閱參考資料) pxssh 使用 shell 的提示符來同步遠程主機的輸出,為了使程序更加穩定,pxssh 還可以設置prompt 為更加唯一的字符串,而不僅僅是“ $ ”和“ # ”。
* login方法
- login (self,server,username,password='',terminal_type='ansi',
- iginal_prompt=r"[#$]",login_timeout=10,port=None,auto_prompt_reset=True)
* prompt方法
- prompt (self, timeout=20)
例5:telnet 的使用
本例實現瞭如下功能:telnet 登錄到某遠程主機上,輸入命令“ls -l”後,將子程序的執行權交還給用戶,用戶可以與生成的telnet 子程序進行交互。
- 清單6. telnet 的例子代碼
- #!/usr/bin/env python
- import pexpect
- # 即將telnet 所要登錄的遠程主機的域名
- ipAddress = 'develperWorks.ibm.com'
- # 登錄用戶名
- loginName = 'root'
- # 用戶名密碼
- loginPassword = 'passw0rd'
- # 提示符,可能是' $ ' , ' # '或' > '
- loginprompt = '[$#>]'
- # 拼湊telnet 命令
- cmd = 'telnet ' + ipAddress
- # 為telnet 生成spawn 類子程序
- child = pexpect.spawn(cmd)
- # 期待'login'字符串出現,從而接下來可以輸入用戶名
- index = child.expect(["login", "(?i)Unknown host", pexpect.EOF, pexpect.TIMEOUT])
- if ( index == 0 ):
- # 匹配'login'字符串成功,輸入用戶名.
- child.sendline(loginName)
- # 期待"[pP]assword" 出現.
- index = child.expect(["[pP]assword", pexpect.EOF, pexpect.TIMEOUT])
- # 匹配"[pP]assword" 字符串成功,輸入密碼.
- child.sendline(loginPassword)
- # 期待提示符出現.
- child.expect(loginprompt)
- if (index == 0):
- # 匹配提示符成功,輸入執行命令'ls -l'
- child.sendline('ls -l')
- # 立馬匹配'ls -l',目的是為了清除剛剛被echo 回顯的命令.
- child.expect('ls -l')
- # 期待提示符出現.
- child.expect(loginprompt)
- # 將'ls -l' 的命令結果輸出.
- print child.before
- print "Script recording started. Type ^] (ASCII 29) to escape from the script
- shell."
- # 將telnet 子程序的執行權交給用戶.
- child.interact()
- print 'Left interactve mode.'
- else:
- # 匹配到了pexpect.EOF 或pexpect.TIMEOUT,表示超時或者EOF,程序打印提示信息並退出.
- print "telnet login failed, due to TIMEOUT or EOF"
- child.close(force=True)
- else:
- # 匹配到了pexpect.EOF 或pexpect.TIMEOUT,表示超時或者EOF,程序打印提示信息並退出.
- print "telnet login failed, due to TIMEOUT or EOF"
- child.close(force=True)
此時程序會block 住,等待用戶的輸入,比如用戶輸入' pwd ',輸出/home/root . 接下來用戶敲入ctrl+] 結束子程序.
* interact 方法
- interact(self, escape_character = chr(29), input_filter = None, output_filter = None)
pexpect 使用 tips
獲得 pexpect.spawn 對象的字符串value 值,將會給debug提供很多有用信息。
- 清單7. 打印 pexpect.spawn 對象的字符串value 值的例子代碼
- try:
- i = child.expect ([pattern1, pattern2, pattern3, etc])
- except:
- print "Exception was thrown"
- print "debug information:"
- print str(child)
- 清單8. 記錄log 的例子代碼
- # 打開loggging 功能並將結果輸出到屏幕上
- child = pexpect.spawn (foo)
- child.logfile = sys.stdout
- 清單9. 重新啟動一個shell 來規避pexpect 對元字符的不解釋
- child = pexpect.spawn('/bin/bash -c "ls -l | grep LOG > log_list.txt"')
- child.expect(pexpect.EOF)
如果子程序沒有在指定的時間內生成任何output,那麼expect() 和read() 都會產生TIMEOUT 異常。超時默認是30s,可以在expect() 和spawn 構造函數初始化時指定為其它時間,如: (如果你想讓expect() 和read() 忽略超時限制,即無限期阻塞住直到有output 產生,設置timeout 參數為None。)
- child.expect('password:', timeout=120) # 等待120s
- child = pexpect.spawn( "telnet develperWorks.ibm.com" )
- child.expect( "login", timeout=None )
有些UNIX 平台,當你讀取一個處於EOF 狀態的文件描述符時,會拋出異常,其他UNIX 平台,卻只會靜靜地返回一個空字符串來表明該文件已經達到了狀態。 pexpect 模塊除了提供 spawn 類以外,還提供了 run() 函數,使用其可以取代一些spawn 的使用,而且更加簡單明了。
- 清單11. 使用run() 來替代spawn 的使用的例子代碼
- # 使用spawn 的例子
- from pexpect import *
- child = spawn('scp foo myname@host.example.com:.')
- child.expect ('(?i)password')
- child.sendline (mypassword)
- from pexpect import *
- run ('scp foo myname@host.example.com:.', events={'(?i)password': mypassword})
- 清單12. 其它一些使用run() 的例子代碼
- # 在local 機器上啟動apache 的daemon
- from pexpect import *
- run ("/usr/local/apache/bin/apachectl start")
- # 使用SVN check in 文件
- from pexpect import *
- run ("svn ci -m 'automatic commit' my_file.py")
- # 運行一個命令並捕獲exit status
- from pexpect import *
- command_output, exitstatus) = run ('ls -l /bin', withexitstatus=1)
- # 運行SSH,並在遠程機器上執行' ls -l ',如果pattern '(?i)password' 被匹配住,密碼'secret'
- # 將會被發送出去
- run ("ssh username@machine.example.com 'ls -l'", events={'(?i)password':'secret\\n'})
- # 啟動mencoder 來rip 一個video,同樣每5s 鐘顯示進度記號
- from pexpect import *
- def print_ticks(d):
- print d['event_count']
- run ("mencoder dvd://1 -o video.avi -oac copy -ovc copy", events={TIMEOUT:print_ticks})
- 清單13. expect_exact() 的例子代碼
- import pexpect
- child = pexpect.spawn('ls -l')
- child.expect_exact('pexpect.txt')
- print child.after
- 清單14. expect() 的最小匹配例子代碼
- # 如果輸入是'foobar'
- index = p.expect (['bar', 'foo', 'foobar'])
- #index 返回是1 ('foo') 而不是2 ('foobar'),即使'foobar' 是個更好的匹配。原因是輸入是個流stream,
- # 當收到foo 時,第1 個pattern ('foo') 就被匹配了,不會等到' bar '的出現了,所以返回 1
- 清單15. 匹配一行結束 1
- child.expect ('\r\n')
- # 成功在一行結束前匹配一個單詞
- child.expect ('\w+\r\n')
- # 以下兩種情況都會失敗
- child.expect ('\w+\n')
- child.expect ('\w+$')
child.expect ('.+'); 因為是最小匹配,所以只會返回一個字符,而不是一個整個一行(雖然pexpect 設置了re.DOTALL,會匹配一個新行)。 child.expect ('.*' ); 每次匹配都會成功,但是總是沒有字符返回,因為'*' 表明前面的字符可以出現0 次, 在pexpect 中,一般來說,任何'*' 都會盡量少的匹配。
測試子程序是否還在運行。這個方法是非阻塞的,如果子程序被終止了,那麼該方法會去讀取子程序的exitstatus 或signalstatus 這兩個域。返回True 表明子程序好像是在運行,返回False 表示不再運行。當平台是Solaris 時,可能需要幾秒鐘才能得到正確的狀態。當子程序退出後立馬執行isalive() 有時可能會返回1 (True),這是一個race condition,原因是子程序已經關閉了其文件描述符,但是在isalive() 執行前還沒有完全的退出。增加一個小小的延時會對isalive() 的結果有效性有幫助。
- 清單17. isalive() 的例子代碼
- # 以下程序有時會返回1 (True)
- child = pexpect.spawn('ls')
- child.expect(pexpect.EOF)
- print child.isalive()
- # 但是如果在isalive() 之前加個小延時,就會一直返回0 (False)
- child = pexpect.spawn('ls')
- child.expect(pexpect.EOF)
- time.sleep(0.1) # 之前要import time,單位是秒 s
- print child.isalive()
- 清單18. delaybeforesend 的例子代碼
- child.expect ('[pP]assword:')
- child.sendline (my_password)
- # 在expect 之後,某些應用程序,如SSH,會做如下動作:
- #1. SSH 打印"password:" 提示符給用戶
- #2. SSH 關閉echo.
- #3. SSH 等待用戶輸入密碼
- # 但是現在第二條語句sendline 可能會發生在1 和2 之間,即在SSH 關掉echo 之前輸入了password 給子程序, 從
- # 而在stdout,該password 被echo 回顯出來,出現了security 的問題
- # 所以此時可以通過設置delaybeforesend 來在將數據寫(發送)給子程序之前增加一點點的小延時,因為該問題經
- # 常出現,所以默認就sleep 50ms. 許多linux 機器必須需要0.03s 以上的delay
- self.delaybeforesend = 0.05 # 單位秒
Supplement
* 探索 Pexpect,第 1 部分:剖析 Pexpect
* IBM Linux developerWork 技巧
沒有留言:
張貼留言