FM / conch

/usr/lib/python3/dist-packages/twisted/conch/

/usr/lib/python3/dist-packages/twisted/conch

Name	Size	Mode	Actions
client/	-	0755	rm
insults/	-	0755	rm
openssh_compat/	-	0755	rm
scripts/	-	0755	rm
ssh/	-	0755	rm
test/	-	0755	rm
ui/	-	0755	rm
__pycache__/	-	0755	rm
avatar.py	1641	0644	edit dl rm
checkers.py	19422	0644	edit dl rm
endpoints.py	29961	0644	edit dl rm
error.py	2659	0644	edit dl rm
interfaces.py	14918	0644	edit dl rm
ls.py	2693	0644	edit dl rm
manhole.py	11844	0644	edit dl rm
manhole_ssh.py	4427	0644	edit dl rm
manhole_tap.py	5484	0644	edit dl rm
mixin.py	1373	0644	edit dl rm
recvline.py	19163	0644	edit dl rm
stdio.py	2773	0644	edit dl rm
tap.py	3190	0644	edit dl rm
telnet.py	38054	0644	edit dl rm
ttymodes.py	2195	0644	edit dl rm
unix.py	16550	0644	edit dl rm
__init__.py	198	0644	edit dl rm

Edit: /usr/lib/python3/dist-packages/twisted/conch/stdio.py (2773B)

# -*- test-case-name: twisted.conch.test.test_manhole -*-
# Copyright (c) Twisted Matrix Laboratories.
# See LICENSE for details.

"""
Asynchronous local terminal input handling

@author: Jp Calderone
"""

import os
import sys
import termios
import tty

from twisted.conch.insults.insults import ServerProtocol
from twisted.conch.manhole import ColoredManhole
from twisted.internet import defer, protocol, reactor, stdio
from twisted.python import failure, log, reflect

class UnexpectedOutputError(Exception):
    pass

class TerminalProcessProtocol(protocol.ProcessProtocol):
    def __init__(self, proto):
        self.proto = proto
        self.onConnection = defer.Deferred()

def connectionMade(self):
        self.proto.makeConnection(self)
        self.onConnection.callback(None)
        self.onConnection = None

def write(self, data):
        """
        Write to the terminal.

@param data: Data to write.
        @type data: L{bytes}
        """
        self.transport.write(data)

def outReceived(self, data):
        """
        Receive data from the terminal.

@param data: Data received.
        @type data: L{bytes}
        """
        self.proto.dataReceived(data)

def errReceived(self, data):
        """
        Report an error.

@param data: Data to include in L{Failure}.
        @type data: L{bytes}
        """
        self.transport.loseConnection()
        if self.proto is not None:
            self.proto.connectionLost(failure.Failure(UnexpectedOutputError(data)))
            self.proto = None

def childConnectionLost(self, childFD):
        if self.proto is not None:
            self.proto.childConnectionLost(childFD)

def processEnded(self, reason):
        if self.proto is not None:
            self.proto.connectionLost(reason)
            self.proto = None

class ConsoleManhole(ColoredManhole):
    """
    A manhole protocol specifically for use with L{stdio.StandardIO}.
    """

def connectionLost(self, reason):
        """
        When the connection is lost, there is nothing more to do.  Stop the
        reactor so that the process can exit.
        """
        reactor.stop()

def runWithProtocol(klass):
    fd = sys.__stdin__.fileno()
    oldSettings = termios.tcgetattr(fd)
    tty.setraw(fd)
    try:
        stdio.StandardIO(ServerProtocol(klass))
        reactor.run()
    finally:
        termios.tcsetattr(fd, termios.TCSANOW, oldSettings)
        os.write(fd, b"\r\x1bc\r")

def main(argv=None):
    log.startLogging(open("child.log", "w"))

if argv is None:
        argv = sys.argv[1:]
    if argv:
        klass = reflect.namedClass(argv[0])
    else:
        klass = ConsoleManhole
    runWithProtocol(klass)

if __name__ == "__main__":
    main()

# -*- test-case-name: twisted.conch.test.test_manhole -*- # Copyright (c) Twisted Matrix Laboratories. # See LICENSE for details. """ Asynchronous local terminal input handling @author: Jp Calderone """ import os import sys import termios import tty from twisted.conch.insults.insults import ServerProtocol from twisted.conch.manhole import ColoredManhole from twisted.internet import defer, protocol, reactor, stdio from twisted.python import failure, log, reflect class UnexpectedOutputError(Exception): pass class TerminalProcessProtocol(protocol.ProcessProtocol): def __init__(self, proto): self.proto = proto self.onConnection = defer.Deferred() def connectionMade(self): self.proto.makeConnection(self) self.onConnection.callback(None) self.onConnection = None def write(self, data): """ Write to the terminal. @param data: Data to write. @type data: L{bytes} """ self.transport.write(data) def outReceived(self, data): """ Receive data from the terminal. @param data: Data received. @type data: L{bytes} """ self.proto.dataReceived(data) def errReceived(self, data): """ Report an error. @param data: Data to include in L{Failure}. @type data: L{bytes} """ self.transport.loseConnection() if self.proto is not None: self.proto.connectionLost(failure.Failure(UnexpectedOutputError(data))) self.proto = None def childConnectionLost(self, childFD): if self.proto is not None: self.proto.childConnectionLost(childFD) def processEnded(self, reason): if self.proto is not None: self.proto.connectionLost(reason) self.proto = None class ConsoleManhole(ColoredManhole): """ A manhole protocol specifically for use with L{stdio.StandardIO}. """ def connectionLost(self, reason): """ When the connection is lost, there is nothing more to do. Stop the reactor so that the process can exit. """ reactor.stop() def runWithProtocol(klass): fd = sys.__stdin__.fileno() oldSettings = termios.tcgetattr(fd) tty.setraw(fd) try: stdio.StandardIO(ServerProtocol(klass)) reactor.run() finally: termios.tcsetattr(fd, termios.TCSANOW, oldSettings) os.write(fd, b"\r\x1bc\r") def main(argv=None): log.startLogging(open("child.log", "w")) if argv is None: argv = sys.argv[1:] if argv: klass = reflect.namedClass(argv[0]) else: klass = ConsoleManhole runWithProtocol(klass) if __name__ == "__main__": main()