opensource/galaxy
3
0
Fork 0
mirror of https://github.com/galaxyproject/galaxy.git synced 2026-09-01 15:37:32 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #21821 from arash77/fix-21715-tool-credentials-containerized-destinations

Browse Source 
[25.1] Fix tool credentials on containerized (Singularity/Docker) destinations
This commit is contained in:
Marius van den Beek
2026-02-20 14:22:57 +01:00
committed by GitHub
parent 650f559210 e84e5ba700
commit 4da5a5f5ee
6 changed files with 432 additions and 162 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/galaxy/tools/__init__.py
+11
View File
@@ -1482,6 +1482,17 @@ class Tool(UsesDictVisibleKeys, ToolParameterBundle):
self.javascript_requirements = javasscript_requirements
self.credentials = credentials
# Add credential inject_as_env names to docker_env_pass_through
# so they are passed into containerized environments (Docker -e, Singularity SINGULARITYENV_)
if self.credentials:
if not self.docker_env_pass_through:
self.docker_env_pass_through = []
for credential in self.credentials:
for secret in credential.secrets:
self.docker_env_pass_through.append(secret.inject_as_env)
for variable in credential.variables:
self.docker_env_pass_through.append(variable.inject_as_env)
required_files = tool_source.parse_required_files()
if required_files is None:
old_id = self.old_id
lib/galaxy/workflow/modules.py
+40
View File
@@ -29,6 +29,7 @@ from galaxy.exceptions import (
)
from galaxy.job_execution.actions.post import ActionBox
from galaxy.job_execution.compute_environment import ComputeEnvironment
from galaxy.managers.credentials import _build_user_credentials_query
from galaxy.model import (
Job,
PostJobAction,
@@ -42,6 +43,11 @@ from galaxy.model.dataset_collections import matching
from galaxy.model.dataset_collections.query import HistoryQuery
from galaxy.model.dataset_collections.type_description import COLLECTION_TYPE_DESCRIPTION_FACTORY
from galaxy.model.dataset_collections.types.sample_sheet_util import validate_column_definitions
from galaxy.schema.credentials import (
CredentialsContext,
SelectedGroup,
ServiceCredentialsContext,
)
from galaxy.schema.invocation import (
CancelReason,
FailureReason,
@@ -2503,6 +2509,7 @@ class ToolModule(WorkflowModule):
if pja.action_type == "ValidateOutputsAction":
validate_outputs = True
credentials_context = self._resolve_credentials_context(tool)
execution_tracker = execute(
trans=self.trans,
tool=tool,
@@ -2518,6 +2525,7 @@ class ToolModule(WorkflowModule):
),
completed_jobs=completed_jobs,
workflow_resource_parameters=resource_parameters,
credentials_context=credentials_context,
)
complete = True
except PartialJobExecution as pje:
@@ -2581,6 +2589,38 @@ class ToolModule(WorkflowModule):
self.trans, self.trans.sa_session, pja, step_inputs, step_outputs, replacement_dict
)
def _resolve_credentials_context(self, tool: "Tool") -> Optional[CredentialsContext]:
"""Auto-resolve the user's current credentials for a tool in workflow execution."""
if not tool.credentials:
return None
trans = self.trans
if not trans.user:
return None
stmt = _build_user_credentials_query(
user_id=trans.user.id,
source_type="tool",
source_id=tool.id,
current_group_only=True,
)
results = trans.sa_session.execute(stmt).tuples().all()
if not results:
return None
encode = trans.security.encode_id
seen = {}
for user_cred, group, _cred in results:
key = (user_cred.id, user_cred.name, user_cred.version)
if key not in seen:
seen[key] = ServiceCredentialsContext(
user_credentials_id=encode(user_cred.id),
name=user_cred.name,
version=user_cred.version,
selected_group=SelectedGroup(
id=encode(group.id),
name=group.name,
),
)
return CredentialsContext(root=list(seen.values()))
def _handle_post_job_actions(self, step, job, replacement_dict):
# Create new PJA associations with the created job, to be run on completion.
# PJA Parameter Replacement (only applies to immediate actions-- rename specifically, for now)
lib/galaxy_test/base/populators.py
+200
View File
@@ -1199,6 +1199,8 @@ class BaseDatasetPopulator(BasePopulator):
kwds["__files"][key] = value
del inputs[key]
if "credentials_context" in kwds and not isinstance(kwds["credentials_context"], str):
kwds["credentials_context"] = json.dumps(kwds["credentials_context"])
return dict(tool_id=tool_id, inputs=json.dumps(inputs), history_id=history_id, **kwds)
def build_tool_state(self, tool_id: str, history_id: str):
@@ -2031,6 +2033,204 @@ class DatasetPopulator(GalaxyInteractorHttpMixin, BaseDatasetPopulator):
yield history_id
class BaseCredentialsPopulator(BasePopulator):
"""Abstract base class for credential operations in Galaxy tests."""
DEFAULT_SOURCE_TYPE = "tool"
DEFAULT_SOURCE_VERSION = "test"
DEFAULT_SERVICE_NAME = "service1"
DEFAULT_SERVICE_VERSION = "v1"
def build_credentials_payload(
self,
tool_id: str,
variables: list,
secrets: list,
source_type: str = DEFAULT_SOURCE_TYPE,
source_version: str = DEFAULT_SOURCE_VERSION,
service_name: str = DEFAULT_SERVICE_NAME,
service_version: str = DEFAULT_SERVICE_VERSION,
group_name: Optional[str] = None,
) -> dict:
"""Build and return a credentials payload dict without posting it."""
if group_name is None:
group_name = random_name()
return {
"source_type": source_type,
"source_id": tool_id,
"source_version": source_version,
"service_credential": {
"name": service_name,
"version": service_version,
"group": {
"name": group_name,
"variables": [dict(v) for v in variables],
"secrets": [dict(s) for s in secrets],
},
},
}
def post_credentials(self, payload: dict, expected_status: int = 200, anon: bool = False) -> dict:
"""Post credentials payload, assert expected status, and return the response JSON."""
response = self._post("/api/users/current/credentials", data=payload, json=True, anon=anon)
api_asserts.assert_status_code_is(response, expected_status)
return response.json()
def create_credentials(
self,
tool_id: str,
variables: list,
secrets: list,
expected_status: int = 200,
anon: bool = False,
**kwargs,
) -> dict:
"""Build and post credentials in one step. Returns response JSON."""
payload = self.build_credentials_payload(tool_id=tool_id, variables=variables, secrets=secrets, **kwargs)
return self.post_credentials(payload, expected_status=expected_status, anon=anon)
def list_credentials(
self,
source_type: Optional[str] = None,
source_id: Optional[str] = None,
include_definition: bool = False,
expected_status: int = 200,
) -> list:
"""Return credentials for the current user, optionally filtered by source or definition."""
params = []
if source_type is not None:
params.append(f"source_type={source_type}")
if source_id is not None:
params.append(f"source_id={source_id}")
if include_definition:
params.append("include_definition=true")
url = "/api/users/current/credentials"
if params:
url += "?" + "&".join(params)
response = self._get(url)
api_asserts.assert_status_code_is(response, expected_status)
return response.json()
def get_credentials(self, source_type: str, source_id: str) -> list:
"""Return credentials list for the given source."""
return self.list_credentials(source_type=source_type, source_id=source_id)
def update_credentials_group(
self, user_credentials_id: str, group_id: str, payload: dict, expected_status: int = 200
) -> dict:
"""PUT /api/users/current/credentials/{id}/groups/{gid} and return response JSON."""
response = self._put(
f"/api/users/current/credentials/{user_credentials_id}/groups/{group_id}", data=payload, json=True
)
api_asserts.assert_status_code_is(response, expected_status)
return response.json()
def select_current_group(
self,
source_type: str,
source_id: str,
source_version: str,
user_credentials_id: str,
current_group_id: Optional[str],
expected_status: int = 204,
) -> None:
"""PUT /api/users/current/credentials to select (or unset) the current group."""
payload = {
"source_type": source_type,
"source_id": source_id,
"source_version": source_version,
"service_credentials": [{"user_credentials_id": user_credentials_id, "current_group_id": current_group_id}],
}
response = self._put("/api/users/current/credentials", data=payload, json=True)
api_asserts.assert_status_code_is(response, expected_status)
def delete_service_credentials(self, user_credentials_id: str, expected_status: int = 204) -> None:
"""DELETE /api/users/current/credentials/{id}."""
response = self._delete(f"/api/users/current/credentials/{user_credentials_id}")
api_asserts.assert_status_code_is(response, expected_status)
def delete_credentials_group(self, user_credentials_id: str, group_id: str, expected_status: int = 204) -> None:
"""DELETE /api/users/current/credentials/{id}/groups/{gid}."""
response = self._delete(f"/api/users/current/credentials/{user_credentials_id}/groups/{group_id}")
api_asserts.assert_status_code_is(response, expected_status)
def setup_credentials_context(
self,
tool_id: str,
variables: list,
secrets: list,
service_name: str = DEFAULT_SERVICE_NAME,
service_version: str = DEFAULT_SERVICE_VERSION,
group_name: str = "default",
source_version: str = DEFAULT_SOURCE_VERSION,
) -> list:
"""Create credentials and return a ready-to-use credentials_context list for tool execution.
Idempotent: if the service credentials and group already exist (e.g. from a prior test
in the same server session), reuses them instead of failing with a 409 conflict.
Also selects the group as the current group so server-side resolution
(e.g. during workflow execution) can find it via current_group_id.
"""
# Check whether credentials already exist for this service.
existing = self.get_credentials("tool", tool_id)
user_cred_entry = next(
(c for c in existing if c.get("name") == service_name and c.get("version") == service_version),
None,
)
if user_cred_entry:
user_credentials_id = user_cred_entry["id"]
created_group = next(
(g for g in user_cred_entry.get("groups", []) if g["name"] == group_name),
None,
)
if created_group is None:
# User credentials exist but this group doesn't — create only the group.
created_group = self.create_credentials(
tool_id=tool_id,
variables=variables,
secrets=secrets,
service_name=service_name,
service_version=service_version,
group_name=group_name,
source_version=source_version,
)
else:
created_group = self.create_credentials(
tool_id=tool_id,
variables=variables,
secrets=secrets,
service_name=service_name,
service_version=service_version,
group_name=group_name,
source_version=source_version,
)
credentials_list = self.get_credentials("tool", tool_id)
user_credentials_id = credentials_list[0]["id"]
self.select_current_group(
source_type="tool",
source_id=tool_id,
source_version=source_version,
user_credentials_id=user_credentials_id,
current_group_id=created_group["id"],
)
return [
{
"user_credentials_id": user_credentials_id,
"name": service_name,
"version": service_version,
"selected_group": {
"id": created_group["id"],
"name": created_group["name"],
},
}
]
class CredentialsPopulator(GalaxyInteractorHttpMixin, BaseCredentialsPopulator):
def __init__(self, galaxy_interactor: ApiTestInteractor) -> None:
self.galaxy_interactor = galaxy_interactor
# Things gxformat2 knows how to upload as workflows
YamlContentT = Union[StrPath, dict]
test/functional/tools/secret_tool.xml
+1
View File
@@ -1,5 +1,6 @@
<tool id="secret_tool" name="secret_tool" version="test" profile="23.0">
<requirements>
<container type="docker">busybox:1.36.1-glibc</container>
<credentials name="service1" version="v1" label="Your credentials set" description="Optional description of the service using credentials">
<variable name="server" inject_as_env="service1_url" optional="false" label="Your Service1 server" description="You can set the server..."/>
<secret name="username" inject_as_env="service1_user" optional="false" label="Your Service1 username" description="Your username is your email"/>
test/integration/test_containerized_jobs.py
+95 -6
View File
@@ -3,13 +3,19 @@
import json
import os
import unittest
from typing import (
Any,
)
from typing import Any
from galaxy.util.commands import which
from galaxy_test.base.populators import DatasetPopulator
from galaxy_test.driver.integration_util import IntegrationTestCase
from galaxy_test.base.populators import (
CredentialsPopulator,
DatasetPopulator,
skip_without_tool,
WorkflowPopulator,
)
from galaxy_test.driver.integration_util import (
ConfiguresDatabaseVault,
IntegrationTestCase,
)
from .test_job_environments import BaseJobEnvironmentIntegrationTestCase
SCRIPT_DIRECTORY = os.path.abspath(os.path.dirname(__file__))
@@ -22,6 +28,10 @@ SINGULARITY_JOB_CONFIG_FILE = os.path.join(SCRIPT_DIRECTORY, "singularity_job_co
EXTENDED_TIMEOUT = 120
CREDENTIALS_TEST_TOOL = "secret_tool"
CONTAINER_TEST_VARIABLES = [{"name": "server", "value": "http://test-server:8080"}]
CONTAINER_TEST_SECRETS = [{"name": "username", "value": "test_user"}, {"name": "password", "value": "test_pass"}]
class MulledJobTestCases:
"""
@@ -29,6 +39,7 @@ class MulledJobTestCases:
"""
dataset_populator: DatasetPopulator
credentials_populator: CredentialsPopulator
container_type: str
def _run_and_get_contents(self, tool_id: str, history_id: str):
@@ -98,7 +109,7 @@ def skip_if_container_type_unavailable(cls) -> None:
raise unittest.SkipTest(f"Executable '{cls.container_type}' not found on PATH")
class TestDockerizedJobsIntegration(BaseJobEnvironmentIntegrationTestCase, MulledJobTestCases):
class TestDockerizedJobsIntegration(BaseJobEnvironmentIntegrationTestCase, MulledJobTestCases, ConfiguresDatabaseVault):
dataset_populator: DatasetPopulator
jobs_directory: str
job_config_file = DOCKERIZED_JOB_CONFIG_FILE
@@ -112,6 +123,7 @@ class TestDockerizedJobsIntegration(BaseJobEnvironmentIntegrationTestCase, Mulle
config["jobs_directory"] = cls.jobs_directory
config["job_config_file"] = cls.job_config_file
disable_dependency_resolution(config)
cls._configure_database_vault(config)
@classmethod
def setUpClass(cls) -> None:
@@ -120,6 +132,14 @@ class TestDockerizedJobsIntegration(BaseJobEnvironmentIntegrationTestCase, Mulle
def setUp(self) -> None:
super().setUp()
self.credentials_populator = CredentialsPopulator(self.galaxy_interactor)
self.workflow_populator = WorkflowPopulator(self.galaxy_interactor)
def _setup_credentials_context(self, **kwargs):
kwargs.setdefault("tool_id", CREDENTIALS_TEST_TOOL)
kwargs.setdefault("variables", CONTAINER_TEST_VARIABLES)
kwargs.setdefault("secrets", CONTAINER_TEST_SECRETS)
return self.credentials_populator.setup_credentials_context(**kwargs)
def test_container_job_environment(self) -> None:
"""
@@ -224,6 +244,75 @@ class TestDockerizedJobsIntegration(BaseJobEnvironmentIntegrationTestCase, Mulle
"""
assert identifier == f"quay.io/local/{expected_hash}"
@skip_without_tool("secret_tool")
def test_credentials_passed_to_container(self) -> None:
"""
Test that tool credentials are passed as environment variables into containerized environments.
"""
credentials_context = self._setup_credentials_context()
# Run the containerized tool that outputs credential environment variables
with self.dataset_populator.test_history() as history_id:
run_response = self.dataset_populator.run_tool(
"secret_tool", {}, history_id, credentials_context=credentials_context
)
job_id = run_response["jobs"][0]["id"]
self.dataset_populator.wait_for_job(job_id=job_id, assert_ok=True, timeout=EXTENDED_TIMEOUT)
# Get the output - should contain the credential environment variables
output = self.dataset_populator.get_history_dataset_content(
history_id, content_id=run_response["outputs"][0]["id"]
)
# Verify that credential environment variables were available in the container
lines = output.strip().split("\n")
assert len(lines) == 3, f"Expected 3 lines in output, got {len(lines)}: {lines}"
assert lines[0] == "http://test-server:8080", f"Expected server URL in first line, got: {lines[0]}"
assert lines[1] == "test_user", f"Expected username in second line, got: {lines[1]}"
assert lines[2] == "test_pass", f"Expected password in third line, got: {lines[2]}"
@skip_without_tool("secret_tool")
def test_credentials_passed_to_container_in_workflow(self) -> None:
"""
Test that tool credentials are passed into containerized workflow steps.
This is a regression test for issue #21715: when a tool using credentials
is run as a workflow step, the credentials are silently dropped because
ToolModule.execute() does not forward credentials_context.
"""
self._setup_credentials_context()
workflow_yaml = """
class: GalaxyWorkflow
steps:
secret_step:
tool_id: secret_tool
"""
workflow_id = self.workflow_populator.upload_yaml_workflow(workflow_yaml)
with self.dataset_populator.test_history() as history_id:
self.workflow_populator.invoke_workflow_and_wait(
workflow_id,
history_id=history_id,
assert_ok=True,
)
# Get the single output dataset from the history
history_contents = self.dataset_populator.get_history_contents(history_id)
datasets = [item for item in history_contents if item["history_content_type"] == "dataset"]
assert len(datasets) == 1, f"Expected 1 output dataset, got {len(datasets)}"
output = self.dataset_populator.get_history_dataset_content(
history_id, content_id=datasets[0]["id"], timeout=EXTENDED_TIMEOUT
)
# Verify that credential environment variables were available in the container.
# This will fail because credentials are not forwarded through workflow execution.
lines = output.strip().split("\n")
assert len(lines) == 3, f"Expected 3 lines in output, got {len(lines)}: {lines}"
assert lines[0] == "http://test-server:8080", f"Expected server URL in first line, got: {lines[0]}"
assert lines[1] == "test_user", f"Expected username in second line, got: {lines[1]}"
assert lines[2] == "test_pass", f"Expected password in third line, got: {lines[2]}"
class TestMappingContainerResolver(IntegrationTestCase):
"""
test/integration/test_credentials.py
+85 -156
View File
@@ -3,10 +3,15 @@ from typing import Optional
from galaxy.model.db.user import get_user_by_email
from galaxy.security.vault import UserVaultWrapper
from galaxy_test.base.api_util import random_name
from galaxy_test.base.populators import skip_without_tool
from galaxy_test.base.populators import (
CredentialsPopulator,
skip_without_tool,
)
from galaxy_test.driver import integration_util
CREDENTIALS_TEST_TOOL = "secret_tool"
DEFAULT_TOOL_VARIABLES = [{"name": "server", "value": "http://localhost:8080"}]
DEFAULT_TOOL_SECRETS = [{"name": "username", "value": "user"}, {"name": "password", "value": "pass"}]
class TestCredentialsApi(integration_util.IntegrationTestCase, integration_util.ConfiguresDatabaseVault):
@@ -15,28 +20,27 @@ class TestCredentialsApi(integration_util.IntegrationTestCase, integration_util.
super().handle_galaxy_config_kwds(config)
cls._configure_database_vault(config)
def setUp(self):
super().setUp()
self.credentials_populator = CredentialsPopulator(self.galaxy_interactor)
@skip_without_tool(CREDENTIALS_TEST_TOOL)
def test_provide_credential(self):
payload = self._build_credentials_payload(group_name="default")
created_credential_group = self._provide_user_credentials(payload=payload)
created_credential_group = self._create_credentials(group_name="default")
assert created_credential_group["name"] == "default"
assert len(created_credential_group["variables"]) == 1
assert len(created_credential_group["secrets"]) == 2
@skip_without_tool(CREDENTIALS_TEST_TOOL)
def test_anon_users_cannot_provide_credentials(self):
payload = self._build_credentials_payload()
response = self._post("/api/users/current/credentials", data=payload, json=True, anon=True)
self._assert_status_code_is(response, 403)
self._create_credentials(expected_status=403, anon=True)
@skip_without_tool(CREDENTIALS_TEST_TOOL)
def test_list_user_credentials(self):
self._provide_user_credentials()
self._create_credentials()
# Check there is at least one credential
response = self._get("/api/users/current/credentials")
self._assert_status_code_is(response, 200)
list_user_credentials = response.json()
list_user_credentials = self.credentials_populator.list_credentials()
assert len(list_user_credentials) > 0
# Check the specific credential exists
@@ -44,7 +48,7 @@ class TestCredentialsApi(integration_util.IntegrationTestCase, integration_util.
@skip_without_tool(CREDENTIALS_TEST_TOOL)
def test_other_users_cannot_list_credentials(self):
self._provide_user_credentials()
self._create_credentials()
self._check_credentials_exist()
@@ -52,12 +56,10 @@ class TestCredentialsApi(integration_util.IntegrationTestCase, integration_util.
self._check_credentials_exist(num_credentials=0)
def test_list_by_source_id_requires_source_type(self):
response = self._get("/api/users/current/credentials?source_id={CREDENTIALS_TEST_TOOL}")
self._assert_status_code_is(response, 400)
self.credentials_populator.list_credentials(source_id=CREDENTIALS_TEST_TOOL, expected_status=400)
def test_list_unsupported_source_type(self):
response = self._get("/api/users/current/credentials?source_type=invalid")
self._assert_status_code_is(response, 400)
self.credentials_populator.list_credentials(source_type="invalid", expected_status=400)
@skip_without_tool(CREDENTIALS_TEST_TOOL)
def test_add_group_to_credentials(self):
@@ -65,8 +67,7 @@ class TestCredentialsApi(integration_util.IntegrationTestCase, integration_util.
initial_group = random_name()
# First, create initial credentials with a unique group
payload = self._build_credentials_payload(group_name=initial_group)
self._provide_user_credentials(payload)
self._create_credentials(group_name=initial_group)
initial_credentials = self._check_credentials_exist()
assert len(initial_credentials) == 1
@@ -76,8 +77,7 @@ class TestCredentialsApi(integration_util.IntegrationTestCase, integration_util.
# Create a new group with the same service credentials
second_group = random_name()
new_payload = self._build_credentials_payload(group_name=second_group)
self._provide_user_credentials(new_payload)
self._create_credentials(group_name=second_group)
# Check that both our groups exist
updated_credentials = self._check_credentials_exist()
@@ -90,8 +90,7 @@ class TestCredentialsApi(integration_util.IntegrationTestCase, integration_util.
@skip_without_tool(CREDENTIALS_TEST_TOOL)
def test_update_credentials_update_time(self):
payload = self._build_credentials_payload()
created_group = self._provide_user_credentials(payload)
created_group = self._create_credentials()
created_group_id = created_group["id"]
list_user_credentials = self._check_credentials_exist()
@@ -117,7 +116,7 @@ class TestCredentialsApi(integration_util.IntegrationTestCase, integration_util.
@skip_without_tool(CREDENTIALS_TEST_TOOL)
def test_update_credentials(self):
# Create initial credentials
initial_group = self._provide_user_credentials()
initial_group = self._create_credentials()
group_id = initial_group["id"]
list_user_credentials = self._check_credentials_exist()
@@ -162,7 +161,7 @@ class TestCredentialsApi(integration_util.IntegrationTestCase, integration_util.
@skip_without_tool(CREDENTIALS_TEST_TOOL)
def test_update_credentials_error_cases(self):
"""Test update error scenarios."""
group = self._provide_user_credentials()
group = self._create_credentials()
group_id = group["id"]
list_user_credentials = self._check_credentials_exist()
@@ -227,15 +226,14 @@ class TestCredentialsApi(integration_util.IntegrationTestCase, integration_util.
@skip_without_tool(CREDENTIALS_TEST_TOOL)
def test_delete_service_credentials(self):
# Create credentials
self._provide_user_credentials()
self._create_credentials()
# Check credentials exist and get the service credentials ID
credentials_list = self._check_credentials_exist()
service_credentials_id = credentials_list[0]["id"]
# Delete the entire service credentials
response = self._delete(f"/api/users/current/credentials/{service_credentials_id}")
self._assert_status_code_is(response, 204)
self.credentials_populator.delete_service_credentials(service_credentials_id)
# Check credentials are deleted
self._check_credentials_exist(num_credentials=0)
@@ -247,12 +245,10 @@ class TestCredentialsApi(integration_util.IntegrationTestCase, integration_util.
target_group_name = random_name()
# Create initial credentials with unique group
payload1 = self._build_credentials_payload(group_name=initial_group)
self._provide_user_credentials(payload1)
self._create_credentials(group_name=initial_group)
# Add a new group
new_payload = self._build_credentials_payload(group_name=target_group_name)
self._provide_user_credentials(new_payload)
self._create_credentials(group_name=target_group_name)
# Check credentials exist with both our groups
list_user_credentials = self._check_credentials_exist()
@@ -267,22 +263,16 @@ class TestCredentialsApi(integration_util.IntegrationTestCase, integration_util.
target_group_id = groups_before[target_group_name]["id"]
# Set the new group as current
select_payload = {
"source_type": "tool",
"source_id": CREDENTIALS_TEST_TOOL,
"source_version": "test",
"service_credentials": [{"user_credentials_id": user_credentials_id, "current_group_id": target_group_id}],
}
response = self._put("/api/users/current/credentials", data=select_payload, json=True)
self._assert_status_code_is(response, 204)
self.credentials_populator.select_current_group(
"tool", CREDENTIALS_TEST_TOOL, "test", user_credentials_id, target_group_id
)
# Verify it's set as current
list_user_credentials = self._check_credentials_exist()
assert list_user_credentials[0]["current_group_id"] == target_group_id
# Delete the group
response = self._delete(f"/api/users/current/credentials/{user_credentials_id}/groups/{target_group_id}")
self._assert_status_code_is(response, 204)
self.credentials_populator.delete_credentials_group(user_credentials_id, target_group_id)
# Check group is deleted - should only have our initial group left
list_user_credentials = self._check_credentials_exist()
@@ -293,49 +283,41 @@ class TestCredentialsApi(integration_util.IntegrationTestCase, integration_util.
@skip_without_tool(CREDENTIALS_TEST_TOOL)
def test_provide_credential_invalid_group(self):
payload = self._build_credentials_payload(group_name="")
self._provide_user_credentials(payload, status_code=400)
self._create_credentials(group_name="", expected_status=400)
def test_invalid_source_type(self):
payload = self._build_credentials_payload(source_type="invalid_source_type")
self._provide_user_credentials(payload, status_code=400)
self._create_credentials(source_type="invalid_source_type", expected_status=400)
def test_not_existing_tool(self):
payload = self._build_credentials_payload(source_id="nonexistent_tool")
self._provide_user_credentials(payload, status_code=404)
self._create_credentials(tool_id="nonexistent_tool", expected_status=404)
@skip_without_tool(CREDENTIALS_TEST_TOOL)
def test_not_existing_tool_version(self):
payload = self._build_credentials_payload(source_version="nonexistent_tool_version")
self._provide_user_credentials(payload, status_code=404)
self._create_credentials(source_version="nonexistent_tool_version", expected_status=404)
def test_not_existing_service_name(self):
payload = self._build_credentials_payload(service_name="nonexistent_service")
self._provide_user_credentials(payload, status_code=404)
self._create_credentials(service_name="nonexistent_service", expected_status=404)
@skip_without_tool(CREDENTIALS_TEST_TOOL)
def test_not_existing_service_version(self):
payload = self._build_credentials_payload(service_version="nonexistent_service_version")
self._provide_user_credentials(payload, status_code=404)
self._create_credentials(service_version="nonexistent_service_version", expected_status=404)
@skip_without_tool(CREDENTIALS_TEST_TOOL)
def test_invalid_credential_name(self):
for key in ["variables", "secrets"]:
payload = self._build_credentials_payload()
payload["service_credential"]["group"][key][0]["name"] = "invalid_name"
self._provide_user_credentials(payload, status_code=400)
self.credentials_populator.post_credentials(payload, expected_status=400)
def test_delete_nonexistent_service_credentials(self):
response = self._delete("/api/users/current/credentials/f2db41e1fa331b3e")
self._assert_status_code_is(response, 400)
self.credentials_populator.delete_service_credentials("f2db41e1fa331b3e", expected_status=400)
def test_delete_nonexistent_credentials_group(self):
response = self._delete("/api/users/current/credentials/f2db41e1fa331b3e/groups/f2db41e1fa331b3e")
self._assert_status_code_is(response, 400)
self.credentials_populator.delete_credentials_group("f2db41e1fa331b3e", "f2db41e1fa331b3e", expected_status=400)
@skip_without_tool(CREDENTIALS_TEST_TOOL)
def test_delete_default_credential_group(self):
created_user_credentials = self._provide_user_credentials()
created_user_credentials = self._create_credentials()
# The new API returns a single ServiceCredentialGroupResponse, not a list
group_id = created_user_credentials["id"]
@@ -343,15 +325,13 @@ class TestCredentialsApi(integration_util.IntegrationTestCase, integration_util.
user_credentials_list = self._check_credentials_exist()
user_credentials_id = user_credentials_list[0]["id"]
response = self._delete(f"/api/users/current/credentials/{user_credentials_id}/groups/{group_id}")
self._assert_status_code_is(response, 204)
self.credentials_populator.delete_credentials_group(user_credentials_id, group_id)
@skip_without_tool(CREDENTIALS_TEST_TOOL)
def test_unset_current_group(self):
# First create credentials with a unique group
group_name = random_name()
payload = self._build_credentials_payload(group_name=group_name)
self._provide_user_credentials(payload)
self._create_credentials(group_name=group_name)
# Set this group as current
user_credentials_list = self._check_credentials_exist()
@@ -362,14 +342,9 @@ class TestCredentialsApi(integration_util.IntegrationTestCase, integration_util.
default_group_id = group["id"]
break
select_payload = {
"source_type": "tool",
"source_id": CREDENTIALS_TEST_TOOL,
"source_version": "test",
"service_credentials": [{"user_credentials_id": user_credentials_id, "current_group_id": default_group_id}],
}
response = self._put("/api/users/current/credentials", data=select_payload, json=True)
self._assert_status_code_is(response, 204)
self.credentials_populator.select_current_group(
"tool", CREDENTIALS_TEST_TOOL, "test", user_credentials_id, default_group_id
)
# Verify it's set as current
list_user_credentials = self._check_credentials_exist()
@@ -382,14 +357,9 @@ class TestCredentialsApi(integration_util.IntegrationTestCase, integration_util.
assert current_group_name == group_name
# Now unset the current group (set to None)
unset_payload = {
"source_type": "tool",
"source_id": CREDENTIALS_TEST_TOOL,
"source_version": "test",
"service_credentials": [{"user_credentials_id": user_credentials_id, "current_group_id": None}],
}
response = self._put("/api/users/current/credentials", data=unset_payload, json=True)
self._assert_status_code_is(response, 204)
self.credentials_populator.select_current_group(
"tool", CREDENTIALS_TEST_TOOL, "test", user_credentials_id, None
)
# Verify current group is unset
list_user_credentials = self._check_credentials_exist()
@@ -399,50 +369,42 @@ class TestCredentialsApi(integration_util.IntegrationTestCase, integration_util.
def test_required_credentials_validation(self):
"""Test that required (non-optional) credentials are properly validated."""
# Test missing required variable
payload = self._build_credentials_payload()
payload["service_credential"]["group"]["variables"] = [] # Remove required 'server' variable
self._provide_user_credentials(payload, status_code=400)
self._create_credentials(variables=[], expected_status=400)
# Test missing required secret
payload = self._build_credentials_payload()
payload["service_credential"]["group"]["secrets"] = [
{"name": "password", "value": "pass"} # Remove required 'username' secret
]
self._provide_user_credentials(payload, status_code=400)
self._create_credentials(secrets=[{"name": "password", "value": "pass"}], expected_status=400)
# Test empty required variable
payload = self._build_credentials_payload()
payload["service_credential"]["group"]["variables"] = [{"name": "server", "value": ""}]
self._provide_user_credentials(payload, status_code=400)
self._create_credentials(variables=[{"name": "server", "value": ""}], expected_status=400)
# Test empty required secret
payload = self._build_credentials_payload()
payload["service_credential"]["group"]["secrets"] = [
{"name": "username", "value": ""}, # Empty required secret
{"name": "password", "value": "pass"},
]
self._provide_user_credentials(payload, status_code=400)
self._create_credentials(
secrets=[
{"name": "username", "value": ""}, # Empty required secret
{"name": "password", "value": "pass"},
],
expected_status=400,
)
# Test that optional credentials can be omitted (password is optional)
payload = self._build_credentials_payload()
payload["service_credential"]["group"]["secrets"] = [
{"name": "username", "value": "user"} # Only required secret, optional 'password' omitted
]
self._provide_user_credentials(payload, status_code=200)
self._create_credentials(
secrets=[{"name": "username", "value": "user"}], # Only required secret, optional 'password' omitted
expected_status=200,
)
@skip_without_tool(CREDENTIALS_TEST_TOOL)
def test_vault_integration(self):
test_user_email = "user@vault.test"
with self._different_user(test_user_email):
payload = self._build_credentials_payload()
self._provide_user_credentials(payload)
self.credentials_populator.post_credentials(payload)
credentials_list = self._check_credentials_exist()
assert len(credentials_list) == 1
group = credentials_list[0]["groups"][0]
# Check that secrets are stored in the vault
for secret in payload["service_credential"]["group"]["secrets"]:
for secret in DEFAULT_TOOL_SECRETS:
vault_ref = self._get_vault_ref(payload, group["id"], secret["name"])
expected_value = secret["value"]
self._check_vault_entry_exists(test_user_email, vault_ref, expected_value)
@@ -450,11 +412,10 @@ class TestCredentialsApi(integration_util.IntegrationTestCase, integration_util.
# Delete the credentials group
user_credentials_id = credentials_list[0]["id"]
group_id = group["id"]
response = self._delete(f"/api/users/current/credentials/{user_credentials_id}/groups/{group_id}")
self._assert_status_code_is(response, 204)
self.credentials_populator.delete_credentials_group(user_credentials_id, group_id)
# Check that secrets are removed from the vault
for secret in payload["service_credential"]["group"]["secrets"]:
for secret in DEFAULT_TOOL_SECRETS:
vault_ref = self._get_vault_ref(payload, group["id"], secret["name"])
self._check_vault_entry_exists(test_user_email, vault_ref, should_exist=False)
@@ -462,7 +423,7 @@ class TestCredentialsApi(integration_util.IntegrationTestCase, integration_util.
def test_list_credentials_with_missing_tool(self):
# Create credentials for the test tool
payload = self._build_credentials_payload()
self._provide_user_credentials(payload)
self.credentials_populator.post_credentials(payload)
# Verify credentials exist normally
credentials_list = self._check_credentials_exist()
@@ -482,9 +443,7 @@ class TestCredentialsApi(integration_util.IntegrationTestCase, integration_util.
assert self._app.toolbox.get_tool(CREDENTIALS_TEST_TOOL) is None
# Test 1: List credentials with include_definition=True
response = self._get("/api/users/current/credentials?include_definition=true")
self._assert_status_code_is(response, 200)
credentials_with_definition = response.json()
credentials_with_definition = self.credentials_populator.list_credentials(include_definition=True)
assert len(credentials_with_definition) == 1
credential = credentials_with_definition[0]
@@ -509,9 +468,7 @@ class TestCredentialsApi(integration_util.IntegrationTestCase, integration_util.
assert len(credential["groups"]) > 0
# Test 2: List credentials without include_definition
response = self._get("/api/users/current/credentials")
self._assert_status_code_is(response, 200)
credentials_without_definition = response.json()
credentials_without_definition = self.credentials_populator.list_credentials()
assert len(credentials_without_definition) == 1
credential_no_def = credentials_without_definition[0]
@@ -525,49 +482,23 @@ class TestCredentialsApi(integration_util.IntegrationTestCase, integration_util.
if tool is not None:
self._app.toolbox.register_tool(tool)
def _provide_user_credentials(self, payload=None, status_code=200):
payload = payload or self._build_credentials_payload()
response = self._post("/api/users/current/credentials", data=payload, json=True)
self._assert_status_code_is(response, status_code)
return response.json()
def _build_credentials_payload(
self,
source_type: str = "tool",
source_id: str = CREDENTIALS_TEST_TOOL,
source_version: str = "test",
service_name: str = "service1",
service_version: str = "v1",
group_name=None,
):
if group_name is None:
group_name = random_name()
return {
"source_type": source_type,
"source_id": source_id,
"source_version": source_version,
"service_credential": {
"name": service_name,
"version": service_version,
"group": {
"name": group_name,
"variables": [{"name": "server", "value": "http://localhost:8080"}],
"secrets": [
{"name": "username", "value": "user"},
{"name": "password", "value": "pass"},
],
},
},
}
def _update_credentials(self, user_credentials_id, group_id, payload=None, status_code=200):
payload = payload or self._build_update_credentials_payload()
response = self._put(
f"/api/users/current/credentials/{user_credentials_id}/groups/{group_id}", data=payload, json=True
return self.credentials_populator.update_credentials_group(
user_credentials_id, group_id, payload, expected_status=status_code
)
self._assert_status_code_is(response, status_code)
return response.json()
def _build_credentials_payload(self, **kwargs):
kwargs.setdefault("tool_id", CREDENTIALS_TEST_TOOL)
kwargs.setdefault("variables", DEFAULT_TOOL_VARIABLES)
kwargs.setdefault("secrets", DEFAULT_TOOL_SECRETS)
return self.credentials_populator.build_credentials_payload(**kwargs)
def _create_credentials(self, **kwargs):
kwargs.setdefault("tool_id", CREDENTIALS_TEST_TOOL)
kwargs.setdefault("variables", DEFAULT_TOOL_VARIABLES)
kwargs.setdefault("secrets", DEFAULT_TOOL_SECRETS)
return self.credentials_populator.create_credentials(**kwargs)
def _build_update_credentials_payload(
self,
@@ -585,9 +516,7 @@ class TestCredentialsApi(integration_util.IntegrationTestCase, integration_util.
return update_payload
def _check_credentials_exist(self, source_id: str = CREDENTIALS_TEST_TOOL, num_credentials: int = 1):
response = self._get(f"/api/users/current/credentials?source_type=tool&source_id={source_id}")
self._assert_status_code_is(response, 200)
list_user_credentials = response.json()
list_user_credentials = self.credentials_populator.list_credentials(source_type="tool", source_id=source_id)
assert len(list_user_credentials) == num_credentials
if num_credentials > 0:
assert list_user_credentials[0]["source_id"] == source_id
@@ -612,6 +541,6 @@ class TestCredentialsApi(integration_util.IntegrationTestCase, integration_util.
or stored_value == ""
), f"Expected vault entry '{vault_ref}' to not exist, but found value '{stored_value}'"
def _get_vault_ref(self, payload: dict, group_id: str, secret_name: str):
def _get_vault_ref(self, payload: dict, group_id: str, secret_name: str) -> str:
decoded_group_id = self._app.security.decode_id(group_id)
return f"{payload['source_type']}|{payload['source_id']}|{payload['service_credential']['name']}|{payload['service_credential']['version']}|{decoded_group_id}|{secret_name}"