Merge pull request #40 from phate999/master

New Sample Apps

cli_sample/cli_sample.py

@@ -0,0 +1,37 @@
+# cli - execute CLI command and return output
+
+from csclient import EventingCSClient
+
+
+def cli(username, password, cmd):
+    """
+    :param username: string
+        username for SSH login
+    :param password: string
+        user password for SSH login
+    :param cmd: string
+        CLI command to execute
+        Example: "sms 2081234567 'hello world' int1"
+        Example: "arpdump"
+    :return: string
+        Text output returned from CLI command
+    """
+    import cppxssh
+    ssh_tunnel = cppxssh.cppxssh()
+    ssh_tunnel.login('localhost', username, password, auto_prompt_reset=False)
+    ssh_tunnel.PROMPT = '\[[0-9A-Za-z_]+@.+\]\$ '
+    ssh_tunnel.sendline(cmd)
+    ssh_tunnel.prompt()
+    output = ssh_tunnel.before.decode()
+    del ssh_tunnel
+    try:
+        # try to remove the command echo
+        return output.split(cmd + '\r\n')[1]
+    except IndexError:
+        # for some reason we failed to remove so return original
+        return output
+
+
+cp = EventingCSClient('cli_sample')
+cp.log('Starting...')
+cp.log('Output:\n' + cli('admin', '11', 'arpdump'))

cli_sample/cppxssh.py

@@ -0,0 +1,74 @@
+from pexpect import ExceptionPexpect, TIMEOUT, EOF, spawn
+from pexpect.pxssh import pxssh, ExceptionPxssh
+
+class cppxssh (pxssh):
+
+    def login (self, server, username, password='', terminal_type='ansi',
+                original_prompt=r"[#$]", login_timeout=10, port=None,
+                auto_prompt_reset=True, ssh_key=None, quiet=True,
+                sync_multiplier=1, check_local_ip=True):
+
+    # cp ssh doesn't have an options field so ignore ALL options
+        ssh_options = ''
+        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)Do you trust the host key", original_prompt, "(?i)(?:password:)|(?:passphrase for key)", "(?i)permission denied", "(?i)terminal type", TIMEOUT, "(?i)connection closed by remote host", EOF], 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)Do you trust the host key", 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)Do you trust the host key", 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)Do you trust the host key", original_prompt, "(?i)(?:password:)|(?:passphrase for key)", "(?i)permission denied", "(?i)terminal type", TIMEOUT])
+        if i==7:
+            self.close()
+            raise ExceptionPxssh('Could not establish connection to host')
+
+        # 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_sample 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_sample 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')
+        return True