Metadata API: add dsse support

lukpueh · lukpueh · commit 9beb8a33e675 · 2022-12-22T15:31:19.000+01:00
Add TUF-specific DSSE (`Envelope`) implementation and define
abstract interface (`BaseMetadata`) for common `Envelope` and
`Metadata` operations:
- get_payload() -&gt; Signed
- sign() -&gt; Signature
- verify_delegate() -&gt; None

**Details**

- `Envelope` inherits and calls generic methods from base
  `Envelope` in securesystemslib to sign and verify using the
   DSSE protocol.

- `Envelope` overrides `sign` to add an `append` option, which
  is not available in the base `Envelope`.

- `Envelope` provides a `from_signed` factory method, which
  serializes a `Signed` instance as payload.

- `Envelope.get_payload` takes a `SignedDeserializer` instance to
  deserialize the payload contents (default:
  `SignedJSONDeserializer`).
  `Metadata.get_payload` just returns the already deserialized
  `signed` attribute.

- `Metadata.[sign|verify_delegate]` methods take a
  `SignedSerializer` instance to serialize the payload prior to
  signing/verifying (default: `CanonicalJSONSerializer`).
  `Envelope.[sign|verify_delegate]` just signs/verifies the
  already serialized payload.

- `BaseMetadata` subclasses inherit `[to|from]_[bytes|file]`
  convenience methods from `SerializationMixin`. In turn they must
provide `_default_[de|s]erializer`s to be used by those methods.

- `BaseMetadata` provides default `JSON[Des|S]erializer` for
  both `Envelope` and `Metadata`.

- `JSONSerializer` requires a class to implement a `to_dict`
  method, which is defined by the `JSONSerializable` interface.
 `BaseMetadata` classes are `JSONSerializable`.

- `JSONDeserializer` can deserialize json bytes into both
  `Envelope` and `Metadata`. It case handles based on the presence
  of a certain field.

Signed-off-by: Lukas Puehringer &lt;lukas.puehringer@nyu.edu&gt;
diff --git a/tuf/api/metadata.py b/tuf/api/metadata.py
@@ -51,14 +51,16 @@
 
 from securesystemslib import exceptions as sslib_exceptions
 from securesystemslib import hash as sslib_hash
+from securesystemslib.metadata import Envelope as BaseEnvelope
 from securesystemslib.serialization import JSONSerializable
 from securesystemslib.signer import Key, Signature, Signer
 
 from tuf.api.exceptions import LengthOrHashMismatchError, UnsignedMetadataError
 from tuf.api.serialization import (
-    MetadataDeserializer,
-    MetadataSerializer,
+    BaseDeserializer,
+    BaseSerializer,
     SerializationMixin,
+    SignedDeserializer,
     SignedSerializer,
 )
 
@@ -80,7 +82,138 @@
 T = TypeVar("T", "Root", "Timestamp", "Snapshot", "Targets")
 
 
