Validate the webhook credential

- we should allow a null credential, so that the admin can choose to configure not posting back status changes of the triggered job
- the credential must be of the new 'token' kind
- if we do configure a credential, its type must match the selected SCM service

awx/api/serializers.py

@@ -2827,6 +2827,23 @@ class JobTemplateMixin(object):
         d['recent_jobs'] = self._recent_jobs(obj)
         return d
 
+    def validate(self, attrs):
+        webhook_service = attrs.get('webhook_service', getattr(self.instance, 'webhook_service', None))
+        webhook_credential = attrs.get('webhook_credential', getattr(self.instance, 'webhook_credential', None))
+
+        if webhook_credential and webhook_credential.credential_type.kind != 'token':
+            raise serializers.ValidationError({
+                'webhook_credential': _("Must be a Personal Access Token."),
+            })
+
+        if webhook_service and webhook_credential:
+            if webhook_credential.kind != '{}_token'.format(webhook_service):
+                raise serializers.ValidationError({
+                    'webhook_credential': _("Must match the selected webhook service."),
+                })
+
+        return super().validate(attrs)
+
 
 class JobTemplateSerializer(JobTemplateMixin, UnifiedJobTemplateSerializer, JobOptionsSerializer):
     show_capabilities = ['start', 'schedule', 'copy', 'edit', 'delete']
@@ -2894,7 +2911,6 @@ class JobTemplateSerializer(JobTemplateMixin, UnifiedJobTemplateSerializer, JobO
     def validate_extra_vars(self, value):
         return vars_validate_or_raise(value)
 
-
     def get_summary_fields(self, obj):
         summary_fields = super(JobTemplateSerializer, self).get_summary_fields(obj)
         all_creds = []

awx/main/tests/functional/api/test_webhooks.py

@@ -2,6 +2,7 @@ import pytest
 
 from awx.api.versioning import reverse
 from awx.main.models.mixins import WebhookMixin
+from awx.main.models.credential import Credential, CredentialType
 
 
 @pytest.mark.django_db
@@ -146,3 +147,56 @@ def test_unset_webhook_service(organization_factory, job_template_factory, patch
     jt.refresh_from_db()
 
     assert (jt.webhook_service, jt.webhook_key) == ('', '')
+
+
+@pytest.mark.django_db
+@pytest.mark.parametrize(
+    "service", [s for s, _ in WebhookMixin.SERVICES]
+)
+def test_set_webhook_credential(organization_factory, job_template_factory, patch, service):
+    objs = organization_factory("org", superusers=['admin'])
+    jt = job_template_factory("jt", organization=objs.organization, webhook_service=service,
+                              inventory='test_inv', project='test_proj').job_template
+    admin = objs.superusers.admin
+    assert jt.webhook_service == service
+    assert jt.webhook_key != ''
+
+    cred_type = CredentialType.defaults['{}_token'.format(service)]()
+    cred_type.save()
+    cred = Credential.objects.create(credential_type=cred_type, name='test-cred',
+                                     inputs={'token': 'secret'})
+
+    url = reverse('api:job_template_detail', kwargs={'pk': jt.pk})
+    patch(url, {'webhook_credential': cred.pk}, user=admin, expect=200)
+    jt.refresh_from_db()
+
+    assert jt.webhook_service == service
+    assert jt.webhook_key != ''
+    assert jt.webhook_credential == cred
+
+
+@pytest.mark.django_db
+@pytest.mark.parametrize(
+    "service,token", [(s, WebhookMixin.SERVICES[i - 1][0]) for i, (s, _) in enumerate(WebhookMixin.SERVICES)]
+)
+def test_set_wrong_service_webhook_credential(organization_factory, job_template_factory, patch, service, token):
+    objs = organization_factory("org", superusers=['admin'])
+    jt = job_template_factory("jt", organization=objs.organization, webhook_service=service,
+                              inventory='test_inv', project='test_proj').job_template
+    admin = objs.superusers.admin
+    assert jt.webhook_service == service
+    assert jt.webhook_key != ''
+
+    cred_type = CredentialType.defaults['{}_token'.format(token)]()
+    cred_type.save()
+    cred = Credential.objects.create(credential_type=cred_type, name='test-cred',
+                                     inputs={'token': 'secret'})
+
+    url = reverse('api:job_template_detail', kwargs={'pk': jt.pk})
+    response = patch(url, {'webhook_credential': cred.pk}, user=admin, expect=400)
+    jt.refresh_from_db()
+
+    assert jt.webhook_service == service
+    assert jt.webhook_key != ''
+    assert jt.webhook_credential is None
+    assert response.data == {'webhook_credential': ["Must match the selected webhook service."]}