Produits Entr'ouvert » Authentic 2

import datetime

from django.db import models, transaction

from django.utils.timezone import now

from authentic2.apps.journal.journal import journal

        keepalive = serializers.BooleanField(required=False)

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

    def _update_keep_alive(self, actor, targeted_users):

        # do not write to db uselessly, one keepalive event by month is ok

        start = now()

        threshold = start - datetime.timedelta(days=30)

        users_to_check = User.objects.filter(

            (models.Q(last_login__isnull=True) | models.Q(last_login__lt=threshold))

            & (models.Q(keepalive__isnull=True) | models.Q(keepalive__lt=threshold)),

            pk__in=targeted_users,

        ).select_related('ou')

        with transaction.atomic(savepoint=False):

            users_pks_to_update = {}

            for user in users_to_check:

                if user.ou and user.ou.clean_unused_accounts_alert:

                    activity_date = max(

                        user.date_joined,

                        user.last_login or user.date_joined,

                        user.keepalive or user.date_joined,

                    )

                    time_since_last_activity = now() - activity_date

                    if time_since_last_activity.days > (user.ou.clean_unused_accounts_alert / 2):

                        users_pks_to_update[user.pk] = user

            User.objects.filter(pk__in=users_pks_to_update.keys()).update(keepalive=start)

            for user in users_pks_to_update.values():

                journal.record('user.notification.activity', actor=actor, target_user=user, api=True)

    def __str__(self):

        return f'OIDC Client "{self.oidc_client}"'

class UserNotificationActivity(EventTypeDefinition):

    name = 'user.notification.activity'

    label = _('user activity notification')

    @classmethod

    def record(cls, *, actor, target_user):

        user = actor if getattr(actor, 'pk', None) is not None else None

        service = user.oidc_client if getattr(user, 'oidc_client', None) is not None else None

        data = {

            'actor': str(actor),

            'target_user': str(target_user),

            'target_user_pk': target_user.pk,

        }

        references = [target_user]

        if service is not None:

            references.append(service)

        if user is not None:

            references.append(user)

        super().record(user=user, service=service, data=data, references=references)

    @classmethod

    def get_message(cls, event, context):

        user, actor_service = event.get_typed_references(User, Service)

        user, actor_user = event.get_typed_references(User, User)

        if actor_service:

            actor = _('service "{0}"').format(actor_service)

        elif actor_user:

            actor = _('user "{0}"').format(actor_user)

        else:

            actor = _('unknown actor')

        if context == user:

            return _('user activity notified by {0}').format(actor)

        else:

            return _('user "{0}" activity notified by {1}').format(user, actor)

from django.db import models

from django.utils.timezone import now

from authentic2.a2_rbac.models import Permission, Role

from authentic2.a2_rbac.utils import get_default_ou

from django_rbac.models import ADMIN_OP, SEARCH_OP

from django_rbac.utils import get_operation

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