recursive DRF and rest_permission boolean permissions

Author: n2ygk

example/tests/test_openapi.py

@@ -2,13 +2,21 @@
 import pytest
 from django.conf.urls import url
 from django.test import RequestFactory, TestCase, override_settings
+from rest_condition import And, Not, Or
 from rest_framework import VERSION as DRFVERSION
 from rest_framework.authentication import BasicAuthentication, SessionAuthentication
+from rest_framework.permissions import (
+    AllowAny,
+    DjangoModelPermissions,
+    IsAdminUser,
+    IsAuthenticated
+)
 from rest_framework.request import Request
 
 from rest_framework_json_api.optional import OAuth2Authentication, TokenMatchesOASRequirements
+from rest_framework_json_api.views import ModelViewSet
 
-from example import views
+from example import models, serializers, views
 
 drf_version = tuple(int(x) for x in DRFVERSION.split('.'))
 
@@ -806,40 +814,42 @@ def test_delete_request(self):
         }
 
     @pytest.mark.skipif(TokenMatchesOASRequirements is None, reason="requires oauth")
-    def test_schema_security(self):
+    class OauthProtectedAuthorViewSet(views.AuthorViewSet):
+        authentication_classes = (OAuth2Authentication, BasicAuthentication,
+                                  SessionAuthentication)
+        permission_classes = (TokenMatchesOASRequirements, IsAuthenticated)
+        required_alternate_scopes = {
+            'GET': [['scope1', 'scope2'], ['scope3', 'scope4']],
+        }
+
+    oauth2_server = 'oauth.example.com'
+    oauth2_config = {
+        'authorization_endpoint': oauth2_server + '/authorize',
+        'token_endpoint': oauth2_server + '/token',
+        'scopes_supported': ['scope1', 'scope2', 'scope3', 'scope4'],
+        'grant_types_supported': ['implicit', 'authorization_code', 'client_credentials',
+                                  'password'],
+    }
+
+    def test_schema_security_list(self):
         """
-        Checks for security object
-        :return:
+        Checks for security objects
         """
-        class OauthProtectedAuthorViewSet(views.AuthorViewSet):
-            authentication_classes = (OAuth2Authentication, BasicAuthentication,
-                                      SessionAuthentication)
-            permission_classes = (TokenMatchesOASRequirements,)
-            required_alternate_scopes = {
-                'GET': [['scope1', 'scope2'], ['scope3', 'scope4']],
-            }
+        # TODO: also test permission combinations with rest_condition & DRF bitwise operators
 
-        oauth2_server = 'oauth.example.com'
-        oauth2_config = {
-            'authorization_endpoint': oauth2_server + '/authorize',
-            'token_endpoint': oauth2_server + '/token',
-            'scopes_supported': ['scope1', 'scope2', 'scope3', 'scope4'],
-            'grant_types_supported': ['implicit', 'authorization_code', 'client_credentials',
-                                      'password'],
-        }
         path = '/authors/'
         method = 'GET'
 
         view = create_view_with_kw(
-            OauthProtectedAuthorViewSet,
+            TestOperationIntrospection.OauthProtectedAuthorViewSet,
             method,
             create_request(path),
             {'get': 'list'}
         )
         inspector = AutoSchema()
         inspector.view = view
 
-        with override_settings(OAUTH2_CONFIG=oauth2_config):
+        with override_settings(OAUTH2_CONFIG=TestOperationIntrospection.oauth2_config):
             operation = inspector.get_operation(path, method)
 
         assert 'security' in operation
@@ -849,39 +859,73 @@ class OauthProtectedAuthorViewSet(views.AuthorViewSet):
         assert operation['security'][2] == {'basicAuth': []}
         assert operation['security'][3] == {'cookieAuth': []}
 
