Merge pull request #2473 from ryanpetrello/disable-oauth2-for-external-users

don't allow OAuth2 token creation for "external" users

awx/api/conf.py

@@ -47,3 +47,14 @@ register(
     category=_('Authentication'),
     category_slug='authentication',
 )
+register(
+    'ALLOW_OAUTH2_FOR_EXTERNAL_USERS',
+    field_class=fields.BooleanField,
+    default=False,
+    label=_('Allow External Users to Create OAuth2 Tokens'),
+    help_text=_('For security reasons, users from external auth providers (LDAP, SAML, '
+                'SSO, Radius, and others) are not allowed to create OAuth2 tokens. '
+                'To change this behavior, enable this setting.'),
+    category=_('Authentication'),
+    category_slug='authentication',
+)

awx/api/serializers.py

@@ -13,6 +13,7 @@ from collections import OrderedDict
 from datetime import timedelta
 
 # OAuth2
+from oauthlib import oauth2
 from oauthlib.common import generate_token
 
 # Django
@@ -54,7 +55,7 @@ from awx.main.utils import (
     get_type_for_model, get_model_for_type, timestamp_apiformat,
     camelcase_to_underscore, getattrd, parse_yaml_or_json,
     has_model_field_prefetched, extract_ansible_vars, encrypt_dict,
-    prefetch_page_capabilities)
+    prefetch_page_capabilities, get_external_account)
 from awx.main.utils.filters import SmartFilter
 from awx.main.redact import REPLACE_STR
 
@@ -932,23 +933,7 @@ class UserSerializer(BaseSerializer):
             obj.save(update_fields=['password'])
 
     def get_external_account(self, obj):
-        account_type = None
-        if getattr(settings, 'AUTH_LDAP_SERVER_URI', None) and feature_enabled('ldap'):
-            try:
-                if obj.pk and obj.profile.ldap_dn and not obj.has_usable_password():
-                    account_type = "ldap"
-            except AttributeError:
-                pass
-        if (getattr(settings, 'SOCIAL_AUTH_GOOGLE_OAUTH2_KEY', None) or
-                getattr(settings, 'SOCIAL_AUTH_GITHUB_KEY', None) or
-                getattr(settings, 'SOCIAL_AUTH_GITHUB_ORG_KEY', None) or
-                getattr(settings, 'SOCIAL_AUTH_GITHUB_TEAM_KEY', None) or
-                getattr(settings, 'SOCIAL_AUTH_SAML_ENABLED_IDPS', None)) and obj.social_auth.all():
-            account_type = "social"
-        if (getattr(settings, 'RADIUS_SERVER', None) or
-                getattr(settings, 'TACACSPLUS_HOST', None)) and obj.enterprise_auth.all():
-            account_type = "enterprise"
-        return account_type
+        return get_external_account(obj)
 
     def create(self, validated_data):
         new_password = validated_data.pop('password', None)
@@ -1082,7 +1067,13 @@ class BaseOAuth2TokenSerializer(BaseSerializer):
                 'Must be a simple space-separated string with allowed scopes {}.'
             ).format(self.ALLOWED_SCOPES))
         return value
-            
+
+    def create(self, validated_data):
+        try:
+            return super(BaseOAuth2TokenSerializer, self).create(validated_data)
+        except oauth2.AccessDeniedError as e:
+            raise PermissionDenied(str(e))
+
 
 class UserAuthorizedTokenSerializer(BaseOAuth2TokenSerializer):  
 

awx/api/urls/oauth2_root.py

@@ -3,16 +3,29 @@
 
 from django.conf.urls import url
 
-from oauth2_provider.urls import base_urlpatterns
+from oauthlib import oauth2
+from oauth2_provider import views
 
 from awx.api.views import (
     ApiOAuthAuthorizationRootView,
 )
 
 
+class TokenView(views.TokenView):
+
+    def create_token_response(self, request):
+        try:
+            return super(TokenView, self).create_token_response(request)
+        except oauth2.AccessDeniedError as e:
+            return request.build_absolute_uri(), {}, str(e), '403'
+
+
 urls = [
     url(r'^$', ApiOAuthAuthorizationRootView.as_view(), name='oauth_authorization_root_view'),
-] + base_urlpatterns
+    url(r"^authorize/$", views.AuthorizationView.as_view(), name="authorize"),
+    url(r"^token/$", TokenView.as_view(), name="token"),
+    url(r"^revoke_token/$", views.RevokeTokenView.as_view(), name="revoke-token"),
+]
 
 
 __all__ = ['urls']

awx/main/models/oauth.py

@@ -11,7 +11,9 @@ from django.conf import settings
 # Django OAuth Toolkit
 from oauth2_provider.models import AbstractApplication, AbstractAccessToken
 from oauth2_provider.generators import generate_client_secret
+from oauthlib import oauth2
 
+from awx.main.utils import get_external_account
 from awx.main.fields import OAuth2ClientSecretField
 
 
@@ -123,3 +125,12 @@ class OAuth2AccessToken(AbstractAccessToken):
             self.last_used = now()
             self.save(update_fields=['last_used'])
         return valid
+
+    def save(self, *args, **kwargs):
+        if self.user and settings.ALLOW_OAUTH2_FOR_EXTERNAL_USERS is False:
+            external_account = get_external_account(self.user)
+            if external_account is not None:
+                raise oauth2.AccessDeniedError(_(
+                    'OAuth2 Tokens cannot be created by users associated with an external authentication provider ({})'
+                ).format(external_account))
+        super(OAuth2AccessToken, self).save(*args, **kwargs)

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

