feat(oauth2_provider): check for OAuth2ScopePermission on all APIViews

Add a check for OAuth2ScopePermission on all active views in an application.
- ANSIBLE_BASE_OAUTH2_PROVIDER_PERMISSIONS_CHECK_IGNORED_VIEWS django setting
  for setting ignores by class path.
- Uses django's check framework, should be compatible with the related
  functionality.
- Set as a deployment check, so it will not block app startup.

Attach OAuth2ScopePermission to all necessary views with permission_classes
defined when ansible_base.oauth2_provider is installed.

AAP-26507

ansible_base/authentication/views/authenticator_users.py

@@ -7,7 +7,7 @@
 from django.http import Http404
 
 from ansible_base.authentication.models import Authenticator
-from ansible_base.lib.utils.views.permissions import IsSuperuserOrAuditor
+from ansible_base.lib.utils.views.permissions import IsSuperuserOrAuditor, try_add_oauth2_scope_permission
 
 logger = logging.getLogger('ansible_base.authentication.views.authenticator_users')
 
@@ -22,7 +22,7 @@ def get_authenticator_user_view():
             raise ModuleNotFoundError()
 
         class AuthenticatorPluginRelatedUsersView(user_viewset_view):
-            permission_classes = [IsSuperuserOrAuditor]
+            permission_classes = try_add_oauth2_scope_permission([IsSuperuserOrAuditor])
 
             def get_queryset(self, **kwargs):
                 # during unit testing we get the pk from kwargs

ansible_base/lib/dynamic_config/dynamic_urls.py

@@ -10,12 +10,14 @@
 for url_type in url_types:
     globals()[url_type] = []
 
-for app in getattr(settings, 'INSTALLED_APPS', []):
+installed_apps = getattr(settings, 'INSTALLED_APPS', [])
+for app in installed_apps:
     if app.startswith('ansible_base.'):
         if not importlib.util.find_spec(f'{app}.urls'):
             logger.debug(f'Module {app} does not specify urls.py')
             continue
         url_module = __import__(f'{app}.urls', fromlist=url_types)
         logger.debug(f'Including URLS from {app}.urls')
         for url_type in ['api_version_urls', 'root_urls', 'api_urls']:
-            globals()[url_type].extend(getattr(url_module, url_type, []))
+            urls = getattr(url_module, url_type, [])
+            globals()[url_type].extend(urls)

ansible_base/lib/dynamic_config/settings_logic.py

