diff --git a/pyproject.toml b/pyproject.toml index 72fa2a7738..5091c22100 100644 --- a/pyproject.toml +++ b/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", diff --git a/requirements-pinned.txt b/requirements-pinned.txt index 190a020922..f4c07db03b 100644 --- a/requirements-pinned.txt +++ b/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 diff --git a/tuf/api/metadata.py b/tuf/api/metadata.py index 2f40499ce8..f2e525aed1 100644 --- a/tuf/api/metadata.py +++ b/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, diff --git a/tuf/api/serialization/__init__.py b/tuf/api/serialization/__init__.py index 7aef8b9884..cb2244ae8b 100644 --- a/tuf/api/serialization/__init__.py +++ b/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 diff --git a/tuf/api/serialization/json.py b/tuf/api/serialization/json.py index 3355511a66..43333b5732 100644 --- a/tuf/api/serialization/json.py +++ b/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: