Add auditor implementation

The auditor verifies all nodes in the snapshot
merkle tree to check for rollback attacks.

This PoC implementation re-uses a lot of functionality
from the updater. It may be better to move some of the
re-used functionality to a separate place (ie separating
file downloading from update logic).

The auditor implementation currently requires there to be
snapshot metadata on the repository in order to iterate over
all of the nodes.

In the future, if the verification succeeds, the auditor
should add a signature to timestamp metadata.

Signed-off-by: Marina Moore <[email protected]>

Author: mnm678

tests/test_auditor.py

@@ -0,0 +1,317 @@
+#!/usr/bin/env python
+
+"""
+<Program Name>
+  test_auditor.py
+
+<Author>
+  Marina Moore
+
+<Started>
+  January 29, 2021
+
+<Copyright>
+  See LICENSE-MIT OR LICENSE for licensing information.
+
+<Purpose>
+  'test-auditor.py' provides a collection of methods that test the public /
+  non-public methods and functions of 'tuf.client.auditor.py'.
+
+"""
+
+import unittest
+import tempfile
+import os
+import logging
+import shutil
+
+import tuf
+import tuf.exceptions
+import tuf.log
+import tuf.keydb
+import tuf.roledb
+import tuf.repository_tool as repo_tool
+import tuf.repository_lib as repo_lib
+import tuf.unittest_toolbox as unittest_toolbox
+import tuf.client.auditor as auditor
+
+from tests import utils
+
+import securesystemslib
+
+logger = logging.getLogger(__name__)
+repo_tool.disable_console_log_messages()
+
+
+class TestAuditor(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())
+
+    # Needed because in some tests simple_server.py cannot be found.
+    # The reason is that the current working directory
+    # has been changed when executing a subprocess.
+    cls.SIMPLE_SERVER_PATH = os.path.join(os.getcwd(), 'simple_server.py')
+
+    # 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_updater.py'
+    # 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_process_handler = utils.TestServerProcess(log=logger,
+        server=cls.SIMPLE_SERVER_PATH)
+
+
+  @classmethod
+  def tearDownClass(cls):
+    # Cleans the resources and flush the logged lines (if any).
+    cls.server_process_handler.clean()
+
+    # 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)
+
+
+  def setUp(self):
+    # We are inheriting from custom class.
+    unittest_toolbox.Modified_TestCase.setUp(self)
+
+    tuf.roledb.clear_roledb(clear_all=True)
+    tuf.keydb.clear_keydb(clear_all=True)
+
+    self.repository_name = 'test_repository1'
+
+    # 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,
+        self.repository_name, '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_process_handler.port) + repository_basepath
+
+    # Setting 'tuf.settings.repository_directory' with the temporary client
+    # directory copied from the original repository files.
+    tuf.settings.repositories_directory = self.client_directory
+
+    # replace timestamp with a merkle timestamp
+    merkle_timestamp = os.path.join(self.repository_directory, 'metadata', 'timestamp-merkle.json')
+    timestamp = os.path.join(self.repository_directory, 'metadata', 'timestamp.json')
+    shutil.move(merkle_timestamp, timestamp)
+
+    # 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)
+
+   # The repository must be rewritten with 'consistent_snapshot' set.
+    repository = repo_tool.load_repository(self.repository_directory)
+
+    # Write metadata for all the top-level roles , since consistent snapshot
+    # is now being set to true (i.e., the pre-generated repository isn't set
+    # to support consistent snapshots.  A new version of targets.json is needed
+    # to ensure <digest>.filename target files are written to disk.
+    repository.targets.load_signing_key(self.role_keys['targets']['private'])
+    repository.root.load_signing_key(self.role_keys['root']['private'])
+    repository.snapshot.load_signing_key(self.role_keys['snapshot']['private'])
+    repository.timestamp.load_signing_key(self.role_keys['timestamp']['private'])
+
+    repository.mark_dirty(['targets', 'root', 'snapshot', 'timestamp'])
+    repository.writeall(snapshot_merkle=True, consistent_snapshot=True)
+
+    # 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'))
+
+    self.repository_mirrors = {'mirror1': {'url_prefix': url_prefix,
+                                           'metadata_path': 'metadata',
+                                           'targets_path': 'targets'}}
+
+
+
+
+  def tearDown(self):
+    # We are inheriting from custom class.
+    unittest_toolbox.Modified_TestCase.tearDown(self)
+    tuf.roledb.clear_roledb(clear_all=True)
+    tuf.keydb.clear_keydb(clear_all=True)
+
+    # Logs stdout and stderr from the sever subprocess.
+    self.server_process_handler.flush_log()
+
+
+  # UNIT TESTS.
+
+  def test_1__init_exceptions(self):
+    # Invalid arguments
+    self.assertRaises(securesystemslib.exceptions.FormatError, auditor.Auditor,
+        5, self.repository_mirrors)
+    self.assertRaises(securesystemslib.exceptions.FormatError, auditor.Auditor,
+        self.repository_name, 5)
+
+
+
+  def test_2__verify_merkle_tree(self):
+    repository_auditor = auditor.Auditor(self.repository_name, self.repository_mirrors)
+    # skip version 1 as it was written without consistent snapshots
+    repository_auditor.last_version_verified = 1
+
+    # The repository must be rewritten with 'consistent_snapshot' set.
+    repository = repo_tool.load_repository(self.repository_directory)
+
+    # Write metadata for all the top-level roles , since consistent snapshot
+    # is now being set to true (i.e., the pre-generated repository isn't set
+    # to support consistent snapshots.  A new version of targets.json is needed
+    # to ensure <digest>.filename target files are written to disk.
+    repository.targets.load_signing_key(self.role_keys['targets']['private'])
+    repository.root.load_signing_key(self.role_keys['root']['private'])
+    repository.snapshot.load_signing_key(self.role_keys['snapshot']['private'])
+    repository.timestamp.load_signing_key(self.role_keys['timestamp']['private'])
+
+    repository.targets.add_target('file1.txt')
+
+    repository.mark_dirty(['targets', 'root', 'snapshot', 'timestamp'])
+    repository.writeall(snapshot_merkle=True, consistent_snapshot=True)
+
+    # 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'))
+
+
+    # Normal case, should not error
+    repository_auditor.verify()
+
+    self.assertEqual(repository_auditor.version_info['role1.json'], 1)
+    self.assertEqual(repository_auditor.version_info['targets.json'], 3)
+    self.assertEqual(repository_auditor.last_version_verified, 3)
+
+    # modify targets
+    repository.targets.add_target('file2.txt')
+
+    repository.targets.load_signing_key(self.role_keys['targets']['private'])
+    repository.root.load_signing_key(self.role_keys['root']['private'])
+    repository.snapshot.load_signing_key(self.role_keys['snapshot']['private'])
+    repository.timestamp.load_signing_key(self.role_keys['timestamp']['private'])
+
+
+    repository.mark_dirty(['targets', 'root', 'snapshot', 'timestamp'])
+    repository.writeall(snapshot_merkle=True, consistent_snapshot=True)
+
+    # 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'))
+
+    repository_auditor.verify()
+
+    # Ensure the auditor checked the latest targets
+    self.assertEqual(repository_auditor.version_info['targets.json'], 4)
+
+    # Test rollback attack detection
+    repository_auditor.version_info['targets.json'] = 5
+    repository_auditor.last_version_verified = 3
+
+    self.assertRaises(tuf.exceptions.RepositoryError, repository_auditor.verify)
+
+
+
+
+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_ed25519_publickey_from_file(targets_key_file+'.pub')
+  role_keys['snapshot']['public'] = \
+    repo_tool.import_ed25519_publickey_from_file(snapshot_key_file+'.pub')
+  role_keys['timestamp']['public'] = \
+      repo_tool.import_ed25519_publickey_from_file(timestamp_key_file+'.pub')
+  role_keys['role1']['public'] = \
+      repo_tool.import_ed25519_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_ed25519_privatekey_from_file(targets_key_file,
+                                              EXPECTED_KEYFILE_PASSWORD)
+  role_keys['snapshot']['private'] = \
+    repo_tool.import_ed25519_privatekey_from_file(snapshot_key_file,
+                                              EXPECTED_KEYFILE_PASSWORD)
+  role_keys['timestamp']['private'] = \
+    repo_tool.import_ed25519_privatekey_from_file(timestamp_key_file,
+                                              EXPECTED_KEYFILE_PASSWORD)
+  role_keys['role1']['private'] = \
+    repo_tool.import_ed25519_privatekey_from_file(delegation_key_file,
+                                              EXPECTED_KEYFILE_PASSWORD)
+
+  return role_keys
+
+
+
+if __name__ == '__main__':
+  unittest.main()