@@ -283,6 +283,8 @@ def get_dab_settings(
 
         dab_data['ALLOW_OAUTH2_FOR_EXTERNAL_USERS'] = False
 
+        dab_data['ANSIBLE_BASE_OAUTH2_PROVIDER_PERMISSIONS_CHECK_DEFAULT_IGNORED_VIEWS'] = []
+
     if caches is not None:
         dab_data['CACHES'] = copy(caches)
         # Ensure proper configuration for fallback cache

ansible_base/lib/utils/views/permissions.py

@@ -1,5 +1,25 @@
 from rest_framework.permissions import SAFE_METHODS, BasePermission
 
+from ansible_base.lib.utils.settings import get_setting
+
+oauth2_provider_installed = "ansible_base.oauth2_provider" in get_setting("INSTALLED_APPS", [])
+
+
+def try_add_oauth2_scope_permission(permission_classes: list):
+    """
+    Attach OAuth2ScopePermission to the provided permission_classes list
+
+    :param permission_classes: list of rest_framework permissions
+    :return: A list of permission_classes, including OAuth2ScopePermission
+        if ansible_base.oauth2_provider is installed; otherwise the same
+        permission_classes list supplied to the function
+    """
+    if oauth2_provider_installed:
+        from ansible_base.oauth2_provider.permissions import OAuth2ScopePermission
+
+        return [OAuth2ScopePermission] + permission_classes
+    return permission_classes
+
 
 class IsSuperuser(BasePermission):
     """

ansible_base/lib/utils/views/urls.py

@@ -0,0 +1,24 @@
+from typing import Type
+
+from rest_framework.schemas.generators import EndpointEnumerator
+from rest_framework.views import APIView
+
+
+def get_api_view_functions(urlpatterns=None) -> set[Type[APIView]]:
+    """
+    Extract view classes from a urlpatterns list using the show_urls helper functions
+
+    :param urlpatterns: django urlpatterns list
+    :return: set of all view classes used by the urlpatterns list
+    """
+    views = set()
+
+    enumerator = EndpointEnumerator()
+    # Get all active APIViews from urlconf
+    endpoints = enumerator.get_api_endpoints(patterns=urlpatterns)
+    for _, _, func in endpoints:
+        # ApiView.as_view() breadcrumb
+        if hasattr(func, 'cls'):
+            views.add(func.cls)
+
+    return views

ansible_base/oauth2_provider/apps.py

@@ -1,7 +1,14 @@
 from django.apps import AppConfig
+from django.core.checks import register
 
 
 class Oauth2ProviderConfig(AppConfig):
     default_auto_field = 'django.db.models.BigAutoField'
     name = 'ansible_base.oauth2_provider'
     label = 'dab_oauth2_provider'
+
+    def ready(self):
+        # Load checks
+        from ansible_base.oauth2_provider.checks.permisssions_check import oauth2_permission_scope_check
+
+        register(oauth2_permission_scope_check, "oauth2_permissions", deploy=True)

ansible_base/oauth2_provider/checks/permisssions_check.py

@@ -0,0 +1,153 @@
+from typing import Iterable, Optional, Type, Union
+
+from django.apps import AppConfig
+from django.conf import settings
+from django.core.checks import CheckMessage, Debug, Error, Warning
+from rest_framework.permissions import AllowAny, OperandHolder, OperationHolderMixin, SingleOperandHolder
+from rest_framework.views import APIView
+
+from ansible_base.lib.utils.views.urls import get_api_view_functions
+from ansible_base.oauth2_provider.permissions import OAuth2ScopePermission
+
+
+class OAuth2ScopePermissionCheck:
+    """
+    Class containing logic for checking view classes for the
+    OAuth2ScopePermission permission_class, and aggregating
+    CheckMessage's for django system checks.
+
+    :param ignore_list: List of python import path strings for view classes exempt from the check logic.
+    :type ignore_list: list
+    """
+
+    def __init__(self, ignore_list: Iterable[str], generate_check_messages=True):
+        self.messages: list[CheckMessage] = []
+        self.current_view: Optional[Type[APIView]] = None
+        self.ignore_list = ignore_list
+        self.generate_check_messages = generate_check_messages
+
+    def check_message(self, message: CheckMessage):
+        if self.generate_check_messages:
+            self.messages.append(message)
+
+    # These are all warning or error conditions, this function is mostly saying to not invert OAuth2ScopePermissions
+    # Returns False.
+    def process_single_operand_holder(self, operand_holder: SingleOperandHolder) -> bool:
+        # The only unary operand for permission classes provided by rest_framework is ~ (not)
+        if self.parse_permission_class(operand_holder.op1_class):
+            self.check_message(
+                Warning(
+                    "~ (not) operand used on OAuth2ScopePermission, probably a bad idea.",
+                    id="ansible_base.oauth2_provider.W001",
+                    obj=self.current_view,
+                )
+            )
+
+        return False
+
+    def process_operand_holder(self, operand_holder: OperandHolder) -> bool:
+        return self.parse_permission_class(operand_holder.op1_class) or self.parse_permission_class(operand_holder.op2_class)
+
+    # Check if permission class is present in nested operands
+    # Sort of recursive? Reasonably this should not be an issue, so long as we don't recurse on an unknown OperationHolderMixin type
+    def parse_permission_class(self, cls: Union[Type[OperationHolderMixin], OperationHolderMixin]) -> bool:
+        # First, most likely case, we're dealing with a BasePermission subclass.
+        if cls is OAuth2ScopePermission:
+            return True
+        elif isinstance(cls, SingleOperandHolder):
+            # Warning or Error case: Will not accept OAuth2 permission nested in NOT
+            return self.process_single_operand_holder(cls)
+        elif isinstance(cls, OperandHolder):
+            return self.process_operand_holder(cls)
+        return False
+
+    def check_view(self, view_class: Type[APIView]) -> bool:
+        """
+        Primary function of the OAuth2ScopePermissionCheck.
+
+        Checks if OAuth2ScopePermission is present on the supplied view's
+        permission_classes; ignores classes that are not APIViews, or that are
+        in the ignore_list.
+
+        Appends CheckMessages to self.messages as a side effect.
+
+        :param view_class: django View class or rest_framework ApiView class
+
+        :return: True if view_class uses the OAuth2ScopePermission permission
+            class, or has some mitigating circumstance that prohibits it, such as
+            view_class not using permission classes, or its import path being in
+            self.ignore_list; returns False otherwise.
+        """
+        if f"{view_class.__module__}.{view_class.__name__}" in self.ignore_list:
+            self.check_message(
+                Debug(
+                    "View class in the ignore list. Ignoring.",
+                    obj=view_class,
+                    id="ansible_base.oauth2_provider.D03",
+                )
+            )
+            return True
+
+        self.current_view = view_class
+
+        for permission_class in getattr(self.current_view, "permission_classes", []):
+            if self.parse_permission_class(permission_class):
+                self.check_message(
+                    Debug(
+                        "Found OAuth2ScopePermission permission_class",
+                        obj=self.current_view,
+                        id="ansible_base.oauth2_provider.D02",
+                    )
+                )
+                return True
+
+        if not self.current_view.permission_classes or AllowAny in self.current_view.permission_classes:
+            self.check_message(
+                Debug(
+                    "View object is fully permissive, OAuth2ScopePermission is not required",
+                    obj=self.current_view,
+                    id="ansible_base.oauth2_provider.D04",
+                )
+            )
+            return True
+
+        # if we went though the whole loop without finding a valid permission_class, raise an error
+        self.check_message(
+            Error(
+                "View class has no valid usage of OAuth2ScopePermission",
+                obj=self.current_view,
+                id="ansible_base.oauth2_provider.E002",
+            )
+        )
+        return False
+
+
+def view_in_app_configs(view_class: type, app_configs: Optional[list[AppConfig]]) -> bool:
+    if app_configs:
+        for app_config in app_configs:
+            if view_class.__module__.startswith(app_config.name):
+                return True
+        return False
+    return True
+
+
+def oauth2_permission_scope_check(app_configs: Optional[list[AppConfig]], **kwargs) -> list[CheckMessage]:
+    """
+    Check for OAuth2ScopePermission permission class on all enabled views.
+
+    Ignore views in the ANSIBLE_BASE_OAUTH2_PROVIDER_PERMISSIONS_CHECK_IGNORED_VIEWS setting
+    """
+    ignore_list = set(
+        getattr(settings, "ANSIBLE_BASE_OAUTH2_PROVIDER_PERMISSIONS_CHECK_DEFAULT_IGNORED_VIEWS", [])
+        + getattr(settings, "ANSIBLE_BASE_OAUTH2_PROVIDER_PERMISSIONS_CHECK_IGNORED_VIEWS", [])
+    )
+
+    check = OAuth2ScopePermissionCheck(ignore_list)
+
+    view_functions = get_api_view_functions()
+    for view in view_functions:
+        # Only run checks on included apps (or all if app_configs is None)
+        if view_in_app_configs(view, app_configs):
+            check.check_view(view)
+
+    return check.messages

ansible_base/oauth2_provider/views/application.py

@@ -3,10 +3,11 @@
 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.permissions import OAuth2ScopePermission
 from ansible_base.oauth2_provider.serializers import OAuth2ApplicationSerializer
 
 
 class OAuth2ApplicationViewSet(AnsibleBaseDjangoAppApiView, ModelViewSet):
     queryset = OAuth2Application.objects.all()
     serializer_class = OAuth2ApplicationSerializer
-    permission_classes = [IsSuperuserOrAuditor]
+    permission_classes = [OAuth2ScopePermission, IsSuperuserOrAuditor]

ansible_base/oauth2_provider/views/token.py

@@ -10,6 +10,7 @@
 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.permissions import OAuth2ScopePermission
 from ansible_base.oauth2_provider.serializers import OAuth2TokenSerializer
 from ansible_base.oauth2_provider.views.permissions import OAuth2TokenPermission
 
@@ -73,4 +74,4 @@ def create_token_response(self, request):
 class OAuth2TokenViewSet(ModelViewSet, AnsibleBaseDjangoAppApiView):
     queryset = OAuth2AccessToken.objects.all()
     serializer_class = OAuth2TokenSerializer
-    permission_classes = [OAuth2TokenPermission]
+    permission_classes = [OAuth2ScopePermission, OAuth2TokenPermission]

ansible_base/rbac/api/views.py

@@ -11,6 +11,7 @@
 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 try_add_oauth2_scope_permission
 from ansible_base.rbac.api.permissions import RoleDefinitionPermissions
 from ansible_base.rbac.api.serializers import (
     RoleDefinitionDetailSerializer,
@@ -44,7 +45,7 @@ class RoleMetadataView(AnsibleBaseDjangoAppApiView, GenericAPIView):
     allowed_permissions: Valid permissions for a role of a given content_type
     """
 
-    permission_classes = [permissions.IsAuthenticated]
+    permission_classes = try_add_oauth2_scope_permission([permissions.IsAuthenticated])
     serializer_class = RoleMetadataSerializer
 
     def get(self, request, format=None):
@@ -88,7 +89,7 @@ class RoleDefinitionViewSet(AnsibleBaseDjangoAppApiView, ModelViewSet):
 
     queryset = RoleDefinition.objects.prefetch_related('created_by', 'modified_by', 'content_type', 'permissions')
     serializer_class = RoleDefinitionSerializer
-    permission_classes = [RoleDefinitionPermissions]
+    permission_classes = try_add_oauth2_scope_permission([RoleDefinitionPermissions])
 
     def get_serializer_class(self):
         if self.action == 'update':
@@ -112,7 +113,7 @@ def perform_destroy(self, instance):
 
 
 class BaseAssignmentViewSet(AnsibleBaseDjangoAppApiView, ModelViewSet):
-    permission_classes = [permissions.IsAuthenticated]
+    permission_classes = try_add_oauth2_scope_permission([permissions.IsAuthenticated])
     # PUT and PATCH are not allowed because these are immutable
     http_method_names = ['get', 'post', 'head', 'options', 'delete']
     prefetch_related = ()

ansible_base/resource_registry/views.py

@@ -13,6 +13,7 @@
 
 from ansible_base.lib.utils.response import CSVStreamResponse, get_relative_url
 from ansible_base.lib.utils.views.django_app_api import AnsibleBaseDjangoAppApiView
+from ansible_base.lib.utils.views.permissions import try_add_oauth2_scope_permission
 from ansible_base.resource_registry.models import Resource, ResourceType, service_id
 from ansible_base.resource_registry.registry import get_registry
 from ansible_base.resource_registry.serializers import ResourceListSerializer, ResourceSerializer, ResourceTypeSerializer, UserAuthenticationSerializer
@@ -66,9 +67,11 @@ class ResourceAPIMixin:
     """
 
     filter_backends = (FieldLookupBackend, TypeFilterBackend, OrderByBackend)
-    permission_classes = [
-        HasResourceRegistryPermissions,
-    ]
+    permission_classes = try_add_oauth2_scope_permission(
+        [
+            HasResourceRegistryPermissions,
+        ]
+    )
     pagination_class = ResourcesPagination
 
 
@@ -153,9 +156,11 @@ def manifest(self, request, name, *args, **kwargs):
 class ServiceMetadataView(
     AnsibleBaseDjangoAppApiView,
 ):
-    permission_classes = [
-        HasResourceRegistryPermissions,
-    ]
+    permission_classes = try_add_oauth2_scope_permission(
+        [
+            HasResourceRegistryPermissions,
+        ]
+    )
 
     # Corresponds to viewset action but given a different name so schema generators are not messed up
     custom_action_label = "service-metadata"
@@ -166,7 +171,7 @@ def get(self, request, **kwargs):
 
 
 class ServiceIndexRootView(AnsibleBaseDjangoAppApiView):
-    permission_classes = [permissions.IsAuthenticated]
+    permission_classes = try_add_oauth2_scope_permission([permissions.IsAuthenticated])
 
     def get(self, request, format=None):
         '''Link other resource registry endpoints'''