POC: Use securesystemslib base+json de/serializer and mixin

pyproject.toml

@@ -44,7 +44,7 @@ classifiers = [
 ]
 dependencies = [
   "requests>=2.19.1",
-  "securesystemslib>=0.26.0",
+  "securesystemslib @ git+https://github.com/PradyumnaKrishna/securesystemslib@4c6be46"
 ]
 dynamic = ["version"]
 
@@ -57,6 +57,9 @@ Source = "https://github.com/theupdateframework/python-tuf"
 [tool.hatch.version]
 path = "tuf/__init__.py"
 
+[tool.hatch.metadata]
+allow-direct-references = true
+
 [tool.hatch.build.targets.sdist]
 include = [
   "/docs",

requirements-pinned.txt

@@ -6,5 +6,5 @@ idna==3.4                 # via requests
 pycparser==2.21           # via cffi
 pynacl==1.5.0             # via securesystemslib
 requests==2.28.1
-securesystemslib[crypto,pynacl]==0.26.0
+git+https://github.com/PradyumnaKrishna/securesystemslib@4c6be46
 urllib3==1.26.14           # via requests

tuf/api/metadata.py

@@ -32,7 +32,6 @@
 import fnmatch
 import io
 import logging
-import tempfile
 from datetime import datetime
 from typing import (
     IO,
@@ -53,14 +52,14 @@
 
 from securesystemslib import exceptions as sslib_exceptions
 from securesystemslib import hash as sslib_hash
+from securesystemslib.serialization import JSONSerializable
 from securesystemslib.signer import Key, Signature, Signer
-from securesystemslib.storage import FilesystemBackend, StorageBackendInterface
-from securesystemslib.util import persist_temp_file
 
 from tuf.api.exceptions import LengthOrHashMismatchError, UnsignedMetadataError
 from tuf.api.serialization import (
     MetadataDeserializer,
     MetadataSerializer,
+    SerializationMixin,
     SignedSerializer,
 )
 
@@ -82,7 +81,7 @@
 T = TypeVar("T", "Root", "Timestamp", "Snapshot", "Targets")
 
 
-class Metadata(Generic[T]):
+class Metadata(Generic[T], JSONSerializable, SerializationMixin):
     """A container for signed TUF metadata.
 
     Provides methods to convert to and from dictionary, read and write to and
@@ -200,97 +199,21 @@ def from_dict(cls, metadata: Dict[str, Any]) -> "Metadata[T]":
             unrecognized_fields=metadata,
         )
 
-    @classmethod
-    def from_file(
-        cls,
-        filename: str,
-        deserializer: Optional[MetadataDeserializer] = None,
-        storage_backend: Optional[StorageBackendInterface] = None,
-    ) -> "Metadata[T]":
-        """Load TUF metadata from file storage.
-
-        Args:
-            filename: Path to read the file from.
-            deserializer: ``MetadataDeserializer`` subclass instance that
-                implements the desired wireline format deserialization. Per
-                default a ``JSONDeserializer`` is used.
-            storage_backend: Object that implements
-                ``securesystemslib.storage.StorageBackendInterface``.
-                Default is ``FilesystemBackend`` (i.e. a local file).
-        Raises:
-            StorageError: The file cannot be read.
-            tuf.api.serialization.DeserializationError:
-                The file cannot be deserialized.
-
-        Returns:
-            TUF ``Metadata`` object.
-        """
-
-        if storage_backend is None:
-            storage_backend = FilesystemBackend()
-
-        with storage_backend.get(filename) as file_obj:
-            return cls.from_bytes(file_obj.read(), deserializer)
-
-    @classmethod
-    def from_bytes(
-        cls,
-        data: bytes,
-        deserializer: Optional[MetadataDeserializer] = None,
-    ) -> "Metadata[T]":
-        """Load TUF metadata from raw data.
-
-        Args:
-            data: Metadata content.
-            deserializer: ``MetadataDeserializer`` implementation to use.
-                Default is ``JSONDeserializer``.
-
-        Raises:
-            tuf.api.serialization.DeserializationError:
-                The file cannot be deserialized.
-
-        Returns:
-            TUF ``Metadata`` object.
-        """
-
-        if deserializer is None:
-            # Use local scope import to avoid circular import errors
-            # pylint: disable=import-outside-toplevel
-            from tuf.api.serialization.json import JSONDeserializer
-
-            deserializer = JSONDeserializer()
-
-        return deserializer.deserialize(data)
-
-    def to_bytes(
-        self, serializer: Optional[MetadataSerializer] = None
-    ) -> bytes:
-        """Return the serialized TUF file format as bytes.
-
-        Note that if bytes are first deserialized into ``Metadata`` and then
-        serialized with ``to_bytes()``, the two are not required to be
-        identical even though the signatures are guaranteed to stay valid. If
-        byte-for-byte equivalence is required (which is the case when content
-        hashes are used in other metadata), the original content should be used
-        instead of re-serializing.
-
-        Args:
-            serializer: ``MetadataSerializer`` instance that implements the
-                desired serialization format. Default is ``JSONSerializer``.
-
-        Raises:
-            tuf.api.serialization.SerializationError:
-                The metadata object cannot be serialized.
-        """
+    @staticmethod
+    def _default_deserializer() -> MetadataDeserializer:
+        """Default deserializer for ``Metadata.from_{bytes, file}``."""
+        # pylint: disable=import-outside-toplevel
+        from tuf.api.serialization.json import JSONDeserializer
 
-        if serializer is None:
-            # Use local scope import to avoid circular import errors
-            # pylint: disable=import-outside-toplevel
-            from tuf.api.serialization.json import JSONSerializer
+        return JSONDeserializer()
 
-            serializer = JSONSerializer(compact=True)
+    @staticmethod
+    def _default_serializer() -> MetadataSerializer:
+        """Default serializer for ``Metadata.to_{bytes, file}``."""
+        # pylint: disable=import-outside-toplevel
+        from tuf.api.serialization.json import JSONSerializer
 
-        return serializer.serialize(self)
+        return JSONSerializer(compact=True)
 
     def to_dict(self) -> Dict[str, Any]:
         """Return the dict representation of self."""
@@ -303,40 +226,6 @@ def to_dict(self) -> Dict[str, Any]:
             **self.unrecognized_fields,
         }
 
-    def to_file(
-        self,
-        filename: str,
-        serializer: Optional[MetadataSerializer] = None,
-        storage_backend: Optional[StorageBackendInterface] = None,
-    ) -> None:
-        """Write TUF metadata to file storage.
-
-        Note that if a file is first deserialized into ``Metadata`` and then
-        serialized with ``to_file()``, the two files are not required to be
-        identical even though the signatures are guaranteed to stay valid. If
-        byte-for-byte equivalence is required (which is the case when file
-        hashes are used in other metadata), the original file should be used
-        instead of re-serializing.
-
-        Args:
-            filename: Path to write the file to.
-            serializer: ``MetadataSerializer`` instance that implements the
-                desired serialization format. Default is ``JSONSerializer``.
-            storage_backend: ``StorageBackendInterface`` implementation. Default
-                is ``FilesystemBackend`` (i.e. a local file).
-
-        Raises:
-            tuf.api.serialization.SerializationError:
-                The metadata object cannot be serialized.
-            StorageError: The file cannot be written.
-        """
-
-        bytes_data = self.to_bytes(serializer)
-
-        with tempfile.TemporaryFile() as temp_file:
-            temp_file.write(bytes_data)
-            persist_temp_file(temp_file, filename, storage_backend)
-
     # Signatures.
     def sign(
         self,

tuf/api/serialization/__init__.py

@@ -15,13 +15,15 @@
 """
 
 import abc
-from typing import TYPE_CHECKING
+from typing import TypeAlias
+
+from securesystemslib.serialization import BaseDeserializer, BaseSerializer
 
 from tuf.api.exceptions import RepositoryError
 
-if TYPE_CHECKING:
-    # pylint: disable=cyclic-import
-    from tuf.api.metadata import Metadata, Signed
+MetadataSerializer: TypeAlias = BaseSerializer
+MetadataDeserializer: TypeAlias = BaseDeserializer
+SignedSerializer: TypeAlias = BaseSerializer
 
 
 class SerializationError(RepositoryError):
@@ -30,30 +32,3 @@ class SerializationError(RepositoryError):
 
 class DeserializationError(RepositoryError):
     """Error during deserialization."""
-
-
-class MetadataDeserializer(metaclass=abc.ABCMeta):
-    """Abstract base class for deserialization of Metadata objects."""
-
-    @abc.abstractmethod
-    def deserialize(self, raw_data: bytes) -> "Metadata":
-        """Deserialize bytes to Metadata object."""
-        raise NotImplementedError
-
-
-class MetadataSerializer(metaclass=abc.ABCMeta):
-    """Abstract base class for serialization of Metadata objects."""
-
-    @abc.abstractmethod
-    def serialize(self, metadata_obj: "Metadata") -> bytes:
-        """Serialize Metadata object to bytes."""
-        raise NotImplementedError
-
-
-class SignedSerializer(metaclass=abc.ABCMeta):
-    """Abstract base class for serialization of Signed objects."""
-
-    @abc.abstractmethod
-    def serialize(self, signed_obj: "Signed") -> bytes:
-        """Serialize Signed object to bytes."""
-        raise NotImplementedError

tuf/api/serialization/json.py

@@ -7,11 +7,13 @@
 metadata to the OLPC Canonical JSON format for signature generation and
 verification.
 """
-
-import json
 from typing import Optional
 
 from securesystemslib.formats import encode_canonical
+from securesystemslib.serialization import (
+    JSONDeserializer as BaseJSONDeserializer,
+)
+from securesystemslib.serialization import JSONSerializer as BaseJSONSerializer
 
 # pylint: disable=cyclic-import
 # ... to allow de/serializing Metadata and Signed objects here, while also
@@ -20,20 +22,19 @@
 from tuf.api.metadata import Metadata, Signed
 from tuf.api.serialization import (
     DeserializationError,
-    MetadataDeserializer,
-    MetadataSerializer,
     SerializationError,
     SignedSerializer,
 )
 
 
-class JSONDeserializer(MetadataDeserializer):
+class JSONDeserializer(BaseJSONDeserializer):
     """Provides JSON to Metadata deserialize method."""
 
     def deserialize(self, raw_data: bytes) -> Metadata:
         """Deserialize utf-8 encoded JSON bytes into Metadata object."""
+
         try:
-            json_dict = json.loads(raw_data.decode("utf-8"))
+            json_dict = super().deserialize(raw_data)
             metadata_obj = Metadata.from_dict(json_dict)
 
         except Exception as e:
@@ -42,7 +43,7 @@ def deserialize(self, raw_data: bytes) -> Metadata:
         return metadata_obj
 
 
-class JSONSerializer(MetadataSerializer):
+class JSONSerializer(BaseJSONSerializer):
     """Provides Metadata to JSON serialize method.
 
     Args:
@@ -55,26 +56,19 @@ class JSONSerializer(MetadataSerializer):
     """
 
     def __init__(self, compact: bool = False, validate: Optional[bool] = False):
-        self.compact = compact
+        super().__init__(compact)
         self.validate = validate
 
-    def serialize(self, metadata_obj: Metadata) -> bytes:
+    def serialize(self, obj: Metadata) -> bytes:
         """Serialize Metadata object into utf-8 encoded JSON bytes."""
 
         try:
-            indent = None if self.compact else 1
-            separators = (",", ":") if self.compact else (",", ": ")
-            json_bytes = json.dumps(
-                metadata_obj.to_dict(),
-                indent=indent,
-                separators=separators,
-                sort_keys=True,
-            ).encode("utf-8")
+            json_bytes = BaseJSONSerializer.serialize(self, obj)
 
             if self.validate:
                 try:
                     new_md_obj = JSONDeserializer().deserialize(json_bytes)
-                    if metadata_obj != new_md_obj:
+                    if obj != new_md_obj:
                         raise ValueError(
                             "Metadata changes if you serialize and deserialize."
                         )
@@ -90,12 +84,12 @@ def serialize(self, metadata_obj: Metadata) -> bytes:
 class CanonicalJSONSerializer(SignedSerializer):
     """Provides Signed to OLPC Canonical JSON serialize method."""
 
-    def serialize(self, signed_obj: Signed) -> bytes:
+    def serialize(self, obj: Signed) -> bytes:
         """Serialize Signed object into utf-8 encoded OLPC Canonical JSON
         bytes.
         """
         try:
-            signed_dict = signed_obj.to_dict()
+            signed_dict = obj.to_dict()
             canonical_bytes = encode_canonical(signed_dict).encode("utf-8")
 
         except Exception as e: