ynerant/nk20
1
0
Fork 0
mirror of https://gitlab.crans.org/bde/nk20 synced 2025-11-20 21:38:28 +01:00
Code Issues Releases Wiki Activity

Merge branch 'main' into note_sheet

Browse Source
This commit is contained in:
Ehouarn
2025-11-20 17:50:11 +01:00
parent 0fad2a876d 0240ea5388
commit bac9ed2353
30 changed files with 1258 additions and 254 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
apps/activity/views.py
View File

@@ -152,9 +152,11 @@ class ActivityDetailView(ProtectQuerysetMixin, LoginRequiredMixin, MultiTableMix
def get_tables_data(self):
return [
Guest.objects.filter(activity=self.object)
.filter(PermissionBackend.filter_queryset(self.request, Guest, "view")),
.filter(PermissionBackend.filter_queryset(self.request, Guest, "view"))
.distinct(),
self.object.opener.filter(activity=self.object)
.filter(PermissionBackend.filter_queryset(self.request, Opener, "view")),
.filter(PermissionBackend.filter_queryset(self.request, Opener, "view"))
.distinct(),
]
def render_to_response(self, context, **response_kwargs):
@@ -309,7 +311,7 @@ class ActivityInviteView(ProtectQuerysetMixin, ProtectedCreateView):
@transaction.atomic
def form_valid(self, form):
form.instance.activity = Activity.objects\
.filter(PermissionBackend.filter_queryset(self.request, Activity, "view")).get(pk=self.kwargs["pk"])
.filter(PermissionBackend.filter_queryset(self.request, Activity, "view")).distinct().get(pk=self.kwargs["pk"])
return super().form_valid(form)
def get_success_url(self, **kwargs):

6
apps/food/views.py
View File

@@ -81,11 +81,15 @@ class FoodListView(ProtectQuerysetMixin, LoginRequiredMixin, MultiTableMixin, Li
search_table = qs.filter(PermissionBackend.filter_queryset(self.request, Food, 'view'))
# table open
open_table = self.get_queryset().order_by('expiry_date').filter(
open_table = self.get_queryset().filter(
Q(polymorphic_ctype__model='transformedfood')
| Q(polymorphic_ctype__model='basicfood', basicfood__date_type='DLC')).filter(
expiry_date__lt=timezone.now(), end_of_life='').filter(
PermissionBackend.filter_queryset(self.request, Food, 'view'))
open_table = open_table.union(self.get_queryset().filter(
Q(end_of_life='', order__iexact='open')
).filter(
PermissionBackend.filter_queryset(self.request, Food, 'view'))).order_by('expiry_date')
# table served
served_table = self.get_queryset().order_by('-pk').filter(
end_of_life='', is_ready=True).exclude(

2
apps/member/models.py
View File

@@ -417,7 +417,7 @@ class Membership(models.Model):
A membership is valid if today is between the start and the end date.
"""
if self.date_end is not None:
return self.date_start.toordinal() <= datetime.datetime.now().toordinal() < self.date_end.toordinal()
return self.date_start.toordinal() <= datetime.datetime.now().toordinal() <= self.date_end.toordinal()
else:
return self.date_start.toordinal() <= datetime.datetime.now().toordinal()

9
apps/member/views.py
View File

@@ -50,6 +50,15 @@ class CustomLoginView(LoginView):
self.request.session['permission_mask'] = form.cleaned_data['permission_mask'].rank
return super().form_valid(form)
def get_context_data(self, **kwargs):
context = super().get_context_data(**kwargs)
user_agent = self.request.META.get('HTTP_USER_AGENT', '').lower()
context['display_appstore_badge'] = 'iphone' in user_agent or 'android' not in user_agent
context['display_playstore_badge'] = 'android' in user_agent or 'iphone' not in user_agent
return context
class UserUpdateView(ProtectQuerysetMixin, LoginRequiredMixin, UpdateView):
"""

2
apps/note/static/note/js/consos.js
View File

@@ -228,7 +228,7 @@ function consume (source, source_alias, dest, quantity, amount, reason, type, ca
addMsg(interpolate(gettext('Warning, the transaction from the note %s succeed, ' +
'but the emitter note %s is negative.'), [source_alias, source_alias]), 'warning', 30000)
}
if (source.membership && source.membership.date_end < new Date().toISOString()) {
if (source.membership && source.membership.date_end <= new Date().toISOString()) {
addMsg(interpolate(gettext('Warning, the emitter note %s is no more a BDE member.'), [source_alias]),
'danger', 30000)
}

6
apps/note/static/note/js/transfer.js
View File

@@ -310,10 +310,10 @@ $('#btn_transfer').click(function () {
destination: dest.note.id,
destination_alias: dest.name
}).done(function () {
if (source.note.membership && source.note.membership.date_end < new Date().toISOString()) {
if (source.note.membership && source.note.membership.date_end <= new Date().toISOString()) {
addMsg(interpolate(gettext('Warning, the emitter note %s is no more a BDE member.'), [source.name]), 'danger', 30000)
}
if (dest.note.membership && dest.note.membership.date_end < new Date().toISOString()) {
if (dest.note.membership && dest.note.membership.date_end <= new Date().toISOString()) {
addMsg(interpolate(gettext('Warning, the destination note %s is no more a BDE member.'), [dest.name]), 'danger', 30000)
}
@@ -414,7 +414,7 @@ $('#btn_transfer').click(function () {
bank: $('#bank').val()
}).done(function () {
addMsg(gettext('Credit/debit succeed!'), 'success', 10000)
if (user_note.membership && user_note.membership.date_end < new Date().toISOString()) { addMsg(gettext('Warning, the emitter note %s is no more a BDE member.'), 'danger', 10000) }
if (user_note.membership && user_note.membership.date_end <= new Date().toISOString()) { addMsg(gettext('Warning, the emitter note %s is no more a BDE member.'), 'danger', 10000) }
reset()
}).fail(function (err) {
const errObj = JSON.parse(err.responseText)

23
apps/permission/backends.py
View File

@@ -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

32
apps/permission/fixtures/initial.json
View File

@@ -927,7 +927,7 @@
"note",
"transactiontemplate"
],
"query": "{\"destination\": [\"club\", \"note\"]}",
"query": "[\"AND\", {\"destination\": [\"club\", \"note\"]}, {\"category__name\": \"Clubs\"}]",
"type": "view",
"mask": 2,
"field": "",
@@ -943,7 +943,7 @@
"note",
"transactiontemplate"
],
"query": "{\"destination\": [\"club\", \"note\"]}",
"query": "[\"AND\", {\"destination\": [\"club\", \"note\"]}, {\"category__name\": \"Clubs\"}]",
"type": "add",
"mask": 3,
"field": "",
@@ -959,7 +959,7 @@
"note",
"transactiontemplate"
],
"query": "{\"destination\": [\"club\", \"note\"]}",
"query": "[\"AND\", {\"destination\": [\"club\", \"note\"]}, {\"category__name\": \"Clubs\"}]",
"type": "change",
"mask": 3,
"field": "",
@@ -3484,7 +3484,23 @@
"mask": 1,
"permanent": false,
"description": "Voir la bouffe servie"
}
}
},
{
"model": "permission.permission",
"pk": 223,
"fields": {
"model": [
"note",
"templatecategory"
],
"query": "{\"name\": \"Clubs\"}",
"type": "view",
"mask": 2,
"field": "",
"permanent": false,
"description": "Voir la catégorie de bouton Clubs"
}
},
{
"model": "permission.permission",
@@ -5091,7 +5107,6 @@
19,
20,
21,
27,
59,
60,
61,
@@ -5102,6 +5117,7 @@
182,
184,
185,
223,
239,
240,
241
@@ -5466,6 +5482,12 @@
176,
177,
197,
211,
212,
213,
214,
215,
216,
311,
319
]

102
apps/permission/scopes.py
View File

@@ -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

38
apps/permission/tests/test_oauth2.py → apps/permission/tests/test_oauth2_access.py
View File

@@ -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
View 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)

18
apps/treasury/migrations/0011_sogecredit_valid.py Normal file
View File

@@ -0,0 +1,18 @@
# Generated by Django 5.2.6 on 2025-09-28 20:12
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('treasury', '0010_alter_invoice_bde'),
]
operations = [
migrations.AddField(
model_name='sogecredit',
name='valid',
field=models.BooleanField(blank=True, default=False, verbose_name='Valid'),
),
]

15
apps/treasury/models.py
View File

@@ -308,6 +308,12 @@ class SogeCredit(models.Model):
null=True,
)
valid = models.BooleanField(
default=False,
verbose_name=_("Valid"),
blank=True,
)
class Meta:
verbose_name = _("Credit from the Société générale")
verbose_name_plural = _("Credits from the Société générale")
@@ -332,7 +338,7 @@ class SogeCredit(models.Model):
last_name=self.user.last_name,
first_name=self.user.first_name,
bank="Société générale",
valid=False,
valid=True,
)
credit_transaction._force_save = True
credit_transaction.save()
@@ -346,12 +352,12 @@ class SogeCredit(models.Model):
return super().save(*args, **kwargs)
@property
def valid(self):
def valid_legacy(self):
return self.credit_transaction and self.credit_transaction.valid
@property
def amount(self):
if self.valid:
if self.valid_legacy:
return self.credit_transaction.total
amount = 0
transactions_wei = self.transactions.filter(membership__club__weiclub__isnull=False)
@@ -397,7 +403,7 @@ class SogeCredit(models.Model):
self.transactions.add(m.transaction)
for tr in self.transactions.all():
tr.valid = False
tr.valid = True
tr.save()
def invalidate(self):
@@ -422,6 +428,7 @@ class SogeCredit(models.Model):
self.invalidate()
# Refresh credit amount
self.save()
self.valid = True
self.credit_transaction.valid = True
self.credit_transaction._force_save = True
self.credit_transaction.save()

BIN
apps/treasury/static/img/Diolistos_bg.jpg
View File

Binary file not shown.
Side by Side Swipe Overlay

Before

Width:  |  Height:  |  Size: 284 KiB

After

Width:  |  Height:  |  Size: 104 KiB

1
apps/treasury/tables.py
View File

@@ -56,6 +56,7 @@ class InvoiceTable(tables.Table):
model = Invoice
template_name = 'django_tables2/bootstrap4.html'
fields = ('id', 'name', 'object', 'acquitted', 'invoice',)
order_by = ('-id',)
class RemittanceTable(tables.Table):

2
apps/treasury/views.py
View File

@@ -417,7 +417,7 @@ class SogeCreditListView(LoginRequiredMixin, ProtectQuerysetMixin, SingleTableVi
)
if "valid" not in self.request.GET or not self.request.GET["valid"]:
qs = qs.filter(credit_transaction__valid=False)
qs = qs.filter(valid=False)
return qs

2
apps/wei/tests/test_wei_registration.py
View File

@@ -680,7 +680,7 @@ class TestWEIRegistration(TestCase):
self.assertTrue(soge_credit.exists())
soge_credit = soge_credit.get()
self.assertTrue(membership.transaction in soge_credit.transactions.all())
self.assertFalse(membership.transaction.valid)
self.assertTrue(membership.transaction.valid)
# Check that if the WEI is started, we can't update a wei
self.wei.date_start = date(2000, 1, 1)