2

我正在阅读 Jessica McKellar 和 Abe Fetting 所著的《twisted network programming essentials 2nd edition》。我被困在以下扭曲的信用身份验证示例 9-1 上。我逐行、逐字地仔细检查了我的代码。代码将运行，当我连接到 telnet 客户端并输入“用户密码”时，它不会让我登录。我只是收到未授权的消息。它自己的脚本不会生成任何错误消息或异常。

谢谢

import sys
from zope.interface import implements, Interface

from twisted.cred import checkers, credentials, portal
from twisted.internet import protocol, reactor
from twisted.protocols import basic
from twisted.python.log import startLogging; startLogging(sys.stdout)

class IPortocolAvatar(Interface):
    def logout():
        """
        Clean up per-login resource allocated to this avatar.
        """

class EchoAvatar(object):
    implements(IPortocolAvatar)

    def logout(self):
        pass

class Echo(basic.LineReceiver):
    portal = None
    avatar = None
    logout = None

    def connectionLost(self, reason):
        if self.logout:
            self.logout()
            self.avatar = None
            self.logout = None

    def lineReceived(self, line):
        if not self.avatar:
            # print "line [32]: %s" % (line,)
            username, password = line.strip().split(" ")
            self.tryLogin(username, password)

        else:
            self.sendLine(line)

    def tryLogin(self, username, password):
        self.portal.login(credentials.UsernamePassword(username, password), None, PortocolAvatar).addCallbacks(self._cbLogin, self._ebLogin)

    def _cbLogin(self, (interface, avatar, logout)):
        self.avatar = avatar
        self.logout = logout
        self.sendLine("Login sucessful, please proceed.")

    def _ebLogin(self, failure):
        self.sendLine("Login denied, goodbye.")
        self.transport.loseConnection()

class EchoFactory(protocol.Factory):
    def __init__(self, portal):
        self.portal = portal

    def buildProtocol(self, addr):
        proto = Echo()
        proto.portal = self.portal
        return proto

class Realm(object):
    implements(portal.IRealm)

    def requestAvatar(self, avatarId, mind, *interfaces):
        if IPortocolAvatar in interfaces:
            avatar = EchoAvatar()
            return IPortocolAvatar, avatar, avatar.logout

        raise NotImplementedError(
            "This realm only supports the IPortocolAvatar interface.")

realm = Realm()
myPortal = portal.Portal(realm)
checker = checkers.InMemoryUsernamePasswordDatabaseDontUse()
checker.addUser("user", "pass")
myPortal.registerChecker(checker)

reactor.listenTCP(8000, EchoFactory(myPortal))
reactor.run()
4

2 回答 2

3

tryLogin您在( ProtocolAvatar-> )中有错字IProtocolAvatar：

def tryLogin(self, username, password):
    self.portal.login(credentials.UsernamePassword(username, password), None, IProtocolAvatar).addCallbacks(self._cbLogin, self._ebLogin)
    #                                                                         ^

更新

还有一个错字：IPortocolAvatar-> IProtocolAvatar。

运行服务器后，尝试以下测试客户端代码：

import telnetlib
import time

t = telnetlib.Telnet('localhost', 8000)
t.write(b'user pass\r\n')
t.write(b'blah blah\r\n')
time.sleep(1)
print(t.read_eager())
于 2013-10-04T01:52:48.653 回答
0

我还让示例使用 python3 运行。这是我为“示例 9-1”更改的代码。echo_cred.py 来自“Twisted Network Programming Essentials, 2nd Edition”一书

from zope.interface import Interface, implementer
from twisted.cred import checkers, credentials, portal, error
from twisted.internet import protocol, reactor, defer

from twisted.protocols import basic

class IProtocolAvatar(Interface):
    def logout():
        """
        Clean up per-login resources allocated to this avatar.
        """

@implementer(IProtocolAvatar)    
class EchoAvatar():

    def logout(self):
        pass

class Echo(basic.LineReceiver):
    portal = None
    avatar = None
    logout = None

    def connectionLost(self, reason):
        if self.logout:
            self.logout()
            self.avatar = None
            self.logout = None

    def lineReceived(self, line):
        print('line received')
        if not self.avatar:
            username, password = line.strip().split()
            # convert 2 text code
            username, password = username.decode(), password.decode()

            d = defer.maybeDeferred(self.tryLogin, username, password)
            d.addCallbacks(self._cbLogin, self._ebLogin)
        else:
            self.sendLine(line)

    def tryLogin(self, username, password):
        if self.portal is not None:
            return self.portal.login(
                credentials.UsernamePassword(username, password),
                None,
                IProtocolAvatar
            )
        raise error.UnauthorizedLogin()        

    def _cbLogin(self, ial):
        interface, avatar, logout = ial
        self.avatar = avatar
        self.logout = logout
        # convert 2 byte code
        self.sendLine("Login successful, please proceed.".encode())

    def _ebLogin(self, failure):
        self.sendLine("Login denied, goodbye.".encode())
        self.transport.loseConnection()

class EchoFactory(protocol.Factory):
    def __init__(self, portal):
        self.portal = portal

    def buildProtocol(self, addr):
        proto = Echo()
        proto.portal = self.portal
        return proto

@implementer(portal.IRealm)    
class Realm():

    def requestAvatar(self, avatarId, mind, *interfaces):
        if IProtocolAvatar in interfaces:
            avatar = EchoAvatar()
            return IProtocolAvatar, avatar, avatar.logout
        raise NotImplementedError("This realm only supports the IProtocolAvatar interface.")

if __name__ == '__main__':
    realm = Realm()
    myPortal = portal.Portal(realm)
    checker = checkers.InMemoryUsernamePasswordDatabaseDontUse()
    checker.addUser("user", "pass")
    myPortal.registerChecker(checker)
    reactor.listenTCP(8000, EchoFactory(myPortal))
    reactor.run()
于 2018-04-23T09:56:47.653 回答