| """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.sendline() |
| time.sleep(0.5) |
| 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: |
