Skip to content

Instantly share code, notes, and snippets.

@michaellihs
Last active August 27, 2024 17:07
Show Gist options
  • Save michaellihs/d2070d7a6d3bb65be18c to your computer and use it in GitHub Desktop.
Save michaellihs/d2070d7a6d3bb65be18c to your computer and use it in GitHub Desktop.
Write your own ssh Server with the Python Twisted library

SSH Server with the Python Twisted Library

Installing the library

Assuming you have Python installed on your system:

pip install twisted
pip install pyOpenSSL
pip install service_identity

Testing the installation

$ python
>>> import twisted
>>> twisted.__version__
'15.2.1'
>>> import OpenSSL
>>> import twisted.internet.ssl
>>> twisted.internet.ssl.SSL
<module 'OpenSSL.SSL' from '/usr/local/lib/python2.7/site-packages/OpenSSL/SSL.pyc'>

Implementing your own ssh-sever with Twisted

Put this script into a file sshserver.py

from twisted.conch import avatar, recvline
from twisted.conch.interfaces import IConchUser, ISession
from twisted.conch.ssh import factory, keys, session
from twisted.conch.insults import insults
from twisted.cred import portal, checkers
from twisted.internet import reactor
from zope.interface import implements
 
class SSHDemoProtocol(recvline.HistoricRecvLine):
    def __init__(self, user):
       self.user = user
 
    def connectionMade(self):
        recvline.HistoricRecvLine.connectionMade(self)
        self.terminal.write("Welcome to my test SSH server.")
        self.terminal.nextLine()
        self.do_help()
        self.showPrompt()
 
    def showPrompt(self):
        self.terminal.write("$ ")
 
    def getCommandFunc(self, cmd):
        return getattr(self, 'do_' + cmd, None)
 
    def lineReceived(self, line):
        line = line.strip()
        if line:
            print line
            f = open('logfile.log', 'w')
            f.write(line + '\n')
            f.close
            cmdAndArgs = line.split()
            cmd = cmdAndArgs[0]
            args = cmdAndArgs[1:]
            func = self.getCommandFunc(cmd)
            if func:
                try:
                    func(*args)
                except Exception, e:
                    self.terminal.write("Error: %s" % e)
                    self.terminal.nextLine()
            else:
                self.terminal.write("No such command.")
                self.terminal.nextLine()
        self.showPrompt()
 
    def do_help(self):
        publicMethods = filter(
            lambda funcname: funcname.startswith('do_'), dir(self))
        commands = [cmd.replace('do_', '', 1) for cmd in publicMethods]
        self.terminal.write("Commands: " + " ".join(commands))
        self.terminal.nextLine()
 
    def do_echo(self, *args):
        self.terminal.write(" ".join(args))
        self.terminal.nextLine()
 
    def do_whoami(self):
        self.terminal.write(self.user.username)
        self.terminal.nextLine()
 
    def do_quit(self):
        self.terminal.write("Thanks for playing!")
        self.terminal.nextLine()
        self.terminal.loseConnection()
 
    def do_clear(self):
        self.terminal.reset()
 
class SSHDemoAvatar(avatar.ConchUser):
    implements(ISession)
 
 
    def __init__(self, username):
        avatar.ConchUser.__init__(self)
        self.username = username
        self.channelLookup.update({'session': session.SSHSession})
 
 
    def openShell(self, protocol):
        serverProtocol = insults.ServerProtocol(SSHDemoProtocol, self)
        serverProtocol.makeConnection(protocol)
        protocol.makeConnection(session.wrapProtocol(serverProtocol))
 
 
    def getPty(self, terminal, windowSize, attrs):
        return None
 
 
    def execCommand(self, protocol, cmd):
        raise NotImplementedError()
 
 
    def closed(self):
        pass
 
 
class SSHDemoRealm(object):
    implements(portal.IRealm)
     
    def requestAvatar(self, avatarId, mind, *interfaces):
        if IConchUser in interfaces:
            return interfaces[0], SSHDemoAvatar(avatarId), lambda: None
        else:
            raise NotImplementedError("No supported interfaces found.")
def getRSAKeys():
 
 
    with open('id_rsa') as privateBlobFile:
        privateBlob = privateBlobFile.read()
        privateKey = keys.Key.fromString(data=privateBlob)
 
 
    with open('id_rsa.pub') as publicBlobFile:
        publicBlob = publicBlobFile.read()
        publicKey = keys.Key.fromString(data=publicBlob)
 
 
    return publicKey, privateKey
 
 
if __name__ == "__main__":
    sshFactory = factory.SSHFactory()
    sshFactory.portal = portal.Portal(SSHDemoRealm())
 
 
users = {'admin': 'aaa', 'guest': 'bbb'}
sshFactory.portal.registerChecker(
    checkers.InMemoryUsernamePasswordDatabaseDontUse(**users))
