2

我正在尝试使用简单的 ssh 服务器捕获 ssh 暴力破解尝试，代码如下并且可以正常工作，但是我无法将用户名/密码组合与其原始 IP 地址匹配。

到目前为止，这很简单，它没有记录，只是打印到标准输出。我使用 buildProtocol 定义在建立新连接时打印出新连接的 IP 地址。但是，我想获取 IP 地址以及用户名和密码凭据，以便我可以同时跟踪来自不同客户端的多个 ssh 暴力破解尝试。就目前而言，我只能在建立连接时获取 IP 地址，这使得同时跟踪多个连接变得不可能。

只是为了澄清一下，当我提到一个或多个连接时，我并不是指成功登录，这在我的代码中是不可能的，也不是有意的。我指的是与 ssh 服务器的单个连接，它允许在重新连接之前尝试 3 次密码。

from zope.interface import implements
from twisted.conch.unix import UnixSSHRealm
from twisted.cred import portal
from twisted.cred.credentials import IUsernamePassword
from twisted.cred.checkers import ICredentialsChecker
from twisted.cred.error import UnauthorizedLogin
from twisted.conch.ssh import factory, userauth, keys, session
from twisted.internet import reactor, defer

with open('id_rsa') as privateBlobFile:
    privateKey = privateBlobFile.read()

with open('id_rsa.pub') as publicBlobFile:
    publicKey = publicBlobFile.read()

class FailDB:
    credentialInterfaces = IUsernamePassword, implements(ICredentialsChecker)

    def requestAvatarId(self, credentials):
        print"%s:%s" % ( credentials.username, credentials.password)
        return defer.fail(UnauthorizedLogin("invalid password"))

class UnixSSHdFactory(factory.SSHFactory):
    publicKeys = {
        'ssh-rsa': keys.Key.fromString(data=publicKey)
    }
    privateKeys = {
        'ssh-rsa': keys.Key.fromString(data=privateKey)
    }
    services = {
        'ssh-userauth': userauth.SSHUserAuthServer
    }

    def buildProtocol(self, addr):
        print addr
        return factory.SSHFactory.buildProtocol(self, addr)

if __name__ == '__main__':
    portal = portal.Portal(UnixSSHRealm())
    portal.registerChecker(FailDB())
    UnixSSHdFactory.portal = portal
    reactor.listenTCP(2022, UnixSSHdFactory())
    reactor.run()

示例输出：

IPv4Address(TCP, '127.0.0.1', 42141)
root:password123
root:123456
root:letmein
4

1 回答 1

0

您可以通过派生SSHUserAuthServer和覆盖该_ebBadAuth方法从失败的登录尝试中获取地址。例子：

class AuthServer(userauth.SSHUserAuthServer):
    def _ebBadAuth(self, reason):
        addr = self.transport.getPeer().address.host
        print("addr {} failed to log in as {} using {}"
              .format(addr, self.user, self.method))
        userauth.SSHUserAuthServer._ebBadAuth(self, reason)

如果您还希望允许连接成功，则不必覆盖SSHFactory.services，而只需使用派生类更新它，例如factory.services.update({'ssh-userauth': AuthServer}).

于 2015-02-10T20:12:05.530 回答