Jocay
178 lines
8.1 KiB
Python
178 lines
8.1 KiB
Python
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)
|