如何使用python建立ssh连接？

Question

任何人都可以推荐一些在python中进行SSH连接吗？ 我需要它与任何操作系统兼容。

我已经试过pyssh只是为了得到与SIGCHLD，我读过的错误是因为Windows缺乏这个。 我试过让paramiko工作，但我之间paramiko和加密之间的错误，其中每个最新版本不会在一起工作的地步。

Python 2.6.1目前在Windows机器上。

Answer 1

模块pxssh不正是你想要的。
例如，运行 'ls -l命令'，并打印输出，你需要做这样的事情：

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() 

一些链接：
Pxssh文档：http://dsnra.jpl.nasa.gov/software/Python/site-packages/Contrib/pxssh.html
Pexpect的（pxssh基于pexpect）：http://www.noah.org/wiki/pexpect

Answer 2

扭曲了SSH支持：http://www.devshed.com/c/a/Python/SSH-with-Twisted/

的twisted.conch包添加SSH支持扭曲。本章介绍如何使用twisted.conch中的模块构建SSH服务器和客户端。

设置自定义SSH服务器

的命令行是对某些任务令人难以置信的高效的接口。系统管理员喜欢通过输入命令来管理应用程序的功能，而无需点击图形用户界面。 SSH shell甚至更好，因为它可以从互联网上的任何地方访问。

您可以使用twisted.conch创建一个SSH服务器，该服务器使用您定义的命令提供对定制shell的访问。这个shell甚至会支持一些额外的功能，比如命令历史记录，这样你就可以滚动你已经输入的命令。

我该怎么做？ 编写一个twisted.conch.recvline.HistoricRecvLine的子类，实现您的shell协议。 HistoricRecvLine类似于twisted.protocols.basic.LineReceiver，但具有控制终端的更高级功能。

编写twisted.conch.recvline.HistoricRecvLine的一个子类，实现您的shell协议。 HistoricRecvLine类似于twisted.protocols.basic.LineReceiver，但具有控制终端的更高级功能。

要使您的shell可以通过SSH获得，您需要实现twisted.conch需要构建SSH服务器的几个不同的类。首先，您需要twisted.cred认证类：门户，凭证检查器和返回头像的领域。使用twisted.conch.avatar.ConchUser作为头像的基类。你的头像类还应该实现twisted.conch.interfaces.ISession，其中包括一个openShell方法，在该方法中创建一个协议来管理用户的交互式会话。最后，创建一个twisted.conch.ssh.factory.SSHFactory对象并将其门户属性设置为您的门户实例。

示例10-1演示了一个定制的SSH服务器，它通过用户名和密码对用户进行身份验证。它为每个用户提供一个提供多个命令的shell。

例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将在端口2222上运行SSH服务器。连接到该服务器使用的用户名admin和密码AAA SSH客户端，并尝试输入一些命令：

$ 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.