Transition to the newest version of TUF

taf/repository_tool.py

@@ -416,56 +416,6 @@ def get_all_target_files_state(self):
 
         return added_target_files, removed_target_files
 
-    def get_signed_target_files(self):
-        """Return all target files signed by all roles.
-
-        Args:
-        - None
-
-        Returns:
-        - Set of all target paths relative to targets directory
-        """
-        all_roles = self.get_all_targets_roles()
-        return self.get_singed_target_files_of_roles(all_roles)
-
-    def get_singed_target_files_of_roles(self, roles):
-        """Return all target files signed by the specified roles
-
-        Args:
-        - roles whose target files will be returned
-
-        Returns:
-        - Set of paths of target files of a role relative to targets directory
-        """
-        if roles is None:
-            roles = self.get_all_targets_roles()
-        return set(
-            reduce(
-                operator.iconcat,
-                [self._role_obj(role).target_files for role in roles],
-                [],
-            )
-        )
-
-    def get_signed_targets_with_custom_data(self, roles):
-        """Return all target files signed by the specified roles and and their custom data
-        as specified in the metadata files
-
-        Args:
-        - roles whose target files will be returned
-
-        Returns:
-        - A dictionary whose keys are parts of target files relative to the targets directory
-        and values are custom data dictionaries.
-        """
-        if roles is None:
-            roles = self.get_all_targets_roles()
-        target_files = {}
-        for role in roles:
-            roles_targets = self._role_obj(role).target_files
-            for target_file, custom_data in roles_targets.items():
-                target_files.setdefault(target_file, {}).update(custom_data)
-        return target_files
 
     def modify_targets(self, added_data=None, removed_data=None):
         """Creates a target.json file containing a repository's commit for each repository.

taf/tests/tuf/conftest.py

@@ -1,65 +1,60 @@
-from collections import defaultdict
 import json
 import re
+
 import pytest
-from taf.models.types import RolesKeysData
 from taf.tests.test_api.conftest import REPOSITORY_DESCRIPTION_INPUT_DIR
-from taf.tuf.keys import load_public_key_from_file, load_signer_from_file
-from taf.tests.test_repository.test_repo import MetadataRepository
-from taf.models.converter import from_dict
+from taf.tuf.keys import load_signer_from_file
 
 from taf.tests.tuf import TEST_DATA_PATH
 NO_YUBIKEYS_INPUT = REPOSITORY_DESCRIPTION_INPUT_DIR / "no_yubikeys.json"
 WITH_DELEGATIONS = REPOSITORY_DESCRIPTION_INPUT_DIR / "with_delegations_no_yubikeys.json"
 
 
-@pytest.fixture
+
+@pytest.fixture(scope="module")
 def keystore():
     """Create signer from some rsa test key."""
     return TEST_DATA_PATH / "keystores" / "keystore"
 
-@pytest.fixture
+
+@pytest.fixture(scope="module")
 def keystore_delegations():
     """Create signer from some rsa test key."""
     return TEST_DATA_PATH / "keystores" / "keystore_no_delegations"
 
 
-@pytest.fixture
+@pytest.fixture(scope="module")
 def no_yubikeys_input():
     return json.loads(NO_YUBIKEYS_INPUT.read_text())
 
 
-@pytest.fixture
+@pytest.fixture(scope="module")
 def with_delegations_no_yubikeys_input():
     return json.loads(WITH_DELEGATIONS.read_text())
 
 
-@pytest.fixture
+@pytest.fixture(scope="module")
 def signers(keystore):
     return _load_signers_from_keystore(keystore)
 
 
-@pytest.fixture
+@pytest.fixture(scope="module")
 def signers_with_delegations(keystore_delegations):
     return _load_signers_from_keystore(keystore_delegations)
 
 
-@pytest.fixture
-def tuf_repo(tmp_path, signers, no_yubikeys_input):
-    # Create new metadata repository
-    tuf_repo = MetadataRepository(tmp_path)
-    roles_keys_data = from_dict(no_yubikeys_input, RolesKeysData)
-    tuf_repo.create(roles_keys_data, signers)
-    yield tuf_repo
+@pytest.fixture(scope="module")
+def public_keys(signers):
+    return {
+        role_name: [signer.public_key for signer in signers] for role_name, signers in signers.items()
+    }
 
 
-@pytest.fixture
-def tuf_repo_with_delegations(tmp_path, signers_with_delegations, with_delegations_no_yubikeys_input):
-    # Create new metadata repository
-    tuf_repo = MetadataRepository(tmp_path)
-    roles_keys_data = from_dict(with_delegations_no_yubikeys_input, RolesKeysData)
-    tuf_repo.create(roles_keys_data, signers_with_delegations)
-    yield tuf_repo
+@pytest.fixture(scope="module")
+def public_keys_with_delegations(signers_with_delegations):
+    return {
+        role_name: [signer.public_key for signer in signers] for role_name, signers in signers_with_delegations.items()
+    }
 
 
 def _load_signers_from_keystore(keystore):

taf/tests/tuf/test_create_repository.py → taf/tests/tuf/test_create_edit_repo/test_create_repository.py

@@ -3,7 +3,6 @@
 from taf.tests.test_repository.test_repo import MetadataRepository
 from taf.models.converter import from_dict
 from taf.tuf.keys import _get_legacy_keyid
-from tuf.api.metadata import Targets
 
 
 def test_create_without_delegations(tmp_path, signers, no_yubikeys_input):
@@ -96,3 +95,4 @@ def _get_pub_key_ids(role):
     # assert repo cannot be created twice
     with pytest.raises(FileExistsError):
         tuf_repo.create(roles_keys_data, signers_with_delegations)
+

taf/tests/tuf/test_create_edit_repo/test_targets.py

@@ -0,0 +1,70 @@
+# def test_add_target_files(self, tmp_path, test_signers):
+#     """Edit metadata repository.
+
+#     If we edit manually, we need to make sure to create a valid snapshot.
+#     """
+#     # Create new metadata repository
+#     repo = MetadataRepository(tmp_path)
+#     repo.create(test_signers)
+
+#     target_file = TargetFile.from_data("foo.txt", b"foo", ["sha256", "sha512"])
+
+#     # assert add target file and correct version bumps
+#     repo.add_target_files([target_file])
+#     assert repo.targets().targets[target_file.path] == target_file
+#     assert repo.root().version == 1
+#     assert repo.timestamp().version == 2
+#     assert repo.snapshot().version == 2
+#     assert repo.targets().version == 2
+#     assert repo.timestamp().snapshot_meta.version == 2
+#     assert repo.snapshot().meta["root.json"].version == 1
+#     assert repo.snapshot().meta["targets.json"].version == 2
+
+# def test_add_keys(self, tmp_path, test_signers, test_signer2):
+#     repo = MetadataRepository(tmp_path)
+#     repo.create(test_signers)
+
+#     # assert add new root key and version bumps (all but targets)
+#     repo.add_keys([test_signer2], "root")
+#     assert test_signer2.public_key.keyid in repo.root().keys
+#     assert test_signer2.public_key.keyid in repo.root().roles["root"].keyids
+#     assert repo.root().version == 2
+#     assert repo.timestamp().version == 2
+#     assert repo.snapshot().version == 2
+#     assert repo.targets().version == 1
+#     assert repo.timestamp().snapshot_meta.version == 2
+#     assert repo.snapshot().meta["root.json"].version == 2
+#     assert repo.snapshot().meta["targets.json"].version == 1
+
+#     # assert add new timestamp key and version bumps (all but targets)
+#     repo.add_keys([test_signer2], "timestamp")
+#     assert test_signer2.public_key.keyid in repo.root().roles["timestamp"].keyids
+#     assert repo.root().version == 3
+#     assert repo.timestamp().version == 3
+#     assert repo.snapshot().version == 3
+#     assert repo.targets().version == 1
+#     assert repo.timestamp().snapshot_meta.version == 3
+#     assert repo.snapshot().meta["root.json"].version == 3
+#     assert repo.snapshot().meta["targets.json"].version == 1
+
+#     # assert add new snapshot key and version bumps (all but targets)
+#     repo.add_keys([test_signer2], "snapshot")
+#     assert test_signer2.public_key.keyid in repo.root().roles["snapshot"].keyids
+#     assert repo.root().version == 4
+#     assert repo.timestamp().version == 4
+#     assert repo.snapshot().version == 4
+#     assert repo.targets().version == 1
+#     assert repo.timestamp().snapshot_meta.version == 4
+#     assert repo.snapshot().meta["root.json"].version == 4
+#     assert repo.snapshot().meta["targets.json"].version == 1
+
+#     # assert add new targets key and version bumps (all)
+#     repo.add_keys([test_signer2], "targets")
+#     assert test_signer2.public_key.keyid in repo.root().roles["targets"].keyids
+#     assert repo.root().version == 5
+#     assert repo.timestamp().version == 5
+#     assert repo.snapshot().version == 5
+#     assert repo.targets().version == 2
+#     assert repo.timestamp().snapshot_meta.version == 5
+#     assert repo.snapshot().meta["root.json"].version == 5
+#     assert repo.snapshot().meta["targets.json"].version == 2

