Lock down OAuth2 views (#434)

Doing this properly with RBAC depends on #424.

For now, we limit Application views to superusers. We limit Token
views using a custom DRF permission class based on what was specified
in AWX's access.py

This also fixes a problem in the activity stream tests where INSTALLED_APPS
got permanently modified which made other tests unpredictable.

It also fixes a bug where token scope choices were previously limited to
only one option.

Signed-off-by: Rick Elrod <[email protected]>
Co-authored-by: John Westcott IV <[email protected]>

ansible_base/lib/utils/views/permissions.py

@@ -1,4 +1,4 @@
-from rest_framework.permissions import BasePermission
+from rest_framework.permissions import SAFE_METHODS, BasePermission
 
 
 class IsSuperuser(BasePermission):
@@ -7,4 +7,20 @@ class IsSuperuser(BasePermission):
     """
 
     def has_permission(self, request, view):
-        return bool(request.user and request.user.is_superuser)
+        return bool(request.user and request.user.is_authenticated and request.user.is_superuser)
+
+
+class IsSuperuserOrAuditor(BasePermission):
+    """
+    Allows write access only to system admin users.
+    Allows read access only to system auditor users.
+    """
+
+    def has_permission(self, request, view):
+        if not (request.user and request.user.is_authenticated):
+            return False
+        if request.user.is_superuser:
+            return True
+        if request.method in SAFE_METHODS:
+            return getattr(request.user, 'is_system_auditor', False)
+        return False

ansible_base/oauth2_provider/migrations/0004_alter_oauth2accesstoken_scope.py

@@ -0,0 +1,19 @@
+# Generated by Django 4.2.11 on 2024-06-06 22:46
+
+import ansible_base.oauth2_provider.models.access_token
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('dab_oauth2_provider', '0003_remove_oauth2application_logo_data'),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name='oauth2accesstoken',
+            name='scope',
+            field=models.CharField(default='write', help_text="Allowed scopes, further restricts user's permissions. Must be a simple space-separated string with allowed scopes ['read', 'write'].", max_length=32, validators=[ansible_base.oauth2_provider.models.access_token.validate_scope]),
+        ),
+    ]

ansible_base/oauth2_provider/models/access_token.py

@@ -1,5 +1,6 @@
 import oauth2_provider.models as oauth2_models
 from django.conf import settings
+from django.core.exceptions import ValidationError
 from django.db import connection, models
 from django.utils.timezone import now
 from django.utils.translation import gettext_lazy as _
@@ -10,6 +11,18 @@
 from ansible_base.lib.utils.settings import get_setting
 from ansible_base.oauth2_provider.utils import is_external_account
 
+SCOPES = ['read', 'write']
+
+
+def validate_scope(value):
+    given_scopes = value.split(' ')
+    if not given_scopes:
+        raise ValidationError(_('Scope must be a simple space-separated string with allowed scopes: %(scopes)s') % {'scopes': ', '.join(SCOPES)})
+    for scope in given_scopes:
+        if scope not in SCOPES:
+            raise ValidationError(_('Invalid scope: %(scope)s. Must be one of: %(scopes)s') % {'scope': scope, 'scopes': ', '.join(SCOPES)})
+
+
 activitystream = object
 if 'ansible_base.activitystream' in settings.INSTALLED_APPS:
     from ansible_base.activitystream.models import AuditableModel
@@ -26,11 +39,6 @@ class Meta(oauth2_models.AbstractAccessToken.Meta):
         ordering = ('id',)
         swappable = "OAUTH2_PROVIDER_ACCESS_TOKEN_MODEL"
 
-    SCOPE_CHOICES = [
-        ('read', _('Read')),
-        ('write', _('Write')),
-    ]
-
     user = models.ForeignKey(
         settings.AUTH_USER_MODEL,
         on_delete=models.CASCADE,
@@ -57,11 +65,10 @@ class Meta(oauth2_models.AbstractAccessToken.Meta):
         editable=False,
     )
     scope = models.CharField(
-        blank=True,
         default='write',
         max_length=32,
-        choices=SCOPE_CHOICES,
         help_text=_("Allowed scopes, further restricts user's permissions. Must be a simple space-separated string with allowed scopes ['read', 'write']."),
+        validators=[validate_scope],
     )
     token = prevent_search(
         models.CharField(

ansible_base/oauth2_provider/serializers/token.py

@@ -15,14 +15,14 @@
 from ansible_base.lib.utils.encryption import ENCRYPTED_STRING
 from ansible_base.lib.utils.settings import get_setting
 from ansible_base.oauth2_provider.models import OAuth2AccessToken, OAuth2RefreshToken
+from ansible_base.oauth2_provider.models.access_token import SCOPES
 
 logger = logging.getLogger("ansible_base.oauth2_provider.serializers.token")
 
 
 class BaseOAuth2TokenSerializer(CommonModelSerializer):
     refresh_token = SerializerMethodField()
     token = SerializerMethodField()
-    ALLOWED_SCOPES = [x[0] for x in OAuth2AccessToken.SCOPE_CHOICES]
 
     class Meta:
         model = OAuth2AccessToken
@@ -69,13 +69,13 @@ def _is_valid_scope(self, value):
         for word in words:
             if words.count(word) > 1:
                 return False  # do not allow duplicates
-            if word not in self.ALLOWED_SCOPES:
+            if word not in SCOPES:
                 return False
         return True
 
     def validate_scope(self, value):
         if not self._is_valid_scope(value):
-            raise ValidationError(_('Must be a simple space-separated string with allowed scopes {}.').format(self.ALLOWED_SCOPES))
+            raise ValidationError(_('Must be a simple space-separated string with allowed scopes {}.').format(SCOPES))
         return value
 
     def create(self, validated_data):

ansible_base/oauth2_provider/views/application.py

@@ -1,12 +1,12 @@
-from rest_framework import permissions
 from rest_framework.viewsets import ModelViewSet
 
 from ansible_base.lib.utils.views.django_app_api import AnsibleBaseDjangoAppApiView
+from ansible_base.lib.utils.views.permissions import IsSuperuserOrAuditor
 from ansible_base.oauth2_provider.models import OAuth2Application
 from ansible_base.oauth2_provider.serializers import OAuth2ApplicationSerializer
 
 
 class OAuth2ApplicationViewSet(AnsibleBaseDjangoAppApiView, ModelViewSet):
     queryset = OAuth2Application.objects.all()
     serializer_class = OAuth2ApplicationSerializer
-    permission_classes = [permissions.IsAuthenticated]
+    permission_classes = [IsSuperuserOrAuditor]

ansible_base/oauth2_provider/views/permissions.py

@@ -0,0 +1,37 @@
+from django.conf import settings
+from rest_framework.permissions import SAFE_METHODS, BasePermission
+
+
+class OAuth2TokenPermission(BasePermission):
+    # An app token is a token that has an application attached to it
+    # A personal access token (PAT) is a token with no application attached to it
+    # With that in mind:
+    # - An app token can be read, changed, or deleted if:
+    #   - I am the superuser
+    #   - I am the admin of the organization of the application of the token
+    #   - I am the user of the token
+    # - An app token can be created if:
+    #   - I have read access to the application (currently this means: I am the superuser)
+    # - A PAT can be read, changed, or deleted if:
+    #   - I am the superuser
+    #   - I am the user of the token
+    # - A PAT can be created if:
+    #   - I am a user
+
+    def has_permission(self, request, view):
+        # Handle PAT and app token creation separately
+        if request.method == 'POST':
+            if request.data.get('application'):
+                # TODO: Change this once ansible/django-ansible-base#424 is fixed
+                return request.user.is_superuser
+        return request.user.is_authenticated
+
+    def has_object_permission(self, request, view, obj):
+        if request.method in SAFE_METHODS and getattr(request.user, 'is_system_auditor', False):
+            return True
+        if request.user.is_superuser:
+            return True
+        if 'ansible_base.rbac' in settings.INSTALLED_APPS:
+            if obj.application and obj.application.organization.access_qs(request.user, "change").exists():
+                return True
+        return request.user == obj.user

ansible_base/oauth2_provider/views/token.py

@@ -3,13 +3,13 @@
 from django.utils.timezone import now
 from oauth2_provider import views as oauth_views
 from oauthlib import oauth2
-from rest_framework import permissions
 from rest_framework.viewsets import ModelViewSet
 
 from ansible_base.lib.utils.settings import get_setting
 from ansible_base.lib.utils.views.django_app_api import AnsibleBaseDjangoAppApiView
 from ansible_base.oauth2_provider.models import OAuth2AccessToken, OAuth2RefreshToken
 from ansible_base.oauth2_provider.serializers import OAuth2TokenSerializer
+from ansible_base.oauth2_provider.views.permissions import OAuth2TokenPermission
 
 
 class TokenView(oauth_views.TokenView, AnsibleBaseDjangoAppApiView):
@@ -54,4 +54,4 @@ def create_token_response(self, request):
 class OAuth2TokenViewSet(ModelViewSet, AnsibleBaseDjangoAppApiView):
     queryset = OAuth2AccessToken.objects.all()
     serializer_class = OAuth2TokenSerializer
-    permission_classes = [permissions.IsAuthenticated]
+    permission_classes = [OAuth2TokenPermission]

test_app/tests/activitystream/test_api.py

@@ -63,12 +63,18 @@ def test_activitystream_api_permission_classes(admin_api_client, user_api_client
     """
     url = reverse("activitystream-list")
 
+    # We *cannot* modify the list in place, pytest-django can't undo that change
+    # at the end of the test.
+    installed_apps = settings.INSTALLED_APPS.copy()
+
     if 'ansible_base.rbac' in settings.INSTALLED_APPS:
         if not has_rbac_app:
-            settings.INSTALLED_APPS.remove('ansible_base.rbac')
+            installed_apps.remove('ansible_base.rbac')
     else:
         if has_rbac_app:
-            settings.INSTALLED_APPS.append('ansible_base.rbac')
+            installed_apps.append('ansible_base.rbac')
+
+    settings.INSTALLED_APPS = installed_apps
 
     # Admin can always access
     response = admin_api_client.get(url)