ropb implementation #137

apps/permission/backends.py

@@ -21,7 +21,7 @@ class PermissionBackend(ModelBackend):
     Manage permissions of users
     """
     supports_object_permissions = True
-    supports_anonymous_user = False
+    supports_anonymous_user = True
     supports_inactive_user = False
 
     @staticmethod
@@ -33,24 +33,18 @@ class PermissionBackend(ModelBackend):
         :param t: The type of the permissions: view, change, add or delete
         :return: The queryset of the permissions of the user (memoized) grouped by clubs
         """
-        if hasattr(request, 'auth') and request.auth is not None and hasattr(request.auth, 'scope'):
+        if hasattr(request, 'oauth2') and request.oauth2 is not None and 'scope' in request.oauth2:
             # OAuth2 Authentication
-            user = request.auth.user
+            user = request.oauth2['user']
 
             def permission_filter(membership_obj):
                 query = Q(pk=-1)
-                if 'mask' in request.GET:
-                    try:
-                        rank = int(request.GET['mask'])
-                    except ValueError:
-                        rank = 42
-                    query &= Q(mask__rank__lte=rank)
-                for scope in request.auth.scope.split(' '):
+                for scope in request.oauth2['scope']:
                     if scope == "openid":
                         continue
                     permission_id, club_id = scope.split('_')
                     if int(club_id) == membership_obj.club_id:
-                        query |= Q(pk=permission_id)
+                        query |= Q(pk=permission_id, mask__rank__lte=request.oauth2['mask'])
                 return query
         else:
             user = request.user

apps/permission/scopes.py

@@ -25,7 +25,9 @@ class PermissionScopes(BaseScopes):
         if 'scopes' in kwargs:
             for scope in kwargs['scopes']:
                 if scope == 'openid':
-                    scopes['openid'] = "OpenID Connect"
+                    scopes['openid'] = _("OpenID Connect (username and email)")
+                elif scope == '0_0':
+                    scopes['0_0'] = _("Useless scope which do nothing")
                 else:
                     p = Permission.objects.get(id=scope.split('_')[0])
                     club = Club.objects.get(id=scope.split('_')[1])
@@ -35,6 +37,7 @@ class PermissionScopes(BaseScopes):
         scopes = {f"{p.id}_{club.id}": f"{p.description} (club {club.name})"
                   for p in Permission.objects.all() for club in Club.objects.all()}
         scopes['openid'] = _("OpenID Connect (username and email)")
+        scopes['0_0'] = _("Useless scope which do nothing")
         return scopes
 
     def get_available_scopes(self, application=None, request=None, *args, **kwargs):
@@ -43,7 +46,7 @@ class PermissionScopes(BaseScopes):
         scopes = [f"{p.id}_{p.membership.club.id}"
                   for t in Permission.PERMISSION_TYPES
                   for p in PermissionBackend.get_raw_permissions(get_current_request(), t[0])]
-        scopes.append('openid')
+        scopes.append('0_0')  # always available
         return scopes
 
     def get_default_scopes(self, application=None, request=None, *args, **kwargs):
@@ -51,7 +54,7 @@ class PermissionScopes(BaseScopes):
             return []
         scopes = [f"{p.id}_{p.membership.club.id}"
                   for p in PermissionBackend.get_raw_permissions(get_current_request(), 'view')]
-        scopes.append('openid')
+        scopes.append('0_0')
         return scopes
 
 
@@ -73,6 +76,70 @@ class PermissionOAuth2Validator(OAuth2Validator):
         claims = super().get_discovery_claims(self)
         return claims + ["name", "normalized_name", "email"]
 
+    def validate_client_credentials_scopes(self, client_id, scopes, client, request, *args, **kwargs):
+        """
+        For client credentials valid scopes are scope of the app owner
+        """
+        valid_scopes = set()
+        request.oauth2 = {}
+        request.oauth2['user'] = client.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_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,
@@ -83,19 +150,20 @@ 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)
 
-        if 'openid' in scopes:
-            valid_scopes.add('openid')
+        if '0_0' in scopes:
+            valid_scopes.add('0_0')
 
         request.scopes = valid_scopes
         return valid_scopes

apps/permission/tests/test_oauth2_access.py

@@ -126,6 +126,7 @@ class OAuth2TestCase(TestCase):
                                **{'Authorization': f'Bearer {token.token}'})
 
         # Token is not granted to see other api
-        resp = self.client.get(f'/api/user/{self.user.pk}/',
+        resp = self.client.get(f'/api/members/profile/{self.user.profile.pk}/',
                                **{'Authorization': f'Bearer {token.token}'})
+
         self.assertEqual(resp.status_code, 404)

apps/permission/tests/test_oauth2_flow.py

@@ -3,24 +3,30 @@
 
 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
 from note.models import NoteUser
-from oauth2_provider.models import Application
+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,
@@ -81,14 +140,10 @@ class OAuth2TestCase(TestCase):
 
         self.assertEqual(resp.status_code, 200)
 
-        token = resp.json()['access_token']
+        token = AccessToken.objects.get(token=resp.json()['access_token'])
 
-        # Token is valid but has no right
-        resp = self.client.get('/api/user/{self.user.pk}',
-                               **{'Authorization': f'Bearer {token}'}
-                               )
-
-        self.assertEqual(resp.status_code, 403)
+        # Token do nothing, it should be have the useless scope
+        self.assertEqual(token.scope, '0_0')
 
         # RFC6749 4.4.2 allows use of scope in client credential flow
         resp = self.client.post('/o/token/',
@@ -100,13 +155,10 @@ class OAuth2TestCase(TestCase):
 
         self.assertEqual(resp.status_code, 200)
 
-        token = resp.json()['access_token']
+        token = AccessToken.objects.get(token=resp.json()['access_token'])
 
-        # Now app can see his creator
-        resp = self.client.post(f'/api/user/{self.user.pk}/',
-                                **{'Authorization': f'Bearer {token}'})
-
-        self.assertEqual(resp.status_code, 200)
+        # Token can have access, it shouldn't have the useless scope
+        self.assertEqual(token.scope, self.base_scope)
 
     def test_oidc_flow(self):
         """