Merge pull request #1605 from MVrachev/snapshot-hashes-length-check

Introduce the idea of trusted/untrusted snapshot

Author: Jussi Kukkonen

tests/repository_simulator.py

@@ -50,14 +50,14 @@
 import logging
 import os
 import tempfile
-from securesystemslib.hash import digest
+import securesystemslib.hash as sslib_hash
 from securesystemslib.keys import generate_ed25519_key
 from securesystemslib.signer import SSlibSigner
 from typing import Dict, Iterator, List, Optional, Tuple
 from urllib import parse
 
 from tuf.api.serialization.json import JSONSerializer
-from tuf.exceptions import FetcherHTTPError, RepositoryError
+from tuf.exceptions import FetcherHTTPError
 from tuf.api.metadata import (
     Key,
     Metadata,
@@ -100,6 +100,9 @@ def __init__(self):
         # target downloads are served from this dict
         self.target_files: Dict[str, RepositoryTarget] = {}
 
+        # Whether to compute hashes and legth for meta in snapshot/timestamp
+        self.compute_metafile_hashes_length = False
+
         self.dump_dir = None
         self.dump_version = 0
 
@@ -121,7 +124,8 @@ def snapshot(self) -> Snapshot:
     def targets(self) -> Targets:
         return self.md_targets.signed
 
-    def delegates(self) -> Iterator[Tuple[str, Targets]]:
+    def all_targets(self) -> Iterator[Tuple[str, Targets]]:
+        yield "targets", self.md_targets.signed
         for role, md in self.md_delegates.items():
             yield role, md.signed
 
@@ -243,16 +247,34 @@ def _fetch_metadata(self, role: str, version: Optional[int] = None) -> bytes:
             )
             return md.to_bytes(JSONSerializer())
 
+    def _compute_hashes_and_length(
+        self, role: str
+    ) -> Tuple[Dict[str, str], int]:
+        data = self._fetch_metadata(role)
+        digest_object = sslib_hash.digest(sslib_hash.DEFAULT_HASH_ALGORITHM)
+        digest_object.update(data)
+        hashes = {sslib_hash.DEFAULT_HASH_ALGORITHM:  digest_object.hexdigest()}
+        return hashes, len(data)
+
     def update_timestamp(self):
         self.timestamp.snapshot_meta.version = self.snapshot.version
 
+        if self.compute_metafile_hashes_length:
+            hashes, length = self._compute_hashes_and_length("snapshot")
+            self.timestamp.snapshot_meta.hashes = hashes
+            self.timestamp.snapshot_meta.length = length
+
         self.timestamp.version += 1
 
     def update_snapshot(self):
-        self.snapshot.meta["targets.json"].version = self.targets.version
-        for role, delegate in self.delegates():
+        for role, delegate in self.all_targets():
             self.snapshot.meta[f"{role}.json"].version = delegate.version
 
+            if self.compute_metafile_hashes_length:
+                hashes, length = self._compute_hashes_and_length(role)
+                self.snapshot.meta[f"{role}.json"].hashes = hashes
+                self.snapshot.meta[f"{role}.json"].length = length
+
         self.snapshot.version += 1
         self.update_timestamp()
 

tests/test_updater_with_simulator.py

@@ -6,18 +6,18 @@
 """Test ngclient Updater using the repository simulator
 """
 
-import logging
 import os
 import sys
 import tempfile
 from typing import Optional, Tuple
-from tuf.exceptions import UnsignedMetadataError
+from tuf.exceptions import UnsignedMetadataError, BadVersionNumberError
 import unittest
 
 from tuf.ngclient import Updater
 
 from tests import utils
 from tests.repository_simulator import RepositorySimulator
+from securesystemslib import hash as sslib_hash
 
 
 class TestUpdater(unittest.TestCase):
@@ -164,6 +164,37 @@ def test_keys_and_signatures(self):
 
         self._run_refresh()
 
