使用twisted.conch跟踪ssh登录尝试的IP

Question

我正在尝试使用一个简单的ssh服务器捕获ssh bruteforce尝试，代码如下并且工作正常，但是我无法将用户名/密码组合与它们的原始IP地址相匹配。

到目前为止，这是我能得到的最简单的结果，它还没有记录到stdout中。在建立新连接时，我使用buildProtocol定义打印出新连接的IP地址。但是，我希望获得IP地址以及用户名和密码凭据，以便可以同时跟踪来自不同客户端的多个ssh bruteforce尝试。目前，我只能在建立连接时获得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

Answer 1

通过从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}) )更新它。