_payload.py

# Copyright the TUF contributors
# SPDX-License-Identifier: MIT OR Apache-2.0


"""Helper classes for low-level Metadata API.

"""
import abc
import fnmatch
import io
import logging
from dataclasses import dataclass
from datetime import datetime
from typing import (
    IO,
    Any,
    ClassVar,
    Dict,
    Iterator,
    List,
    Optional,
    Tuple,
    TypeVar,
    Union,
)

from securesystemslib import exceptions as sslib_exceptions
from securesystemslib import hash as sslib_hash
from securesystemslib.signer import Key, Signature

from tuf.api.exceptions import LengthOrHashMismatchError, UnsignedMetadataError

_ROOT = "root"
_SNAPSHOT = "snapshot"
_TARGETS = "targets"
_TIMESTAMP = "timestamp"

# We aim to support SPECIFICATION_VERSION and require the input metadata
# files to have the same major version (the first number) as ours.
SPECIFICATION_VERSION = ["1", "0", "31"]
TOP_LEVEL_ROLE_NAMES = {_ROOT, _TIMESTAMP, _SNAPSHOT, _TARGETS}

logger = logging.getLogger(__name__)

# T is a Generic type constraint for container payloads
T = TypeVar("T", "Root", "Timestamp", "Snapshot", "Targets")


class Signed(metaclass=abc.ABCMeta):
    """A base class for the signed part of TUF metadata.

    Objects with base class Signed are usually included in a ``Metadata`` object
    on the signed attribute. This class provides attributes and methods that
    are common for all TUF metadata types (roles).

    *All parameters named below are not just constructor arguments but also
    instance attributes.*

    Args:
        version: Metadata version number. If None, then 1 is assigned.
        spec_version: Supported TUF specification version. If None, then the
            version currently supported by the library is assigned.
        expires: Metadata expiry date. If None, then current date and time is
            assigned.
        unrecognized_fields: Dictionary of all attributes that are not managed
            by TUF Metadata API

    Raises:
        ValueError: Invalid arguments.
    """

    # type is required for static reference without changing the API
    type: ClassVar[str] = "signed"

    # _type and type are identical: 1st replicates file format, 2nd passes lint
    @property
    def _type(self) -> str:
        return self.type

    @property
    def expires(self) -> datetime:
        """Get the metadata expiry date.

        # Use 'datetime' module to e.g. expire in seven days from now
        obj.expires = utcnow() + timedelta(days=7)
        """
        return self._expires

    @expires.setter
    def expires(self, value: datetime) -> None:
        self._expires = value.replace(microsecond=0)

    # 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,
        version: Optional[int],
        spec_version: Optional[str],
        expires: Optional[datetime],
        unrecognized_fields: Optional[Dict[str, Any]],
    ):
        if spec_version is None:
            spec_version = ".".join(SPECIFICATION_VERSION)
        # Accept semver (X.Y.Z) but also X.Y for legacy compatibility
        spec_list = spec_version.split(".")
        if len(spec_list) not in [2, 3] or not all(
            el.isdigit() for el in spec_list
        ):
            raise ValueError(f"Failed to parse spec_version {spec_version}")

        # major version must match
        if spec_list[0] != SPECIFICATION_VERSION[0]:
            raise ValueError(f"Unsupported spec_version {spec_version}")

        self.spec_version = spec_version

        self.expires = expires or datetime.utcnow()

        if version is None:
            version = 1
        elif version <= 0:
            raise ValueError(f"version must be > 0, got {version}")
        self.version = version

        if unrecognized_fields is None:
            unrecognized_fields = {}

        self.unrecognized_fields = unrecognized_fields

    def __eq__(self, other: Any) -> bool:
        if not isinstance(other, Signed):
            return False

        return (
            self.type == other.type
            and self.version == other.version
            and self.spec_version == other.spec_version
            and self.expires == other.expires
            and self.unrecognized_fields == other.unrecognized_fields
        )

    @abc.abstractmethod
    def to_dict(self) -> Dict[str, Any]:
        """Serialize and return a dict representation of self."""
        raise NotImplementedError

    @classmethod
    @abc.abstractmethod
    def from_dict(cls, signed_dict: Dict[str, Any]) -> "Signed":
        """Deserialization helper, creates object from json/dict
        representation.
        """
        raise NotImplementedError

    @classmethod
    def _common_fields_from_dict(
        cls, signed_dict: Dict[str, Any]
    ) -> Tuple[int, str, datetime]:
        """Return common fields of ``Signed`` instances from the passed dict
        representation, and returns an ordered list to be passed as leading
        positional arguments to a subclass constructor.

        See ``{Root, Timestamp, Snapshot, Targets}.from_dict``
        methods for usage.

        """
        _type = signed_dict.pop("_type")
        if _type != cls.type:
            raise ValueError(f"Expected type {cls.type}, got {_type}")

        version = signed_dict.pop("version")
        spec_version = signed_dict.pop("spec_version")
        expires_str = signed_dict.pop("expires")
        # Convert 'expires' TUF metadata string to a datetime object, which is
        # what the constructor expects and what we store. The inverse operation
        # is implemented in '_common_fields_to_dict'.
        expires = datetime.strptime(expires_str, "%Y-%m-%dT%H:%M:%SZ")

        return version, spec_version, expires

    def _common_fields_to_dict(self) -> Dict[str, Any]:
        """Return a dict representation of common fields of
        ``Signed`` instances.

        See ``{Root, Timestamp, Snapshot, Targets}.to_dict`` methods for usage.

        """
        return {
            "_type": self._type,
            "version": self.version,
            "spec_version": self.spec_version,
            "expires": self.expires.isoformat() + "Z",
            **self.unrecognized_fields,
        }

    def is_expired(self, reference_time: Optional[datetime] = None) -> bool:
        """Check metadata expiration against a reference time.

        Args:
            reference_time: Time to check expiration date against. A naive
                datetime in UTC expected. Default is current UTC date and time.

        Returns:
            ``True`` if expiration time is less than the reference time.
        """
        if reference_time is None:
            reference_time = datetime.utcnow()

        return reference_time >= self.expires


