Compare commits

..

1 Commits

Author SHA1 Message Date
Peter Braun
c71a49e044 fix: do not include secret values in the credentials test endpoint an… (#16425)
fix: do not include secret values in the credentials test endpoint and add a guard to make sure credentials are testable
2026-04-24 12:35:12 +00:00
4 changed files with 17 additions and 51 deletions

View File

@@ -4142,28 +4142,9 @@ class LaunchConfigurationBaseSerializer(BaseSerializer):
attrs['extra_data'][key] = db_extra_data[key]
# Build unsaved version of this config, use it to detect prompts errors
# Capture keys before _build_mock_obj pops pseudo-fields from attrs
incoming_attr_keys = set(attrs.keys())
mock_obj = self._build_mock_obj(attrs)
ask_mapping_keys = set(ujt.get_ask_mapping().keys())
requested_prompt_fields = incoming_attr_keys & ask_mapping_keys
if 'extra_data' in incoming_attr_keys:
requested_prompt_fields.add('extra_vars')
requested_prompt_fields.add('survey_passwords')
# prompts_dict() pulls persisted M2M state (labels, credentials,
# instance_groups) via the instance pk. Only re-validate the full prompt
# state when the caller is switching the underlying template; otherwise
# restrict validation to the fields the request explicitly provided.
if 'unified_job_template' in attrs:
prompts_to_validate = mock_obj.prompts_dict()
elif requested_prompt_fields:
prompts_to_validate = {k: v for k, v in mock_obj.prompts_dict().items() if k in requested_prompt_fields}
else:
prompts_to_validate = None
if prompts_to_validate is not None:
accepted, rejected, errors = ujt._accept_or_ignore_job_kwargs(_exclude_errors=self.exclude_errors, **prompts_to_validate)
if set(list(ujt.get_ask_mapping().keys()) + ['extra_data']) & set(attrs.keys()):
accepted, rejected, errors = ujt._accept_or_ignore_job_kwargs(_exclude_errors=self.exclude_errors, **mock_obj.prompts_dict())
else:
# Only perform validation of prompts if prompts fields are provided
errors = {}

View File

@@ -1720,12 +1720,10 @@ class OIDCCredentialTestMixin:
return {'details': {'sent_jwt_payload': self._decode_jwt_payload_for_display(jwt_token)}}
def _call_backend_with_error_handling(self, plugin, backend_kwargs, response_body):
"""Call credential backend and handle errors, adding secret_value to response if OIDC details present."""
"""Call credential backend and handle errors."""
try:
with set_environ(**settings.AWX_TASK_ENV):
secret_value = plugin.backend(**backend_kwargs)
if 'details' in response_body:
response_body['details']['secret_value'] = secret_value
plugin.backend(**backend_kwargs)
return Response(response_body, status=status.HTTP_202_ACCEPTED)
except requests.exceptions.HTTPError as exc:
message = self._extract_http_error_message(exc)
@@ -1791,6 +1789,8 @@ class CredentialExternalTest(OIDCCredentialTestMixin, SubDetailAPIView):
It does not support standard credential types such as Machine, SCM, and Cloud."""})
def post(self, request, *args, **kwargs):
obj = self.get_object()
if obj.credential_type.kind != 'external':
raise ParseError(_('Credential is not testable.'))
backend_kwargs = {}
for field_name, value in obj.inputs.items():
backend_kwargs[field_name] = obj.get_input(field_name)
@@ -1858,6 +1858,8 @@ class CredentialTypeExternalTest(OIDCCredentialTestMixin, SubDetailAPIView):
@extend_schema_if_available(extensions={"x-ai-description": "Test a complete set of input values for an external credential"})
def post(self, request, *args, **kwargs):
obj = self.get_object()
if obj.kind != 'external':
raise ParseError(_('Credential type is not testable.'))
backend_kwargs = request.data.get('inputs', {})
backend_kwargs.update(request.data.get('metadata', {}))

View File

@@ -257,3 +257,12 @@ def test_credential_type_test_success_returns_jwt_payload(mock_flag, post, admin
assert response.status_code == 202
assert 'details' in response.data
assert 'sent_jwt_payload' in response.data['details']
@pytest.mark.django_db
def test_credential_external_test_returns_400_for_non_external_credential(post, admin, credential):
# credential fixture creates a non-external credential (e.g. SSH/vault kind)
url = reverse('api:credential_external_test', kwargs={'pk': credential.pk})
response = post(url, {'metadata': {}}, admin)
assert response.status_code == 400
assert 'not testable' in response.data.get('detail', '').lower()

View File

@@ -13,7 +13,6 @@ from awx.main.models.workflow import (
WorkflowJobTemplateNode,
)
from awx.main.models.credential import Credential
from awx.main.models.label import Label
from awx.main.scheduler import TaskManager, WorkflowManager, DependencyManager
# Django
@@ -52,31 +51,6 @@ def test_node_accepts_prompted_fields(inventory, project, workflow_job_template,
post(url, {'unified_job_template': job_template.pk, 'limit': 'webservers'}, user=admin_user, expect=201)
@pytest.mark.django_db
def test_node_extra_data_patch_with_unprompted_labels(inventory, project, organization, workflow_job_template, patch, admin_user):
"""AAP-41742: PATCH extra_data on a workflow node should succeed even when
the node has labels associated but the JT has ask_labels_on_launch=False."""
jt = JobTemplate.objects.create(
inventory=inventory,
project=project,
playbook='helloworld.yml',
ask_variables_on_launch=True,
ask_labels_on_launch=False,
)
label = Label.objects.create(name='repro-label', organization=organization)
node = WorkflowJobTemplateNode.objects.create(
workflow_job_template=workflow_job_template,
unified_job_template=jt,
extra_data={'foo': 'bar'},
)
node.labels.add(label)
url = reverse('api:workflow_job_template_node_detail', kwargs={'pk': node.pk})
r = patch(url, {'extra_data': {'foo': 'edited'}}, user=admin_user, expect=200)
assert r.data['extra_data'] == {'foo': 'edited'}
@pytest.mark.django_db
@pytest.mark.parametrize(
"field_name, field_value",