Viewing file:  manhole_tap.py (4.3 KB) -rw-r--r--
Select action/file-type:
 (+) |  (+) |  (+) | Code (+) | Session (+) |  (+) | SDB (+) |  (+) |  (+) |  (+) |  (+) |  (+) |
# Copyright (c) 2001-2004 Twisted Matrix Laboratories.
# See LICENSE for details.
"""
TAP plugin for creating telnet- and ssh-accessible manhole servers.
@author: Jp Calderone
"""
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 manhole, manhole_ssh, telnet
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(manhole.ColoredManhole, self.namespace)
class _StupidRealm:
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"],
["passwd", "p", "/etc/passwd", "name of a passwd(5)-format username/password file"]]
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 manhole 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.
"passwd": Name of a passwd(5)-format username/password file.
@rtype: L{twisted.application.service.IService}
@return: A manhole service.
"""
svc = service.MultiService()
namespace = options['namespace']
if namespace is None:
namespace = {}
checker = checkers.FilePasswordDB(options['passwd'])
if options['telnetPort']:
telnetRealm = _StupidRealm(telnet.TelnetBootstrapProtocol,
insults.ServerProtocol,
manhole.ColoredManhole,
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
|