0003-provisionning-add-synchronous-provisionning-to-DRF-a.patch

Benjamin Dauvergne, 30 juin 2022 23:24

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

Subject: [PATCH 3/3] provisionning: add synchronous provisionning to DRF
 authentication (#59135)

 hobo/provisionning/utils.py               | 52 ++++++++++++
 hobo/rest_authentication.py               |  5 +-
 tests_multitenant/conftest.py             |  8 +-
 tests_multitenant/test_provisionning.py   | 98 ++++++++++++++++++++++-
 tests_multitenant/test_uwsgidecorators.py |  5 +-
 5 files changed, 160 insertions(+), 8 deletions(-)

     import hashlib
     import logging
     from urllib.parse import quote, urljoin
     import requests
     from django.conf import settings
     from django.contrib.auth import get_user_model
     from django.contrib.auth.models import Group
     from django.db import IntegrityError
-...
     from django.db.transaction import atomic
     from mellon.models import Issuer
     from hobo import signature
     from hobo.agent.common.models import Role
     from hobo.multitenant.utils import provision_user_groups
-...
         return s
     def get_idp_service():
         idp_services = list((settings.KNOWN_SERVICES or {}).get('authentic', {}).values())
         return (idp_services or [None])[0]
     def get_issuer(entity_id):
         issuer, created = Issuer.objects.get_or_create(entity_id=entity_id)
         return issuer
-...
                 except TryAgain:
                     continue
                 break
     def get_user_from_name_id(name_id, entity_id=None, raise_on_missing=False):
         User = get_user_model()
         try:
             user = User.objects.get(
                 saml_identifiers__name_id=name_id, saml_identifiers__issuer__entity_id=entity_id
+            )
         except User.DoesNotExist:
             try:
                 user = User.objects.get(Q(username=name_id[:30]) | Q(username=name_id[:150]))
             except User.DoesNotExist:
                 user = None
         if not user:
             if raise_on_missing:
                 raise User.DoesNotExist
         return user
     def get_or_create_user_from_name_id(name_id, raise_on_missing=False):
         User = get_user_model()
         user = get_user_from_name_id(name_id=name_id)
         if user:
             return user
         idp_service = get_idp_service()
         if not idp_service:
             if raise_on_missing:
                 raise User.DoesNotExist
             return None
         entity_id = idp_service['saml-idp-metadata-url']
         issuer = get_issuer(entity_id)
         users_api_url = urljoin(idp_service['url'], 'api/users/%s/' % quote(name_id, safe=''))
         try:
             response = requests.get(signature.sign_url(users_api_url, idp_service['secret']), timeout=5)
             response.raise_for_status()
         except requests.RequestException:
             if raise_on_missing:
                 raise User.DoesNotExist('Failed to reach IdP')
             return None
         return provision_user(issuer.entity_id, o=response.json())

     from rest_framework import authentication, exceptions, status
     from hobo import signature
     from hobo.provisionning.utils import get_or_create_user_from_name_id
     try:
         from mellon.models import UserSAMLIdentifier
-...
                 elif UserSAMLIdentifier:
                     try:
                         return UserSAMLIdentifier.objects.get(name_id=name_id).user
                     except UserSAMLIdentifier.DoesNotExist:
                         return get_or_create_user_from_name_id(name_id, raise_on_missing=True)
                     except User.DoesNotExist:
                         raise PublikAuthenticationFailed('user-not-found')
                 else:
                     raise PublikAuthenticationFailed('no-usable-model')

     import pytest
     from tenant_schemas.utils import tenant_context
     @pytest.fixture
-...
                                 'service-id': 'authentic',
                                 'base_url': 'http://other.example.net',
                                 'legacy_urls': [{'base_url': 'http://olda2.example.net'}],
                                 'saml-idp-metadata-url': 'https://other.example.net/idp/saml2/metadata',
                             },
+                            {
                                 'slug': 'another',
-...
                                 'service-id': 'combo',
                                 'template_name': '...portal-user...',
                                 'base_url': 'http://portal-user.example.net',
                                 'secret_key': 'abcdefg',
                             },
                         ],
                     },