tests/test_updater.py

@@ -1791,15 +1791,15 @@ def test_snapshot_merkle(self):
     repository_updater.refresh()
 
     # Test verify merkle path
-    snapshot_info = repository_updater._verify_merkle_path('targets')
+    snapshot_info = repository_updater.verify_merkle_path('targets')
     self.assertEqual(snapshot_info['version'], 1)
 
-    snapshot_info = repository_updater._verify_merkle_path('role1')
+    snapshot_info = repository_updater.verify_merkle_path('role1')
     self.assertEqual(snapshot_info['version'], 1)
 
     # verify merkle path with invalid role
     self.assertRaises(tuf.exceptions.NoWorkingMirrorError,
-        repository_updater._verify_merkle_path, 'foo')
+        repository_updater.verify_merkle_path, 'foo')
 
     # Test get_one_valid_targetinfo with snapshot merkle
     repository_updater.get_one_valid_targetinfo('file1.txt')

tuf/client/auditor.py

@@ -0,0 +1,137 @@
+#!/usr/bin/env python
+
+# Copyright 2012 - 2017, New York University and the TUF contributors
+# SPDX-License-Identifier: MIT OR Apache-2.0
+
+"""
+<Program Name>
+  auditor.py
+
+<Author>
+  Marina Moore <mnm678@gmail.com>
+<Started>
+  January 28, 2021
+<Copyright>
+  See LICENSE-MIT OR LICENSE for licensing information.
+<Purpose>
+  'auditor.py' provides an implementation of an auditor for
+  snapshot merkle metadata.
+
+"""
+
+import tuf
+import tuf.download
+import tuf.formats
+import tuf.client.updater
+
+import securesystemslib.hash
+
+
+
+class Auditor(object):
+  """
+  <Purpose>
+    Provide a class that downloads and verifies snapshot merkle metadata
+    from a repository.
+
+  <Arguments>
+    repository_name:
+      Name of the repository to be audited
+
+    repository_mirrors:
+      Dictionary holding repository mirror information, conformant to
+      `tuf.formats.MIRRORDICT_SCHEMA`.
+
+  <Exceptions>
+    securesystemslib.exceptions.FormatError:
+      If the arguments are improperly formatted.
+
+  <Side Effects>
+    None.
+
+  <Returns>
+    None.
+  """
+
+  def __init__(self, repository_name, repository_mirrors):
+    securesystemslib.formats.NAME_SCHEMA.check_match(repository_name)
+    tuf.formats.MIRRORDICT_SCHEMA.check_match(repository_mirrors)
+
+    self.repository_name = repository_name
+    self.mirrors = repository_mirrors
+
+    # Create a dictionary to store current version information
+    # for all targets metadata
+    self.version_info = {}
+
+    # Keep track of the last timestamp version number checked
+    self.last_version_verified = 0
+
+    # Updater will be used to update top-level metadata
+    self.updater = tuf.client.updater.Updater(repository_name, repository_mirrors)
+
+
+  def verify(self):
+    # download most recent top-level metadata, determine current timestamp key
+    self.updater.refresh()
+
+    cur_timestamp_keys = self.updater.metadata['current']['root']['roles']['timestamp']['keyids']
+
+    # Download all trees since last_version_verified that use cur_timestamp_key
+
+    next_version = self.last_version_verified + 1
+    version_exists = True
+
+    while(version_exists):
+      verification_fn = self.updater.signable_verification
+
+      # Attempt to download this version of timestamp. If it does not exist,
+      # break out of the loop
+      timestamp = self.updater.download_metadata_version_if_exists("timestamp",
+          next_version, verification_fn,
+          tuf.settings.DEFAULT_TIMESTAMP_REQUIRED_LENGTH)
+
+      if not timestamp:
+        version_exists = False
+        break
+
+
+      # Compare with the current timestamp keys. We only verify any trees
+      # that use the current keys for fast forward attack recovery
+      # TODO support more than one timestamp key
+      if timestamp['signatures'][0]['keyid'] != cur_timestamp_keys[0]:
+        # TODO: Should the auditor also verify older trees?
+        break
+
+      merkle_root = timestamp['signed']['merkle_root']
+
+      # Download and verify Merkle trees
+
+      # First, download snapshot to get a list of nodes
+      snapshot = self.updater.download_metadata_version_if_exists("snapshot",
+          next_version, verification_fn,
+          tuf.settings.DEFAULT_SNAPSHOT_REQUIRED_LENGTH)
+
+      for metadata_filename in snapshot['signed']['meta']:
+        # Download the node and verify its path
+        versioninfo = self.updater.verify_merkle_path(
+            metadata_filename[:-len('.json')], next_version, merkle_root)
+
+        # Have we seen this metadata file before?
+        # If yes, compare the version info
+        if metadata_filename in self.version_info:
+          if self.version_info[metadata_filename] > versioninfo['version']:
+            raise tuf.exceptions.RepositoryError('Rollback attack detected' +
+                'for ' + metadata_filename + '. Version ' +
+                str(versioninfo['version']) + ' is less than ' +
+                str(self.version_info[metadata_filename]))
+
+        # Update `version_info` with the latest seen version
+        self.version_info[metadata_filename] = versioninfo['version']
+
+
+      self.last_version_verified = next_version
+      next_version = next_version + 1
+
+
+