pubKey, privKey = getRSAKeys()
sshFactory.publicKeys = {'ssh-rsa': pubKey}
sshFactory.privateKeys = {'ssh-rsa': privKey}
reactor.listenTCP(22222, sshFactory)
reactor.run()

Create an RSA key for encryption

ssh-keygen -t rsa -b 4096 -C "[email protected]"

ATTENTION: as a standard, the key gets written into the .ssh directory of the current user - make sure not to override your keys!!!

Start the server

python sshserver.py

Connect to the server

ssh admin@localhost -p 22222
 
((password 'aaa'))
 
>>> Welcome to my test SSH server.
Commands: clear echo help quit whoami
$

Testing commands

The above server is implemented in such a way, that it outputs all commands to STDOUT. So you can redirect the output into a file and after a ssh session assure that the expected commands where called on the server. This is neat for functional testing of programs using an ssh connection.

That's it - hope you like it!

@XxB1a
Copy link

XxB1a commented Mar 25, 2021

Hey. I'm having trouble with this..
Server side error:

Unhandled Error
Traceback (most recent call last):
  File "C:\Users\Demo\AppData\Local\Programs\Python\Python38-32\lib\site-packages\twisted\python\log.py", line 85, in callWithContext
    return context.call({ILogContext: newCtx}, func, *args, **kw)
  File "C:\Users\Demo\AppData\Local\Programs\Python\Python38-32\lib\site-packages\twisted\python\context.py", line 118, in callWithContext
    return self.currentContext().callWithContext(ctx, func, *args, **kw)
  File "C:\Users\Demo\AppData\Local\Programs\Python\Python38-32\lib\site-packages\twisted\python\context.py", line 83, in callWithContext
    return func(*args, **kw)
  File "C:\Users\Demo\AppData\Local\Programs\Python\Python38-32\lib\site-packages\twisted\internet\selectreactor.py", line 149, in _doReadOrWrite
    why = getattr(selectable, method)()
--- <exception caught here> ---
  File "C:\Users\Demo\AppData\Local\Programs\Python\Python38-32\lib\site-packages\twisted\internet\tcp.py", line 1403, in doRead
    protocol.makeConnection(transport)
  File "C:\Users\Demo\AppData\Local\Programs\Python\Python38-32\lib\site-packages\twisted\internet\protocol.py", line 508, in makeConnection
    self.connectionMade()
  File "C:\Users\Demo\AppData\Local\Programs\Python\Python38-32\lib\site-packages\twisted\conch\ssh\transport.py", line 512, in connectionMade
    self.sendKexInit()
  File "C:\Users\Demo\AppData\Local\Programs\Python\Python38-32\lib\site-packages\twisted\conch\ssh\transport.py", line 535, in sendKexInit
    NS(b",".join(self.supportedPublicKeys)),
builtins.TypeError: sequence item 0: expected a bytes-like object, str found

Client sided error:

Received disconnect from 127.0.0.1 port 69:3: couldn't match all kex parts
Disconnected from 127.0.0.1 port 69

C:\Windows\system32>

Command to make an SSH key (no passphrase):

ssh-keygen -t rsa -b 1028

users.json:

{"admin": "admin"}

server.py:

from twisted.conch import avatar, recvline
from twisted.conch.interfaces import IConchUser, ISession
from twisted.conch.ssh import factory, keys, session
from twisted.conch.insults import insults
from twisted.cred import portal, checkers
from twisted.internet import reactor
from zope.interface import implementer
from colorama import *
import json
import os

