Add hash and length verification to MetaFile and TargetFile

tests/test_api.py

@@ -20,7 +20,7 @@
 
 from tests import utils
 
-import tuf.exceptions
+from tuf import exceptions
 from tuf.api.metadata import (
     Metadata,
     Root,
@@ -178,7 +178,7 @@ def test_sign_verify(self):
         self.assertTrue(len(metadata_obj.signatures) == 1)
         # ... which is valid for the correct key.
         targets_key.verify_signature(metadata_obj)
-        with self.assertRaises(tuf.exceptions.UnsignedMetadataError):
+        with self.assertRaises(exceptions.UnsignedMetadataError):
             snapshot_key.verify_signature(metadata_obj)
 
         sslib_signer = SSlibSigner(self.keystore['snapshot'])
@@ -197,7 +197,7 @@ def test_sign_verify(self):
         self.assertTrue(len(metadata_obj.signatures) == 1)
         # ... valid for that key.
         timestamp_key.verify_signature(metadata_obj)
-        with self.assertRaises(tuf.exceptions.UnsignedMetadataError):
+        with self.assertRaises(exceptions.UnsignedMetadataError):
             targets_key.verify_signature(metadata_obj)
 
 
@@ -280,7 +280,6 @@ def test_targetfile_class(self):
         targetfile_obj = TargetFile.from_dict(copy.copy(data))
         self.assertEqual(targetfile_obj.to_dict(), data)
 
-
     def test_metadata_snapshot(self):
         snapshot_path = os.path.join(
                 self.repo_dir, 'metadata', 'snapshot.json')
@@ -352,6 +351,7 @@ def test_metadata_timestamp(self):
         timestamp_test = Timestamp.from_dict(test_dict)
         self.assertEqual(timestamp_dict['signed'], timestamp_test.to_dict())
 
+
     def test_key_class(self):
         keys = {
             "59a4df8af818e9ed7abe0764c0b47b4240952aa0d179b5b78346c470ac30278d":{
@@ -638,6 +638,66 @@ def test_support_for_unrecognized_fields(self):
                 metadata_obj.signed.to_dict(), metadata_obj2.signed.to_dict()
             )
 
+    def  test_length_and_hash_validation(self):
+
+        # Test metadata files' hash and length verification.
+        # Use timestamp to get a MetaFile object and snapshot
+        # for untrusted metadata file to verify.
+        timestamp_path = os.path.join(
+            self.repo_dir, 'metadata', 'timestamp.json')
+        timestamp = Metadata.from_file(timestamp_path)
+        snapshot_metafile = timestamp.signed.meta["snapshot.json"]
+
+        snapshot_path = os.path.join(
+            self.repo_dir, 'metadata', 'snapshot.json')
+
+        with open(snapshot_path, "rb") as file:
+            # test with  data as a file object
+            snapshot_metafile.verify_length_and_hashes(file)
+            file.seek(0)
+            data = file.read()
+            # test with data as bytes
+            snapshot_metafile.verify_length_and_hashes(data)
+
+            # test exceptions
+            expected_length = snapshot_metafile.length
+            snapshot_metafile.length = 2345
+            self.assertRaises(exceptions.LengthOrHashMismatchError,
+                snapshot_metafile.verify_length_and_hashes, data)
+
+            snapshot_metafile.length = expected_length
+            snapshot_metafile.hashes = {'sha256': 'incorrecthash'}
+            self.assertRaises(exceptions.LengthOrHashMismatchError,
+                snapshot_metafile.verify_length_and_hashes, data)
+
+            # test optional length and hashes
+            snapshot_metafile.length = None
+            snapshot_metafile.hashes = None
+            snapshot_metafile.verify_length_and_hashes(data)
+
+
+        # Test target files' hash and length verification
+        targets_path = os.path.join(
+            self.repo_dir, 'metadata', 'targets.json')
+        targets = Metadata.from_file(targets_path)
+        file1_targetfile = targets.signed.targets['file1.txt']
+        filepath = os.path.join(
+            self.repo_dir, 'targets', 'file1.txt')
+
+        with open(filepath, "rb") as file1:
+            file1_targetfile.verify_length_and_hashes(file1)
+
+            # test exceptions
+            expected_length = file1_targetfile.length
+            file1_targetfile.length = 2345
+            self.assertRaises(exceptions.LengthOrHashMismatchError,
+                file1_targetfile.verify_length_and_hashes, file1)
+
+            file1_targetfile.length = expected_length
+            file1_targetfile.hashes = {'sha256': 'incorrecthash'}
+            self.assertRaises(exceptions.LengthOrHashMismatchError,
+                file1_targetfile.verify_length_and_hashes, file1)
+
 
 # Run unit test.
 if __name__ == '__main__':

tuf/api/metadata.py

@@ -16,10 +16,23 @@
 
 """
 import abc
+import io
 import tempfile
 from datetime import datetime, timedelta
-from typing import Any, ClassVar, Dict, List, Mapping, Optional, Tuple, Type
+from typing import (
+    Any,
+    BinaryIO,
+    ClassVar,
+    Dict,
+    List,
+    Mapping,
+    Optional,
+    Tuple,
+    Type,
+    Union,
+)
 
+from securesystemslib import hash as sslib_hash
 from securesystemslib import keys as sslib_keys
 from securesystemslib.signer import Signature, Signer
 from securesystemslib.storage import FilesystemBackend, StorageBackendInterface
@@ -622,7 +635,53 @@ def remove_key(self, role: str, keyid: str) -> None:
         del self.keys[keyid]
 
 
-class MetaFile:
+class BaseFile:
+    """A base class of MetaFile and TargetFile.
+
+    Encapsulates common static methods for length and hash verification.
+    """
+
+    @staticmethod
+    def _verify_hashes(
+        data: Union[bytes, BinaryIO], expected_hashes: Dict[str, str]
+    ) -> None:
+        """Verifies that the hash of 'data' matches 'expected_hashes'"""
+        is_bytes = isinstance(data, bytes)
+        for algo, exp_hash in expected_hashes.items():
+            if is_bytes:
+                digest_object = sslib_hash.digest(algo)
+                digest_object.update(data)
+            else:
+                # if data is not bytes, assume it is a file object
+                digest_object = sslib_hash.digest_fileobject(data, algo)
+
+            observed_hash = digest_object.hexdigest()
+            if observed_hash != exp_hash:
+                raise exceptions.LengthOrHashMismatchError(
+                    f"Observed hash {observed_hash} does not match"
+                    f"expected hash {exp_hash}"
+                )
+
+    @staticmethod
+    def _verify_length(
+        data: Union[bytes, BinaryIO], expected_length: int
+    ) -> None:
+        """Verifies that the length of 'data' matches 'expected_length'"""
+        if isinstance(data, bytes):
+            observed_length = len(data)
+        else:
+            # if data is not bytes, assume it is a file object
+            data.seek(0, io.SEEK_END)
+            observed_length = data.tell()
+
+        if observed_length != expected_length:
+            raise exceptions.LengthOrHashMismatchError(
+                f"Observed length {observed_length} does not match"
+                f"expected length {expected_length}"
+            )
+
+
+class MetaFile(BaseFile):
     """A container with information about a particular metadata file.
 
     Attributes:
@@ -660,6 +719,13 @@ def from_dict(cls, meta_dict: Dict[str, Any]) -> "MetaFile":
         version = meta_dict.pop("version")
         length = meta_dict.pop("length", None)
         hashes = meta_dict.pop("hashes", None)
+
+        # Do some basic input validation
+        if version <= 0:
+            raise ValueError(f"Metafile version must be > 0, got {version}")
+        if length is not None and length <= 0:
+            raise ValueError(f"Metafile length must be > 0, got {length}")
+
         # All fields left in the meta_dict are unrecognized.
         return cls(version, length, hashes, meta_dict)
 
@@ -678,6 +744,22 @@ def to_dict(self) -> Dict[str, Any]:
 
         return res_dict
 
+    def verify_length_and_hashes(self, data: Union[bytes, BinaryIO]):
+        """Verifies that the length and hashes of "data" match expected
+            values.
+        Args:
+            data: File object or its content in bytes.
+        Raises:
+            LengthOrHashMismatchError: Calculated length or hashes do not
+                match expected values.
+        """
+        if self.length is not None:
+            self._verify_length(data, self.length)
+
+        # Skip the check in case of an empty dictionary too
+        if self.hashes:
+            self._verify_hashes(data, self.hashes)
+
 
 class Timestamp(Signed):
     """A container for the signed part of timestamp metadata.
@@ -905,7 +987,7 @@ def to_dict(self) -> Dict[str, Any]:
         }
 
 
-class TargetFile:
+class TargetFile(BaseFile):
     """A container with information about a particular target file.
 
     Attributes:
@@ -923,12 +1005,6 @@ class TargetFile:
 
     """
 
-    @property
-    def custom(self):
-        if self.unrecognized_fields is None:
-            return None
-        return self.unrecognized_fields.get("custom", None)
-
     def __init__(
         self,
         length: int,
@@ -939,11 +1015,24 @@ def __init__(
         self.hashes = hashes
         self.unrecognized_fields = unrecognized_fields or {}
 
+    @property
+    def custom(self):
+        if self.unrecognized_fields is None:
+            return None
+        return self.unrecognized_fields.get("custom", None)
+
     @classmethod
     def from_dict(cls, target_dict: Dict[str, Any]) -> "TargetFile":
         """Creates TargetFile object from its dict representation."""
         length = target_dict.pop("length")
         hashes = target_dict.pop("hashes")
+
+        # Do some basic validation checks
+        if length <= 0:
+            raise ValueError(f"Targetfile length must be > 0, got {length}")
+        if not hashes:
+            raise ValueError("Missing targetfile hashes")
+
         # All fields left in the target_dict are unrecognized.
         return cls(length, hashes, target_dict)
 
@@ -955,6 +1044,18 @@ def to_dict(self) -> Dict[str, Any]:
             **self.unrecognized_fields,
         }
 
+    def verify_length_and_hashes(self, data: Union[bytes, BinaryIO]):
+        """Verifies that the length and hashes of "data" match expected
+            values.
+        Args:
+            data: File object or its content in bytes.
+        Raises:
+            LengthOrHashMismatchError: Calculated length or hashes do not
+                match expected values.
+        """
+        self._verify_length(data, self.length)
+        self._verify_hashes(data, self.hashes)
+
 
 class Targets(Signed):
     """A container for the signed part of targets metadata.

tuf/exceptions.py

@@ -65,6 +65,8 @@ def __repr__(self):
 class UnsupportedAlgorithmError(Error):
   """Indicate an error while trying to identify a user-specified algorithm."""
 
+class LengthOrHashMismatchError(Error):
+  """Indicate an error while checking the length and hash values of an object"""
 
 class BadHashError(Error):
   """Indicate an error while checking the value of a hash object."""
@@ -88,9 +90,6 @@ def __repr__(self):
     #     self.__class__.__name__ + '(' + repr(self.expected_hash) + ', ' +
     #     repr(self.observed_hash) + ')')
 
-
-
-
 class BadVersionNumberError(Error):
   """Indicate an error for metadata that contains an invalid version number."""