1
# Copyright (c) 2001-2004 Twisted Matrix Laboratories.
2
# See LICENSE for details.
4
# You can run this .tac file directly with:
5
# twistd -ny demo_scroll.tac
7
"""Simple echo-ish server that uses the scroll-region.
9
This demo sets up two listening ports: one on 6022 which accepts ssh
10
connections; one on 6023 which accepts telnet connections. No login
11
for the telnet server is required; for the ssh server, \"username\" is
12
the username and \"password\" is the password.
14
The TerminalProtocol subclass defined here sets up a scroll-region occupying
15
most of the screen. It positions the cursor at the bottom of the screen and
16
then echos back printable input. When return is received, the line is
17
copied to the upper area of the screen (scrolling anything older up) and
18
clears the input line.
23
from twisted.python import log
24
from twisted.internet import protocol
25
from twisted.application import internet, service
26
from twisted.cred import checkers, portal
28
from twisted.conch.insults import insults
29
from twisted.conch.telnet import TelnetTransport, TelnetBootstrapProtocol
30
from twisted.conch.manhole_ssh import ConchFactory, TerminalRealm
32
class DemoProtocol(insults.TerminalProtocol):
33
"""Copies input to an upwards scrolling region.
38
def connectionMade(self):
40
self.terminalSize(self.width, self.height)
43
def terminalSize(self, width, height):
47
self.terminal.setScrollRegion(0, height - 1)
48
self.terminal.cursorPosition(0, height)
49
self.terminal.write('> ')
51
def unhandledControlSequence(self, seq):
52
log.msg("Client sent something weird: %r" % (seq,))
54
def keystrokeReceived(self, keyID, modifier):
56
self.terminal.cursorPosition(0, self.height - 2)
57
self.terminal.nextLine()
58
self.terminal.write(''.join(self.buffer))
59
self.terminal.cursorPosition(0, self.height - 1)
60
self.terminal.eraseToLineEnd()
61
self.terminal.write('> ')
63
elif keyID in list(string.printable):
64
self.terminal.write(keyID)
65
self.buffer.append(keyID)
68
def makeService(args):
69
checker = checkers.InMemoryUsernamePasswordDatabaseDontUse(username="password")
71
f = protocol.ServerFactory()
72
f.protocol = lambda: TelnetTransport(TelnetBootstrapProtocol,
73
insults.ServerProtocol,
74
args['protocolFactory'],
75
*args.get('protocolArgs', ()),
76
**args.get('protocolKwArgs', {}))
77
tsvc = internet.TCPServer(args['telnet'], f)
79
def chainProtocolFactory():
80
return insults.ServerProtocol(
81
args['protocolFactory'],
82
*args.get('protocolArgs', ()),
83
**args.get('protocolKwArgs', {}))
86
rlm.chainedProtocolFactory = chainProtocolFactory
87
ptl = portal.Portal(rlm, [checker])
89
csvc = internet.TCPServer(args['ssh'], f)
91
m = service.MultiService()
92
tsvc.setServiceParent(m)
93
csvc.setServiceParent(m)
96
application = service.Application("Scroll Region Demo App")
98
makeService({'protocolFactory': DemoProtocol,
100
'ssh': 6022}).setServiceParent(application)