class SSHDemoProtocol(recvline.HistoricRecvLine):
    def __init__(self, user):
        self.user = user

    def connectionMade(self):
        recvline.HistoricRecvLine.connectionMade(self)

        banner = f"""
                {Fore.RED};) ██╗  █{Fore.YELLOW}█╗███████{Fore.GREEN}╗███╗   █{Fore.CYAN}█╗███████{Fore.BLUE}█╗ █████╗{Fore.MAGENTA} ██╗ :-).
                {Fore.RED};) ██║  █{Fore.YELLOW}█║██╔════{Fore.GREEN}╝████╗  █{Fore.CYAN}█║╚══██╔═{Fore.BLUE}═╝██╔══██{Fore.MAGENTA}╗██║ :-).
                {Fore.RED};) ██████{Fore.YELLOW}█║█████╗ {Fore.GREEN} ██╔██╗ █{Fore.CYAN}█║   ██║ {Fore.BLUE}  ███████{Fore.MAGENTA}║██║ :-).
                {Fore.RED};) ██╔══█{Fore.YELLOW}█║██╔══╝ {Fore.GREEN} ██║╚██╗█{Fore.CYAN}█║   ██║ {Fore.BLUE}  ██╔══██{Fore.MAGENTA}║██║ :-).
                {Fore.RED};) ██║  █{Fore.YELLOW}█║███████{Fore.GREEN}╗██║ ╚███{Fore.CYAN}█║   ██║ {Fore.BLUE}  ██║  ██{Fore.MAGENTA}║██║ :-).
                {Fore.RED};) ╚═╝  ╚{Fore.YELLOW}═╝╚══════{Fore.GREEN}╝╚═╝  ╚══{Fore.CYAN}═╝   ╚═╝ {Fore.BLUE}  ╚═╝  ╚═{Fore.MAGENTA}╝╚═╝ :-).
                {Fore.RESET}"""
        self.terminal.write(banner)
        self.terminal.nextLine()
        self.do_help()
        self.showPrompt()

    def showPrompt(self):
        self.terminal.write("elo")

    def getCommandFunc(self, cmd):
        return getattr(self, 'do_' + cmd, None)

    def lineReceived(self, line):
        line = line.strip()

        if line:
            with open('logfile.log', 'w') as f:
                f.write(f'{line}\n')

            cmdAndArgs = line.split()
            cmd = cmdAndArgs[0]
            args = cmdAndArgs[1:]
            func = self.getCommandFunc(cmd)

            if func:
                try:
                    func(*args)

                except Exception as e:
                    self.terminal.write(f"Error: {e}")
                    self.terminal.nextLine()

            else:
                self.terminal.write(f"{Fore.RED}{Fore.MAGENTA}──────────────────{Fore.RED}{Fore.RESET}")
                self.terminal.write(f"  {Fore.RED}{Fore.CYAN}Invalid command!{Fore.RED}{Fore.RESET}")
                self.terminal.write(f"{Fore.RED}{Fore.MAGENTA}──────────────────{Fore.RED}{Fore.RESET}")
                # █──────────────────█ >>> ^^^
                # ┼ Invalid command! ┼
                # █──────────────────█
                self.terminal.nextLine()

        self.showPrompt()

    def do_help(self):
        publicMethods = filter(lambda funcname: funcname.startswith('do_'), dir(self))
        commands = [cmd.replace('do_', '', 1) for cmd in publicMethods]
        self.terminal.write(f"Commands: {' '.join(commands)}")
        self.terminal.nextLine()

    def do_echo(self, *args):
        self.terminal.write(" ".join(args))
        self.terminal.nextLine()

    def do_whoami(self):
        self.terminal.write(self.user.username)
        self.terminal.nextLine()

    def do_quit(self):
        self.terminal.write(f"{Fore.RED}Cyaaaa")
        self.terminal.nextLine()
        self.terminal.loseConnection()

    def do_clear(self):
        self.terminal.reset()

@implementer(ISession)
class SSHDemoAvatar(avatar.ConchUser):
    def __init__(self, username):
        avatar.ConchUser.__init__(self)
        self.username = username
        self.channelLookup.update({'session': session.SSHSession})

    def openShell(self, protocol):
        serverProtocol = insults.ServerProtocol(SSHDemoProtocol, self)
        serverProtocol.makeConnection(protocol)
        protocol.makeConnection(session.wrapProtocol(serverProtocol))

    def getPty(self, terminal, windowSize, attrs):
        return None

    def execCommand(self, protocol, cmd):
        raise NotImplementedError()

    def closed(self):
        pass

@implementer(portal.IRealm)
class SSHDemoRealm(object):
    def requestAvatar(self, avatarId, mind, *interfaces):
        if IConchUser in interfaces:
            return interfaces[0], SSHDemoAvatar(avatarId), lambda: None

        else:
            raise NotImplementedError("UwU! No supported interfaces found. ;-;")

def getRSAKeys():
    with open(f'C:\\Users\\{os.getlogin()}\\.ssh\\id_rsa') as privateBlobFile:
        privateBlob = privateBlobFile.read()
        privateKey = keys.Key.fromString(data=privateBlob)

    with open(f'C:\\Users\\{os.getlogin()}\\.ssh\\id_rsa.pub') as publicBlobFile:
        publicBlob = publicBlobFile.read()
        publicKey = keys.Key.fromString(data=publicBlob)

    return publicKey, privateKey

