Metadata API: Move "typed constructors" to Signed

Change the way Metadata[Root] and other typed Metadata are constructed.
Before:
    root_md = Metadata._from_file(filename, signed_type=Root)
after:
    root_md = Root.metadata_from_file(filename)

This nicely removes the requirement for an extra argument.

This unfortunately adds a lot of lines to metadata.py but actual LOC
count does not really change -- it's just 2 additional abstract methods
and 4 one-liner implementations for each of those. The only things
these implementations do are:
 * annotates returns type as e.g. Metadata[Root]
 * raises if the signed type in input is incorrect

Signed-off-by: Jussi Kukkonen <jkukkonen@vmware.com>

Author: Jussi Kukkonen

tests/test_api.py

@@ -140,15 +140,15 @@ def test_typed_read(self):
             data = f.read()
 
         # Loading a root file as "Metadata[Root]" succeeds
-        md = Metadata.from_bytes(data, signed_type=Root)
-        md2 = Metadata.from_file(path, signed_type=Root)
+        md = Root.metadata_from_bytes(data)
+        md2 = Root.metadata_from_file(path)
 
         # Loading the file fails with non-"Root" type constraints
         for expected_type in [Timestamp, Snapshot, Targets]:
             with self.assertRaises(DeserializationError):
-                Metadata.from_bytes(data, signed_type=expected_type)
+                expected_type.metadata_from_bytes(data)
             with self.assertRaises(DeserializationError):
-                Metadata.from_file(path, signed_type=expected_type)
+                expected_type.metadata_from_file(path)
 
 
     def test_compact_json(self):
@@ -177,7 +177,7 @@ def test_read_write_read_compare(self):
 
     def test_sign_verify(self):
         root_path = os.path.join(self.repo_dir, 'metadata', 'root.json')
-        root = Metadata.from_file(root_path, signed_type=Root).signed
+        root = Root.metadata_from_file(root_path).signed
 
         # Locate the public keys we need from root
         targets_keyid = next(iter(root.roles["targets"].keyids))
@@ -189,7 +189,7 @@ def test_sign_verify(self):
 
         # Load sample metadata (targets) and assert ...
         path = os.path.join(self.repo_dir, 'metadata', 'targets.json')
-        metadata_obj = Metadata.from_file(path, signed_type=Targets)
+        metadata_obj = Targets.metadata_from_file(path)
 
         # ... it has a single existing signature,
         self.assertEqual(len(metadata_obj.signatures), 1)
@@ -306,7 +306,7 @@ def test_targetfile_class(self):
     def test_metadata_snapshot(self):
         snapshot_path = os.path.join(
                 self.repo_dir, 'metadata', 'snapshot.json')
-        snapshot = Metadata.from_file(snapshot_path, signed_type=Snapshot)
+        snapshot = Snapshot.metadata_from_file(snapshot_path)
 
         # Create a MetaFile instance representing what we expect
         # the updated data to be.
@@ -332,7 +332,7 @@ def test_metadata_snapshot(self):
     def test_metadata_timestamp(self):
         timestamp_path = os.path.join(
                 self.repo_dir, 'metadata', 'timestamp.json')
-        timestamp = Metadata.from_file(timestamp_path, signed_type=Timestamp)
+        timestamp = Timestamp.metadata_from_file(timestamp_path)
 
         self.assertEqual(timestamp.signed.version, 1)
         timestamp.signed.bump_version()
@@ -441,8 +441,7 @@ def test_role_class(self):
     def test_metadata_root(self):
         root_path = os.path.join(
                 self.repo_dir, 'metadata', 'root.json')
-        root = Metadata.from_file(root_path, signed_type=Root)
-
+        root = Root.metadata_from_file(root_path)
         # Add a second key to root role
         root_key2 =  import_ed25519_publickey_from_file(
                     os.path.join(self.keystore_dir, 'root_key2.pub'))
