[Twisted-Python] I'm stuck

Curt curtferguson at cfl.rr.com
Tue May 6 09:56:22 EDT 2008


Ok, first, I'm a true newbie, I've been playing with twisted, and python
for that matter, for about 3 days.  I've been trying to get this script
to work, and I *know* I've got to be doing something simple wrong.  All
it is, is a slap-together of various examples available in the twisted
documentation to see if I can learn how this works.  Here's the script

All it is supposed to do at this point is accept a normal telnet
connection on one port and authenticate it, and an ssh on another, and
authenticate, I'm attempting to write an interpreter to connect both
inputs to, but right now I'm stuck getting the connections to hold.

Any assistance would be most welcome.

from zope.interface import implements

from twisted.internet import protocol
from twisted.application import service, strports
from twisted.conch.ssh import session
from twisted.conch import interfaces as iconch
from twisted.cred import portal, checkers
from twisted.python import usage

from twisted.conch.insults import insults
from twisted.conch import recvline, manhole_ssh, telnet

class parseInput(recvline.HistoricRecvLine):

    def connectionMade(self):
        print("Got Connection to parseInput")
        recvline.HistoricRecvLine.connectionMade(self)
        print("Came back from the undercall")
        self.interpreter=commandInterpreter

    def handle_QUIT(self):
        self.terminal.loseConnection()

    def lineReceived(self, line):
        print("Input, " + line)
        self.terminal.write("I got, " + line)

class makeTelnetProtocol:
    def __init__(self, portal):
        self.portal = portal

    def __call__(self):
        auth = telnet.AuthenticatingTelnetProtocol
        args = (self.portal,)
        return telnet.TelnetTransport(auth, *args)

class chainedProtocolFactory:
    def __init__(self, namespace):
        self.namespace = namespace

    def __call__(self):
        return insults.ServerProtocol(parseInput, self.namespace)

class _BaseTelnetRealm:
    implements(portal.IRealm)

    def __init__(self, proto, *a, **kw):
        self.protocolFactory = proto
        self.protocolArgs = a
        self.protocolKwArgs = kw

    def requestAvatar(self, avatarId, *interfaces):
        if telnet.ITelnetProtocol in interfaces:
            return (telnet.ITelnetProtocol,
                    self.protocolFactory(*self.protocolArgs,
**self.protocolKwArgs),
                    lambda: None)
        raise NotImplementedError()

class Options(usage.Options):
    optParameters = [
        ["telnetPort", "t", None, "strports description of the address
on which to listen for telnet connections"],
        ["sshPort", "s", None, "strports description of the address on
which to listen for ssh connections"]]

    def __init__(self):
        usage.Options.__init__(self)
        self.users = []
        self['namespace'] = None

    def opt_user(self, name):
        self.users.append(name)

    def postOptions(self):
        if self['telnetPort'] is None and self['sshPort'] is None:
            raise usage.UsageError("At least one of --telnetPort and
--sshPort must be specified")

def makeService(options):
    """Create a server service.

    @type options: C{dict}
    @param options: A mapping describing the configuration of
    the desired service.  Recognized key/value pairs are::

        "telnetPort": strports description of the address on which
                      to listen for telnet connections.  If None,
                      no telnet service will be started.

        "sshPort": strports description of the address on which to
                   listen for ssh connections.  If None, no ssh
                   service will be started.

        "namespace": dictionary containing desired initial locals
                     for manhole connections.  If None, an empty
                     dictionary will be used.

    @rtype: L{twisted.application.service.IService}
    @return: A server service.
    """

    svc = service.MultiService()

    namespace = options['namespace']
    if namespace is None:
        namespace = {}

    checker = checkers.InMemoryUsernamePasswordDatabaseDontUse()
    checker.addUser("guest", "testing")

    if options['telnetPort']:
        telnetRealm = _BaseTelnetRealm(telnet.TelnetBootstrapProtocol,
                                   insults.ServerProtocol,
                                   parseInput,
                                   namespace)

        telnetPortal = portal.Portal(telnetRealm, [checker])

        telnetFactory = protocol.ServerFactory()
        telnetFactory.protocol = makeTelnetProtocol(telnetPortal)
        telnetService = strports.service(options['telnetPort'],
                                         telnetFactory)
        telnetService.setServiceParent(svc)

    if options['sshPort']:
        sshRealm = manhole_ssh.TerminalRealm()
        sshRealm.chainedProtocolFactory = chainedProtocolFactory(namespace)

        sshPortal = portal.Portal(sshRealm, [checker])
        sshFactory = manhole_ssh.ConchFactory(sshPortal)
        sshService = strports.service(options['sshPort'],
                                      sshFactory)
        sshService.setServiceParent(svc)

    return svc

application = service.Application("Learn Twisted")
makeService({"telnetPort": "tcp:9900",
             "sshPort": "tcp:9901",
             "namespace": {"foo": "bar"}}).setServiceParent(application)





More information about the Twisted-Python mailing list