if __name__ == "__main__":
    init(convert=True)

    banner = f"""
        {Fore.RED};) ██╗  █{Fore.YELLOW}█╗███████{Fore.GREEN}╗███╗   █{Fore.CYAN}█╗███████{Fore.BLUE}█╗ █████╗{Fore.MAGENTA} ██╗ :-).
        {Fore.RED};) ██║  █{Fore.YELLOW}█║██╔════{Fore.GREEN}╝████╗  █{Fore.CYAN}█║╚══██╔═{Fore.BLUE}═╝██╔══██{Fore.MAGENTA}╗██║ :-).
        {Fore.RED};) ██████{Fore.YELLOW}█║█████╗ {Fore.GREEN} ██╔██╗ █{Fore.CYAN}█║   ██║ {Fore.BLUE}  ███████{Fore.MAGENTA}║██║ :-).
        {Fore.RED};) ██╔══█{Fore.YELLOW}█║██╔══╝ {Fore.GREEN} ██║╚██╗█{Fore.CYAN}█║   ██║ {Fore.BLUE}  ██╔══██{Fore.MAGENTA}║██║ :-).
        {Fore.RED};) ██║  █{Fore.YELLOW}█║███████{Fore.GREEN}╗██║ ╚███{Fore.CYAN}█║   ██║ {Fore.BLUE}  ██║  ██{Fore.MAGENTA}║██║ :-).
        {Fore.RED};) ╚═╝  ╚{Fore.YELLOW}═╝╚══════{Fore.GREEN}╝╚═╝  ╚══{Fore.CYAN}═╝   ╚═╝ {Fore.BLUE}  ╚═╝  ╚═{Fore.MAGENTA}╝╚═╝ :-).
        {Fore.RESET}"""
    print(banner)

    sshFactory = factory.SSHFactory()
    sshFactory.portal = portal.Portal(SSHDemoRealm())

    with open('users.json', 'r') as f:
        uwu = f.readlines()
        uwu = str(uwu)
        uwu = uwu.replace("[\'", "")
        uwu = uwu.replace("\']", "")
        users = json.loads(uwu)

    sshFactory.portal.registerChecker(checkers.InMemoryUsernamePasswordDatabaseDontUse(**users))
    pubKey, privKey = getRSAKeys()
    sshFactory.publicKeys = {'ssh-rsa': pubKey}
    sshFactory.privateKeys = {'ssh-rsa': privKey}
    reactor.listenTCP(69, sshFactory)
    reactor.run()

Note: the port for the SSH server is 69

@XxB1a
Copy link

XxB1a commented Mar 30, 2021

fixxed it
b'ssh-rsa' did the job

@tekasok
Copy link

tekasok commented Jun 26, 2021

i need some help, how can i use time.sleep on it? i tried but it not run on ssh session, they run direct on my vps

@alexisespinosa
Copy link

Another fix {b'session': session.SSHSession}

@lasse-lenting
Copy link

i get access denied

Hey. I'm having trouble with this..
Server side error:

Unhandled Error
Traceback (most recent call last):
  File "C:\Users\Demo\AppData\Local\Programs\Python\Python38-32\lib\site-packages\twisted\python\log.py", line 85, in callWithContext
    return context.call({ILogContext: newCtx}, func, *args, **kw)
  File "C:\Users\Demo\AppData\Local\Programs\Python\Python38-32\lib\site-packages\twisted\python\context.py", line 118, in callWithContext
    return self.currentContext().callWithContext(ctx, func, *args, **kw)
  File "C:\Users\Demo\AppData\Local\Programs\Python\Python38-32\lib\site-packages\twisted\python\context.py", line 83, in callWithContext
    return func(*args, **kw)
  File "C:\Users\Demo\AppData\Local\Programs\Python\Python38-32\lib\site-packages\twisted\internet\selectreactor.py", line 149, in _doReadOrWrite
    why = getattr(selectable, method)()
--- <exception caught here> ---
  File "C:\Users\Demo\AppData\Local\Programs\Python\Python38-32\lib\site-packages\twisted\internet\tcp.py", line 1403, in doRead
    protocol.makeConnection(transport)
  File "C:\Users\Demo\AppData\Local\Programs\Python\Python38-32\lib\site-packages\twisted\internet\protocol.py", line 508, in makeConnection
    self.connectionMade()
  File "C:\Users\Demo\AppData\Local\Programs\Python\Python38-32\lib\site-packages\twisted\conch\ssh\transport.py", line 512, in connectionMade
    self.sendKexInit()
  File "C:\Users\Demo\AppData\Local\Programs\Python\Python38-32\lib\site-packages\twisted\conch\ssh\transport.py", line 535, in sendKexInit
    NS(b",".join(self.supportedPublicKeys)),
builtins.TypeError: sequence item 0: expected a bytes-like object, str found

Client sided error:

Received disconnect from 127.0.0.1 port 69:3: couldn't match all kex parts
Disconnected from 127.0.0.1 port 69

C:\Windows\system32>

Command to make an SSH key (no passphrase):

ssh-keygen -t rsa -b 1028

users.json:

{"admin": "admin"}

server.py:

from twisted.conch import avatar, recvline
from twisted.conch.interfaces import IConchUser, ISession
from twisted.conch.ssh import factory, keys, session
from twisted.conch.insults import insults
from twisted.cred import portal, checkers
from twisted.internet import reactor
from zope.interface import implementer
from colorama import *
import json
import os