class Role:
    """Container that defines which keys are required to sign roles metadata.

    Role defines how many keys are required to successfully sign the roles
    metadata, and which keys are accepted.

    *All parameters named below are not just constructor arguments but also
    instance attributes.*

    Args:
        keyids: Roles signing key identifiers.
        threshold: Number of keys required to sign this role's metadata.
        unrecognized_fields: Dictionary of all attributes that are not managed
            by TUF Metadata API

    Raises:
        ValueError: Invalid arguments.
    """

    def __init__(
        self,
        keyids: List[str],
        threshold: int,
        unrecognized_fields: Optional[Dict[str, Any]] = None,
    ):
        if len(set(keyids)) != len(keyids):
            raise ValueError(f"Nonunique keyids: {keyids}")
        if threshold < 1:
            raise ValueError("threshold should be at least 1!")
        self.keyids = keyids
        self.threshold = threshold
        if unrecognized_fields is None:
            unrecognized_fields = {}

        self.unrecognized_fields = unrecognized_fields

    def __eq__(self, other: Any) -> bool:
        if not isinstance(other, Role):
            return False

        return (
            self.keyids == other.keyids
            and self.threshold == other.threshold
            and self.unrecognized_fields == other.unrecognized_fields
        )

    @classmethod
    def from_dict(cls, role_dict: Dict[str, Any]) -> "Role":
        """Create ``Role`` object from its json/dict representation.

        Raises:
            ValueError, KeyError: Invalid arguments.
        """
        keyids = role_dict.pop("keyids")
        threshold = role_dict.pop("threshold")
        # All fields left in the role_dict are unrecognized.
        return cls(keyids, threshold, role_dict)

    def to_dict(self) -> Dict[str, Any]:
        """Return the dictionary representation of self."""
        return {
            "keyids": self.keyids,
            "threshold": self.threshold,
            **self.unrecognized_fields,
        }


@dataclass
class VerificationResult:
    """Signature verification result for delegated role metadata.

    Attributes:
        threshold: Number of required signatures.
        signed: dict of keyid to Key, containing keys that have signed.
        unsigned: dict of keyid to Key, containing keys that have not signed.
    """

    threshold: int
    signed: Dict[str, Key]
    unsigned: Dict[str, Key]

    def __bool__(self) -> bool:
        return self.verified

    @property
    def verified(self) -> bool:
        """True if threshold of signatures is met."""
        return len(self.signed) >= self.threshold

    @property
    def missing(self) -> int:
        """Number of additional signatures required to reach threshold."""
        return max(0, self.threshold - len(self.signed))


