New API: accept metadata with unrecognized fields

In order to support ADR 0008 we would want to accept unrecognized
fields in all metadata classes, including the classes that would be
added representing a subportion of a role like "meta", "delegations"
and "roles".

Also, we should test that we support unrecognized fields when adding
new classes or modifying existing ones to make sure we support
ADR 0008.

Signed-off-by: Martin Vrachev <mvrachev@vmware.com>

Author: Martin Vrachev

tests/test_api.py

@@ -330,6 +330,30 @@ def test_metadata_targets(self):
         # Verify that data is updated
         self.assertEqual(targets.signed.targets[filename], fileinfo)
 
+    def setup_json_dict_with_unrecognized_field(self, file_path):
+        json_dict = {}
+        with open(file_path) as f:
+            json_dict = json.loads(f.read())
+        # We are changing the json dict without changing the signature.
+        # This could be a problem if we want to do verification on this dict.
+        json_dict["signed"]["foo"] = "bar"
+        return json_dict
+
+    def test_support_for_unrecognized_fields(self):
+        for metadata in ["root", "timestamp", "snapshot", "targets"]:
+            path = os.path.join(self.repo_dir, "metadata", metadata + ".json")
+            json_dict = self.setup_json_dict_with_unrecognized_field(path)
+            # Test that the metadata classes store unrecognized fields when
+            # initializing and passes them when casting the instance to a dict.
+
+            # TODO: Remove the deepcopy when Metadata.from_dict() doesn't have
+            # the side effect to destroy the passed dictionary.
+            temp_copy = copy.deepcopy(json_dict)
+            metadata_obj = Metadata.from_dict(temp_copy)
+
+            self.assertEqual(json_dict["signed"], metadata_obj.signed.to_dict())
+
+
 # Run unit test.
 if __name__ == '__main__':
     utils.configure_test_logging(sys.argv)

tuf/api/metadata.py

@@ -300,9 +300,13 @@ class Signed:
         spec_version: The TUF specification version number (semver) the
             metadata format adheres to.
         expires: The metadata expiration datetime object.
+        known_fields: A dictionary containing the known fields(attributes). Used
+            when filtering the known fields from all fields in order to retrieve
+            and store the unrecognized fields. See ADR 0008 for context.
 
     """
 
+    known_fields = ["_type", "spec_version", "expires", "version"]
     # NOTE: Signed is a stupid name, because this might not be signed yet, but
     # we keep it to match spec terminology (I often refer to this as "payload",
     # or "inner metadata")
@@ -319,6 +323,19 @@ def __init__(
             raise ValueError(f"version must be >= 0, got {version}")
         self.version = version
 
+    @classmethod
+    def _store_unrecognized_fields(
+        cls, all_fields: Mapping[str, Any]
+    ) -> Mapping[str, Any]:
+        """Utility function filtering the known fields from all in order to
+        retrieve and store the unrecognized fields."""
+        unrecognized_fields = {}
+        for key, value in all_fields.items():
+            if key not in cls.known_fields:
+                unrecognized_fields[key] = value
+
+        cls.unrecognized_fields = unrecognized_fields
+
     @staticmethod
     def _common_fields_from_dict(signed_dict: Mapping[str, Any]) -> list:
         """Returns common fields of 'Signed' instances from the passed dict
@@ -423,6 +440,8 @@ def from_dict(cls, root_dict: Mapping[str, Any]) -> "Root":
         consistent_snapshot = root_dict.pop("consistent_snapshot")
         keys = root_dict.pop("keys")
         roles = root_dict.pop("roles")
+        cls.known_fields += ["consistent_snapshot", "keys", "roles"]
+        cls._store_unrecognized_fields(root_dict)
         return cls(*common_args, consistent_snapshot, keys, roles)
 
     def to_dict(self) -> Dict[str, Any]:
@@ -433,6 +452,7 @@ def to_dict(self) -> Dict[str, Any]:
                 "consistent_snapshot": self.consistent_snapshot,
                 "keys": self.keys,
                 "roles": self.roles,
+                **self.unrecognized_fields,
             }
         )
         return root_dict
@@ -495,12 +515,14 @@ def from_dict(cls, timestamp_dict: Mapping[str, Any]) -> "Timestamp":
         """Creates Timestamp object from its dict representation. """
         common_args = cls._common_fields_from_dict(timestamp_dict)
         meta = timestamp_dict.pop("meta")
+        cls.known_fields += ["meta"]
+        cls._store_unrecognized_fields(timestamp_dict)
         return cls(*common_args, meta)
 
     def to_dict(self) -> Dict[str, Any]:
         """Returns the dict representation of self. """
         timestamp_dict = self._common_fields_to_dict()
-        timestamp_dict.update({"meta": self.meta})
+        timestamp_dict.update({"meta": self.meta, **self.unrecognized_fields})
         return timestamp_dict
 
     # Modification.
@@ -559,12 +581,14 @@ def from_dict(cls, snapshot_dict: Mapping[str, Any]) -> "Snapshot":
         """Creates Snapshot object from its dict representation. """
         common_args = cls._common_fields_from_dict(snapshot_dict)
         meta = snapshot_dict.pop("meta")
+        cls.known_fields += ["meta"]
+        cls._store_unrecognized_fields(snapshot_dict)
         return cls(*common_args, meta)
 
     def to_dict(self) -> Dict[str, Any]:
         """Returns the dict representation of self. """
         snapshot_dict = self._common_fields_to_dict()
-        snapshot_dict.update({"meta": self.meta})
+        snapshot_dict.update({"meta": self.meta, **self.unrecognized_fields})
         return snapshot_dict
 
     # Modification.
@@ -664,6 +688,8 @@ def from_dict(cls, targets_dict: Mapping[str, Any]) -> "Targets":
         common_args = cls._common_fields_from_dict(targets_dict)
         targets = targets_dict.pop("targets")
         delegations = targets_dict.pop("delegations")
+        cls.known_fields += ["meta", "targets", "delegations"]
+        cls._store_unrecognized_fields(targets_dict)
         return cls(*common_args, targets, delegations)
 
     def to_dict(self) -> Dict[str, Any]:
@@ -673,6 +699,7 @@ def to_dict(self) -> Dict[str, Any]:
             {
                 "targets": self.targets,
                 "delegations": self.delegations,
+                **self.unrecognized_fields,
             }
         )
         return targets_dict