Add TokenHasMethodScopeAlternative

n2ygk · jleclanche · commit b4378b3d3197 · 2018-05-12T06:58:56.000+03:00
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -2,6 +2,7 @@
 
 * **Compatibility**: Python 3.4 is the new minimum required version.
 * **Compatibility**: Django 2.0 is the new minimum required version.
+* **New feature**: Added TokenHasMethodScopeAlternative Permissions.
 
 ### 1.1.1 [2018-05-08]
 
@@ -25,6 +26,7 @@
   refresh tokens may be re-used.
 * An `app_authorized` signal is fired when a token is generated.
 
+
 ### 1.0.0 [2017-06-07]
 
 * **New feature**: AccessToken, RefreshToken and Grant models are now swappable.
diff --git a/docs/rest-framework/openapi.yaml b/docs/rest-framework/openapi.yaml
@@ -0,0 +1,49 @@
+openapi: "3.0.0"
+info:
+  title: songs
+  version: v1
+components:
+  securitySchemes:
+    song_auth:
+      type: oauth2
+      flows:
+        implicit:
+          authorizationUrl: http://localhost:8000/o/authorize
+          scopes:
+            read: read about a song
+            create: create a new song
+            update: update an existing song
+            delete: delete a song
+            post: create a new song
+            widget: widget scope
+            scope2: scope too
+            scope3: another scope
+paths:
+  /songs:
+    get:
+      security:
+        - song_auth: [read]
+      responses:
+        '200':
+          description: A list of songs.
+    post:
+      security:
+        - song_auth: [create]
+        - song_auth: [post, widget]
+      responses:
+        '201':
+          description: new song added
+    put:
+      security:
+        - song_auth: [update]
+        - song_auth: [put, widget]
+      responses:
+        '204':
+          description: song updated
+    delete:
+      security:
+        - song_auth: [delete]
+        - song_auth: [scope2, scope3]
+      responses:
+        '200':
+          description: song deleted
diff --git a/docs/rest-framework/permissions.rst b/docs/rest-framework/permissions.rst
@@ -48,6 +48,7 @@ For example:
 
 When a request is performed both the `READ_SCOPE` \\ `WRITE_SCOPE` and 'music' scopes are required to be authorized for the current access token.
 
+
 TokenHasResourceScope
 ----------------------
 The `TokenHasResourceScope` permission class allows access only when the current access token has been authorized for **all** the scopes listed in the `required_scopes` field of the view but according of request's method.
@@ -81,3 +82,36 @@ For example:
         required_scopes = ['music']
 
 The `required_scopes` attribute is mandatory.
+
+
+TokenHasMethodScopeAlternative
+------------------------------
+
+The `TokenHasMethodScopeAlternative` permission class allows the access based on a per-method basis
+and with alternative lists of required scopes. This permission provides full functionality
+required by REST API specifications like the
+`OpenAPI Specification (OAS) security requirement object <https://github.com/OAI/OpenAPI-Specification/blob/master/versions/3.0.0.md#securityRequirementObject>`_.
+
+The `required_alternate_scopes` attribute is a required map keyed by HTTP method name where each value is
+a list of alternative lists of required scopes.
+
+In the follow example GET requires "read" scope, POST requires either "create" scope **OR** "post" and "widget" scopes,
+etc.
+
+.. code-block:: python
+
+    class SongView(views.APIView):
+        authentication_classes = [OAuth2Authentication]
+        permission_classes = [TokenHasMethodScopeAlternative]
+        required_alternate_scopes = {
+            "GET": [["read"]],
+            "POST": [["create"], ["post", "widget"]],
+            "PUT":  [["update"], ["put", "widget"]],
+            "DELETE": [["delete"], ["scope2", "scope3"]],
+        }
+
+The following is a minimal OAS declaration that shows the same required alternate scopes. It is complete enough
+to try it in the `swagger editor <https://editor.swagger.io>`_.
+
+.. literalinclude:: openapi.yaml
+  :language: YAML
diff --git a/oauth2_provider/contrib/rest_framework/__init__.py b/oauth2_provider/contrib/rest_framework/__init__.py
@@ -1,4 +1,6 @@
 # flake8: noqa
 from .authentication import OAuth2Authentication
-from .permissions import TokenHasScope, TokenHasReadWriteScope, TokenHasResourceScope
-from .permissions import IsAuthenticatedOrTokenHasScope
+from .permissions import (
+    TokenHasScope, TokenHasReadWriteScope, TokenHasMethodScopeAlternative,
+    TokenHasResourceScope, IsAuthenticatedOrTokenHasScope
+)
diff --git a/oauth2_provider/contrib/rest_framework/permissions.py b/oauth2_provider/contrib/rest_framework/permissions.py
@@ -121,3 +121,58 @@ def has_permission(self, request, view):
 
         token_has_scope = TokenHasScope()
         return (is_authenticated and not oauth2authenticated) or token_has_scope.has_permission(request, view)
+
+
+class TokenHasMethodScopeAlternative(BasePermission):
+    """
+    :attr:alternate_required_scopes: dict keyed by HTTP method name with value: iterable alternate scope lists
+
+    This fulfills the [Open API Specification (OAS; formerly Swagger)](https://www.openapis.org/)
+    list of alternative Security Requirements Objects for oauth2 or openIdConnect:
+      When a list of Security Requirement Objects is defined on the Open API object or Operation Object,
+      only one of Security Requirement Objects in the list needs to be satisfied to authorize the request.
+    [1](https://github.com/OAI/OpenAPI-Specification/blob/master/versions/3.0.0.md#securityRequirementObject)
+
+    For each method, a list of lists of allowed scopes is tried in order and the first to match succeeds.
+
+    @example
+    required_alternate_scopes = {
+       'GET': [['read']],
+       'POST': [['create1','scope2'], ['alt-scope3'], ['alt-scope4','alt-scope5']],
+    }
+
+    TODO: DRY: subclass TokenHasScope and iterate over values of required_scope?
+    """
+
+    def has_permission(self, request, view):
+        token = request.auth
+
+        if not token:
+            return False
+
+        if hasattr(token, "scope"):  # OAuth 2
+            required_alternate_scopes = self.get_required_alternate_scopes(request, view)
+
+            m = request.method.upper()
+            if m in required_alternate_scopes:
+                log.debug("Required scopes alternatives to access resource: {0}"
+                          .format(required_alternate_scopes[m]))
+                for alt in required_alternate_scopes[m]:
+                    if token.is_valid(alt):
+                        return True
+                return False
+            else:
+                log.warning("no scope alternates defined for method {0}".format(m))
+                return False
+
+        assert False, ("TokenHasMethodScope requires the"
+                       "`oauth2_provider.rest_framework.OAuth2Authentication` authentication "
+                       "class to be used.")
+
+    def get_required_alternate_scopes(self, request, view):
+        try:
+            return getattr(view, "required_alternate_scopes")
+        except AttributeError:
+            raise ImproperlyConfigured(
+                "TokenHasMethodScopeAlternative requires the view to"
+                " define the required_alternate_scopes attribute")
diff --git a/tests/test_rest_framework.py b/tests/test_rest_framework.py
@@ -2,17 +2,20 @@
 
 from django.conf.urls import include, url
 from django.contrib.auth import get_user_model
+from django.core.exceptions import ImproperlyConfigured
 from django.http import HttpResponse
 from django.test import TestCase
 from django.test.utils import override_settings
 from django.utils import timezone
 from rest_framework import permissions
+from rest_framework.authentication import BaseAuthentication
 from rest_framework.test import APIRequestFactory, force_authenticate
 from rest_framework.views import APIView
 
 from oauth2_provider.contrib.rest_framework import (
     IsAuthenticatedOrTokenHasScope, OAuth2Authentication,
-    TokenHasReadWriteScope, TokenHasResourceScope, TokenHasScope
+    TokenHasMethodScopeAlternative, TokenHasReadWriteScope,
+    TokenHasResourceScope, TokenHasScope
 )
 from oauth2_provider.models import get_access_token_model, get_application_model
 from oauth2_provider.settings import oauth2_settings
@@ -38,14 +41,17 @@ def get(self, request):
     def post(self, request):
         return HttpResponse({"a": 1, "b": 2, "c": 3})
 
+    def put(self, request):
+        return HttpResponse({"a": 1, "b": 2, "c": 3})
+
 
 class OAuth2View(MockView):
     authentication_classes = [OAuth2Authentication]
 
 
 class ScopedView(OAuth2View):
     permission_classes = [permissions.IsAuthenticated, TokenHasScope]
-    required_scopes = ["scope1"]
+    required_scopes = ["scope1", "another"]
 
 
 class AuthenticatedOrScopedView(OAuth2View):
@@ -62,13 +68,48 @@ class ResourceScopedView(OAuth2View):
     required_scopes = ["resource1"]
 
 
+class MethodScopeAltView(OAuth2View):
+    permission_classes = [TokenHasMethodScopeAlternative]
+    required_alternate_scopes = {
+        "GET": [["read"]],
+        "POST": [["create"]],
+        "PUT": [["update", "put"], ["update", "edit"]],
+        "DELETE": [["delete"], ["deleter", "write"]],
+    }
+
+
+class MethodScopeAltViewBad(OAuth2View):
+    permission_classes = [TokenHasMethodScopeAlternative]
+
+
+class MissingAuthentication(BaseAuthentication):
+    def authenticate(self, request):
+        return ("junk", "junk",)
+
+
+class BrokenOAuth2View(MockView):
+    authentication_classes = [MissingAuthentication]
+
+
+class TokenHasScopeViewWrongAuth(BrokenOAuth2View):
+    permission_classes = [TokenHasScope]
+
+
+class MethodScopeAltViewWrongAuth(BrokenOAuth2View):
+    permission_classes = [TokenHasMethodScopeAlternative]
+
+
 urlpatterns = [
     url(r"^oauth2/", include("oauth2_provider.urls")),
     url(r"^oauth2-test/$", OAuth2View.as_view()),
     url(r"^oauth2-scoped-test/$", ScopedView.as_view()),
+    url(r"^oauth2-scoped-missing-auth/$", TokenHasScopeViewWrongAuth.as_view()),
     url(r"^oauth2-read-write-test/$", ReadWriteScopedView.as_view()),
     url(r"^oauth2-resource-scoped-test/$", ResourceScopedView.as_view()),
     url(r"^oauth2-authenticated-or-scoped-test/$", AuthenticatedOrScopedView.as_view()),
+    url(r"^oauth2-method-scope-test/.*$", MethodScopeAltView.as_view()),
+    url(r"^oauth2-method-scope-fail/$", MethodScopeAltViewBad.as_view()),
+    url(r"^oauth2-method-scope-missing-auth/$", MethodScopeAltViewWrongAuth.as_view()),
 ]
 
 
@@ -142,13 +183,19 @@ def test_authentication_or_scope_denied(self):
         self.assertEqual(response.status_code, 403)
 
     def test_scoped_permission_allow(self):
-        self.access_token.scope = "scope1"
+        self.access_token.scope = "scope1 another"
         self.access_token.save()
 
         auth = self._create_authorization_header(self.access_token.token)
         response = self.client.get("/oauth2-scoped-test/", HTTP_AUTHORIZATION=auth)
         self.assertEqual(response.status_code, 200)
 
+    def test_scope_missing_scope_attr(self):
+        auth = self._create_authorization_header("fake-token")
+        with self.assertRaises(AssertionError) as e:
+            self.client.get("/oauth2-scoped-missing-auth/", HTTP_AUTHORIZATION=auth)
+        self.assertTrue("`oauth2_provider.rest_framework.OAuth2Authentication`" in str(e.exception))
+
     def test_authenticated_or_scoped_permission_allow(self):
         self.access_token.scope = "scope1"
         self.access_token.save()
@@ -255,7 +302,7 @@ def test_required_scope_in_response(self):
         auth = self._create_authorization_header(self.access_token.token)
         response = self.client.get("/oauth2-scoped-test/", HTTP_AUTHORIZATION=auth)
         self.assertEqual(response.status_code, 403)
-        self.assertEqual(response.data["required_scopes"], ["scope1"])
+        self.assertEqual(response.data["required_scopes"], ["scope1", "another"])
 
     def test_required_scope_not_in_response_by_default(self):
         self.access_token.scope = "scope2"