@dataclass
class RootVerificationResult:
    """Signature verification result for root metadata.

    Root must be verified by itself and the previous root version. This
    dataclass represents both results. For the edge case of first version
    of root, these underlying results are identical.

    Note that `signed` and `unsigned` correctness requires the underlying
    VerificationResult keys to not conflict (no reusing the same keyid for
    different keys).

    Attributes:
        first: First underlying VerificationResult
        second: Second underlying VerificationResult
    """

    first: VerificationResult
    second: VerificationResult

    def __bool__(self) -> bool:
        return self.verified

    @property
    def verified(self) -> bool:
        """True if threshold of signatures is met in both underlying
        VerificationResults.
        """
        return self.first.verified and self.second.verified

    @property
    def signed(self) -> Dict[str, Key]:
        """Dictionary of all signing keys that have signed, from both
        VerificationResults.
        return a union of all signed (in python<3.9 this requires
        dict unpacking)
        """
        return {**self.first.signed, **self.second.signed}

    @property
    def unsigned(self) -> Dict[str, Key]:
        """Dictionary of all signing keys that have not signed, from both
        VerificationResults.
        return a union of all unsigned (in python<3.9 this requires
        dict unpacking)
        """
        return {**self.first.unsigned, **self.second.unsigned}


class _DelegatorMixin(metaclass=abc.ABCMeta):
    """Class that implements verify_delegate() for Root and Targets"""

    @abc.abstractmethod
    def get_delegated_role(self, delegated_role: str) -> Role:
        """Return the role object for the given delegated role.

        Raises ValueError if delegated_role is not actually delegated.
        """
        raise NotImplementedError

    @abc.abstractmethod
    def get_key(self, keyid: str) -> Key:
        """Return the key object for the given keyid.

        Raises ValueError if key is not found.
        """
        raise NotImplementedError

    def get_verification_result(
        self,
        delegated_role: str,
        payload: bytes,
        signatures: Dict[str, Signature],
    ) -> VerificationResult:
        """Return signature threshold verification result for delegated role.

        NOTE: Unlike `verify_delegate()` this method does not raise, if the
        role metadata is not fully verified.

        Args:
            delegated_role: Name of the delegated role to verify
            payload: Signed payload bytes for the delegated role
            signatures: Signatures over payload bytes

        Raises:
            ValueError: no delegation was found for ``delegated_role``.
        """
        role = self.get_delegated_role(delegated_role)

        signed = {}
        unsigned = {}

        for keyid in role.keyids:
            try:
                key = self.get_key(keyid)
            except ValueError:
                logger.info("No key for keyid %s", keyid)
                continue

            if keyid not in signatures:
                unsigned[keyid] = key
                logger.info("No signature for keyid %s", keyid)
                continue

            sig = signatures[keyid]
            try:
                key.verify_signature(sig, payload)
                signed[keyid] = key
            except sslib_exceptions.UnverifiedSignatureError:
                unsigned[keyid] = key
                logger.info("Key %s failed to verify %s", keyid, delegated_role)

        return VerificationResult(role.threshold, signed, unsigned)

    def verify_delegate(
        self,
        delegated_role: str,
        payload: bytes,
        signatures: Dict[str, Signature],
    ) -> None:
        """Verify signature threshold for delegated role.

        Verify that there are enough valid ``signatures`` over ``payload``, to
        meet the threshold of keys for ``delegated_role``, as defined by the
        delegator (``self``).

        Args:
            delegated_role: Name of the delegated role to verify
            payload: Signed payload bytes for the delegated role
            signatures: Signatures over payload bytes

        Raises:
            UnsignedMetadataError: ``delegated_role`` was not signed with
                required threshold of keys for ``role_name``.
            ValueError: no delegation was found for ``delegated_role``.
        """
        result = self.get_verification_result(
            delegated_role, payload, signatures
        )
        if not result:
            raise UnsignedMetadataError(
                f"{delegated_role} was signed by {len(result.signed)}/"
                f"{result.threshold} keys"
            )


