Generate RSA keys job (#223)

* rsa_keys: save before merge

* rsa_keys: make current tests work

* rsa_keys: add tests

* rsa_keys: use newer uv to have deps cooldown

* rsa_keys: refactor

Author: pawbu

tests/integration/resources/README.md .test.envtests/integration/resources/README.md renamed to .test.env

@@ -1,4 +1,4 @@
-FROM ghcr.io/astral-sh/uv:bookworm-slim as builder
+FROM ghcr.io/astral-sh/uv:trixie-slim as builder
 
 WORKDIR /app
 COPY pyproject.toml uv.lock /app/

Dockerfile

@@ -48,7 +48,7 @@ uv run pytest
 ### Build docker image
 
 ```
-docker build --tag job_executor .
+docker buildx build --tag job-executor:test-local .
 ```
 
 ## Built with

README.md

@@ -16,10 +16,11 @@
     Operation,
 )
 from job_executor.common.exceptions import HttpRequestError, HttpResponseError
-from job_executor.config import environment
+from job_executor.config import environment, secrets
 
 DATASTORE_API_URL = environment.datastore_api_url
 DEFAULT_REQUESTS_TIMEOUT = (10, 60)  # (read timeout, connect timeout)
+DATASTORE_API_SERVICE_KEY = secrets.datastore_api_service_key
 
 logger = logging.getLogger()
 
@@ -117,6 +118,25 @@ def get_datastore_directory(rdn: str) -> Path:
     return Path(DatastoreResponse.model_validate(response.json()).directory)
 
 
+def post_public_key(datastore_rdn: str, public_key_pem: bytes) -> None:
+    """
+    Post the public RSA key to the datastore-api.
+
+    :param datastore_rdn: The RDN of the datastore
+    :param public_key_pem: The public key in PEM format as bytes
+    """
+    request_url = f"{DATASTORE_API_URL}/datastores/{datastore_rdn}/public-key"
+    execute_request(
+        "POST",
+        request_url,
+        data=public_key_pem,
+        headers={
+            "Content-Type": "application/x-pem-file",
+            "X-API-Key": DATASTORE_API_SERVICE_KEY,
+        },
+    )
+
+
 def query_for_jobs() -> JobQueryResult:
     """
     Retrieves different types of jobs based on the system's state
@@ -147,6 +167,7 @@ def query_for_jobs() -> JobQueryResult:
                         Operation.REMOVE,
                         Operation.ROLLBACK_REMOVE,
                         Operation.DELETE_ARCHIVE,
+                        Operation.GENERATE_RSA_KEYS,
                     ],
                 ),
                 queued_worker_jobs=get_jobs(

job_executor/adapter/datastore_api/__init__.py

@@ -50,6 +50,7 @@ class Operation(StrEnum):
     REMOVE = "REMOVE"
     ROLLBACK_REMOVE = "ROLLBACK_REMOVE"
     DELETE_ARCHIVE = "DELETE_ARCHIVE"
+    GENERATE_RSA_KEYS = "GENERATE_RSA_KEYS"
 
 
 class ReleaseStatus(StrEnum):

job_executor/adapter/datastore_api/models.py

@@ -4,7 +4,9 @@
 
 from job_executor.adapter.fs.datastore_files import DatastoreDirectory
 from job_executor.adapter.fs.input_files import InputDirectory
+from job_executor.adapter.fs.private_keys_directory import PrivateKeysDirectory
 from job_executor.adapter.fs.working_files import WorkingDirectory
+from job_executor.config import environment
 
 
 class FileSystemAdapter(Protocol):
@@ -21,13 +23,17 @@ class LocalStorageAdapter:
     datastore_dir: DatastoreDirectory
     working_dir: WorkingDirectory
     input_dir: InputDirectory
+    private_keys_dir: PrivateKeysDirectory
 
-    def __init__(self, datastore_dir_path: Path) -> None:
+    def __init__(self, datastore_dir_path: Path, datastore_rdn: str) -> None:
         self.datastore_dir = DatastoreDirectory(datastore_dir_path)
         self.working_dir = WorkingDirectory(
             Path(f"{datastore_dir_path}_working")
         )
         self.input_dir = InputDirectory(Path(f"{datastore_dir_path}_input"))
+        self.private_keys_dir = PrivateKeysDirectory(
+            Path(environment.private_keys_dir) / datastore_rdn
+        )
 
     def move_working_dir_parquet_to_datastore(self, dataset_name: str) -> None:
         """

