feat: Implement directory validation and shell command quoting
- Added is_valid_directory function to validate directory paths, ensuring they do not contain control characters and are within a specified length. - Introduced quote_shell_arg to safely quote shell arguments, preventing command injection. - Created build_cd_command to generate a command for changing directories in a shell. - Enhanced the LoginHandler to utilize a login rate limiter, preventing brute-force attacks by tracking failed login attempts. - Implemented an EncodingCache to optimize encoding detection for SSH connections. - Updated the UI to include an input field for specifying an initial directory upon login, with appropriate validation and hints. - Added a quickbar in the terminal interface for easy access to copy and paste functionality. - Introduced a toast notification system to provide feedback on copy actions. - Refactored connection storage to encrypt passwords at rest, improving security. - Updated various templates and styles to accommodate new features and improve user experience.
This commit is contained in:
Jocay
+150
-4
@@ -3,19 +3,27 @@ import shutil
|
||||
import tempfile
|
||||
import unittest
|
||||
|
||||
from webssh.crypto import is_encrypted
|
||||
from webssh.storage import ConnectionStore
|
||||
|
||||
|
||||
SECRET = 'test-server-secret'
|
||||
|
||||
|
||||
class TestConnectionStore(unittest.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
self.tmpdir = tempfile.mkdtemp()
|
||||
self.store = ConnectionStore(os.path.join(self.tmpdir,
|
||||
'connections.json'))
|
||||
self.filename = os.path.join(self.tmpdir, 'connections.json')
|
||||
self.store = ConnectionStore(self.filename, SECRET)
|
||||
|
||||
def tearDown(self):
|
||||
shutil.rmtree(self.tmpdir)
|
||||
|
||||
def read_raw(self):
|
||||
with open(self.filename, encoding='utf-8') as f:
|
||||
return f.read()
|
||||
|
||||
def test_upsert_list_and_delete(self):
|
||||
profile = self.store.upsert('admin', {
|
||||
'hostname': '127.0.0.1',
|
||||
@@ -23,11 +31,12 @@ class TestConnectionStore(unittest.TestCase):
|
||||
'username': 'root',
|
||||
'password': 'root-secret',
|
||||
'term': 'xterm-256color',
|
||||
'directory': '/var/www',
|
||||
'auth_type': 'password'
|
||||
})
|
||||
|
||||
self.assertEqual(profile['title'], 'root@127.0.0.1:22')
|
||||
self.assertEqual(profile['password'], 'root-secret')
|
||||
self.assertEqual(profile['directory'], '/var/www')
|
||||
self.assertEqual([profile], self.store.list('admin'))
|
||||
self.assertEqual([], self.store.list('other'))
|
||||
|
||||
@@ -43,7 +52,144 @@ class TestConnectionStore(unittest.TestCase):
|
||||
self.assertEqual(profile['id'], updated['id'])
|
||||
self.assertEqual(1, len(self.store.list('admin')))
|
||||
self.assertEqual('xterm', self.store.list('admin')[0]['term'])
|
||||
self.assertEqual('new-secret', self.store.list('admin')[0]['password'])
|
||||
self.assertEqual('', self.store.list('admin')[0]['directory'])
|
||||
self.assertTrue(self.store.delete('admin', profile['id']))
|
||||
self.assertFalse(self.store.delete('admin', profile['id']))
|
||||
self.assertEqual([], self.store.list('admin'))
|
||||
|
||||
def test_password_never_leaves_the_store(self):
|
||||
profile = self.store.upsert('admin', {
|
||||
'hostname': '127.0.0.1',
|
||||
'username': 'root',
|
||||
'password': 'root-secret'
|
||||
})
|
||||
|
||||
self.assertNotIn('password', profile)
|
||||
self.assertNotIn('password_enc', profile)
|
||||
self.assertTrue(profile['has_password'])
|
||||
|
||||
listed = self.store.list('admin')[0]
|
||||
self.assertNotIn('password', listed)
|
||||
self.assertNotIn('password_enc', listed)
|
||||
self.assertTrue(listed['has_password'])
|
||||
|
||||
# ... but it is still usable server side
|
||||
self.assertEqual(
|
||||
'root-secret', self.store.get_password('admin', profile['id'])
|
||||
)
|
||||
|
||||
def test_password_is_encrypted_at_rest(self):
|
||||
self.store.upsert('admin', {
|
||||
'hostname': '127.0.0.1',
|
||||
'username': 'root',
|
||||
'password': 'root-secret'
|
||||
})
|
||||
|
||||
raw = self.read_raw()
|
||||
self.assertNotIn('root-secret', raw)
|
||||
self.assertIn('password_enc', raw)
|
||||
|
||||
def test_password_is_scoped_to_owner_and_destination(self):
|
||||
profile = self.store.upsert('admin', {
|
||||
'hostname': '127.0.0.1',
|
||||
'username': 'root',
|
||||
'password': 'root-secret'
|
||||
})
|
||||
|
||||
self.assertEqual(
|
||||
'', self.store.get_password('mallory', profile['id'])
|
||||
)
|
||||
self.assertEqual('', self.store.get_password('admin', 'deadbeef'))
|
||||
|
||||
# an edited hostname yields a different id, so nothing is found
|
||||
other_id = self.store.make_id('admin', 'evil.example.com', 22, 'root')
|
||||
self.assertNotEqual(profile['id'], other_id)
|
||||
self.assertEqual('', self.store.get_password('admin', other_id))
|
||||
|
||||
def test_profile_without_password(self):
|
||||
profile = self.store.upsert('admin', {
|
||||
'hostname': '127.0.0.1',
|
||||
'username': 'root'
|
||||
})
|
||||
self.assertFalse(profile['has_password'])
|
||||
self.assertEqual('', self.store.get_password('admin', profile['id']))
|
||||
|
||||
def test_a_wrong_secret_does_not_break_the_store(self):
|
||||
profile = self.store.upsert('admin', {
|
||||
'hostname': '127.0.0.1',
|
||||
'username': 'root',
|
||||
'password': 'root-secret'
|
||||
})
|
||||
|
||||
rotated = ConnectionStore(self.filename, 'a-different-secret')
|
||||
self.assertEqual('', rotated.get_password('admin', profile['id']))
|
||||
self.assertTrue(rotated.list('admin')[0]['has_password'])
|
||||
|
||||
def test_encrypt_plaintext_passwords_migrates_old_files(self):
|
||||
legacy = ConnectionStore(self.filename)
|
||||
profile = legacy.upsert('admin', {
|
||||
'hostname': '127.0.0.1',
|
||||
'username': 'root',
|
||||
'password': 'legacy-secret'
|
||||
})
|
||||
self.assertIn('legacy-secret', self.read_raw())
|
||||
self.assertEqual(
|
||||
'legacy-secret', legacy.get_password('admin', profile['id'])
|
||||
)
|
||||
|
||||
self.assertEqual(1, self.store.encrypt_plaintext_passwords())
|
||||
|
||||
raw = self.read_raw()
|
||||
self.assertNotIn('legacy-secret', raw)
|
||||
self.assertEqual(
|
||||
'legacy-secret', self.store.get_password('admin', profile['id'])
|
||||
)
|
||||
self.assertTrue(self.store.list('admin')[0]['has_password'])
|
||||
|
||||
# already migrated, nothing left to do
|
||||
self.assertEqual(0, self.store.encrypt_plaintext_passwords())
|
||||
|
||||
def test_directory_is_normalized_and_optional(self):
|
||||
profile = self.store.upsert('admin', {
|
||||
'hostname': 'example.com',
|
||||
'username': 'deploy',
|
||||
'directory': ' /srv/app '
|
||||
})
|
||||
self.assertEqual('/srv/app', profile['directory'])
|
||||
|
||||
profile = self.store.upsert('admin', {
|
||||
'hostname': 'example.org',
|
||||
'username': 'deploy'
|
||||
})
|
||||
self.assertEqual('', profile['directory'])
|
||||
|
||||
|
||||
class TestSecretBox(unittest.TestCase):
|
||||
|
||||
def test_round_trip(self):
|
||||
from webssh.crypto import SecretBox
|
||||
|
||||
box = SecretBox(SECRET)
|
||||
token = box.encrypt('hunter2')
|
||||
self.assertTrue(is_encrypted(token))
|
||||
self.assertNotIn('hunter2', token)
|
||||
self.assertEqual('hunter2', box.decrypt(token))
|
||||
|
||||
def test_nonce_is_not_reused(self):
|
||||
from webssh.crypto import SecretBox
|
||||
|
||||
box = SecretBox(SECRET)
|
||||
self.assertNotEqual(box.encrypt('same'), box.encrypt('same'))
|
||||
|
||||
def test_empty_and_tampered_values(self):
|
||||
from webssh.crypto import SecretBox
|
||||
|
||||
box = SecretBox(SECRET)
|
||||
self.assertEqual('', box.encrypt(''))
|
||||
self.assertEqual('', box.decrypt(''))
|
||||
self.assertEqual('', box.decrypt('not-a-token'))
|
||||
self.assertEqual('', box.decrypt('aesgcm$####'))
|
||||
|
||||
token = box.encrypt('hunter2')
|
||||
self.assertEqual('', box.decrypt(token[:-4] + 'AAAA'))
|
||||
self.assertEqual('', SecretBox('other').decrypt(token))
|
||||
|
||||
Reference in New Issue
Block a user