0001-idp_oidc-restrict-client-s-returned-qs-to-authorized.patch

Paul Marillonnet, 03 juin 2022 11:43

Voir les différences: en ligne côte à côte

Subject: [PATCH] idp_oidc: restrict client's returned qs to authorized users
 (#65942)

 src/authentic2_idp_oidc/apps.py | 28 ++++++++++++---
 tests/api/test_all.py           | 39 +++++++++++++++++++-
 tests/conftest.py               |  4 +++
 tests/idp_oidc/test_api.py      | 63 ++++++++++++++++++++++++++++++++-
 4 files changed, 128 insertions(+), 6 deletions(-)

         def a2_hook_api_modify_queryset(self, view, qs):
             from django.contrib.auth import get_user_model
             from django.db.models import Q
             from django.utils.timezone import now
             from .models import OIDCAuthorization, OIDCClient
             client = self.get_oidc_client(view)
             # fast path
             User = get_user_model()
             # reduce qs to users having previously granted an authorization to the client
             if (
                 not issubclass(qs.model, get_user_model())
                 or client is None
                 or not client.authorized_roles.exists()
                 client
                 and client.authorization_mode != OIDCClient.AUTHORIZATION_MODE_NONE
                 and view.request.method == 'GET'
                 and issubclass(qs.model, User)
             ):
                 authorizations = OIDCAuthorization.objects.none()
                 if client.authorization_mode == OIDCClient.AUTHORIZATION_MODE_BY_SERVICE:
                     authorizations = client.authorizations
                 elif client.ou and hasattr(client.ou, 'oidc_authorizations'):
                     authorizations = client.ou.oidc_authorizations
                 qs = qs.filter(
                     id__in=authorizations.filter(Q(expired__isnull=True) | Q(expired__gt=now())).values_list(
                         'user__id', flat=True
+                    )
+                )
             # fast path
             if not issubclass(qs.model, User) or client is None or not client.authorized_roles.exists():
                 return qs
             qs = qs.filter(roles__in=client.authorized_roles.children())

     from django.urls import reverse
     from django.utils.encoding import force_text
     from django.utils.text import slugify
     from django.utils.timezone import now
     from requests.models import Response
     from authentic2.a2_rbac.models import OrganizationalUnit as OU
-...
     from authentic2.models import Attribute, AttributeValue, AuthorizedRole, PasswordReset, Service
     from authentic2.utils.misc import good_next_url
     from django_rbac.models import SEARCH_OP
     from django_rbac.utils import get_ou_model
     from ..utils import assert_event, basic_authorization_header, get_link_from_mail, login
-...
     def test_api_users_create(settings, app, api_user):
         from authentic2_idp_oidc.models import OIDCAuthorization, OIDCClient
         at = Attribute.objects.create(kind='title', name='title', label='title')
         app.authorization = ('Basic', (api_user.username, api_user.username))
         # test missing first_name
-...
             assert AttributeValue.objects.with_owner(new_user).filter(verified=True).count() == 0
             assert AttributeValue.objects.with_owner(new_user).filter(attribute=at).exists()
             assert AttributeValue.objects.with_owner(new_user).get(attribute=at).content == payload['title']
             if (
                 hasattr(api_user, 'oidc_client')
                 and api_user.oidc_client.authorization_mode != OIDCClient.AUTHORIZATION_MODE_NONE
             ):
                 if api_user.oidc_client.authorization_mode == OIDCClient.AUTHORIZATION_MODE_BY_SERVICE:
                     client_id = api_user.oidc_client.id
                     client_ct = ContentType.objects.get_for_model(OIDCClient)
                 else:
                     client_id = api_user.oidc_client.ou.id
                     client_ct = ContentType.objects.get_for_model(get_ou_model())
                 OIDCAuthorization.objects.create(
                     user=new_user,
                     client_id=client_id,
                     client_ct=client_ct,
                     expired=now() + datetime.timedelta(hours=1),
+                )
             resp2 = app.get('/api/users/%s/' % resp.json['uuid'])
             assert resp.json == resp2.json
             payload.update({'uuid': '1234567890', 'email': 'foo@example.com', 'username': 'foobar'})
-...
             assert 'title' in resp.json
             at.disabled = True
             at.save()
             if (
                 hasattr(api_user, 'oidc_client')
                 and api_user.oidc_client.authorization_mode != OIDCClient.AUTHORIZATION_MODE_NONE
             ):
                 authz = OIDCAuthorization.objects.get(user=new_user)
                 authz.user = User.objects.get(uuid='1234567890')
                 authz.save()
             resp = app.get('/api/users/1234567890/')
             assert 'title' not in resp.json
-...
     def test_api_drf_authentication_class(app, admin, user_ou1, oidc_client):
         from authentic2_idp_oidc.models import OIDCAuthorization, OIDCClient
         url = '/api/users/%s/' % user_ou1.uuid
         # test invalid client
         app.authorization = ('Basic', ('foo', 'bar'))
-...
         resp = app.get(url, status=401)
         assert resp.json['result'] == 0
         assert resp.json['errors'] == "User inactive or deleted."
         # test oidc client
         # test oidc client unauthorized for user_ou1
         app.authorization = ('Basic', (oidc_client.username, oidc_client.username))
         app.get(url, status=404)
         OIDCAuthorization.objects.create(
             client_id=oidc_client.id,
             client_ct=ContentType.objects.get_for_model(OIDCClient),
             user=user_ou1,
             expired=now() + datetime.timedelta(hours=1),
+        )
         # test oidc client
         app.get(url, status=200)
         # test oidc client without has API access
         oidc_client.oidc_client.has_api_access = False

             def username(self):
                 return self.oidc_client.client_id
             @property
             def id(self):
                 return self.oidc_client.id
             @property
             def is_superuser(self):
                 return False

     # along with this program.  If not, see <http://www.gnu.org/licenses/>.
     import random
     from datetime import timedelta
     from django.contrib.contenttypes.models import ContentType
     from django.utils.timezone import now
     from authentic2.a2_rbac.utils import get_default_ou
     from authentic2.custom_user.models import User
     from authentic2_idp_oidc.models import OIDCClaim, OIDCClient
     from authentic2_idp_oidc.models import OIDCAuthorization, OIDCClaim, OIDCClient
     from authentic2_idp_oidc.utils import make_sub
     from django_rbac.utils import get_ou_model
     def test_api_synchronization(app, oidc_client):
-...
+        )
         users = [User.objects.create(username=f'user-{i}', last_name=f'Name-{i}') for i in range(10)]
         expired = now() + timedelta(hours=1)
         for user in users:
             OIDCAuthorization.objects.create(
                 client_id=oidc_client.id,
                 client_ct=ContentType.objects.get_for_model(OIDCClient),
                 user=user,
                 expired=expired,
+            )
         pre_modification = now().strftime('%Y-%m-%dT%H:%M:%S')
         for count, user in enumerate(users):
             user.first_name = f'User {count}'
