Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
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 "lihs@punkt.de"

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!

@ignertic

This comment has been minimized.

Copy link

@ignertic ignertic commented Mar 23, 2019

Awesome thanks

@Aakash134

This comment has been minimized.

Copy link

@Aakash134 Aakash134 commented Jun 5, 2019

Not able to login
The server reply's
admin@127.0.0.1's password: aaa
Access denied
made changes to run on python 3

@jrtechs

This comment has been minimized.

Copy link

@jrtechs jrtechs commented Aug 25, 2019

@Aakash134 what changes did you make?

@sumitgpl

This comment has been minimized.

Copy link

@sumitgpl sumitgpl commented Mar 12, 2020

@Aakash134 what changes did you make?

@ralbra

This comment has been minimized.

Copy link

@ralbra ralbra commented Apr 7, 2020

@sumitgpl
I also tried to make it work under Python 3 and ended up having trouble with the login not being accepted.
My guess is that something is going wrong the registerChecker but i dont know.
Thats my code:

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
 
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 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({'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'C:\workspace\SSH-Server\id_rsa', "rb") as privateBlobFile:
        privateBlob = privateBlobFile.read()
        privateKey = keys.Key.fromString(data=privateBlob)
 
 
    with open(r'C:\workspace\SSH-Server\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': 'aaa', 'guest': '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()
@unkn0wn-user

This comment has been minimized.

Copy link

@unkn0wn-user unkn0wn-user commented Nov 13, 2020

@ralbra
Have you got it working in python3 correctly ?
I'm currently facing the same issue with my login not being accepted whilst username and password are absolutely correct.

Running the same code in python2 everything seems to be working fine.

@rwp40

This comment has been minimized.

Copy link

@rwp40 rwp40 commented Nov 13, 2020

@ralbra
Have you got it working in python3 correctly ?
I'm currently facing the same issue with my login not being accepted whilst username and password are absolutely correct.

Running the same code in python2 everything seems to be working fine.

Try:
users = {'admin': b'aaa', 'guest': b'bbb'}

@ralbra

This comment has been minimized.

Copy link

@ralbra ralbra commented Nov 14, 2020

Hi, I solved this by simply using the original twited ssh server example: https://twistedmatrix.com/documents/current/conch/examples/

The only major difference i can see on the fly is that user and password have to be a byte strings so
users = {b'admin': b'aaa', b'guest': b'bbb'}

Hope this helps.

@unkn0wn-user

This comment has been minimized.

Copy link

@unkn0wn-user unkn0wn-user commented Nov 17, 2020

Thanks a lot for your help, it now works like a charm.

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