How to make a ssh connection with python?

Question

Can anyone recommend something for making a ssh connection in python? I need it to be compatible with any OS.

I've already tried pyssh only to get an error with SIGCHLD, which I've read is because Windows lacks this. I've tried getting paramiko to work, but I've had errors between paramiko and Crypto to the point where the latest versions of each won't work together.

Python 2.6.1 currently on a Windows machine.

Answer 1

Notice that this doesn't work in Windows.
The module pxssh does exactly what you want:
For example, to run 'ls -l' and to print the output, you need to do something like that :

from pexpect import pxssh
s = pxssh.pxssh()
if not s.login ('localhost', 'myusername', 'mypassword'):
    print "SSH session failed on login."
    print str(s)
else:
    print "SSH session login successful"
    s.sendline ('ls -l')
    s.prompt()         # match the prompt
    print s.before     # print everything before the prompt.
    s.logout()

Some links :
Pxssh docs : http://dsnra.jpl.nasa.gov/software/Python/site-packages/Contrib/pxssh.html
Pexpect (pxssh is based on pexpect) : http://pexpect.readthedocs.io/en/stable/

Answer 2

Twisted has SSH support : http://www.devshed.com/c/a/Python/SSH-with-Twisted/

The twisted.conch package adds SSH support to Twisted. This chapter shows how you can use the modules in twisted.conch to build SSH servers and clients.

Setting Up a Custom SSH Server

The command line is an incredibly efficient interface for certain tasks. System administrators love the ability to manage applications by typing commands without having to click through a graphical user interface. An SSH shell is even better, as it’s accessible from anywhere on the Internet.

You can use twisted.conch to create an SSH server that provides access to a custom shell with commands you define. This shell will even support some extra features like command history, so that you can scroll through the commands you’ve already typed.

How Do I Do That? Write a subclass of twisted.conch.recvline.HistoricRecvLine that implements your shell protocol. HistoricRecvLine is similar to twisted.protocols.basic.LineReceiver , but with higher-level features for controlling the terminal.

Write a subclass of twisted.conch.recvline.HistoricRecvLine that implements your shell protocol. HistoricRecvLine is similar to twisted.protocols.basic.LineReceiver, but with higher-level features for controlling the terminal.

To make your shell available through SSH, you need to implement a few different classes that twisted.conch needs to build an SSH server. First, you need the twisted.cred authentication classes: a portal, credentials checkers, and a realm that returns avatars. Use twisted.conch.avatar.ConchUser as the base class for your avatar. Your avatar class should also implement twisted.conch.interfaces.ISession , which includes an openShell method in which you create a Protocol to manage the user’s interactive session. Finally, create a twisted.conch.ssh.factory.SSHFactory object and set its portal attribute to an instance of your portal.

Example 10-1 demonstrates a custom SSH server that authenticates users by their username and password. It gives each user a shell that provides several commands.

Example 10-1. sshserver.py

from twisted.cred import portal, checkers, credentials
from twisted.conch import error, avatar, recvline, interfaces as conchinterfaces
from twisted.conch.ssh import factory, userauth, connection, keys, session, common from twisted.conch.insults import insults from twisted.application import service, internet
from zope.interface import implements
import os