job_executor/adapter/fs/__init__.py

@@ -19,7 +19,6 @@
 
 class DatastoreDirectory:
     root_dir: Path
-    vault_dir: Path
     data_dir: Path
     metadata_dir: Path
     draft_metadata_all_path: Path
@@ -31,7 +30,6 @@ def __init__(self, root_dir: Path) -> None:
         self.root_dir = root_dir
         self.data_dir = root_dir / "data"
         self.metadata_dir = root_dir / "datastore"
-        self.vault_dir = root_dir / "vault"
         self.draft_version_path = self.metadata_dir / "draft_version.json"
         self.archive_dir = self.root_dir / "archive"
         self.draft_metadata_all_path = (

job_executor/adapter/fs/datastore_files.py

@@ -0,0 +1,27 @@
+import os
+from dataclasses import dataclass
+from pathlib import Path
+
+
+@dataclass
+class PrivateKeysDirectory:
+    path_with_rdn: Path
+
+    def create(self) -> bool:
+        if not self.path_with_rdn.exists():
+            os.makedirs(self.path_with_rdn)
+            return True
+        return False
+
+    def save_private_key(self, microdata_private_key_pem: bytes) -> None:
+        with open(self._get_private_key_location(), "wb") as file:
+            file.write(microdata_private_key_pem)
+
+    def clean_up(self) -> bool:
+        if self._get_private_key_location().exists():
+            os.remove(self._get_private_key_location())
+            return True
+        return False
+
+    def _get_private_key_location(self) -> Path:
+        return self.path_with_rdn / "microdata_private_key.pem"

job_executor/adapter/fs/private_keys_directory.py

@@ -24,7 +24,7 @@ def initialize_app() -> Manager:
         rollback.fix_interrupted_jobs()
         for rdn in datastore_api.get_datastores():
             local_storage = LocalStorageAdapter(
-                datastore_api.get_datastore_directory(rdn)
+                datastore_api.get_datastore_directory(rdn), rdn
             )
             if local_storage.datastore_dir.temporary_backup_exists():
                 raise StartupException(f"tmp directory exists for {rdn}")

job_executor/app.py

@@ -14,6 +14,7 @@ class Environment:
     docker_host_name: str
     commit_id: str
     max_gb_all_workers: int
+    private_keys_dir: str
 
 
 def _initialize_environment() -> Environment:
@@ -27,6 +28,7 @@ def _initialize_environment() -> Environment:
         docker_host_name=os.environ["DOCKER_HOST_NAME"],
         commit_id=os.environ["COMMIT_ID"],
         max_gb_all_workers=int(os.environ["MAX_GB_ALL_WORKERS"]),
+        private_keys_dir=os.environ["PRIVATE_KEYS_DIR"],
     )
 
 
@@ -36,13 +38,15 @@ def _initialize_environment() -> Environment:
 @dataclass
 class Secrets:
     pseudonym_service_api_key: str
+    datastore_api_service_key: str
 
 
 def _initialize_secrets() -> Secrets:
     with open(environment.secrets_file, encoding="utf-8") as f:
         secrets_file = json.load(f)
     return Secrets(
-        pseudonym_service_api_key=secrets_file["PSEUDONYM_SERVICE_API_KEY"]
+        pseudonym_service_api_key=secrets_file["PSEUDONYM_SERVICE_API_KEY"],
+        datastore_api_service_key=secrets_file["DATASTORE_API_SERVICE_KEY"],
     )