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

Jump to bottom

Offline mode #2472

Closed
wants to merge 9 commits into from
Closed

Offline mode #2472

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
da0ef28
ngclient: Add an optional offline mode
jku Sep 14, 2023
4f3d2c9
ngclient: Refactor offline mode
jku Sep 26, 2023
9084b22
ngclient: Improve delegated targets offline handling
jku Sep 26, 2023
7e6fbbd
tests: Refactor to keep mypy happy
jku Sep 26, 2023
e138cb6
ngclient: Standardize the offline exceptions
jku Sep 27, 2023
96d55dc
tests: Add initial tests for ngclient offline mode
jku Sep 27, 2023
3a35c04
ngclient: Document the offline config
jku Sep 27, 2023
dcad0ab
tests: Create subtest matrix for TrustedMetadataSet
jku Sep 27, 2023
1229e54
ngclient: Refactor offline Updater
jku Sep 27, 2023
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
197 changes: 162 additions & 35 deletions tests/test_trusted_metadata_set.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,12 @@

logger = logging.getLogger(__name__)

# test matrix for all tests that should behave similarly regardless of respect_expiry
respect_expiry_matrix: utils.DataSet = {
"respect_expiry=True": True,
"respect_expiry=False": False,
}


# pylint: disable=too-many-public-methods
class TestTrustedMetadataSet(unittest.TestCase):
Expand Down Expand Up @@ -93,7 +99,12 @@ def hashes_length_modifier(timestamp: Timestamp) -> None:
)

def setUp(self) -> None:
self.trusted_set = TrustedMetadataSet(self.metadata[Root.type])
self.trusted_set: TrustedMetadataSet

def _init(self, respect_expiry: bool) -> None:
self.trusted_set = TrustedMetadataSet(
self.metadata[Root.type], respect_expiry
)

