0006-api-add-keepalive-option-to-user-syncronization-API-.patch

Benjamin Dauvergne, 02 novembre 2022 14:22

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

Subject: [PATCH 6/7] api: add keepalive option to user syncronization API
 (#67901)

 src/authentic2/api_views.py            | 33 ++++++++++++-
 src/authentic2/authentication.py       |  3 ++
 src/authentic2/journal_event_types.py  | 33 +++++++++++++
 tests/api/test_user_synchronization.py | 66 +++++++++++++++++++++++++-
 tests/test_manager_journal.py          | 30 +++++++++++-
 5 files changed, 161 insertions(+), 4 deletions(-)

     # You should have received a copy of the GNU Affero General Public License
     # along with this program.  If not, see <http://www.gnu.org/licenses/>.
     import datetime
     import logging
     import smtplib
     from functools import partial
-...
     from django.contrib.auth.hashers import identify_hasher
     from django.contrib.contenttypes.models import ContentType
     from django.core.exceptions import MultipleObjectsReturned
     from django.db import models
     from django.db import models, transaction
     from django.shortcuts import get_object_or_404
     from django.utils.dateparse import parse_datetime
     from django.utils.encoding import force_str
     from django.utils.text import slugify
     from django.utils.timezone import now
     from django.utils.translation import gettext_lazy as _
     from django.views.decorators.cache import cache_control
     from django.views.decorators.vary import vary_on_headers
-...
     from rest_framework.views import APIView
     from rest_framework.viewsets import ModelViewSet, ViewSet
     from authentic2.apps.journal.journal import journal
     from authentic2.apps.journal.models import reference_integer
     from authentic2.compat.drf import action
-...
             known_uuids = serializers.ListField(child=serializers.CharField())
             full_known_users = serializers.BooleanField(required=False)
             timestamp = serializers.DateTimeField(required=False)
             keepalive = serializers.BooleanField(required=False)
         def check_unknown_uuids(self, remote_uuids, users):
             return set(remote_uuids) - {user.uuid for user in users}
-...
                 # reload users to get all fields
                 known_users = User.objects.filter(pk__in=[user.pk for user in users[:1000]])
                 data['known_users'] = [BaseUserSerializer(user).data for user in known_users]
             # update keepalive if requested and:
             # - user is an administrator of users,
             # - user is a publik service using publik signature.
             # It currently excludes APIClient and OIDCClient
             keepalive = serializer.validated_data.get('keepalive', False)
             if keepalive:
                 if not (
                     str(request.user).startswith('Publik Service')
                     or (isinstance(request.user, User) and request.user.has_perm('custom_user.admin_user'))
                 ):
                     raise PermissionDenied('keepalive requires the admin_user permission')
                 self._update_keep_alive(actor=request.user, targeted_users=users)
             hooks.call_hooks('api_modify_response', self, 'synchronization', data)
             return Response(data)
         def _update_keep_alive(self, actor, targeted_users, period_in_days=30):
             # do not write to db uselessly, one keepalive event by month is ok
             start = now()
             threshold = start - datetime.timedelta(days=period_in_days)
             users_to_update = User.objects.filter(pk__in=targeted_users).exclude(
                 models.Q(date_joined__gt=threshold)
                 | models.Q(last_login__gt=threshold)
                 | models.Q(keepalive__gt=threshold)
+            )
             with transaction.atomic(savepoint=False):
                 users_to_update.update(keepalive=start)
                 actor = actor if isinstance(actor, User) else getattr(actor, 'oidc_client', None)
                 for user in users_to_update.only('id'):
                     journal.record('user.notification.activity', actor=actor, target_user=user, api=True)
         @action(
             detail=True,
             methods=['post'],

         def is_authenticated(self):
             return CallableTrue
         def __str__(self):
             return f'OIDC Client "{self.oidc_client}"'
     class Authentic2Authentication(BasicAuthentication):
         def authenticate_credentials(self, userid, password, request=None):

                 'notification sent to "{email}" after {days_of_inactivity} days of inactivity. '
                 'Account will be deleted in {days_to_deletion} days.'
             ).format(days_of_inactivity=days_of_inactivity, days_to_deletion=days_to_deletion, email=email)
     class UserNotificationActivity(EventTypeWithService):
         name = 'user.notification.activity'
         label = _('user activity notification')
         @classmethod
         def record(cls, *, actor, target_user):
             user = actor if isinstance(actor, User) else None
             service = actor if isinstance(actor, Service) else None
             data = {
                 'target_user': str(target_user),
                 'target_user_pk': target_user.pk,
+            }
             references = [target_user]
             super().record(user=user, service=service, data=data, references=references)
         @classmethod
         def get_message(cls, event, context):
             actor_user = event.user
             actor_service, target_user = event.get_typed_references(Service, User)
             if actor_service is None:
                 (target_user,) = event.get_typed_references(User)
             if actor_user is not None:
                 actor = _('user "{0}"').format(actor_user)
             elif actor_service:
                 actor = _('service "{0}"').format(actor_service)
             else:
                 actor = _('unknown actor')
             if context == target_user:
                 return _('user activity notified by {0}').format(actor)
             else:
                 return _('user "{0}" activity notified by {1}').format(target_user, actor)

     import pytest
     from django.contrib.contenttypes.models import ContentType
     from django.db import models
     from django.urls import reverse
     from django.utils.timezone import now
     from authentic2.a2_rbac.models import SEARCH_OP, Role
     from authentic2.a2_rbac.models import ADMIN_OP, SEARCH_OP, Permission, Role
     from authentic2.a2_rbac.utils import get_default_ou
     from authentic2.apps.journal.models import Event, EventType
     from authentic2.custom_user.models import User
     from django_rbac.utils import get_operation
     URL = '/api/users/synchronization/'
-...
         assert len(response.json['unknown_uuids']) == 3
         for user in users[3:]:
             assert user.uuid in response.json['unknown_uuids']
     def test_keepalive_false(app, payload, unknown_uuids):
         app.post_json(URL, params=payload)
         assert User.objects.filter(keepalive__isnull=False).count() == 0
         payload['keepalive'] = False
         app.post_json(URL, params=payload)
         assert User.objects.filter(keepalive__isnull=False).count() == 0
     def test_keepalive_missing_permission(app, user, payload, freezer):
         payload['keepalive'] = True
         app.post_json(URL, params=payload, status=403)
     class TestWithPermission:
         @pytest.fixture(autouse=True)
         def configure_ou(self, users, db):
             ou = get_default_ou()
             User.objects.all().update(ou=ou)
             User.objects.update(last_login=models.F('date_joined'))
             ou.clean_unused_accounts_alert = 60
             ou.clean_unused_accounts_deletion = 63
             ou.save()
         @pytest.fixture
         def user(self, user):
             perm, _ = Permission.objects.get_or_create(
                 ou__isnull=True,
                 operation=get_operation(ADMIN_OP),
                 target_ct=ContentType.objects.get_for_model(ContentType),
                 target_id=ContentType.objects.get_for_model(User).pk,
+            )
             user.roles.all()[0].permissions.add(perm)
             return user
         @pytest.fixture
         def payload(self, payload):
             payload['keepalive'] = True
             return payload
         def test_keepalive_true(self, app, user, users, payload, freezer):
             freezer.move_to(datetime.timedelta(days=50, hours=1))
             app.post_json(URL, params=payload)
             assert User.objects.filter(keepalive__isnull=False).count() == len(users)
         def test_keepalive_one_time_by_clean_unused_period_alert(self, app, user, users, payload, freezer):
             # set last keepalive 29 days ago
             User.objects.exclude(pk=user.pk).update(keepalive=now())
             app.post_json(URL, params=payload)
             freezer.move_to(datetime.timedelta(days=30))
             # keepalive did not change
             assert User.objects.filter(keepalive__lt=now() - datetime.timedelta(days=1)).count() == 10
             # move 2 days in the future
             freezer.move_to(datetime.timedelta(days=1))
             app.post_json(URL, params=payload)
             # keepalive did change
             assert User.objects.filter(keepalive__lt=now() - datetime.timedelta(days=1)).count() == 0

     @pytest.fixture(autouse=True)
     def events(db, freezer):
     def events(db, superuser, freezer):
         session1 = Session(session_key="1234")
         session2 = Session(session_key="abcd")
-...
             session=session2,
             related_object=set_attribute_action,
+        )
         make('user.notification.activity', actor=service, target_user=user)
         make('user.notification.activity', actor=superuser, target_user=user)
         # verify we created at least one event for each type
         assert set(Event.objects.values_list("type__name", flat=True)) == set(_registry)