@@ -579,7 +578,7 @@ def test_delegation_class(self):
     def test_metadata_targets(self):
         targets_path = os.path.join(
                 self.repo_dir, 'metadata', 'targets.json')
-        targets = Metadata.from_file(targets_path, signed_type=Targets)
+        targets = Targets.metadata_from_file(targets_path)
 
         # Create a fileinfo dict representing what we expect the updated data to be
         filename = 'file2.txt'
@@ -668,7 +667,7 @@ def  test_length_and_hash_validation(self):
         # for untrusted metadata file to verify.
         timestamp_path = os.path.join(
             self.repo_dir, 'metadata', 'timestamp.json')
-        timestamp = Metadata.from_file(timestamp_path, signed_type=Timestamp)
+        timestamp = Timestamp.metadata_from_file(timestamp_path)
         snapshot_metafile = timestamp.signed.meta["snapshot.json"]
 
         snapshot_path = os.path.join(
@@ -702,7 +701,7 @@ def  test_length_and_hash_validation(self):
         # Test target files' hash and length verification
         targets_path = os.path.join(
             self.repo_dir, 'metadata', 'targets.json')
-        targets = Metadata.from_file(targets_path, signed_type=Targets)
+        targets = Targets.metadata_from_file(targets_path)
         file1_targetfile = targets.signed.targets['file1.txt']
         filepath = os.path.join(
             self.repo_dir, 'targets', 'file1.txt')

tuf/api/metadata.py

@@ -73,14 +73,11 @@ class Metadata(Generic[T]):
     [Root, Timestamp, Snapshot, Targets]. The purpose of this is to allow
     type checking of the signed attribute in code using Metadata::
 
-        root_md = Metadata.from_file("root.json", signed_type=Root)
-        # root_md type is now Metadata[Root]. This means signed and its
+        root_md = Root.metadata_from_file("root.json")
+        print(root_md.signed.consistent_snapshot)
+        # root_md type is now Metadata[Root]. This means root_md.signed and its
         # attributes like consistent_snapshot are now statically typed and the
         # types can be verified by static type checkers and shown by IDEs
-        print(root_md.signed.consistent_snapshot)
-
-    Using the signed_type argument in factory constructors is not required but
-    not doing so means T is not a specific type so static typing cannot happen.
 
     Attributes:
         signed: A subclass of Signed, which has the actual metadata payload,
@@ -147,7 +144,6 @@ def from_file(
         filename: str,
         deserializer: Optional[MetadataDeserializer] = None,
         storage_backend: Optional[StorageBackendInterface] = None,
-        signed_type: Optional[Type[T]] = None,
     ) -> "Metadata[T]":
         """Loads TUF metadata from file storage.
 
@@ -159,7 +155,6 @@ def from_file(
             storage_backend: An object that implements
                 securesystemslib.storage.StorageBackendInterface. Per default
                 a (local) FilesystemBackend is used.
-            signed_type: Optional; Expected type of deserialized signed object.
 
         Raises:
             securesystemslib.exceptions.StorageError: The file cannot be read.
@@ -174,21 +169,19 @@ def from_file(
             storage_backend = FilesystemBackend()
 
         with storage_backend.get(filename) as f:
-            return Metadata.from_bytes(f.read(), deserializer, signed_type)
+            return Metadata.from_bytes(f.read(), deserializer)
 
     @staticmethod
     def from_bytes(
         data: bytes,
         deserializer: Optional[MetadataDeserializer] = None,
-        signed_type: Optional[Type[T]] = None,
     ) -> "Metadata[T]":
         """Loads TUF metadata from raw data.
 
         Arguments:
             data: metadata content as bytes.
             deserializer: Optional; A MetadataDeserializer instance that
                 implements deserialization. Default is JSONDeserializer.
-            signed_type: Optional; Expected type of deserialized signed object.
 
         Raises:
             tuf.api.serialization.DeserializationError:
@@ -207,11 +200,6 @@ def from_bytes(
 
         md = deserializer.deserialize(data)
 
-        # Ensure deserialized signed type matches the requested type
-        if signed_type is not None and signed_type != type(md.signed):
-            raise DeserializationError(
-                f"Expected {signed_type}, got {type(md.signed)}"
-            )
         return md
 
     def to_dict(self) -> Dict[str, Any]:
@@ -364,6 +352,60 @@ def to_dict(self) -> Dict[str, Any]:
         """Serialization helper that returns dict representation of self"""
         raise NotImplementedError
 
+    @classmethod
+    @abc.abstractmethod
+    def metadata_from_bytes(
+        cls,
+        data: bytes,
+        deserializer: Optional[MetadataDeserializer] = None,
+    ) -> Metadata:
+        """Loads a Metadata object from bytes.
+
+        Like Metadata.from_bytes() but also raises DeserializationError if
+        bytes does not contain the correct metadata type."""
+        raise NotImplementedError
+
+    @classmethod
+    @abc.abstractmethod
+    def metadata_from_file(
+        cls,
+        filename: str,
+        deserializer: Optional[MetadataDeserializer] = None,
+        storage_backend: Optional[StorageBackendInterface] = None,
+    ) -> Metadata:
+        """Loads a Metadata object from file.
+
+        Like Metadata.from_file() but also raises DeserializationError if
+        file does not contain the correct metadata type."""
+        raise NotImplementedError
+
+    @classmethod
+    def _metadata_from_bytes(
+        cls, data: bytes, deserializer: Optional[MetadataDeserializer]
+    ) -> Metadata:
+        """Like Metadata.from_bytes() but raises on wrong type"""
+        metadata = Metadata.from_bytes(data, deserializer)
+        if not isinstance(metadata.signed, cls):
+            raise DeserializationError(
+                f"Expected {cls}, got {type(metadata.signed)}"
+            )
+        return metadata
+
+    @classmethod
+    def _metadata_from_file(
+        cls,
+        filename: str,
+        deserializer: Optional[MetadataDeserializer],
+        storage_backend: Optional[StorageBackendInterface],
+    ) -> Metadata:
+        """Like Metadata.from_file() but raises on wrong type"""
+        metadata = Metadata.from_file(filename, deserializer, storage_backend)
+        if not isinstance(metadata.signed, cls):
+            raise DeserializationError(
+                f"Expected {cls}, got {type(metadata.signed)}"
+            )
+        return metadata
+
     @classmethod
     @abc.abstractmethod
     def from_dict(cls, signed_dict: Dict[str, Any]) -> "Signed":
@@ -632,6 +674,31 @@ def __init__(
         self.keys = keys
         self.roles = roles
 
+    @classmethod
+    def metadata_from_bytes(
+        cls,
+        data: bytes,
+        deserializer: Optional[MetadataDeserializer] = None,
+    ) -> Metadata["Root"]:
+        """Loads a Metadata[Root] from raw data.
+
+        Like Metadata.from_bytes() but also raises DeserializationError if
+        bytes does not contain root metadata."""
+        return cls._metadata_from_bytes(data, deserializer)
+
+    @classmethod
+    def metadata_from_file(
+        cls,
+        filename: str,
+        deserializer: Optional[MetadataDeserializer] = None,
+        storage_backend: Optional[StorageBackendInterface] = None,
+    ) -> Metadata["Root"]:
+        """Loads a Metadata[Root] from file.
+
+        Like Metadata.from_file() but also raises DeserializationError if file
+        does not contain root metadata."""
+        return cls._metadata_from_file(filename, deserializer, storage_backend)
+
     @classmethod
     def from_dict(cls, signed_dict: Dict[str, Any]) -> "Root":
         """Creates Root object from its dict representation."""
@@ -846,6 +913,31 @@ def from_dict(cls, signed_dict: Dict[str, Any]) -> "Timestamp":
         # All fields left in the timestamp_dict are unrecognized.
         return cls(*common_args, meta, signed_dict)
 
+    @classmethod
+    def metadata_from_bytes(
+        cls,
+        data: bytes,
+        deserializer: Optional[MetadataDeserializer] = None,
+    ) -> Metadata["Timestamp"]:
+        """Loads a Metadata[Timestamp] from raw data.
+
+        Like Metadata.from_bytes() but also raises DeserializationError if
+        bytes does not contain timestamp metadata."""
+        return cls._metadata_from_bytes(data, deserializer)
+
+    @classmethod
+    def metadata_from_file(
+        cls,
+        filename: str,
+        deserializer: Optional[MetadataDeserializer] = None,
+        storage_backend: Optional[StorageBackendInterface] = None,
+    ) -> Metadata["Timestamp"]:
+        """Loads a Metadata[Timestamp] from file.
+
+        Like Metadata.from_file() but also raises DeserializationError if file
+        does not contain timestamp metadata."""
+        return cls._metadata_from_file(filename, deserializer, storage_backend)
+
     def to_dict(self) -> Dict[str, Any]:
         """Returns the dict representation of self."""
         res_dict = self._common_fields_to_dict()
@@ -898,6 +990,31 @@ def from_dict(cls, signed_dict: Dict[str, Any]) -> "Snapshot":
         # All fields left in the snapshot_dict are unrecognized.
         return cls(*common_args, meta, signed_dict)
 
+    @classmethod
+    def metadata_from_bytes(
+        cls,
+        data: bytes,
+        deserializer: Optional[MetadataDeserializer] = None,
+    ) -> Metadata["Snapshot"]:
+        """Loads a Metadata[Snapshot] from raw data.
+
+        Like Metadata.from_bytes() but also raises DeserializationError if
+        bytes does not contain snapshot metadata."""
+        return cls._metadata_from_bytes(data, deserializer)
+
+    @classmethod
+    def metadata_from_file(
+        cls,
+        filename: str,
+        deserializer: Optional[MetadataDeserializer] = None,
+        storage_backend: Optional[StorageBackendInterface] = None,
+    ) -> Metadata["Snapshot"]:
+        """Loads a Metadata[Snapshot] from file.
+
+        Like Metadata.from_file() but also raises DeserializationError if file
+        does not contain snapshot metadata."""
+        return cls._metadata_from_file(filename, deserializer, storage_backend)
+
     def to_dict(self) -> Dict[str, Any]:
         """Returns the dict representation of self."""
         snapshot_dict = self._common_fields_to_dict()
@@ -1161,6 +1278,31 @@ def from_dict(cls, signed_dict: Dict[str, Any]) -> "Targets":
         # All fields left in the targets_dict are unrecognized.
         return cls(*common_args, res_targets, delegations, signed_dict)
 
+    @classmethod
+    def metadata_from_bytes(
+        cls,
+        data: bytes,
+        deserializer: Optional[MetadataDeserializer] = None,
+    ) -> Metadata["Targets"]:
+        """Loads a Metadata[Targets] from raw data.
+
+        Like Metadata.from_bytes() but also raises DeserializationError if
+        bytes does not contain targets metadata."""
+        return cls._metadata_from_bytes(data, deserializer)
+
+    @classmethod
+    def metadata_from_file(
+        cls,
+        filename: str,
+        deserializer: Optional[MetadataDeserializer] = None,
+        storage_backend: Optional[StorageBackendInterface] = None,
+    ) -> Metadata["Targets"]:
+        """Loads a Metadata[Targets] from file.
+
+        Like Metadata.from_file() but also raises DeserializationError if file
+        does not contain targets metadata."""
+        return cls._metadata_from_file(filename, deserializer, storage_backend)
+
     def to_dict(self) -> Dict[str, Any]:
         """Returns the dict representation of self."""
         targets_dict = self._common_fields_to_dict()