-class Metadata(Generic[T], JSONSerializable, SerializationMixin):
+class BaseMetadata(SerializationMixin, JSONSerializable, metaclass=abc.ABCMeta):
+    """A common metadata interface for Envelope (DSSE) and Metadata objects."""
+
+    @staticmethod
+    def _default_deserializer() -> BaseDeserializer:
+        """Default deserializer for Serialization Mixin."""
+        # pylint: disable=import-outside-toplevel
+        from tuf.api.serialization.json import JSONDeserializer
+
+        return JSONDeserializer()
+
+    @staticmethod
+    def _default_serializer() -> BaseSerializer:
+        """Default serializer for Serialization Mixin."""
+        # pylint: disable=import-outside-toplevel
+        from tuf.api.serialization.json import JSONSerializer
+
+        return JSONSerializer(compact=True)
+
+    @staticmethod
+    def _get_role_and_keys(
+        signed: "Signed", delegated_role: str
+    ) -> Tuple["Role", Dict[str, Key]]:
+        """Return the keys and role for delegated_role"""
+
+        role: Optional[Role] = None
+        if isinstance(signed, Root):
+            keys = signed.keys
+            role = signed.roles.get(delegated_role)
+        elif isinstance(signed, Targets):
+            if signed.delegations is None:
+                raise ValueError(f"No delegation found for {delegated_role}")
+
+            keys = signed.delegations.keys
+            if signed.delegations.roles is not None:
+                role = signed.delegations.roles.get(delegated_role)
+            elif signed.delegations.succinct_roles is not None:
+                if signed.delegations.succinct_roles.is_delegated_role(
+                    delegated_role
+                ):
+                    role = signed.delegations.succinct_roles
+        else:
+            raise TypeError("Call is valid only on delegator metadata")
+
+        if role is None:
+            raise ValueError(f"No delegation found for {delegated_role}")
+
+        return (role, keys)
+
+    @abc.abstractmethod
+    def get_signed(self) -> "Signed":
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    def sign(
+        self,
+        signer: Signer,
+    ) -> Signature:
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    def verify_delegate(
+        self,
+        delegated_role: str,
+        delegated_metadata: "BaseMetadata",
+    ) -> None:
+        raise NotImplementedError
+
+
+class Envelope(Generic[T], BaseMetadata, BaseEnvelope):
+    """DSSE Envelope for tuf payloads."""
+
+    DEFAULT_PAYLOAD_TYPE = "application/vnd.tuf"
+
+    @classmethod
+    def from_signed(
+        cls, signed: "Signed", serializer: SignedSerializer = None
+    ) -> "Envelope":
+        """Creates DSSE envelope with signed bytes as payload."""
+
+        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
+
+            serializer = JSONSerializer(compact=True)
+
+        return cls(
+            payload=serializer.serialize(signed),
+            payload_type=cls.DEFAULT_PAYLOAD_TYPE,
+            signatures=[],
+        )
+
+    def get_signed(self, deserializer: SignedDeserializer = None) -> "Signed":
+        if deserializer is None:
+            # Use local scope import to avoid circular import errors
+            # pylint: disable=import-outside-toplevel
+            from tuf.api.serialization.json import SignedJSONDeserializer
+
+            deserializer = SignedJSONDeserializer()
+
+        return BaseEnvelope.get_payload(self, deserializer)
+
+    def sign(
+        self,
+        signer: Signer,
+        append: bool = False,
+    ) -> Signature:
+
+        if not append:
+            self.signatures.clear()
+
+        return BaseEnvelope.sign(self, signer)
+
+    def verify_delegate(
+        self,
+        delegated_role: str,
+        delegated_metadata: "BaseMetadata",
+    ) -> None:
+        signed = self.get_signed(None)
+        role, keys = self._get_role_and_keys(signed, delegated_role)
+
+        try:
+            _ = BaseEnvelope.verify(
+                delegated_metadata, keys.values(), role.threshold
+            )
+
+        except sslib_exceptions.UnverifiedSignatureError as e:
+            raise UnsignedMetadataError from e
+
+
+class Metadata(Generic[T], BaseMetadata):
     """A container for signed TUF metadata.
 
     Provides methods to convert to and from dictionary, read and write to and
@@ -149,6 +282,9 @@ def __eq__(self, other: Any) -> bool:
             and self.unrecognized_fields == other.unrecognized_fields
         )
 
+    def get_signed(self) -> "Signed":
+        return self.signed
+
     @classmethod
     def from_dict(cls, metadata: Dict[str, Any]) -> "Metadata[T]":
         """Creates ``Metadata`` object from its json/dict representation.
@@ -198,22 +334,6 @@ def from_dict(cls, metadata: Dict[str, Any]) -> "Metadata[T]":
             unrecognized_fields=metadata,
         )
 
-    @staticmethod
-    def _default_deserializer() -> MetadataDeserializer:
-        """Default Deserializer to be used for deserialization."""
-        # pylint: disable=import-outside-toplevel
-        from tuf.api.serialization.json import JSONDeserializer
-
-        return JSONDeserializer()
-
-    @staticmethod
-    def _default_serializer() -> MetadataSerializer:
-        """Default Serializer to be used for serialization."""
-        # pylint: disable=import-outside-toplevel
-        from tuf.api.serialization.json import JSONSerializer
-
-        return JSONSerializer(compact=True)
-
     def to_dict(self) -> Dict[str, Any]:
         """Returns the dict representation of self."""
 
