import io import json import shutil import stat import tempfile import threading import unittest from unittest.mock import Mock 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.worker import clients from webssh.workspace import ( TRANSFER_LIMIT, decrypt_backup, encrypt_backup, transfer, validate_path ) class TestBackup(unittest.TestCase): def test_roundtrip_wrong_password_and_tampering(self): connections = [{'hostname': 'example.com', 'password': 'secret'}] document = encrypt_backup(connections, 'long-backup-password') self.assertNotIn('secret', json.dumps(document)) self.assertEqual(connections, decrypt_backup(document, 'long-backup-password')) with self.assertRaises(ValueError): decrypt_backup(document, 'wrong-backup-password') document['data'] = document['data'][:-4] + 'AAAA' with self.assertRaises(ValueError): decrypt_backup(document, 'long-backup-password') def test_short_password_is_rejected(self): with self.assertRaises(ValueError): encrypt_backup([], 'short') class TestTransfer(unittest.TestCase): def make_sftp(self): ssh, sftp = Mock(), Mock() ssh.open_sftp.return_value = sftp sftp.normalize.side_effect = lambda path: path return ssh, sftp def test_list_sorts_directories_before_files(self): ssh, sftp = self.make_sftp() sftp.listdir_iter.return_value = [ Mock(filename='file', st_mode=stat.S_IFREG, st_size=10, st_mtime=0), Mock(filename='dir', st_mode=stat.S_IFDIR, st_size=0, st_mtime=0) ] result = transfer(ssh, 'list', '/home') self.assertEqual(['dir', 'file'], [item['name'] for item in result['entries']]) sftp.close.assert_called_once() sftp.get_channel().settimeout.assert_called_once_with(30) def test_upload_is_exclusive_and_rejects_path_in_filename(self): ssh, sftp = self.make_sftp() remote = Mock() sftp.open.return_value.__enter__ = Mock(return_value=remote) sftp.open.return_value.__exit__ = Mock(return_value=False) transfer(ssh, 'upload', '/home', ('file.txt', b'content')) sftp.open.assert_called_once_with('/home/file.txt', 'wx') remote.write.assert_called_once_with(b'content') for filename in ('../secret', 'a/b', 'a\\b', '..'): with self.assertRaises(ValueError): transfer(ssh, 'upload', '/home', (filename, b'content')) def test_download_enforces_size_and_regular_file(self): ssh, sftp = self.make_sftp() sftp.stat.return_value = Mock(st_mode=stat.S_IFREG, st_size=3) sftp.open.return_value = io.BytesIO(b'abc') self.assertEqual(b'abc', transfer(ssh, 'download', '/home/file')) sftp.stat.return_value.st_size = TRANSFER_LIMIT + 1 with self.assertRaises(ValueError): transfer(ssh, 'download', '/home/file') sftp.stat.return_value = Mock(st_mode=stat.S_IFDIR, st_size=0) with self.assertRaises(ValueError): transfer(ssh, 'download', '/home') def test_sftp_is_closed_when_upload_fails(self): ssh, sftp = self.make_sftp() sftp.open.side_effect = OSError('Already exists') with self.assertRaises(OSError): transfer(ssh, 'upload', '/home', ('existing', b'new')) sftp.close.assert_called_once() def test_paths_reject_control_characters(self): for path in ('a\x00b', 'a\nb', 'x' * 4097): with self.assertRaises(ValueError): validate_path(path) class TestWorkspaceAPI(AsyncHTTPTestCase): def setUp(self): self.tmpdir = tempfile.mkdtemp() super(TestWorkspaceAPI, self).setUp() self.store = self._app.settings['connection_store'] def tearDown(self): clients.pop('user:admin', None) super(TestWorkspaceAPI, self).tearDown() shutil.rmtree(self.tmpdir) def get_app(self): options.auth = True options.auth_username = 'admin' options.auth_password = 'login-password' options.auth_password_hash = '' options.auth_session_days = 7 options.data_dir = self.tmpdir options.debug = False options.xsrf = True options.policy = 'warning' options.hostfile = options.syshostfile = options.tdstream = '' options.origin = 'same' return make_app(make_handlers(self.io_loop, options), get_app_settings(options)) def headers(self): from tornado.web import create_signed_value cookie = create_signed_value(self._app.settings['cookie_secret'], 'webssh_user', 'admin') return {'Cookie': 'webssh_user=' + cookie.decode('utf-8') + '; _xsrf=token', 'X-XSRFToken': 'token', 'Content-Type': 'application/json'} def post_json(self, path, data, headers=None): return self.fetch(path, method='POST', body=json.dumps(data), headers=headers or self.headers(), follow_redirects=False) def test_plain_export_and_metadata_keep_password_private(self): profile = self.store.upsert('admin', {'hostname': 'example.com', 'username': 'root', 'password': 'secret'}) response = self.post_json('/connections/' + profile['id'] + '/metadata', {'name': 'Production'}) self.assertEqual(200, response.code) self.assertNotIn(b'password_enc', response.body) response = self.post_json('/connections/export', {}) self.assertEqual(200, response.code) self.assertNotIn(b'secret', response.body) self.assertEqual('Production', json.loads(response.body)['connections'][0]['name']) def test_backup_is_encrypted_and_can_restore(self): self.store.upsert('admin', {'hostname': 'example.com', 'username': 'root', 'privatekey': 'key-material'}) response = self.post_json('/connections/export', {'credentials': True, 'password': 'long-backup-password'}) self.assertEqual(200, response.code) self.assertNotIn(b'key-material', response.body) document = json.loads(response.body) response = self.post_json('/connections/import', {'document': document, 'password': 'wrong-password'}) self.assertEqual(400, response.code) response = self.post_json('/connections/import', {'document': document, 'password': 'long-backup-password'}) self.assertEqual(200, response.code) self.assertNotIn(b'privatekey_enc', response.body) def test_private_routes_require_login_and_xsrf_and_same_origin(self): for path in ('/connections/export', '/connections/import'): response = self.post_json(path, {}, {'Content-Type': 'application/json', 'Cookie': '_xsrf=token', 'X-XSRFToken': 'token'}) self.assertEqual(403, response.code) headers = self.headers() headers.pop('X-XSRFToken') self.assertEqual(403, self.post_json('/connections/export', {}, headers).code) headers = self.headers() headers['Origin'] = 'https://other.example' self.assertEqual(403, self.post_json('/connections/export', {}, headers).code) def test_file_worker_is_scoped_to_user_and_rejects_busy_sessions(self): response = self.fetch('/files/not-owned', headers=self.headers()) self.assertEqual(409, response.code) worker = Mock(closed=False, handler=Mock(), file_lock=threading.Lock()) worker.file_lock.acquire() clients['user:admin'] = {'busy': worker} response = self.fetch('/files/busy', headers=self.headers()) self.assertEqual(409, response.code) worker.file_lock.release() def test_workspace_controls_are_present(self): response = self.fetch('/', headers=self.headers()) for element in ('session-tabs', 'terminal-search', 'terminal-theme', 'command-list', 'files-panel', 'import-config', 'connection-search', 'connection-group'): self.assertIn(('id="' + element + '"').encode('utf-8'), response.body) self.assertIn(b'workspace.js', response.body)