diff -Naur ../pexpect-2.3/pexpect.py ./pexpect.py--- ../pexpect-2.3/pexpect.py 2007-12-26 18:40:52.000000000 -0800+++ ./pexpect.py 2009-05-20 17:18:58.000000000 -0700@@ -84,7 +84,7 @@ A critical module was not found. Probably this operating system does not support it. Pexpect is intended for UNIX-like operating systems.""")-__version__ = '2.3'+__version__ = '2.3.p2' __revision__ = '$Revision: 399 $' __all__ = ['ExceptionPexpect', 'EOF', 'TIMEOUT', 'spawn', 'run', 'which', 'split_command_line', '__version__', '__revision__']@@ -480,13 +480,16 @@ s.append('delayafterterminate: ' + str(self.delayafterterminate)) return '\n'.join(s)- def _spawn(self,command,args=[]):+ def _spawn(self, command, args=None): """This starts the given command in a child process. This does all the fork/exec type of stuff for a pty. This is called by __init__. If args is empty then command will be parsed (split on spaces) and args will be set to parsed arguments. """+ if args is None:+ args = []+ # The pid and child_fd of this object get set by this method. # Note that it is difficult for this method to fail. # You cannot detect if the child process cannot start.@@ -497,10 +500,13 @@ # that performs some task; creates no stdout output; and then dies. # If command is an int type then it may represent a file descriptor.- if type(command) == type(0):- raise ExceptionPexpect ('Command is an int type. If this is a file descriptor then maybe you want to use fdpexpect.fdspawn which takes an existing file descriptor instead of a command string.')+ if type(command) == int:+ raise ExceptionPexpect ('Command is an int type. If this is a file '+ 'descriptor then maybe you want to use fdpexpect.fdspawn which '+ 'takes an existing file descriptor instead of a command '+ 'string.')- if type (args) != type([]):+ if type (args) != list: raise TypeError ('The argument, args, must be a list.') if args == []:@@ -1218,16 +1224,26 @@ compiled_pattern_list = [] for p in patterns: if type(p) in types.StringTypes:- compiled_pattern_list.append(re.compile(p, compile_flags))+ try:+ compiled_pattern_list.append(re.compile(p, compile_flags))+ except Exception, err:+ raise Exception("""Error encountered when compiling regexp:+%s+Message:+%s""" % (p, err) ) elif p is EOF: compiled_pattern_list.append(EOF) elif p is TIMEOUT: compiled_pattern_list.append(TIMEOUT)- elif type(p) is type(re.compile('')):+ elif hasattr(p, "search") and hasattr(p, "pattern"): compiled_pattern_list.append(p) else:- raise TypeError ('Argument must be one of StringTypes, EOF, TIMEOUT, SRE_Pattern, or a list of those type. %s' % str(type(p)))+ raise TypeError ('Argument must be one of StringTypes, EOF, '+ 'TIMEOUT, SRE_Pattern, or a list of those type. %s' % \+ str(type(p)))+ for pat in compiled_pattern_list:+ print "Pattern: %s" % str(pat) return compiled_pattern_list def expect(self, pattern, timeout = -1, searchwindowsize=None):diff -Naur ../pexpect-2.3/pxsh.py ./pxsh.py--- ../pexpect-2.3/pxsh.py 1969-12-31 16:00:00.000000000 -0800+++ ./pxsh.py 2009-05-13 23:45:07.000000000 -0700@@ -0,0 +1,204 @@+""" This class extends pexpect.spawn to provide a base for non-login based+shells. This adds methods for expecting and matching a shell prompt.++$Id: pxsh.py 487 2008-11-21 22:33:29Z gcooper $+"""++from pexpect import spawn, TIMEOUT+import pexpect, re, time++__all__ = ['ExceptionPxsh', 'levenshtein_distance', 'pxsh']++# Exception classes used by this module.+class ExceptionPxsh(pexpect.ExceptionPexpect):+ """Raised for pxsh exceptions. """++class pxsh(spawn):++ """+ pxsh uses the shell prompt to synchronize output from the local machine. In+ order to make this more robust it sets the shell prompt to something more+ unique than just $ or #. This should work on most Bourne/Bash or Csh style+ shells.++ Example that runs a few commands on a local machine and prints the result:++ import pxsh++ s = None+ try: + s = pxsh.pxsh()+ except pxsh.ExceptionPxsh, e:+ print "pxsh failed to spawn a shell."+ raise++ for cmd in ['uptime', 'ls -l', 'df']:+ s.sendline (cmd) # run a command+ s.prompt() # match the prompt+ print s.before # print everything before the prompt.++ """++ def __init__ (self, cmd="/bin/sh", timeout=10, maxread=2000,+ searchwindowsize=None, logfile=None, cwd=None, env=None):++ self.cmd_is_spawned = False ++ spawn.__init__(self, cmd, timeout=timeout, maxread=maxread,+ searchwindowsize=searchwindowsize, logfile=logfile, cwd=cwd,+ env=env)++ self.name = '<pxsh>'++ # Used to match the command-line prompt++ self.set_prompt = "[PEXPECT]$ "+ self.expected_prompt = re.escape(self.set_prompt)++ def configure_shell (self, auto_prompt_reset=True, timeout=15):++ """ Detect the shell by first attempting to sync the prompt, and+ second by setting up a valid `unique' prompt. """++ if not pxsh.synch_original_prompt(self):+ self.close()+ raise ExceptionPxsh('Could not synchronize with original prompt')++ # We appear to be logged in. Set the shell prompt to something unique.+ if auto_prompt_reset:+ if not pxsh.set_unique_prompt(self, timeout=timeout):+ self.close()+ raise ExceptionPxsh("\n".join(['Could not set shell prompt',+ self.before]))++ def synch_original_prompt (self):++ """ This attempts to find the prompt. Basically, press enter and+ record the response; press enter again and record the response; if the+ two responses are similar then assume we are at the original+ prompt. """++ # All of these timing pace values are magic.+ # I came up with these based on what seemed reliable for+ # connecting to a heavily loaded machine I have.+ # If latency is worse than these values then this will fail.++ # GAS: Clear out the cache before getting the prompt+ try:+ # If the buffer is empty (couldn't find prompt on login, etc)+ # this will toss an Exception.+ self.sendline()+ self.sendline()+ self.read_nonblocking(size=10000, timeout=1)+ except pexpect.TIMEOUT:+ pass+ time.sleep(0.1)+ self.sendline()+ time.sleep(0.5)+ self.read_nonblocking(size=1000, timeout=1)+ time.sleep(0.1)+ self.sendline()+ time.sleep(0.5)+ a = self.read_nonblocking(size=1000, timeout=1)+ time.sleep(0.1)+ self.sendline()+ time.sleep(0.5)+ b = self.read_nonblocking(size=1000, timeout=1)+ ld = levenshtein_distance(a, b)+ len_a = len(a)+ if len_a == 0:+ return False+ if float(ld) / len_a < 0.4:+ return True+ return False++ def prompt (self, timeout=20):++ """This matches the shell prompt. This is little more than a short-cut+ to the expect() method. This returns True if the shell prompt was+ matched. This returns False if there was a timeout. Note that if you+ called login() with auto_prompt_reset set to False then you should have+ manually set the PROMPT attribute to a regex pattern for matching the+ prompt. """++ if self.expect([self.expected_prompt, TIMEOUT], timeout=timeout) == 0:+ # The prompt was matched.+ return True+ # A timeout occurred.+ return False++ def set_unique_prompt (self, timeout=15):++ """ This sets the remote prompt to something more unique than # or $.+ This makes it easier for the prompt() method to match the shell prompt+ unambiguously. This method is called automatically by the login()+ method, but you may want to call it manually if you somehow reset the+ shell prompt. For example, if you 'su' to a different user then you+ will need to manually reset the prompt. This sends shell commands to+ the remote host to set the prompt, so this assumes the remote host is+ ready to receive commands.++ Alternatively, you may use your own prompt pattern. Just set the PROMPT+ attribute to a regular expression that matches it. In this case you+ should call login() with auto_prompt_reset=False; then set the PROMPT+ attribute. After that the prompt() method will try to match your prompt+ pattern. """++ # Why echoing both variables? Because ?t?csh uses $shell and bourne+ # shells use $SHELL..++ self.sendline ("unset PROMPT_COMMAND; echo $SHELL\necho $shell")++ i = self.expect(["csh", "[akz]*sh", TIMEOUT], timeout=timeout)++ if i is 0:+ # ?t?csh+ self.sendline("set prompt='%s'" % self.set_prompt)+ elif i is 1:+ # Bourne shell compatible (this includes, ash, bash, ksh, sh, and+ # zsh.+ self.sendline("export PS1='%s'" % self.set_prompt)+ else:+ # Are you insane?!?! Running php or tclsh as a backend shell (or+ # something else unsupported like psh)????+ raise ExceptionPxsh("Unknown shell; before: %s" % self.before)++ try:+ # Clear the buffer+ self.expect([ TIMEOUT ], timeout=0)+ except:+ pass++ return pxsh.prompt(self, timeout=timeout)++ def _spawn(self, cmd, args=None):++ """ Execute a command via spawn. Don't execute a command if one is+ already executed. """++ if self.cmd_is_spawned is False:+ spawn._spawn(self, cmd, args=args)+ self.cmd_is_spawned = True+ else:+ raise ExceptionPxsh("Command already spawned.")++def levenshtein_distance (a, b):++ """ This calculates the Levenshtein distance between a and b. """++ n, m = len(a), len(b)+ if n > m:+ a, b = b, a+ n, m = m, n+ current = range(n+1)+ for i in range(1, m+1):+ previous, current = current, [i]+[0]*n+ for j in range(1, n+1):+ add, delete = previous[j]+1, current[j-1]+1+ change = previous[j-1]+ if a[j-1] != b[i-1]:+ change += 1+ current[j] = min(add, delete, change)+ return current[n]++# vi:ts=4:sw=4:expandtab:ft=python:diff -Naur ../pexpect-2.3/pxssh.py ./pxssh.py--- ../pexpect-2.3/pxssh.py 2007-08-29 15:33:30.000000000 -0700+++ ./pxssh.py 2009-05-13 23:45:09.000000000 -0700@@ -1,32 +1,33 @@-"""This class extends pexpect.spawn to specialize setting up SSH connections.-This adds methods for login, logout, and expecting the shell prompt.+"""This class extends pexpect.pxsh to specialize setting up SSH connections.+This adds methods for logging in and logging out of SSH connections. $Id: pxssh.py 487 2007-08-29 22:33:29Z noah $ """-from pexpect import *-import pexpect-import time+from getpass import getuser+from pexpect import EOF, ExceptionPexpect, TIMEOUT+from pxsh import pxsh+import sys __all__ = ['ExceptionPxssh', 'pxssh'] # Exception classes used by this module. class ExceptionPxssh(ExceptionPexpect):- """Raised for pxssh exceptions.- """+ """Raised for pxssh exceptions. """+ pass-class pxssh (spawn):+class pxssh (pxsh):- """This class extends pexpect.spawn to specialize setting up SSH- connections. This adds methods for login, logout, and expecting the shell- prompt. It does various tricky things to handle many situations in the SSH+ """This class extends pexpect.pxsh to specialize setting up SSH+ connections. This adds methods for logging in and logging out of SSH+ connections. It does various tricky things to handle many situations in the login process. For example, if the session is your first login, then pxssh automatically accepts the remote certificate; or if you have public key authentication setup then pxssh won't wait for the password prompt. pxssh uses the shell prompt to synchronize output from the remote host. In order to make this more robust it sets the shell prompt to something more- unique than just $ or #. This should work on most Borne/Bash or Csh style+ unique than just $ or #. This should work on most Bourne/Bash or Csh style shells. Example that runs a few commands on a remote server and prints the result::@@ -70,93 +71,44 @@ s.login (hostname, username, password) """- def __init__ (self, timeout=30, maxread=2000, searchwindowsize=None, logfile=None, cwd=None, env=None):- spawn.__init__(self, None, timeout=timeout, maxread=maxread, searchwindowsize=searchwindowsize, logfile=logfile, cwd=cwd, env=env)+ def __init__ (self, timeout=30, maxread=2000, searchwindowsize=None,+ logfile=None, cwd=None, env=None, ssh_opts=""):- self.name = '<pxssh>'+ pxsh.__init__(self, cmd=None, timeout=timeout, maxread=maxread,+ searchwindowsize=searchwindowsize, logfile=logfile, cwd=cwd,+ env=env)- #SUBTLE HACK ALERT! Note that the command to set the prompt uses a- #slightly different string than the regular expression to match it. This- #is because when you set the prompt the command will echo back, but we- #don't want to match the echoed command. So if we make the set command- #slightly different than the regex we eliminate the problem. To make the- #set command different we add a backslash in front of $. The $ doesn't- #need to be escaped, but it doesn't hurt and serves to make the set- #prompt command different than the regex.-- # used to match the command-line prompt- self.UNIQUE_PROMPT = "\[PEXPECT\][\$\#] "- self.PROMPT = self.UNIQUE_PROMPT-- # used to set shell command-line prompt to UNIQUE_PROMPT.- self.PROMPT_SET_SH = "PS1='[PEXPECT]\$ '"- self.PROMPT_SET_CSH = "set prompt='[PEXPECT]\$ '"- self.SSH_OPTS = "-o'RSAAuthentication=no' -o 'PubkeyAuthentication=no'"- # Disabling X11 forwarding gets rid of the annoying SSH_ASKPASS from- # displaying a GUI password dialog. I have not figured out how to- # disable only SSH_ASKPASS without also disabling X11 forwarding.- # Unsetting SSH_ASKPASS on the remote side doesn't disable it! Annoying!- #self.SSH_OPTS = "-x -o'RSAAuthentication=no' -o 'PubkeyAuthentication=no'"- self.force_password = False- self.auto_prompt_reset = True+ self.name = '<pxssh>'- def levenshtein_distance(self, a,b):+ self.user_ssh_opts = ''- """This calculates the Levenshtein distance between a and b.- """+ if ssh_opts:+ self.user_ssh_opts = ssh_opts++ # Disable all keys, X11 agents, and pubkey auth. This is for+ # noisome authentication situations that pexpect can't handle out of+ # the box.+ self.FORCE_PASSWORD_SSH_OPTS = (" -o 'RSAAuthentication=no' "+ "-o 'ForwardAgent=no' -o 'PubkeyAuthentication=no'")++ self.force_password = False- n, m = len(a), len(b)- if n > m:- a,b = b,a- n,m = m,n- current = range(n+1)- for i in range(1,m+1):- previous, current = current, [i]+[0]*n- for j in range(1,n+1):- add, delete = previous[j]+1, current[j-1]+1- change = previous[j-1]- if a[j-1] != b[i-1]:- change = change + 1- current[j] = min(add, delete, change)- return current[n]+ # A flag to ensure that we don't use logout or any login related+ # methods before they should be used. So, basically protecting the user+ # against themselves.+ self.logged_in = False def synch_original_prompt (self):- """This attempts to find the prompt. Basically, press enter and record- the response; press enter again and record the response; if the two- responses are similar then assume we are at the original prompt. """-- # All of these timing pace values are magic.- # I came up with these based on what seemed reliable for- # connecting to a heavily loaded machine I have.- # If latency is worse than these values then this will fail.-- self.read_nonblocking(size=10000,timeout=1) # GAS: Clear out the cache before getting the prompt- time.sleep(0.1)- self.sendline()- time.sleep(0.5)- x = self.read_nonblocking(size=1000,timeout=1)- time.sleep(0.1)- self.sendline()- time.sleep(0.5)- a = self.read_nonblocking(size=1000,timeout=1)- time.sleep(0.1)- self.sendline()- time.sleep(0.5)- b = self.read_nonblocking(size=1000,timeout=1)- ld = self.levenshtein_distance(a,b)- len_a = len(a)- if len_a == 0:- return False- if float(ld)/len_a < 0.4:- return True- return False-- ### TODO: This is getting messy and I'm pretty sure this isn't perfect.- ### TODO: I need to draw a flow chart for this.- def login (self,server,username,password='',terminal_type='ansi',original_prompt=r"[#$]",login_timeout=10,port=None,auto_prompt_reset=True):+ """ Synchronizes the prompt using pxsh.synch_original_prompt(..) """++ return pxsh.synch_original_prompt(self)++ def login (self, server, username=getuser(), password='', + terminal_type='ansi', original_prompt=r"[#\$]", login_timeout=10,+ port=22, auto_prompt_reset=True):- """This logs the user into the given server. It uses the+ """ This logs the user into the given server. It uses the 'original_prompt' to try to find the prompt right after login. When it finds the prompt it immediately tries to reset the prompt to something more easily matched. The default 'original_prompt' is very optimistic@@ -170,100 +122,305 @@ prompt was so weird that we could not match it, so we use a few tricks to guess when we have reached the prompt. Then we hope for the best and blindly try to reset the prompt to something more unique. If that fails- then login() raises an ExceptionPxssh exception.-+ then login() raises either an ExceptionPxsh or ExceptionPxssh+ exception.+ In some situations it is not possible or desirable to reset the original prompt. In this case, set 'auto_prompt_reset' to False to- inhibit setting the prompt to the UNIQUE_PROMPT. Remember that pxssh+ inhibit setting the prompt to the pxsh.set_prompt. Remember that pxssh uses a unique prompt in the prompt() method. If the original prompt is not reset then this will disable the prompt() method unless you- manually set the PROMPT attribute. """+ manually set the 'original_prompt' keyed value in this method, or set+ the 'expected_prompt' value in the object itself to an appropriate+ value.++ There isn't a space behind the default original_prompt value, because+ this breaks colorized logins (I know -- who does colorized logins+ nowadays?).++ Please see the comment in login() regarding TIMEOUT's if you get stuck+ with whacky prompt / login issues.+ """++ login_timeout = int(login_timeout)++ if login_timeout < 0:+ #+ # We want to get people into the habit of specifying integer+ # numbers for login_timeouts to increase robustness on error+ # checking.+ #+ # Sidenote: A value < 0 will print out:+ #+ # command-line line (int): invalid time value.+ #+ raise ExceptionPxssh("Login time must be an integer value greater "+ "than 0")++ if self.logged_in:+ raise ExceptionPxssh("Already logged in; call logout method first")++ ssh_options = "".join([self.user_ssh_opts, ' ',+ "-o NumberOfPasswordPrompts=1"])- ssh_options = '-q' if self.force_password:- ssh_options = ssh_options + ' ' + self.SSH_OPTS- if port is not None:- ssh_options = ssh_options + ' -p %s'%(str(port))- cmd = "ssh %s -l %s %s" % (ssh_options, username, server)+ ssh_options = " ".join([ssh_options, self.FORCE_PASSWORD_SSH_OPTS])++ if port:++ port = int(port)++ if port < 1 or 65536 < port:+ raise TypeError('Port number must be in the range: 1-65536.')+ ssh_options += ' -p %d' % port++ cmd = "ssh %s -o ConnectTimeout=%d -l %s %s" % \+ (ssh_options, login_timeout, username, server)++ sys.stderr.write("Will connect using `%s'\n" % cmd) # This does not distinguish between a remote server 'password' prompt # and a local ssh 'passphrase' prompt (for unlocking a private key).- spawn._spawn(self, cmd)- i = self.expect(["(?i)are you sure you want to continue connecting", original_prompt, "(?i)(?:password)|(?:passphrase for key)", "(?i)permission denied", "(?i)terminal type", TIMEOUT, "(?i)connection closed by remote host"], timeout=login_timeout)+ pxsh._spawn(self, cmd)- # First phase- if i==0: - # New certificate -- always accept it.- # This is what you get if SSH does not have the remote host's- # public key stored in the 'known_hosts' cache.- self.sendline("yes")- i = self.expect(["(?i)are you sure you want to continue connecting", original_prompt, "(?i)(?:password)|(?:passphrase for key)", "(?i)permission denied", "(?i)terminal type", TIMEOUT])- if i==2: # password or passphrase- self.sendline(password)- i = self.expect(["(?i)are you sure you want to continue connecting", original_prompt, "(?i)(?:password)|(?:passphrase for key)", "(?i)permission denied", "(?i)terminal type", TIMEOUT])- if i==4:- self.sendline(terminal_type)- i = self.expect(["(?i)are you sure you want to continue connecting", original_prompt, "(?i)(?:password)|(?:passphrase for key)", "(?i)permission denied", "(?i)terminal type", TIMEOUT])-- # Second phase- if i==0:- # This is weird. This should not happen twice in a row.- self.close()- raise ExceptionPxssh ('Weird error. Got "are you sure" prompt twice.')- elif i==1: # can occur if you have a public key pair set to authenticate. - ### TODO: May NOT be OK if expect() got tricked and matched a false prompt.- pass- elif i==2: # password prompt again- # For incorrect passwords, some ssh servers will- # ask for the password again, others return 'denied' right away.- # If we get the password prompt again then this means- # we didn't get the password right the first time. - self.close()- raise ExceptionPxssh ('password refused')- elif i==3: # permission denied -- password was bad.- self.close()- raise ExceptionPxssh ('permission denied')- elif i==4: # terminal type again? WTF?- self.close()- raise ExceptionPxssh ('Weird error. Got "terminal type" prompt twice.')- elif i==5: # Timeout- #This is tricky... I presume that we are at the command-line prompt.- #It may be that the shell prompt was so weird that we couldn't match- #it. Or it may be that we couldn't log in for some other reason. I- #can't be sure, but it's safe to guess that we did login because if- #I presume wrong and we are not logged in then this should be caught- #later when I try to set the shell prompt.- pass- elif i==6: # Connection closed by remote host- self.close()- raise ExceptionPxssh ('connection closed')- else: # Unexpected - self.close()- raise ExceptionPxssh ('unexpected login response')- if not self.synch_original_prompt():+ # + # 0 : new certificate (host not in known_hosts)+ # 1 : login prompt match (we're done, except for state 4, maybe).+ # 2 : password prompt+ # 3 : permission denied (bad password or login not allowed -- this can+ # be a null password or root login, if appropriate settings aren't+ # configured in sshd_config)+ # 4 : terminal type (after login -- printed out by ncurses apps when+ # determining what terminal to emulate)+ # 5 : TIMEOUT (expression not matched) - let it pass.+ #+ # THIS IS NOT A BUG BUT A FEATURE.+ #+ # Master Noah's comment in the original code about TIMEOUT (which I+ # _completely_ agree with after 2 engineers hit their head against a+ # brick wall trying to solve prompt matching issues):+ #+ # This is tricky... I presume that we are at the command-line prompt.+ # It may be that the shell prompt was so weird that we couldn't match+ # it. Or it may be that we couldn't log in for some other reason. I+ # can't be sure, but it's safe to guess that we did login because if+ # I presume wrong and we are not logged in then this should be caught+ # later when I try to set the shell prompt.+ #+ # Signed,+ # Padawan Garrett+ #+ # 6 : connection closed by remote host (notable mostly for timeout+ # based autologout functionality with [t]csh and shells that exit+ # immediately because of shlib issues, e.g. bash depends on iconv+ # and iconv is broken after an OS upgrade, etc, or simply user is+ # setup with /sbin/nologin or their password has expired).+ # 7 : couldn't resolve DNS host+ # 8 : Connection refused by remote host (server not listening / invalid+ # server specified).+ # 9 : Bad option passed to ssh_options in __init__.+ #++ login_tuples = \+ [ "(?i)are you sure you want to continue connecting",+ original_prompt, "(?i)(?:password)|(?:passphrase for key)",+ "(?i)permission denied", "(?i)terminal type",+ "(?i)connection closed by remote host",+ "(?i)connection timed out",+ ("(?i)(?:could not resolve)|(?:couldn't connect to:)|"+ "(?:connection refused)|(?:network is unreachable)|"+ "(?:could not connect to server)"),+ "(?i)(?:bad configuration option:)",+ TIMEOUT ]++ # An exception message.+ exc_msg = None+ # Tries to go through the FSM.+ stage = 0++ while stage < 4 and exc_msg is None:++ # We should run through these steps a max of 4 times. Why?+ # Because we want to ensure we hit all cases, as they can be hit+ # variably depending on the env setup, in particular whether or not+ # the host is in the known_hosts file, the host allows for+ # passwordless logins, etc.+ #+ # The longest path is:+ #+ # 1. Host not in known_hosts (send yes).+ # 2. Password prompt (send password).+ # 3. Terminal not known (send terminal type).+ # 4. Login prompt++ #+ # Debug messages:+ #+ #if self.before is not None:+ # print "DEBUG: BEFORE -- before it all began...\n", self.before++ #if self.after is not None:+ # print "DEBUG: AFTER -- after it all fell down...\n", self.after++ # Wait login_timeout + 2 seconds (fudge time).+ i = self.expect(login_tuples, timeout=(login_timeout + 2) )++ if i == 0:+ # New certificate -- always accept it.+ #+ # This is what you get if SSH does not have the remote host's+ # public key stored in the `known_hosts' cache.+ self.sendline("yes")++ elif i == 1: + # Found a prompt.+ #+ # May NOT be OK if expect() got tricked and matches a+ # false prompt or a motd item. This shouldn't be a problem in+ # the grand scheme of things, because it'll be caught later on+ # when syncing the prompt.+ #+ break++ elif i == 2:+ # Password or passphrase prompt.+ self.sendline(password)++ elif i == 3:+ # Permission denied+ #+ # Backstory:+ #+ # For incorrect passwords, some ssh servers will ask for the+ # password again, others return 'denied' right away. If we get+ # the password prompt again then this means we didn't get the+ # password right the first time.+ #+ # We shouldn't hit this more than once though because we set:+ # -o 'NumberOfPasswordPrompts=1'+ #+ # Also, if you fail to login with a pubkey, or any other kind+ # of auth method, because someone fubars the options passed+ # into ssh(1), it'll immediately fire back this message.+ #+ # Example:+ #+ # test@headless-horseman ~ $ ssh -o PubkeyAuthentication=no \+ # -o PreferredAuthentications=publickey test@0.0.0.0+ # Permission denied (publickey,keyboard-interactive).+ #+ # IMPORTANT: MAKE SURE TO READ THROUGH THIS BEFORE REPORTING A+ # BUG.+ #+ # Some versions of PAM released on different Unix versions+ # (*cough* Gentoo Linux) are by default extremely paranoid and+ # if permissions are screwed up, passkey auth WILL fail --+ # horribly. If you do have root access or can read+ # /var/log/messages, look for authentication errors reported by+ # PAM relating along the lines of:+ #+ # pam_unix(sshd:auth): authentication failure; logname= uid=0 \+ # euid=0 tty=ssh ruser= rhost=127.0.0.1 user={some username}+ #+ # So, this issue is an infrastructure problem that needs to be+ # properly worked around.+ #++ # print "BEFORE --", self.before, "AFTER --", self.after++ exc_msg = "Permission denied"++ elif i == 4:+ # Ok, so we need to make sure that the remote machine+ # understands what terminal we're dealing with. This can be+ # something of the flavor, rxt, vt100, etc.++ self.sendline(terminal_type)++ elif i == 5:+ # Connection closed by remote host (bad terminal setup,+ # /sbin/nologin set, etc).++ exc_msg = ('Connection closed (does remote login function for '+ 'user?)')++ elif i == 6:+ # Host can't be resolved via DNS or bad IP addr specified.++ exc_msg = "\n".join(["Couldn't resolve requested host",+ self.before])++ elif i == 7:+ # Couldn't connect to server (wrong port, not listening, network+ # down, bad hostname, etc).++ exc_msg = "\n".join([ "Could not connect to server",+ self.before ])++ elif i == 8:+ # Bad config option.++ exc_msg = "\n".join([ "Bad configuration option",+ self.before ])++ else:+ # Timeout -- see above comment.+ break++ stage += 1++ if exc_msg is not None:+ # Houston, we have a problem. Let's close out and raise an+ # exception. self.close()- raise ExceptionPxssh ('could not synchronize with original prompt')- # We appear to be in.- # set shell prompt to something unique.- if auto_prompt_reset:- if not self.set_unique_prompt():- self.close()- raise ExceptionPxssh ('could not set shell prompt\n'+self.before)- return True+ raise ExceptionPxssh(exc_msg)- def logout (self):+ # Let's assume we're logged in.+ pxsh.configure_shell(self, auto_prompt_reset=auto_prompt_reset)++ self.logged_in = True++ # sys.stderr.write("DEBUG (pxssh.before):\n" + self.before)++ return self.logged_in+ def logout (self): """This sends exit to the remote shell. If there are stopped jobs then- this automatically sends exit twice. """+ this automatically sends exit twice.++ This is in place to avoid confusing EOF / ctrl-D messages from Python.+ """- self.sendline("exit")- index = self.expect([EOF, "(?i)there are stopped jobs"])- if index==1:+ if self.is_logged_in():+ self.sendline("exit")++ if self.expect([EOF, "(?i)there are stopped jobs"]) == 1: self.sendline("exit") self.expect(EOF)- self.close()+ self.close()+ self.logged_in = False++ else:+ raise ExceptionPxssh ("You aren't logged in yet..")++ def is_logged_in(self):+ """ It'd be nice if just querying self.logged_in was sufficient, but a+ user can logout by themselves, thus causing the underlying connection+ to close, s.t. the logged_in attribute would be out of sync. """++ self.logged_in = self.logged_in and self.isalive()++ return self.logged_in- def prompt (self, timeout=20):+ def set_unique_prompt (self, timeout=20):+ """ Set a unique prompt via pxsh.set_unique_prompt(..). """++ if not self.is_logged_in():+ return False++ return pxsh.set_unique_prompt(self, timeout=timeout)++ def prompt (self, timeout=10): """This matches the shell prompt. This is little more than a short-cut to the expect() method. This returns True if the shell prompt was@@ -272,36 +429,9 @@ manually set the PROMPT attribute to a regex pattern for matching the prompt. """- i = self.expect([self.PROMPT, TIMEOUT], timeout=timeout)- if i==1:+ if not self.is_logged_in(): return False- return True- - def set_unique_prompt (self):- """This sets the remote prompt to something more unique than # or $.- This makes it easier for the prompt() method to match the shell prompt- unambiguously. This method is called automatically by the login()- method, but you may want to call it manually if you somehow reset the- shell prompt. For example, if you 'su' to a different user then you- will need to manually reset the prompt. This sends shell commands to- the remote host to set the prompt, so this assumes the remote host is- ready to receive commands.-- Alternatively, you may use your own prompt pattern. Just set the PROMPT- attribute to a regular expression that matches it. In this case you- should call login() with auto_prompt_reset=False; then set the PROMPT- attribute. After that the prompt() method will try to match your prompt- pattern."""-- self.sendline ("unset PROMPT_COMMAND")- self.sendline (self.PROMPT_SET_SH) # sh-style- i = self.expect ([TIMEOUT, self.PROMPT], timeout=10)- if i == 0: # csh-style- self.sendline (self.PROMPT_SET_CSH)- i = self.expect ([TIMEOUT, self.PROMPT], timeout=10)- if i == 0:- return False- return True+ return pxsh.prompt(self, timeout=timeout) # vi:ts=4:sw=4:expandtab:ft=python:diff -Naur ../pexpect-2.3/setup.py ./setup.py--- ../pexpect-2.3/setup.py 2008-01-05 14:00:44.000000000 -0800+++ ./setup.py 2009-05-01 20:36:34.000000000 -0700@@ -4,8 +4,8 @@ ''' from distutils.core import setup setup (name='pexpect',- version='2.3',- py_modules=['pexpect', 'pxssh', 'fdpexpect', 'FSM', 'screen', 'ANSI'],+ version='2.3-p2',+ py_modules=['__init__', 'pexpect', 'pxsh', 'pxssh', 'fdpexpect', 'FSM', 'screen', 'ANSI'], description='Pexpect is a pure Python Expect. It allows easy control of other applications.', author='Noah Spurrier', author_email='noah@noah.org',diff -Naur ../pexpect-2.3/tests/test_pxsh.py ./tests/test_pxsh.py--- ../pexpect-2.3/tests/test_pxsh.py 1969-12-31 16:00:00.000000000 -0800+++ ./tests/test_pxsh.py 2009-06-09 12:46:49.000000000 -0700@@ -0,0 +1,87 @@+"""+A nose capable test module for pxsh.++Garrett Cooper, April 2009+"""++import getpass, platform, pxsh, re++def setup_package():+ """ Check to see what platform we're using (this only works on Unix right+ now).. """+ assert platform.system().find("Windows") == -1++class TestPxsh(object):+ """ Test out pxsh. """++ console = None++ def setUp(self):+ self.console = pxsh.pxsh()+ # Set it to the default prompt in pxsh.+ self.console.set_unique_prompt(timeout=5)++ def tearDown(self):+ self.console = None++ def ensure_matched_prompt(self, expected_prompt, expected_truth):+ """ Ensure that the new prompt matches the expected truth+ condition. """++ if expected_prompt is not None:+ self.console.expected_prompt = expected_prompt++ print 'set prompt =>', self.console.set_prompt, "expected prompt =>", \+ self.console.expected_prompt+ assert self.console.prompt(timeout=5) is expected_truth++ def ensure_unique_prompt(self, set_prompt, expected_prompt, expected_truth):+ """ Ensure that the new prompt `uniquely' set, matches the expected+ truth condition. """++ if set_prompt is not None:+ self.console.set_prompt = set_prompt+ if expected_prompt is not None:+ self.console.expected_prompt = expected_prompt++ print 'set prompt =>', self.console.set_prompt, "expected prompt =>", \+ self.console.expected_prompt+ assert self.console.set_unique_prompt(timeout=5) is expected_truth++ def test_set_unique_prompt(self):+ """ Test pxsh.set_unique_prompt() """++ expected_tokens = [+ # set prompt, expected prompt, expected truth++ # -- Positive testcases+ ( None, None, True),+ ( r"[\u] ", r"\[" + getpass.getuser() + "\] ", True),+ ( '[0-9]>', re.escape('[0-9]>'), True),+ ( 'rommon 42 >', 'rommon \d+ >', True),+ # -- Negative testcases+ ( 'foo', 'bar', False),+ ( '[0-9]>', '[0-9]>', False)+ ]++ for expected_tuples in expected_tokens:++ set_prompt, expected_prompt, expected_truth = expected_tuples++ yield self.ensure_unique_prompt, set_prompt, expected_prompt, \+ expected_truth++ def test_prompt(self):+ """ Test pxsh.prompt() """++ expected_tokens = [+ # expected prompt, expected truth+ # -- Positive testcases+ ( None, True),+ # -- Negative testcases+ ( 'bar', False),+ ]++ for expected_tuples in expected_tokens:+ expected_prompt, expected_truth = expected_tuples+ yield self.ensure_matched_prompt, expected_prompt, expected_truthdiff -Naur ../pexpect-2.3/tests/test_pxssh.py ./tests/test_pxssh.py--- ../pexpect-2.3/tests/test_pxssh.py 1969-12-31 16:00:00.000000000 -0800+++ ./tests/test_pxssh.py 2009-06-09 12:47:42.000000000 -0700@@ -0,0 +1,306 @@+"""+A nose capable test module for pxssh.++This is split into:+- TestPexpect - Testing base class.+- TestPexpectSanity - Sanity check tests, primarily for boundary+ conditions.+- TestPasskeyFunctionality - Passkey login check tests.+- TestPasswordFunctionality - Password login check tests.++Garrett Cooper, April 2009+"""++import getpass, glob, os, pexpect, pxssh+from os import getenv+from pxssh import ExceptionPxssh+from random import choice, randint+import re, shutil, socket, string, sys, tempfile, unittest++def gen_random_password():+ """ Generate a random password for testing, somewhere to the tune of\+ 10~25 characters. """++ return ''.join([choice(string.letters + string.digits) \+ for i in range(randint(10,25))])++class TestPexpect(unittest.TestCase):+ """ A class that serves as a basis for all pexpect based tests. """++ hostname = None++ password = None++ pxssh_inst = None++ ssh_backup = None++ ssh_home = None++ ssh_opts = None++ username = None++ def setUp(self):+ """ Setup creds for a proper login (backup current settings) """++ self.pxssh_inst = pxssh.pxssh(ssh_opts=self.ssh_opts)++ try:++ username = getenv("PEXPECT_USERNAME", "")++ # Really simple sanity check. Username's should be 2+ chars long+ # and consist of logical, word characters.+ if len(username) < 2 or re.search("\W+", username):+ self.username = getpass.getuser()++ self.password = getenv("PEXPECT_PASSWORD", gen_random_password())++ hostname = getenv("PEXPECT_HOSTNAME", "")++ try:+ self.hostname = socket.gethostbyname(hostname)+ except socket.gaierror:+ print "Could not resolve host; defaulting to `localhost'"+ self.hostname = socket.gethostname()++ except KeyboardInterrupt:+ raise Exception ("Interrupted by user; indeterminate results")++ self.ssh_home = os.path.join(os.path.expanduser("~"), ".ssh")++ if os.access(self.ssh_home, os.W_OK):+ # If the directory exists, and is writable, let's back it up!++ self.ssh_backup = tempfile.mkdtemp("", \+ os.path.join(os.path.expanduser("~"), "ssh"))++ # Backup a user's ssh directory, if it exists. Move+ # self.ssh_home -> self.ssh_backup+ os.rename(self.ssh_home, self.ssh_backup)++ # Make a copy of the old directory for storage.+ os.makedirs(self.ssh_home)++ def tearDown(self):+ """ Teardown the previously setup login creds (restore backed up\+ settings) """++ #print "Will remove all files in", self.ssh_home, "and restore:", self.ssh_backup++ if self.pxssh_inst is not None and self.pxssh_inst.logged_in is True:+ # Logout if we successfully logged in (cleanup ssh connection).+ self.pxssh_inst.logout()++ if self.ssh_home is not None and self.ssh_backup is not None and \+ os.access(self.ssh_backup, os.W_OK):++ # Restore the backup unless it truly exists -- have to remove the+ # temporary files, first before using rmdir..+ for file in os.listdir(self.ssh_home):+ os.unlink(os.path.join(self.ssh_home, file))++ # .. then remove the test directory..+ os.rmdir(self.ssh_home)++ # .. and finally restore the backup.+ os.rename(self.ssh_backup, self.ssh_home)++class TestPexpectSanity(TestPexpect):+ """ A class to test some sanity of the overall functionality of the+ system. Doesn't require things like passwords, etc. """++ def test_bad_hostname_login(self):+ """ Invalid host login treated properly """+ try:+ self.pxssh_inst.login('999.999.999.999', self.username)+ except ExceptionPxssh:+ pass+ else:+ assert 0++ def test_bad_network_login(self):+ """ Invalid network treated properly """+ try:+ self.pxssh_inst.login('255.255.255.255', self.username)+ except ExceptionPxssh:+ pass+ else:+ assert 0++ def test_logout_before_login(self):+ """ logout() before login() call++ Make sure that state is maintained so logout isn't processed before+ login. """++ try:+ self.pxssh_inst.logout()+ except ExceptionPxssh:+ pass+ else:+ assert 0++ def test_goodport_login_pos(self):+ """ Valid port number (22) """+ try:+ self.pxssh_inst.login(self.hostname, self.username, \+ password=self.password, port=22)+ except ExceptionPxssh:+ # Ignore login related errors -- I'm only looking for TypeError's.+ pass++ def test_badport_login_neg1(self):+ """ Negative out of bounds port case """+ try:+ self.pxssh_inst.login(self.hostname, self.username, port=-1)+ except TypeError:+ pass+ else:+ assert 0++ def test_badport_login_neg2(self):+ """ Positive out of bounds port case """+ try:+ self.pxssh_inst.login(self.hostname, self.username, port=100000)+ except TypeError:+ pass+ else:+ assert 0++ def test_badport_login_neg3(self):+ """ Bogus port case """+ try:+ self.pxssh_inst.login(port = '')+ except TypeError:+ pass+ else:+ assert 0++class TestPasskeyFunctionality(TestPexpect):+ """ A class that defines passkey functionality tests.++ XXX: Need to login and transfer the file (for non-localhost based+ authentication methods). This isn't really secure though, so I don't like+ this methodology... esp since this can be done with a test account on the+ local machine with almost zero work.++ IMPORTANT: MAKE SURE TO READ THROUGH THE PERMISSION DENIED SECTION BEFORE+ REPORTING A TEST BUG.++ Some versions of PAM released on different Unix versions (*cough* Gentoo+ Linux) are by default extremely paranoid and if you screw up the+ permissions, things WILL fail -- horribly. If you do have root access or+ can read /var/log/messages, look for authentication errors reported by+ PAM relating along the lines of:++ pam_unix(sshd:auth): authentication failure; logname= uid=0 euid=0 tty=ssh ruser= rhost=127.0.0.1 user={some username}++ So, basically it's an infrastructure problem.+ """++ hasOldPasskey = False++ ssh_opts = '-o PubkeyAuthentication=yes \+-o ChallengeResponseAuthentication=no -o HostbasedAuthentication=no \+-o PasswordAuthentication=no'++ def test_goodpass_passkey_login(self):+ """ Standard passkey login """++ self.setup_passkey(password=self.password)+ self.pxssh_inst.login(self.hostname, self.username, \+ password=self.password)++ def test_bad_auth_options(self):+ """ Ensure `permission denied' is caught in the FSM with conflicting options to ssh(1)++ See relevant section in pxssh.login() for more details.+ """++ exc = None++ try:+ # Try to force password authentication.+ self.force_password = True+ self.pxssh_inst.login(self.hostname, self.username)+ except ExceptionPxssh:+ pass+ except Exception, e:+ exc = e+ else:+ exc = Exception("Exception not thrown (did the login go through???)")++ # Be sure to set this or the rest of the tests will fail..+ self.force_password = False++ if exc is not None:+ raise exc++ def test_nullpass_passkey_login(self):+ """ No-password passkey login """++ self.setup_passkey()+ self.pxssh_inst.login(self.hostname, self.username)++ def test_badpass_passkey_login(self):+ """ Bad password key login """++ self.setup_passkey(password=gen_random_password())++ try:+ self.pxssh_inst.login(self.hostname, self.username)+ except ExceptionPxssh:+ pass+ else:+ raise Exception("Didn't catch anything. Please check the backlog: %s" %+ self.pxssh_inst.before)+ assert 0++ def setup_nonzero_passkey(self):+ """ Setup a passkey login with an actual legitimate password """+ self.setup_passkey(password=self.password)+ self.pxssh_inst.login(self.hostname, self.username, \+ password=self.password)++ def setup_passkey(self, password=''):+ """ Setup a password for a passkey """++ key_type = 'dsa'++ if self.hasOldPasskey is True:+ os.unlink( glob.glob( '%s/id_%s*' % (self.ssh_home, key_type) ) )++ cmd = "ssh-keygen -q -b 1024 -N '%s' -f %s/id_%s -t %s" % \+ (password, self.ssh_home, key_type, key_type)++ # print "Using command -- %s -- to generate password" % cmd++ if os.system(cmd):+ raise OSError("Couldn't create ssh key")++ # Don't forget to copy the keys.. otherwise we'll get permission+ # denied -- silly goomba!+ shutil.copyfile("%s/id_%s.pub" % (self.ssh_home, key_type),+ "%s/authorized_keys" % self.ssh_home)++ # os.system('ls -l %s' % self.ssh_home)++ self.hasOldPasskey = True++class TestPasswordFunctionality(TestPexpect):+ """ A class that defines password entry tests. """++ def test_password_login(self):+ """ Standard login """+ self.pxssh_inst.login(self.hostname, self.username, self.password)++ def test_nopass_login(self):+ """ No-password login (it's possible if the server supports this) """+ self.pxssh_inst.login(self.hostname, self.username)++ def setUp(self):+ """ Setup the password functionality class """+ TestPexpect.setUp(self)+ self.pxssh_inst.force_password = True