1 """This class extends pexpect.spawn to specialize setting up SSH connections.
2 This adds methods for login, logout, and expecting the shell prompt.
4 $Id: pxssh.py 487 2007-08-29 22:33:29Z noah $
11 __all__ = ['ExceptionPxssh', 'pxssh']
13 # Exception classes used by this module.
14 class ExceptionPxssh(ExceptionPexpect):
15 """Raised for pxssh exceptions.
20 """This class extends pexpect.spawn to specialize setting up SSH
21 connections. This adds methods for login, logout, and expecting the shell
22 prompt. It does various tricky things to handle many situations in the SSH
23 login process. For example, if the session is your first login, then pxssh
24 automatically accepts the remote certificate; or if you have public key
25 authentication setup then pxssh won't wait for the password prompt.
27 pxssh uses the shell prompt to synchronize output from the remote host. In
28 order to make this more robust it sets the shell prompt to something more
29 unique than just $ or #. This should work on most Borne/Bash or Csh style
32 Example that runs a few commands on a remote server and prints the result::
38 hostname = raw_input('hostname: ')
39 username = raw_input('username: ')
40 password = getpass.getpass('password: ')
41 s.login (hostname, username, password)
42 s.sendline ('uptime') # run a command
43 s.prompt() # match the prompt
44 print s.before # print everything before the prompt.
52 except pxssh.ExceptionPxssh, e:
53 print "pxssh failed on login."
56 Note that if you have ssh-agent running while doing development with pxssh
57 then this can lead to a lot of confusion. Many X display managers (xdm,
58 gdm, kdm, etc.) will automatically start a GUI agent. You may see a GUI
59 dialog box popup asking for a password during development. You should turn
60 off any key agents during testing. The 'force_password' attribute will turn
61 off public key authentication. This will only work if the remote SSH server
62 is configured to allow password logins. Example of using 'force_password'
66 s.force_password = True
67 hostname = raw_input('hostname: ')
68 username = raw_input('username: ')
69 password = getpass.getpass('password: ')
70 s.login (hostname, username, password)
73 def __init__ (self, timeout=30, maxread=2000, searchwindowsize=None, logfile=None, cwd=None, env=None):
74 spawn.__init__(self, None, timeout=timeout, maxread=maxread, searchwindowsize=searchwindowsize, logfile=logfile, cwd=cwd, env=env)
78 #SUBTLE HACK ALERT! Note that the command to set the prompt uses a
79 #slightly different string than the regular expression to match it. This
80 #is because when you set the prompt the command will echo back, but we
81 #don't want to match the echoed command. So if we make the set command
82 #slightly different than the regex we eliminate the problem. To make the
83 #set command different we add a backslash in front of $. The $ doesn't
84 #need to be escaped, but it doesn't hurt and serves to make the set
85 #prompt command different than the regex.
87 # used to match the command-line prompt
88 self.UNIQUE_PROMPT = "\[PEXPECT\][\$\#] "
89 self.PROMPT = self.UNIQUE_PROMPT
91 # used to set shell command-line prompt to UNIQUE_PROMPT.
92 self.PROMPT_SET_SH = "PS1='[PEXPECT]\$ '"
93 self.PROMPT_SET_CSH = "set prompt='[PEXPECT]\$ '"
94 self.SSH_OPTS = "-o'RSAAuthentication=no' -o 'PubkeyAuthentication=no'"
95 # Disabling X11 forwarding gets rid of the annoying SSH_ASKPASS from
96 # displaying a GUI password dialog. I have not figured out how to
97 # disable only SSH_ASKPASS without also disabling X11 forwarding.
98 # Unsetting SSH_ASKPASS on the remote side doesn't disable it! Annoying!
99 #self.SSH_OPTS = "-x -o'RSAAuthentication=no' -o 'PubkeyAuthentication=no'"
100 self.force_password = False
101 self.auto_prompt_reset = True
103 def levenshtein_distance(self, a,b):
105 """This calculates the Levenshtein distance between a and b.
108 n, m = len(a), len(b)
113 for i in range(1,m+1):
114 previous, current = current, [i]+[0]*n
115 for j in range(1,n+1):
116 add, delete = previous[j]+1, current[j-1]+1
117 change = previous[j-1]
120 current[j] = min(add, delete, change)
123 def synch_original_prompt (self):
125 """This attempts to find the prompt. Basically, press enter and record
126 the response; press enter again and record the response; if the two
127 responses are similar then assume we are at the original prompt. """
129 # All of these timing pace values are magic.
130 # I came up with these based on what seemed reliable for
131 # connecting to a heavily loaded machine I have.
132 # If latency is worse than these values then this will fail.
134 self.read_nonblocking(size=10000,timeout=1) # GAS: Clear out the cache before getting the prompt
138 x = self.read_nonblocking(size=1000,timeout=1)
142 a = self.read_nonblocking(size=1000,timeout=1)
146 b = self.read_nonblocking(size=1000,timeout=1)
147 ld = self.levenshtein_distance(a,b)
151 if float(ld)/len_a < 0.4:
155 ### TODO: This is getting messy and I'm pretty sure this isn't perfect.
156 ### TODO: I need to draw a flow chart for this.
157 def login (self,server,username,password='',terminal_type='ansi',original_prompt=r"[#$]",login_timeout=10,port=None,auto_prompt_reset=True):
159 """This logs the user into the given server. It uses the
160 'original_prompt' to try to find the prompt right after login. When it
161 finds the prompt it immediately tries to reset the prompt to something
162 more easily matched. The default 'original_prompt' is very optimistic
163 and is easily fooled. It's more reliable to try to match the original
164 prompt as exactly as possible to prevent false matches by server
165 strings such as the "Message Of The Day". On many systems you can
166 disable the MOTD on the remote server by creating a zero-length file
167 called "~/.hushlogin" on the remote server. If a prompt cannot be found
168 then this will not necessarily cause the login to fail. In the case of
169 a timeout when looking for the prompt we assume that the original
170 prompt was so weird that we could not match it, so we use a few tricks
171 to guess when we have reached the prompt. Then we hope for the best and
172 blindly try to reset the prompt to something more unique. If that fails
173 then login() raises an ExceptionPxssh exception.
175 In some situations it is not possible or desirable to reset the
176 original prompt. In this case, set 'auto_prompt_reset' to False to
177 inhibit setting the prompt to the UNIQUE_PROMPT. Remember that pxssh
178 uses a unique prompt in the prompt() method. If the original prompt is
179 not reset then this will disable the prompt() method unless you
180 manually set the PROMPT attribute. """
183 if self.force_password:
184 ssh_options = ssh_options + ' ' + self.SSH_OPTS
186 ssh_options = ssh_options + ' -p %s'%(str(port))
187 cmd = "ssh %s -l %s %s" % (ssh_options, username, server)
189 # This does not distinguish between a remote server 'password' prompt
190 # and a local ssh 'passphrase' prompt (for unlocking a private key).
191 spawn._spawn(self, cmd)
192 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)
196 # New certificate -- always accept it.
197 # This is what you get if SSH does not have the remote host's
198 # public key stored in the 'known_hosts' cache.
200 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])
201 if i==2: # password or passphrase
202 self.sendline(password)
203 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])
205 self.sendline(terminal_type)
206 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])
210 # This is weird. This should not happen twice in a row.
212 raise ExceptionPxssh ('Weird error. Got "are you sure" prompt twice.')
213 elif i==1: # can occur if you have a public key pair set to authenticate.
214 ### TODO: May NOT be OK if expect() got tricked and matched a false prompt.
216 elif i==2: # password prompt again
217 # For incorrect passwords, some ssh servers will
218 # ask for the password again, others return 'denied' right away.
219 # If we get the password prompt again then this means
220 # we didn't get the password right the first time.
222 raise ExceptionPxssh ('password refused')
223 elif i==3: # permission denied -- password was bad.
225 raise ExceptionPxssh ('permission denied')
226 elif i==4: # terminal type again? WTF?
228 raise ExceptionPxssh ('Weird error. Got "terminal type" prompt twice.')
230 #This is tricky... I presume that we are at the command-line prompt.
231 #It may be that the shell prompt was so weird that we couldn't match
232 #it. Or it may be that we couldn't log in for some other reason. I
233 #can't be sure, but it's safe to guess that we did login because if
234 #I presume wrong and we are not logged in then this should be caught
235 #later when I try to set the shell prompt.
237 elif i==6: # Connection closed by remote host
239 raise ExceptionPxssh ('connection closed')
242 raise ExceptionPxssh ('unexpected login response')
243 if not self.synch_original_prompt():
245 raise ExceptionPxssh ('could not synchronize with original prompt')
246 # We appear to be in.
247 # set shell prompt to something unique.
248 if auto_prompt_reset:
249 if not self.set_unique_prompt():
251 raise ExceptionPxssh ('could not set shell prompt\n'+self.before)
256 """This sends exit to the remote shell. If there are stopped jobs then
257 this automatically sends exit twice. """
259 self.sendline("exit")
260 index = self.expect([EOF, "(?i)there are stopped jobs"])
262 self.sendline("exit")
266 def prompt (self, timeout=20):
268 """This matches the shell prompt. This is little more than a short-cut
269 to the expect() method. This returns True if the shell prompt was
270 matched. This returns False if there was a timeout. Note that if you
271 called login() with auto_prompt_reset set to False then you should have
272 manually set the PROMPT attribute to a regex pattern for matching the
275 i = self.expect([self.PROMPT, TIMEOUT], timeout=timeout)
280 def set_unique_prompt (self):
282 """This sets the remote prompt to something more unique than # or $.
283 This makes it easier for the prompt() method to match the shell prompt
284 unambiguously. This method is called automatically by the login()
285 method, but you may want to call it manually if you somehow reset the
286 shell prompt. For example, if you 'su' to a different user then you
287 will need to manually reset the prompt. This sends shell commands to
288 the remote host to set the prompt, so this assumes the remote host is
289 ready to receive commands.
291 Alternatively, you may use your own prompt pattern. Just set the PROMPT
292 attribute to a regular expression that matches it. In this case you
293 should call login() with auto_prompt_reset=False; then set the PROMPT
294 attribute. After that the prompt() method will try to match your prompt
297 self.sendline ("unset PROMPT_COMMAND")
298 self.sendline (self.PROMPT_SET_SH) # sh-style
299 i = self.expect ([TIMEOUT, self.PROMPT], timeout=10)
300 if i == 0: # csh-style
301 self.sendline (self.PROMPT_SET_CSH)
302 i = self.expect ([TIMEOUT, self.PROMPT], timeout=10)
307 # vi:ts=4:sw=4:expandtab:ft=python: