Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Snapshot RSA Accumulator #1510

Closed
wants to merge 4 commits into from
Closed

Snapshot RSA Accumulator #1510

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
e56cd12
Add initial poc for RSA Accumulator snapshots: repo side
mnm678 Jul 27, 2021
90d7aa9
Initial poc for RSA Accumulator snapshot: client side
mnm678 Jul 27, 2021
7ac4a3a
Updates to server side RSA Accumulator for rebase
mnm678 Aug 24, 2021
b688cbf
Update tests
mnm678 Sep 28, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
66 changes: 62 additions & 4 deletions tests/test_repository_lib.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -458,6 +458,64 @@ def test_generate_targets_metadata(self):
False, use_existing_fileinfo=True)


def test_build_rsa_acc(self):
temporary_directory = tempfile.mkdtemp(dir=self.temporary_directory)
storage_backend = securesystemslib.storage.FilesystemBackend()
version = 1

# Test an rsa accumulator with a few nodes to verify the output

test_nodes = {}
test_nodes['file1'] = tuf.formats.make_metadata_fileinfo(5, None, None)


root_1, leaves = repo_lib._build_rsa_acc(test_nodes)
repo_lib._write_rsa_proofs(root_1, leaves, storage_backend,
temporary_directory, version)

# Ensure that the paths are written to the directory
file_path = os.path.join(temporary_directory, 'file1-snapshot.json')
self.assertTrue(os.path.exists(file_path))

file_path = os.path.join(temporary_directory, '1.file1-snapshot.json')
self.assertTrue(os.path.exists(file_path))

test_nodes = {}
test_nodes['targets'] = tuf.formats.make_metadata_fileinfo(1, None, None)
test_nodes['role1'] = tuf.formats.make_metadata_fileinfo(1, None, None)
test_nodes['role2'] = tuf.formats.make_metadata_fileinfo(1, None, None)

root, leaves = repo_lib._build_rsa_acc(test_nodes)
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Not sure what's going on here -- above, there's a whole build/write/check files/check for specific value cycle, whereas here you just seem to be verifying that it doesn't crash?

repo_lib._write_rsa_proofs(root, leaves, storage_backend,
temporary_directory, version)

# Ensure that the paths are written to the directory
file_path = os.path.join(temporary_directory, 'targets-snapshot.json')
self.assertTrue(os.path.exists(file_path))

file_path = os.path.join(temporary_directory, '2.targets-snapshot.json')
self.assertTrue(os.path.exists(file_path))

file_path = os.path.join(temporary_directory, 'role1-snapshot.json')
self.assertTrue(os.path.exists(file_path))

file_path = os.path.join(temporary_directory, '1.role1-snapshot.json')
self.assertTrue(os.path.exists(file_path))

file_path = os.path.join(temporary_directory, 'role2-snapshot.json')
self.assertTrue(os.path.exists(file_path))

file_path = os.path.join(temporary_directory, '1.role2-snapshot.json')
self.assertTrue(os.path.exists(file_path))

# TODO: check against the correct root value
self.assertEqual(root_1, 5)
self.assertEqual(root, 5)





def _setup_generate_snapshot_metadata_test(self):
# Test normal case.
temporary_directory = tempfile.mkdtemp(dir=self.temporary_directory)
Expand Down Expand Up @@ -499,7 +557,7 @@ def test_generate_snapshot_metadata(self):
metadata_directory, version, expiration_date, storage_backend = \
self._setup_generate_snapshot_metadata_test()

snapshot_metadata = \
snapshot_metadata, _ = \
repo_lib.generate_snapshot_metadata(metadata_directory, version,
expiration_date,
storage_backend,
Expand Down Expand Up @@ -527,7 +585,7 @@ def test_generate_snapshot_metadata_with_length(self):
metadata_directory, version, expiration_date, storage_backend = \
self._setup_generate_snapshot_metadata_test()

snapshot_metadata = \
snapshot_metadata, _ = \
repo_lib.generate_snapshot_metadata(metadata_directory, version,
expiration_date,
storage_backend,
Expand Down Expand Up @@ -558,7 +616,7 @@ def test_generate_snapshot_metadata_with_hashes(self):
metadata_directory, version, expiration_date, storage_backend = \
self._setup_generate_snapshot_metadata_test()

snapshot_metadata = \
snapshot_metadata, _ = \
repo_lib.generate_snapshot_metadata(metadata_directory, version,
expiration_date,
storage_backend,
Expand Down Expand Up @@ -589,7 +647,7 @@ def test_generate_snapshot_metadata_with_hashes_and_length(self):
metadata_directory, version, expiration_date, storage_backend = \
self._setup_generate_snapshot_metadata_test()

snapshot_metadata = \
snapshot_metadata, _ = \
repo_lib.generate_snapshot_metadata(metadata_directory, version,
expiration_date,
storage_backend,
Expand Down
20 changes: 19 additions & 1 deletion tests/test_repository_tool.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -247,6 +247,21 @@ def test_writeall(self):
# Verify that status() does not raise an exception.
repository.status()

# Test writeall with generating a snapshot RSA accumulator
repository.mark_dirty(['role1', 'targets', 'root', 'snapshot', 'timestamp'])
repository.writeall(rsa_acc=True)

# Were the RSA proof snapshots written?
targets_snapshot_filepath = os.path.join(metadata_directory,
'targets-snapshot.json')
targets_snapshot = securesystemslib.util.load_json_file(targets_snapshot_filepath)
tuf.formats.SNAPSHOT_RSA_ACC_SCHEMA.check_match(targets_snapshot)

# Does timestamp have the root hash?
timestamp_filepath = os.path.join(metadata_directory, 'timestamp.json')
timestamp = securesystemslib.util.load_json_file(timestamp_filepath)
timestamp['signed']['rsa_acc']

# Verify that status() does not raise
# 'tuf.exceptions.InsufficientKeysError' if a top-level role
# does not contain a threshold of keys.
Expand Down Expand Up @@ -488,10 +503,13 @@ def test_get_filepaths_in_directory(self):
# Construct list of file paths expected, determining absolute paths.
expected_files = []
for filepath in ['1.root.json', 'root.json', 'targets.json',
'snapshot.json', 'timestamp.json', 'role1.json', 'role2.json']:
'snapshot.json', 'timestamp.json', 'role1.json', 'role2.json',
'targets-snapshot.json', 'timestamp-rsa.json',
'role1-snapshot.json', 'role2-snapshot.json']:
expected_files.append(os.path.abspath(os.path.join(
'repository_data', 'repository', 'metadata', filepath)))

print(sorted(metadata_files))
self.assertEqual(sorted(expected_files), sorted(metadata_files))


Expand Down
29 changes: 29 additions & 0 deletions tests/test_updater.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1771,6 +1771,35 @@ def test_13__targets_of_role(self):



def test_snapshot_rsa_acc(self):
# replace timestamp with an RSA accumulator timestamp and create the updater
rsa_acc_timestamp = os.path.join(self.repository_directory, 'metadata', 'timestamp-rsa.json')
timestamp = os.path.join(self.repository_directory, 'metadata', 'timestamp.json')

shutil.move(rsa_acc_timestamp, timestamp)

repository_updater = updater.Updater(self.repository_name,
self.repository_mirrors)
repository_updater.refresh()

# Test verify RSA accumulator proof
snapshot_info = repository_updater.verify_rsa_acc_proof('targets')
self.assertEqual(snapshot_info['version'], 1)

snapshot_info = repository_updater.verify_rsa_acc_proof('role1')
self.assertEqual(snapshot_info['version'], 1)

# verify RSA accumulator with invalid role
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Would like to also see a test for verifying an existing role against an invalid accumulator

self.assertRaises(tuf.exceptions.NoWorkingMirrorError,
repository_updater.verify_rsa_acc_proof, 'foo')

# Test get_one_valid_targetinfo with snapshot RSA accumulator
repository_updater.get_one_valid_targetinfo('file1.txt')





class TestMultiRepoUpdater(unittest_toolbox.Modified_TestCase):

def setUp(self):
Expand Down
Loading