taf/tests/tuf/test_query_repo/conftest.py

@@ -0,0 +1,37 @@
+
+import shutil
+from taf.tests.conftest import CLIENT_DIR_PATH
+from taf.utils import on_rm_error
+import pytest
+from taf.models.types import RolesKeysData
+from taf.tests.test_repository.test_repo import MetadataRepository
+from taf.models.converter import from_dict
+
+
+@pytest.fixture(scope="module", autouse=True)
+def repo_dir():
+    path = CLIENT_DIR_PATH / "tuf"
+    path.mkdir()
+    yield path
+    shutil.rmtree(path, onerror=on_rm_error)
+
+@pytest.fixture(scope="module")
+def tuf_repo(repo_dir, signers, no_yubikeys_input):
+    # Create new metadata repository
+    path = repo_dir / "repository_without_delegations"
+    path.mkdir()
+    tuf_repo = MetadataRepository(path)
+    roles_keys_data = from_dict(no_yubikeys_input, RolesKeysData)
+    tuf_repo.create(roles_keys_data, signers)
+    yield tuf_repo
+
+
+@pytest.fixture(scope="module")
+def tuf_repo_with_delegations(repo_dir, signers_with_delegations, with_delegations_no_yubikeys_input):
+    # Create new metadata repository
+    path = repo_dir / "repository_with_delegations"
+    path.mkdir()
+    tuf_repo = MetadataRepository(path)
+    roles_keys_data = from_dict(with_delegations_no_yubikeys_input, RolesKeysData)
+    tuf_repo.create(roles_keys_data, signers_with_delegations)
+    yield tuf_repo