def _update_all_besides_targets(
self,
Expand All @@ -117,7 +128,9 @@ def _update_all_besides_targets(
snapshot_bytes = snapshot_bytes or self.metadata[Snapshot.type]
self.trusted_set.update_snapshot(snapshot_bytes)

def test_update(self) -> None:
@utils.run_sub_tests_with_dataset(respect_expiry_matrix)
def test_update(self, respect_expiry: bool) -> None:
self._init(respect_expiry)
self.trusted_set.update_timestamp(self.metadata[Timestamp.type])
self.trusted_set.update_snapshot(self.metadata[Snapshot.type])
self.trusted_set.update_targets(self.metadata[Targets.type])
Expand All @@ -137,7 +150,9 @@ def test_update(self) -> None:

self.assertTrue(count, 6)

def test_update_metadata_output(self) -> None:
@utils.run_sub_tests_with_dataset(respect_expiry_matrix)
def test_update_metadata_output(self, respect_expiry: bool) -> None:
self._init(respect_expiry)
timestamp = self.trusted_set.update_timestamp(
self.metadata["timestamp"]
)
Expand All @@ -155,7 +170,9 @@ def test_update_metadata_output(self) -> None:
self.assertIsInstance(delegeted_targets_1.signed, Targets)
self.assertIsInstance(delegeted_targets_2.signed, Targets)

def test_out_of_order_ops(self) -> None:
@utils.run_sub_tests_with_dataset(respect_expiry_matrix)
def test_out_of_order_ops(self, respect_expiry: bool) -> None:
self._init(respect_expiry)
# Update snapshot before timestamp
with self.assertRaises(RuntimeError):
self.trusted_set.update_snapshot(self.metadata[Snapshot.type])
Expand Down Expand Up @@ -192,24 +209,42 @@ def test_out_of_order_ops(self) -> None:
self.metadata["role1"], "role1", Targets.type
)

def test_root_with_invalid_json(self) -> None:
# Test loading initial root and root update
for test_func in [TrustedMetadataSet, self.trusted_set.update_root]:
# root is not json
with self.assertRaises(exceptions.RepositoryError):
test_func(b"")
@utils.run_sub_tests_with_dataset(respect_expiry_matrix)
def test_initial_root_with_invalid_json(self, respect_expiry: bool) -> None:
# root is not json
with self.assertRaises(exceptions.RepositoryError):
TrustedMetadataSet(b"", respect_expiry)

# root is invalid
root = Metadata.from_bytes(self.metadata[Root.type])
root.signed.version += 1
with self.assertRaises(exceptions.UnsignedMetadataError):
test_func(root.to_bytes())
# root is invalid
root = Metadata.from_bytes(self.metadata[Root.type])
root.signed.version += 1
with self.assertRaises(exceptions.UnsignedMetadataError):
TrustedMetadataSet(root.to_bytes(), respect_expiry)

# metadata is of wrong type
with self.assertRaises(exceptions.RepositoryError):
test_func(self.metadata[Snapshot.type])
# metadata is of wrong type
with self.assertRaises(exceptions.RepositoryError):
TrustedMetadataSet(self.metadata[Snapshot.type], respect_expiry)

@utils.run_sub_tests_with_dataset(respect_expiry_matrix)
def test_root_with_invalid_json(self, respect_expiry: bool) -> None:
self._init(respect_expiry)
# root is not json
with self.assertRaises(exceptions.RepositoryError):
self.trusted_set.update_root(b"")

# root is invalid
root = Metadata.from_bytes(self.metadata[Root.type])
root.signed.version += 1
with self.assertRaises(exceptions.UnsignedMetadataError):
self.trusted_set.update_root(root.to_bytes())

# metadata is of wrong type
with self.assertRaises(exceptions.RepositoryError):
self.trusted_set.update_root(self.metadata[Snapshot.type])

def test_top_level_md_with_invalid_json(self) -> None:
@utils.run_sub_tests_with_dataset(respect_expiry_matrix)
def test_top_level_md_with_invalid_json(self, respect_expiry: bool) -> None:
self._init(respect_expiry)
top_level_md: List[Tuple[bytes, Callable[[bytes], Metadata]]] = [
(self.metadata[Timestamp.type], self.trusted_set.update_timestamp),
(self.metadata[Snapshot.type], self.trusted_set.update_snapshot),
Expand All @@ -232,15 +267,23 @@ def test_top_level_md_with_invalid_json(self) -> None:

update_func(metadata)

def test_update_root_new_root(self) -> None:
@utils.run_sub_tests_with_dataset(respect_expiry_matrix)
def test_update_root_new_root(self, respect_expiry: bool) -> None:
self._init(respect_expiry)

# test that root can be updated with a new valid version
def root_new_version_modifier(root: Root) -> None:
root.version += 1

root = self.modify_metadata(Root.type, root_new_version_modifier)
self.trusted_set.update_root(root)

def test_update_root_new_root_fail_threshold_verification(self) -> None:
@utils.run_sub_tests_with_dataset(respect_expiry_matrix)
def test_update_root_new_root_fail_threshold_verification(
self, respect_expiry: bool
) -> None:
self._init(respect_expiry)

# Increase threshold in new root, do not add enough keys
def root_threshold_bump(root: Root) -> None:
root.version += 1
Expand All @@ -250,7 +293,11 @@ def root_threshold_bump(root: Root) -> None:
with self.assertRaises(exceptions.UnsignedMetadataError):
self.trusted_set.update_root(root)

def test_update_root_new_root_ver_same_as_trusted_root_ver(self) -> None:
@utils.run_sub_tests_with_dataset(respect_expiry_matrix)
def test_update_root_new_root_ver_same_as_trusted_root_ver(
self, respect_expiry: bool
) -> None:
self._init(respect_expiry)
with self.assertRaises(exceptions.BadVersionNumberError):
self.trusted_set.update_root(self.metadata[Root.type])

Expand All @@ -260,12 +307,22 @@ def root_expired_modifier(root: Root) -> None:

# intermediate root can be expired
root = self.modify_metadata(Root.type, root_expired_modifier)
tmp_trusted_set = TrustedMetadataSet(root)
tmp_trusted_set = TrustedMetadataSet(root, True)
# update timestamp to trigger final root expiry check
with self.assertRaises(exceptions.ExpiredMetadataError):
tmp_trusted_set.update_timestamp(self.metadata[Timestamp.type])

def test_update_timestamp_new_timestamp_ver_below_trusted_ver(self) -> None:
# If we decide to not respect expiry, it all works:
root = self.modify_metadata(Root.type, root_expired_modifier)
tmp_trusted_set = TrustedMetadataSet(root, False)
tmp_trusted_set.update_timestamp(self.metadata[Timestamp.type])

@utils.run_sub_tests_with_dataset(respect_expiry_matrix)
def test_update_timestamp_new_timestamp_ver_below_trusted_ver(
self, respect_expiry: bool
) -> None:
self._init(respect_expiry)

# new_timestamp.version < trusted_timestamp.version
def version_modifier(timestamp: Timestamp) -> None:
timestamp.version = 3
Expand All @@ -275,7 +332,11 @@ def version_modifier(timestamp: Timestamp) -> None:
with self.assertRaises(exceptions.BadVersionNumberError):
self.trusted_set.update_timestamp(self.metadata[Timestamp.type])

def test_update_timestamp_with_same_timestamp(self) -> None:
@utils.run_sub_tests_with_dataset(respect_expiry_matrix)
def test_update_timestamp_with_same_timestamp(
self, respect_expiry: bool
) -> None:
self._init(respect_expiry)
# Test that timestamp is NOT updated if:
# new_timestamp.version == trusted_timestamp.version
self.trusted_set.update_timestamp(self.metadata[Timestamp.type])
Expand All @@ -289,7 +350,12 @@ def test_update_timestamp_with_same_timestamp(self) -> None:
# was not updated.
self.assertEqual(id(initial_timestamp), id(self.trusted_set.timestamp))

def test_update_timestamp_snapshot_ver_below_current(self) -> None:
@utils.run_sub_tests_with_dataset(respect_expiry_matrix)
def test_update_timestamp_snapshot_ver_below_current(
self, respect_expiry: bool
) -> None:
self._init(respect_expiry)

def bump_snapshot_version(timestamp: Timestamp) -> None:
timestamp.snapshot_meta.version = 2
# The timestamp version must be increased to initiate a update.
Expand All @@ -304,9 +370,12 @@ def bump_snapshot_version(timestamp: Timestamp) -> None:
self.trusted_set.update_timestamp(self.metadata[Timestamp.type])

def test_update_timestamp_expired(self) -> None:
self._init(respect_expiry=True)

# new_timestamp has expired
def timestamp_expired_modifier(timestamp: Timestamp) -> None:
timestamp.expires = datetime(1970, 1, 1)
timestamp.version = 2

# expired intermediate timestamp is loaded but raises
timestamp = self.modify_metadata(
Expand All @@ -319,7 +388,17 @@ def timestamp_expired_modifier(timestamp: Timestamp) -> None:
with self.assertRaises(exceptions.ExpiredMetadataError):
self.trusted_set.update_snapshot(self.metadata[Snapshot.type])

def test_update_snapshot_length_or_hash_mismatch(self) -> None:
# If we decide to not respect expiry, it all works:
self._init(respect_expiry=False)
self.trusted_set.update_timestamp(timestamp)
self.trusted_set.update_snapshot(self.metadata[Snapshot.type])

@utils.run_sub_tests_with_dataset(respect_expiry_matrix)
def test_update_snapshot_length_or_hash_mismatch(
self, respect_expiry: bool
) -> None:
self._init(respect_expiry)

def modify_snapshot_length(timestamp: Timestamp) -> None:
timestamp.snapshot_meta.length = 1

Expand All @@ -330,16 +409,23 @@ def modify_snapshot_length(timestamp: Timestamp) -> None:
with self.assertRaises(exceptions.RepositoryError):
self.trusted_set.update_snapshot(self.metadata[Snapshot.type])

def test_update_snapshot_fail_threshold_verification(self) -> None:
@utils.run_sub_tests_with_dataset(respect_expiry_matrix)
def test_update_snapshot_fail_threshold_verification(
self, respect_expiry: bool
) -> None:
self._init(respect_expiry)
self.trusted_set.update_timestamp(self.metadata[Timestamp.type])
snapshot = Metadata.from_bytes(self.metadata[Snapshot.type])
snapshot.signatures.clear()
with self.assertRaises(exceptions.UnsignedMetadataError):
self.trusted_set.update_snapshot(snapshot.to_bytes())

@utils.run_sub_tests_with_dataset(respect_expiry_matrix)
def test_update_snapshot_version_diverge_timestamp_snapshot_version(
self,
self, respect_expiry: bool
) -> None:
self._init(respect_expiry)

def timestamp_version_modifier(timestamp: Timestamp) -> None:
timestamp.snapshot_meta.version = 2

Expand All @@ -356,7 +442,11 @@ def timestamp_version_modifier(timestamp: Timestamp) -> None:
with self.assertRaises(exceptions.BadVersionNumberError):
self.trusted_set.update_targets(self.metadata[Targets.type])

def test_update_snapshot_file_removed_from_meta(self) -> None:
@utils.run_sub_tests_with_dataset(respect_expiry_matrix)
def test_update_snapshot_file_removed_from_meta(
self, respect_expiry: bool
) -> None:
self._init(respect_expiry)
self._update_all_besides_targets(self.metadata[Timestamp.type])

def remove_file_from_meta(snapshot: Snapshot) -> None:
Expand All @@ -367,7 +457,11 @@ def remove_file_from_meta(snapshot: Snapshot) -> None:
with self.assertRaises(exceptions.RepositoryError):
self.trusted_set.update_snapshot(snapshot)

def test_update_snapshot_meta_version_decreases(self) -> None:
@utils.run_sub_tests_with_dataset(respect_expiry_matrix)
def test_update_snapshot_meta_version_decreases(
self, respect_expiry: bool
) -> None:
self._init(respect_expiry)
self.trusted_set.update_timestamp(self.metadata[Timestamp.type])

def version_meta_modifier(snapshot: Snapshot) -> None:
Expand All @@ -380,6 +474,7 @@ def version_meta_modifier(snapshot: Snapshot) -> None:
self.trusted_set.update_snapshot(self.metadata[Snapshot.type])

def test_update_snapshot_expired_new_snapshot(self) -> None:
self._init(respect_expiry=True)
self.trusted_set.update_timestamp(self.metadata[Timestamp.type])

def snapshot_expired_modifier(snapshot: Snapshot) -> None:
Expand All @@ -396,7 +491,18 @@ def snapshot_expired_modifier(snapshot: Snapshot) -> None:
with self.assertRaises(exceptions.ExpiredMetadataError):
self.trusted_set.update_targets(self.metadata[Targets.type])

def test_update_snapshot_successful_rollback_checks(self) -> None:
# If we decide to not respect expiry, it all works:
self._init(respect_expiry=False)
self.trusted_set.update_timestamp(self.metadata[Timestamp.type])
self.trusted_set.update_snapshot(snapshot)
self.trusted_set.update_targets(self.metadata[Targets.type])

@utils.run_sub_tests_with_dataset(respect_expiry_matrix)
def test_update_snapshot_successful_rollback_checks(
self, respect_expiry: bool
) -> None:
self._init(respect_expiry)

def meta_version_bump(timestamp: Timestamp) -> None:
timestamp.snapshot_meta.version += 1
# The timestamp version must be increased to initiate a update.
Expand All @@ -420,7 +526,12 @@ def version_bump(snapshot: Snapshot) -> None:
# update targets to trigger final snapshot meta version check
self.trusted_set.update_targets(self.metadata[Targets.type])

def test_update_targets_no_meta_in_snapshot(self) -> None:
@utils.run_sub_tests_with_dataset(respect_expiry_matrix)
def test_update_targets_no_meta_in_snapshot(
self, respect_expiry: bool
) -> None:
self._init(respect_expiry)

def no_meta_modifier(snapshot: Snapshot) -> None:
snapshot.meta = {}

Expand All @@ -432,7 +543,12 @@ def no_meta_modifier(snapshot: Snapshot) -> None:
with self.assertRaises(exceptions.RepositoryError):
self.trusted_set.update_targets(self.metadata[Targets.type])

def test_update_targets_hash_diverge_from_snapshot_meta_hash(self) -> None:
@utils.run_sub_tests_with_dataset(respect_expiry_matrix)
def test_update_targets_hash_diverge_from_snapshot_meta_hash(
self, respect_expiry: bool
) -> None:
self._init(respect_expiry)

def meta_length_modifier(snapshot: Snapshot) -> None:
for metafile_path in snapshot.meta:
snapshot.meta[metafile_path] = MetaFile(version=1, length=1)
Expand All @@ -445,7 +561,12 @@ def meta_length_modifier(snapshot: Snapshot) -> None:
with self.assertRaises(exceptions.RepositoryError):
self.trusted_set.update_targets(self.metadata[Targets.type])

def test_update_targets_version_diverge_snapshot_meta_version(self) -> None:
@utils.run_sub_tests_with_dataset(respect_expiry_matrix)
def test_update_targets_version_diverge_snapshot_meta_version(
self, respect_expiry: bool
) -> None:
self._init(respect_expiry)

def meta_modifier(snapshot: Snapshot) -> None:
for metafile_path in snapshot.meta:
snapshot.meta[metafile_path] = MetaFile(version=2)
Expand All @@ -459,6 +580,7 @@ def meta_modifier(snapshot: Snapshot) -> None:
self.trusted_set.update_targets(self.metadata[Targets.type])

def test_update_targets_expired_new_target(self) -> None:
self._init(respect_expiry=True)
self._update_all_besides_targets()

# new_delegated_target has expired
Expand All @@ -469,6 +591,11 @@ def target_expired_modifier(target: Targets) -> None:
with self.assertRaises(exceptions.ExpiredMetadataError):
self.trusted_set.update_targets(targets)

# If we decide to not respect expiry, it all works:
self._init(respect_expiry=False)
self._update_all_besides_targets()
self.trusted_set.update_targets(targets)

# TODO test updating over initial metadata (new keys, newer timestamp, etc)


Expand Down
Loading
Loading