@@ -265,3 +312,90 @@ def test_required_scope_not_in_response_by_default(self):
         response = self.client.get("/oauth2-scoped-test/", HTTP_AUTHORIZATION=auth)
         self.assertEqual(response.status_code, 403)
         self.assertNotIn("required_scopes", response.data)
+
+    def test_method_scope_alt_permission_get_allow(self):
+        self.access_token.scope = "read"
+        self.access_token.save()
+
+        auth = self._create_authorization_header(self.access_token.token)
+        response = self.client.get("/oauth2-method-scope-test/", HTTP_AUTHORIZATION=auth)
+        self.assertEqual(response.status_code, 200)
+
+    def test_method_scope_alt_permission_post_allow(self):
+        self.access_token.scope = "create"
+        self.access_token.save()
+
+        auth = self._create_authorization_header(self.access_token.token)
+        response = self.client.post("/oauth2-method-scope-test/", HTTP_AUTHORIZATION=auth)
+        self.assertEqual(response.status_code, 200)
+
+    def test_method_scope_alt_permission_put_allow(self):
+        self.access_token.scope = "edit update"
+        self.access_token.save()
+
+        auth = self._create_authorization_header(self.access_token.token)
+        response = self.client.put("/oauth2-method-scope-test/123", HTTP_AUTHORIZATION=auth)
+        self.assertEqual(response.status_code, 200)
+
+    def test_method_scope_alt_permission_put_fail(self):
+        self.access_token.scope = "edit"
+        self.access_token.save()
+
+        auth = self._create_authorization_header(self.access_token.token)
+        response = self.client.put("/oauth2-method-scope-test/123", HTTP_AUTHORIZATION=auth)
+        self.assertEqual(response.status_code, 403)
+
+    def test_method_scope_alt_permission_get_deny(self):
+        self.access_token.scope = "write"
+        self.access_token.save()
+
+        auth = self._create_authorization_header(self.access_token.token)
+        response = self.client.get("/oauth2-method-scope-test/", HTTP_AUTHORIZATION=auth)
+        self.assertEqual(response.status_code, 403)
+
+    def test_method_scope_alt_permission_post_deny(self):
+        self.access_token.scope = "read"
+        self.access_token.save()
+
+        auth = self._create_authorization_header(self.access_token.token)
+        response = self.client.post("/oauth2-method-scope-test/", HTTP_AUTHORIZATION=auth)
+        self.assertEqual(response.status_code, 403)
+
+    def test_method_scope_alt_no_token(self):
+        self.access_token.scope = ""
+        self.access_token.save()
+
+        auth = self._create_authorization_header(self.access_token.token)
+        self.access_token = None
+        response = self.client.post("/oauth2-method-scope-test/", HTTP_AUTHORIZATION=auth)
+        self.assertEqual(response.status_code, 403)
+
+    def test_method_scope_alt_missing_attr(self):
+        self.access_token.scope = "read"
+        self.access_token.save()
+
+        auth = self._create_authorization_header(self.access_token.token)
+        with self.assertRaises(ImproperlyConfigured):
+            self.client.post("/oauth2-method-scope-fail/", HTTP_AUTHORIZATION=auth)
+
+    def test_method_scope_alt_missing_patch_method(self):
+        self.access_token.scope = "update"
+        self.access_token.save()
+
+        auth = self._create_authorization_header(self.access_token.token)
+        response = self.client.patch("/oauth2-method-scope-test/", HTTP_AUTHORIZATION=auth)
+        self.assertEqual(response.status_code, 403)
+
+    def test_method_scope_alt_empty_scope(self):
+        self.access_token.scope = ""
+        self.access_token.save()
+
+        auth = self._create_authorization_header(self.access_token.token)
+        response = self.client.patch("/oauth2-method-scope-test/", HTTP_AUTHORIZATION=auth)
+        self.assertEqual(response.status_code, 403)
+
+    def test_method_scope_alt_missing_scope_attr(self):
+        auth = self._create_authorization_header("fake-token")
+        with self.assertRaises(AssertionError) as e:
+            self.client.get("/oauth2-method-scope-missing-auth/", HTTP_AUTHORIZATION=auth)
+        self.assertTrue("`oauth2_provider.rest_framework.OAuth2Authentication`" in str(e.exception))
