fixed bugs and better tests

Author: Massimiliano

oauth2_provider/decorators.py

@@ -9,55 +9,60 @@
 from .settings import oauth2_settings
 
 
-def protected_resource(view_func, scopes=None, validator_cls=OAuth2Validator, server_cls=Server):
+def protected_resource(scopes=None, validator_cls=OAuth2Validator, server_cls=Server):
     """
-    Decorator to protect views by providing OAuth2 authentication out of the box, optionally with scope handling.
+    Decorator to protect views by providing OAuth2 authentication out of the box, optionally with
+    scope handling.
     """
-    if scopes is None:
-        scopes = []
-    
-    @wraps(view_func)
-    def _validate(request, *args, **kwargs):
-        validator = validator_cls()
-        core = OAuthLibCore(server_cls(validator))
-        valid, oauthlib_req = core.verify_request(request, scopes=scopes)
-        if valid:
-            return view_func(request, *args, **kwargs)
-        return HttpResponseForbidden()
-    return _validate
-
-
-def rw_protected_resource(view_func, scopes=None, validator_cls=OAuth2Validator, server_cls=Server):
+    _scopes = scopes or []
+
+    def decorator(view_func):
+        @wraps(view_func)
+        def _validate(request, *args, **kwargs):
+            validator = validator_cls()
+            core = OAuthLibCore(server_cls(validator))
+            valid, oauthlib_req = core.verify_request(request, scopes=_scopes)
+            if valid:
+                return view_func(request, *args, **kwargs)
+            return HttpResponseForbidden()
+        return _validate
+    return decorator
+
+
+def rw_protected_resource(scopes=None, validator_cls=OAuth2Validator, server_cls=Server):
     """
-    Decorator to protect views by providing OAuth2 authentication and read/write scopes out of the box.
+    Decorator to protect views by providing OAuth2 authentication and read/write scopes out of the
+    box.
     GET, HEAD, OPTIONS http methods require "read" scope. Otherwise "write" scope is required.
     """
-    if scopes is None:
-        scopes = []
-
-    @wraps(view_func)
-    def _validate(request, *args, **kwargs):
-        # Check if provided scopes are acceptable
-        provided_scopes = oauth2_settings._SCOPES
-        read_write_scopes = [oauth2_settings.READ_SCOPE, oauth2_settings.WRITE_SCOPE]
-
-        if not set(read_write_scopes).issubset(set(provided_scopes)):
-            raise ImproperlyConfigured(
-                "rw_protected_resource decorator requires following scopes {0}"
-                " to be in OAUTH2_PROVIDER['SCOPES'] list in settings".format(read_write_scopes)
-            )
-
-        # Check if method is safe
-        if request.method.upper() in ['GET', 'HEAD', 'OPTIONS']:
-            scopes.append(oauth2_settings.READ_SCOPE)
-        else:
-            scopes.append(oauth2_settings.WRITE_SCOPE)
-
-        # proceed with validation
-        validator = validator_cls()
-        core = OAuthLibCore(server_cls(validator))
-        valid, oauthlib_req = core.verify_request(request, scopes=scopes)
-        if valid:
-            return view_func(request, *args, **kwargs)
-        return HttpResponseForbidden()
-    return _validate
+    _scopes = scopes or []
+
+    def decorator(view_func):
+        @wraps(view_func)
+        def _validate(request, *args, **kwargs):
+            # Check if provided scopes are acceptable
+            provided_scopes = oauth2_settings._SCOPES
+            read_write_scopes = [oauth2_settings.READ_SCOPE, oauth2_settings.WRITE_SCOPE]
+
+            if not set(read_write_scopes).issubset(set(provided_scopes)):
+                raise ImproperlyConfigured(
+                    "rw_protected_resource decorator requires following scopes {0}"
+                    " to be in OAUTH2_PROVIDER['SCOPES'] list in settings".format(
+                        read_write_scopes)
+                )
+
+            # Check if method is safe
+            if request.method.upper() in ['GET', 'HEAD', 'OPTIONS']:
+                _scopes.append(oauth2_settings.READ_SCOPE)
+            else:
+                _scopes.append(oauth2_settings.WRITE_SCOPE)
+
+            # proceed with validation
+            validator = validator_cls()
+            core = OAuthLibCore(server_cls(validator))
+            valid, oauthlib_req = core.verify_request(request, scopes=_scopes)
+            if valid:
+                return view_func(request, *args, **kwargs)
+            return HttpResponseForbidden()
+        return _validate
+    return decorator

oauth2_provider/tests/test_decorators.py

@@ -36,10 +36,18 @@ def setUp(self):
             application=self.application
         )
 
+        self.another_access_token = AccessToken.objects.create(
+            user=self.user,
+            scope='can_touch_this',
+            expires=timezone.now() + timedelta(seconds=300),
+            token='secret-access-token-key2',
+            application=self.application
+        )
+
         oauth2_settings._SCOPES = ['read', 'write']
 
     def test_access_denied(self):
-        @protected_resource
+        @protected_resource()
         def view(request, *args, **kwargs):
             return 'protected contents'
 
@@ -48,33 +56,45 @@ def view(request, *args, **kwargs):
         self.assertEqual(response.status_code, 403)
 
     def test_access_allowed(self):
-        @protected_resource
+        @protected_resource()
         def view(request, *args, **kwargs):
             return 'protected contents'
 
+        @protected_resource(scopes=['can_touch_this'])
+        def scoped_view(request, *args, **kwargs):
+            return 'moar protected contents'
+
         auth_headers = {
             'HTTP_AUTHORIZATION': 'Bearer ' + self.access_token.token,
         }
         request = self.request_factory.get("/fake-resource", **auth_headers)
         response = view(request)
         self.assertEqual(response, "protected contents")
 
+        # now with scopes
+        auth_headers = {
+            'HTTP_AUTHORIZATION': 'Bearer ' + self.another_access_token.token,
+        }
+        request = self.request_factory.get("/fake-resource", **auth_headers)
+        response = scoped_view(request)
+        self.assertEqual(response, "moar protected contents")
+
     def test_rw_protected(self):
         self.access_token.scope = 'read'
         self.access_token.save()
         auth_headers = {
             'HTTP_AUTHORIZATION': 'Bearer ' + self.access_token.token,
         }
 
-        @rw_protected_resource
+        @rw_protected_resource()
         def scoped_view(request, *args, **kwargs):
             return 'other protected contents'
 
         request = self.request_factory.post("/fake-resource", **auth_headers)
         response = scoped_view(request)
         self.assertEqual(response.status_code, 403)
 
-        @rw_protected_resource
+        @rw_protected_resource()
         def scoped_view(request, *args, **kwargs):
             return 'other protected contents'