feature: import targets

Implements the feature import target `rstuf admin import-targets`.

This feature gives the RSTUF administrator the functionality to
import a large number of existent targets. It helps to roll out and
deploy the RSTUF in existing repositories.

This feature is detailed and explained at these links
- repository-service-tuf/repository-service-tuf#188
- BDD Feature: repository-service-tuf/repository-service-tuf#218

Some changes in the ceremony was done to reuse in the code:

  - The `ceremony._check_server` was converted to
  `helpers.api_client.get_headers`
  - The `ceremony._bootstrap_state` was converted to
  `helpers.api_client.task_status`

Added 100% coverage to `helpsers.api_client`

Signed-off-by: Kairo de Araujo <kdearaujo@vmware.com>

Pipfile

@@ -13,6 +13,8 @@ requests = "*"
 tuf = "==2.0.0"
 dynaconf = {extras = ["ini"], version = "*"}
 isort = "*"
+sqlalchemy = "*"
+psycopg2 = "*"
 
 [dev-packages]
 black = "*"

repository_service_tuf/cli/admin/ceremony.py

@@ -29,8 +29,9 @@
 from repository_service_tuf.helpers.api_client import (
     URL,
     Methods,
-    is_logged,
+    get_headers,
     request_server,
+    task_status,
 )
 from repository_service_tuf.helpers.tuf import (
     RolesKeysInput,
@@ -426,37 +427,6 @@ def _configure_keys(rolename: str, role: RolesKeysInput) -> None:
         key_count += 1
 
 
-def _check_server(settings) -> dict[str, str]:
-    server = settings.get("SERVER")
-    token = settings.get("TOKEN")
-    if server and token:
-        token_access_check = is_logged(server, token)
-        if token_access_check.state is False:
-            raise click.ClickException(
-                f"{str(token_access_check.data)}"
-                "\n\nTry re-login: 'Repository Service for TUF admin login'"
-            )
-
-        expired_admin = token_access_check.data.get("expired")
-        if expired_admin is True:
-            raise click.ClickException(
-                "Token expired. Run 'Repository Service for TUF admin login'"
-            )
-        else:
-            headers = {"Authorization": f"Bearer {token}"}
-            response = request_server(
-                server, URL.bootstrap.value, Methods.get, headers=headers
-            )
-            if response.status_code != 200 and (
-                response.json().get("bootstrap") is True or None
-            ):
-                raise click.ClickException(f"{response.json().get('detail')}")
-    else:
-        raise click.ClickException("Login first. Run 'rstuf-cli admin login'")
-
-    return headers
-
-
 def _bootstrap(server, headers, json_payload) -> Optional[str]:
     task_id = None
     response = request_server(
@@ -599,7 +569,7 @@ def ceremony(context, bootstrap, file, upload, save) -> None:
 
     settings = context.obj["settings"]
     if bootstrap:
-        headers = _check_server(settings)
+        headers = get_headers(settings)
         bs_response = request_server(
             settings.SERVER, URL.bootstrap.value, Methods.get, headers=headers
         )
@@ -775,6 +745,6 @@ def ceremony(context, bootstrap, file, upload, save) -> None:
         if task_id is None:
             raise click.ClickException("task id wasn't received")
 
-        _bootstrap_state(task_id, settings.SERVER, headers)
+        task_status(task_id, settings.SERVER, headers, "Bootstrap status: ")
 
     console.print("\nCeremony done. 🔐 🎉")

repository_service_tuf/cli/admin/import_targets.py

@@ -0,0 +1,168 @@
+import json
+import os
+from datetime import datetime
+from typing import Any, Dict, List
+
+from rich.console import Console
+from sqlalchemy import Connection, MetaData, Table, create_engine
+from sqlalchemy.exc import IntegrityError
+
+from repository_service_tuf.cli import click
+from repository_service_tuf.cli.admin import admin
+from repository_service_tuf.helpers.api_client import (
+    URL,
+    Methods,
+    get_headers,
+    is_logged,
+    request_server,
+    task_status,
+)
+from repository_service_tuf.helpers.tuf import Metadata, SuccinctRoles
+
+console = Console()
+
+
+def _check_csv_files(csv_files: List[str]):
+    not_found_csv_files: List[str] = []
+    for csv_file in csv_files:
+        if not os.path.isfile(csv_file):
+            not_found_csv_files.append(csv_file)
+
+    if len(not_found_csv_files) > 0:
+        raise click.ClickException(
+            f"CSV file(s) not found: {(', ').join(not_found_csv_files)}"
+        )
+
+
+def _parse_csv_data(
+    csv_file: str, succinct_roles: SuccinctRoles
+) -> List[Dict[str, Any]]:
+    rstuf_db_data: List[Dict[str, Any]] = []
+    with open(csv_file, "r") as f:
+        for line in f:
+            rstuf_db_data.append(
+                {
+                    "path": line.split(";")[0],
+                    "info": {
+                        "length": int(line.split(";")[1]),
+                        "hashes": {line.split(";")[2]: line.split(";")[3]},
+                    },
+                    "rolename": succinct_roles.get_role_for_target(
+                        line.split(";")[0]
+                    ),
+                    "published": False,
+                    "action": "ADD",
+                    "last_update": datetime.now(),
+                }
+            )
+
+    return rstuf_db_data
+
+
+def _import_csv_to_rstuf(
+    db_client: Connection,
+    rstuf_table: Table,
+    csv_files: List[str],
+    succinct_roles: SuccinctRoles,
+) -> None:
+    for csv_file in csv_files:
+        console.print(f"Import status: Loading data from {csv_file}")
+        rstuf_db_data = _parse_csv_data(csv_file, succinct_roles)
+        console.print(f"Import status: Importing {csv_file} data")
+        try:
+            db_client.execute(rstuf_table.insert(), rstuf_db_data)
+        except IntegrityError:
+            raise click.ClickException(
+                "Import status: ABORTED due duplicated targets. "
+                "CSV files must to have unique targets (path). "
+                "No data added to RSTUF DB."
+            )
+        console.print(f"Import status: {csv_file} imported")
+
+
+@admin.command()
+@click.option(
+    "-metadata-url",
+    required=True,
+    help="RSTUF Metadata URL i.e.: http://127.0.0.1 .",
+)
+@click.option(
+    "-db-uri",
+    required=True,
+    help="RSTUF DB URI. i.e.: postgresql://postgres:secret@127.0.0.1:5433",
+)
+@click.option(
+    "-csv",
+    required=True,
+    multiple=True,
+    help="CSV file to import. Multiple -csv parameters are allowed.",
+)
+@click.option(
+    "--skip-publish-targets",
+    is_flag=True,
+    help="Skip publish targets process in TUF Metadata.",
+)
+@click.pass_context
+def import_targets(context, metadata_url, db_uri, csv, skip_publish_targets):
+    """
+    Import targets to RSTUF from exported CSV file.
+    """
+    settings = context.obj["settings"]
+    server = settings.get("SERVER")
+    token = settings.get("TOKEN")
+    if server and token:
+        token_access_check = is_logged(server, token)
+        if token_access_check.state is False:
+            raise click.ClickException(
+                f"{str(token_access_check.data)}"
+                "\n\nTry re-login: 'Repository Service for TUF admin login'"
+            )
+    else:
+        raise click.ClickException("Login first. Run 'rstuf admin login'")
+
+    headers = get_headers(settings)
+
+    response = request_server(metadata_url, "1.bin.json", Methods.get)
+    if response.status_code == 404:
+        raise click.ClickException("RSTUF Metadata Targets not found.")
+
+    # load all required infrastructure
+    json_data = json.loads(response.text)
+    targets = Metadata.from_dict(json_data)
+    succinct_roles = targets.signed.delegations.succinct_roles
+    engine = create_engine(f"{db_uri}")
+    db_metadata = MetaData()
+    db_client = engine.connect()
+    rstuf_table = Table("rstuf_targets", db_metadata, autoload_with=engine)
+
+    # validate if the CSV files are accessible
+    _check_csv_files(csv)
+    # import all CSV file(s) data to RSTUF DB without commiting
+    _import_csv_to_rstuf(db_client, rstuf_table, csv, succinct_roles)
+
+    # commit data into RSTUF DB
+    console.print("Import status: Commiting all data to the RSTUF database")
+    db_client.commit()
+    console.print("Import status: All data imported to RSTUF DB")
+
+    if skip_publish_targets:
+        console.print(
+            "Import status: Finshed. "
+            "Not targets published (`--skip-publish-targets`)"
+        )
+    else:
+        console.print("Import status: Submitting action publish targets")
+        publish_targets = request_server(
+            server, URL.publish_targets.value, Methods.post, headers=headers
+        )
+        if publish_targets.status_code != 202:
+            raise click.ClickException(
+                f"Failed to publish targets. {publish_targets.text}"
+            )
+        task_id = publish_targets.json()["data"]["task_id"]
+        console.print(f"Import status: Publish targets task id is {task_id}")
+
+        # monitor task status
+        result = task_status(task_id, server, headers, "Import status: task ")
+        if result is not None:
+            console.print("Import status: [green]Finished.[/]")

repository_service_tuf/helpers/api_client.py

@@ -1,20 +1,25 @@
 # SPDX-FileCopyrightText: 2022-2023 VMware Inc
 #
 # SPDX-License-Identifier: MIT
-
+import time
 from dataclasses import dataclass
 from enum import Enum
 from typing import Any, Dict, Optional
 
 import requests
+from requests.exceptions import ConnectionError
+from rich.console import Console
 
 from repository_service_tuf.cli import click
 
+console = Console()
+
 
 class URL(Enum):
     token = "api/v1/token/"  # nosec bandit: not hard coded password.
     bootstrap = "api/v1/bootstrap/"
     task = "api/v1/task/?task_id="
+    publish_targets = "api/v1/targets/publish/"
 
 
 class Methods(Enum):
@@ -50,7 +55,7 @@ def request_server(
         else:
             raise ValueError("Internal Error. Invalid HTTP/S Method.")
 
-    except requests.exceptions.ConnectionError:
+    except ConnectionError:
         raise click.ClickException(f"Failed to connect to {server}")
 
     return response
@@ -69,6 +74,80 @@ def is_logged(server: str, token: str):
             return Login(state=True, data=data)
 
     else:
-        click.ClickException(
-            f"Error {response.status_code} {response.json()['detail']}"
+        raise click.ClickException(
+            f"Error {response.status_code} {response.text}"
         )
+
+
+def get_headers(settings: Dict[str, str]) -> Dict[str, str]:
+    server = settings.get("SERVER")
+    token = settings.get("TOKEN")
+    if server and token:
+        token_access_check = is_logged(server, token)
+        if token_access_check.state is False:
+            raise click.ClickException(
+                f"{str(token_access_check.data)}"
+                "\n\nTry re-login: 'rstuf admin login'"
+            )
+
+        expired_admin = token_access_check.data.get("expired")
+        if expired_admin is True:
+            raise click.ClickException(
+                "Token expired. Run 'rstuf admin login'"
+            )
+        else:
+            headers = {"Authorization": f"Bearer {token}"}
+            response = request_server(
+                server, URL.bootstrap.value, Methods.get, headers=headers
+            )
+            if response.status_code != 200:
+                raise click.ClickException(
+                    f"Unexpected error: {response.text}"
+                )
+    else:
+        raise click.ClickException("Login first. Run 'rstuf admin login'")
+
+    return headers
+
+
+def task_status(
+    task_id: str, server: str, headers: Dict[str, str], title: Optional[str]
+) -> Dict[Any, str]:
+    received_state = []
+    while True:
+        state_response = request_server(
+            server, f"{URL.task.value}{task_id}", Methods.get, headers=headers
+        )
+
+        if state_response.status_code != 200:
+            raise click.ClickException(
+                f"Unexpected response {state_response.text}"
+            )
+
+        data = state_response.json().get("data")
+
+        if data:
+            if state := data.get("state"):
+                if state not in received_state:
+                    console.print(f"{title}{state}")
+                    received_state.append(state)
+                else:
+                    console.print(".", end="")
+
+                if state == "SUCCESS":
+                    return data
+
+                elif state == "FAILURE":
+                    raise click.ClickException(
+                        f"Failed: {state_response.text}"
+                    )
+
+            else:
+                raise click.ClickException(
+                    f"No state in data received {state_response.text}"
+                )
+        else:
+            raise click.ClickException(
+                f"No data received {state_response.text}"
+            )
+        time.sleep(2)