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/avatar.py (1641B)

# -*- test-case-name: twisted.conch.test.test_conch -*-

from zope.interface import implementer

from twisted.conch.error import ConchError
from twisted.conch.interfaces import IConchUser
from twisted.conch.ssh.connection import OPEN_UNKNOWN_CHANNEL_TYPE
from twisted.logger import Logger
from twisted.python.compat import nativeString

@implementer(IConchUser)
class ConchUser:
    _log = Logger()

def __init__(self):
        self.channelLookup = {}
        self.subsystemLookup = {}

@property
    def conn(self):
        return self._conn

@conn.setter
    def conn(self, value):
        self._conn = value

def lookupChannel(self, channelType, windowSize, maxPacket, data):
        klass = self.channelLookup.get(channelType, None)
        if not klass:
            raise ConchError(OPEN_UNKNOWN_CHANNEL_TYPE, "unknown channel")
        else:
            return klass(
                remoteWindow=windowSize,
                remoteMaxPacket=maxPacket,
                data=data,
                avatar=self,
            )

def lookupSubsystem(self, subsystem, data):
        self._log.debug(
            "Subsystem lookup: {subsystem!r}", subsystem=self.subsystemLookup
        )
        klass = self.subsystemLookup.get(subsystem, None)
        if not klass:
            return False
        return klass(data, avatar=self)

def gotGlobalRequest(self, requestType, data):
        # XXX should this use method dispatch?
        requestType = nativeString(requestType.replace(b"-", b"_"))
        f = getattr(self, "global_%s" % requestType, None)
        if not f:
            return 0
        return f(data)

# -*- test-case-name: twisted.conch.test.test_conch -*- from zope.interface import implementer from twisted.conch.error import ConchError from twisted.conch.interfaces import IConchUser from twisted.conch.ssh.connection import OPEN_UNKNOWN_CHANNEL_TYPE from twisted.logger import Logger from twisted.python.compat import nativeString @implementer(IConchUser) class ConchUser: _log = Logger() def __init__(self): self.channelLookup = {} self.subsystemLookup = {} @property def conn(self): return self._conn @conn.setter def conn(self, value): self._conn = value def lookupChannel(self, channelType, windowSize, maxPacket, data): klass = self.channelLookup.get(channelType, None) if not klass: raise ConchError(OPEN_UNKNOWN_CHANNEL_TYPE, "unknown channel") else: return klass( remoteWindow=windowSize, remoteMaxPacket=maxPacket, data=data, avatar=self, ) def lookupSubsystem(self, subsystem, data): self._log.debug( "Subsystem lookup: {subsystem!r}", subsystem=self.subsystemLookup ) klass = self.subsystemLookup.get(subsystem, None) if not klass: return False return klass(data, avatar=self) def gotGlobalRequest(self, requestType, data): # XXX should this use method dispatch? requestType = nativeString(requestType.replace(b"-", b"_")) f = getattr(self, "global_%s" % requestType, None) if not f: return 0 return f(data)