refactor(secrets): adapt to reana-commons secret-handling changes (#583)

Closes reanahub/reana-commons#455

reana_workflow_controller/consumer.py

@@ -23,7 +23,7 @@
     current_k8s_batchv1_api_client,
     current_k8s_corev1_api_client,
 )
-from reana_commons.k8s.secrets import REANAUserSecretsStore
+from reana_commons.k8s.secrets import UserSecretsStore
 from reana_commons.utils import (
     calculate_file_access_time,
     calculate_hash_of_dir,
@@ -180,8 +180,16 @@ def _update_commit_status(workflow, status):
         state = "canceled"
     else:
         state = "running"
-    secret_store = REANAUserSecretsStore(workflow.owner_id)
-    gitlab_access_token = secret_store.get_secret_value("gitlab_access_token")
+
+    user_secrets = UserSecretsStore.fetch(workflow.owner_id)
+    gitlab_access_token_secret = user_secrets.get_secret("gitlab_access_token")
+    if not gitlab_access_token_secret:
+        logging.error(
+            f"Skipping updating commit status for workflow {workflow.id_}: GitLab access token not found."
+        )
+        return
+    gitlab_access_token = gitlab_access_token_secret.value_str
+
     target_url = f"https://{REANA_HOSTNAME}/api/workflows/{workflow.id_}/logs"
     workflow_name = urlparse.quote_plus(workflow.git_repo)
     system_name = "reana"
@@ -190,7 +198,13 @@ def _update_commit_status(workflow, status):
         f"{workflow.git_ref}?access_token={gitlab_access_token}&state={state}&"
         f"target_url={target_url}&name={system_name}"
     )
-    requests.post(commit_status_url)
+
+    res = requests.post(commit_status_url)
+    if res.status_code >= 400:
+        logging.error(
+            f"Failed to update commit status for workflow {workflow.id_}: "
+            f"status code {res.status_code}, content {res.text}"
+        )
 
 
 def _update_run_progress(workflow_uuid, msg):

reana_workflow_controller/k8s.py

@@ -17,7 +17,7 @@
     current_k8s_corev1_api_client,
     current_k8s_networking_api_client,
 )
-from reana_commons.k8s.secrets import REANAUserSecretsStore
+from reana_commons.k8s.secrets import UserSecretsStore
 from reana_commons.k8s.volumes import (
     get_k8s_cvmfs_volumes,
     get_workspace_volume,
@@ -228,16 +228,16 @@ def add_run_with_root_permissions(self):
 
     def add_user_secrets(self):
         """Mount the "file" secrets and set the "env" secrets in the container."""
-        secrets_store = REANAUserSecretsStore(self.owner_id)
+        user_secrets = UserSecretsStore.fetch(self.owner_id)
 
         # mount file secrets
-        secrets_volume = secrets_store.get_file_secrets_volume_as_k8s_specs()
-        secrets_volume_mount = secrets_store.get_secrets_volume_mount_as_k8s_spec()
+        secrets_volume = user_secrets.get_file_secrets_volume_as_k8s_specs()
+        secrets_volume_mount = user_secrets.get_secrets_volume_mount_as_k8s_spec()
         self._pod_spec.volumes.append(secrets_volume)
         self._session_container.volume_mounts.append(secrets_volume_mount)
 
         # set environment secrets
-        self._session_container.env += secrets_store.get_env_secrets_as_k8s_spec()
+        self._session_container.env += user_secrets.get_env_secrets_as_k8s_spec()
 
     def get_deployment_objects(self):
         """Return the alrady built Kubernetes objects."""

reana_workflow_controller/rest/utils.py

@@ -33,7 +33,7 @@
 from kubernetes.client.rest import ApiException
 from reana_commons import workspace
 from reana_commons.config import REANA_WORKFLOW_UMASK, WORKFLOW_TIME_FORMAT
-from reana_commons.k8s.secrets import REANAUserSecretsStore
+from reana_commons.k8s.secrets import UserSecretsStore
 from reana_commons.utils import (
     get_workflow_status_change_verb,
     remove_upper_level_references,
@@ -367,8 +367,11 @@ def create_workflow_workspace(
     os.umask(REANA_WORKFLOW_UMASK)
     os.makedirs(path, exist_ok=True)
     if git_url and git_ref:
-        secret_store = REANAUserSecretsStore(user_id)
-        gitlab_access_token = secret_store.get_secret_value("gitlab_access_token")
+        user_secrets = UserSecretsStore.fetch(user_id)
+        gitlab_access_token_secret = user_secrets.get_secret("gitlab_access_token")
+        if not gitlab_access_token_secret:
+            raise Exception("GitLab access token not found.")
+        gitlab_access_token = gitlab_access_token_secret.value_str
         url = "https://oauth2:{0}@{1}/{2}.git".format(
             gitlab_access_token, REANA_GITLAB_HOST, git_url
         )

reana_workflow_controller/workflow_run_manager.py

@@ -37,7 +37,7 @@
 )
 from reana_commons.k8s.api_client import current_k8s_batchv1_api_client
 from reana_commons.k8s.kerberos import get_kerberos_k8s_config
-from reana_commons.k8s.secrets import REANAUserSecretsStore
+from reana_commons.k8s.secrets import UserSecretsStore
 from reana_commons.k8s.volumes import (
     create_cvmfs_persistent_volume_claim,
     get_workspace_volume,
@@ -572,12 +572,11 @@ def _create_job_spec(
             namespace=REANA_RUNTIME_KUBERNETES_NAMESPACE,
         )
 
-        secrets_store = REANAUserSecretsStore(owner_id)
-
+        user_secrets = UserSecretsStore.fetch(owner_id)
         kerberos = None
         if self.requires_kerberos():
             kerberos = get_kerberos_k8s_config(
-                secrets_store,
+                user_secrets,
                 kubernetes_uid=WORKFLOW_RUNTIME_USER_UID,
             )
 
@@ -634,9 +633,10 @@ def _create_job_spec(
             workflow_engine_container.volume_mounts += kerberos.volume_mounts
             workflow_engine_container.env += kerberos.env
 
-        job_controller_env_secrets = secrets_store.get_env_secrets_as_k8s_spec()
+        job_controller_env_secrets = user_secrets.get_env_secrets_as_k8s_spec()
 
-        user = secrets_store.get_secret_value("CERN_USER") or WORKFLOW_RUNTIME_USER_NAME
+        user_secret = user_secrets.get_secret("CERN_USER")
+        user = user_secret.value_str if user_secret else WORKFLOW_RUNTIME_USER_NAME
 
         job_controller_container = client.V1Container(
             name=current_app.config["JOB_CONTROLLER_NAME"],
@@ -723,7 +723,7 @@ def _create_job_spec(
                 },
             )
 
-        secrets_volume_mount = secrets_store.get_secrets_volume_mount_as_k8s_spec()
+        secrets_volume_mount = user_secrets.get_secrets_volume_mount_as_k8s_spec()
         job_controller_container.volume_mounts = [workspace_mount, secrets_volume_mount]
 
         job_controller_container.ports = [
@@ -741,7 +741,7 @@ def _create_job_spec(
         )
         volumes = [
             workspace_volume,
-            secrets_store.get_file_secrets_volume_as_k8s_specs(),
+            user_secrets.get_file_secrets_volume_as_k8s_specs(),
         ]
 
         if kerberos:

setup.py

@@ -1,7 +1,7 @@
 # -*- coding: utf-8 -*-
 #
 # This file is part of REANA.
-# Copyright (C) 2017, 2018, 2019, 2020, 2021, 2022, 2023 CERN.
+# Copyright (C) 2017, 2018, 2019, 2020, 2021, 2022, 2023, 2024 CERN.
 #
 # REANA is free software; you can redistribute it and/or modify it
 # under the terms of the MIT License; see LICENSE file for more details.
@@ -50,7 +50,7 @@
     "jsonpickle>=0.9.6",
     "marshmallow>2.13.0,<3.0.0",  # same upper pin as reana-server
     "packaging>=18.0",
-    "reana-commons[kubernetes]>=0.95.0a3,<0.96.0",
+    "reana-commons[kubernetes] @ git+https://github.com/reanahub/[email protected]",
     "reana-db>=0.95.0a3,<0.96.0",
     "requests>=2.25.0",
     "sqlalchemy-utils>=0.31.0",

tests/test_consumer.py

@@ -14,9 +14,10 @@
 from reana_commons.config import MQ_DEFAULT_QUEUES
 from reana_commons.consumer import BaseConsumer
 from reana_commons.publisher import WorkflowStatusPublisher
+from reana_commons.k8s.secrets import UserSecrets, Secret
 from reana_db.models import RunStatus
 
-from reana_workflow_controller.consumer import JobStatusConsumer
+from reana_workflow_controller.consumer import JobStatusConsumer, _update_commit_status
 
 
 def test_workflow_finish_and_kubernetes_not_available(
@@ -45,3 +46,67 @@ def test_workflow_finish_and_kubernetes_not_available(
     ):
         consume_queue(job_status_consumer, limit=1)
     assert sample_serial_workflow_in_db.status == next_status
+
+
+@pytest.mark.parametrize(
+    "status,gitlab_status",
+    [
+        (RunStatus.created, "running"),
+        (RunStatus.deleted, "canceled"),
+        (RunStatus.failed, "failed"),
+        (RunStatus.finished, "success"),
+        (RunStatus.pending, "running"),
+        (RunStatus.queued, "running"),
+        (RunStatus.running, "running"),
+        (RunStatus.stopped, "canceled"),
+    ],
+)
+def test_update_commit_status(
+    session, sample_serial_workflow_in_db, status, gitlab_status
+):
+    """Test update commit status."""
+    workflow = sample_serial_workflow_in_db
+    workflow.git_repo = "foo/bar"
+    session.add(workflow)
+    session.commit()
+
+    post_mock = Mock()
+    post_mock.return_value.status_code = 200
+    secrets = UserSecrets(
+        user_id=str(workflow.owner_id),
+        k8s_secret_name="k8s-secret",
+        secrets=[Secret(name="gitlab_access_token", type_="env", value="my-token")],
+    )
+    fetch_mock = Mock()
+    fetch_mock.return_value = secrets
+    with patch("requests.post", post_mock), patch(
+        "reana_commons.k8s.secrets.UserSecretsStore.fetch", fetch_mock
+    ):
+        _update_commit_status(workflow, status)
+        fetch_mock.assert_called_once_with(workflow.owner_id)
+        post_mock.assert_called_once()
+        url = post_mock.call_args.args[0]
+        assert "access_token=my-token" in url
+        assert f"state={gitlab_status}" in url
+
+
+def test_update_commit_status_without_token(session, sample_serial_workflow_in_db):
+    """Test updating commit status without valid GitLab token."""
+    workflow = sample_serial_workflow_in_db
+    workflow.git_repo = "foo/bar"
+    session.add(workflow)
+    session.commit()
+
+    post_mock = Mock()
+    secrets = UserSecrets(
+        user_id=str(workflow.owner_id),
+        k8s_secret_name="k8s-secret",
+    )
+    fetch_mock = Mock()
+    fetch_mock.return_value = secrets
+    with patch("requests.post", post_mock), patch(
+        "reana_commons.k8s.secrets.UserSecretsStore.fetch", fetch_mock
+    ):
+        _update_commit_status(workflow, RunStatus.finished)
+        fetch_mock.assert_called_once_with(workflow.owner_id)
+        post_mock.assert_not_called()

tests/test_k8s.py

@@ -5,26 +5,24 @@
 # under the terms of the MIT License; see LICENSE file for more details.
 
 from unittest.mock import Mock, patch
+from uuid import uuid4
+
 from reana_workflow_controller.k8s import InteractiveDeploymentK8sBuilder
-from reana_commons.k8s.secrets import REANAUserSecretsStore
+from reana_commons.k8s.secrets import UserSecretsStore, UserSecrets, Secret
 
 
 def test_interactive_deployment_k8s_builder_user_secrets(monkeypatch):
     """Expose user secrets in interactive sessions"""
-    monkeypatch.setattr(
-        REANAUserSecretsStore,
-        "get_file_secrets_volume_as_k8s_specs",
-        lambda _: {"name": "secrets-volume"},
-    )
-    monkeypatch.setattr(
-        REANAUserSecretsStore,
-        "get_secrets_volume_mount_as_k8s_spec",
-        lambda _: {"name": "secrets-volume-mount"},
+    user_id = uuid4()
+    user_secrets = UserSecrets(
+        user_id=str(user_id),
+        k8s_secret_name="k8s-secret",
+        secrets=[Secret(name="third_env", type_="env", value="3")],
     )
     monkeypatch.setattr(
-        REANAUserSecretsStore,
-        "get_env_secrets_as_k8s_spec",
-        lambda _: [{"name": "third_env", "value": "3"}],
+        UserSecretsStore,
+        "fetch",
+        lambda _: user_secrets,
     )
 
     builder = InteractiveDeploymentK8sBuilder(
@@ -42,6 +40,6 @@ def test_interactive_deployment_k8s_builder_user_secrets(monkeypatch):
     deployment = objs["deployment"]
     pod = deployment.spec.template.spec
     assert len(pod.containers) == 1
-    assert {"name": "secrets-volume"} in pod.volumes
-    assert {"name": "secrets-volume-mount"} in pod.containers[0].volume_mounts
-    assert {"name": "third_env", "value": "3"} in pod.containers[0].env
+    assert any(v["name"] == "k8s-secret" for v in pod.volumes)
+    assert any(vm["name"] == "k8s-secret" for vm in pod.containers[0].volume_mounts)
+    assert any(e["name"] == "third_env" for e in pod.containers[0].env)

tests/test_views.py

@@ -1155,8 +1155,6 @@ def test_start_workflow_db_failure(
     app,
     session,
     default_user,
-    user_secrets,
-    corev1_api_client_with_user_secrets,
     sample_serial_workflow_in_db,
 ):
     """Test starting workflow with a DB failure."""
@@ -1193,8 +1191,6 @@ def test_start_workflow_kubernetes_failure(
     app,
     session,
     default_user,
-    user_secrets,
-    corev1_api_client_with_user_secrets,
     sample_serial_workflow_in_db,
 ):
     """Test starting workflow with a Kubernetes failure when creating jobs."""
@@ -1490,7 +1486,7 @@ def test_create_interactive_session(
             current_k8s_corev1_api_client=mock.DEFAULT,
             current_k8s_networking_api_client=mock.DEFAULT,
             current_k8s_appsv1_api_client=mock.DEFAULT,
-            REANAUserSecretsStore=mock.DEFAULT,
+            UserSecretsStore=mock.DEFAULT,
         ):
             res = client.post(
                 url_for(
@@ -1533,7 +1529,7 @@ def test_create_interactive_session_custom_image(
             current_k8s_corev1_api_client=mock.DEFAULT,
             current_k8s_networking_api_client=mock.DEFAULT,
             current_k8s_appsv1_api_client=mock.DEFAULT,
-            REANAUserSecretsStore=mock.DEFAULT,
+            UserSecretsStore=mock.DEFAULT,
         ) as mocks:
             client.post(
                 url_for(

tests/test_workflow_run_manager.py

@@ -19,8 +19,6 @@
     RunStatus,
     InteractiveSession,
     InteractiveSessionType,
-    JobStatus,
-    Job,
 )
 
 from reana_workflow_controller.config import (