summaryrefslogtreecommitdiff
path: root/testing/tests/server
diff options
context:
space:
mode:
Diffstat (limited to 'testing/tests/server')
-rw-r--r--testing/tests/server/__init__.py0
-rw-r--r--testing/tests/server/test_server.py536
2 files changed, 536 insertions, 0 deletions
diff --git a/testing/tests/server/__init__.py b/testing/tests/server/__init__.py
new file mode 100644
index 00000000..e69de29b
--- /dev/null
+++ b/testing/tests/server/__init__.py
diff --git a/testing/tests/server/test_server.py b/testing/tests/server/test_server.py
new file mode 100644
index 00000000..b99d1939
--- /dev/null
+++ b/testing/tests/server/test_server.py
@@ -0,0 +1,536 @@
+# -*- coding: utf-8 -*-
+# test_server.py
+# Copyright (C) 2013 LEAP
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program. If not, see <http://www.gnu.org/licenses/>.
+"""
+Tests for server-related functionality.
+"""
+import binascii
+import mock
+import os
+import tempfile
+
+from hashlib import sha512
+from pkg_resources import resource_filename
+from urlparse import urljoin
+from uuid import uuid4
+
+from twisted.internet import defer
+from twisted.trial import unittest
+
+from leap.soledad.common.couch.state import CouchServerState
+from leap.soledad.common.couch import CouchDatabase
+from test_soledad.u1db_tests import TestCaseWithServer
+from test_soledad.util import CouchDBTestCase
+from test_soledad.util import (
+ make_token_soledad_app,
+ make_soledad_document_for_test,
+ soledad_sync_target,
+ BaseSoledadTest,
+)
+
+from leap.soledad.common import crypto
+from leap.soledad.client import Soledad
+from leap.soledad.server import load_configuration
+from leap.soledad.server import CONFIG_DEFAULTS
+from leap.soledad.server.auth import URLToAuthorization
+from leap.soledad.server.auth import SoledadTokenAuthMiddleware
+
+
+class ServerAuthenticationMiddlewareTestCase(CouchDBTestCase):
+
+ def setUp(self):
+ super(ServerAuthenticationMiddlewareTestCase, self).setUp()
+ app = mock.Mock()
+ self._state = CouchServerState(self.couch_url)
+ app.state = self._state
+ self.auth_middleware = SoledadTokenAuthMiddleware(app)
+ self._authorize('valid-uuid', 'valid-token')
+
+ def _authorize(self, uuid, token):
+ token_doc = {}
+ token_doc['_id'] = sha512(token).hexdigest()
+ token_doc[self._state.TOKENS_USER_ID_KEY] = uuid
+ token_doc[self._state.TOKENS_TYPE_KEY] = \
+ self._state.TOKENS_TYPE_DEF
+ dbname = self._state._tokens_dbname()
+ db = self.couch_server.create(dbname)
+ db.save(token_doc)
+ self.addCleanup(self.delete_db, db.name)
+
+ def test_authorized_user(self):
+ is_authorized = self.auth_middleware._verify_authentication_data
+ self.assertTrue(is_authorized('valid-uuid', 'valid-token'))
+ self.assertFalse(is_authorized('valid-uuid', 'invalid-token'))
+ self.assertFalse(is_authorized('invalid-uuid', 'valid-token'))
+ self.assertFalse(is_authorized('eve', 'invalid-token'))
+
+
+class ServerAuthorizationTestCase(BaseSoledadTest):
+
+ """
+ Tests related to Soledad server authorization.
+ """
+
+ def setUp(self):
+ pass
+
+ def tearDown(self):
+ pass
+
+ def _make_environ(self, path_info, request_method):
+ return {
+ 'PATH_INFO': path_info,
+ 'REQUEST_METHOD': request_method,
+ }
+
+ def test_verify_action_with_correct_dbnames(self):
+ """
+ Test encrypting and decrypting documents.
+
+ The following table lists the authorized actions among all possible
+ u1db remote actions:
+
+ URL path | Authorized actions
+ --------------------------------------------------
+ / | GET
+ /shared-db | GET
+ /shared-db/docs | -
+ /shared-db/doc/{id} | GET, PUT, DELETE
+ /shared-db/sync-from/{source} | -
+ /user-db | GET, PUT, DELETE
+ /user-db/docs | -
+ /user-db/doc/{id} | -
+ /user-db/sync-from/{source} | GET, PUT, POST
+ """
+ uuid = uuid4().hex
+ authmap = URLToAuthorization(uuid,)
+ dbname = authmap._user_db_name
+ # test global auth
+ self.assertTrue(
+ authmap.is_authorized(self._make_environ('/', 'GET')))
+ # test shared-db database resource auth
+ self.assertTrue(
+ authmap.is_authorized(
+ self._make_environ('/shared', 'GET')))
+ self.assertFalse(
+ authmap.is_authorized(
+ self._make_environ('/shared', 'PUT')))
+ self.assertFalse(
+ authmap.is_authorized(
+ self._make_environ('/shared', 'DELETE')))
+ self.assertFalse(
+ authmap.is_authorized(
+ self._make_environ('/shared', 'POST')))
+ # test shared-db docs resource auth
+ self.assertFalse(
+ authmap.is_authorized(
+ self._make_environ('/shared/docs', 'GET')))
+ self.assertFalse(
+ authmap.is_authorized(
+ self._make_environ('/shared/docs', 'PUT')))
+ self.assertFalse(
+ authmap.is_authorized(
+ self._make_environ('/shared/docs', 'DELETE')))
+ self.assertFalse(
+ authmap.is_authorized(
+ self._make_environ('/shared/docs', 'POST')))
+ # test shared-db doc resource auth
+ self.assertTrue(
+ authmap.is_authorized(
+ self._make_environ('/shared/doc/x', 'GET')))
+ self.assertTrue(
+ authmap.is_authorized(
+ self._make_environ('/shared/doc/x', 'PUT')))
+ self.assertTrue(
+ authmap.is_authorized(
+ self._make_environ('/shared/doc/x', 'DELETE')))
+ self.assertFalse(
+ authmap.is_authorized(
+ self._make_environ('/shared/doc/x', 'POST')))
+ # test shared-db sync resource auth
+ self.assertFalse(
+ authmap.is_authorized(
+ self._make_environ('/shared/sync-from/x', 'GET')))
+ self.assertFalse(
+ authmap.is_authorized(
+ self._make_environ('/shared/sync-from/x', 'PUT')))
+ self.assertFalse(
+ authmap.is_authorized(
+ self._make_environ('/shared/sync-from/x', 'DELETE')))
+ self.assertFalse(
+ authmap.is_authorized(
+ self._make_environ('/shared/sync-from/x', 'POST')))
+ # test user-db database resource auth
+ self.assertTrue(
+ authmap.is_authorized(
+ self._make_environ('/%s' % dbname, 'GET')))
+ self.assertTrue(
+ authmap.is_authorized(
+ self._make_environ('/%s' % dbname, 'PUT')))
+ self.assertTrue(
+ authmap.is_authorized(
+ self._make_environ('/%s' % dbname, 'DELETE')))
+ self.assertFalse(
+ authmap.is_authorized(
+ self._make_environ('/%s' % dbname, 'POST')))
+ # test user-db docs resource auth
+ self.assertFalse(
+ authmap.is_authorized(
+ self._make_environ('/%s/docs' % dbname, 'GET')))
+ self.assertFalse(
+ authmap.is_authorized(
+ self._make_environ('/%s/docs' % dbname, 'PUT')))
+ self.assertFalse(
+ authmap.is_authorized(
+ self._make_environ('/%s/docs' % dbname, 'DELETE')))
+ self.assertFalse(
+ authmap.is_authorized(
+ self._make_environ('/%s/docs' % dbname, 'POST')))
+ # test user-db doc resource auth
+ self.assertFalse(
+ authmap.is_authorized(
+ self._make_environ('/%s/doc/x' % dbname, 'GET')))
+ self.assertFalse(
+ authmap.is_authorized(
+ self._make_environ('/%s/doc/x' % dbname, 'PUT')))
+ self.assertFalse(
+ authmap.is_authorized(
+ self._make_environ('/%s/doc/x' % dbname, 'DELETE')))
+ self.assertFalse(
+ authmap.is_authorized(
+ self._make_environ('/%s/doc/x' % dbname, 'POST')))
+ # test user-db sync resource auth
+ self.assertTrue(
+ authmap.is_authorized(
+ self._make_environ('/%s/sync-from/x' % dbname, 'GET')))
+ self.assertTrue(
+ authmap.is_authorized(
+ self._make_environ('/%s/sync-from/x' % dbname, 'PUT')))
+ self.assertFalse(
+ authmap.is_authorized(
+ self._make_environ('/%s/sync-from/x' % dbname, 'DELETE')))
+ self.assertTrue(
+ authmap.is_authorized(
+ self._make_environ('/%s/sync-from/x' % dbname, 'POST')))
+
+ def test_verify_action_with_wrong_dbnames(self):
+ """
+ Test if authorization fails for a wrong dbname.
+ """
+ uuid = uuid4().hex
+ authmap = URLToAuthorization(uuid)
+ dbname = 'somedb'
+ # test wrong-db database resource auth
+ self.assertFalse(
+ authmap.is_authorized(
+ self._make_environ('/%s' % dbname, 'GET')))
+ self.assertFalse(
+ authmap.is_authorized(
+ self._make_environ('/%s' % dbname, 'PUT')))
+ self.assertFalse(
+ authmap.is_authorized(
+ self._make_environ('/%s' % dbname, 'DELETE')))
+ self.assertFalse(
+ authmap.is_authorized(
+ self._make_environ('/%s' % dbname, 'POST')))
+ # test wrong-db docs resource auth
+ self.assertFalse(
+ authmap.is_authorized(
+ self._make_environ('/%s/docs' % dbname, 'GET')))
+ self.assertFalse(
+ authmap.is_authorized(
+ self._make_environ('/%s/docs' % dbname, 'PUT')))
+ self.assertFalse(
+ authmap.is_authorized(
+ self._make_environ('/%s/docs' % dbname, 'DELETE')))
+ self.assertFalse(
+ authmap.is_authorized(
+ self._make_environ('/%s/docs' % dbname, 'POST')))
+ # test wrong-db doc resource auth
+ self.assertFalse(
+ authmap.is_authorized(
+ self._make_environ('/%s/doc/x' % dbname, 'GET')))
+ self.assertFalse(
+ authmap.is_authorized(
+ self._make_environ('/%s/doc/x' % dbname, 'PUT')))
+ self.assertFalse(
+ authmap.is_authorized(
+ self._make_environ('/%s/doc/x' % dbname, 'DELETE')))
+ self.assertFalse(
+ authmap.is_authorized(
+ self._make_environ('/%s/doc/x' % dbname, 'POST')))
+ # test wrong-db sync resource auth
+ self.assertFalse(
+ authmap.is_authorized(
+ self._make_environ('/%s/sync-from/x' % dbname, 'GET')))
+ self.assertFalse(
+ authmap.is_authorized(
+ self._make_environ('/%s/sync-from/x' % dbname, 'PUT')))
+ self.assertFalse(
+ authmap.is_authorized(
+ self._make_environ('/%s/sync-from/x' % dbname, 'DELETE')))
+ self.assertFalse(
+ authmap.is_authorized(
+ self._make_environ('/%s/sync-from/x' % dbname, 'POST')))
+
+
+class EncryptedSyncTestCase(
+ CouchDBTestCase, TestCaseWithServer):
+
+ """
+ Tests for encrypted sync using Soledad server backed by a couch database.
+ """
+
+ # increase twisted.trial's timeout because large files syncing might take
+ # some time to finish.
+ timeout = 500
+
+ @staticmethod
+ def make_app_with_state(state):
+ return make_token_soledad_app(state)
+
+ make_document_for_test = make_soledad_document_for_test
+
+ sync_target = soledad_sync_target
+
+ def _soledad_instance(self, user=None, passphrase=u'123',
+ prefix='',
+ secrets_path='secrets.json',
+ local_db_path='soledad.u1db',
+ server_url='',
+ cert_file=None, auth_token=None):
+ """
+ Instantiate Soledad.
+ """
+
+ # this callback ensures we save a document which is sent to the shared
+ # db.
+ def _put_doc_side_effect(doc):
+ self._doc_put = doc
+
+ if not server_url:
+ # attempt to find the soledad server url
+ server_address = None
+ server = getattr(self, 'server', None)
+ if server:
+ server_address = getattr(self.server, 'server_address', None)
+ else:
+ host = self.port.getHost()
+ server_address = (host.host, host.port)
+ if server_address:
+ server_url = 'http://%s:%d' % (server_address)
+
+ return Soledad(
+ user,
+ passphrase,
+ secrets_path=os.path.join(self.tempdir, prefix, secrets_path),
+ local_db_path=os.path.join(
+ self.tempdir, prefix, local_db_path),
+ server_url=server_url,
+ cert_file=cert_file,
+ auth_token=auth_token,
+ shared_db=self.get_default_shared_mock(_put_doc_side_effect))
+
+ def make_app(self):
+ self.request_state = CouchServerState(self.couch_url)
+ return self.make_app_with_state(self.request_state)
+
+ def setUp(self):
+ # the order of the following initializations is crucial because of
+ # dependencies.
+ # XXX explain better
+ CouchDBTestCase.setUp(self)
+ self.tempdir = tempfile.mkdtemp(prefix="leap_tests-")
+ TestCaseWithServer.setUp(self)
+
+ def tearDown(self):
+ CouchDBTestCase.tearDown(self)
+ TestCaseWithServer.tearDown(self)
+
+ def _test_encrypted_sym_sync(self, passphrase=u'123', doc_size=2,
+ number_of_docs=1):
+ """
+ Test the complete syncing chain between two soledad dbs using a
+ Soledad server backed by a couch database.
+ """
+ self.startTwistedServer()
+ user = 'user-' + uuid4().hex
+
+ # this will store all docs ids to avoid get_all_docs
+ created_ids = []
+
+ # instantiate soledad and create a document
+ sol1 = self._soledad_instance(
+ user=user,
+ # token is verified in test_target.make_token_soledad_app
+ auth_token='auth-token',
+ passphrase=passphrase)
+
+ # instantiate another soledad using the same secret as the previous
+ # one (so we can correctly verify the mac of the synced document)
+ sol2 = self._soledad_instance(
+ user=user,
+ prefix='x',
+ auth_token='auth-token',
+ secrets_path=sol1._secrets_path,
+ passphrase=passphrase)
+
+ # ensure remote db exists before syncing
+ db = CouchDatabase.open_database(
+ urljoin(self.couch_url, 'user-' + user),
+ create=True,
+ ensure_ddocs=True)
+
+ def _db1AssertEmptyDocList(results):
+ _, doclist = results
+ self.assertEqual([], doclist)
+
+ def _db1CreateDocs(results):
+ deferreds = []
+ for i in xrange(number_of_docs):
+ content = binascii.hexlify(os.urandom(doc_size / 2))
+ d = sol1.create_doc({'data': content})
+ d.addCallback(created_ids.append)
+ deferreds.append(d)
+ return defer.DeferredList(deferreds)
+
+ def _db1AssertDocsSyncedToServer(results):
+ self.assertEqual(number_of_docs, len(created_ids))
+ for soldoc in created_ids:
+ couchdoc = db.get_doc(soldoc.doc_id)
+ self.assertTrue(couchdoc)
+ # assert document structure in couch server
+ self.assertEqual(soldoc.doc_id, couchdoc.doc_id)
+ self.assertEqual(soldoc.rev, couchdoc.rev)
+ couch_content = couchdoc.content.keys()
+ self.assertEqual(6, len(couch_content))
+ self.assertTrue(crypto.ENC_JSON_KEY in couch_content)
+ self.assertTrue(crypto.ENC_SCHEME_KEY in couch_content)
+ self.assertTrue(crypto.ENC_METHOD_KEY in couch_content)
+ self.assertTrue(crypto.ENC_IV_KEY in couch_content)
+ self.assertTrue(crypto.MAC_KEY in couch_content)
+ self.assertTrue(crypto.MAC_METHOD_KEY in couch_content)
+
+ d = sol1.get_all_docs()
+ d.addCallback(_db1AssertEmptyDocList)
+ d.addCallback(_db1CreateDocs)
+ d.addCallback(lambda _: sol1.sync())
+ d.addCallback(_db1AssertDocsSyncedToServer)
+
+ def _db2AssertEmptyDocList(results):
+ _, doclist = results
+ self.assertEqual([], doclist)
+
+ def _getAllDocsFromBothDbs(results):
+ d1 = sol1.get_all_docs()
+ d2 = sol2.get_all_docs()
+ return defer.DeferredList([d1, d2])
+
+ d.addCallback(lambda _: sol2.get_all_docs())
+ d.addCallback(_db2AssertEmptyDocList)
+ d.addCallback(lambda _: sol2.sync())
+ d.addCallback(_getAllDocsFromBothDbs)
+
+ def _assertDocSyncedFromDb1ToDb2(results):
+ r1, r2 = results
+ _, (gen1, doclist1) = r1
+ _, (gen2, doclist2) = r2
+ self.assertEqual(number_of_docs, gen1)
+ self.assertEqual(number_of_docs, gen2)
+ self.assertEqual(number_of_docs, len(doclist1))
+ self.assertEqual(number_of_docs, len(doclist2))
+ self.assertEqual(doclist1[0], doclist2[0])
+
+ d.addCallback(_assertDocSyncedFromDb1ToDb2)
+
+ def _cleanUp(results):
+ db.delete_database()
+ db.close()
+ sol1.close()
+ sol2.close()
+
+ d.addCallback(_cleanUp)
+
+ return d
+
+ def test_encrypted_sym_sync(self):
+ return self._test_encrypted_sym_sync()
+
+ def test_encrypted_sym_sync_with_unicode_passphrase(self):
+ """
+ Test the complete syncing chain between two soledad dbs using a
+ Soledad server backed by a couch database, using an unicode
+ passphrase.
+ """
+ return self._test_encrypted_sym_sync(passphrase=u'ãáàäéàëíìïóòöõúùüñç')
+
+ def test_sync_very_large_files(self):
+ """
+ Test if Soledad can sync very large files.
+ """
+ self.skipTest(
+ "Work in progress. For reference, see: "
+ "https://leap.se/code/issues/7370")
+ length = 100 * (10 ** 6) # 100 MB
+ return self._test_encrypted_sym_sync(doc_size=length, number_of_docs=1)
+
+ def test_sync_many_small_files(self):
+ """
+ Test if Soledad can sync many smallfiles.
+ """
+ return self._test_encrypted_sym_sync(doc_size=2, number_of_docs=100)
+
+
+class ConfigurationParsingTest(unittest.TestCase):
+
+ def setUp(self):
+ self.maxDiff = None
+
+ def test_use_defaults_on_failure(self):
+ config = load_configuration('this file will never exist')
+ expected = CONFIG_DEFAULTS
+ self.assertEquals(expected, config)
+
+ def test_security_values_configuration(self):
+ # given
+ config_path = resource_filename('test_soledad',
+ 'fixture_soledad.conf')
+ # when
+ config = load_configuration(config_path)
+
+ # then
+ expected = {'members': ['user1', 'user2'],
+ 'members_roles': ['role1', 'role2'],
+ 'admins': ['user3', 'user4'],
+ 'admins_roles': ['role3', 'role3']}
+ self.assertDictEqual(expected, config['database-security'])
+
+ def test_server_values_configuration(self):
+ # given
+ config_path = resource_filename('test_soledad',
+ 'fixture_soledad.conf')
+ # when
+ config = load_configuration(config_path)
+
+ # then
+ expected = {'couch_url':
+ 'http://soledad:passwd@localhost:5984',
+ 'create_cmd':
+ 'sudo -u soledad-admin /usr/bin/create-user-db',
+ 'admin_netrc':
+ '/etc/couchdb/couchdb-soledad-admin.netrc',
+ 'batching': False}
+ self.assertDictEqual(expected, config['soledad-server'])