tuf/client/updater.py

@@ -1492,7 +1492,7 @@ def _get_metadata_file(self, metadata_role, remote_filename,
       snapshot_merkle:
         Is the metadata file a snapshot merkle file? Snapshot merkle files
         are not signed and so should skip some of the verification steps here.
-        Instead, they must be verified using _verify_merkle_path.
+        Instead, they must be verified using verify_merkle_path.
 
     <Exceptions>
       tuf.exceptions.NoWorkingMirrorError:
@@ -1654,7 +1654,7 @@ def _update_merkle_metadata(self, merkle_filename, upperbound_filelength,
 
       version:
         The expected and required version number of the 'merkle_filename' file
-        downloaded.  'expected_version' is an integer.
+        downloaded.  'version' is an integer.
 
     <Exceptions>
       tuf.exceptions.NoWorkingMirrorError:
@@ -1865,7 +1865,7 @@ def _update_metadata(self, metadata_role, upperbound_filelength, version=None):
 
 
 
-  def _verify_merkle_path(self, metadata_role):
+  def verify_merkle_path(self, metadata_role, version=None, merkle_root=None):
     """
     <Purpose>
       Download the merkle path associated with metadata_role and verify the hashes.
@@ -1879,13 +1879,14 @@ def _verify_merkle_path(self, metadata_role):
       A dictionary containing the snapshot information about metadata role,
       conforming to VERSIONINFO_SCHEMA or METADATA_FILEINFO_SCHEMA
     """
-    merkle_root = self.metadata['current']['timestamp']['merkle_root']
+    if not merkle_root:
+      merkle_root = self.metadata['current']['timestamp']['merkle_root']
 
     metadata_rolename = metadata_role + '-snapshot'
 
     # Download Merkle path
     upperbound_filelength = tuf.settings.MERKLE_FILELENGTH
-    self._update_merkle_metadata(metadata_rolename, upperbound_filelength)
+    self._update_merkle_metadata(metadata_rolename, upperbound_filelength, version)
     metadata_directory = self.metadata_directory['current']
     metadata_filename = metadata_rolename + '.json'
     metadata_filepath = os.path.join(metadata_directory, metadata_filename)
@@ -2040,7 +2041,7 @@ def _update_metadata_if_changed(self, metadata_role,
 
     if 'merkle_root' in self.metadata['current']['timestamp']:
       # Download version information from merkle tree
-      contents = self._verify_merkle_path(metadata_role)
+      contents = self.verify_merkle_path(metadata_role)
       expected_versioninfo = contents
 
     else:
@@ -3419,3 +3420,42 @@ def download_target(self, target, destination_directory,
         trusted_hashes, prefix_filename_with_hash)
 
     securesystemslib.util.persist_temp_file(target_file_object, destination)
+
+  def download_metadata_version_if_exists(self, role_name, version, verification_fn, upperbound_filelength):
+
+    filename = role_name + ".json"
+    dirname, basename = os.path.split(filename)
+    remote_filename = os.path.join(dirname, str(version) + '.' + basename)
+
+
+    def neither_403_nor_404(mirror_error):
+      if isinstance(mirror_error, requests.exceptions.HTTPError):
+        if mirror_error.response.status_code in {403, 404}:
+          return False
+      return True
+
+    updated_metadata_object = None
+
+    try:
+      # Thoroughly verify it.
+      metadata_file_object = \
+        self._get_metadata_file(role_name, remote_filename,
+          upperbound_filelength, version, verification_fn)
+      metadata_file_object.seek(0)
+      updated_metadata_object = \
+        securesystemslib.util.load_json_string(metadata_file_object.read().decode('utf-8'))
+    # When we run into HTTP 403/404 error from ALL mirrors,
+    # metadata file is most likely missing.
+    except tuf.exceptions.NoWorkingMirrorError as exception:
+      for mirror_error in exception.mirror_errors.values():
+        # Otherwise, reraise the error, because it is not a simple HTTP
+        # error.
+        if neither_403_nor_404(mirror_error):
+          logger.exception('Misc error for root version '+str(version))
+          raise
+        else:
+          # Calling this function should give us a detailed stack trace
+          # including an HTTP error code, if any.
+          logger.exception('HTTP error for root version '+str(version))
+
+    return updated_metadata_object