class SSHDemoProtocol(recvline.HistoricRecvLine):
    def __init__(self, user):
        self.user = user

    def connectionMade(self):
        recvline.HistoricRecvLine.connectionMade(self)

        banner = f"""
                {Fore.RED};) ██╗  █{Fore.YELLOW}█╗███████{Fore.GREEN}╗███╗   █{Fore.CYAN}█╗███████{Fore.BLUE}█╗ █████╗{Fore.MAGENTA} ██╗ :-).
                {Fore.RED};) ██║  █{Fore.YELLOW}█║██╔════{Fore.GREEN}╝████╗  █{Fore.CYAN}█║╚══██╔═{Fore.BLUE}═╝██╔══██{Fore.MAGENTA}╗██║ :-).
                {Fore.RED};) ██████{Fore.YELLOW}█║█████╗ {Fore.GREEN} ██╔██╗ █{Fore.CYAN}█║   ██║ {Fore.BLUE}  ███████{Fore.MAGENTA}║██║ :-).
                {Fore.RED};) ██╔══█{Fore.YELLOW}█║██╔══╝ {Fore.GREEN} ██║╚██╗█{Fore.CYAN}█║   ██║ {Fore.BLUE}  ██╔══██{Fore.MAGENTA}║██║ :-).
                {Fore.RED};) ██║  █{Fore.YELLOW}█║███████{Fore.GREEN}╗██║ ╚███{Fore.CYAN}█║   ██║ {Fore.BLUE}  ██║  ██{Fore.MAGENTA}║██║ :-).
                {Fore.RED};) ╚═╝  ╚{Fore.YELLOW}═╝╚══════{Fore.GREEN}╝╚═╝  ╚══{Fore.CYAN}═╝   ╚═╝ {Fore.BLUE}  ╚═╝  ╚═{Fore.MAGENTA}╝╚═╝ :-).
                {Fore.RESET}"""
        self.terminal.write(banner)
        self.terminal.nextLine()
        self.do_help()
        self.showPrompt()

    def showPrompt(self):
        self.terminal.write("elo")

    def getCommandFunc(self, cmd):
        return getattr(self, 'do_' + cmd, None)

    def lineReceived(self, line):
        line = line.strip()

        if line:
            with open('logfile.log', 'w') as f:
                f.write(f'{line}\n')

            cmdAndArgs = line.split()
            cmd = cmdAndArgs[0]
            args = cmdAndArgs[1:]
            func = self.getCommandFunc(cmd)

            if func:
                try:
                    func(*args)

                except Exception as e:
                    self.terminal.write(f"Error: {e}")
                    self.terminal.nextLine()

            else:
                self.terminal.write(f"{Fore.RED}{Fore.MAGENTA}──────────────────{Fore.RED}{Fore.RESET}")
                self.terminal.write(f"  {Fore.RED}{Fore.CYAN}Invalid command!{Fore.RED}{Fore.RESET}")
                self.terminal.write(f"{Fore.RED}{Fore.MAGENTA}──────────────────{Fore.RED}{Fore.RESET}")
                # █──────────────────█ >>> ^^^
                # ┼ Invalid command! ┼
                # █──────────────────█
                self.terminal.nextLine()

        self.showPrompt()

    def do_help(self):
        publicMethods = filter(lambda funcname: funcname.startswith('do_'), dir(self))
        commands = [cmd.replace('do_', '', 1) for cmd in publicMethods]
        self.terminal.write(f"Commands: {' '.join(commands)}")
        self.terminal.nextLine()

    def do_echo(self, *args):
        self.terminal.write(" ".join(args))
        self.terminal.nextLine()

    def do_whoami(self):
        self.terminal.write(self.user.username)
        self.terminal.nextLine()

    def do_quit(self):
        self.terminal.write(f"{Fore.RED}Cyaaaa")
        self.terminal.nextLine()
        self.terminal.loseConnection()

    def do_clear(self):
        self.terminal.reset()

@implementer(ISession)
class SSHDemoAvatar(avatar.ConchUser):
    def __init__(self, username):
        avatar.ConchUser.__init__(self)
        self.username = username
        self.channelLookup.update({'session': session.SSHSession})

    def openShell(self, protocol):
        serverProtocol = insults.ServerProtocol(SSHDemoProtocol, self)
        serverProtocol.makeConnection(protocol)
        protocol.makeConnection(session.wrapProtocol(serverProtocol))

    def getPty(self, terminal, windowSize, attrs):
        return None

    def execCommand(self, protocol, cmd):
        raise NotImplementedError()

    def closed(self):
        pass

@implementer(portal.IRealm)
class SSHDemoRealm(object):
    def requestAvatar(self, avatarId, mind, *interfaces):
        if IConchUser in interfaces:
            return interfaces[0], SSHDemoAvatar(avatarId), lambda: None

        else:
            raise NotImplementedError("UwU! No supported interfaces found. ;-;")