@@ -3,12 +3,14 @@ import base64
 import json
 
 from django.db import connection
+from django.test.utils import override_settings
 
 from awx.main.utils.encryption import decrypt_value, get_encryption_key
 from awx.api.versioning import reverse, drf_reverse
 from awx.main.models.oauth import (OAuth2Application as Application, 
                                    OAuth2AccessToken as AccessToken, 
                                    )
+from awx.sso.models import UserEnterpriseAuth
 from oauth2_provider.models import RefreshToken
 
 
@@ -29,6 +31,29 @@ def test_personal_access_token_creation(oauth_application, post, alice):
     assert 'refresh_token' in resp_json
 
 
+@pytest.mark.django_db
+@pytest.mark.parametrize('allow_oauth, status', [(True, 201), (False, 403)])
+def test_token_creation_disabled_for_external_accounts(oauth_application, post, alice, allow_oauth, status):
+    UserEnterpriseAuth(user=alice, provider='radius').save()
+    url = drf_reverse('api:oauth_authorization_root_view') + 'token/'
+
+    with override_settings(RADIUS_SERVER='example.org', ALLOW_OAUTH2_FOR_EXTERNAL_USERS=allow_oauth):
+        resp = post(
+            url,
+            data='grant_type=password&username=alice&password=alice&scope=read',
+            content_type='application/x-www-form-urlencoded',
+            HTTP_AUTHORIZATION='Basic ' + base64.b64encode(':'.join([
+                oauth_application.client_id, oauth_application.client_secret
+            ])),
+            status=status
+        )
+        if allow_oauth:
+            assert AccessToken.objects.count() == 1
+        else:
+            assert 'OAuth2 Tokens cannot be created by users associated with an external authentication provider' in resp.content
+            assert AccessToken.objects.count() == 0
+
+
 @pytest.mark.django_db
 def test_pat_creation_no_default_scope(oauth_application, post, admin):
     # tests that the default scope is overriden

awx/main/utils/common.py

@@ -52,7 +52,7 @@ __all__ = ['get_object_or_400', 'get_object_or_403', 'camelcase_to_underscore',
            'extract_ansible_vars', 'get_search_fields', 'get_system_task_capacity', 'get_cpu_capacity', 'get_mem_capacity',
            'wrap_args_with_proot', 'build_proot_temp_dir', 'check_proot_installed', 'model_to_dict',
            'model_instance_diff', 'timestamp_apiformat', 'parse_yaml_or_json', 'RequireDebugTrueOrTest',
-           'has_model_field_prefetched', 'set_environ', 'IllegalArgumentError', 'get_custom_venv_choices']
+           'has_model_field_prefetched', 'set_environ', 'IllegalArgumentError', 'get_custom_venv_choices', 'get_external_account']
 
 
 def get_object_or_400(klass, *args, **kwargs):
@@ -1073,3 +1073,25 @@ def has_model_field_prefetched(model_obj, field_name):
     # NOTE: Update this function if django internal implementation changes.
     return getattr(getattr(model_obj, field_name, None),
                    'prefetch_cache_name', '') in getattr(model_obj, '_prefetched_objects_cache', {})
+
+
+def get_external_account(user):
+    from django.conf import settings
+    from awx.conf.license import feature_enabled
+    account_type = None
+    if getattr(settings, 'AUTH_LDAP_SERVER_URI', None) and feature_enabled('ldap'):
+        try:
+            if user.pk and user.profile.ldap_dn and not user.has_usable_password():
+                account_type = "ldap"
+        except AttributeError:
+            pass
+    if (getattr(settings, 'SOCIAL_AUTH_GOOGLE_OAUTH2_KEY', None) or
+            getattr(settings, 'SOCIAL_AUTH_GITHUB_KEY', None) or
+            getattr(settings, 'SOCIAL_AUTH_GITHUB_ORG_KEY', None) or
+            getattr(settings, 'SOCIAL_AUTH_GITHUB_TEAM_KEY', None) or
+            getattr(settings, 'SOCIAL_AUTH_SAML_ENABLED_IDPS', None)) and user.social_auth.all():
+        account_type = "social"
+    if (getattr(settings, 'RADIUS_SERVER', None) or
+            getattr(settings, 'TACACSPLUS_HOST', None)) and user.enterprise_auth.all():
+        account_type = "enterprise"
+    return account_type

awx/settings/defaults.py

@@ -355,6 +355,7 @@ OAUTH2_PROVIDER_REFRESH_TOKEN_MODEL = 'oauth2_provider.RefreshToken'
 
 OAUTH2_PROVIDER = {'ACCESS_TOKEN_EXPIRE_SECONDS': 31536000000,
                    'AUTHORIZATION_CODE_EXPIRE_SECONDS': 600}
+ALLOW_OAUTH2_FOR_EXTERNAL_USERS = False
 
 # LDAP server (default to None to skip using LDAP authentication).
 # Note: This setting may be overridden by database settings.

awx/ui/client/src/configuration/system-form/sub-forms/system-misc.form.js

@@ -40,6 +40,9 @@ export default ['i18n', function(i18n) {
             AUTH_BASIC_ENABLED: {
                 type: 'toggleSwitch',
             },
+            ALLOW_OAUTH2_FOR_EXTERNAL_USERS: {
+                type: 'toggleSwitch',
+            },
             REMOTE_HOST_HEADERS: {
                 type: 'textarea',
                 reset: 'REMOTE_HOST_HEADERS'