mirror of
https://gitlab.crans.org/bde/nk20
synced 2025-11-14 18:51:27 +01:00
Compare commits
10 Commits
c32c4eaf10
...
oauth2
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
13171899c2 | ||
|
|
dacedbff20 | ||
|
|
a61a4667b9 | ||
|
|
9998189dbf | ||
|
|
08593700fc | ||
|
|
54d28b30e5 | ||
|
|
c09f133652 | ||
|
|
bfd50e3cd5 | ||
|
|
68341a2a7e | ||
|
|
d2cc1b902d |
@@ -26,24 +26,42 @@ class PermissionBackend(ModelBackend):
|
||||
|
||||
@staticmethod
|
||||
@memoize
|
||||
def get_raw_permissions(request, t):
|
||||
def get_raw_permissions(request, t): # noqa: C901
|
||||
"""
|
||||
Query permissions of a certain type for a user, then memoize it.
|
||||
:param request: The current request
|
||||
: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'):
|
||||
# Permission for auth
|
||||
if hasattr(request, 'oauth2') and request.oauth2 is not None and 'scope' in request.oauth2:
|
||||
# OAuth2 Authentication
|
||||
user = request.oauth2['user']
|
||||
|
||||
def permission_filter(membership_obj):
|
||||
query = Q(pk=-1)
|
||||
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, mask__rank__lte=request.oauth2['mask'])
|
||||
return query
|
||||
|
||||
# Restreint token permission to his scope
|
||||
elif hasattr(request, 'auth') and request.auth is not None and hasattr(request.auth, 'scope'):
|
||||
user = request.auth.user
|
||||
|
||||
def permission_filter(membership_obj):
|
||||
query = Q(pk=-1)
|
||||
for scope in request.auth.scope.split(' '):
|
||||
if scope == "openid" or scope == "0_0":
|
||||
continue
|
||||
permission_id, club_id = scope.split('_')
|
||||
if int(club_id) == membership_obj.club_id:
|
||||
query |= Q(pk=permission_id)
|
||||
return query
|
||||
|
||||
else:
|
||||
user = request.user
|
||||
|
||||
@@ -77,7 +95,6 @@ class PermissionBackend(ModelBackend):
|
||||
:param type: The type of the permissions: view, change, add or delete
|
||||
:return: A generator of the requested permissions
|
||||
"""
|
||||
|
||||
if hasattr(request, 'auth') and request.auth is not None and hasattr(request.auth, 'scope'):
|
||||
# OAuth2 Authentication
|
||||
user = request.auth.user
|
||||
|
||||
@@ -10,6 +10,8 @@ from note_kfet.middlewares import get_current_request
|
||||
from .backends import PermissionBackend
|
||||
from .models import Permission
|
||||
|
||||
from django.utils.translation import gettext_lazy as _
|
||||
|
||||
|
||||
class PermissionScopes(BaseScopes):
|
||||
"""
|
||||
@@ -23,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])
|
||||
@@ -32,7 +36,8 @@ 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"
|
||||
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):
|
||||
@@ -41,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):
|
||||
@@ -49,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 = ['0_0'] # always default
|
||||
return scopes
|
||||
|
||||
|
||||
@@ -67,10 +72,77 @@ class PermissionOAuth2Validator(OAuth2Validator):
|
||||
"email": request.user.email,
|
||||
}
|
||||
|
||||
def get_userinfo_claims(self, request):
|
||||
claims = super().get_userinfo_claims(request)
|
||||
claims['is_active'] = request.user.is_active
|
||||
return claims
|
||||
|
||||
def get_discovery_claims(self, request):
|
||||
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,
|
||||
@@ -79,17 +151,35 @@ class PermissionOAuth2Validator(OAuth2Validator):
|
||||
This allows clients to request more permission to get finally a
|
||||
subset of permissions.
|
||||
"""
|
||||
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)
|
||||
|
||||
# Authorization code and Implicit are the same for scope, OIDC it's only a layer
|
||||
|
||||
valid_scopes = set()
|
||||
req = get_current_request()
|
||||
request.oauth2 = {}
|
||||
request.oauth2['user'] = req.user
|
||||
request.oauth2['scope'] = scopes
|
||||
# mask implementation
|
||||
request.oauth2['mask'] = req.session.load()['permission_mask']
|
||||
|
||||
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(request, t[0]):
|
||||
scope = f"{p.id}_{p.membership.club.id}"
|
||||
if scope in scopes:
|
||||
valid_scopes.add(scope)
|
||||
|
||||
if 'openid' in scopes:
|
||||
# We grant openid scope if user is active
|
||||
if 'openid' in scopes and req.user.is_active:
|
||||
valid_scopes.add('openid')
|
||||
|
||||
# Always give one scope to generate token
|
||||
if not valid_scopes:
|
||||
valid_scopes.add('0_0')
|
||||
|
||||
request.scopes = valid_scopes
|
||||
return valid_scopes
|
||||
|
||||
@@ -21,6 +21,7 @@ class OAuth2TestCase(TestCase):
|
||||
def setUp(self):
|
||||
self.user = User.objects.create(
|
||||
username="toto",
|
||||
password="toto1234",
|
||||
)
|
||||
self.application = Application.objects.create(
|
||||
name="Test",
|
||||
@@ -92,3 +93,40 @@ class OAuth2TestCase(TestCase):
|
||||
self.assertEqual(resp.status_code, 200)
|
||||
self.assertIn(self.application, resp.context['scopes'])
|
||||
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)
|
||||
444
apps/permission/tests/test_oauth2_flow.py
Normal file
444
apps/permission/tests/test_oauth2_flow.py
Normal file
@@ -0,0 +1,444 @@
|
||||
# Copyright (C) 2018-2025 by BDE ENS Paris-Saclay
|
||||
# SPDX-License-Identifier: GPL-3.0-or-later
|
||||
|
||||
import base64
|
||||
import hashlib
|
||||
|
||||
from django.contrib.auth.hashers import PBKDF2PasswordHasher
|
||||
from django.contrib.auth.models import User
|
||||
from django.utils.crypto import get_random_string
|
||||
from django.test import TestCase
|
||||
from member.models import Membership, Club
|
||||
from note.models import NoteUser
|
||||
from oauth2_provider.models import Application, AccessToken, Grant
|
||||
|
||||
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
|
||||
"""
|
||||
|
||||
app = Application.objects.create(
|
||||
name="Test Authorization Code",
|
||||
client_type=Application.CLIENT_CONFIDENTIAL,
|
||||
authorization_grant_type=Application.GRANT_AUTHORIZATION_CODE,
|
||||
user=self.user,
|
||||
hash_client_secret=False,
|
||||
redirect_uris='http://127.0.0.1:8000/noexist/callback',
|
||||
algorithm=Application.NO_ALGORITHM,
|
||||
)
|
||||
|
||||
credential = base64.b64encode(f'{app.client_id}:{app.client_secret}'.encode('utf-8')).decode()
|
||||
|
||||
############################
|
||||
# Minimal RFC6749 requests #
|
||||
############################
|
||||
|
||||
resp = self.client.get('/o/authorize/',
|
||||
data={"response_type": "code", # REQUIRED
|
||||
"client_id": app.client_id}, # REQUIRED
|
||||
**{"Content-Type": 'application/x-www-form-urlencoded'})
|
||||
|
||||
# Get user authorization
|
||||
|
||||
##################################################################################
|
||||
|
||||
url = resp.url
|
||||
csrf_token = resp.text.split('CSRF_TOKEN = "')[0].split('"')[0]
|
||||
|
||||
resp = self.client.post(url,
|
||||
data={"username": self.user.username,
|
||||
"password": self.user_password,
|
||||
"permission_mask": 1,
|
||||
"csrfmiddlewaretoken": csrf_token})
|
||||
|
||||
url = resp.url
|
||||
resp = self.client.get(url)
|
||||
|
||||
csrf_token = resp.text.split('CSRF_TOKEN = "')[0].split('"')[0]
|
||||
|
||||
resp = self.client.post(url,
|
||||
follow=True,
|
||||
data={"allow": "Authorize",
|
||||
"scope": "0_0",
|
||||
"csrfmiddlewaretoken": csrf_token,
|
||||
"response_type": "code",
|
||||
"client_id": app.client_id,
|
||||
"redirect_uri": app.redirect_uris})
|
||||
|
||||
keys = resp.request['QUERY_STRING'].split("&")
|
||||
for key in keys:
|
||||
if len(key.split('code=')) == 2:
|
||||
code = key.split('code=')[1]
|
||||
|
||||
##################################################################################
|
||||
|
||||
grant = Grant.objects.get(code=code)
|
||||
self.assertEqual(grant.scope, '0_0')
|
||||
|
||||
# Now we can ask an Access Token
|
||||
|
||||
resp = self.client.post('/o/token/',
|
||||
data={"grant_type": 'authorization_code', # REQUIRED
|
||||
"code": code}, # REQUIRED
|
||||
**{"Content-Type": 'application/x-www-form-urlencoded',
|
||||
"HTTP_Authorization": f'Basic {credential}'})
|
||||
|
||||
# We should have refresh token
|
||||
self.assertEqual('refresh_token' in resp.json(), True)
|
||||
|
||||
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')
|
||||
|
||||
# Logout user
|
||||
self.client.logout()
|
||||
|
||||
#############################################
|
||||
# Maximal RFC6749 + RFC7636 (PKCE) requests #
|
||||
#############################################
|
||||
|
||||
state = get_random_string(32)
|
||||
|
||||
# PKCE
|
||||
code_verifier = get_random_string(100) # 43-128 characters [A-Z,a-z,0-9,"-",".","_","~"]
|
||||
code_challenge = hashlib.sha256(code_verifier.encode('utf-8')).digest()
|
||||
code_challenge = base64.urlsafe_b64encode(code_challenge).decode('utf-8').replace('=', '')
|
||||
cc_method = "S256"
|
||||
|
||||
resp = self.client.get('/o/authorize/',
|
||||
data={"response_type": "code", # REQUIRED
|
||||
"code_challenge": code_challenge, # PKCE REQUIRED
|
||||
"code_challenge_method": cc_method, # PKCE REQUIRED
|
||||
"client_id": app.client_id, # REQUIRED
|
||||
"redirect_uri": app.redirect_uris, # OPTIONAL
|
||||
"scope": self.base_scope, # OPTIONAL
|
||||
"state": state}, # RECOMMENDED
|
||||
**{"Content-Type": 'application/x-www-form-urlencoded'})
|
||||
|
||||
# Get user authorization
|
||||
##################################################################################
|
||||
url = resp.url
|
||||
csrf_token = resp.text.split('CSRF_TOKEN = "')[0].split('"')[0]
|
||||
|
||||
resp = self.client.post(url,
|
||||
data={"username": self.user.username,
|
||||
"password": self.user_password,
|
||||
"permission_mask": 1,
|
||||
"csrfmiddlewaretoken": csrf_token})
|
||||
|
||||
url = resp.url
|
||||
resp = self.client.get(url)
|
||||
|
||||
csrf_token = resp.text.split('CSRF_TOKEN = "')[0].split('"')[0]
|
||||
|
||||
resp = self.client.post(url,
|
||||
follow=True,
|
||||
data={"allow": "Authorize",
|
||||
"scope": self.base_scope,
|
||||
"csrfmiddlewaretoken": csrf_token,
|
||||
"response_type": "code",
|
||||
"code_challenge": code_challenge,
|
||||
"code_challenge_method": cc_method,
|
||||
"client_id": app.client_id,
|
||||
"state": state,
|
||||
"redirect_uri": app.redirect_uris})
|
||||
|
||||
keys = resp.request['QUERY_STRING'].split("&")
|
||||
for key in keys:
|
||||
if len(key.split('code=')) == 2:
|
||||
code = key.split('code=')[1]
|
||||
if len(key.split('state=')) == 2:
|
||||
resp_state = key.split('state=')[1]
|
||||
|
||||
##################################################################################
|
||||
|
||||
grant = Grant.objects.get(code=code)
|
||||
self.assertEqual(grant.scope, self.base_scope)
|
||||
self.assertEqual(state, resp_state)
|
||||
|
||||
# Now we can ask an Access Token
|
||||
|
||||
resp = self.client.post('/o/token/',
|
||||
data={"grant_type": 'authorization_code', # REQUIRED
|
||||
"code": code, # REQUIRED
|
||||
"code_verifier": code_verifier, # PKCE REQUIRED
|
||||
"redirect_uri": app.redirect_uris}, # REQUIRED
|
||||
**{"Content-Type": 'application/x-www-form-urlencoded',
|
||||
"HTTP_Authorization": f'Basic {credential}'})
|
||||
|
||||
# We should have refresh token
|
||||
self.assertEqual('refresh_token' in resp.json(), True)
|
||||
|
||||
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_oauth2_implicit_flow(self):
|
||||
"""
|
||||
Ensure OAuth2 Implicit Flow work
|
||||
"""
|
||||
app = Application.objects.create(
|
||||
name="Test Implicit Flow",
|
||||
client_type=Application.CLIENT_CONFIDENTIAL,
|
||||
authorization_grant_type=Application.GRANT_IMPLICIT,
|
||||
user=self.user,
|
||||
hash_client_secret=False,
|
||||
algorithm=Application.NO_ALGORITHM,
|
||||
redirect_uris='http://127.0.0.1:8000/noexist/callback/',
|
||||
)
|
||||
|
||||
############################
|
||||
# Minimal RFC6749 requests #
|
||||
############################
|
||||
|
||||
resp = self.client.get('/o/authorize/',
|
||||
data={'response_type': 'token', # REQUIRED
|
||||
'client_id': app.client_id}, # REQUIRED
|
||||
**{"Content-Type": 'application/x-www-form-urlencoded'}
|
||||
)
|
||||
|
||||
# Get user authorization
|
||||
##################################################################################
|
||||
url = resp.url
|
||||
csrf_token = resp.text.split('CSRF_TOKEN = "')[0].split('"')[0]
|
||||
|
||||
resp = self.client.post(url,
|
||||
data={"username": self.user.username,
|
||||
"password": self.user_password,
|
||||
"permission_mask": 1,
|
||||
"csrfmiddlewaretoken": csrf_token})
|
||||
|
||||
url = resp.url
|
||||
resp = self.client.get(url)
|
||||
|
||||
csrf_token = resp.text.split('CSRF_TOKEN = "')[0].split('"')[0]
|
||||
|
||||
resp = self.client.post(url,
|
||||
follow=True,
|
||||
data={"allow": "Authorize",
|
||||
"scope": '0_0',
|
||||
"csrfmiddlewaretoken": csrf_token,
|
||||
"response_type": "token",
|
||||
"client_id": app.client_id,
|
||||
"redirect_uri": app.redirect_uris})
|
||||
|
||||
url = resp.redirect_chain[0][0]
|
||||
keys = url.split('#')[1]
|
||||
refresh_token = ''
|
||||
|
||||
for couple in keys.split('&'):
|
||||
if couple.split('=')[0] == 'access_token':
|
||||
token = couple.split('=')[1]
|
||||
if couple.split('=')[0] == 'refresh_token':
|
||||
refresh_token = couple.split('=')[1]
|
||||
|
||||
##################################################################################
|
||||
|
||||
self.assertEqual(refresh_token, '')
|
||||
|
||||
access_token = AccessToken.objects.get(token=token)
|
||||
|
||||
# Token do nothing, it should be have the useless scope
|
||||
self.assertEqual(access_token.scope, '0_0')
|
||||
|
||||
# Logout user
|
||||
self.client.logout()
|
||||
|
||||
############################
|
||||
# Maximal RFC6749 requests #
|
||||
############################
|
||||
|
||||
state = get_random_string(32)
|
||||
|
||||
resp = self.client.get('/o/authorize/',
|
||||
data={'response_type': 'token', # REQUIRED
|
||||
'client_id': app.client_id, # REQUIRED
|
||||
'redirect_uri': app.redirect_uris, # OPTIONAL
|
||||
'scope': self.base_scope, # OPTIONAL
|
||||
'state': state}, # RECOMMENDED
|
||||
**{"Content-Type": 'application/x-www-form-urlencoded'}
|
||||
)
|
||||
|
||||
# Get user authorization
|
||||
##################################################################################
|
||||
url = resp.url
|
||||
csrf_token = resp.text.split('CSRF_TOKEN = "')[0].split('"')[0]
|
||||
|
||||
resp = self.client.post(url,
|
||||
data={"username": self.user.username,
|
||||
"password": self.user_password,
|
||||
"permission_mask": 1,
|
||||
"csrfmiddlewaretoken": csrf_token})
|
||||
|
||||
url = resp.url
|
||||
resp = self.client.get(url)
|
||||
|
||||
csrf_token = resp.text.split('CSRF_TOKEN = "')[0].split('"')[0]
|
||||
|
||||
resp = self.client.post(url,
|
||||
follow=True,
|
||||
data={"allow": "Authorize",
|
||||
"scope": self.base_scope,
|
||||
"state": state,
|
||||
"csrfmiddlewaretoken": csrf_token,
|
||||
"response_type": "token",
|
||||
"client_id": app.client_id,
|
||||
"redirect_uri": app.redirect_uris})
|
||||
|
||||
url = resp.redirect_chain[0][0]
|
||||
keys = url.split('#')[1]
|
||||
refresh_token = ''
|
||||
|
||||
for couple in keys.split('&'):
|
||||
if couple.split('=')[0] == 'access_token':
|
||||
token = couple.split('=')[1]
|
||||
if couple.split('=')[0] == 'refresh_token':
|
||||
refresh_token = couple.split('=')[1]
|
||||
if couple.split('=')[0] == 'state':
|
||||
resp_state = couple.split('=')[1]
|
||||
|
||||
##################################################################################
|
||||
|
||||
self.assertEqual(refresh_token, '')
|
||||
|
||||
access_token = AccessToken.objects.get(token=token)
|
||||
|
||||
# Token can have access, it shouldn't have the useless scope
|
||||
self.assertEqual(access_token.scope, self.base_scope)
|
||||
|
||||
self.assertEqual(state, resp_state)
|
||||
|
||||
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", # REQUIRED
|
||||
"username": self.user, # REQUIRED
|
||||
"password": "password"}, # REQUIRED
|
||||
**{"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", # REQUIRED
|
||||
"username": self.user, # REQUIRED
|
||||
"password": self.user_password}, # REQUIRED
|
||||
**{"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", # REQUIRED
|
||||
"username": self.user, # REQUIRED
|
||||
"password": self.user_password, # REQUIRED
|
||||
"scope": self.base_scope}, # OPTIONAL
|
||||
**{"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"}, # REQUIRED
|
||||
**{"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"}, # REQUIRED
|
||||
**{'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", # REQUIRED
|
||||
"scope": self.base_scope}, # OPTIONAL
|
||||
**{'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)
|
||||
@@ -84,7 +84,7 @@ Le script *generate_wrapped* fonctionne de la manière suivante :
|
||||
wrapped·s va/vont être généré·s
|
||||
ou regénéré·s.
|
||||
* ``global_data`` : le script génére ensuite des statistiques globales qui concernent pas qu'une seule
|
||||
note (nombre de soirée, classement, etc).
|
||||
note (nombre de soirée, classement, etc).
|
||||
* ``unique_data`` : le script génére les statitiques uniques à chaque note, et rajoute des données
|
||||
globales si nécessaire, pour chaque note on souhaite avoir un json avec toutes les données qui
|
||||
seront dans le wrapped.
|
||||
|
||||
@@ -18,11 +18,21 @@ note. De cette façon, chaque application peut authentifier ses utilisateur⋅ri
|
||||
et récupérer leurs adhésions, leur nom de note afin d'éventuellement faire des transferts
|
||||
via l'API.
|
||||
|
||||
Deux protocoles d'authentification sont implémentées :
|
||||
Trois protocoles d'authentification sont implémentées :
|
||||
|
||||
* `CAS <cas>`_
|
||||
* `OAuth2 <oauth2>`_
|
||||
* Open ID Connect
|
||||
|
||||
À ce jour, il n'y a pas encore d'exemple d'utilisation d'application qui utilise ce
|
||||
mécanisme, mais on peut imaginer par exemple que la Mediatek ou l'AMAP implémentent
|
||||
ces protocoles pour récupérer leurs adhérent⋅es.
|
||||
À ce jour, ce mécanisme est notamment utilisé par :
|
||||
* Le `serveur photo <https://photos.crans.org>`_
|
||||
* L'`imprimante <https://helloworld.crans.org>`_ du `Cr@ns <https://crans.org>`_
|
||||
* Le serveur `Matrix <https://element.crans.org>`_ du `Cr@ns <https://crans.org>`_
|
||||
* La `base de donnée de la Mediatek <https://med.crans.org>`_
|
||||
* Le site du `K-WEI <https://kwei.crans.org>`_
|
||||
|
||||
Et dans un futur plus ou moins proche :
|
||||
* Le site pour loger les admissibles pendant les oraux (cf. `ici <https://gitlab.crans.org/bde/la25>`_)
|
||||
* L'application mobile de la note
|
||||
* Le site pour les commandes Terre à Terre (cf. `là <https://gitlab.crans.org/tat/blog>`_)
|
||||
* Le futur wiki...
|
||||
|
||||
@@ -47,7 +47,6 @@ On a ensuite besoin de définir nos propres scopes afin d'avoir des permissions
|
||||
'OIDC_ENABLED': True,
|
||||
'OIDC_RSA_PRIVATE_KEY':
|
||||
os.getenv('OIDC_RSA_PRIVATE_KEY', '/var/secrets/oidc.key'),
|
||||
'SCOPES': { 'openid': "OpenID Connect scope" },
|
||||
}
|
||||
|
||||
Cela a pour effet d'avoir des scopes sous la forme ``PERMISSION_CLUB``,
|
||||
@@ -99,7 +98,7 @@ du format renvoyé.
|
||||
|
||||
.. warning::
|
||||
|
||||
Un petit mot sur les scopes : tel qu'implémenté, une scope est une permission unitaire
|
||||
Un petit mot sur les scopes : tel qu'implémenté, un scope est une permission unitaire
|
||||
(telle que décrite dans le modèle ``Permission``) associée à un club. Ainsi, un jeton
|
||||
a accès à une scope si et seulement si læ propriétaire du jeton dispose d'une adhésion
|
||||
courante dans le club lié à la scope qui lui octroie cette permission.
|
||||
@@ -113,6 +112,9 @@ du format renvoyé.
|
||||
Vous pouvez donc contrôler le plus finement possible les permissions octroyées à vos
|
||||
jetons.
|
||||
|
||||
Deux scopes sont un peu particulier, le scope "0_0" qui ne donne aucune permission
|
||||
et le scope "openid" pour l'OIDC.
|
||||
|
||||
.. danger::
|
||||
|
||||
Demander des scopes n'implique pas de les avoir.
|
||||
@@ -134,6 +136,11 @@ du format renvoyé.
|
||||
uniquement dans le cas où l'utilisateur⋅rice connecté⋅e
|
||||
possède la permission problématique.
|
||||
|
||||
Dans le cas extrême ou aucun scope demandé n'est obtenus, vous
|
||||
obtiendriez le scope "0_0" qui ne permet l'accès à rien.
|
||||
Cela permet de générer un token pour toute les requêtes valides.
|
||||
|
||||
|
||||
Avec Django-allauth
|
||||
###################
|
||||
|
||||
@@ -142,6 +149,10 @@ le module pré-configuré disponible ici :
|
||||
`<https://gitlab.crans.org/bde/allauth-note-kfet>`_. Pour l'installer, vous
|
||||
pouvez simplement faire :
|
||||
|
||||
.. warning::
|
||||
À cette heure (11/2025), ce paquet est déprécié et il est plutôt conseillé de créer
|
||||
sa propre application.
|
||||
|
||||
.. code:: bash
|
||||
|
||||
$ pip3 install git+https://gitlab.crans.org/bde/allauth-note-kfet.git
|
||||
@@ -195,6 +206,20 @@ récupérés. Les autres données sont stockées mais inutilisées.
|
||||
Application personnalisée
|
||||
#########################
|
||||
|
||||
.. note::
|
||||
|
||||
Tout les flow (c'est-à-dire les différentes suites de requête possible pour obtenir
|
||||
un token d'accès) de l'OAuth2 sont reproduits dans les
|
||||
`tests <https://gitlab.crans.org/bde/nk20/-/tree/main/apps/permission/tests/test_oauth2_flow.py>`_
|
||||
de l'application permission de la Note. L'OIDC n'étant qu'une extension du protocole
|
||||
OAuth2 vous pouvez facilement reproduire les requêtes en vous inspirant de
|
||||
l'Authorization Code de OAuth2.
|
||||
|
||||
.. danger::
|
||||
|
||||
Pour des raisons de rétrocompatibilité, PKCE (Proof Key for Code Exchange) n'est pas requis,
|
||||
son utilisation est néanmoins très vivement conseillé.
|
||||
|
||||
Ce modèle vous permet de créer vos propres applications à interfacer avec la Note Kfet.
|
||||
|
||||
Commencez par créer une application : `<https://note.crans.org/o/applications/register>`_.
|
||||
@@ -223,6 +248,8 @@ c'est sur cette page qu'il faut rediriger les utilisateur⋅rices. Il faut mettr
|
||||
autorisée par l'application. À des fins de test, peut être `<http://localhost/>`_.
|
||||
* ``state`` : optionnel, peut être utilisé pour permettre au client de détecter des requêtes
|
||||
provenant d'autres sites.
|
||||
* ``code_challenge``: PKCE, le hash d'une chaine d'entre 43 et 128 caractères.
|
||||
* ``code_challenge_method``: PKCE, ``S256`` si le hasher est sha256.
|
||||
|
||||
Sur cette page, les permissions demandées seront listées, et l'utilisateur⋅rice aura le
|
||||
choix d'accepter ou non. Dans les deux cas, l'utilisateur⋅rice sera redirigée vers
|
||||
@@ -283,4 +310,4 @@ de rafraichissement à usage unique. Il suffit pour cela de refaire une requête
|
||||
Le serveur vous fournira alors une nouvelle paire de jetons, comme précédemment.
|
||||
À noter qu'un jeton de rafraîchissement est à usage unique.
|
||||
|
||||
N'hésitez pas à vous renseigner sur OAuth2 pour plus d'informations.
|
||||
N'hésitez pas à vous renseigner sur `OAuth2 <https://www.rfc-editor.org/rfc/rfc6749.html>`_ ou sur le protocole `OIDC <https://openid.net/specs/openid-connect-core-1_0.html>`_ pour plus d'informations.
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -273,9 +273,9 @@ OAUTH2_PROVIDER = {
|
||||
'REFRESH_TOKEN_EXPIRE_SECONDS': timedelta(days=14),
|
||||
'PKCE_REQUIRED': False, # PKCE (fix a breaking change of django-oauth-toolkit 2.0.0)
|
||||
'OIDC_ENABLED': True,
|
||||
'OIDC_RP_INITIATED_LOGOUT_ENABLED': False,
|
||||
'OIDC_RSA_PRIVATE_KEY':
|
||||
os.getenv('OIDC_RSA_PRIVATE_KEY', 'CHANGE_ME_IN_ENV_SETTINGS').replace('\\n', '\n'), # for multilines
|
||||
'SCOPES': { 'openid': "OpenID Connect scope" },
|
||||
}
|
||||
|
||||
# Take control on how widget templates are sourced
|
||||
|
||||
Reference in New Issue
Block a user