mirror of
https://gitlab.crans.org/bde/nk20
synced 2025-11-15 19:17:42 +01:00
Compare commits
2 Commits
c09f133652
...
translatio
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
7e664c2d4e | ||
|
|
95fa3c1bc9 |
@@ -21,30 +21,28 @@ class PermissionBackend(ModelBackend):
|
|||||||
Manage permissions of users
|
Manage permissions of users
|
||||||
"""
|
"""
|
||||||
supports_object_permissions = True
|
supports_object_permissions = True
|
||||||
supports_anonymous_user = True
|
supports_anonymous_user = False
|
||||||
supports_inactive_user = False
|
supports_inactive_user = False
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
@memoize
|
@memoize
|
||||||
def get_raw_permissions(request, t): # noqa: C901
|
def get_raw_permissions(request, t):
|
||||||
"""
|
"""
|
||||||
Query permissions of a certain type for a user, then memoize it.
|
Query permissions of a certain type for a user, then memoize it.
|
||||||
:param request: The current request
|
:param request: The current request
|
||||||
:param t: The type of the permissions: view, change, add or delete
|
: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
|
:return: The queryset of the permissions of the user (memoized) grouped by clubs
|
||||||
"""
|
"""
|
||||||
if hasattr(request, 'oauth2') and request.oauth2 is not None and 'scope' in request.oauth2:
|
if hasattr(request, 'auth') and request.auth is not None and hasattr(request.auth, 'scope'):
|
||||||
# OAuth2 Authentication
|
# OAuth2 Authentication
|
||||||
user = request.oauth2['user']
|
user = request.auth.user
|
||||||
|
|
||||||
def permission_filter(membership_obj):
|
def permission_filter(membership_obj):
|
||||||
query = Q(pk=-1)
|
query = Q(pk=-1)
|
||||||
for scope in request.oauth2['scope']:
|
for scope in request.auth.scope.split(' '):
|
||||||
if scope == "openid":
|
|
||||||
continue
|
|
||||||
permission_id, club_id = scope.split('_')
|
permission_id, club_id = scope.split('_')
|
||||||
if int(club_id) == membership_obj.club_id:
|
if int(club_id) == membership_obj.club_id:
|
||||||
query |= Q(pk=permission_id, mask__rank__lte=request.oauth2['mask'])
|
query |= Q(pk=permission_id)
|
||||||
return query
|
return query
|
||||||
else:
|
else:
|
||||||
user = request.user
|
user = request.user
|
||||||
|
|||||||
@@ -10,8 +10,6 @@ from note_kfet.middlewares import get_current_request
|
|||||||
from .backends import PermissionBackend
|
from .backends import PermissionBackend
|
||||||
from .models import Permission
|
from .models import Permission
|
||||||
|
|
||||||
from django.utils.translation import gettext_lazy as _
|
|
||||||
|
|
||||||
|
|
||||||
class PermissionScopes(BaseScopes):
|
class PermissionScopes(BaseScopes):
|
||||||
"""
|
"""
|
||||||
@@ -25,9 +23,7 @@ class PermissionScopes(BaseScopes):
|
|||||||
if 'scopes' in kwargs:
|
if 'scopes' in kwargs:
|
||||||
for scope in kwargs['scopes']:
|
for scope in kwargs['scopes']:
|
||||||
if scope == 'openid':
|
if scope == 'openid':
|
||||||
scopes['openid'] = _("OpenID Connect (username and email)")
|
scopes['openid'] = "OpenID Connect"
|
||||||
elif scope == '0_0':
|
|
||||||
scopes['0_0'] = _("Useless scope which do nothing")
|
|
||||||
else:
|
else:
|
||||||
p = Permission.objects.get(id=scope.split('_')[0])
|
p = Permission.objects.get(id=scope.split('_')[0])
|
||||||
club = Club.objects.get(id=scope.split('_')[1])
|
club = Club.objects.get(id=scope.split('_')[1])
|
||||||
@@ -36,8 +32,7 @@ class PermissionScopes(BaseScopes):
|
|||||||
|
|
||||||
scopes = {f"{p.id}_{club.id}": f"{p.description} (club {club.name})"
|
scopes = {f"{p.id}_{club.id}": f"{p.description} (club {club.name})"
|
||||||
for p in Permission.objects.all() for club in Club.objects.all()}
|
for p in Permission.objects.all() for club in Club.objects.all()}
|
||||||
scopes['openid'] = _("OpenID Connect (username and email)")
|
scopes['openid'] = "OpenID Connect"
|
||||||
scopes['0_0'] = _("Useless scope which do nothing")
|
|
||||||
return scopes
|
return scopes
|
||||||
|
|
||||||
def get_available_scopes(self, application=None, request=None, *args, **kwargs):
|
def get_available_scopes(self, application=None, request=None, *args, **kwargs):
|
||||||
@@ -46,7 +41,7 @@ class PermissionScopes(BaseScopes):
|
|||||||
scopes = [f"{p.id}_{p.membership.club.id}"
|
scopes = [f"{p.id}_{p.membership.club.id}"
|
||||||
for t in Permission.PERMISSION_TYPES
|
for t in Permission.PERMISSION_TYPES
|
||||||
for p in PermissionBackend.get_raw_permissions(get_current_request(), t[0])]
|
for p in PermissionBackend.get_raw_permissions(get_current_request(), t[0])]
|
||||||
scopes.append('0_0') # always available
|
scopes.append('openid')
|
||||||
return scopes
|
return scopes
|
||||||
|
|
||||||
def get_default_scopes(self, application=None, request=None, *args, **kwargs):
|
def get_default_scopes(self, application=None, request=None, *args, **kwargs):
|
||||||
@@ -54,7 +49,7 @@ class PermissionScopes(BaseScopes):
|
|||||||
return []
|
return []
|
||||||
scopes = [f"{p.id}_{p.membership.club.id}"
|
scopes = [f"{p.id}_{p.membership.club.id}"
|
||||||
for p in PermissionBackend.get_raw_permissions(get_current_request(), 'view')]
|
for p in PermissionBackend.get_raw_permissions(get_current_request(), 'view')]
|
||||||
scopes.append('0_0')
|
scopes.append('openid')
|
||||||
return scopes
|
return scopes
|
||||||
|
|
||||||
|
|
||||||
@@ -76,70 +71,6 @@ class PermissionOAuth2Validator(OAuth2Validator):
|
|||||||
claims = super().get_discovery_claims(self)
|
claims = super().get_discovery_claims(self)
|
||||||
return claims + ["name", "normalized_name", "email"]
|
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):
|
def validate_scopes(self, client_id, scopes, client, request, *args, **kwargs):
|
||||||
"""
|
"""
|
||||||
User can request as many scope as he wants, including invalid scopes,
|
User can request as many scope as he wants, including invalid scopes,
|
||||||
@@ -150,11 +81,6 @@ class PermissionOAuth2Validator(OAuth2Validator):
|
|||||||
"""
|
"""
|
||||||
|
|
||||||
valid_scopes = set()
|
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)
|
|
||||||
|
|
||||||
|
|
||||||
for t in Permission.PERMISSION_TYPES:
|
for t in Permission.PERMISSION_TYPES:
|
||||||
for p in PermissionBackend.get_raw_permissions(get_current_request(), t[0]):
|
for p in PermissionBackend.get_raw_permissions(get_current_request(), t[0]):
|
||||||
@@ -162,8 +88,8 @@ class PermissionOAuth2Validator(OAuth2Validator):
|
|||||||
if scope in scopes:
|
if scope in scopes:
|
||||||
valid_scopes.add(scope)
|
valid_scopes.add(scope)
|
||||||
|
|
||||||
if '0_0' in scopes:
|
if 'openid' in scopes:
|
||||||
valid_scopes.add('0_0')
|
valid_scopes.add('openid')
|
||||||
|
|
||||||
request.scopes = valid_scopes
|
request.scopes = valid_scopes
|
||||||
return valid_scopes
|
return valid_scopes
|
||||||
|
|||||||
@@ -21,7 +21,6 @@ class OAuth2TestCase(TestCase):
|
|||||||
def setUp(self):
|
def setUp(self):
|
||||||
self.user = User.objects.create(
|
self.user = User.objects.create(
|
||||||
username="toto",
|
username="toto",
|
||||||
password="toto1234",
|
|
||||||
)
|
)
|
||||||
self.application = Application.objects.create(
|
self.application = Application.objects.create(
|
||||||
name="Test",
|
name="Test",
|
||||||
@@ -93,40 +92,3 @@ class OAuth2TestCase(TestCase):
|
|||||||
self.assertEqual(resp.status_code, 200)
|
self.assertEqual(resp.status_code, 200)
|
||||||
self.assertIn(self.application, resp.context['scopes'])
|
self.assertIn(self.application, resp.context['scopes'])
|
||||||
self.assertIn('1_1', resp.context['scopes'][self.application]) # Now the user has this permission
|
self.assertIn('1_1', resp.context['scopes'][self.application]) # Now the user has this permission
|
||||||
|
|
||||||
def test_oidc(self):
|
|
||||||
"""
|
|
||||||
Ensure OIDC work
|
|
||||||
"""
|
|
||||||
# Create access token that has access to our own user detail
|
|
||||||
token = AccessToken.objects.create(
|
|
||||||
user=self.user,
|
|
||||||
application=self.application,
|
|
||||||
scope="openid",
|
|
||||||
token=get_random_string(64),
|
|
||||||
expires=timezone.now() + timedelta(days=365),
|
|
||||||
)
|
|
||||||
|
|
||||||
# No access without token
|
|
||||||
resp = self.client.get('/o/userinfo/') # userinfo endpoint
|
|
||||||
self.assertEqual(resp.status_code, 401)
|
|
||||||
|
|
||||||
# Valid token
|
|
||||||
resp = self.client.get('/o/userinfo/', **{'Authorization': f'Bearer {token.token}'})
|
|
||||||
self.assertEqual(resp.status_code, 200)
|
|
||||||
|
|
||||||
# Create membership to test api
|
|
||||||
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"))
|
|
||||||
membership.save()
|
|
||||||
|
|
||||||
# Token can always be use to see yourself
|
|
||||||
resp = self.client.get('/api/me/',
|
|
||||||
**{'Authorization': f'Bearer {token.token}'})
|
|
||||||
|
|
||||||
# Token is not granted to see other api
|
|
||||||
resp = self.client.get(f'/api/members/profile/{self.user.profile.pk}/',
|
|
||||||
**{'Authorization': f'Bearer {token.token}'})
|
|
||||||
|
|
||||||
self.assertEqual(resp.status_code, 404)
|
|
||||||
@@ -1,167 +0,0 @@
|
|||||||
# Copyright (C) 2018-2025 by BDE ENS Paris-Saclay
|
|
||||||
# SPDX-License-Identifier: GPL-3.0-or-later
|
|
||||||
|
|
||||||
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, AccessToken
|
|
||||||
|
|
||||||
from ..models import Role, Permission
|
|
||||||
|
|
||||||
|
|
||||||
class OAuth2FlowTestCase(TestCase):
|
|
||||||
fixtures = ('initial', )
|
|
||||||
|
|
||||||
def setUp(self):
|
|
||||||
self.user_password = "toto1234"
|
|
||||||
hasher = PBKDF2PasswordHasher()
|
|
||||||
|
|
||||||
self.user = User.objects.create(
|
|
||||||
username="toto",
|
|
||||||
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"))
|
|
||||||
membership.save()
|
|
||||||
|
|
||||||
bde = Club.objects.get(name="BDE")
|
|
||||||
view_user_perm = Permission.objects.get(pk=1) # View own user detail
|
|
||||||
|
|
||||||
self.base_scope = f'{view_user_perm.pk}_{bde.pk}'
|
|
||||||
|
|
||||||
def test_oauth2_authorization_code_flow(self):
|
|
||||||
"""
|
|
||||||
Ensure OAuth2 Authorization Code Flow work
|
|
||||||
"""
|
|
||||||
pass
|
|
||||||
|
|
||||||
def test_oauth2_implicit_flow(self):
|
|
||||||
"""
|
|
||||||
Ensure OAuth2 Implicit Flow work
|
|
||||||
"""
|
|
||||||
pass
|
|
||||||
|
|
||||||
def test_oauth2_resource_owner_password_credentials_flow(self):
|
|
||||||
"""
|
|
||||||
Ensure OAuth2 Resource Owner Password Credentials Flow work
|
|
||||||
"""
|
|
||||||
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 client_credentials",
|
|
||||||
client_type=Application.CLIENT_CONFIDENTIAL,
|
|
||||||
authorization_grant_type=Application.GRANT_CLIENT_CREDENTIALS,
|
|
||||||
user=self.user,
|
|
||||||
hash_client_secret=False,
|
|
||||||
algorithm=Application.NO_ALGORITHM,
|
|
||||||
)
|
|
||||||
|
|
||||||
# No token without credential
|
|
||||||
resp = self.client.post('/o/token/',
|
|
||||||
data={"grant_type": "client_credentials"},
|
|
||||||
**{"Content-Type": 'application/x-www-form-urlencoded'}
|
|
||||||
)
|
|
||||||
|
|
||||||
self.assertEqual(resp.status_code, 401)
|
|
||||||
|
|
||||||
# Access with credential
|
|
||||||
credential = base64.b64encode(f'{app.client_id}:{app.client_secret}'.encode('utf-8')).decode()
|
|
||||||
|
|
||||||
resp = self.client.post('/o/token/',
|
|
||||||
data={"grant_type": "client_credentials"},
|
|
||||||
**{'HTTP_Authorization': f'Basic {credential}',
|
|
||||||
"Content-Type": 'application/x-www-form-urlencoded'}
|
|
||||||
)
|
|
||||||
|
|
||||||
self.assertEqual(resp.status_code, 200)
|
|
||||||
|
|
||||||
token = AccessToken.objects.get(token=resp.json()['access_token'])
|
|
||||||
|
|
||||||
# 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/',
|
|
||||||
data={"grant_type": "client_credentials",
|
|
||||||
"scope": self.base_scope},
|
|
||||||
**{'http_Authorization': f'Basic {credential}',
|
|
||||||
"Content-Type": 'application/x-www-form-urlencoded'}
|
|
||||||
)
|
|
||||||
|
|
||||||
self.assertEqual(resp.status_code, 200)
|
|
||||||
|
|
||||||
token = AccessToken.objects.get(token=resp.json()['access_token'])
|
|
||||||
|
|
||||||
# Token can have access, it shouldn't have the useless scope
|
|
||||||
self.assertEqual(token.scope, self.base_scope)
|
|
||||||
|
|
||||||
def test_oidc_flow(self):
|
|
||||||
"""
|
|
||||||
Ensure OIDC Flow work
|
|
||||||
"""
|
|
||||||
pass
|
|
||||||
@@ -4399,7 +4399,7 @@ msgstr "Géré par le BDE"
|
|||||||
|
|
||||||
#: note_kfet/templates/base.html:231
|
#: note_kfet/templates/base.html:231
|
||||||
msgid "Hosted by Cr@ns"
|
msgid "Hosted by Cr@ns"
|
||||||
msgstr "Hébergé par le Cr@ans"
|
msgstr "Hébergé par le Cr@ns"
|
||||||
|
|
||||||
#: note_kfet/templates/base.html:273
|
#: note_kfet/templates/base.html:273
|
||||||
msgid "The note is not available for now"
|
msgid "The note is not available for now"
|
||||||
|
|||||||
Reference in New Issue
Block a user