diff --git a/tests/integration/test_mix_and_match_attack.py b/tests/integration/test_mix_and_match_attack.py index 3026894a..7d4512a0 100755 --- a/tests/integration/test_mix_and_match_attack.py +++ b/tests/integration/test_mix_and_match_attack.py @@ -1,36 +1,30 @@ +#!/usr/bin/env python + """ test_mix_and_match_attack.py - Konstantin Andrianov + Konstantin Andrianov. - March 27, 2012 + March 27, 2012. + + April 6, 2014. + Refactored to use the 'unittest' module (test conditions in code, rather + than verifying text output), use pre-generated repository files, and + discontinue use of the old repository tools. Modify the previous scenario + simulated for the mix-and-match attack. -vladimir.v.diaz See LICENSE for licensing information. - Simulate slow retrieval attack. A simple client update vs. client - update implementing TUF. - - In the mix-and-match attack, attacker is able to trick clients into using - combination of metadata that never existed together on the repository at - the same time. - - NOTE: The interposition provided by 'tuf.interposition' is used to intercept - all calls made by urllib/urillib2 to certain network locations specified in - the interposition configuration file. Look up interposition.py for more - information and illustration of a sample contents of the interposition - configuration file. Interposition was meant to make TUF integration with an - existing software updater an easy process. This allows for more flexibility - to the existing software updater. However, if you are planning to solely use - TUF there should be no need for interposition, all necessary calls will be - generated from within TUF. - - There is no difference between 'updates' and 'target' files. + Simulate a mix-and-match attack. In a mix-and-match attack, an attacker is + able to trick clients into using a combination of metadata that never existed + together on the repository at the same time. + Note: There is no difference between 'updates' and 'target' files. """ # Help with Python 3 compatability, where the print statement is a function, an @@ -41,164 +35,218 @@ from __future__ import division import os -import shutil import urllib import tempfile +import random import time +import shutil +import json +import subprocess +import unittest +import logging -import tuf -import tuf.interposition -import tuf.tests.util_test_tools as util_test_tools +import tuf.formats +import tuf.util +import tuf.log +import tuf.client.updater as updater +import tuf.repository_tool as repo_tool +import tuf.tests.unittest_toolbox as unittest_toolbox + +# The repository tool is imported and logs console messages by default. Disable +# console log messages generated by this unit test. +repo_tool.disable_console_log_messages() + +logger = logging.getLogger('tuf.test_mix_and_match_attack') -class MixAndMatchAttackAlert(Exception): - pass +class TestMixAndMatchAttack(unittest_toolbox.Modified_TestCase): -def _download(url, filename, using_tuf=False): - if using_tuf: - tuf.interposition.urllib_tuf.urlretrieve(url, filename) + @classmethod + def setUpClass(cls): + # setUpClass() is called before any of the test cases are executed. - else: - urllib.urlretrieve(url, filename) - - - -def test_mix_and_match_attack(using_tuf=False): - """ - Attack design: - There are 3 stages: - Stage 1: Consists of a usual mode of operations using tuf. Client - downloads a target file. (Initial download) - - Stage 2: The target file is legitimately modified and metadata correctly - updated. Client downloads the target file again. (Patched target download) - - Stage 3: The target file is legitimately modified and metadata correctly - updated again. However, before client gets to download the newly patched - target file the attacker replaces the release metadata, targets metadata - and the target file with the ones from stage 1 (mix-and-match attack). - Note that timestamp metadata is untouched. Further note that same would - happen if only target metadata, and target file are reverted. - """ - - ERROR_MSG = '\tMix-And-Match Attack was Successful!\n\n' - - - try: - # Setup / Stage 1 - # --------------- - root_repo, url, server_proc, keyids = util_test_tools.init_repo(using_tuf) - reg_repo = os.path.join(root_repo, 'reg_repo') - downloads = os.path.join(root_repo, 'downloads') - evil_dir = tempfile.mkdtemp(dir=root_repo) + # 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()) - # Add file to 'repo' directory: {root_repo} - filepath = util_test_tools.add_file_to_repository(reg_repo, 'A'*10) - file_basename = os.path.basename(filepath) - url_to_file = url+'reg_repo/'+file_basename - downloaded_file = os.path.join(downloads, file_basename) + # 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 this unit test assume + # the pre-generated metadata files have a specific structure, such + # as a delegated role 'targets/role1', 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('Server process started.') + logger.info('Server process id: '+str(cls.server_process.pid)) + logger.info('Serving on port: '+str(cls.SERVER_PORT)) + cls.url = 'http://localhost:'+str(cls.SERVER_PORT) + os.path.sep - # Attacker saves the initial file. - shutil.copy(filepath, evil_dir) - unpatched_file = os.path.join(evil_dir, file_basename) + # NOTE: Following error is raised if a delay is not applied: + # + time.sleep(.2) - if using_tuf: - print('TUF ...') - tuf_repo = os.path.join(root_repo, 'tuf_repo') - tuf_targets = os.path.join(tuf_repo, 'targets') - metadata_dir = os.path.join(tuf_repo, 'metadata') - release_meta_file = os.path.join(metadata_dir, 'release.txt') - targets_meta_file = os.path.join(metadata_dir, 'targets.txt') - target = os.path.join(tuf_targets, file_basename) - - # Update TUF metadata before attacker modifies anything. - util_test_tools.tuf_refresh_repo(root_repo, keyids) - # Attacker saves the original metadata and the target file. - #shutil.copy(target, evil_dir) - shutil.copy(release_meta_file, evil_dir) - shutil.copy(targets_meta_file, evil_dir) - #target_old = os.path.join(evil_dir, file_basename) - release_meta_file_old = os.path.join(evil_dir, 'release.txt') - targets_meta_file_old = os.path.join(evil_dir, 'targets.txt') - - # Modify the url. Remember that the interposition will intercept - # urls that have 'localhost:9999' hostname, which was specified in - # the json interposition configuration file. Look for 'hostname' - # in 'util_test_tools.py'. Further, the 'file_basename' is the target - # path relative to 'targets_dir'. - url_to_file = 'http://localhost:9999/'+file_basename + @classmethod + def tearDownClass(cls): + # tearDownModule() is called after all the test cases 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 of all the test cases. + shutil.rmtree(cls.temporary_directory) + + unittest_toolbox.Modified_TestCase.clear_toolbox() + + # Kill the SimpleHTTPServer process. + if cls.server_process.returncode is None: + logger.info('Server process '+str(cls.server_process.pid)+' terminated.') + cls.server_process.kill() - # Wait for some time to let program set up local http server - time.sleep(1) - # Client's initial download. - _download(url_to_file, downloaded_file, using_tuf) - # Stage 2 - # ------- - # Developer patches the file and updates the repository. - util_test_tools.modify_file_at_repository(filepath, 'B'*11) + 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(), os.pardir, + '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_client = os.path.join(original_repository_files, 'client') + original_keystore = os.path.join(original_repository_files, 'keystore') + + # 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.client_directory = os.path.join(temporary_repository_root, 'client') + self.keystore_directory = os.path.join(temporary_repository_root, 'keystore') + + # 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) + + # Set the url prefix required by the 'tuf/client/updater.py' updater. + # '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': ['']}} - # Updating tuf repository. This will copy files from regular repository - # into tuf repository and refresh the metadata - if using_tuf: - util_test_tools.tuf_refresh_repo(root_repo, keyids) + # Create the 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) - # Client downloads the patched file. - _download(url_to_file, downloaded_file, using_tuf) - downloaded_content = util_test_tools.read_file_content(downloaded_file) + def tearDown(self): + # Modified_TestCase.tearDown() automatically deletes temporary files and + # directories that may have been created during each test case. + unittest_toolbox.Modified_TestCase.tearDown(self) - # Stage 3 - # ------- - # Developer patches the file and updates the repository again. - util_test_tools.modify_file_at_repository(filepath, 'C'*10) - # Updating tuf repository. This will copy files from regular repository - # into tuf repository and refresh the metadata - if using_tuf: - util_test_tools.tuf_refresh_repo(root_repo, keyids) - # Attacker replaces the metadata and the target file. - shutil.copyfile(unpatched_file, target) - shutil.copyfile(release_meta_file_old, release_meta_file) - shutil.copyfile(targets_meta_file_old, targets_meta_file) + def test_with_tuf(self): + # Scenario: + # An attacker tries to trick the client into installing files indicated by + # a previous release of its corresponding metatadata. The outdated metadata + # is properly named and was previously valid, but is no longer current + # according to the latest 'snapshot.json' role. Generate a new snapshot of + # the repository after modifying a target file of 'role1.json'. + # Backup 'role1.json' (the delegated role to be updated, and then inserted + # again for the mix-and-match attack.) + role1_path = os.path.join(self.repository_directory, 'metadata', 'targets', + 'role1.json') + backup_role1 = os.path.join(self.repository_directory, 'role1.json.backup') + shutil.copy(role1_path, backup_role1) - # Attacker replaces the patched file with the unpatched one. - shutil.copyfile(unpatched_file, filepath) + # Backup 'file3.txt', specified by 'role1.json'. + file3_path = os.path.join(self.repository_directory, 'targets', 'file3.txt') + shutil.copy(file3_path, file3_path + '.backup') + + # Re-generate the required metadata on the remote repository. The affected + # metadata must be properly updated and signed with 'repository_tool.py', + # otherwise the client will reject them as invalid metadata. The resulting + # metadata should be valid metadata. + repository = repo_tool.load_repository(self.repository_directory) + + # Load the signing keys so that newly generated metadata is properly signed. + timestamp_keyfile = os.path.join(self.keystore_directory, 'timestamp_key') + role1_keyfile = os.path.join(self.keystore_directory, 'delegation_key') + snapshot_keyfile = os.path.join(self.keystore_directory, 'snapshot_key') + timestamp_private = \ + repo_tool.import_rsa_privatekey_from_file(timestamp_keyfile, 'password') + role1_private = \ + repo_tool.import_rsa_privatekey_from_file(role1_keyfile, 'password') + snapshot_private = \ + repo_tool.import_rsa_privatekey_from_file(snapshot_keyfile, 'password') + + repository.targets('role1').load_signing_key(role1_private) + repository.snapshot.load_signing_key(snapshot_private) + repository.timestamp.load_signing_key(timestamp_private) + + # Modify a 'role1.json' target file, and add it to its metadata so that a + # new version is generated. + with open(file3_path, 'wb') as file_object: + file_object.write('update file3') + repository.targets('role1').add_target(file3_path) + + 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')) + + # Insert the previously valid 'role1.json'. The TUF client should reject it. + shutil.move(backup_role1, role1_path) + + # Verify that the TUF client detects unexpected metadata (previously valid, + # but not up-to-date with the latest snashot of the repository) and refuses + # to continue the update process. + # Refresh top-level metadata so that the client is aware of the latest + # snapshot of the repository. + self.repository_updater.refresh() - # Client tries to downloads the newly patched file. try: - _download(url_to_file, downloaded_file, using_tuf) - except tuf.NoWorkingMirrorError as errors: - for mirror_url, mirror_error in errors.mirror_errors.iteritems(): - if type(mirror_error) == tuf.BadHashError: - print('Caught a Bad Hash Error!') + self.repository_updater.targets_of_role('targets/role1') + + # Verify that the specific 'tuf.BadHashError' exception is raised by each + # mirror. + except tuf.NoWorkingMirrorError, exception: + for mirror_url, mirror_error in exception.mirror_errors.iteritems(): + url_prefix = self.repository_mirrors['mirror1']['url_prefix'] + url_file = os.path.join(url_prefix, 'metadata', 'targets', 'role1.json') + + # Verify that 'timestamp.json' is the culprit. + self.assertEqual(url_file, mirror_url) + self.assertTrue(isinstance(mirror_error, tuf.BadHashError)) - # Check whether the attack succeeded by inspecting the content of the - # update. The update should contain 'Test NOT A'. - downloaded_content = util_test_tools.read_file_content(downloaded_file) - if ('B'*11) != downloaded_content: - raise MixAndMatchAttackAlert(ERROR_MSG) + else: + self.fail('TUF did not prevent a mix-and-match attack.') - finally: - util_test_tools.cleanup(root_repo, server_proc) - - - - - -try: - test_mix_and_match_attack(using_tuf=False) -except MixAndMatchAttackAlert, error: - print(error) - - -try: - test_mix_and_match_attack(using_tuf=True) -except MixAndMatchAttackAlert, error: - print(error) +if __name__ == '__main__': + unittest.main()