webssh/tests/test_storage.py

import os
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.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',
            'port': 22,
            '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['directory'], '/var/www')
        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',
            'password': 'new-secret',
            '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.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))