Create constants for top-level rolenames

This is a change in the metadata API to remove hardcoded rolenames
and use constants instead.
Adresses theupdateframework#1648

Signed-off-by: Ivana Atanasova <[email protected]>

Author: Ivana Atanasova

tests/repository_simulator.py

@@ -73,6 +73,7 @@
     Timestamp,
 )
 from tuf.api.serialization.json import JSONSerializer
+from tuf.api.metadata import Rolename
 from tuf.exceptions import FetcherHTTPError, RepositoryError
 from tuf.ngclient.fetcher import FetcherInterface
 
@@ -136,7 +137,7 @@ def targets(self) -> Targets:
         return self.md_targets.signed
 
     def all_targets(self) -> Iterator[Tuple[str, Targets]]:
-        yield "targets", self.md_targets.signed
+        yield Rolename.targets, self.md_targets.signed
         for role, md in self.md_delegates.items():
             yield role, md.signed
 
@@ -178,7 +179,7 @@ def _initialize(self):
     def publish_root(self):
         """Sign and store a new serialized version of root"""
         self.md_root.signatures.clear()
-        for signer in self.signers["root"].values():
+        for signer in self.signers[Rolename.root].values():
             self.md_root.sign(signer, append=True)
 
         self.signed_roots.append(self.md_root.to_bytes(JSONSerializer()))
@@ -191,9 +192,9 @@ def fetch(self, url: str) -> Iterator[bytes]:
         if path.startswith("/metadata/") and path.endswith(".json"):
             ver_and_name = path[len("/metadata/") :][: -len(".json")]
             # only consistent_snapshot supported ATM: timestamp is special case
-            if ver_and_name == "timestamp":
+            if ver_and_name == Rolename.timestamp:
                 version = None
-                role = "timestamp"
+                role = Rolename.timestamp
             else:
                 version, _, role = ver_and_name.partition(".")
                 version = int(version)
@@ -230,19 +231,19 @@ def _fetch_metadata(
 
         If version is None, non-versioned metadata is being requested
         """
-        if role == "root":
+        if role == Rolename.root:
             # return a version previously serialized in publish_root()
             if version is None or version > len(self.signed_roots):
                 raise FetcherHTTPError(f"Unknown root version {version}", 404)
             logger.debug("fetched root version %d", role, version)
             return self.signed_roots[version - 1]
         else:
             # sign and serialize the requested metadata
-            if role == "timestamp":
+            if role == Rolename.timestamp:
                 md: Metadata = self.md_timestamp
-            elif role == "snapshot":
+            elif role == Rolename.snapshot:
                 md = self.md_snapshot
-            elif role == "targets":
+            elif role == Rolename.targets:
                 md = self.md_targets
             else:
                 md = self.md_delegates[role]
@@ -275,7 +276,7 @@ def update_timestamp(self):
         self.timestamp.snapshot_meta.version = self.snapshot.version
 
         if self.compute_metafile_hashes_length:
-            hashes, length = self._compute_hashes_and_length("snapshot")
+            hashes, length = self._compute_hashes_and_length(Rolename.snapshot)
             self.timestamp.snapshot_meta.hashes = hashes
             self.timestamp.snapshot_meta.length = length
 
@@ -296,7 +297,7 @@ def update_snapshot(self):
         self.update_timestamp()
 
     def add_target(self, role: str, data: bytes, path: str):
-        if role == "targets":
+        if role == Rolename.targets:
             targets = self.targets
         else:
             targets = self.md_delegates[role].signed
@@ -314,7 +315,7 @@ def add_delegation(
         paths: Optional[List[str]],
         hash_prefixes: Optional[List[str]],
     ):
-        if delegator_name == "targets":
+        if delegator_name == Rolename.targets:
             delegator = self.targets
         else:
             delegator = self.md_delegates[delegator_name].signed
@@ -350,9 +351,9 @@ def write(self):
 
         for ver in range(1, len(self.signed_roots) + 1):
             with open(os.path.join(dir, f"{ver}.root.json"), "wb") as f:
-                f.write(self._fetch_metadata("root", ver))
+                f.write(self._fetch_metadata(Rolename.root, ver))
 
-        for role in ["timestamp", "snapshot", "targets"]:
+        for role in [Rolename.timestamp, Rolename.snapshot, Rolename.targets]:
             with open(os.path.join(dir, f"{role}.json"), "wb") as f:
                 f.write(self._fetch_metadata(role))
 

tuf/api/metadata.py

@@ -61,14 +61,20 @@
     SignedSerializer,
 )
 
+class Rolename:
+    root = "root"
+    timestamp = "timestamp"
+    snapshot = "snapshot"
+    targets = "targets"
+
 # pylint: disable=too-many-lines
 
 logger = logging.getLogger(__name__)
 
 # 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", "19"]
-TOP_LEVEL_ROLE_NAMES = {"root", "timestamp", "snapshot", "targets"}
+TOP_LEVEL_ROLE_NAMES = {Rolename.root, Rolename.timestamp, Rolename.snapshot, Rolename.targets}
 
 # T is a Generic type constraint for Metadata.signed
 T = TypeVar("T", "Root", "Timestamp", "Snapshot", "Targets")
@@ -130,13 +136,13 @@ def from_dict(cls, metadata: Dict[str, Any]) -> "Metadata[T]":
         # Dispatch to contained metadata class on metadata _type field.
         _type = metadata["signed"]["_type"]
 
-        if _type == "targets":
+        if _type == Rolename.targets:
             inner_cls: Type[Signed] = Targets
-        elif _type == "snapshot":
+        elif _type == Rolename.snapshot:
             inner_cls = Snapshot
-        elif _type == "timestamp":
+        elif _type == Rolename.timestamp:
             inner_cls = Timestamp
-        elif _type == "root":
+        elif _type == Rolename.root:
             inner_cls = Root
         else:
             raise ValueError(f'unrecognized metadata type "{_type}"')
@@ -712,7 +718,7 @@ class Root(Signed):
         unrecognized_fields: Dictionary of all unrecognized fields.
     """
 
-    _signed_type = "root"
+    _signed_type = Rolename.root
 
     # TODO: determine an appropriate value for max-args
     # pylint: disable=too-many-arguments
@@ -965,7 +971,7 @@ class Timestamp(Signed):
         snapshot_meta: Meta information for snapshot metadata.
     """
 
-    _signed_type = "timestamp"
+    _signed_type = Rolename.timestamp
 
     def __init__(
         self,
@@ -1015,7 +1021,7 @@ class Snapshot(Signed):
         meta: A dictionary of target metadata filenames to MetaFile objects.
     """
 
-    _signed_type = "snapshot"
+    _signed_type = Rolename.snapshot
 
     def __init__(
         self,
@@ -1402,7 +1408,7 @@ class Targets(Signed):
         unrecognized_fields: Dictionary of all unrecognized fields.
     """
 
-    _signed_type = "targets"
+    _signed_type = Rolename.targets
 
     # TODO: determine an appropriate value for max-args
     # pylint: disable=too-many-arguments
@@ -1423,7 +1429,7 @@ def __init__(
     def from_dict(cls, signed_dict: Dict[str, Any]) -> "Targets":
         """Creates Targets object from its dict representation."""
         common_args = cls._common_fields_from_dict(signed_dict)
-        targets = signed_dict.pop("targets")
+        targets = signed_dict.pop(Rolename.targets)
         try:
             delegations_dict = signed_dict.pop("delegations")
         except KeyError:
@@ -1444,7 +1450,7 @@ def to_dict(self) -> Dict[str, Any]:
         targets = {}
         for target_path, target_file_obj in self.targets.items():
             targets[target_path] = target_file_obj.to_dict()
-        targets_dict["targets"] = targets
+        targets_dict[Rolename.targets] = targets
         if self.delegations is not None:
             targets_dict["delegations"] = self.delegations.to_dict()
         return targets_dict

tuf/ngclient/_internal/trusted_metadata_set.py

@@ -10,7 +10,7 @@
 network IO, which are not handled here.
 
 Loaded metadata can be accessed via index access with rolename as key
-(trusted_set["root"]) or, in the case of top-level metadata, using the helper
+(trusted_set[Rolename.root]) or, in the case of top-level metadata, using the helper
 properties (trusted_set.root).
 
 The rules that TrustedMetadataSet follows for top-level metadata are
@@ -35,7 +35,7 @@
 >>>     trusted_set = TrustedMetadataSet(f.read())
 >>>
 >>> # update root from remote until no more are available
->>> with download("root", trusted_set.root.signed.version + 1) as f:
+>>> with download(Rolename.root, trusted_set.root.signed.version + 1) as f:
 >>>     trusted_set.update_root(f.read())
 >>>
 >>> # load local timestamp, then update from remote
@@ -45,7 +45,7 @@
 >>> except (RepositoryError, OSError):
 >>>     pass # failure to load a local file is ok
 >>>
->>> with download("timestamp") as f:
+>>> with download(Rolename.timestamp) as f:
 >>>     trusted_set.update_timestamp(f.read())
 >>>
 >>> # load local snapshot, then update from remote if needed
@@ -55,7 +55,7 @@
 >>> except (RepositoryError, OSError):
 >>>     # local snapshot is not valid, load from remote
 >>>     # (RepositoryErrors here stop the update)
->>>     with download("snapshot", version) as f:
+>>>     with download(Rolename.snapshot, version) as f:
 >>>         trusted_set.update_snapshot(f.read())
 
 TODO:
@@ -73,7 +73,7 @@
 from typing import Dict, Iterator, Optional
 
 from tuf import exceptions
-from tuf.api.metadata import Metadata, Root, Snapshot, Targets, Timestamp
+from tuf.api.metadata import Metadata, Root, Snapshot, Targets, Timestamp, Rolename
 from tuf.api.serialization import DeserializationError
 
 logger = logging.getLogger(__name__)
@@ -123,22 +123,22 @@ def __iter__(self) -> Iterator[Metadata]:
     @property
     def root(self) -> Metadata[Root]:
         """Current root Metadata"""
-        return self._trusted_set["root"]
+        return self._trusted_set[Rolename.root]
 
     @property
     def timestamp(self) -> Optional[Metadata[Timestamp]]:
         """Current timestamp Metadata or None"""
-        return self._trusted_set.get("timestamp")
+        return self._trusted_set.get(Rolename.timestamp)
 
     @property
     def snapshot(self) -> Optional[Metadata[Snapshot]]:
         """Current snapshot Metadata or None"""
-        return self._trusted_set.get("snapshot")
+        return self._trusted_set.get(Rolename.snapshot)
 
     @property
     def targets(self) -> Optional[Metadata[Targets]]:
         """Current targets Metadata or None"""
-        return self._trusted_set.get("targets")
+        return self._trusted_set.get(Rolename.targets)
 
     # Methods for updating metadata
     def update_root(self, data: bytes) -> None:
@@ -163,23 +163,23 @@ def update_root(self, data: bytes) -> None:
         except DeserializationError as e:
             raise exceptions.RepositoryError("Failed to load root") from e
 
-        if new_root.signed.type != "root":
+        if new_root.signed.type != Rolename.root:
             raise exceptions.RepositoryError(
                 f"Expected 'root', got '{new_root.signed.type}'"
             )
 
         # Verify that new root is signed by trusted root
-        self.root.verify_delegate("root", new_root)
+        self.root.verify_delegate(Rolename.root, new_root)
 
         if new_root.signed.version != self.root.signed.version + 1:
             raise exceptions.ReplayedMetadataError(
-                "root", new_root.signed.version, self.root.signed.version
+                Rolename.root, new_root.signed.version, self.root.signed.version
             )
 
         # Verify that new root is signed by itself
-        new_root.verify_delegate("root", new_root)
+        new_root.verify_delegate(Rolename.root, new_root)
 
-        self._trusted_set["root"] = new_root
+        self._trusted_set[Rolename.root] = new_root
         logger.info("Updated root v%d", new_root.signed.version)
 
     def update_timestamp(self, data: bytes) -> None:
@@ -214,20 +214,20 @@ def update_timestamp(self, data: bytes) -> None:
         except DeserializationError as e:
             raise exceptions.RepositoryError("Failed to load timestamp") from e
 
-        if new_timestamp.signed.type != "timestamp":
+        if new_timestamp.signed.type != Rolename.timestamp:
             raise exceptions.RepositoryError(
                 f"Expected 'timestamp', got '{new_timestamp.signed.type}'"
             )
 
-        self.root.verify_delegate("timestamp", new_timestamp)
+        self.root.verify_delegate(Rolename.timestamp, new_timestamp)
 
         # If an existing trusted timestamp is updated,
         # check for a rollback attack
         if self.timestamp is not None:
             # Prevent rolling back timestamp version
             if new_timestamp.signed.version < self.timestamp.signed.version:
                 raise exceptions.ReplayedMetadataError(
-                    "timestamp",
+                    Rolename.timestamp,
                     new_timestamp.signed.version,
                     self.timestamp.signed.version,
                 )
@@ -237,15 +237,15 @@ def update_timestamp(self, data: bytes) -> None:
                 < self.timestamp.signed.snapshot_meta.version
             ):
                 raise exceptions.ReplayedMetadataError(
-                    "snapshot",
+                    Rolename.snapshot,
                     new_timestamp.signed.snapshot_meta.version,
                     self.timestamp.signed.snapshot_meta.version,
                 )
 
         # expiry not checked to allow old timestamp to be used for rollback
         # protection of new timestamp: expiry is checked in update_snapshot()
 
-        self._trusted_set["timestamp"] = new_timestamp
+        self._trusted_set[Rolename.timestamp] = new_timestamp
         logger.info("Updated timestamp v%d", new_timestamp.signed.version)
 
         # timestamp is loaded: raise if it is not valid _final_ timestamp
@@ -310,12 +310,12 @@ def update_snapshot(
         except DeserializationError as e:
             raise exceptions.RepositoryError("Failed to load snapshot") from e
 
-        if new_snapshot.signed.type != "snapshot":
+        if new_snapshot.signed.type != Rolename.snapshot:
             raise exceptions.RepositoryError(
                 f"Expected 'snapshot', got '{new_snapshot.signed.type}'"
             )
 
-        self.root.verify_delegate("snapshot", new_snapshot)
+        self.root.verify_delegate(Rolename.snapshot, new_snapshot)
 
         # version not checked against meta version to allow old snapshot to be
         # used in rollback protection: it is checked when targets is updated
@@ -341,7 +341,7 @@ def update_snapshot(
         # expiry not checked to allow old snapshot to be used for rollback
         # protection of new snapshot: it is checked when targets is updated
 
-        self._trusted_set["snapshot"] = new_snapshot
+        self._trusted_set[Rolename.snapshot] = new_snapshot
         logger.info("Updated snapshot v%d", new_snapshot.signed.version)
 
         # snapshot is loaded, but we raise if it's not valid _final_ snapshot
@@ -371,7 +371,7 @@ def update_targets(self, data: bytes) -> None:
             RepositoryError: Metadata failed to load or verify. The actual
                 error type and content will contain more details.
         """
-        self.update_delegated_targets(data, "targets", "root")
+        self.update_delegated_targets(data, Rolename.targets, Rolename.root)
 
     def update_delegated_targets(
         self, data: bytes, role_name: str, delegator_name: str
@@ -419,7 +419,7 @@ def update_delegated_targets(
         except DeserializationError as e:
             raise exceptions.RepositoryError("Failed to load snapshot") from e
 
-        if new_delegate.signed.type != "targets":
+        if new_delegate.signed.type != Rolename.targets:
             raise exceptions.RepositoryError(
                 f"Expected 'targets', got '{new_delegate.signed.type}'"
             )
@@ -449,12 +449,12 @@ def _load_trusted_root(self, data: bytes) -> None:
         except DeserializationError as e:
             raise exceptions.RepositoryError("Failed to load root") from e
 
-        if new_root.signed.type != "root":
+        if new_root.signed.type != Rolename.root:
             raise exceptions.RepositoryError(
                 f"Expected 'root', got '{new_root.signed.type}'"
             )
 
-        new_root.verify_delegate("root", new_root)
+        new_root.verify_delegate(Rolename.root, new_root)
 
-        self._trusted_set["root"] = new_root
+        self._trusted_set[Rolename.root] = new_root
         logger.info("Loaded trusted root v%d", new_root.signed.version)