taf/tests/tuf/test_metadata_repository.py → taf/tests/tuf/test_query_repo/test_query_repo.py

@@ -102,10 +102,21 @@ def test_signing_roles(tuf_repo_with_delegations):
 def test_get_role_from_target_paths(tuf_repo_with_delegations):
     assert tuf_repo_with_delegations.get_role_from_target_paths(["dir1/file1.txt", "dir1/file2.txt"]) == "delegated_role"
 
-# def test_find_keys_roles(targets_key):
-#     test_group_dir = TEST_DATA_REPOS_PATH / "test-repository-tool/test-delegated-roles-pkcs1v15" / "taf"
-#     tuf_repo = MetadataRepository(test_group_dir)
-#     tuf_repo.find_keys_roles([targets_key])
+def test_find_keys_roles(tuf_repo_with_delegations, public_keys_with_delegations):
+    target_keys = public_keys_with_delegations["targets"]
+    delegated_role_keys = public_keys_with_delegations["delegated_role"]
+    actual = tuf_repo_with_delegations.find_keys_roles(target_keys + delegated_role_keys)
+    assert actual == ["targets", "delegated_role"]
+    actual = tuf_repo_with_delegations.find_keys_roles(target_keys[2:] + delegated_role_keys)
+    assert actual == ["delegated_role"]
+    root_keys = public_keys_with_delegations["root"]
+    actual = tuf_repo_with_delegations.find_keys_roles(root_keys)
+    assert actual == ["root"]
+
+def test_find_associated_roles_of_key(tuf_repo_with_delegations, public_keys_with_delegations):
+    for role in ("root", "targets", "snapshot", "timestamp", "delegated_role", "inner_role"):
+        key = public_keys_with_delegations[role][0]
+        assert tuf_repo_with_delegations.find_associated_roles_of_key(key) == [role]
 
 
 # def test_all_target_files():