-...
     @pytest.fixture
     def tenant(make_tenant):
         return make_tenant('tenant.example.net')
         tenant = make_tenant('tenant.example.net')
         with tenant_context(tenant):
             yield tenant
     @pytest.fixture

     from hobo.provisionning.utils import NotificationProcessing
     import json
     import pytest
     import requests
     from httmock import HTTMock, urlmatch
     from mellon.models import Issuer
     from rest_framework import permissions
     from rest_framework.response import Response
     from rest_framework.views import APIView
     from hobo import rest_authentication, signature
     from hobo.provisionning.utils import (
         NotificationProcessing,
         get_or_create_user_from_name_id,
         get_user_from_name_id,
+    )
     def test_truncate_role_name():
-...
             assert len(truncated) == max_length
             assert truncated not in seen
             seen.add(truncated)
     @urlmatch()
     def request_exception(url, request):
         raise requests.ConnectionError
     NAME_ID = '1234' * 8
     @urlmatch(path='/api/users/')
     def user_payload(url, request):
         return {
             'status_code': 200,
             'content': json.dumps(
+                {
                     'uuid': NAME_ID,
                     'first_name': 'John',
                     'last_name': 'Doe',
                     'email': 'john.doe@example.net',
                     'is_superuser': False,
                     'is_active': True,
+                }
             ),
+        }
     def test_get_or_create_user_from_name_id(tenant, django_user_model):
         Issuer.objects.create(entity_id='https://idp.examle.net/idp/saml2/metadata')
         assert get_user_from_name_id(NAME_ID) is None
         with pytest.raises(django_user_model.DoesNotExist):
             get_user_from_name_id(NAME_ID, raise_on_missing=True)
         with HTTMock(request_exception):
             assert get_or_create_user_from_name_id(NAME_ID) is None
             with pytest.raises(django_user_model.DoesNotExist):
                 get_or_create_user_from_name_id(NAME_ID, raise_on_missing=True)
         with HTTMock(user_payload):
             user = get_or_create_user_from_name_id(NAME_ID)
             assert user is not None
             assert user.first_name == 'John'
             assert user.username == NAME_ID
             assert user.saml_identifiers.get().name_id == NAME_ID
             assert django_user_model.objects.count() == 1
     class DummyAPIView(APIView):
         authentication_classes = (rest_authentication.PublikAuthentication,)
         permission_classes = (permissions.IsAuthenticated,)
         def get(self, request, format=None):
             return Response({'err': 0})
     def test_rest_authentication_provisionning_ok(tenant, settings, rf, django_user_model):
         key = settings.KNOWN_SERVICES['combo']['another2']['secret']
         request = rf.get(signature.sign_url('/api/misc/?param1=2&NameID=%s&orig=portal-user.example.net', key))
         view = DummyAPIView.as_view()
         with HTTMock(user_payload):
             response = view(request)
         assert response.status_code == 200
         assert response.data == {'err': 0}
         user = django_user_model.objects.get()
         assert user.first_name == 'John'
         assert user.username == NAME_ID
         assert user.saml_identifiers.get().name_id == NAME_ID
     def test_rest_authentication_provisionning_nok(tenant, settings, rf):
         key = settings.KNOWN_SERVICES['combo']['another2']['secret']
         request = rf.get(signature.sign_url('/api/misc/?param1=2&NameID=%s&orig=portal-user.example.net', key))
         view = DummyAPIView.as_view()
         response = view(request)
         assert response.status_code == 401
         assert response.data == {'err': 1, 'err_desc': 'user-not-found'}

     def test_mocked_uwsgi_tenant(uwsgi, tenant):
         from tenant_schemas.utils import tenant_context
         @hobo.multitenant.uwsgidecorators.spool
         def function(a, b):
             pass
         with tenant_context(tenant):
             function.spool(1, 2)
         function.spool(1, 2)
         assert set(uwsgi.spool.call_args[1].keys()) == {'body', 'tenant', 'name'}
         assert pickle.loads(uwsgi.spool.call_args[1]['body']) == {'args': (1, 2), 'kwargs': {}}
+    -

Projet

Général

Profil

Produits Entr'ouvert » Hobo

0003-provisionning-add-synchronous-provisionning-to-DRF-a.patch