Support partially signed metadata in metadata_update

Update `_root_metadata_update` subroutine of `metadata_update` task
interface to accept partially signed metadata. If the required threshold
is not met, the passed metadata is written to the "ROOT_SIGNING"
repository setting and the task returns with a "pending signatures"
message

Missing signatures can then be added using the `sign_metadata` task
interface, which also finalizes the metadata update, as soon as the
threshold is reached.

NOTE: Currently, there is no sanity check of signatures below the
threshold.  A useful check might be, that passed metadata has at least 1
initial and only valid signatures, akin to bootstrap. repository-service-tuf#367 will make
this a lot easier.

This change also includes a reordering of the validation routine to check
the version increment prior to signature threshold. Otherwise, a bad
version would only be detected after all signatures have been added.

Signed-off-by: Lukas Puehringer <[email protected]>

repository_service_tuf_worker/repository.py

@@ -1109,19 +1109,19 @@ def _trusted_root_update(
                 f"Expected 'root', got '{new_root.signed.type}'"
             )
 
-        # Verify that new root is signed by trusted root
-        current_root.verify_delegate(Root.type, new_root)
-
-        # Verify that new root is signed by itself
-        new_root.verify_delegate(Root.type, new_root)
-
         # Verify the new root version
         if new_root.signed.version != current_root.signed.version + 1:
             raise BadVersionNumberError(
                 f"Expected root version {current_root.signed.version + 1}"
                 f" instead got version {new_root.signed.version}"
             )
 
+        # Verify that new root is signed by trusted root
+        current_root.verify_delegate(Root.type, new_root)
+
+        # Verify that new root is signed by itself
+        new_root.verify_delegate(Root.type, new_root)
+
     def _root_metadata_update(
         self, new_root: Metadata[Root]
     ) -> Dict[str, Any]:
@@ -1130,9 +1130,26 @@ def _root_metadata_update(
 
         try:
             self._trusted_root_update(current_root, new_root)
+
+        except UnsignedMetadataError:
+            # TODO: Add missing sanity check - new root must have at least 1
+            # and only valid signature - use `get_verification_status` (#367)
+            self.write_repository_settings("ROOT_SIGNING", new_root.to_dict())
+            return self._task_result(
+                TaskName.METADATA_UPDATE,
+                True,
+                {
+                    "message": "Metadata Update Processed",
+                    "role": Root.type,
+                    "update": (
+                        f"Root v{new_root.signed.version} is "
+                        "pending signatures"
+                    ),
+                },
+            )
+
         except (
             ValueError,
-            UnsignedMetadataError,
             TypeError,
             BadVersionNumberError,
             RepositoryError,

tests/unit/tuf_repository_service_worker/test_repository.py

@@ -2513,25 +2513,17 @@ def test__trusted_root_update_bad_version(self, test_repo):
                 version=4,
                 type=repository.Root.type,
             ),
-            verify_delegate=pretend.call_recorder(lambda *a: None),
         )
         fake_old_root_md = pretend.stub(
             signed=pretend.stub(
                 roles={"timestamp": pretend.stub(keyids={"k1": "v1"})},
                 version=1,
             ),
-            verify_delegate=pretend.call_recorder(lambda *a: None),
         )
 
         with pytest.raises(repository.BadVersionNumberError) as err:
             test_repo._trusted_root_update(fake_old_root_md, fake_new_root_md)
         assert "Expected root version 2 instead got version 4" in str(err)
-        assert fake_new_root_md.verify_delegate.calls == [
-            pretend.call(repository.Root.type, fake_new_root_md)
-        ]
-        assert fake_old_root_md.verify_delegate.calls == [
-            pretend.call(repository.Root.type, fake_new_root_md)
-        ]
 
     def test__trusted_root_update_bad_type(self, test_repo):
         fake_new_root_md = pretend.stub(
@@ -2593,6 +2585,53 @@ def test__root_metadata_update(self, test_repo, mocked_datetime):
             pretend.call(fake_new_root_md, repository.Root.type)
         ]
 
+    def test__root_metadata_update_signatures_pending(
+        self, test_repo, mocked_datetime
+    ):
+        fake_datetime = mocked_datetime
+        fake_new_root_md = pretend.stub(
+            signed=pretend.stub(
+                roles={"timestamp": pretend.stub(keyids={"k1": "v1"})},
+                version=2,
+            )
+        )
+        fake_old_root_md = pretend.stub(
+            signed=pretend.stub(
+                roles={"timestamp": pretend.stub(keyids={"k1": "v1"})},
+                version=1,
+            )
+        )
+        test_repo._storage_backend.get = pretend.call_recorder(
+            lambda *a: fake_old_root_md
+        )
+        test_repo._trusted_root_update = pretend.raiser(
+            repository.UnsignedMetadataError()
+        )
+
+        fake_new_root_md.to_dict = pretend.call_recorder(lambda: "fake dict")
+        test_repo.write_repository_settings = pretend.call_recorder(
+            lambda *a: "fake"
+        )
+
+        result = test_repo._root_metadata_update(fake_new_root_md)
+
+        assert result == {
+            "task": "metadata_update",
+            "status": True,
+            "last_update": fake_datetime.now(),
+            "details": {
+                "message": "Metadata Update Processed",
+                "role": "root",
+                "update": "Root v2 is pending signatures",
+            },
+        }
+        assert test_repo._storage_backend.get.calls == [
+            pretend.call(repository.Root.type)
+        ]
+        assert test_repo.write_repository_settings.calls == [
+            pretend.call("ROOT_SIGNING", "fake dict")
+        ]
+
     def test__root_metadata_update_not_trusted(
         self, test_repo, mocked_datetime
     ):