class Root(Signed, _DelegatorMixin):
    """A container for the signed part of root metadata.

    Parameters listed below are also instance attributes.

    Args:
        version: Metadata version number. Default is 1.
        spec_version: Supported TUF specification version. Default is the
            version currently supported by the library.
        expires: Metadata expiry date. Default is current date and time.
        keys: Dictionary of keyids to Keys. Defines the keys used in ``roles``.
            Default is empty dictionary.
        roles: Dictionary of role names to Roles. Defines which keys are
            required to sign the metadata for a specific role. Default is
            a dictionary of top level roles without keys and threshold of 1.
        consistent_snapshot: ``True`` if repository supports consistent
        snapshots. Default is True.
        unrecognized_fields: Dictionary of all attributes that are not managed
            by TUF Metadata API

    Raises:
        ValueError: Invalid arguments.
    """

    type = _ROOT

    def __init__(
        self,
        version: Optional[int] = None,
        spec_version: Optional[str] = None,
        expires: Optional[datetime] = None,
        keys: Optional[Dict[str, Key]] = None,
        roles: Optional[Dict[str, Role]] = None,
        consistent_snapshot: Optional[bool] = True,
        unrecognized_fields: Optional[Dict[str, Any]] = None,
    ):
        super().__init__(version, spec_version, expires, unrecognized_fields)
        self.consistent_snapshot = consistent_snapshot
        self.keys = keys if keys is not None else {}

        if roles is None:
            roles = {r: Role([], 1) for r in TOP_LEVEL_ROLE_NAMES}
        elif set(roles) != TOP_LEVEL_ROLE_NAMES:
            raise ValueError("Role names must be the top-level metadata roles")
        self.roles = roles

    def __eq__(self, other: Any) -> bool:
        if not isinstance(other, Root):
            return False

        return (
            super().__eq__(other)
            and self.keys == other.keys
            and self.roles == other.roles
            and self.consistent_snapshot == other.consistent_snapshot
        )

    @classmethod
    def from_dict(cls, signed_dict: Dict[str, Any]) -> "Root":
        """Create ``Root`` object from its json/dict representation.

        Raises:
            ValueError, KeyError, TypeError: Invalid arguments.
        """
        common_args = cls._common_fields_from_dict(signed_dict)
        consistent_snapshot = signed_dict.pop("consistent_snapshot", None)
        keys = signed_dict.pop("keys")
        roles = signed_dict.pop("roles")

        for keyid, key_dict in keys.items():
            keys[keyid] = Key.from_dict(keyid, key_dict)
        for role_name, role_dict in roles.items():
            roles[role_name] = Role.from_dict(role_dict)

        # All fields left in the signed_dict are unrecognized.
        return cls(*common_args, keys, roles, consistent_snapshot, signed_dict)

    def to_dict(self) -> Dict[str, Any]:
        """Return the dict representation of self."""
        root_dict = self._common_fields_to_dict()
        keys = {keyid: key.to_dict() for (keyid, key) in self.keys.items()}
        roles = {}
        for role_name, role in self.roles.items():
            roles[role_name] = role.to_dict()
        if self.consistent_snapshot is not None:
            root_dict["consistent_snapshot"] = self.consistent_snapshot

        root_dict.update(
            {
                "keys": keys,
                "roles": roles,
            }
        )
        return root_dict

    def add_key(self, key: Key, role: str) -> None:
        """Add new signing key for delegated role ``role``.

        Args:
            key: Signing key to be added for ``role``.
            role: Name of the role, for which ``key`` is added.

        Raises:
            ValueError: If the argument order is wrong or if ``role`` doesn't
                exist.
        """
        # Verify that our users are not using the old argument order.
        if isinstance(role, Key):
            raise ValueError("Role must be a string, not a Key instance")

        if role not in self.roles:
            raise ValueError(f"Role {role} doesn't exist")
        if key.keyid not in self.roles[role].keyids:
            self.roles[role].keyids.append(key.keyid)
        self.keys[key.keyid] = key

    def revoke_key(self, keyid: str, role: str) -> None:
        """Revoke key from ``role`` and updates the key store.

        Args:
            keyid: Identifier of the key to be removed for ``role``.
            role: Name of the role, for which a signing key is removed.

        Raises:
            ValueError: If ``role`` doesn't exist or if ``role`` doesn't include
                the key.
        """
        if role not in self.roles:
            raise ValueError(f"Role {role} doesn't exist")
        if keyid not in self.roles[role].keyids:
            raise ValueError(f"Key with id {keyid} is not used by {role}")
        self.roles[role].keyids.remove(keyid)
        for keyinfo in self.roles.values():
            if keyid in keyinfo.keyids:
                return

        del self.keys[keyid]

    def get_delegated_role(self, delegated_role: str) -> Role:
        """Return the role object for the given delegated role.

        Raises ValueError if delegated_role is not actually delegated.
        """
        if delegated_role not in self.roles:
            raise ValueError(f"Delegated role {delegated_role} not found")

        return self.roles[delegated_role]

    def get_key(self, keyid: str) -> Key:  # noqa: D102
        if keyid not in self.keys:
            raise ValueError(f"Key {keyid} not found")

        return self.keys[keyid]

    def get_root_verification_result(
        self,
        previous: Optional["Root"],
        payload: bytes,
        signatures: Dict[str, Signature],
    ) -> RootVerificationResult:
        """Return signature threshold verification result for two root roles.

        Verify root metadata with two roles (`self` and optionally `previous`).

        If the repository has no root role versions yet, `previous` can be left
        None. In all other cases, `previous` must be the previous version of
        the Root.

        NOTE: Unlike `verify_delegate()` this method does not raise, if the
        root metadata is not fully verified.

        Args:
            previous: The previous `Root` to verify payload with, or None
            payload: Signed payload bytes for root
            signatures: Signatures over payload bytes

        Raises:
            ValueError: no delegation was found for ``root`` or given Root
                versions are not sequential.
        """

        if previous is None:
            previous = self
        elif self.version != previous.version + 1:
            versions = f"v{previous.version} and v{self.version}"
            raise ValueError(
                f"Expected sequential root versions, got {versions}."
            )

        return RootVerificationResult(
            previous.get_verification_result(Root.type, payload, signatures),
            self.get_verification_result(Root.type, payload, signatures),
        )


class BaseFile:
    """A base class of ``MetaFile`` and ``TargetFile``.

    Encapsulates common static methods for length and hash verification.
    """

    @staticmethod
    def _verify_hashes(
        data: Union[bytes, IO[bytes]], expected_hashes: Dict[str, str]
    ) -> None:
        """Verify that the hash of ``data`` matches ``expected_hashes``."""
        is_bytes = isinstance(data, bytes)
        for algo, exp_hash in expected_hashes.items():
            try:
                if is_bytes:
                    digest_object = sslib_hash.digest(algo)
                    digest_object.update(data)
                else:
                    # if data is not bytes, assume it is a file object
                    digest_object = sslib_hash.digest_fileobject(data, algo)
            except (
                sslib_exceptions.UnsupportedAlgorithmError,
                sslib_exceptions.FormatError,
            ) as e:
                raise LengthOrHashMismatchError(
                    f"Unsupported algorithm '{algo}'"
                ) from e

            observed_hash = digest_object.hexdigest()
            if observed_hash != exp_hash:
                raise LengthOrHashMismatchError(
                    f"Observed hash {observed_hash} does not match "
                    f"expected hash {exp_hash}"
                )

    @staticmethod
    def _verify_length(
        data: Union[bytes, IO[bytes]], expected_length: int
    ) -> None:
        """Verify that the length of ``data`` matches ``expected_length``."""
        if isinstance(data, bytes):
            observed_length = len(data)
        else:
            # if data is not bytes, assume it is a file object
            data.seek(0, io.SEEK_END)
            observed_length = data.tell()

        if observed_length != expected_length:
            raise LengthOrHashMismatchError(
                f"Observed length {observed_length} does not match "
                f"expected length {expected_length}"
            )

    @staticmethod
    def _validate_hashes(hashes: Dict[str, str]) -> None:
        if not hashes:
            raise ValueError("Hashes must be a non empty dictionary")
        for key, value in hashes.items():
            if not (isinstance(key, str) and isinstance(value, str)):
                raise TypeError("Hashes items must be strings")

    @staticmethod
    def _validate_length(length: int) -> None:
        if length < 0:
            raise ValueError(f"Length must be >= 0, got {length}")

    @staticmethod
    def _get_length_and_hashes(
        data: Union[bytes, IO[bytes]], hash_algorithms: Optional[List[str]]
    ) -> Tuple[int, Dict[str, str]]:
        """Calculate length and hashes of ``data``."""
        if isinstance(data, bytes):
            length = len(data)
        else:
            data.seek(0, io.SEEK_END)
            length = data.tell()

        hashes = {}

        if hash_algorithms is None:
            hash_algorithms = [sslib_hash.DEFAULT_HASH_ALGORITHM]

        for algorithm in hash_algorithms:
            try:
                if isinstance(data, bytes):
                    digest_object = sslib_hash.digest(algorithm)
                    digest_object.update(data)
                else:
                    digest_object = sslib_hash.digest_fileobject(
                        data, algorithm
                    )
            except (
                sslib_exceptions.UnsupportedAlgorithmError,
                sslib_exceptions.FormatError,
            ) as e:
                raise ValueError(f"Unsupported algorithm '{algorithm}'") from e

            hashes[algorithm] = digest_object.hexdigest()

        return (length, hashes)


class MetaFile(BaseFile):
    """A container with information about a particular metadata file.

    *All parameters named below are not just constructor arguments but also
    instance attributes.*

    Args:
        version: Version of the metadata file.
        length: Length of the metadata file in bytes.
        hashes: Dictionary of hash algorithm names to hashes of the metadata
            file content.
        unrecognized_fields: Dictionary of all attributes that are not managed
            by TUF Metadata API

    Raises:
        ValueError, TypeError: Invalid arguments.
    """

    def __init__(
        self,
        version: int = 1,
        length: Optional[int] = None,
        hashes: Optional[Dict[str, str]] = None,
        unrecognized_fields: Optional[Dict[str, Any]] = None,
    ):
        if version <= 0:
            raise ValueError(f"Metafile version must be > 0, got {version}")
        if length is not None:
            self._validate_length(length)
        if hashes is not None:
            self._validate_hashes(hashes)

        self.version = version
        self.length = length
        self.hashes = hashes
        if unrecognized_fields is None:
            unrecognized_fields = {}

        self.unrecognized_fields = unrecognized_fields

    def __eq__(self, other: Any) -> bool:
        if not isinstance(other, MetaFile):
            return False

        return (
            self.version == other.version
            and self.length == other.length
            and self.hashes == other.hashes
            and self.unrecognized_fields == other.unrecognized_fields
        )

    @classmethod
    def from_dict(cls, meta_dict: Dict[str, Any]) -> "MetaFile":
        """Create ``MetaFile`` object from its json/dict representation.

        Raises:
            ValueError, KeyError: Invalid arguments.
        """
        version = meta_dict.pop("version")
        length = meta_dict.pop("length", None)
        hashes = meta_dict.pop("hashes", None)

        # All fields left in the meta_dict are unrecognized.
        return cls(version, length, hashes, meta_dict)

    @classmethod
    def from_data(
        cls,
        version: int,
        data: Union[bytes, IO[bytes]],
        hash_algorithms: List[str],
    ) -> "MetaFile":
        """Creates MetaFile object from bytes.
        This constructor should only be used if hashes are wanted.
        By default, MetaFile(ver) should be used.
        Args:
            version: Version of the metadata file.
            data: Metadata bytes that the metafile represents.
            hash_algorithms: Hash algorithms to create the hashes with. If not
            specified, the securesystemslib default hash algorithm is used.

        Raises:
            ValueError: The hash algorithms list contains an unsupported
            algorithm.
        """
        length, hashes = cls._get_length_and_hashes(data, hash_algorithms)
        return cls(version, length, hashes)

    def to_dict(self) -> Dict[str, Any]:
        """Return the dictionary representation of self."""
        res_dict: Dict[str, Any] = {
            "version": self.version,
            **self.unrecognized_fields,
        }

        if self.length is not None:
            res_dict["length"] = self.length

        if self.hashes is not None:
            res_dict["hashes"] = self.hashes

        return res_dict

    def verify_length_and_hashes(self, data: Union[bytes, IO[bytes]]) -> None:
        """Verify that the length and hashes of ``data`` match expected values.

        Args:
            data: File object or its content in bytes.

        Raises:
            LengthOrHashMismatchError: Calculated length or hashes do not
                match expected values or hash algorithm is not supported.
        """
        if self.length is not None:
            self._verify_length(data, self.length)

        if self.hashes is not None:
            self._verify_hashes(data, self.hashes)


