1# -*- test-case-name: twisted.conch.test.test_manhole -*-
2# Copyright (c) Twisted Matrix Laboratories.
3# See LICENSE for details.
4
5"""
6insults/SSH integration support.
7
8@author: Jp Calderone
9"""
10
11from typing import Dict
12
13from zope.interface import implementer
14
15from twisted.conch import avatar, error as econch, interfaces as iconch
16from twisted.conch.insults import insults
17from twisted.conch.ssh import factory, session
18from twisted.python import components
19
20
21class _Glue:
22    """
23    A feeble class for making one attribute look like another.
24
25    This should be replaced with a real class at some point, probably.
26    Try not to write new code that uses it.
27    """
28
29    def __init__(self, **kw):
30        self.__dict__.update(kw)
31
32    def __getattr__(self, name):
33        raise AttributeError(self.name, "has no attribute", name)
34
35
36class TerminalSessionTransport:
37    def __init__(self, proto, chainedProtocol, avatar, width, height):
38        self.proto = proto
39        self.avatar = avatar
40        self.chainedProtocol = chainedProtocol
41
42        protoSession = self.proto.session
43
44        self.proto.makeConnection(
45            _Glue(
46                write=self.chainedProtocol.dataReceived,
47                loseConnection=lambda: avatar.conn.sendClose(protoSession),
48                name="SSH Proto Transport",
49            )
50        )
51
52        def loseConnection():
53            self.proto.loseConnection()
54
55        self.chainedProtocol.makeConnection(
56            _Glue(
57                write=self.proto.write,
58                loseConnection=loseConnection,
59                name="Chained Proto Transport",
60            )
61        )
62
63        # XXX TODO
64        # chainedProtocol is supposed to be an ITerminalTransport,
65        # maybe.  That means perhaps its terminalProtocol attribute is
66        # an ITerminalProtocol, it could be.  So calling terminalSize
67        # on that should do the right thing But it'd be nice to clean
68        # this bit up.
69        self.chainedProtocol.terminalProtocol.terminalSize(width, height)
70
71
72@implementer(iconch.ISession)
73class TerminalSession(components.Adapter):
74    transportFactory = TerminalSessionTransport
75    chainedProtocolFactory = insults.ServerProtocol
76
77    def getPty(self, term, windowSize, attrs):
78        self.height, self.width = windowSize[:2]
79
80    def openShell(self, proto):
81        self.transportFactory(
82            proto,
83            self.chainedProtocolFactory(),
84            iconch.IConchUser(self.original),
85            self.width,
86            self.height,
87        )
88
89    def execCommand(self, proto, cmd):
90        raise econch.ConchError("Cannot execute commands")
91
92    def windowChanged(self, newWindowSize):
93        # ISession.windowChanged
94        raise NotImplementedError("Unimplemented: TerminalSession.windowChanged")
95
96    def eofReceived(self):
97        # ISession.eofReceived
98        raise NotImplementedError("Unimplemented: TerminalSession.eofReceived")
99
100    def closed(self):
101        # ISession.closed
102        pass
103
104
105class TerminalUser(avatar.ConchUser, components.Adapter):
106    def __init__(self, original, avatarId):
107        components.Adapter.__init__(self, original)
108        avatar.ConchUser.__init__(self)
109        self.channelLookup[b"session"] = session.SSHSession
110
111
112class TerminalRealm:
113    userFactory = TerminalUser
114    sessionFactory = TerminalSession
115
116    transportFactory = TerminalSessionTransport
117    chainedProtocolFactory = insults.ServerProtocol
118
119    def _getAvatar(self, avatarId):
120        comp = components.Componentized()
121        user = self.userFactory(comp, avatarId)
122        sess = self.sessionFactory(comp)
123
124        sess.transportFactory = self.transportFactory
125        sess.chainedProtocolFactory = self.chainedProtocolFactory
126
127        comp.setComponent(iconch.IConchUser, user)
128        comp.setComponent(iconch.ISession, sess)
129
130        return user
131
132    def __init__(self, transportFactory=None):
133        if transportFactory is not None:
134            self.transportFactory = transportFactory
135
136    def requestAvatar(self, avatarId, mind, *interfaces):
137        for i in interfaces:
138            if i is iconch.IConchUser:
139                return (iconch.IConchUser, self._getAvatar(avatarId), lambda: None)
140        raise NotImplementedError()
141
142
143class ConchFactory(factory.SSHFactory):
144    publicKeys: Dict[bytes, bytes] = {}
145    privateKeys: Dict[bytes, bytes] = {}
146
147    def __init__(self, portal):
148        self.portal = portal
149