+    def test_snapshot_rollback_with_local_snapshot_hash_mismatch(self):
+        # Test triggering snapshot rollback check on a newly downloaded snapshot
+        # when the local snapshot is loaded even when there is a hash mismatch
+        # with timestamp.snapshot_meta.
+
+        # By raising this flag on timestamp update the simulator would:
+        # 1) compute the hash of the new modified version of snapshot
+        # 2) assign the hash to timestamp.snapshot_meta
+        # The purpose is to create a hash mismatch between timestamp.meta and
+        # the local snapshot, but to have hash match between timestamp.meta and
+        # the next snapshot version.
+        self.sim.compute_metafile_hashes_length = True
+
+        # Initialize all metadata and assign targets version higher than 1.
+        self.sim.targets.version = 2
+        self.sim.update_snapshot()
+        self._run_refresh()
+
+        # The new targets should have a lower version than the local trusted one.
+        self.sim.targets.version = 1
+        self.sim.update_snapshot()
+
+        # During the snapshot update, the local snapshot will be loaded even if
+        # there is a hash mismatch with timestamp.snapshot_meta, because it will
+        # be considered as trusted.
+        # Should fail as a new version of snapshot will be fetched which lowers
+        # the snapshot.meta["targets.json"] version by 1 and throws an error.
+        with self.assertRaises(BadVersionNumberError):
+            self._run_refresh()
+
+
 if __name__ == "__main__":
     if "--dump" in sys.argv:
         TestUpdater.dump_dir = tempfile.mkdtemp()

tuf/ngclient/_internal/trusted_metadata_set.py

@@ -258,7 +258,9 @@ def _check_final_timestamp(self) -> None:
         if self.timestamp.signed.is_expired(self.reference_time):
             raise exceptions.ExpiredMetadataError("timestamp.json is expired")
 
-    def update_snapshot(self, data: bytes) -> None:
+    def update_snapshot(
+        self, data: bytes, trusted: Optional[bool] = False
+    ) -> None:
         """Verifies and loads 'data' as new snapshot metadata.
 
         Note that an intermediate snapshot is allowed to be expired and version
@@ -271,6 +273,11 @@ def update_snapshot(self, data: bytes) -> None:
 
         Args:
             data: unverified new snapshot metadata as bytes
+            trusted: whether data has at some point been verified by
+                TrustedMetadataSet as a valid snapshot. Purpose of trusted is
+                to allow loading of locally stored snapshot as intermediate
+                snapshot even if hashes in current timestamp meta no longer
+                match data. Default is False.
 
         Raises:
             RepositoryError: data failed to load or verify as final snapshot.
@@ -288,13 +295,15 @@ def update_snapshot(self, data: bytes) -> None:
 
         snapshot_meta = self.timestamp.signed.snapshot_meta
 
-        # Verify against the hashes in timestamp, if any
-        try:
-            snapshot_meta.verify_length_and_hashes(data)
-        except exceptions.LengthOrHashMismatchError as e:
-            raise exceptions.RepositoryError(
-                "Snapshot length or hashes do not match"
-            ) from e
+        # Verify non-trusted data against the hashes in timestamp, if any.
+        # Trusted snapshot data has already been verified once.
+        if not trusted:
+            try:
+                snapshot_meta.verify_length_and_hashes(data)
+            except exceptions.LengthOrHashMismatchError as e:
+                raise exceptions.RepositoryError(
+                    "Snapshot length or hashes do not match"
+                ) from e
 
         try:
             new_snapshot = Metadata[Snapshot].from_bytes(data)

tuf/ngclient/updater.py

@@ -346,7 +346,7 @@ def _load_snapshot(self) -> None:
         """Load local (and if needed remote) snapshot metadata"""
         try:
             data = self._load_local_metadata("snapshot")
-            self._trusted_set.update_snapshot(data)
+            self._trusted_set.update_snapshot(data, trusted=True)
             logger.debug("Local snapshot is valid: not downloading new one")
         except (OSError, exceptions.RepositoryError) as e:
             # Local snapshot does not exist or is invalid: update from remote