refact: reimplement repository_at_revision

taf/api/metadata.py

@@ -65,7 +65,7 @@ def print_expiration_dates(
     expired: Dict, will_expire: Dict, start_date: datetime, interval: Optional[int] = 30
 ) -> None:
     if expired or will_expire:
-        now = datetime.now()
+        now = datetime.now(timezone.utc)
         print(
             f"Given a {interval} day interval from ({start_date.strftime('%Y-%m-%d')}):"
         )
@@ -170,7 +170,7 @@ def update_metadata_expiration_date(
     reraise=True,
 )
 def _update_expiration_date_of_role(
-    auth_repo: TUFRepository,
+    taf_repo: TUFRepository,
     role: str,
     loaded_yubikeys: Dict,
     keystore: str,
@@ -180,7 +180,7 @@ def _update_expiration_date_of_role(
     prompt_for_keys: bool,
 ) -> None:
     keystore_signers, yubikeys = load_signers(
-        auth_repo,
+        taf_repo,
         role,
         loaded_yubikeys=loaded_yubikeys,
         keystore=keystore,
@@ -189,7 +189,7 @@ def _update_expiration_date_of_role(
     )
     # sign with keystore
     if len(keystore_signers):
-        auth_repo.set_metadata_expiration_date(
+        taf_repo.set_metadata_expiration_date(
             role, keystore_signers, start_date=start_date, interval=interval
         )
     if len(yubikeys):  # sign with yubikey

taf/api/repository.py

@@ -1,3 +1,4 @@
+import json
 from logging import ERROR, INFO
 from typing import Optional
 import click
@@ -193,7 +194,7 @@ def taf_status(path: str, library_dir: Optional[str] = None, indent: int = 0) ->
     print(f"{indent_str}Something to commit: {auth_repo.something_to_commit()}")
     print(f"{indent_str}Target Repositories Status:")
     # Call the list_targets function
-    list_targets(path=path)
+    print(json.dumps(list_targets(path=path), indent=1))
 
     # Load dependencies using repositoriesdb.get_auth_repositories
     repositoriesdb.load_dependencies(auth_repo, library_dir=library_dir)

taf/auth_repo.py

@@ -7,11 +7,14 @@
 from collections import defaultdict
 from contextlib import contextmanager
 from pathlib import Path
+from taf.exceptions import GitError, TAFError
 from tuf.api.metadata import Metadata
 from taf.git import GitRepository
-from taf.tuf.repository import METADATA_DIRECTORY_NAME, Repository as TUFRepository, get_role_metadata_path, get_target_path
+from taf.tuf.repository import METADATA_DIRECTORY_NAME, MetadataRepository as TUFRepository, get_role_metadata_path, get_target_path
 from taf.constants import INFO_JSON_PATH
 
+from securesystemslib.exceptions import StorageError
+
 
 class AuthenticationRepository(GitRepository, TUFRepository):
 
@@ -71,6 +74,7 @@ def __init__(
 
         self.conf_directory_root = conf_directory_root_path.resolve()
         self.out_of_band_authentication = out_of_band_authentication
+        self._current_commit = None
 
     # TODO rework conf_dir
 
@@ -147,7 +151,11 @@ def log_prefix(self) -> str:
         return f"Auth repo {self.name}: "
 
     def close(self, role: str, md: Metadata) -> None:
-        super(role, md)
+        if self._current_commit is None:
+            super(role, md)
+        else:
+            raise TAFError("Cannot update metadata file. File read from git")
+
 
     def commit_and_push(
         self,
@@ -262,7 +270,19 @@ def is_commit_authenticated(self, target_name: str, commit: str) -> bool:
 
 
     def open(self, role: str) -> Metadata:
-        return super().open(role)
+        """Read role metadata from disk."""
+        try:
+            role_path = self.metadata_path / f"{role}.json"
+            if self._current_commit is None:
+                return Metadata.from_file(role_path)
+            try:
+                file_content = self.get_file(self._current_commit, Path(METADATA_DIRECTORY_NAME, f"{role}.json"))
+                file_bytes = file_content.encode()
+                return Metadata.from_bytes(file_bytes)
+            except GitError as e:
+               raise StorageError(e)
+        except StorageError:
+            raise TAFError(f"Metadata file {self.metadata_path} does not exist")
 
     @contextmanager
     def repository_at_revision(self, commit: str):
@@ -272,22 +292,9 @@ def repository_at_revision(self, commit: str):
         and metadata files inside it. Deleted the temp directory when no longer
         needed.
         """
-        tuf_repository = self._tuf_repository
-        with tempfile.TemporaryDirectory() as temp_dir:
-            metadata_files = self.list_files_at_revision(
-                commit, METADATA_DIRECTORY_NAME
-            )
-            Path(temp_dir, METADATA_DIRECTORY_NAME).mkdir(parents=True)
-            for file_name in metadata_files:
-                path = Path(temp_dir, METADATA_DIRECTORY_NAME, file_name)
-                with open(path, "w") as f:
-                    data = self.get_json(
-                        commit, f"{METADATA_DIRECTORY_NAME}/{file_name}"
-                    )
-                    json.dump(data, f)
-            self._load_tuf_repository(temp_dir)
-            yield
-            self._tuf_repository = tuf_repository
+        self._current_commit = commit
+        yield
+        self._current_commit = None
 
     def set_last_validated_commit(self, commit: str):
         """

taf/tests/tuf/test_create_edit_repo/test_targets.py

@@ -4,7 +4,7 @@ def test_add_target_files(tuf_repo):
     # assert add target file and correct version bumps
     path1 = "test1.txt"
     tuf_repo.add_target_files_to_role({path1: {"target": "test1"}})
-    assert (tuf_repo._path / "targets" / path1).is_file()
+    assert (tuf_repo.path / "targets" / path1).is_file()
     assert tuf_repo.targets().targets[path1]
     assert tuf_repo.targets().targets[path1].length > 0
     assert len(tuf_repo.targets().targets[path1].hashes) == 2
@@ -20,7 +20,7 @@ def test_add_target_files(tuf_repo):
     path2 = "test2.txt"
     custom =  {"custom_attr": "custom_val"}
     tuf_repo.add_target_files_to_role({path2: {"target": "test2", "custom": custom}})
-    assert (tuf_repo._path / "targets" / path2).is_file()
+    assert (tuf_repo.path / "targets" / path2).is_file()
     assert tuf_repo.targets().targets[path2].length > 0
     assert tuf_repo.targets().targets[path2].custom ==  custom
 
@@ -35,12 +35,12 @@ def test_repo_target_files(tuf_repo):
         }
     )
     for path in (path1, path2):
-        assert (tuf_repo._path / "targets" / path).is_file()
+        assert (tuf_repo.path / "targets" / path).is_file()
         assert tuf_repo.targets().targets[path].length > 0
 
     tuf_repo.modify_targets(added_data=None, removed_data={path1: None})
-    assert not (tuf_repo._path / "targets" / path1).is_file()
-    assert (tuf_repo._path / "targets" / path2).is_file()
+    assert not (tuf_repo.path / "targets" / path1).is_file()
+    assert (tuf_repo.path / "targets" / path2).is_file()
     assert path1 not in tuf_repo.targets().targets
     assert path2 in tuf_repo.targets().targets
 
@@ -56,7 +56,7 @@ def test_repo_target_files_with_delegations(tuf_repo):
         }
     )
     for path in (target_path1, target_path2):
-        assert (tuf_repo._path / "targets" / path).is_file()
+        assert (tuf_repo.path / "targets" / path).is_file()
         assert tuf_repo.targets().targets[path].length > 0
 
     delegated_path1 = "dir1/path1"
@@ -68,7 +68,7 @@ def test_repo_target_files_with_delegations(tuf_repo):
         }
     )
     for path in (delegated_path1, delegated_path2):