-...
             assert user_dict['email']
             assert user_dict['email'].startswith(user_dict['last_name'])
             assert user_dict['email'].endswith('@templated.nowhere.null')
     def test_api_users_list_queryset_reduction(app, oidc_client):
         oidc_client.has_api_access = True
         oidc_client.identifier_policy = OIDCClient.POLICY_PAIRWISE_REVERSIBLE
         oidc_client.save()
         pre_modification = now().strftime('%Y-%m-%dT%H:%M:%S')
         users = [User.objects.create(username=f'user-{i}', last_name=f'Name-{i}') for i in range(20)]
         expired = now() + timedelta(hours=1)
         for user in random.sample(users, k=5):
             OIDCAuthorization.objects.create(
                 client_id=oidc_client.id,
                 client_ct=ContentType.objects.get_for_model(OIDCClient),
                 user=user,
                 expired=expired,
+            )
         app.authorization = ('Basic', (oidc_client.client_id, oidc_client.client_secret))
         url = f'/api/users/?modified__gt={pre_modification}&claim_resolution'
         response = app.get(url, status=200)
         assert len(response.json['results']) == 5
         oidc_client.authorization_mode = OIDCClient.AUTHORIZATION_MODE_NONE
         oidc_client.save()
         response = app.get(url, status=200)
         assert len(response.json['results']) == User.objects.count()
         oidc_client.authorization_mode = OIDCClient.AUTHORIZATION_MODE_BY_OU
         oidc_client.ou = get_default_ou()
         oidc_client.save()
         response = app.get(url, status=200)
         assert not response.json['results']
         for user in random.sample(users, k=8):
             OIDCAuthorization.objects.create(
                 client_id=get_default_ou().id,
                 client_ct=ContentType.objects.get_for_model(get_ou_model()),
                 user=user,
                 expired=expired,
+            )
         response = app.get(url, status=200)
         assert len(response.json['results']) == 8
+    -

Projet

Général

Profil

Produits Entr'ouvert » Authentic 2

0001-idp_oidc-restrict-client-s-returned-qs-to-authorized.patch