class SSHDemoProtocol(recvline.HistoricRecvLine):
    def __init__(self, user):
        self.user = user

    def connectionMade(self) : 
     recvline.HistoricRecvLine.connectionMade(self)
        self.terminal.write("Welcome to my test SSH server.")
        self.terminal.nextLine() 
        self.do_help()
        self.showPrompt()

    def showPrompt(self): 
        self.terminal.write("$ ")

    def getCommandFunc(self, cmd):
        return getattr(self, ‘do_’ + cmd, None)

    def lineReceived(self, line):
        line = line.strip()
        if line: 
            cmdAndArgs = line.split()
            cmd = cmdAndArgs[0]
            args = cmdAndArgs[1:]
            func = self.getCommandFunc(cmd)
            if func: 
               try:
                   func(*args)
               except Exception, e: 
                   self.terminal.write("Error: %s" % e)
                   self.terminal.nextLine()
            else:
               self.terminal.write("No such command.")
               self.terminal.nextLine()
        self.showPrompt()

    def do_help(self, cmd=”):
        "Get help on a command. Usage: help command"
        if cmd: 
            func = self.getCommandFunc(cmd)
            if func:
                self.terminal.write(func.__doc__)
                self.terminal.nextLine()
                return

        publicMethods = filter(
            lambda funcname: funcname.startswith(‘do_’), dir(self)) 
        commands = [cmd.replace(‘do_’, ”, 1) for cmd in publicMethods] 
        self.terminal.write("Commands: " + " ".join(commands))
        self.terminal.nextLine()

    def do_echo(self, *args):
        "Echo a string. Usage: echo my line of text"
        self.terminal.write(" ".join(args)) 
        self.terminal.nextLine()

    def do_whoami(self):
        "Prints your user name. Usage: whoami"
        self.terminal.write(self.user.username)
        self.terminal.nextLine()

    def do_quit(self):
        "Ends your session. Usage: quit" 
        self.terminal.write("Thanks for playing!")
        self.terminal.nextLine() 
        self.terminal.loseConnection()

    def do_clear(self):
        "Clears the screen. Usage: clear" 
        self.terminal.reset()

class SSHDemoAvatar(avatar.ConchUser): 
    implements(conchinterfaces.ISession)

    def __init__(self, username): 
        avatar.ConchUser.__init__(self) 
        self.username = username 
        self.channelLookup.update({‘session’:session.SSHSession})

    def openShell(self, protocol): 
        serverProtocol = insults.ServerProtocol(SSHDemoProtocol, self)
        serverProtocol.makeConnection(protocol)
        protocol.makeConnection(session.wrapProtocol(serverProtocol))

    def getPty(self, terminal, windowSize, attrs):
        return None

    def execCommand(self, protocol, cmd): 
        raise NotImplementedError

    def closed(self):
        pass

class SSHDemoRealm:
    implements(portal.IRealm)

    def requestAvatar(self, avatarId, mind, *interfaces):
        if conchinterfaces.IConchUser in interfaces:
            return interfaces[0], SSHDemoAvatar(avatarId), lambda: None
        else:
            raise Exception, "No supported interfaces found."

def getRSAKeys():
    if not (os.path.exists(‘public.key’) and os.path.exists(‘private.key’)):
        # generate a RSA keypair
        print "Generating RSA keypair…" 
        from Crypto.PublicKey import RSA 
        KEY_LENGTH = 1024
        rsaKey = RSA.generate(KEY_LENGTH, common.entropy.get_bytes)
        publicKeyString = keys.makePublicKeyString(rsaKey) 
        privateKeyString = keys.makePrivateKeyString(rsaKey)
        # save keys for next time
        file(‘public.key’, ‘w+b’).write(publicKeyString)
        file(‘private.key’, ‘w+b’).write(privateKeyString)
        print "done."
    else:
        publicKeyString = file(‘public.key’).read()
        privateKeyString = file(‘private.key’).read() 
    return publicKeyString, privateKeyString

if __name__ == "__main__":
    sshFactory = factory.SSHFactory() 
    sshFactory.portal = portal.Portal(SSHDemoRealm())
    users = {‘admin’: ‘aaa’, ‘guest’: ‘bbb’}
    sshFactory.portal.registerChecker(
 checkers.InMemoryUsernamePasswordDatabaseDontUse(**users))

    pubKeyString, privKeyString =
getRSAKeys()
    sshFactory.publicKeys = {
        ‘ssh-rsa’: keys.getPublicKeyString(data=pubKeyString)}
    sshFactory.privateKeys = {
        ‘ssh-rsa’: keys.getPrivateKeyObject(data=privKeyString)}

    from twisted.internet import reactor 
    reactor.listenTCP(2222, sshFactory) 
    reactor.run()

{mospagebreak title=Setting Up a Custom SSH Server continued}

sshserver.py will run an SSH server on port 2222. Connect to this server with an SSH client using the username admin and password aaa, and try typing some commands:

$ ssh admin@localhost -p 2222 
admin@localhost’s password: aaa

>>> Welcome to my test SSH server.  
Commands: clear echo help quit whoami
$ whoami
admin
$ help echo
Echo a string. Usage: echo my line of text
$ echo hello SSH world!
hello SSH world!
$ quit

Connection to localhost closed.