def getRSAKeys():
    with open(f'C:\\Users\\{os.getlogin()}\\.ssh\\id_rsa') as privateBlobFile:
        privateBlob = privateBlobFile.read()
        privateKey = keys.Key.fromString(data=privateBlob)

    with open(f'C:\\Users\\{os.getlogin()}\\.ssh\\id_rsa.pub') as publicBlobFile:
        publicBlob = publicBlobFile.read()
        publicKey = keys.Key.fromString(data=publicBlob)

    return publicKey, privateKey

if __name__ == "__main__":
    init(convert=True)

    banner = f"""
        {Fore.RED};) ██╗  █{Fore.YELLOW}█╗███████{Fore.GREEN}╗███╗   █{Fore.CYAN}█╗███████{Fore.BLUE}█╗ █████╗{Fore.MAGENTA} ██╗ :-).
        {Fore.RED};) ██║  █{Fore.YELLOW}█║██╔════{Fore.GREEN}╝████╗  █{Fore.CYAN}█║╚══██╔═{Fore.BLUE}═╝██╔══██{Fore.MAGENTA}╗██║ :-).
        {Fore.RED};) ██████{Fore.YELLOW}█║█████╗ {Fore.GREEN} ██╔██╗ █{Fore.CYAN}█║   ██║ {Fore.BLUE}  ███████{Fore.MAGENTA}║██║ :-).
        {Fore.RED};) ██╔══█{Fore.YELLOW}█║██╔══╝ {Fore.GREEN} ██║╚██╗█{Fore.CYAN}█║   ██║ {Fore.BLUE}  ██╔══██{Fore.MAGENTA}║██║ :-).
        {Fore.RED};) ██║  █{Fore.YELLOW}█║███████{Fore.GREEN}╗██║ ╚███{Fore.CYAN}█║   ██║ {Fore.BLUE}  ██║  ██{Fore.MAGENTA}║██║ :-).
        {Fore.RED};) ╚═╝  ╚{Fore.YELLOW}═╝╚══════{Fore.GREEN}╝╚═╝  ╚══{Fore.CYAN}═╝   ╚═╝ {Fore.BLUE}  ╚═╝  ╚═{Fore.MAGENTA}╝╚═╝ :-).
        {Fore.RESET}"""
    print(banner)

    sshFactory = factory.SSHFactory()
    sshFactory.portal = portal.Portal(SSHDemoRealm())

    with open('users.json', 'r') as f:
        uwu = f.readlines()
        uwu = str(uwu)
        uwu = uwu.replace("[\'", "")
        uwu = uwu.replace("\']", "")
        users = json.loads(uwu)

    sshFactory.portal.registerChecker(checkers.InMemoryUsernamePasswordDatabaseDontUse(**users))
    pubKey, privKey = getRSAKeys()
    sshFactory.publicKeys = {'ssh-rsa': pubKey}
    sshFactory.privateKeys = {'ssh-rsa': privKey}
    reactor.listenTCP(69, sshFactory)
    reactor.run()

Note: the port for the SSH server is 69

i get access denied after fixing all the bugs

Copy link

ghost commented Sep 29, 2021

hi bia

@lasse-lenting
Copy link

is there a way to change the title of a terminal window using twisted?

@MRLPYT
Copy link

MRLPYT commented Jan 11, 2022

Hey, can sb help me? I got this error everytime logged in to the server:

Server sided error:

