michaellihs · August 27, 2024 17:07 · Jul 29, 2015 · Jul 29, 2015 · Jul 29, 2015
diff --git a/twisted.md b/twisted.md
@@ -191,4 +191,8 @@ 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!
diff --git a/twisted.md b/twisted.md
@@ -182,7 +182,7 @@ python sshserver.py
 ## Connect to the server
 
 ```` shell
-ssh admin@localhost -p 2222
+ssh admin@localhost -p 22222
 
 ((password 'aaa'))
 

diff --git a/twisted.md b/twisted.md
@@ -0,0 +1,194 @@
+# SSH Server with the Python Twisted Library
+
+## Installing the library
+
+Assuming you have Python installed on your system:
+
+```` shell
+pip install twisted
+pip install pyOpenSSL
+pip install service_identity
+````
+
+## Testing the installation
+
+```` shell
+$ 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`
+
+```` python
+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
+
+```` shell
+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
+
+```` shell
+python sshserver.py
+````
+
+## Connect to the server
+
+```` shell
+ssh admin@localhost -p 2222
+
+((password 'aaa'))
+
+>>> Welcome to my test SSH server.
+Commands: clear echo help quit whoami
+$
+````
+
+That's it - hope you like it!