Refactor test_mix_and_match_attack.py.

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.
This commit is contained in:
vladdd 2014-04-06 16:17:14 -04:00
parent 31464663d9
commit 75a7124bb1

View file

@ -1,36 +1,30 @@
#!/usr/bin/env python
"""
<Program Name>
test_mix_and_match_attack.py
<Author>
Konstantin Andrianov
Konstantin Andrianov.
<Started>
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
<Copyright>
See LICENSE for licensing information.
<Purpose>
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:
# <urlopen error [Errno 111] Connection refused>
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()