Configure unrecognized_fields per instance

It's important to make unrecognized_fields an instance level variable
instead of class-level variable, because two root files for example
could have a different set of unrecognized fields.

Additional changes were necessary to make it work.

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

Author: MVrachev

tests/test_api.py

@@ -330,28 +330,37 @@ 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):
+    def setup_dict_with_unrecognized_field(self, file_path, field, value):
         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"
+        json_dict["signed"][field] = value
         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)
+            dict1 = self.setup_dict_with_unrecognized_field(path, "f", "b")
             # 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)
+            temp_copy = copy.deepcopy(dict1)
             metadata_obj = Metadata.from_dict(temp_copy)
 
-            self.assertEqual(json_dict["signed"], metadata_obj.signed.to_dict())
+            self.assertEqual(dict1["signed"], metadata_obj.signed.to_dict())
+
+            # Test that two instances of the same class could have different
+            # unrecognized fields.
+            dict2 = self.setup_dict_with_unrecognized_field(path, "f2", "b2")
+            temp_copy2 = copy.deepcopy(dict2)
+            metadata_obj2 = Metadata.from_dict(temp_copy2)
+            self.assertNotEqual(
+                metadata_obj.signed.to_dict(), metadata_obj2.signed.to_dict()
+            )
 
 
 # Run unit test.

tuf/api/metadata.py

@@ -300,18 +300,24 @@ 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.
+        unrecognized_fields: An optional dictionary storing all unrecognized
+            fields. Used for backward compatibility.
 
     """
 
+    # This list is used when filtering the known fields from all fields
+    # in order to retrieve and store the unrecognized fields.
     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")
     def __init__(
-        self, _type: str, version: int, spec_version: str, expires: datetime
+        self,
+        _type: str,
+        version: int,
+        spec_version: str,
+        expires: datetime,
+        unrecognized_fields: Optional[Mapping[str, Any]] = {},
     ) -> None:
 
         self._type = _type
@@ -322,19 +328,29 @@ def __init__(
         if version < 0:
             raise ValueError(f"version must be >= 0, got {version}")
         self.version = version
+        self.unrecognized_fields = unrecognized_fields
 
     @classmethod
-    def _store_unrecognized_fields(
+    def _get_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."""
+        get the unrecognized fields."""
+
+        _type = all_fields["_type"]
+        if _type == "root":
+            cls.known_fields += ["consistent_snapshot", "keys", "roles"]
+        elif _type == "snapshot" or _type == "timestamp":
+            cls.known_fields += ["meta"]
+        elif _type == "targets":
+            cls.known_fields += ["targets", "delegations"]
+
         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
+        return unrecognized_fields
 
     @staticmethod
     def _common_fields_from_dict(signed_dict: Mapping[str, Any]) -> list:
@@ -366,6 +382,7 @@ def _common_fields_to_dict(self) -> Dict[str, Any]:
             "version": self.version,
             "spec_version": self.spec_version,
             "expires": self.expires.isoformat() + "Z",
+            **self.unrecognized_fields,
         }
 
     # Modification.
@@ -426,8 +443,11 @@ def __init__(
         consistent_snapshot: bool,
         keys: Mapping[str, Any],
         roles: Mapping[str, Any],
+        unrecognized_fields: Optional[Mapping[str, Any]] = {},
     ) -> None:
-        super().__init__(_type, version, spec_version, expires)
+        super().__init__(
+            _type, version, spec_version, expires, unrecognized_fields
+        )
         # TODO: Add classes for keys and roles
         self.consistent_snapshot = consistent_snapshot
         self.keys = keys
@@ -436,13 +456,14 @@ def __init__(
     @classmethod
     def from_dict(cls, root_dict: Mapping[str, Any]) -> "Root":
         """Creates Root object from its dict representation. """
+        unrecognized_fields = cls._get_unrecognized_fields(root_dict)
         common_args = cls._common_fields_from_dict(root_dict)
         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)
+        return cls(
+            *common_args, consistent_snapshot, keys, roles, unrecognized_fields
+        )
 
     def to_dict(self) -> Dict[str, Any]:
         """Returns the dict representation of self. """
@@ -452,7 +473,6 @@ def to_dict(self) -> Dict[str, Any]:
                 "consistent_snapshot": self.consistent_snapshot,
                 "keys": self.keys,
                 "roles": self.roles,
-                **self.unrecognized_fields,
             }
         )
         return root_dict
@@ -505,24 +525,26 @@ def __init__(
         spec_version: str,
         expires: datetime,
         meta: Mapping[str, Any],
+        unrecognized_fields: Optional[Mapping[str, Any]] = {},
     ) -> None:
-        super().__init__(_type, version, spec_version, expires)
+        super().__init__(
+            _type, version, spec_version, expires, unrecognized_fields
+        )
         # TODO: Add class for meta
         self.meta = meta
 
     @classmethod
     def from_dict(cls, timestamp_dict: Mapping[str, Any]) -> "Timestamp":
         """Creates Timestamp object from its dict representation. """
+        unrecognized_fields = cls._get_unrecognized_fields(timestamp_dict)
         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)
+        return cls(*common_args, meta, unrecognized_fields)
 
     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, **self.unrecognized_fields})
+        timestamp_dict.update({"meta": self.meta})
         return timestamp_dict
 
     # Modification.
@@ -571,24 +593,27 @@ def __init__(
         spec_version: str,
         expires: datetime,
         meta: Mapping[str, Any],
+        unrecognized_fields: Optional[Mapping[str, Any]] = {},
     ) -> None:
-        super().__init__(_type, version, spec_version, expires)
+        super().__init__(
+            _type, version, spec_version, expires, unrecognized_fields
+        )
         # TODO: Add class for meta
         self.meta = meta
 
     @classmethod
     def from_dict(cls, snapshot_dict: Mapping[str, Any]) -> "Snapshot":
         """Creates Snapshot object from its dict representation. """
+        unrecognized_fields = cls._get_unrecognized_fields(snapshot_dict)
         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)
+        return cls(*common_args, meta, unrecognized_fields)
 
     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, **self.unrecognized_fields})
+        snapshot_dict.update({"meta": self.meta})
         return snapshot_dict
 
     # Modification.
@@ -676,21 +701,23 @@ def __init__(
         expires: datetime,
         targets: Mapping[str, Any],
         delegations: Mapping[str, Any],
+        unrecognized_fields: Optional[Mapping[str, Any]] = {},
     ) -> None:
-        super().__init__(_type, version, spec_version, expires)
+        super().__init__(
+            _type, version, spec_version, expires, unrecognized_fields
+        )
         # TODO: Add class for meta
         self.targets = targets
         self.delegations = delegations
 
     @classmethod
     def from_dict(cls, targets_dict: Mapping[str, Any]) -> "Targets":
         """Creates Targets object from its dict representation. """
+        unrecognized_fields = cls._get_unrecognized_fields(targets_dict)
         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)
+        return cls(*common_args, targets, delegations, unrecognized_fields)
 
     def to_dict(self) -> Dict[str, Any]:
         """Returns the dict representation of self. """
@@ -699,7 +726,6 @@ def to_dict(self) -> Dict[str, Any]:
             {
                 "targets": self.targets,
                 "delegations": self.delegations,
-                **self.unrecognized_fields,
             }
         )
         return targets_dict