Prepare WebSSH for self-hosted deployment

tests/test_app.py

@@ -99,6 +99,7 @@ class TestAppBasic(TestAppBase):
         options.syshostfile = ''
         options.tdstream = ''
         options.delay = 0.1
+        options.auth = False
         app = make_app(make_handlers(loop, options), get_app_settings(options))
         return app
 
@@ -536,6 +537,7 @@ class OtherTestBase(TestAppBase):
         options.tdstream = self.tdstream
         options.maxconn = self.maxconn
         options.origin = self.origin
+        options.auth = False
         app = make_app(make_handlers(loop, options), get_app_settings(options))
         return app
 

tests/test_auth.py

@@ -0,0 +1,36 @@
+import os
+import shutil
+import tempfile
+import unittest
+
+from webssh.auth import AuthManager, make_password_hash, verify_password
+
+
+class TestAuth(unittest.TestCase):
+
+    def setUp(self):
+        self.tmpdir = tempfile.mkdtemp()
+
+    def tearDown(self):
+        shutil.rmtree(self.tmpdir)
+
+    def test_password_hash(self):
+        encoded = make_password_hash('secret')
+        self.assertTrue(verify_password('secret', encoded))
+        self.assertFalse(verify_password('wrong', encoded))
+        self.assertFalse(verify_password('', encoded))
+
+    def test_auth_manager_setup_and_verify(self):
+        auth_file = os.path.join(self.tmpdir, 'auth.json')
+        manager = AuthManager(auth_file=auth_file)
+        self.assertFalse(manager.is_configured())
+
+        manager.setup('admin', 'secret')
+        self.assertTrue(manager.is_configured())
+        self.assertTrue(manager.verify('admin', 'secret'))
+        self.assertFalse(manager.verify('admin', 'wrong'))
+        self.assertFalse(manager.verify('other', 'secret'))
+
+        loaded = AuthManager(auth_file=auth_file)
+        self.assertTrue(loaded.verify('admin', 'secret'))
+

tests/test_auth_app.py

@@ -0,0 +1,64 @@
+import json
+import shutil
+import tempfile
+
+from tornado.options import options
+from tornado.testing import AsyncHTTPTestCase
+
+from webssh.main import make_app, make_handlers
+from webssh.settings import get_app_settings
+from webssh.utils import to_str
+
+
+class TestAppAuth(AsyncHTTPTestCase):
+
+    def setUp(self):
+        self.tmpdir = tempfile.mkdtemp()
+        super(TestAppAuth, self).setUp()
+
+    def tearDown(self):
+        super(TestAppAuth, self).tearDown()
+        shutil.rmtree(self.tmpdir)
+
+    def get_app(self):
+        loop = self.io_loop
+        options.auth = True
+        options.auth_username = 'admin'
+        options.auth_password = ''
+        options.auth_password_hash = ''
+        options.auth_session_days = 7
+        options.data_dir = self.tmpdir
+        options.debug = False
+        options.xsrf = False
+        options.policy = 'warning'
+        options.hostfile = ''
+        options.syshostfile = ''
+        options.tdstream = ''
+        options.origin = 'same'
+        handlers = make_handlers(loop, options)
+        return make_app(handlers, get_app_settings(options))
+
+    def get_cookie_header(self, response):
+        cookie = response.headers.get_list('Set-Cookie')[0]
+        return {'Cookie': cookie.split(';', 1)[0]}
+
+    def test_auth_setup_and_saved_connections_endpoint(self):
+        response = self.fetch('/', follow_redirects=False)
+        self.assertEqual(response.code, 302)
+        self.assertIn('/login', response.headers['Location'])
+
+        response = self.fetch('/login')
+        self.assertIn('创建管理员账号'.encode('utf-8'), response.body)
+
+        body = 'username=admin&password=secret&confirm=secret'
+        response = self.fetch('/login', method='POST', body=body,
+                              follow_redirects=False)
+        self.assertEqual(response.code, 302)
+        headers = self.get_cookie_header(response)
+
+        response = self.fetch('/', headers=headers)
+        self.assertIn('已保存连接'.encode('utf-8'), response.body)
+
+        response = self.fetch('/connections', headers=headers)
+        data = json.loads(to_str(response.body))
+        self.assertEqual(data, {'connections': []})

tests/test_storage.py

@@ -0,0 +1,46 @@
+import os
+import shutil
+import tempfile
+import unittest
+
+from webssh.storage import ConnectionStore
+
+
+class TestConnectionStore(unittest.TestCase):
+
+    def setUp(self):
+        self.tmpdir = tempfile.mkdtemp()
+        self.store = ConnectionStore(os.path.join(self.tmpdir,
+                                                  'connections.json'))
+
+    def tearDown(self):
+        shutil.rmtree(self.tmpdir)
+
+    def test_upsert_list_and_delete(self):
+        profile = self.store.upsert('admin', {
+            'hostname': '127.0.0.1',
+            'port': 22,
+            'username': 'root',
+            'term': 'xterm-256color',
+            'auth_type': 'password'
+        })
+
+        self.assertEqual(profile['title'], 'root@127.0.0.1:22')
+        self.assertEqual([profile], self.store.list('admin'))
+        self.assertEqual([], self.store.list('other'))
+
+        updated = self.store.upsert('admin', {
+            'hostname': '127.0.0.1',
+            'port': 22,
+            'username': 'root',
+            'term': 'xterm',
+            'auth_type': 'privatekey'
+        })
+
+        self.assertEqual(profile['id'], updated['id'])
+        self.assertEqual(1, len(self.store.list('admin')))
+        self.assertEqual('xterm', self.store.list('admin')[0]['term'])
+        self.assertTrue(self.store.delete('admin', profile['id']))
+        self.assertFalse(self.store.delete('admin', profile['id']))
+        self.assertEqual([], self.store.list('admin'))
+