如何用Python建立SSH连接?

26 投票
3 回答
145077 浏览
提问于 2025-04-16 18:41

有没有人能推荐一个可以在Python中建立SSH连接的工具?我希望它能在任何操作系统上都能用。

我之前试过pyssh,但遇到了一个关于SIGCHLD的错误,我查了一下,发现这是因为Windows不支持这个功能。我还尝试让paramiko工作,但paramiko和Crypto之间出现了错误,导致它们的最新版本根本无法一起使用。

我现在在一台Windows电脑上使用的是Python 2.6.1。

3 个回答

1

在这里补充一下driquet的回答,因为编辑的等待列表已经满了。

对于Python3:

from pexpect import pxssh
s = pxssh.pxssh()
if not s.login ('10.6.192.18', 'root', 'paloalto'):
    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.decode('UTF-8'))     # decode to string and print everything before the prompt. 
    s.logout()

4

Twisted支持SSH功能:http://www.devshed.com/c/a/Python/SSH-with-Twisted/

twisted.conch这个包为Twisted添加了SSH支持。本章将展示如何使用twisted.conch中的模块来构建SSH服务器和客户端。

搭建一个自定义的SSH服务器

命令行是一种非常高效的操作界面,特别适合某些任务。系统管理员喜欢通过输入命令来管理应用程序,而不需要在图形界面中点击。SSH终端更棒,因为它可以从互联网上的任何地方访问。

你可以使用twisted.conch来创建一个SSH服务器,提供一个你定义的自定义命令行。这个命令行还支持一些额外的功能,比如命令历史记录,这样你就可以查看之前输入过的命令。

我该怎么做呢?

你需要写一个twisted.conch.recvline.HistoricRecvLine的子类,来实现你的命令行协议。HistoricRecvLine类似于twisted.protocols.basic.LineReceiver,但提供了更高级的功能来控制终端。

为了通过SSH让你的命令行可用,你需要实现几个不同的类,twisted.conch需要这些类来构建SSH服务器。首先,你需要twisted.cred的认证类:一个门户、凭证检查器和一个返回用户信息的领域。使用twisted.conch.avatar.ConchUser作为你的用户信息类的基类。你的用户信息类还应该实现twisted.conch.interfaces.ISession接口,这个接口包含一个openShell方法,用于创建一个协议来管理用户的交互会话。最后,创建一个twisted.conch.ssh.factory.SSHFactory对象,并将它的portal属性设置为你的门户实例。

示例10-1展示了一个自定义的SSH服务器,它通过用户名和密码来验证用户。它为每个用户提供一个命令行,支持多个命令。

示例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服务器。使用SSH客户端连接到这个服务器,用户名为admin,密码为aaa,试着输入一些命令:

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

请注意,这个在Windows上不适用
模块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://pexpect.readthedocs.io/en/stable/

撰写回答