ropb implementation #137

apps/permission/scopes.py

@@ -107,6 +107,39 @@ class PermissionOAuth2Validator(OAuth2Validator):
         request.scopes = valid_scopes
         return valid_scopes
 
+    def validate_ropb_scopes(self, client_id, scopes, client, request, *args, **kwargs):
+        """
+        For ROPB valid scopes are scope of the user
+        """
+        valid_scopes = set()
+        request.oauth2 = {}
+        request.oauth2['user'] = request.user
+        request.oauth2['user'].is_anomymous = False
+        request.oauth2['scope'] = scopes
+        # mask implementation
+        if hasattr(request.decoded_body, 'mask'):
+            try:
+                request.oauth2['mask'] = int(request.decoded_body['mask'])
+            except ValueError:
+                request.oauth2['mask'] = 42
+        else:
+            request.oauth2['mask'] = 42
+
+        for t in Permission.PERMISSION_TYPES:
+            for p in PermissionBackend.get_raw_permissions(request, t[0]):
+                scope = f"{p.id}_{p.membership.club.id}"
+                if scope in scopes:
+                    valid_scopes.add(scope)
+
+        # Always give one scope to generate token
+        if not valid_scopes:
+            valid_scopes.add('0_0')
+
+        request.scopes = valid_scopes
+        return valid_scopes
+
+
+
     def validate_scopes(self, client_id, scopes, client, request, *args, **kwargs):
         """
         User can request as many scope as he wants, including invalid scopes,
@@ -117,16 +150,14 @@ class PermissionOAuth2Validator(OAuth2Validator):
         """
 
         valid_scopes = set()
-
         if hasattr(request, 'grant_type') and request.grant_type == 'client_credentials':
             return self.validate_client_credentials_scopes(client_id, scopes, client, request, args, kwargs)
+        if hasattr(request, 'grant_type') and request.grant_type == 'password':
+            return self.validate_ropb_scopes(client_id, scopes, client, request, args, kwargs)
+
 
-        # simple patch for have functionnal ROPB flow
-        # TODO rewrite
-        r = get_current_request()
-        r.user = request.user
         for t in Permission.PERMISSION_TYPES:
-            for p in PermissionBackend.get_raw_permissions(r, t[0]):
+            for p in PermissionBackend.get_raw_permissions(get_current_request(), t[0]):
                 scope = f"{p.id}_{p.membership.club.id}"
                 if scope in scopes:
                     valid_scopes.add(scope)

apps/permission/tests/test_oauth2_flow.py

@@ -3,6 +3,7 @@
 
 import base64
 
+from django.contrib.auth.hashers import PBKDF2PasswordHasher
 from django.contrib.auth.models import User
 from django.test import TestCase
 from member.models import Membership, Club
@@ -12,15 +13,20 @@ from oauth2_provider.models import Application, AccessToken
 from ..models import Role, Permission
 
 
-class OAuth2TestCase(TestCase):
+class OAuth2FlowTestCase(TestCase):
     fixtures = ('initial', )
 
     def setUp(self):
+        self.user_password = "toto1234"
+        hasher = PBKDF2PasswordHasher()
+        
         self.user = User.objects.create(
             username="toto",
-            password="toto1234",
+            password=hasher.encode(self.user_password, hasher.salt()),
         )
 
+        
+
         NoteUser.objects.create(user=self.user)
         membership = Membership.objects.create(user=self.user, club_id=1)
         membership.roles.add(Role.objects.get(name="Adhérent⋅e BDE"))
@@ -47,14 +53,67 @@ class OAuth2TestCase(TestCase):
         """
         Ensure OAuth2 Resource Owner Password Credentials Flow work
         """
-        pass
+        app = Application.objects.create(
+            name="Test ROPB",
+            client_type=Application.CLIENT_CONFIDENTIAL,
+            authorization_grant_type=Application.GRANT_PASSWORD,
+            user=self.user,
+            hash_client_secret=False,
+            algorithm=Application.NO_ALGORITHM,
+        )
+        
+        credential = base64.b64encode(f'{app.client_id}:{app.client_secret}'.encode('utf-8')).decode()
+
+        # No token without real password
+        resp = self.client.post('/o/token/',
+                                data={"grant_type": "password",
+                                      "username": self.user,
+                                      "password": "password"},
+                                **{"Content-Type": 'application/x-www-form-urlencoded',
+                                   "Http_Authorization": f'Basic {credential}'}
+                                )
+
+        self.assertEqual(resp.status_code, 400)
+
+        resp = self.client.post('/o/token/',
+                                data={"grant_type": "password",
+                                      "username": self.user,
+                                      "password": self.user_password},
+                                **{"Content-Type": 'application/x-www-form-urlencoded',
+                                   "HTTP_Authorization": f'Basic {credential}'}
+                                )
+        
+        self.assertEqual(resp.status_code, 200)
+
+        access_token = AccessToken.objects.get(token=resp.json()['access_token'])
+        self.assertEqual('refresh_token' in resp.json(), True)
+
+        self.assertEqual(access_token.scope, '0_0') # token do nothing
+
+        # RFC6749 4.3.2 allows use of scope in ROPB token access request
+        
+        resp = self.client.post('/o/token/',
+                                data={"grant_type": "password",
+                                      #"client_id": app.client_id,
+                                      "username": self.user,
+                                      "password": self.user_password,
+                                      "scope": self.base_scope},
+                                **{"Content-Type": 'application/x-www-form-urlencoded',
+                                   "HTTP_Authorization": f'Basic {credential}'}
+                                )
+
+        token = AccessToken.objects.get(token=resp.json()['access_token'])
+
+        self.assertEqual(token.scope, self.base_scope) # token do nothing more than base_scope
+
+
 
     def test_oauth2_client_credentials(self):
         """
         Ensure OAuth2 Client Credentials work
         """
         app = Application.objects.create(
-            name="Test credentials",
+            name="Test client_credentials",
             client_type=Application.CLIENT_CONFIDENTIAL,
             authorization_grant_type=Application.GRANT_CLIENT_CREDENTIALS,
             user=self.user,