-        assert (tuf_repo._path / "targets" / path).is_file()
+        assert (tuf_repo.path / "targets" / path).is_file()
         assert tuf_repo._signed_obj("delegated_role").targets[path].length > 0
 
     path_delegated = "dir2/path2"
@@ -92,7 +92,7 @@ def test_get_all_target_files_state(tuf_repo):
         }
     )
 
-    (tuf_repo._path / "targets" / target_path1).unlink()
+    (tuf_repo.path / "targets" / target_path1).unlink()
 
     delegated_path1 = "dir1/path1"
     delegated_path2 = "dir2/path1"
@@ -102,7 +102,7 @@ def test_get_all_target_files_state(tuf_repo):
         delegated_path2: {"target": "test2"}
         }
     )
-    path = tuf_repo._path / "targets" / delegated_path1
+    path = tuf_repo.path / "targets" / delegated_path1
     path.write_text("Updated content")
 
     actual = tuf_repo.get_all_target_files_state()
@@ -123,9 +123,9 @@ def test_delete_unregistered_target_files(tuf_repo):
         "dir2/path1": {"target": "test2"}
         }
     )
-    new_target1 = tuf_repo._path / "targets" / "new"
+    new_target1 = tuf_repo.path / "targets" / "new"
     new_target1.touch()
-    new_target2 = tuf_repo._path / "targets" / "dir1" / "new"
+    new_target2 = tuf_repo.path / "targets" / "dir1" / "new"
     new_target2.touch()
     assert new_target1.is_file()
     assert new_target2.is_file()

taf/tuf/repository.py

@@ -111,18 +111,18 @@ def certs_dir(self):
 
     def __init__(self, path: Path) -> None:
         self.signer_cache: Dict[str, Dict[str, Signer]] = defaultdict(dict)
-        self._path = path
+        self.path = path
 
         self._snapshot_info = MetaFile(1)
         self._targets_infos: Dict[str, MetaFile] = defaultdict(lambda: MetaFile(1))
 
     @property
     def metadata_path(self) -> Path:
-        return self._path / METADATA_DIRECTORY_NAME
+        return self.path / METADATA_DIRECTORY_NAME
 
     @property
     def targets_path(self):
-        return self._path / TARGETS_DIRECTORY_NAME
+        return self.path / TARGETS_DIRECTORY_NAME
 
     @property
     def targets_infos(self) -> Dict[str, MetaFile]:
@@ -698,10 +698,13 @@ def get_signed_targets_with_custom_data(self, roles: Optional[List[str]]=None) -
         if roles is None:
             roles = self.get_all_targets_roles()
         target_files = {}
-        for role in roles:
-            roles_targets = self.get_targets_of_role(role)
-            for target_path, target_file in roles_targets.items():
-                target_files.setdefault(target_path, {}).update(target_file.custom or {})
+        try:
+            for role in roles:
+                roles_targets = self.get_targets_of_role(role)
+                for target_path, target_file in roles_targets.items():
+                    target_files.setdefault(target_path, {}).update(target_file.custom or {})
+        except StorageError:
+            pass
         return target_files
 
     def get_target_file_custom_data(self, target_path: str) -> Optional[Dict]:
@@ -731,7 +734,7 @@ def get_key_length_and_scheme_from_metadata(self, parent_role, keyid):
         try:
             metadata = json.loads(
                 Path(
-                    self._path, METADATA_DIRECTORY_NAME, f"{parent_role}.json"
+                    self.path, METADATA_DIRECTORY_NAME, f"{parent_role}.json"
                 ).read_text()
             )
             metadata = metadata["signed"]