@@ -275,35 +395,6 @@ def sign(
 
         return signature
 
-    def _get_role_and_keys(
-        self, delegated_role: str
-    ) -> Tuple["Role", Dict[str, Key]]:
-        """Return the keys and role for delegated_role"""
-
-        role: Optional[Role] = None
-        if isinstance(self.signed, Root):
-            keys = self.signed.keys
-            role = self.signed.roles.get(delegated_role)
-        elif isinstance(self.signed, Targets):
-            if self.signed.delegations is None:
-                raise ValueError(f"No delegation found for {delegated_role}")
-
-            keys = self.signed.delegations.keys
-            if self.signed.delegations.roles is not None:
-                role = self.signed.delegations.roles.get(delegated_role)
-            elif self.signed.delegations.succinct_roles is not None:
-                if self.signed.delegations.succinct_roles.is_delegated_role(
-                    delegated_role
-                ):
-                    role = self.signed.delegations.succinct_roles
-        else:
-            raise TypeError("Call is valid only on delegator metadata")
-
-        if role is None:
-            raise ValueError(f"No delegation found for {delegated_role}")
-
-        return (role, keys)
-
     def verify_delegate(
         self,
         delegated_role: str,
@@ -333,14 +424,15 @@ def verify_delegate(
             signed_serializer = CanonicalJSONSerializer()
 
         data = signed_serializer.serialize(delegated_metadata.signed)
-        role, keys = self._get_role_and_keys(delegated_role)
+        role, keys = self._get_role_and_keys(self.signed, delegated_role)
 
         # verify that delegated_metadata is signed by threshold of unique keys
         signing_keys = set()
         for keyid in role.keyids:
             if keyid not in keys:
                 logger.info("No key for keyid %s", keyid)
                 continue
+
             if keyid not in delegated_metadata.signatures:
                 logger.info("No signature for keyid %s", keyid)
                 continue
diff --git a/tuf/api/serialization/__init__.py b/tuf/api/serialization/__init__.py
@@ -32,6 +32,7 @@
 MetadataSerializer: TypeAlias = BaseSerializer
 MetadataDeserializer: TypeAlias = BaseDeserializer
 SignedSerializer: TypeAlias = BaseSerializer
+SignedDeserializer: TypeAlias = BaseDeserializer
 
 
 class SerializationError(RepositoryError):
diff --git a/tuf/api/serialization/json.py b/tuf/api/serialization/json.py
@@ -7,7 +7,7 @@
 metadata to the OLPC Canonical JSON format for signature generation and
 verification.
 """
-from typing import Optional
+from typing import Optional, Type
 
 from securesystemslib.formats import encode_canonical
 from securesystemslib.serialization import (
@@ -19,7 +19,16 @@
 # ... to allow de/serializing Metadata and Signed objects here, while also
 # creating default de/serializers there (see metadata local scope imports).
 # NOTE: A less desirable alternative would be to add more abstraction layers.
-from tuf.api.metadata import Metadata, Signed
+from tuf.api.metadata import (
+    BaseMetadata,
+    Envelope,
+    Metadata,
+    Root,
+    Signed,
+    Snapshot,
+    Targets,
+    Timestamp,
+)
 from tuf.api.serialization import (
     DeserializationError,
     SerializationError,
@@ -28,23 +37,31 @@
 
 
 class JSONDeserializer(BaseJSONDeserializer):
-    """Provides JSON to Metadata deserialize method."""
+    """Provides JSON to ``BaseMetadata`` deserialize method."""
+
+    def deserialize(self, raw_data: bytes) -> BaseMetadata:
+        """Deserialize utf-8 encoded JSON bytes into ``BaseMetadata`` instance.
 
-    def deserialize(self, raw_data: bytes) -> Metadata:
-        """Deserialize utf-8 encoded JSON bytes into Metadata object."""
+        Creates ``Metadata`` or ``Envelope`` instance based on presence of
+        ``payload`` or ``signed`` field."""
 
         try:
             json_dict = super().deserialize(raw_data)
-            metadata_obj = Metadata.from_dict(json_dict)
+
+            if "payload" in json_dict:
+                return Envelope.from_dict(json_dict)
+
+            if "signed" in json_dict:
+                return Metadata.from_dict(json_dict)
+
+            raise ValueError("unrecognized metadata")
 
         except Exception as e:
             raise DeserializationError("Failed to deserialize JSON") from e
 
-        return metadata_obj
-
 
 class JSONSerializer(BaseJSONSerializer):
-    """Provides Metadata to JSON serialize method.
+    """Provides ``BaseMetadata`` to JSON serialize method.
 
     Args:
         compact: A boolean indicating if the JSON bytes generated in
@@ -59,8 +76,8 @@ def __init__(self, compact: bool = False, validate: Optional[bool] = False):
         super().__init__(compact)
         self.validate = validate
 
-    def serialize(self, obj: Metadata) -> bytes:
-        """Serialize Metadata object into utf-8 encoded JSON bytes."""
+    def serialize(self, obj: BaseMetadata) -> bytes:
+        """Serialize ``BaseMetadata`` object into utf-8 encoded JSON bytes."""
 
         try:
             json_bytes = BaseJSONSerializer.serialize(self, obj)
@@ -81,6 +98,36 @@ def serialize(self, obj: Metadata) -> bytes:
         return json_bytes
 
 
+class SignedJSONDeserializer(BaseJSONDeserializer):
+    """Provides JSON to ``Signed`` deserialize method."""
+
+    def deserialize(self, raw_data: bytes) -> Signed:
+        """Deserialize utf-8 encoded JSON bytes into ``Signed`` instance.
+
+        Creates ``Targets``, ``Snapshot``, ``Timestamp`` or ``Root`` instance
+        based on value in ``_type`` field."""
+        try:
+            json_dict = super().deserialize(raw_data)
+
+            _type = json_dict["_type"]
+
+            if _type == Targets.type:
+                _cls: Type[Signed] = Targets
+            elif _type == Snapshot.type:
+                _cls = Snapshot
+            elif _type == Timestamp.type:
+                _cls = Timestamp
+            elif _type == Root.type:
+                _cls = Root
+            else:
+                raise ValueError(f'unrecognized metadata type "{_type}"')
+
+        except Exception as e:
+            raise SerializationError("Failed to serialize JSON") from e
+
+        return _cls.from_dict(json_dict)
+
+
 class CanonicalJSONSerializer(SignedSerializer):
     """Provides Signed to OLPC Canonical JSON serialize method."""
 
