doc/conch/examples/demo_scroll.tac

   1 # Copyright (c) Twisted Matrix Laboratories.
   2 # See LICENSE for details.
   3
   4 # You can run this .tac file directly with:
   5 #    twistd -ny demo_scroll.tac
   6
   7 """Simple echo-ish server that uses the scroll-region.
   8
   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.
  13
  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.
  19 """
  20
  21 import string
  22
  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
  27
  28 from twisted.conch.insults import insults
  29 from twisted.conch.telnet import TelnetTransport, TelnetBootstrapProtocol
  30 from twisted.conch.manhole_ssh import ConchFactory, TerminalRealm
  31
  32 class DemoProtocol(insults.TerminalProtocol):
  33     """Copies input to an upwards scrolling region.
  34     """
  35     width = 80
  36     height = 24
  37
  38     def connectionMade(self):
  39         self.buffer = []
  40         self.terminalSize(self.width, self.height)
  41
  42     # ITerminalListener
  43     def terminalSize(self, width, height):
  44         self.width = width
  45         self.height = height
  46
  47         self.terminal.setScrollRegion(0, height - 1)
  48         self.terminal.cursorPosition(0, height)
  49         self.terminal.write('> ')
  50
  51     def unhandledControlSequence(self, seq):
  52         log.msg("Client sent something weird: %r" % (seq,))
  53
  54     def keystrokeReceived(self, keyID, modifier):
  55         if keyID == '\r':
  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('> ')
  62             self.buffer = []
  63         elif keyID in list(string.printable):
  64             self.terminal.write(keyID)
  65             self.buffer.append(keyID)
  66
  67
  68 def makeService(args):
  69     checker = checkers.InMemoryUsernamePasswordDatabaseDontUse(username="password")
  70
  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)
  78
  79     def chainProtocolFactory():
  80         return insults.ServerProtocol(
  81             args['protocolFactory'],
  82             *args.get('protocolArgs', ()),
  83             **args.get('protocolKwArgs', {}))
  84
  85     rlm = TerminalRealm()
  86     rlm.chainedProtocolFactory = chainProtocolFactory
  87     ptl = portal.Portal(rlm, [checker])
  88     f = ConchFactory(ptl)
  89     csvc = internet.TCPServer(args['ssh'], f)
  90
  91     m = service.MultiService()
  92     tsvc.setServiceParent(m)
  93     csvc.setServiceParent(m)
  94     return m
  95
  96 application = service.Application("Scroll Region Demo App")
  97
  98 makeService({'protocolFactory': DemoProtocol,
  99              'telnet': 6023,
 100              'ssh': 6022}).setServiceParent(application)