Add test for oauth2 flow, add temporary ROPB for NoteApp #137

Small features

See merge request bde/nk20!355

apps/food/views.py

@@ -74,11 +74,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(

apps/member/models.py

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

apps/note/static/note/js/consos.js

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

apps/note/static/note/js/transfer.js

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

apps/permission/backends.py

@@ -26,7 +26,7 @@ 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
@@ -39,7 +39,15 @@ class PermissionBackend(ModelBackend):
 
             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(' '):
+                    if scope == "openid":
+                        continue
                     permission_id, club_id = scope.split('_')
                     if int(club_id) == membership_obj.club_id:
                         query |= Q(pk=permission_id)

apps/permission/scopes.py

@@ -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):
     """
@@ -32,7 +34,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"
+        scopes['openid'] = _("OpenID Connect (username and email)")
         return scopes
 
     def get_available_scopes(self, application=None, request=None, *args, **kwargs):
@@ -82,8 +84,12 @@ class PermissionOAuth2Validator(OAuth2Validator):
 
         valid_scopes = set()
 
+        # 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(get_current_request(), t[0]):
+            for p in PermissionBackend.get_raw_permissions(r, t[0]):
                 scope = f"{p.id}_{p.membership.club.id}"
                 if scope in scopes:
                     valid_scopes.add(scope)

apps/permission/tests/test_oauth2_access.py

@@ -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,39 @@ 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/user/{self.user.pk}/',
+                               **{'Authorization': f'Bearer {token.token}'})
+        self.assertEqual(resp.status_code, 404)

apps/permission/tests/test_oauth2_flow.py

@@ -0,0 +1,115 @@
+# Copyright (C) 2018-2025 by BDE ENS Paris-Saclay
+# SPDX-License-Identifier: GPL-3.0-or-later
+
+import base64
+
+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 ..models import Role, Permission
+
+
+class OAuth2TestCase(TestCase):
+    fixtures = ('initial', )
+
+    def setUp(self):
+        self.user = User.objects.create(
+            username="toto",
+            password="toto1234",
+        )
+
+        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
+        """
+        pass
+
+    def test_oauth2_client_credentials(self):
+        """
+        Ensure OAuth2 Client Credentials work
+        """
+        app = Application.objects.create(
+            name="Test 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 = 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)
+
+        # 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 = 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)
+
+    def test_oidc_flow(self):
+        """
+        Ensure OIDC Flow work
+        """
+        pass

apps/treasury/migrations/0011_sogecredit_valid.py

@@ -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'),
+        ),
+    ]

apps/treasury/models.py

@@ -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")
@@ -338,7 +344,7 @@ class SogeCredit(models.Model):
             credit_transaction.save()
             credit_transaction.refresh_from_db()
             self.credit_transaction = credit_transaction
-        elif not self.valid:
+        elif not self.valid_legacy:
             self.credit_transaction.amount = self.amount
             self.credit_transaction._force_save = True
             self.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)
@@ -365,7 +371,7 @@ class SogeCredit(models.Model):
         The Sogé credit may be created after the user already paid its memberships.
         We query transactions and update the credit, if it is unvalid.
         """
-        if self.valid or not self.pk:
+        if self.valid_legacy or not self.pk:
             return
 
 # Soge do not pay BDE and kfet memberships since 2022
@@ -405,7 +411,7 @@ class SogeCredit(models.Model):
         Invalidating a Société générale delete the transaction of the bank if it was already created.
         Treasurers must know what they do, With Great Power Comes Great Responsibility...
         """
-        if self.valid:
+        if self.valid_legacy:
             self.credit_transaction.valid = False
             self.credit_transaction.save()
         for tr in self.transactions.all():
@@ -414,7 +420,7 @@ class SogeCredit(models.Model):
             tr.save()
 
     def validate(self, force=False):
-        if self.valid and not force:
+        if self.valid_legacy and not force:
             # The credit is already done
             return
 

apps/treasury/tests/test_treasury.py

@@ -359,7 +359,7 @@ class TestSogeCredits(TestCase):
         ))
         self.assertRedirects(response, reverse("treasury:manage_soge_credit", args=(soge_credit.pk,)), 302, 200)
         soge_credit.refresh_from_db()
-        self.assertTrue(soge_credit.valid)
+        self.assertTrue(soge_credit.valid_legacy)
         self.user.note.refresh_from_db()
         self.assertEqual(
             Transaction.objects.filter(Q(source=self.user.note) | Q(destination=self.user.note)).count(), 3)

requirements.txt

@@ -12,7 +12,7 @@ django-filter~=25.1
 django-mailer~=2.3.2
 django-oauth-toolkit~=3.0.1
 django-phonenumber-field~=8.1.0
-django-polymorphic~=3.1.0
+django-polymorphic~=4.1.0
 djangorestframework~=3.16.0
 django-rest-polymorphic~=0.1.10
 django-tables2~=2.7.5