I am going the book twisted network programming essentials 2nd edition by Jessica McKellar and Abe Fetting. I am stuck on the following twisted cred auth example 9-1. I have double checked my code line by line, letter by letter. The code will run and when I connect with a telnet client and enter 'user pass' it will not log me in. I just get the not authorized message. The script it self does not generate any error message or exceptions.
Thanks
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()
You have a typo in tryLogin
(ProtocolAvatar
-> IProtocolAvatar
):
def tryLogin(self, username, password):
self.portal.login(credentials.UsernamePassword(username, password), None, IProtocolAvatar).addCallbacks(self._cbLogin, self._ebLogin)
# ^
UPDATE
There is another typo: IPortocolAvatar
-> IProtocolAvatar
.
After run the server, try following test client code:
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())