--- /dev/null
+"""This class extends pexpect.spawn to specialize setting up SSH connections.
+This adds methods for login, logout, and expecting the shell prompt.
+
+$Id: pxssh.py 487 2007-08-29 22:33:29Z noah $
+"""
+
+from pexpect import *
+import pexpect
+import time
+
+__all__ = ['ExceptionPxssh', 'pxssh']
+
+# Exception classes used by this module.
+class ExceptionPxssh(ExceptionPexpect):
+ """Raised for pxssh exceptions.
+ """
+
+class pxssh (spawn):
+
+ """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
+ 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
+ shells.
+
+ Example that runs a few commands on a remote server and prints the result::
+
+ import pxssh
+ import getpass
+ try:
+ s = pxssh.pxssh()
+ hostname = raw_input('hostname: ')
+ username = raw_input('username: ')
+ password = getpass.getpass('password: ')
+ s.login (hostname, username, password)
+ s.sendline ('uptime') # run a command
+ s.prompt() # match the prompt
+ print s.before # print everything before the prompt.
+ s.sendline ('ls -l')
+ s.prompt()
+ print s.before
+ s.sendline ('df')
+ s.prompt()
+ print s.before
+ s.logout()
+ except pxssh.ExceptionPxssh, e:
+ print "pxssh failed on login."
+ print str(e)
+
+ Note that if you have ssh-agent running while doing development with pxssh
+ then this can lead to a lot of confusion. Many X display managers (xdm,
+ gdm, kdm, etc.) will automatically start a GUI agent. You may see a GUI
+ dialog box popup asking for a password during development. You should turn
+ off any key agents during testing. The 'force_password' attribute will turn
+ off public key authentication. This will only work if the remote SSH server
+ is configured to allow password logins. Example of using 'force_password'
+ attribute::
+
+ s = pxssh.pxssh()
+ s.force_password = True
+ hostname = raw_input('hostname: ')
+ username = raw_input('username: ')
+ password = getpass.getpass('password: ')
+ 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)
+
+ self.name = '<pxssh>'
+
+ #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
+
+ def levenshtein_distance(self, 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 = change + 1
+ current[j] = min(add, delete, change)
+ return current[n]
+
+ 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):
+
+ """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
+ and is easily fooled. It's more reliable to try to match the original
+ prompt as exactly as possible to prevent false matches by server
+ strings such as the "Message Of The Day". On many systems you can
+ disable the MOTD on the remote server by creating a zero-length file
+ called "~/.hushlogin" on the remote server. If a prompt cannot be found
+ then this will not necessarily cause the login to fail. In the case of
+ a timeout when looking for the prompt we assume that the original
+ 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.
+
+ 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
+ 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. """
+
+ 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)
+
+ # 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)
+
+ # 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():
+ 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
+
+ def logout (self):
+
+ """This sends exit to the remote shell. If there are stopped jobs then
+ this automatically sends exit twice. """
+
+ self.sendline("exit")
+ index = self.expect([EOF, "(?i)there are stopped jobs"])
+ if index==1:
+ self.sendline("exit")
+ self.expect(EOF)
+ self.close()
+
+ 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. """
+
+ i = self.expect([self.PROMPT, TIMEOUT], timeout=timeout)
+ if i==1:
+ 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
+
+# vi:ts=4:sw=4:expandtab:ft=python:
git.ucc.asn.au / progcomp10.git / blobdiff
UCC git Repository :: git.ucc.asn.au