From 09163247b6da660dc543d0bc1bc98447400e7f10 Mon Sep 17 00:00:00 2001 From: Vladimir Diaz Date: Thu, 28 Apr 2016 09:02:58 -0400 Subject: [PATCH] Initial integration test for key revocations --- tests/test_key_revocation.py | 285 +++++++++++++++++++++++++++++++++++ 1 file changed, 285 insertions(+) create mode 100755 tests/test_key_revocation.py diff --git a/tests/test_key_revocation.py b/tests/test_key_revocation.py new file mode 100755 index 00000000..b120137f --- /dev/null +++ b/tests/test_key_revocation.py @@ -0,0 +1,285 @@ +#!/usr/bin/env python + +""" + + test_key_revocation.py + + + Vladimir Diaz. + + + April 28, 2016. + + + See LICENSE for licensing information. + + + Integration test that verifies how a top-level role like Root is updated + after all of its keys have been revoked. There are unit tests in + 'test_repository_tool.py' that verify key and role revocation of specific + roles, but these should be expanded to verify key revocations over the span + of multiple snapshots of the repository. + + The 'unittest_toolbox.py' module was created to provide additional testing + tools, such as automatically deleting temporary files created in test cases. + For more information, see 'tests/unittest_toolbox.py'. +""" + +# Help with Python 3 compatibility, where the print statement is a function, an +# implicit relative import is invalid, and the '/' operator performs true +# division. Example: print 'hello world' raises a 'SyntaxError' exception. +from __future__ import print_function +from __future__ import absolute_import +from __future__ import division +from __future__ import unicode_literals + +import os +import time +import shutil +import copy +import tempfile +import logging +import random +import subprocess +import sys + +# 'unittest2' required for testing under Python < 2.7. +if sys.version_info >= (2, 7): + import unittest + +else: + import unittest2 as unittest + +import tuf +import tuf.log +import tuf.conf +import tuf.roledb +import tuf.repository_tool as repo_tool +import tuf.unittest_toolbox as unittest_toolbox +import tuf.client.updater as updater +import six + +logger = logging.getLogger('tuf.test_key_revocation') +repo_tool.disable_console_log_messages() + + +class TestUpdater(unittest_toolbox.Modified_TestCase): + + @classmethod + def setUpClass(cls): + # setUpClass() is called before tests in an individual class are executed. + + # Create a temporary directory to store the repository, metadata, and target + # files. 'temporary_directory' must be deleted in TearDownModule() so that + # temporary files are always removed, even when exceptions occur. + cls.temporary_directory = tempfile.mkdtemp(dir=os.getcwd()) + + # Launch a SimpleHTTPServer (serves files in the current directory). Test + # cases will request metadata and target files that have been pre-generated + # in 'tuf/tests/repository_data', which will be served by the + # SimpleHTTPServer launched here. The test cases of + # 'test_key_revocation.py' assume the pre-generated metadata files have a + # specific structure, such as a delegated role, three target files, five + # key files, etc. + cls.SERVER_PORT = random.randint(30000, 45000) + command = ['python', 'simple_server.py', str(cls.SERVER_PORT)] + cls.server_process = subprocess.Popen(command, stderr=subprocess.PIPE) + logger.info('\n\tServer process started.') + logger.info('\tServer process id: '+str(cls.server_process.pid)) + logger.info('\tServing on port: '+str(cls.SERVER_PORT)) + cls.url = 'http://localhost:'+str(cls.SERVER_PORT) + os.path.sep + + # NOTE: Following error is raised if a delay is not applied: + # + time.sleep(1) + + + + @classmethod + def tearDownClass(cls): + # tearDownModule() is called after all the tests have run. + # http://docs.python.org/2/library/unittest.html#class-and-module-fixtures + + # Remove the temporary repository directory, which should contain all the + # metadata, targets, and key files generated for the test cases. + shutil.rmtree(cls.temporary_directory) + + # Kill the SimpleHTTPServer process. + if cls.server_process.returncode is None: + logger.info('\tServer process '+str(cls.server_process.pid)+' terminated.') + cls.server_process.kill() + + + + def setUp(self): + # We are inheriting from custom class. + unittest_toolbox.Modified_TestCase.setUp(self) + + # Copy the original repository files provided in the test folder so that + # any modifications made to repository files are restricted to the copies. + # The 'repository_data' directory is expected to exist in 'tuf.tests/'. + original_repository_files = os.path.join(os.getcwd(), 'repository_data') + temporary_repository_root = \ + self.make_temp_directory(directory=self.temporary_directory) + + # The original repository, keystore, and client directories will be copied + # for each test case. + original_repository = os.path.join(original_repository_files, 'repository') + original_keystore = os.path.join(original_repository_files, 'keystore') + original_client = os.path.join(original_repository_files, 'client') + + # Save references to the often-needed client repository directories. + # Test cases need these references to access metadata and target files. + self.repository_directory = \ + os.path.join(temporary_repository_root, 'repository') + self.keystore_directory = \ + os.path.join(temporary_repository_root, 'keystore') + self.client_directory = os.path.join(temporary_repository_root, 'client') + self.client_metadata = os.path.join(self.client_directory, 'metadata') + self.client_metadata_current = os.path.join(self.client_metadata, 'current') + self.client_metadata_previous = \ + os.path.join(self.client_metadata, 'previous') + + # Copy the original 'repository', 'client', and 'keystore' directories + # to the temporary repository the test cases can use. + shutil.copytree(original_repository, self.repository_directory) + shutil.copytree(original_client, self.client_directory) + shutil.copytree(original_keystore, self.keystore_directory) + + # 'path/to/tmp/repository' -> 'localhost:8001/tmp/repository'. + repository_basepath = self.repository_directory[len(os.getcwd()):] + url_prefix = \ + 'http://localhost:' + str(self.SERVER_PORT) + repository_basepath + + # Setting 'tuf.conf.repository_directory' with the temporary client + # directory copied from the original repository files. + tuf.conf.repository_directory = self.client_directory + + self.repository_mirrors = {'mirror1': {'url_prefix': url_prefix, + 'metadata_path': 'metadata', + 'targets_path': 'targets', + 'confined_target_dirs': ['']}} + + # Creating repository instance. The test cases will use this client + # updater to refresh metadata, fetch target files, etc. + self.repository_updater = updater.Updater('test_repository', + self.repository_mirrors) + + # Metadata role keys are needed by the test cases to make changes to the + # repository (e.g., adding a new target file to 'targets.json' and then + # requesting a refresh()). + self.role_keys = _load_role_keys(self.keystore_directory) + + + + def tearDown(self): + # We are inheriting from custom class. + unittest_toolbox.Modified_TestCase.tearDown(self) + tuf.roledb.clear_roledb() + + + + + + # UNIT TESTS. + def test_timestamp_key_revocation(self): + # First verify that the Timestamp role is properly signed. + self.repository_updater.refresh() + + + # Verify that the client is able to recognize that a new set of keys have + # added to the Timestamp role. + + + repository = repo_tool.load_repository(self.repository_directory) + + repository.timestamp.load_signing_key(self.role_keys['timestamp']['private']) + repository.write() + + # Move the staged metadata to the "live" metadata. + shutil.rmtree(os.path.join(self.repository_directory, 'metadata')) + shutil.copytree(os.path.join(self.repository_directory, 'metadata.staged'), + os.path.join(self.repository_directory, 'metadata')) + + + + + + def test_snapshot_key_revocation(self): + pass + + + + + + def test_targets_key_revocation(self): + pass + + + + + + def test_root_key_revocation(self): + pass + + + +def _load_role_keys(keystore_directory): + + # Populating 'self.role_keys' by importing the required public and private + # keys of 'tuf/tests/repository_data/'. The role keys are needed when + # modifying the remote repository used by the test cases in this unit test. + + # The pre-generated key files in 'repository_data/keystore' are all encrypted with + # a 'password' passphrase. + EXPECTED_KEYFILE_PASSWORD = 'password' + + # Store and return the cryptography keys of the top-level roles, including 1 + # delegated role. + role_keys = {} + + root_key_file = os.path.join(keystore_directory, 'root_key') + targets_key_file = os.path.join(keystore_directory, 'targets_key') + snapshot_key_file = os.path.join(keystore_directory, 'snapshot_key') + timestamp_key_file = os.path.join(keystore_directory, 'timestamp_key') + delegation_key_file = os.path.join(keystore_directory, 'delegation_key') + + role_keys = {'root': {}, 'targets': {}, 'snapshot': {}, 'timestamp': {}, + 'role1': {}} + + # Import the top-level and delegated role public keys. + role_keys['root']['public'] = \ + repo_tool.import_rsa_publickey_from_file(root_key_file+'.pub') + role_keys['targets']['public'] = \ + repo_tool.import_rsa_publickey_from_file(targets_key_file+'.pub') + role_keys['snapshot']['public'] = \ + repo_tool.import_rsa_publickey_from_file(snapshot_key_file+'.pub') + role_keys['timestamp']['public'] = \ + repo_tool.import_rsa_publickey_from_file(timestamp_key_file+'.pub') + role_keys['role1']['public'] = \ + repo_tool.import_rsa_publickey_from_file(delegation_key_file+'.pub') + + # Import the private keys of the top-level and delegated roles. + role_keys['root']['private'] = \ + repo_tool.import_rsa_privatekey_from_file(root_key_file, + EXPECTED_KEYFILE_PASSWORD) + role_keys['targets']['private'] = \ + repo_tool.import_rsa_privatekey_from_file(targets_key_file, + EXPECTED_KEYFILE_PASSWORD) + role_keys['snapshot']['private'] = \ + repo_tool.import_rsa_privatekey_from_file(snapshot_key_file, + EXPECTED_KEYFILE_PASSWORD) + role_keys['timestamp']['private'] = \ + repo_tool.import_rsa_privatekey_from_file(timestamp_key_file, + EXPECTED_KEYFILE_PASSWORD) + role_keys['role1']['private'] = \ + repo_tool.import_rsa_privatekey_from_file(delegation_key_file, + EXPECTED_KEYFILE_PASSWORD) + + return role_keys + + + + +if __name__ == '__main__': + unittest.main()