--- <exception caught here> ---
  File "C:\Users\xnumb\AppData\Roaming\Python\Python310\site-packages\twisted\conch\ssh\connection.py", line 147, in ssh_CHANNEL_OPEN
    channel = self.getChannel(channelType, windowSize, maxPacket, packet)
  File "C:\Users\xnumb\AppData\Roaming\Python\Python310\site-packages\twisted\conch\ssh\connection.py", line 584, in getChannel
    chan = self.transport.avatar.lookupChannel(
  File "C:\Users\xnumb\AppData\Roaming\Python\Python310\site-packages\twisted\conch\avatar.py", line 32, in lookupChannel
    raise ConchError(OPEN_UNKNOWN_CHANNEL_TYPE, "unknown channel")
twisted.conch.error.ConchError: (3, 'unknown channel')

channel open failed
Traceback (most recent call last):
  File "C:\Users\xnumb\AppData\Roaming\Python\Python310\site-packages\twisted\internet\tcp.py", line 252, in _dataReceived
    rval = self.protocol.dataReceived(data)
  File "C:\Users\xnumb\AppData\Roaming\Python\Python310\site-packages\twisted\conch\ssh\transport.py", line 721, in dataReceived
    self.dispatchMessage(messageNum, packet[1:])
  File "C:\Users\xnumb\AppData\Roaming\Python\Python310\site-packages\twisted\conch\ssh\transport.py", line 747, in dispatchMessage
    self.service.packetReceived(messageNum, payload)
  File "C:\Users\xnumb\AppData\Roaming\Python\Python310\site-packages\twisted\conch\ssh\service.py", line 49, in packetReceived
    return f(packet)
--- <exception caught here> ---
  File "C:\Users\xnumb\AppData\Roaming\Python\Python310\site-packages\twisted\conch\ssh\connection.py", line 147, in ssh_CHANNEL_OPEN
    channel = self.getChannel(channelType, windowSize, maxPacket, packet)
  File "C:\Users\xnumb\AppData\Roaming\Python\Python310\site-packages\twisted\conch\ssh\connection.py", line 584, in getChannel
    chan = self.transport.avatar.lookupChannel(
  File "C:\Users\xnumb\AppData\Roaming\Python\Python310\site-packages\twisted\conch\avatar.py", line 32, in lookupChannel
    raise ConchError(OPEN_UNKNOWN_CHANNEL_TYPE, "unknown channel")
twisted.conch.error.ConchError: (3, 'unknown channel')

Client sided error:

C:\Windows\System32>ssh admin@localhost -p 22222
admin@localhost's password:
channel 0: open failed: unknown channel type: unknown channel
Connection to localhost closed.

@ssn-2
Copy link

ssn-2 commented Jan 29, 2022

Please contact me on Telegram if you're interested in creating a C&C project in python. t.me/tcpsyn

@CDWimmer
Copy link

CDWimmer commented May 16, 2022

https://gist.github.com/michaellihs/d2070d7a6d3bb65be18c?permalink_comment_id=3803369#gistcomment-3803369

This was the last fix it needed!

edit: also got add .decode() to a couple of byte strings in certain places. line in a f.write() and the cmd in return getattr(...) of the getCommandFunc method.

another one:

    def do_echo(self, *args):
        self.terminal.write(" ".join([arg.decode() for arg in args]))
        self.terminal.nextLine()

@Zakgeki
Copy link

Zakgeki commented Aug 26, 2022

@MRLPYT

Hey, can sb help me? I got this error everytime logged in to the server:

Server sided error:

--- <exception caught here> ---
  File "C:\Users\xnumb\AppData\Roaming\Python\Python310\site-packages\twisted\conch\ssh\connection.py", line 147, in ssh_CHANNEL_OPEN
    channel = self.getChannel(channelType, windowSize, maxPacket, packet)
  File "C:\Users\xnumb\AppData\Roaming\Python\Python310\site-packages\twisted\conch\ssh\connection.py", line 584, in getChannel
    chan = self.transport.avatar.lookupChannel(
  File "C:\Users\xnumb\AppData\Roaming\Python\Python310\site-packages\twisted\conch\avatar.py", line 32, in lookupChannel
    raise ConchError(OPEN_UNKNOWN_CHANNEL_TYPE, "unknown channel")
twisted.conch.error.ConchError: (3, 'unknown channel')

channel open failed
Traceback (most recent call last):
  File "C:\Users\xnumb\AppData\Roaming\Python\Python310\site-packages\twisted\internet\tcp.py", line 252, in _dataReceived
    rval = self.protocol.dataReceived(data)
  File "C:\Users\xnumb\AppData\Roaming\Python\Python310\site-packages\twisted\conch\ssh\transport.py", line 721, in dataReceived
    self.dispatchMessage(messageNum, packet[1:])
  File "C:\Users\xnumb\AppData\Roaming\Python\Python310\site-packages\twisted\conch\ssh\transport.py", line 747, in dispatchMessage
    self.service.packetReceived(messageNum, payload)
  File "C:\Users\xnumb\AppData\Roaming\Python\Python310\site-packages\twisted\conch\ssh\service.py", line 49, in packetReceived
    return f(packet)
--- <exception caught here> ---
  File "C:\Users\xnumb\AppData\Roaming\Python\Python310\site-packages\twisted\conch\ssh\connection.py", line 147, in ssh_CHANNEL_OPEN
    channel = self.getChannel(channelType, windowSize, maxPacket, packet)
  File "C:\Users\xnumb\AppData\Roaming\Python\Python310\site-packages\twisted\conch\ssh\connection.py", line 584, in getChannel
    chan = self.transport.avatar.lookupChannel(
  File "C:\Users\xnumb\AppData\Roaming\Python\Python310\site-packages\twisted\conch\avatar.py", line 32, in lookupChannel
    raise ConchError(OPEN_UNKNOWN_CHANNEL_TYPE, "unknown channel")
twisted.conch.error.ConchError: (3, 'unknown channel')

Client sided error:

C:\Windows\System32>ssh admin@localhost -p 22222
admin@localhost's password:
channel 0: open failed: unknown channel type: unknown channel
Connection to localhost closed.

I had to specify a byte string in the following function for the class SSHDemoAvatar to fix this bug. Sorry if it's a little late, but I hope it helps!

def __init__(self, username):
    avatar.ConchUser.__init__(self)
    self.username = username
    self.channelLookup.update({b'session': session.SSHSession})

edit: It looks like I missed it but @alexisespinosa mentioned this fix first.

@onlyquestionmark
Copy link

onlyquestionmark commented Sep 15, 2022

Hey.
It seems like there is no supported changing of window size.
builtins.AttributeError: 'SSHDemoAvatar' object has no attribute 'windowChanged'
This error appeared when i tried to resize my putty client on linux.
The login and entering command works fine thanks to @Zakgeki 's result.

This is my code:

from twisted.conch.interfaces import IConchUser, ISession
from twisted.conch.ssh import factory, keys, session
from twisted.conch.insults import insults
from twisted.cred import portal, checkers
from twisted.internet import reactor
from zope.interface import implementer


class SSHDemoProtocol(recvline.HistoricRecvLine):
    def __init__(self, user):
       self.user = user
 
    def connectionMade(self):
        recvline.HistoricRecvLine.connectionMade(self)
        self.terminal.write("Welcome to my test SSH server.")
        self.terminal.nextLine()
        self.do_help()
        self.showPrompt()
 
    def showPrompt(self):
        self.terminal.write(f'$ ')
 
    def getCommandFunc(self, cmd):
        return getattr(self, 'do_' + cmd, None)
 
    def lineReceived(self, line):
        line = line.strip()
        if line:
            print(line)
            f = open('logfile.log', 'w')
            f.write(line + '\n')
            f.close
            cmdAndArgs = line.split()
            cmd = cmdAndArgs[0]
            args = cmdAndArgs[1:]
            func = self.getCommandFunc(cmd)
            if func:
                try:
                    func(*args)
                except Exception as e:
                    self.terminal.write("Error: %s" % e)
                    self.terminal.nextLine()
            else:
                self.terminal.write("No such command.")
                self.terminal.nextLine()
        self.showPrompt()
 
    def do_help(self):
        publicMethods = filter(
            lambda funcname: funcname.startswith('do_'), dir(self))
        commands = [cmd.replace('do_', '', 1) for cmd in publicMethods]
        self.terminal.write("Commands: " + " ".join(commands))
        self.terminal.nextLine()
 
    def do_echo(self, *args):
        self.terminal.write(" ".join(args))
        self.terminal.nextLine()
 
    def do_whoami(self):
        self.terminal.write(self.user.username)
        self.terminal.nextLine()
 
    def do_quit(self):
        self.terminal.write("Thanks for playing!")
        self.terminal.nextLine()
        self.terminal.loseConnection()
 
    def do_clear(self):
        self.terminal.reset()
@implementer(ISession)
class SSHDemoAvatar(avatar.ConchUser):
     
    def __init__(self, username):
        avatar.ConchUser.__init__(self)
        self.username = username
        self.channelLookup.update({b'session': session.SSHSession})
 
 
    def openShell(self, protocol):
        serverProtocol = insults.ServerProtocol(SSHDemoProtocol, self)
        serverProtocol.makeConnection(protocol)
        protocol.makeConnection(session.wrapProtocol(serverProtocol))
 
 
    def getPty(self, terminal, windowSize, attrs):
        return None
 
 
    def execCommand(self, protocol, cmd):
        raise NotImplementedError()
 
 
    def closed(self):
        pass
 
@implementer(portal.IRealm)
class SSHDemoRealm(object):
     
    def requestAvatar(self, avatarId, mind, *interfaces):
        if IConchUser in interfaces:
            return interfaces[0], SSHDemoAvatar(avatarId), lambda: None
        else:
            raise NotImplementedError("No supported interfaces found.")

def getRSAKeys():
    with open(r'id_rsa', "rb") as privateBlobFile:
        privateBlob = privateBlobFile.read()
        privateKey = keys.Key.fromString(data=privateBlob)

    with open(r'id_rsa.pub', "rb") as publicBlobFile:
        publicBlob = publicBlobFile.read()
        publicKey = keys.Key.fromString(data=publicBlob)

    return publicKey, privateKey
 
if __name__ == "__main__":
    sshFactory = factory.SSHFactory()
    sshFactory.portal = portal.Portal(SSHDemoRealm())
 
users = {'admin': b'aaa', 'guest': b'bbb'}
sshFactory.portal.registerChecker(
    checkers.InMemoryUsernamePasswordDatabaseDontUse(**users))
pubKey, privKey = getRSAKeys()
sshFactory.publicKeys = {b'ssh-rsa': pubKey}
sshFactory.privateKeys = {b'ssh-rsa': privKey}
reactor.listenTCP(22222, sshFactory)
reactor.run()```

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment