Merge pull request #2321 from jku/repository-annotations

Repository annotations

Author: Lukas Pühringer

examples/repository/_simplerepo.py

@@ -71,7 +71,7 @@ def __init__(self) -> None:
         )
 
         # setup a basic repository, generate signing key per top-level role
-        with self.edit("root") as root:
+        with self.edit_root() as root:
             for role in ["root", "timestamp", "snapshot", "targets"]:
                 key = keys.generate_ed25519_key()
                 self.signer_cache[role].append(SSlibSigner(key))
@@ -127,14 +127,14 @@ def add_target(self, path: str, content: str) -> None:
         self.target_cache[path] = data
 
         # add a target in the targets metadata
-        with self.edit("targets") as targets:
+        with self.edit_targets() as targets:
             targets.targets[path] = TargetFile.from_data(path, data)
 
         logger.debug("Targets v%d", targets.version)
 
         # update snapshot, timestamp
-        self.snapshot()
-        self.timestamp()
+        self.do_snapshot()
+        self.do_timestamp()
 
     def submit_delegation(self, rolename: str, data: bytes) -> bool:
         """Add a delegation to a (offline signed) delegated targets metadata"""
@@ -145,10 +145,11 @@ def submit_delegation(self, rolename: str, data: bytes) -> bool:
 
             # add delegation and key
             role = DelegatedRole(rolename, [], 1, True, [f"{rolename}/*"])
-            with self.edit("targets") as targets:
+            with self.edit_targets() as targets:
                 if targets.delegations is None:
                     targets.delegations = Delegations({}, {})
-
+                if targets.delegations.roles is None:
+                    targets.delegations.roles = {}
                 targets.delegations.roles[rolename] = role
                 targets.add_key(key, rolename)
 
@@ -159,8 +160,8 @@ def submit_delegation(self, rolename: str, data: bytes) -> bool:
         logger.debug("Targets v%d", targets.version)
 
         # update snapshot, timestamp
-        self.snapshot()
-        self.timestamp()
+        self.do_snapshot()
+        self.do_timestamp()
 
         return True
 
@@ -176,15 +177,10 @@ def submit_role(self, role: str, data: bytes) -> bool:
                 if not targetpath.startswith(f"{role}/"):
                     raise ValueError(f"targets allowed under {role}/ only")
 
-            targets_md = self.role_cache["targets"][-1]
+            targets_md = self.open("targets")
             targets_md.verify_delegate(role, md)
-            if role in self.role_cache:
-                current_md = self.role_cache[role][-1]
-                current_ver = current_md.signed.version
-            else:
-                current_ver = 0
 
-            if md.signed.version != current_ver + 1:
+            if md.signed.version != self.targets(role).version + 1:
                 raise ValueError("Invalid version {md.signed.version}")
 
         except (RepositoryError, ValueError) as e:
@@ -201,7 +197,7 @@ def submit_role(self, role: str, data: bytes) -> bool:
             self.target_cache[targetpath] = bytes(f"{targetpath}", "utf-8")
 
         # update snapshot, timestamp
-        self.snapshot()
-        self.timestamp()
+        self.do_snapshot()
+        self.do_timestamp()
 
         return True

examples/uploader/_localrepo.py

@@ -102,7 +102,7 @@ def add_target(self, role: str, targetpath: str) -> bool:
         data = bytes(targetpath, "utf-8")
         targetfile = TargetFile.from_data(targetpath, data)
         try:
-            with self.edit(role) as delegated:
+            with self.edit_targets(role) as delegated:
                 delegated.targets[targetpath] = targetfile
 
         except Exception as e:  # pylint: disable=broad-except

tuf/repository/_repository.py

@@ -9,7 +9,15 @@
 from copy import deepcopy
 from typing import Dict, Generator, Optional, Tuple
 
-from tuf.api.metadata import Metadata, MetaFile, Signed
+from tuf.api.metadata import (
+    Metadata,
+    MetaFile,
+    Root,
+    Signed,
+    Snapshot,
+    Targets,
+    Timestamp,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -57,8 +65,9 @@ def close(self, role: str, md: Metadata) -> None:
     def targets_infos(self) -> Dict[str, MetaFile]:
         """Returns the MetaFiles for current targets metadatas
 
-        This property is used by snapshot() to update Snapshot.meta: Repository
-        implementations should override this property to enable snapshot().
+        This property is used by do_snapshot() to update Snapshot.meta:
+        Repository implementations should override this property to enable
+        do_snapshot().
 
         Note that there is a difference between this return value and
         Snapshot.meta: This dictionary reflects the targets metadata that
@@ -71,9 +80,9 @@ def targets_infos(self) -> Dict[str, MetaFile]:
     def snapshot_info(self) -> MetaFile:
         """Returns the MetaFile for current snapshot metadata
 
-        This property is used by timestamp() to update Timestamp.meta:
+        This property is used by do_timestamp() to update Timestamp.meta:
         Repository implementations should override this property to enable
-        timestamp().
+        do_timestamp().
         """
         raise NotImplementedError
 
@@ -94,7 +103,71 @@ def edit(self, role: str) -> Generator[Signed, None, None]:
             yield md.signed
             self.close(role, md)
 
-    def snapshot(self, force: bool = False) -> Tuple[bool, Dict[str, MetaFile]]:
+    @contextmanager
+    def edit_root(self) -> Generator[Root, None, None]:
+        """Context manager for editing root metadata. See edit()"""
+        with self.edit(Root.type) as root:
+            if not isinstance(root, Root):
+                raise RuntimeError("Unexpected root type")
+            yield root
+
+    @contextmanager
+    def edit_timestamp(self) -> Generator[Timestamp, None, None]:
+        """Context manager for editing timestamp metadata. See edit()"""
+        with self.edit(Timestamp.type) as timestamp:
+            if not isinstance(timestamp, Timestamp):
+                raise RuntimeError("Unexpected timestamp type")
+            yield timestamp
+
+    @contextmanager
+    def edit_snapshot(self) -> Generator[Snapshot, None, None]:
+        """Context manager for editing snapshot metadata. See edit()"""
+        with self.edit(Snapshot.type) as snapshot:
+            if not isinstance(snapshot, Snapshot):
+                raise RuntimeError("Unexpected snapshot type")
+            yield snapshot
+
+    @contextmanager
+    def edit_targets(
+        self, rolename: str = Targets.type
+    ) -> Generator[Targets, None, None]:
+        """Context manager for editing targets metadata. See edit()"""
+        with self.edit(rolename) as targets:
+            if not isinstance(targets, Targets):
+                raise RuntimeError(f"Unexpected targets ({rolename}) type")
+            yield targets
+
+    def root(self) -> Root:
+        """Read current root metadata"""
+        root = self.open(Root.type).signed
+        if not isinstance(root, Root):
+            raise RuntimeError("Unexpected root type")
+        return root
+
+    def timestamp(self) -> Timestamp:
+        """Read current timestamp metadata"""
+        timestamp = self.open(Timestamp.type).signed
+        if not isinstance(timestamp, Timestamp):
+            raise RuntimeError("Unexpected timestamp type")
+        return timestamp
+
+    def snapshot(self) -> Snapshot:
+        """Read current snapshot metadata"""
+        snapshot = self.open(Snapshot.type).signed
+        if not isinstance(snapshot, Snapshot):
+            raise RuntimeError("Unexpected snapshot type")
+        return snapshot
+
+    def targets(self, rolename: str = Targets.type) -> Targets:
+        """Read current targets metadata"""
+        targets = self.open(rolename).signed
+        if not isinstance(targets, Targets):
+            raise RuntimeError("Unexpected targets type")
+        return targets
+
+    def do_snapshot(
+        self, force: bool = False
+    ) -> Tuple[bool, Dict[str, MetaFile]]:
         """Update snapshot meta information
 
         Updates the snapshot meta information according to current targets
@@ -115,7 +188,7 @@ def snapshot(self, force: bool = False) -> Tuple[bool, Dict[str, MetaFile]]:
         update_version = force
         removed: Dict[str, MetaFile] = {}
 
-        with self.edit("snapshot") as snapshot:
+        with self.edit_snapshot() as snapshot:
             for keyname, new_meta in self.targets_infos.items():
                 if keyname not in snapshot.meta:
                     update_version = True
@@ -131,18 +204,20 @@ def snapshot(self, force: bool = False) -> Tuple[bool, Dict[str, MetaFile]]:
                     removed[keyname] = old_meta
 
             if not update_version:
-                # prevent edit() from storing a new snapshot version
+                # prevent edit_snapshot() from storing a new version
                 raise AbortEdit("Skip snapshot: No targets version changes")
 
         if not update_version:
-            # this is reachable as edit() handles AbortEdit
+            # this is reachable as edit_snapshot() handles AbortEdit
             logger.debug("Snapshot update not needed")  # type: ignore[unreachable]
         else:
             logger.debug("Snapshot v%d", snapshot.version)
 
         return update_version, removed
 
-    def timestamp(self, force: bool = False) -> Tuple[bool, Optional[MetaFile]]:
+    def do_timestamp(
+        self, force: bool = False
+    ) -> Tuple[bool, Optional[MetaFile]]:
         """Update timestamp meta information
 
         Updates timestamp according to current snapshot state
@@ -153,7 +228,7 @@ def timestamp(self, force: bool = False) -> Tuple[bool, Optional[MetaFile]]:
         """
         update_version = force
         removed = None
-        with self.edit("timestamp") as timestamp:
+        with self.edit_timestamp() as timestamp:
             if self.snapshot_info.version < timestamp.snapshot_meta.version:
                 raise ValueError("snapshot version rollback")
 
@@ -166,7 +241,7 @@ def timestamp(self, force: bool = False) -> Tuple[bool, Optional[MetaFile]]:
                 raise AbortEdit("Skip timestamp: No snapshot version changes")
 
         if not update_version:
-            # this is reachable as edit() handles AbortEdit
+            # this is reachable as edit_timestamp() handles AbortEdit
             logger.debug("Timestamp update not needed")  # type: ignore[unreachable]
         else:
             logger.debug("Timestamp v%d", timestamp.version)