aboutsummaryrefslogtreecommitdiffstats
path: root/contrib/python/Twisted/py2/twisted/conch/manhole_tap.py
blob: b5fb78b60ca7d39918d9c69110d4f5d370c8ad47 (plain) (blame)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
# Copyright (c) Twisted Matrix Laboratories.
# See LICENSE for details.

"""
TAP plugin for creating telnet- and ssh-accessible manhole servers.

@author: Jp Calderone
"""

from zope.interface import implementer

from twisted.internet import protocol
from twisted.application import service, strports
from twisted.cred import portal, checkers
from twisted.python import usage, filepath

from twisted.conch import manhole, manhole_ssh, telnet
from twisted.conch.insults import insults
from twisted.conch.ssh import keys



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)



@implementer(portal.IRealm)
class _StupidRealm:
    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"],
        ["sshKeyDir", None, "<USER DATA DIR>",
         "Directory where the autogenerated SSH key is kept."],
        ["sshKeyName", None, "server.key",
         "Filename of the autogenerated SSH key."],
        ["sshKeySize", None, 4096,
         "Size of the automatically generated SSH key."],
    ]

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

    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: L{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.

        "sshKeyDir": The folder that the SSH server key will be kept in.

        "sshKeyName": The filename of the key.

        "sshKeySize": The size of the key, in bits. Default is 4096.

    @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)

        if options['sshKeyDir'] != "<USER DATA DIR>":
            keyDir = options['sshKeyDir']
        else:
            from twisted.python._appdirs import getDataDirectory
            keyDir = getDataDirectory()

        keyLocation = filepath.FilePath(keyDir).child(options['sshKeyName'])

        sshKey = keys._getPersistentRSAKey(keyLocation,
                                           int(options['sshKeySize']))
        sshFactory.publicKeys[b"ssh-rsa"] = sshKey
        sshFactory.privateKeys[b"ssh-rsa"] = sshKey

        sshService = strports.service(options['sshPort'], sshFactory)
        sshService.setServiceParent(svc)

    return svc