class Timestamp(Signed):
    """A container for the signed part of timestamp metadata.

    TUF file format uses a dictionary to contain the snapshot information:
    this is not the case with ``Timestamp.snapshot_meta`` which is a
    ``MetaFile``.

    *All parameters named below are not just constructor arguments but also
    instance attributes.*

    Args:
        version: Metadata version number. Default is 1.
        spec_version: Supported TUF specification version. Default is the
            version currently supported by the library.
        expires: Metadata expiry date. Default is current date and time.
        unrecognized_fields: Dictionary of all attributes that are not managed
            by TUF Metadata API
        snapshot_meta: Meta information for snapshot metadata. Default is a
            MetaFile with version 1.

    Raises:
        ValueError: Invalid arguments.
    """

    type = _TIMESTAMP

    def __init__(
        self,
        version: Optional[int] = None,
        spec_version: Optional[str] = None,
        expires: Optional[datetime] = None,
        snapshot_meta: Optional[MetaFile] = None,
        unrecognized_fields: Optional[Dict[str, Any]] = None,
    ):
        super().__init__(version, spec_version, expires, unrecognized_fields)
        self.snapshot_meta = snapshot_meta or MetaFile(1)

    def __eq__(self, other: Any) -> bool:
        if not isinstance(other, Timestamp):
            return False

        return (
            super().__eq__(other) and self.snapshot_meta == other.snapshot_meta
        )

    @classmethod
    def from_dict(cls, signed_dict: Dict[str, Any]) -> "Timestamp":
        """Create ``Timestamp`` object from its json/dict representation.

        Raises:
            ValueError, KeyError: Invalid arguments.
        """
        common_args = cls._common_fields_from_dict(signed_dict)
        meta_dict = signed_dict.pop("meta")
        snapshot_meta = MetaFile.from_dict(meta_dict["snapshot.json"])
        # All fields left in the timestamp_dict are unrecognized.
        return cls(*common_args, snapshot_meta, signed_dict)

    def to_dict(self) -> Dict[str, Any]:
        """Return the dict representation of self."""
        res_dict = self._common_fields_to_dict()
        res_dict["meta"] = {"snapshot.json": self.snapshot_meta.to_dict()}
        return res_dict


class Snapshot(Signed):
    """A container for the signed part of snapshot metadata.

    Snapshot contains information about all target Metadata files.

    *All parameters named below are not just constructor arguments but also
    instance attributes.*

    Args:
        version: Metadata version number. Default is 1.
        spec_version: Supported TUF specification version. Default is the
            version currently supported by the library.
        expires: Metadata expiry date. Default is current date and time.
        unrecognized_fields: Dictionary of all attributes that are not managed
            by TUF Metadata API
        meta: Dictionary of targets filenames to ``MetaFile`` objects. Default
            is a dictionary with a Metafile for "snapshot.json" version 1.

    Raises:
        ValueError: Invalid arguments.
    """

    type = _SNAPSHOT

    def __init__(
        self,
        version: Optional[int] = None,
        spec_version: Optional[str] = None,
        expires: Optional[datetime] = None,
        meta: Optional[Dict[str, MetaFile]] = None,
        unrecognized_fields: Optional[Dict[str, Any]] = None,
    ):
        super().__init__(version, spec_version, expires, unrecognized_fields)
        self.meta = meta if meta is not None else {"targets.json": MetaFile(1)}

    def __eq__(self, other: Any) -> bool:
        if not isinstance(other, Snapshot):
            return False

        return super().__eq__(other) and self.meta == other.meta

    @classmethod
    def from_dict(cls, signed_dict: Dict[str, Any]) -> "Snapshot":
        """Create ``Snapshot`` object from its json/dict representation.

        Raises:
            ValueError, KeyError: Invalid arguments.
        """
        common_args = cls._common_fields_from_dict(signed_dict)
        meta_dicts = signed_dict.pop("meta")
        meta = {}
        for meta_path, meta_dict in meta_dicts.items():
            meta[meta_path] = MetaFile.from_dict(meta_dict)
        # All fields left in the snapshot_dict are unrecognized.
        return cls(*common_args, meta, signed_dict)

    def to_dict(self) -> Dict[str, Any]:
        """Return the dict representation of self."""
        snapshot_dict = self._common_fields_to_dict()
        meta_dict = {}
        for meta_path, meta_info in self.meta.items():
            meta_dict[meta_path] = meta_info.to_dict()

        snapshot_dict["meta"] = meta_dict
        return snapshot_dict