Feature/tuf repositoty

setup.py

@@ -54,9 +54,9 @@
         "cattrs>=23.1.2",
         "click==8.*",
         "colorama>=0.3.9",
-        "oll-tuf==0.20.0.dev2",
-        "cryptography==38.0.*",
-        "securesystemslib==0.25.*",
+        "tuf==5.*",
+        # "cryptography>=40.0.0",
+        "securesystemslib==1.*",
         "loguru==0.7.*",
         'pygit2==1.9.*; python_version < "3.11"',
         'pygit2==1.14.*; python_version >= "3.11"',

taf/api/keystore.py

@@ -4,18 +4,15 @@
 from pathlib import Path
 from taf.models.types import RolesKeysData
 from taf.api.utils._conf import find_taf_directory
-from tuf.repository_tool import (
-    generate_and_write_rsa_keypair,
-    generate_and_write_unencrypted_rsa_keypair,
-)
-from securesystemslib import keys
+
 
 from taf.api.roles import _initialize_roles_and_keystore
 from taf.keys import get_key_name
 from taf.log import taf_logger
 from taf.models.types import RolesIterator
 from taf.models.converter import from_dict
 from taf.exceptions import KeystoreError
+from taf.tuf.keys import generate_and_write_rsa_keypair, generate_rsa_keypair
 
 
 @log_on_start(INFO, "Generating '{key_path:s}'", logger=taf_logger)
@@ -42,12 +39,7 @@ def _generate_rsa_key(key_path: str, password: str, bits: Optional[int] = None)
         None
     """
     try:
-        if password:
-            generate_and_write_rsa_keypair(
-                filepath=key_path, bits=bits, password=password
-            )
-        else:
-            generate_and_write_unencrypted_rsa_keypair(filepath=key_path, bits=bits)
+        generate_and_write_rsa_keypair(path=key_path, key_size=bits, password=password)
         taf_logger.log("NOTICE", f"Generated key {key_path}")
     except Exception:
         taf_logger.error(f"An error occurred while generating rsa key {key_path}")
@@ -80,6 +72,7 @@ def generate_keys(
     Raises:
         KeystoreError if an error occurs while initializing the keystore directory or generating a key
     """
+    # TODO handle scheme
     if keystore is None:
         taf_directory = find_taf_directory(Path())
         if taf_directory:
@@ -104,6 +97,5 @@ def generate_keys(
                     key_path = str(Path(keystore, key_name))
                     _generate_rsa_key(key_path, password, role.length)
                 else:
-                    rsa_key = keys.generate_rsa_key(role.length)
-                    private_key_val = rsa_key["keyval"]["private"]
-                    print(f"{role.name} key:\n\n{private_key_val}\n\n")
+                    rsa_key, _ = generate_rsa_keypair(role.length)
+                    print(f"{role.name} key:\n\n{rsa_key.decode()}\n\n")

taf/api/metadata.py

@@ -1,13 +1,15 @@
-from datetime import datetime
+from datetime import datetime, timezone
 from logging import INFO, ERROR
 from typing import Dict, List, Optional, Tuple
 from logdecorator import log_on_end, log_on_error
+from taf.api.utils._conf import find_keystore
 from taf.api.utils._git import check_if_clean
+from taf.api.utils._repo import manage_repo_and_signers
 from taf.exceptions import TAFError
-from taf.keys import load_signing_keys
+from taf.keys import load_signers
 from taf.constants import DEFAULT_RSA_SIGNATURE_SCHEME
 from taf.messages import git_commit_message
-from taf.repository_tool import Repository, is_delegated_role
+from taf.tuf.repository import MetadataRepository as TUFRepository, is_delegated_role
 from taf.log import taf_logger
 from taf.auth_repo import AuthenticationRepository
 
@@ -42,10 +44,12 @@ def check_expiration_dates(
     Returns:
         None
     """
-    taf_repo = Repository(path)
+    taf_repo = TUFRepository(path)
 
     if start_date is None:
-        start_date = datetime.now()
+        start_date = datetime.now(timezone.utc)
+    elif start_date.tzinfo is None:
+        start_date = start_date.replace(tzinfo=timezone.utc)
 
     expired_dict, will_expire_dict = taf_repo.check_roles_expiration_dates(
         interval, start_date, excluded_roles
@@ -63,7 +67,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')}):"
         )
@@ -118,79 +122,27 @@ def update_metadata_expiration_date(
     Returns:
         None
     """
+
     if start_date is None:
         start_date = datetime.now()
 
-    taf_repo = Repository(path)
-    loaded_yubikeys: Dict = {}
-    roles_to_update = []
-
-    if "root" in roles:
-        roles_to_update.append("root")
-    if "targets" in roles:
-        roles_to_update.append("targets")
-    for role in roles:
-        if is_delegated_role(role):
-            roles_to_update.append(role)
-
-    if len(roles_to_update) or "snapshot" in roles:
-        roles_to_update.append("snapshot")
-    roles_to_update.append("timestamp")
-
-    for role in roles_to_update:
-        _update_expiration_date_of_role(
-            taf_repo,
-            role,
-            loaded_yubikeys,
-            keystore,
-            start_date,
-            interval,
-            scheme,
-            prompt_for_keys,
-        )
+    roles_to_update = set(roles)
+    update_snapshot_and_timestamp = "timestamp" not in roles_to_update or len(roles_to_update) > 1
+    if update_snapshot_and_timestamp:
+        roles_to_update.add("snapshot")
+        roles_to_update.add("timestamp")
 
-    if not commit:
-        print("\nPlease commit manually.\n")
-    else:
-        auth_repo = AuthenticationRepository(path=path)
-        commit_msg = git_commit_message(
-            "update-expiration-dates", roles=",".join(roles)
-        )
-        auth_repo.commit_and_push(commit_msg=commit_msg, push=push)
+    with manage_repo_and_signers(path, roles_to_update, keystore, scheme, prompt_for_keys, load_snapshot_and_timestamp=update_snapshot_and_timestamp) as auth_repo:
+        for role in roles_to_update:
+            auth_repo.set_metadata_expiration_date(
+                role, start_date=start_date, interval=interval
+            )
 
+        if not commit:
+            print("\nPlease commit manually.\n")
+        else:
+            commit_msg = git_commit_message(
+                "update-expiration-dates", roles=",".join(roles)
+            )
+            auth_repo.commit_and_push(commit_msg=commit_msg, push=push)
 
-@log_on_end(INFO, "Updated expiration date of {role:s}", logger=taf_logger)
-@log_on_error(
-    ERROR,
-    "Error: could not update expiration date: {e}",
-    logger=taf_logger,
-    on_exceptions=TAFError,
-    reraise=True,
-)
-def _update_expiration_date_of_role(
-    auth_repo: Repository,
-    role: str,
-    loaded_yubikeys: Dict,
-    keystore: str,
-    start_date: datetime,
-    interval: int,
-    scheme: str,
-    prompt_for_keys: bool,
-) -> None:
-    keys, yubikeys = load_signing_keys(
-        auth_repo,
-        role,
-        loaded_yubikeys=loaded_yubikeys,
-        keystore=keystore,
-        scheme=scheme,
-        prompt_for_keys=prompt_for_keys,
-    )
-    # sign with keystore
-    if len(keys):
-        auth_repo.update_role_keystores(
-            role, keys, start_date=start_date, interval=interval
-        )
-    if len(yubikeys):  # sign with yubikey
-        auth_repo.update_role_yubikeys(
-            role, yubikeys, start_date=start_date, interval=interval
-        )

taf/api/repository.py

@@ -1,8 +1,11 @@
+import json
 from logging import ERROR, INFO
+import shutil
 from typing import Optional
 import click
 from logdecorator import log_on_end, log_on_error, log_on_start
 from taf.api.utils._roles import setup_role
+from taf.git import GitRepository
 from taf.messages import git_commit_message
 from taf.models.types import RolesIterator
 from taf.models.types import RolesKeysData
@@ -20,8 +23,8 @@
 from taf.keys import load_sorted_keys_of_new_roles
 import taf.repositoriesdb as repositoriesdb
 from taf.api.utils._conf import find_keystore
+from taf.tuf.repository import METADATA_DIRECTORY_NAME, MetadataRepository
 from taf.utils import ensure_pre_push_hook
-from tuf.repository_tool import create_new_repository
 from taf.log import taf_logger
 
 
@@ -61,52 +64,42 @@ def create_repository(
     Returns:
         None
     """
-    auth_repo = AuthenticationRepository(path=path)
-    if not _check_if_can_create_repository(auth_repo):
+    # TODO support yubikeys
+
+    if not _check_if_can_create_repository(Path(path)):
         return
 
-    if not keystore and auth_repo.path is not None:
-        keystore_path = find_keystore(auth_repo.path)
+    if not keystore and path is not None:
+        keystore_path = find_keystore(path)
         if keystore_path is not None:
             keystore = str(keystore_path)
     roles_key_infos_dict, keystore, skip_prompt = _initialize_roles_and_keystore(
         roles_key_infos, keystore
     )
 
     roles_keys_data = from_dict(roles_key_infos_dict, RolesKeysData)
-    repository = create_new_repository(
-        str(auth_repo.path), repository_name=auth_repo.name
-    )
-    signing_keys, verification_keys = load_sorted_keys_of_new_roles(
-        auth_repo=auth_repo,
+    auth_repo = AuthenticationRepository(path=path)
+    signers, verification_keys = load_sorted_keys_of_new_roles(
         roles=roles_keys_data.roles,
         yubikeys_data=roles_keys_data.yubikeys,
         keystore=keystore,
         skip_prompt=skip_prompt,
+        certs_dir=auth_repo.certs_dir
     )
-    if signing_keys is None:
+    if signers is None:
         return
 
-    for role in RolesIterator(roles_keys_data.roles, include_delegations=False):
-        setup_role(
-            role,
-            repository,
-            verification_keys[role.name],
-            signing_keys.get(role.name),
-        )
+    repository = MetadataRepository(path)
+    repository.create(roles_keys_data, signers, verification_keys)
 
-    create_delegations(
-        roles_keys_data.roles.targets, repository, verification_keys, signing_keys
-    )
 
     if test:
         test_auth_file = (
             Path(auth_repo.path, auth_repo.targets_path) / auth_repo.TEST_REPO_FLAG_FILE
         )
         test_auth_file.touch()
 
-    auth_repo._tuf_repository = repository
-    updated = register_target_files(
+    register_target_files(
         path,
         keystore,
         roles_key_infos,
@@ -118,8 +111,6 @@ def create_repository(
 
     ensure_pre_push_hook(auth_repo.path)
 
-    if not updated:
-        repository.writeall()
 
     if commit:
         auth_repo.init_repo()
@@ -129,7 +120,7 @@ def create_repository(
         print("\nPlease commit manually.\n")
 
 
-def _check_if_can_create_repository(auth_repo: AuthenticationRepository) -> bool:
+def _check_if_can_create_repository(path: Path) -> bool:
     """
     Check if a new authentication repository can be created at the specified location.
     A repository can be created if there is not directory at the repository's location
@@ -144,11 +135,12 @@ def _check_if_can_create_repository(auth_repo: AuthenticationRepository) -> bool
     Returns:
         True if a new authentication repository can be created, False otherwise.
     """
-    path = Path(auth_repo.path)
+    repo = GitRepository(path=path)
     if path.is_dir():
         # check if there is non-empty metadata directory
-        if auth_repo.metadata_path.is_dir() and any(auth_repo.metadata_path.iterdir()):
-            if auth_repo.is_git_repository:
+        metadata_dir = path / METADATA_DIRECTORY_NAME
+        if metadata_dir.is_dir() and any(metadata_dir.iterdir()):
+            if repo.is_git_repository:
                 print(
                     f'"{path}" is a git repository containing the metadata directory. Generating new metadata files could make the repository invalid. Aborting.'
                 )
@@ -157,6 +149,8 @@ def _check_if_can_create_repository(auth_repo: AuthenticationRepository) -> bool
                 f'Metadata directory found inside "{path}". Recreate metadata files?'
             ):
                 return False
+            else:
+                shutil.rmtree(metadata_dir)
     return True
 
 
@@ -193,7 +187,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)