-...
         set_attribute_action = SetAttributeAction.objects.get()
         # remove event about admin login
         Event.objects.filter(user=superuser).delete()
         Event.objects.order_by('-id').filter(type__name='user.login', user=superuser)[0].delete()
         response = response.click("Global journal")
-...
                 'type': 'authenticator.related_object.deletion',
                 'user': 'agent',
             },
+            {
                 'message': 'user "Johnny doe" activity notified by service "service"',
                 'timestamp': 'Jan. 3, 2020, 11 a.m.',
                 'type': 'user.notification.activity',
                 'user': '-',
             },
+            {
                 'message': 'user "Johnny doe" activity notified by user "super user"',
                 'timestamp': 'Jan. 3, 2020, noon',
                 'type': 'user.notification.activity',
                 'user': 'super user',
             },
+        ]
         agent_page = response.click('agent', index=1)
-...
                 'type': 'user.deletion.inactivity',
                 'user': 'Johnny doe',
             },
+            {
                 'message': 'user activity notified by service "service"',
                 'timestamp': 'Jan. 3, 2020, 11 a.m.',
                 'type': 'user.notification.activity',
                 'user': '-',
             },
+            {
                 'message': 'user activity notified by user "super user"',
                 'timestamp': 'Jan. 3, 2020, noon',
                 'type': 'user.notification.activity',
                 'user': 'super user',
             },
+        ]
+    -

Projet

Général

Profil

Produits Entr'ouvert » Authentic 2

0006-api-add-keepalive-option-to-user-syncronization-API-.patch