feat: support for oidc credential /test endpoint#16370
Open
fincamd wants to merge 14 commits intoansible:develfrom
Open
feat: support for oidc credential /test endpoint#16370fincamd wants to merge 14 commits intoansible:develfrom
fincamd wants to merge 14 commits intoansible:develfrom
Conversation
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
📝 WalkthroughWalkthroughAdded a workload-identity JWT helper and refactored token retrieval to use it; injected generated workload-identity JWTs into external credential-test endpoints when feature-flagged (with job template loading and permission checks); and filtered out internal credential input fields during serialization. Changes
Sequence DiagramsequenceDiagram
participant Client as Client
participant Handler as TestEndpoint
participant Schema as CredentialSchema
participant JT as JobTemplateSvc
participant Auth as PermissionCheck
participant Claims as ClaimsBuilder
participant JWTSvc as WorkloadJWTSvc
participant Decoder as JWTDecoder
Client->>Handler: POST /credential/test (backend_kwargs incl. job_template_id)
Handler->>Schema: Inspect credential inputs for internal workload_identity_token
Schema-->>Handler: indicates workload_identity_token present
Handler->>JT: Load JobTemplate by id
JT-->>Handler: JobTemplate instance
Handler->>Auth: user.can_access(job_template, "read")
Auth-->>Handler: permission result
Handler->>Claims: Build claims from JobTemplate/org/project
Claims-->>Handler: claims dict
Handler->>JWTSvc: retrieve_workload_identity_jwt_with_claims(claims, aud, scope)
JWTSvc-->>Handler: workload_identity_token
Handler->>Decoder: Decode token (RS256, verification disabled)
Decoder-->>Handler: sent_jwt_payload
Handler->>Client: 202 Accepted + {"details": {"sent_jwt_payload": ...}}
Estimated code review effort🎯 4 (Complex) | ⏱️ ~45 minutes 🚥 Pre-merge checks | ✅ 2 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (2 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches🧪 Generate unit tests (beta)
Comment |
PabloHiro
requested changes
Mar 25, 2026
awx/api/views/__init__.py
Outdated
| ) | ||
| backend_kwargs['workload_identity_token'] = workload_identity_token | ||
| from jwt import decode as _jwt_decode | ||
| response_body['sent_jwt_payload'] = _jwt_decode(workload_identity_token, algorithms=["RS256"], options={"verify_signature": False}) |
Contributor
There was a problem hiding this comment.
In the conversation we had around the PoC we were thinking about nesting sent_jwt_payload under something more generic:
{
"details": {
"sent_jwt_payload": "..."
}
}This pattern is followed some lines above, but not here. Was this intentional?
awx/api/views/__init__.py
Outdated
| return Response({'detail': _('Job template ID is required.')}, status=status.HTTP_400_BAD_REQUEST) | ||
| job_template = models.JobTemplate.objects.get(id=int()) | ||
| if not request.user.can_access(models.JobTemplate, 'read', job_template): | ||
| raise PermissionDenied(_(f'You do not have access to job template with id: {job_template.id}.')) |
Contributor
There was a problem hiding this comment.
What happens in this case? Do we get a 500 HTTP error? Or something else? Shouldn't we be returning another HTTP_400_BAD_REQUEST here, instead of raising an exception?
Contributor
There was a problem hiding this comment.
Contributor
There was a problem hiding this comment.
@fincamd feel free to resolve this conversation if we get a 403 when the user tries to use a Job Template it does not have access to
awx/api/serializers.py
Outdated
Comment on lines
+2945
to
+2955
| # If workload identity token field exists, add job_template to metadata | ||
| if found_wit_field: | ||
| metadata = value['inputs'].get('metadata', []) | ||
| metadata.append({ | ||
| "id": "job_template_id", | ||
| "label": "ID of a Job Template", | ||
| "type": "string", | ||
| "help_text": "Job template ID to use when generating a token." | ||
| }) | ||
| value['inputs']['metadata'] = metadata | ||
|
|
Contributor
There was a problem hiding this comment.
We talked about making this a dynamic input in the UI instead of making it a dynamic field in the backend.
Both solutions are ugly, but my personal take is that the UI workaround is less ugly. Please take a look that the drafted PR linked in AAP-64624 (the corresponding UI change). Thoughts?
Contributor
There was a problem hiding this comment.
Agreed that it's less ugly for the UI to special-case the inclusion of job_template_id when performing a test. This just came up on Slack and I said the same thing 😄 . So I think we can delete 2945-2955
dleehr
suggested changes
Mar 26, 2026
Contributor
There was a problem hiding this comment.
Kudos to @matoval for testing this with the UI. Some early feedback:
- Both
CredentialExternalTestandCredentialTypeExternalTestviews need to implement the new behavior. Right now it's just happening inCredentialExternalTest - Calling the API with a
job_template_idfails because the plugin's backend function doesn't expect this argument. Matthew suggested to.pop()the job template id instead of.get(). I don't remember running into this problem in my PoC 🤔, but we obviously don't want it to fail. - The test operation re-uses
retrieve_workload_identity_jwtfunction from jobs.py to get a JWT, but that's intended for jobs not templates and isn't working. We'll need to do something different there - we can't use that function as-is with job templates, it's meant for jobs.
dleehr
reviewed
Mar 26, 2026
awx/api/views/__init__.py
Outdated
| raise PermissionDenied(_(f'You do not have access to job template with id: {job_template.id}.')) | ||
|
|
||
| # Get a Workload Identity Token | ||
| workload_identity_token = retrieve_workload_identity_jwt( |
Contributor
There was a problem hiding this comment.
Credit to @matoval but this function expects a job, not a job template.
Contributor
Author
fincamd
force-pushed
the
AAP-67727-oidc-vault-creds-testing
branch
from
5b0693e to
8154edc
Compare
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 3
🧹 Nitpick comments (2)
awx/api/views/__init__.py (2)
167-183: Consider defensive null checks for related objects.The function directly accesses
job_template.organization.nameandjob_template.project.name. While these fields are typically required on JobTemplate, adding null checks (similar togetattr_dnepattern used inpopulate_claims_for_workload) would make this more resilient to edge cases.♻️ Optional defensive approach
def _get_workload_identity_token(job_template: models.JobTemplate, jwt_aud: str) -> str: - claims = { - AutomationControllerJobScope.CLAIM_ORGANIZATION_NAME: job_template.organization.name, - AutomationControllerJobScope.CLAIM_ORGANIZATION_ID: job_template.organization.id, - AutomationControllerJobScope.CLAIM_PROJECT_NAME: job_template.project.name, - AutomationControllerJobScope.CLAIM_PROJECT_ID: job_template.project.id, - AutomationControllerJobScope.CLAIM_JOB_TEMPLATE_NAME: job_template.name, - AutomationControllerJobScope.CLAIM_JOB_TEMPLATE_ID: job_template.id, - AutomationControllerJobScope.CLAIM_PLAYBOOK_NAME: job_template.playbook, - } + claims = { + AutomationControllerJobScope.CLAIM_JOB_TEMPLATE_NAME: job_template.name, + AutomationControllerJobScope.CLAIM_JOB_TEMPLATE_ID: job_template.id, + AutomationControllerJobScope.CLAIM_PLAYBOOK_NAME: job_template.playbook, + } + if job_template.organization: + claims[AutomationControllerJobScope.CLAIM_ORGANIZATION_NAME] = job_template.organization.name + claims[AutomationControllerJobScope.CLAIM_ORGANIZATION_ID] = job_template.organization.id + if job_template.project: + claims[AutomationControllerJobScope.CLAIM_PROJECT_NAME] = job_template.project.name + claims[AutomationControllerJobScope.CLAIM_PROJECT_ID] = job_template.project.id🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@awx/api/views/__init__.py` around lines 167 - 183, The function _get_workload_identity_token accesses nested attributes on job_template (job_template.organization.name, .id, job_template.project.name, .id, etc.) without null checks; make these accesses defensive by using the existing getattr_dne helper (as used in populate_claims_for_workload) or explicit None-safe lookups to obtain organization name/id, project name/id, playbook, and job_template fields before building claims, and fall back to empty strings or safe defaults so retrieve_workload_identity_jwt_with_claims always receives defined claim values.
1644-1665: Consider extracting duplicated OIDC handling logic.The OIDC workload identity token logic is nearly identical between
CredentialExternalTestandCredentialTypeExternalTest. This duplication could lead to maintenance burden if changes are needed later.♻️ Suggested helper extraction
def _handle_oidc_workload_identity(request, fields, backend_kwargs): """ Handle OIDC workload identity token generation for credential test endpoints. Returns (response_body_update, error_response) tuple. """ if not flag_enabled('FEATURE_OIDC_WORKLOAD_IDENTITY_ENABLED'): return {}, None for field in fields: if field.get('internal') and field.get('id') == 'workload_identity_token': job_template_id = backend_kwargs.get('job_template_id') if job_template_id is None: return None, Response({'detail': _('Job template ID is required.')}, status=status.HTTP_400_BAD_REQUEST) try: job_template = models.JobTemplate.objects.get(id=int(job_template_id)) except models.JobTemplate.DoesNotExist: return None, Response({'detail': _('Job template not found.')}, status=status.HTTP_400_BAD_REQUEST) if not request.user.can_access(models.JobTemplate, 'read', job_template): raise PermissionDenied(_('You do not have access to job template with id: %(id)s.') % {'id': job_template.id}) backend_kwargs['workload_identity_token'] = _get_workload_identity_token(job_template, backend_kwargs.get('jwt_aud')) from jwt import decode as _jwt_decode return { 'details': { 'sent_jwt_payload': _jwt_decode(backend_kwargs['workload_identity_token'], algorithms=["RS256"], options={"verify_signature": False}), } }, None return {}, NoneAlso applies to: 1729-1750
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@awx/api/views/__init__.py` around lines 1644 - 1665, Extract the duplicated OIDC workload identity token logic from CredentialExternalTest and CredentialTypeExternalTest into a single helper (e.g. _handle_oidc_workload_identity) that takes (request, fields, backend_kwargs) and returns (response_body_update, error_response). Inside the helper check flag_enabled('FEATURE_OIDC_WORKLOAD_IDENTITY_ENABLED'), iterate fields for an internal id == 'workload_identity_token', validate job_template_id exists in backend_kwargs, catch models.JobTemplate.DoesNotExist and return a 400 Response, verify request.user.can_access(models.JobTemplate, 'read', job_template) (raise PermissionDenied if not), set backend_kwargs['workload_identity_token'] using _get_workload_identity_token(job_template, backend_kwargs.get('jwt_aud')) and return the decoded sent_jwt_payload for the response_body; update both callers (CredentialExternalTest and CredentialTypeExternalTest) to call the helper and short-circuit on error_response, merging response_body_update into their response_body.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@awx/api/serializers.py`:
- Around line 2936-2942: The code assumes value['inputs'] is a dict and calls
.get on it, which breaks for inputs=None or skips deepcopy for inputs={} ;
update the logic around value['inputs'] so you first normalize/guard it (e.g.,
if not isinstance(value.get('inputs'), dict): value['inputs'] = {}), then
perform copy.deepcopy on that dict to avoid mutating the original, and finally
compute fields = value['inputs'].get('fields', []) before filtering internal
fields with value['inputs']['fields'] = [f for f in fields if not
f.get('internal')]; ensure the normalization and deepcopy happen before any .get
calls on value['inputs'].
In `@awx/api/views/__init__.py`:
- Around line 1741-1743: Wrap the JobTemplate lookup in a try/except to catch
models.JobTemplate.DoesNotExist and raise an appropriate 404 (e.g., NotFound)
with a translated message that includes the id, and avoid using an f-string
inside the translation call; after retrieving job_template via
models.JobTemplate.objects.get(id=int(job_template_id)) handle the exception and
when checking access use request.user.can_access(...) and, if denied, call
PermissionDenied with a translation string created either by formatting outside
the _() call or by using a translation placeholder (e.g., _('You do not have
access to job template with id: %(id)s.') % {'id': job_template.id}) so the
translation pattern is correct.
- Around line 1656-1658: Wrap the JobTemplate lookup
(models.JobTemplate.objects.get(id=int(job_template_id))) in a try/except that
catches models.JobTemplate.DoesNotExist (or ObjectDoesNotExist) and raise a
BadRequest (400) with a clear message using the raw job_template_id; after a
successful get, keep the permission check but change the translation pattern to
use a translatable string with interpolation (e.g. _('You do not have access to
job template with id: %(id)s.') % {'id': job_template.id}) so the message is
translated correctly; refer to the job_template lookup and the permission check
around job_template = models.JobTemplate.objects.get(...) and
request.user.can_access(...).
---
Nitpick comments:
In `@awx/api/views/__init__.py`:
- Around line 167-183: The function _get_workload_identity_token accesses nested
attributes on job_template (job_template.organization.name, .id,
job_template.project.name, .id, etc.) without null checks; make these accesses
defensive by using the existing getattr_dne helper (as used in
populate_claims_for_workload) or explicit None-safe lookups to obtain
organization name/id, project name/id, playbook, and job_template fields before
building claims, and fall back to empty strings or safe defaults so
retrieve_workload_identity_jwt_with_claims always receives defined claim values.
- Around line 1644-1665: Extract the duplicated OIDC workload identity token
logic from CredentialExternalTest and CredentialTypeExternalTest into a single
helper (e.g. _handle_oidc_workload_identity) that takes (request, fields,
backend_kwargs) and returns (response_body_update, error_response). Inside the
helper check flag_enabled('FEATURE_OIDC_WORKLOAD_IDENTITY_ENABLED'), iterate
fields for an internal id == 'workload_identity_token', validate job_template_id
exists in backend_kwargs, catch models.JobTemplate.DoesNotExist and return a 400
Response, verify request.user.can_access(models.JobTemplate, 'read',
job_template) (raise PermissionDenied if not), set
backend_kwargs['workload_identity_token'] using
_get_workload_identity_token(job_template, backend_kwargs.get('jwt_aud')) and
return the decoded sent_jwt_payload for the response_body; update both callers
(CredentialExternalTest and CredentialTypeExternalTest) to call the helper and
short-circuit on error_response, merging response_body_update into their
response_body.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 897d0054-153e-4d13-92a6-b2578217a10a
📒 Files selected for processing (3)
awx/api/serializers.pyawx/api/views/__init__.pyawx/main/tasks/jobs.py
PabloHiro
approved these changes
Mar 30, 2026
Contributor
PabloHiro
left a comment
PabloHiro
left a comment
There was a problem hiding this comment.
LGTM, please address CI checks 🙏
credential external test plugin will gather known claims to perform the test operation over vault by using OIDC auth
fincamd
force-pushed
the
AAP-67727-oidc-vault-creds-testing
branch
from
e0cc026 to
8926db5
Compare
There was a problem hiding this comment.
Actionable comments posted: 1
♻️ Duplicate comments (1)
awx/api/views/__init__.py (1)
1653-1658:⚠️ Potential issue | 🟠 MajorValidate
job_template_idbefore dereferencing it.
int(job_template_id)andJobTemplate.objects.get()can both fail here, and this happens before the backend call is guarded, so malformed or unknown IDs currently surface as 500s in both/testendpoints. This block is duplicated too, so a small helper would keep the two paths aligned.Suggested fix pattern
if job_template_id is None: return Response({'detail': _('Job template ID is required.')}, status=status.HTTP_400_BAD_REQUEST) - job_template = models.JobTemplate.objects.get(id=int(job_template_id)) + try: + job_template = models.JobTemplate.objects.get(id=int(job_template_id)) + except (TypeError, ValueError, models.JobTemplate.DoesNotExist): + return Response({'detail': _('Invalid job template ID.')}, status=status.HTTP_400_BAD_REQUEST) if not request.user.can_access(models.JobTemplate, 'read', job_template): raise PermissionDenied(...)Also applies to: 1738-1743
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@awx/api/views/__init__.py` around lines 1653 - 1658, The code dereferences job_template_id and calls models.JobTemplate.objects.get without validating/handling bad input, causing 500s; create a small helper (e.g., validate_or_fetch_job_template or _get_job_template_from_kwargs) used in both places to: extract job_template_id from backend_kwargs, attempt to coerce to int (catch ValueError and return a 400 Response with a clear message), attempt to fetch models.JobTemplate (catch models.JobTemplate.DoesNotExist and return a 404 Response), and then perform the existing permission check using request.user.can_access (raise PermissionDenied as before); replace the duplicated blocks at the current location and at the other occurrence (around lines 1738-1743) with calls to this helper.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Duplicate comments:
In `@awx/api/views/__init__.py`:
- Around line 1653-1658: The code dereferences job_template_id and calls
models.JobTemplate.objects.get without validating/handling bad input, causing
500s; create a small helper (e.g., validate_or_fetch_job_template or
_get_job_template_from_kwargs) used in both places to: extract job_template_id
from backend_kwargs, attempt to coerce to int (catch ValueError and return a 400
Response with a clear message), attempt to fetch models.JobTemplate (catch
models.JobTemplate.DoesNotExist and return a 404 Response), and then perform the
existing permission check using request.user.can_access (raise PermissionDenied
as before); replace the duplicated blocks at the current location and at the
other occurrence (around lines 1738-1743) with calls to this helper.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 7d2270ea-9c12-4d5b-b9ec-b0f443e84efe
📒 Files selected for processing (3)
awx/api/serializers.pyawx/api/views/__init__.pyawx/main/tasks/jobs.py
✅ Files skipped from review due to trivial changes (1)
- awx/api/serializers.py
🚧 Files skipped from review as they are similar to previous changes (1)
- awx/main/tasks/jobs.py
There was a problem hiding this comment.
♻️ Duplicate comments (2)
awx/api/views/__init__.py (2)
1657-1662:⚠️ Potential issue | 🟠 MajorHandle bad
job_template_idinputs in both OIDC test paths.
int(job_template_id)andJobTemplate.objects.get()still turn malformed or unknown IDs into 500s here, and_(f"...")won't translate correctly. CatchTypeError/ValueError/DoesNotExistonce in a shared helper, then use placeholder interpolation for the permission message in both endpoints.Example patch for one branch
- job_template = models.JobTemplate.objects.get(id=int(job_template_id)) + try: + job_template = models.JobTemplate.objects.get(id=int(job_template_id)) + except (TypeError, ValueError, models.JobTemplate.DoesNotExist): + return Response({'detail': _('Job template not found.')}, status=status.HTTP_400_BAD_REQUEST) if not request.user.can_access(models.JobTemplate, 'read', job_template): - raise PermissionDenied(_(f'You do not have access to job template with id: {job_template.id}.')) + raise PermissionDenied(_('You do not have access to job template with id: %(id)s.') % {'id': job_template.id})Verification: confirm both duplicated branches still use the raw lookup and f-string translation path.
#!/bin/bash rg -n -C2 "JobTemplate\.objects\.get\(id=int\(job_template_id\)\)|PermissionDenied\(_\(f'You do not have access to job template with id:" awx/api/views/__init__.pyAlso applies to: 1745-1750
167-182:⚠️ Potential issue | 🟠 MajorDon't let workload-identity setup failures escape as 500s.
retrieve_workload_identity_jwt_with_claims()inawx/main/tasks/jobs.py:161-177explicitly raisesRuntimeErrorwhen the workload identity client is not configured. Because_get_workload_identity_token()and the JWT decode both run before either endpoint's existingtry,/teststill returns a 500 instead of a controlled API error.Also applies to: 1664-1668, 1752-1756
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@awx/api/views/__init__.py` around lines 167 - 182, The workload-identity client failure currently raises a RuntimeError from retrieve_workload_identity_jwt_with_claims and bubbles out as a 500; catch RuntimeError in _get_workload_identity_token (and any places that call it before their existing try blocks) and convert it into a controlled API-level exception (e.g., raise rest_framework.exceptions.APIException or ValidationError with a clear message) so endpoints like the /test path return a proper API error instead of a 500; specifically wrap the call to retrieve_workload_identity_jwt_with_claims in _get_workload_identity_token and re-raise a DRF exception, and also add similar try/except around any pre-try JWT decode callers to ensure consistent API error handling.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Duplicate comments:
In `@awx/api/views/__init__.py`:
- Around line 167-182: The workload-identity client failure currently raises a
RuntimeError from retrieve_workload_identity_jwt_with_claims and bubbles out as
a 500; catch RuntimeError in _get_workload_identity_token (and any places that
call it before their existing try blocks) and convert it into a controlled
API-level exception (e.g., raise rest_framework.exceptions.APIException or
ValidationError with a clear message) so endpoints like the /test path return a
proper API error instead of a 500; specifically wrap the call to
retrieve_workload_identity_jwt_with_claims in _get_workload_identity_token and
re-raise a DRF exception, and also add similar try/except around any pre-try JWT
decode callers to ensure consistent API error handling.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 15b93346-9f9b-4fe7-bb91-e57a3cbfde8a
📒 Files selected for processing (1)
awx/api/views/__init__.py
dleehr
suggested changes
Mar 31, 2026
Contributor
dleehr
left a comment
dleehr
left a comment
There was a problem hiding this comment.
Requesting changes primarily because there are no tests.
I left a fair bit of other feedback too. Please explore how to refactor the implementation that is duplicated across two views, address the error handling feedback and other items. The feedback from coderabbit looks sound too.
Also, please clarify what the behavior is with get vs pop. Does one fail?
awx/api/views/__init__.py
Outdated
Comment on lines
+63
to
+64
| from awx.main.tasks.jobs import AutomationControllerJobScope, retrieve_workload_identity_jwt_with_claims | ||
| from awx.main.tasks.system import flag_enabled, send_notifications, update_inventory_computed_fields |
Contributor
There was a problem hiding this comment.
I find these imports from awx.main.tasks.jobs are a bit confusing
First, AutomationControllerJobScope is defined in DAB, why do we import it from awx.main.tasks.jobs?
Second, I do like how you refactored retrieve_workload_identity_jwt_with_claims to support both this test operation and job execution 👍. But maybe that function should be in some common awx module rather than awx.main.tasks.jobs. This one's not critical but please have a look.
Contributor
Author
There was a problem hiding this comment.
Modified that import so it draws from dab, not jobs.py (I must have had Vscode autoimport, I never checked back) Done here: 86c0364
Contributor
Author
There was a problem hiding this comment.
Second part, moving that helper function to a separate shared file: 149773a
awx/api/views/__init__.py
Outdated
| raise PermissionDenied(_(f'You do not have access to job template with id: {job_template.id}.')) | ||
|
|
||
| backend_kwargs['workload_identity_token'] = _get_workload_identity_token(job_template, backend_kwargs.get('jwt_aud')) | ||
| from jwt import decode as _jwt_decode |
Contributor
There was a problem hiding this comment.
Can we move this import to the top of the file? I don't think there's a need for it to be nested.
Comment on lines
+1736
to
+1758
|
|
||
| # Add extra test functionality for OIDC-enabled credential types | ||
| response_body = {} | ||
| if flag_enabled('FEATURE_OIDC_WORKLOAD_IDENTITY_ENABLED'): | ||
| # Get a Workload Identity Token if credential contains an internal 'workload_identity_token' field | ||
| fields = obj.inputs.get('fields', []) | ||
| for field in fields: | ||
| if field.get('internal') and field.get('id') == 'workload_identity_token': | ||
| # Make sure that the requesting user has access to the job template | ||
| job_template_id = backend_kwargs.get('job_template_id') | ||
| if job_template_id is None: | ||
| return Response({'detail': _('Job template ID is required.')}, status=status.HTTP_400_BAD_REQUEST) | ||
| job_template = models.JobTemplate.objects.get(id=int(job_template_id)) | ||
| if not request.user.can_access(models.JobTemplate, 'read', job_template): | ||
| raise PermissionDenied(_(f'You do not have access to job template with id: {job_template.id}.')) | ||
|
|
||
| backend_kwargs['workload_identity_token'] = _get_workload_identity_token(job_template, backend_kwargs.get('jwt_aud')) | ||
| from jwt import decode as _jwt_decode | ||
|
|
||
| response_body['details'] = { | ||
| 'sent_jwt_payload': _jwt_decode(backend_kwargs['workload_identity_token'], algorithms=["RS256"], options={"verify_signature": False}), | ||
| } | ||
|
|
Contributor
There was a problem hiding this comment.
This code is duplicated in both view implementations. Can we refactor this so we don't have two copies of the same thing?
Member
There was a problem hiding this comment.
Refactored into helper function.
awx/api/views/__init__.py
Outdated
| return Response({'detail': _('Job template ID is required.')}, status=status.HTTP_400_BAD_REQUEST) | ||
| job_template = models.JobTemplate.objects.get(id=int(job_template_id)) | ||
| if not request.user.can_access(models.JobTemplate, 'read', job_template): | ||
| raise PermissionDenied(_(f'You do not have access to job template with id: {job_template.id}.')) |
Contributor
There was a problem hiding this comment.
This use of _() (translation lookup) is invalid - we need to look up the format string without interpolating the id variable first:
# Wrong — string interpolated before translation lookup
raise PermissionDenied(_(f'You do not have access to job template with id: {job_template.id}.'))
# Correct
raise PermissionDenied(_('You do not have access to job template with id: %(id)s.') % {'id': job_template.id})
Contributor
There was a problem hiding this comment.
The bot called this out too. Please have a look.
awx/api/views/__init__.py
Outdated
| for field in fields: | ||
| if field.get('internal') and field.get('id') == 'workload_identity_token': | ||
| # Make sure that the requesting user has access to the job template | ||
| job_template_id = backend_kwargs.get('job_template_id') |
Contributor
There was a problem hiding this comment.
I relayed some earlier feedback from @matoval that if we leave the job_template_id in the backend_kwargs dict when we invoke the plugin's backend() function, it fails.
He recommended using .pop() instead of .get(), since pop will remove the value from the dictionary.
What was the resolution on this feedback? Did you test it and find this was not necessary?
awx/api/views/__init__.py
Outdated
| job_template_id = backend_kwargs.get('job_template_id') | ||
| if job_template_id is None: | ||
| return Response({'detail': _('Job template ID is required.')}, status=status.HTTP_400_BAD_REQUEST) | ||
| job_template = models.JobTemplate.objects.get(id=int(job_template_id)) |
Contributor
There was a problem hiding this comment.
This has a couple more failure modes - how are we handling those?
- Since we're not using a serializer to validate the incoming data, the incoming
job_template_idmight not be an integer, and we may getValueErrorwhen callingint()on it. That should return 400 bad request - If the ID references a template that does not exist, that should either return 400 bad request or 404 not found, but I assume it returns a 500 right now.
Member
There was a problem hiding this comment.
Updated to add error handling.
There was a problem hiding this comment.
Actionable comments posted: 1
♻️ Duplicate comments (1)
awx/api/views/__init__.py (1)
1746-1753:⚠️ Potential issue | 🟠 MajorSame issues as CredentialExternalTest: unhandled exceptions and invalid translation pattern.
This code block has the same issues flagged for
CredentialExternalTest:
- Line 1749: Missing
ValueError/TypeErrorhandling forint()conversion- Line 1749: Missing
DoesNotExisthandling forJobTemplate.objects.get()- Line 1751: f-string inside
_()breaks translations- Line 1753:
RuntimeErrorfrom_get_workload_identity_token()not handled- Line 1746: Should use
.pop()instead of.get()Apply the same fixes as proposed for
CredentialExternalTest.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@awx/api/views/__init__.py` around lines 1746 - 1753, Replace backend_kwargs.get('job_template_id') with backend_kwargs.pop('job_template_id', None) and validate it; wrap int(job_template_id) in try/except catching ValueError/TypeError and return a 400 Response on bad int conversion. Wrap models.JobTemplate.objects.get(id=...) in try/except catching models.JobTemplate.DoesNotExist and return a 404/400 Response as appropriate. Change the translation call to avoid f-strings, e.g. raise PermissionDenied(_('You do not have access to job template with id: %s.') % job_template.id). Finally, call _get_workload_identity_token(job_template, backend_kwargs.get('jwt_aud')) inside a try/except that catches RuntimeError and returns a 400/500 Response with the error message, and only set backend_kwargs['workload_identity_token'] when it succeeds.
🧹 Nitpick comments (2)
awx/api/views/__init__.py (2)
1666-1666: Move import to top of file.The
jwtimport should be at the top of the file with other imports, not nested inside the function. This was flagged in a previous review.♻️ Proposed refactor
Add this import near the top of the file with other imports:
from jwt import decode as jwt_decodeThen update line 1669 to use
jwt_decodedirectly:- from jwt import decode as _jwt_decode - response_body['details'] = { - 'sent_jwt_payload': _jwt_decode(backend_kwargs['workload_identity_token'], algorithms=["RS256"], options={"verify_signature": False}), + 'sent_jwt_payload': jwt_decode(backend_kwargs['workload_identity_token'], algorithms=["RS256"], options={"verify_signature": False}), }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@awx/api/views/__init__.py` at line 1666, Move the nested import "from jwt import decode as _jwt_decode" out of the function and place "from jwt import decode as jwt_decode" with the module's other imports at the top of the file, then update all local usages of _jwt_decode to jwt_decode (e.g., the call that currently references _jwt_decode) so the decode function is imported once at module load time and used consistently.
1740-1758: Refactor to eliminate code duplication.This OIDC workload identity block (lines 1740-1758) is nearly identical to the one in
CredentialExternalTest.post(lines 1652-1670). Consider extracting this into a shared helper function to avoid maintaining duplicate logic.♻️ Proposed refactor
Extract to a helper function at module level:
def _prepare_workload_identity_for_test(request, backend_kwargs, fields): """ Prepare workload identity token for credential test endpoints. Returns (response_body_details, error_response) where error_response is None on success. """ if not flag_enabled('FEATURE_OIDC_WORKLOAD_IDENTITY_ENABLED'): return None, None for field in fields: if field.get('internal') and field.get('id') == 'workload_identity_token': job_template_id = backend_kwargs.pop('job_template_id', None) if job_template_id is None: return None, Response({'detail': _('Job template ID is required.')}, status=status.HTTP_400_BAD_REQUEST) try: job_template = models.JobTemplate.objects.get(id=int(job_template_id)) except (ValueError, TypeError): return None, Response({'detail': _('Job template ID must be a valid integer.')}, status=status.HTTP_400_BAD_REQUEST) except models.JobTemplate.DoesNotExist: return None, Response({'detail': _('Job template not found.')}, status=status.HTTP_400_BAD_REQUEST) if not request.user.can_access(models.JobTemplate, 'read', job_template): raise PermissionDenied(_('You do not have access to job template with id: %(id)s.') % {'id': job_template.id}) try: backend_kwargs['workload_identity_token'] = _get_workload_identity_token(job_template, backend_kwargs.get('jwt_aud')) except RuntimeError as exc: return None, Response({'detail': str(exc)}, status=status.HTTP_400_BAD_REQUEST) return { 'sent_jwt_payload': jwt_decode(backend_kwargs['workload_identity_token'], algorithms=["RS256"], options={"verify_signature": False}), }, None return None, NoneThen call from both views:
details, error_response = _prepare_workload_identity_for_test(request, backend_kwargs, fields) if error_response: return error_response if details: response_body['details'] = details🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@awx/api/views/__init__.py` around lines 1740 - 1758, The OIDC workload identity handling is duplicated; extract it into a shared module-level helper (e.g., _prepare_workload_identity_for_test) that takes (request, backend_kwargs, fields) and returns (details_dict, error_response) where error_response is a DRF Response on validation/failure or None on success. Implement the helper to check flag_enabled('FEATURE_OIDC_WORKLOAD_IDENTITY_ENABLED'), find the internal field with id 'workload_identity_token', validate and coerce job_template_id (handle missing/invalid int and JobTemplate.DoesNotExist), enforce request.user.can_access(models.JobTemplate, 'read', job_template) (raise PermissionDenied if unauthorized), call _get_workload_identity_token(job_template, backend_kwargs.get('jwt_aud')) (catch and convert runtime errors to a 400 Response), populate backend_kwargs['workload_identity_token'], build sent_jwt_payload using jwt.decode/options as before, and return (details, None); then call this helper from both the current view block and CredentialExternalTest.post, returning error_response if present and setting response_body['details'] when details is returned.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@awx/api/views/__init__.py`:
- Around line 1658-1665: Validate and remove job_template_id from backend_kwargs
and handle all exceptions: attempt to convert
backend_kwargs.pop('job_template_id') to int inside a try/except and return a
Response with status 400 if ValueError; wrap
models.JobTemplate.objects.get(id=...) in try/except to catch
models.JobTemplate.DoesNotExist and return a 404 Response; replace the f-string
translation with a translation-safe call like _('You do not have access to job
template with id: %(id)s.') % {'id': job_template.id} when raising
PermissionDenied; call _get_workload_identity_token(job_template,
backend_kwargs.get('jwt_aud')) inside a try/except that catches RuntimeError and
returns a 400 Response with the error message; ensure you reference
job_template_id, backend_kwargs.pop, models.JobTemplate.objects.get,
request.user.can_access, _get_workload_identity_token, Response, and
PermissionDenied when making these changes.
---
Duplicate comments:
In `@awx/api/views/__init__.py`:
- Around line 1746-1753: Replace backend_kwargs.get('job_template_id') with
backend_kwargs.pop('job_template_id', None) and validate it; wrap
int(job_template_id) in try/except catching ValueError/TypeError and return a
400 Response on bad int conversion. Wrap models.JobTemplate.objects.get(id=...)
in try/except catching models.JobTemplate.DoesNotExist and return a 404/400
Response as appropriate. Change the translation call to avoid f-strings, e.g.
raise PermissionDenied(_('You do not have access to job template with id: %s.')
% job_template.id). Finally, call _get_workload_identity_token(job_template,
backend_kwargs.get('jwt_aud')) inside a try/except that catches RuntimeError and
returns a 400/500 Response with the error message, and only set
backend_kwargs['workload_identity_token'] when it succeeds.
---
Nitpick comments:
In `@awx/api/views/__init__.py`:
- Line 1666: Move the nested import "from jwt import decode as _jwt_decode" out
of the function and place "from jwt import decode as jwt_decode" with the
module's other imports at the top of the file, then update all local usages of
_jwt_decode to jwt_decode (e.g., the call that currently references _jwt_decode)
so the decode function is imported once at module load time and used
consistently.
- Around line 1740-1758: The OIDC workload identity handling is duplicated;
extract it into a shared module-level helper (e.g.,
_prepare_workload_identity_for_test) that takes (request, backend_kwargs,
fields) and returns (details_dict, error_response) where error_response is a DRF
Response on validation/failure or None on success. Implement the helper to check
flag_enabled('FEATURE_OIDC_WORKLOAD_IDENTITY_ENABLED'), find the internal field
with id 'workload_identity_token', validate and coerce job_template_id (handle
missing/invalid int and JobTemplate.DoesNotExist), enforce
request.user.can_access(models.JobTemplate, 'read', job_template) (raise
PermissionDenied if unauthorized), call
_get_workload_identity_token(job_template, backend_kwargs.get('jwt_aud')) (catch
and convert runtime errors to a 400 Response), populate
backend_kwargs['workload_identity_token'], build sent_jwt_payload using
jwt.decode/options as before, and return (details, None); then call this helper
from both the current view block and CredentialExternalTest.post, returning
error_response if present and setting response_body['details'] when details is
returned.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 4f3230ae-5bd9-46d2-acef-e2d28721f82c
📒 Files selected for processing (3)
awx/api/views/__init__.pyawx/main/tasks/jobs.pyawx/main/utils/workload_identity.py
There was a problem hiding this comment.
Actionable comments posted: 1
♻️ Duplicate comments (1)
awx/api/views/__init__.py (1)
1746-1789:⚠️ Potential issue | 🔴 CriticalSame critical issues as
CredentialExternalTest:KeyErrorand unhandled exceptions.This code block has identical issues:
- Lines 1757, 1768, 1780, 1788:
response_body['details']['error_message']will raiseKeyError.- Line 1755: Use
.pop()instead of.get().- Line 1759: Missing exception handling for
ValueErrorandDoesNotExist.- Line 1761: f-string inside
_()breaks translations.As noted in past reviews, this logic is duplicated between both endpoints. Consider extracting the OIDC workload identity handling into a shared helper function to avoid maintaining two copies.
🐛 Proposed fix (same pattern as above)
- job_template_id = backend_kwargs.get('job_template_id') + job_template_id = backend_kwargs.pop('job_template_id', None) if job_template_id is None: - response_body['details']['error_message'] = _('Job template ID is required.') - return Response(response_body, status=status.HTTP_400_BAD_REQUEST) - job_template = models.JobTemplate.objects.get(id=int(job_template_id)) + return Response({'details': {'error_message': _('Job template ID is required.')}}, status=status.HTTP_400_BAD_REQUEST) + try: + job_template = models.JobTemplate.objects.get(id=int(job_template_id)) + except (ValueError, TypeError): + return Response({'details': {'error_message': _('Job template ID must be a valid integer.')}}, status=status.HTTP_400_BAD_REQUEST) + except models.JobTemplate.DoesNotExist: + return Response({'details': {'error_message': _('Job template not found.')}}, status=status.HTTP_400_BAD_REQUEST) if not request.user.can_access(models.JobTemplate, 'read', job_template): - raise PermissionDenied(_(f'You do not have access to job template with id: {job_template.id}.')) + raise PermissionDenied(_('You do not have access to job template with id: %(id)s.') % {'id': job_template.id})🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@awx/api/views/__init__.py` around lines 1746 - 1789, Initialize response_body['details'] before writing into it and replace backend_kwargs.get('job_template_id') with backend_kwargs.pop('job_template_id', None); validate and convert the popped value inside a try/except catching ValueError and models.JobTemplate.DoesNotExist and return a 400 with a clear error message; when checking permission use a translation-safe string (e.g. _('You do not have access to job template with id: %(id)s.') % {'id': job_template.id}) instead of an f-string inside _(); wrap the models.JobTemplate.objects.get(...) call and int conversion in the same try/except and handle RuntimeError from _get_workload_identity_token as before; finally consider extracting the OIDC handling into a helper (e.g. _handle_workload_identity_token that uses _get_workload_identity_token and _jwt_decode) and call it from both places to remove duplication and centralize error handling involving response_body, backend_kwargs, and request.user.can_access.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@awx/api/views/__init__.py`:
- Around line 1650-1699: response_body is used before 'details' exists and
backend_kwargs wrongly retains job_template_id and lacks error handling;
initialize response_body['details'] = {} early, replace
backend_kwargs.get('job_template_id') with backend_kwargs.pop('job_template_id',
None) when extracting job_template_id, add try/except around converting/getting
JobTemplate to catch ValueError and models.JobTemplate.DoesNotExist and set
response_body['details']['error_message'] accordingly, avoid using an f-string
inside _() for the permission message (use string concatenation or format with
_()), and ensure every except block (including the RuntimeError path from
_get_workload_identity_token and the generic Exception handler) writes into
response_body['details']['error_message'] only after confirming
response_body['details'] exists so the endpoint cannot raise KeyError before
returning.
---
Duplicate comments:
In `@awx/api/views/__init__.py`:
- Around line 1746-1789: Initialize response_body['details'] before writing into
it and replace backend_kwargs.get('job_template_id') with
backend_kwargs.pop('job_template_id', None); validate and convert the popped
value inside a try/except catching ValueError and
models.JobTemplate.DoesNotExist and return a 400 with a clear error message;
when checking permission use a translation-safe string (e.g. _('You do not have
access to job template with id: %(id)s.') % {'id': job_template.id}) instead of
an f-string inside _(); wrap the models.JobTemplate.objects.get(...) call and
int conversion in the same try/except and handle RuntimeError from
_get_workload_identity_token as before; finally consider extracting the OIDC
handling into a helper (e.g. _handle_workload_identity_token that uses
_get_workload_identity_token and _jwt_decode) and call it from both places to
remove duplication and centralize error handling involving response_body,
backend_kwargs, and request.user.can_access.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: a53fb747-69a8-4e0a-b763-5628bdb6887a
📒 Files selected for processing (1)
awx/api/views/__init__.py
- Fix KeyError when building error responses (8 occurrences) - Use .pop() to prevent passing job_template_id/jwt_aud to backend - Add exception handling for ValueError and JobTemplate.DoesNotExist - Fix Django translation pattern (use % formatting instead of f-strings) - Refactor duplicated OIDC logic into _handle_oidc_credential_test()
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
4 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
SUMMARY
This PR addresses the credential test functionality for existing and new credentials when connecting to OIDC secret providers.
ISSUE TYPE
COMPONENT NAME
STEPS TO REPRODUCE AND EXTRA INFO
Summary by CodeRabbit
New Features
API Changes
Bug Fixes