-    # TODO: figure these out
-    # def test_retrieve_relationships(self):
-    #     path = '/authors/{id}/relationships/bio/'
-    #     method = 'GET'
-    #
-    #     view = create_view_with_kw(
-    #         views.AuthorRelationshipView,
-    #         method,
-    #         create_request(path),
-    #         {'get': 'retrieve'}
-    #     )
-    #     inspector = AutoSchema()
-    #     inspector.view = view
-    #
-    #     operation = inspector.get_operation(path, method)
-    #     assert operation == {}
+    def test_schema_security_drf_condition(self):
+        """
+        Checks for security objects with DRF bitwise conditional operators
+        """
+        class DRF_Cond_ViewSet(TestOperationIntrospection.OauthProtectedAuthorViewSet):
+            # this is a crazy example just to make sure all the recursive code is covered
+            permission_classes = [
+                (IsAuthenticated & DjangoModelPermissions) |
+                ~(TokenMatchesOASRequirements & AllowAny),
+                ~AllowAny | (IsAdminUser & IsAuthenticated),
+                (TokenMatchesOASRequirements & AllowAny) |
+                (IsAuthenticated & DjangoModelPermissions),
+                ~TokenMatchesOASRequirements
+            ]
 
-    # def test_retrieve_related(self):
-    #     path = '/authors/{id}/{related_field}/'
-    #     method = 'GET'
-    #
-    #     view = create_view_with_kw(
-    #         views.AuthorViewSet,
-    #         method,
-    #         create_request(path),
-    #         {'get': 'retrieve_related',
-    #          'related_field': 'bio'}
-    #     )
-    #     inspector = AutoSchema()
-    #     inspector.view = view
-    #
-    #     operation = inspector.get_operation(path, method)
-    #     assert operation == {}
+        path = '/authors/'
+        method = 'GET'
+
+        view = create_view_with_kw(
+            DRF_Cond_ViewSet,
+            method,
+            create_request(path),
+            {'get': 'list'}
+        )
+        inspector = AutoSchema()
+        inspector.view = view
+
+        with override_settings(OAUTH2_CONFIG=TestOperationIntrospection.oauth2_config):
+            operation = inspector.get_operation(path, method)
+
+        assert 'security' in operation
+        assert {'oauth': ['scope1', 'scope2']} in operation['security']
+        assert {'oauth': ['scope3', 'scope4']} in operation['security']
+        assert {'basicAuth': []} in operation['security']
+        assert {'cookieAuth': []} in operation['security']
+
+    def test_schema_security_rest_condition(self):
+        """
+        Checks for security objects with rest_condition operator methods
+        """
+        class Rest_Cond_ViewSet(TestOperationIntrospection.OauthProtectedAuthorViewSet):
+            permission_classes = [
+                Or(
+                    And(IsAuthenticated, DjangoModelPermissions),
+                    And(Not(TokenMatchesOASRequirements), AllowAny)),
+            ]
+
+        path = '/authors/'
+        method = 'GET'
+
+        view = create_view_with_kw(
+            Rest_Cond_ViewSet,
+            method,
+            create_request(path),
+            {'get': 'list'}
+        )
+        inspector = AutoSchema()
+        inspector.view = view
+
+        with override_settings(OAUTH2_CONFIG=TestOperationIntrospection.oauth2_config):
+            operation = inspector.get_operation(path, method)
+
+        assert 'security' in operation
+        assert {'oauth': ['scope1', 'scope2']} in operation['security']
+        assert {'oauth': ['scope3', 'scope4']} in operation['security']
+        assert {'basicAuth': []} in operation['security']
+        assert {'cookieAuth': []} in operation['security']
 
 
 @override_settings(REST_FRAMEWORK={
@@ -901,3 +945,46 @@ def test_schema_construction(self):
         assert 'info' in schema
         assert 'paths' in schema
         assert 'components' in schema
+
+    # TODO: figure these out
+    def test_schema_related(self):
+        class AuthorBioViewSet(ModelViewSet):
+            queryset = models.AuthorBio.objects.all()
+            serializer_class = serializers.AuthorBioSerializer
+
+        patterns = [
+            url(r'^authors/(?P<pk>[^/.]+)/(?P<related_field>\w+)/$',
+                views.AuthorViewSet.as_view({'get': 'retrieve_related'}),
+                name='author-related'),
+            url(r'^bios/(?P<pk>[^/.]+)/$',
+                AuthorBioViewSet,
+                name='author-bio')
+        ]
+        generator = SchemaGenerator(patterns=patterns)
+
+        request = create_request('/authors/123/bio/')
+        schema = generator.get_schema(request=request)
+        # TODO: finish this test
+        print(schema)
+
+    # def test_retrieve_relationships(self):
+    #     path = '/authors/{id}/relationships/bio/'
+    #     method = 'GET'
+    #
+    #     view = create_view_with_kw(
+    #         views.AuthorViewSet,
+    #         method,
+    #         create_request(path),
+    #         {'get': 'retrieve_related'}
+    #     )
+    #     inspector = AutoSchema()
+    #     inspector.view = view
+    #
+    #     operation = inspector.get_operation(path, method)
+    #     assert 'responses' in operation
+    #     assert '200' in operation['responses']
+    #     resp = operation['responses']['200']['content']
+    #     data = resp['application/vnd.api+json']['schema']['properties']['data']
+    #     assert data['type'] == 'object'
+    #     assert data['required'] == ['type', 'id']
+    #     assert data['properties']['type'] == {'$ref': '#/components/schemas/type'}

rest_framework_json_api/optional.py

@@ -6,3 +6,10 @@
 except ImportError:  # pragma: no cover
     OAuth2Authentication = None
     TokenMatchesOASRequirements = None
+
+# DRF 3.9+ has native boolean conditions now.
+# But older code may use rest_condition (or other packages).
+try:
+    from rest_condition import Condition
+except ImportError:
+    Condition = None

rest_framework_json_api/schemas/openapi.py

@@ -6,23 +6,19 @@
 from django.utils.module_loading import import_string as import_class_from_dotted_path
 from rest_framework import exceptions
 from rest_framework.authentication import BasicAuthentication, SessionAuthentication
-from rest_framework.permissions import OperandHolder, SingleOperandHolder
+from rest_framework.permissions import AND, NOT, OR
 from rest_framework.relations import ManyRelatedField
 from rest_framework.schemas import openapi as drf_openapi
 from rest_framework.schemas.utils import is_list_view
 
 from rest_framework_json_api import serializers
-from rest_framework_json_api.optional import OAuth2Authentication, TokenMatchesOASRequirements
+from rest_framework_json_api.optional import (
+    Condition,
+    OAuth2Authentication,
+    TokenMatchesOASRequirements
+)
 from rest_framework_json_api.views import RelationshipView
 
-# DRF 3.9+ has native boolean conditions now
-# But older code may use rest_condition or other packages.
-try:
-    from rest_condition import Condition as rest_condition_Condition
-except ImportError:
-    rest_condition_Condition = None
-
-
 #: static OAS 3.0 component definitions that are referenced by AutoSchema.
 JSONAPI_COMPONENTS = {
     'schemas': {
@@ -576,38 +572,68 @@ def _get_oauth_security(self, path, method):
         # TODO: add JWT and SAML2 bearer
         content = []
         # permission_classes can be a direct list of classes, or instances of Operands, etc.
-        # TODO: this is kind of ugly. modularize it.
-        for perm_class_or_condition in self.view.permission_classes:
-            # check if DRF conditional operands were specified
-            if isinstance(perm_class_or_condition, OperandHolder):
-                if issubclass(perm_class_or_condition.op1_class, TokenMatchesOASRequirements):
-                    perm_class_instance = perm_class_or_condition.op1_class()
-                elif issubclass(perm_class_or_condition.op2_class, TokenMatchesOASRequirements):
-                    perm_class_instance = perm_class_or_condition.op2_class()
-                else:
-                    perm_class_instance = None
-            elif isinstance(perm_class_or_condition, SingleOperandHolder):
-                if issubclass(perm_class_or_condition.op1_class, TokenMatchesOASRequirements):
-                    perm_class_instance = perm_class_or_condition.op1_class()
-                else:
-                    perm_class_instance = None
-            # check for rest_condition.Condition
-            elif (rest_condition_Condition and
-                  isinstance(perm_class_or_condition, rest_condition_Condition)):
-                for cond in perm_class_or_condition.perms_or_conds:
-                    if issubclass(cond, TokenMatchesOASRequirements):
-                        perm_class_instance = cond()
-                        break
-            else:
-                perm_class_instance = perm_class_or_condition()
-            if isinstance(perm_class_instance, TokenMatchesOASRequirements):
+        for perm in self.view.permission_classes:
+            if (
+                    isinstance(perm(), TokenMatchesOASRequirements) or
+                    self._drf_conditional_contains(perm(), TokenMatchesOASRequirements) or
+                    self._rest_cond_contains(perm(), TokenMatchesOASRequirements)
+            ):
                 alt_scopes = self.view.required_alternate_scopes
                 if method not in alt_scopes:
                     continue
                 for scopes in alt_scopes[method]:
                     content.append({'oauth': scopes})
         return content
 
+    def _drf_conditional_contains(self, perm_inst, the_class):
+        """
+        Recursively check if DRF conditional operands were specified.
+        If there's any reference to `the_class` then return True.
+        Don't care what the boolean logic is, just if there's an instance of the_class.
+        """
+        binary = (AND, OR)
+        unary = (NOT,)
+        ops = binary + unary
+
+        if not isinstance(perm_inst, ops):
+            return False
+
+        if isinstance(perm_inst, binary):
+            if isinstance(perm_inst.op1, the_class):
+                return True
+            if isinstance(perm_inst.op2, the_class):
+                return True
+            if isinstance(perm_inst.op1, ops):
+                if self._drf_conditional_contains(perm_inst.op1, the_class):
+                    return True
+            if isinstance(perm_inst.op2, ops):
+                if self._drf_conditional_contains(perm_inst.op2, the_class):
+                    return True
+        elif isinstance(perm_inst, unary):
+            if isinstance(perm_inst.op1, the_class):
+                return True
+            if isinstance(perm_inst.op1, ops):
+                if self._drf_conditional_contains(perm_inst.op1, the_class):
+                    return True
+        return False
+
+    def _rest_cond_contains(self, perm_inst, the_class):
+        """
+        Recursively check if rest_condition conditional operands were specified.
+        If there's any reference to `the_class` then return True.
+        Don't care what the boolean logic is, just if there's an instance of the_class.
+        """
+        if Condition is None or not isinstance(perm_inst, Condition):
+            return False
+
+        for cond in perm_inst.perms_or_conds:
+            if isinstance(cond(), the_class):
+                return True
+            if isinstance(cond(), Condition):
+                if self._rest_cond_contains(cond, the_class):
+                    return True
+        return False
+
     def _get_include_parameters(self, path, method):
         """
         includes parameter: https://jsonapi.org/format/#fetching-includes

setup.cfg

@@ -22,6 +22,7 @@ known_first_party = rest_framework_json_api
 # This is to "trick" isort into putting example below DJA imports.
 known_localfolder = example
 known_standard_library = mock
+known_third_party = rest_condition
 line_length = 100
 multi_line_output = 3
 skip=

tox.ini

@@ -13,6 +13,7 @@ deps =
     drfmaster: https://github.com/encode/django-rest-framework/archive/master.zip
     oauth12: django-oauth-toolkit>=1.2.0
     coreapi>=2.3.1
+    rest-condition>=1.0.3
 
 setenv =
     PYTHONPATH = {toxinidir}