0002-commands-per-OU-unused-user-accounts-cleaning-policy.patch

Valentin Deniaud, 02 avril 2020 11:03

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

Subject: [PATCH 2/3] commands: per-OU unused user accounts cleaning policy
 (#26909)

 .../migrations/0022_auto_20200402_1101.py     |  26 ++++
 src/authentic2/a2_rbac/models.py              |  22 +++
 .../0018_user_last_account_deletion_alert.py  |  20 +++
 src/authentic2/custom_user/models.py          |   4 +
 .../commands/clean-unused-accounts.py         | 130 +++++++-----------
 src/authentic2/manager/forms.py               |   2 +-
 src/authentic2/utils/__init__.py              |   2 +
 tests/test_a2_rbac.py                         |  10 ++
 tests/test_all.py                             |   1 +
 tests/test_api.py                             |   6 +-
 tests/test_commands.py                        |  80 ++++++++++-
 11 files changed, 218 insertions(+), 85 deletions(-)
 create mode 100644 src/authentic2/a2_rbac/migrations/0022_auto_20200402_1101.py
 create mode 100644 src/authentic2/custom_user/migrations/0018_user_last_account_deletion_alert.py

     # -*- coding: utf-8 -*-
     # Generated by Django 1.11.18 on 2020-04-02 09:01
     from __future__ import unicode_literals
     import django.core.validators
     from django.db import migrations, models
     class Migration(migrations.Migration):
         dependencies = [
             ('a2_rbac', '0021_auto_20200317_1514'),
+        ]
         operations = [
             migrations.AddField(
                 model_name='organizationalunit',
                 name='clean_unused_accounts_alert',
                 field=models.PositiveIntegerField(blank=True, null=True, validators=[django.core.validators.MinValueValidator(30, 'Ensure that this value is greater than 30 days, or leave blank for deactivating.')], verbose_name='Days after which the user receives an account deletion alert'),
             ),
             migrations.AddField(
                 model_name='organizationalunit',
                 name='clean_unused_accounts_deletion',
                 field=models.PositiveIntegerField(blank=True, null=True, validators=[django.core.validators.MinValueValidator(30, 'Ensure that this value is greater than 30 days, or leave blank for deactivating.')], verbose_name='Delay in days before cleaning unused accounts'),
             ),
+        ]

     from collections import namedtuple
     from django.core.exceptions import ValidationError
     from django.core.validators import MinValueValidator
     from django.utils import six
     from django.utils.translation import ugettext_lazy as _
     from django.utils.text import slugify
-...
             choices=USER_ADD_PASSWD_POLICY_CHOICES,
             default=0)
         clean_unused_accounts_alert = models.PositiveIntegerField(
             verbose_name=_('Days after which the user receives an account deletion alert'),
             validators=[MinValueValidator(
 , _('Ensure that this value is greater than 30 days, or leave blank for deactivating.')
             )],
             null=True,
             blank=True)
         clean_unused_accounts_deletion = models.PositiveIntegerField(
             verbose_name=_('Delay in days before cleaning unused accounts'),
             validators=[MinValueValidator(
 , _('Ensure that this value is greater than 30 days, or leave blank for deactivating.')
             )],
             null=True,
             blank=True)
         objects = managers.OrganizationalUnitManager()
         class Meta:
-...
                 raise ValidationError(_('You cannot unset this organizational '
                                         'unit as the default, but you can set '
                                         'another one as the default.'))
             if bool(self.clean_unused_accounts_alert) ^ bool(self.clean_unused_accounts_deletion):
                 raise ValidationError(_('Deletion and alert delays must be set together.'))
             if self.clean_unused_accounts_alert and \
                     self.clean_unused_accounts_alert >= self.clean_unused_accounts_deletion:
                 raise ValidationError(_('Deletion alert delay must be less than actual deletion delay.'))
             super(OrganizationalUnit, self).clean()
         def get_admin_role(self):

     # -*- coding: utf-8 -*-
     # Generated by Django 1.11.18 on 2020-03-17 14:16
     from __future__ import unicode_literals
     from django.db import migrations, models
     class Migration(migrations.Migration):
         dependencies = [
             ('custom_user', '0017_auto_20200305_1645'),
+        ]
         operations = [
             migrations.AddField(
                 model_name='user',
                 name='last_account_deletion_alert',
                 field=models.DateTimeField(blank=True, null=True, verbose_name='Last account deletion alert'),
             ),
+        ]

             verbose_name=_('Last modification time'),
             db_index=True,
             auto_now=True)
         last_account_deletion_alert = models.DateTimeField(
             verbose_name=_('Last account deletion alert'),
             null=True,
             blank=True)
         objects = UserManager.from_queryset(UserQuerySet)()
         attributes = AttributesDescriptor()

     from __future__ import print_function
     import json
     import logging
     import datetime
     import smtplib
     from datetime import timedelta
     from django.contrib.auth import get_user_model
     from django.core.management.base import BaseCommand, CommandError
     from django.core.mail import send_mail
     from django.utils.timezone import now
     from django.utils import timezone
     from django.template.loader import render_to_string
     from django_rbac.utils import get_ou_model
     from authentic2.models import DeletedUser
     from authentic2.utils import send_templated_mail
     from django.conf import settings
     logger = logging.getLogger(__name__)
     User = get_user_model()
     def print_table(table):
         col_width = [max(len(x) for x in col) for col in zip(*table)]
-...
         help = '''Clean unused accounts'''
         def add_arguments(self, parser):
             parser.add_argument('clean_threshold', type=int)
             parser.add_argument(
                 "--alert-thresholds",
                 help='list of durations before sending an alert '
                      'message for unused account, default is none',
                 default=None)
             parser.add_argument(
                 "--period", type=int,
                 help='period between two calls to '
                      'clean-unused-accounts as days, default is 1',
                 default=1
+            )
             parser.add_argument("--fake", action='store_true', help='do nothing', default=False)
             parser.add_argument(
                 "--filter", help='filter to apply to the user queryset, '
                 'the Django filter key and value are separated by character =', action='append',
                 default=[]
+            )
             parser.add_argument(
                 '--from-email', default=settings.DEFAULT_FROM_EMAIL,
                 help='sender address for notifications, default is DEFAULT_FROM_EMAIL from settings'
+            )
         def handle(self, *args, **options):
             try:
-...
                 logger.exception('failure while cleaning unused accounts')
         def clean_unused_acccounts(self, *args, **options):
             if options['period'] < 1:
                 raise CommandError('period must be > 0')
             clean_threshold = options['clean_threshold']
             if clean_threshold < 1:
                 raise CommandError('clean_threshold must be an integer > 0')
             if options['verbosity'] == '0':
                 logging.basicConfig(level=logging.CRITICAL)
             if options['verbosity'] == '1':
-...
             elif options['verbosity'] == '3':
                 logging.basicConfig(level=logging.DEBUG)
             n = now().replace(hour=0, minute=0, second=0, microsecond=0)
             now = timezone.now()
             self.fake = options['fake']
             self.from_email = options['from_email']
             if self.fake:
                 logger.info('fake call to clean-unused-accounts')
             users = get_user_model().objects.all()
             if options['filter']:
                 for f in options['filter']:
                     key, value = f.split('=', 1)
                     try:
                         users = users.filter(**{key: value})
                     except Exception:
                         raise CommandError('invalid --filter %s' % f)
             if options['alert_thresholds']:
                 alert_thresholds = options['alert_thresholds']
                 alert_thresholds = alert_thresholds.split(',')
                 try:
                     alert_thresholds = map(int, alert_thresholds)
                 except ValueError:
                     raise CommandError('alert_thresholds must be a comma separated list of integers')
                 for threshold in alert_thresholds:
                     if not (0 < threshold < clean_threshold):
                         raise CommandError(
                             'alert-threshold must a positive integer inferior to clean-threshold: 0 < %d < %d' % (
                                 threshold, clean_threshold))
                 for threshold in alert_thresholds:
                     a = n - datetime.timedelta(days=threshold)
                     b = n - datetime.timedelta(days=threshold - options['period'])
                     for user in users.filter(last_login__lt=b, last_login__gte=a):
                         logger.info('%s last login %d days ago, sending alert', user, threshold)
                         self.send_alert(user, threshold, clean_threshold - threshold)
             threshold = n - datetime.timedelta(days=clean_threshold)
             for user in users.filter(last_login__lt=threshold):
                 d = n - user.last_login
                 logger.info('%s last login %d days ago, deleting user', user, d.days)
                 self.delete_user(user, clean_threshold)
         def send_alert(self, user, threshold, clean_threshold):
             for ou in get_ou_model().objects.filter(clean_unused_accounts_alert__isnull=False):
                 alert_delay = timedelta(days=ou.clean_unused_accounts_alert)
                 deletion_delay = timedelta(days=ou.clean_unused_accounts_deletion)
                 users = User.objects.filter(ou=ou, last_login__lte=now-alert_delay)
                 for user in users.filter(last_account_deletion_alert__isnull=True):
                     logger.info('%s last login %d days ago, sending alert', user,
                              ou.clean_unused_accounts_alert)
                     self.send_alert(user)
                 to_delete = users.filter(
                     last_login__lte=now - deletion_delay,
                     # ensure respect of alert delay before deletion
                     last_account_deletion_alert__lte=now - (deletion_delay - alert_delay)
+                )
                 for user in to_delete:
                     logger.info('%s last login more than %d days ago, deleting user',
                              user, ou.clean_unused_accounts_deletion)
                     self.delete_user(user)
         def send_alert(self, user):
             alert_delay = user.ou.clean_unused_accounts_alert
             days_to_deletion = user.ou.clean_unused_accounts_deletion - alert_delay
             ctx = {
                 'user': user,
                 'threshold': threshold,
                 'clean_threshold': clean_threshold
                 'threshold': alert_delay,
                 'clean_threshold': days_to_deletion,
+            }
             self.send_mail('authentic2/unused_account_alert', user, ctx)
             try:
                 self.send_mail('authentic2/unused_account_alert', user, ctx)
             except smtplib.SMTPException as e:
                 logger.exception('email sending failure: %s', e)
             else:
                 if not self.fake:
                     user.last_account_deletion_alert = timezone.now()
                     user.save()
         def send_mail(self, prefix, user, ctx):
             if not user.email:
                 logger.debug('%s has no email, no mail sent', user)
             subject = render_to_string(prefix + '_subject.txt', ctx).strip()
             body = render_to_string(prefix + '_body.txt', ctx)
             if not self.fake:
                 try:
                     logger.debug('sending mail to %s', user.email)
                     send_mail(subject, body, self.from_email, [user.email])
                 except Exception:
                     logger.exception('email sending failure')
                 logger.debug('sending mail to %s', user.email)
                 send_templated_mail(user.email, prefix, ctx)
         def delete_user(self, user, threshold):
         def delete_user(self, user):
             ctx = {
                 'user': user,
                 'threshold': threshold
                 'threshold': user.ou.clean_unused_accounts_deletion
+            }
             self.send_mail('authentic2/unused_account_delete', user, ctx)
             try:
                 self.send_mail('authentic2/unused_account_delete', user, ctx)
             except smtplib.SMTPException as e:
                 logger.exception('email sending failure: %s', e)
             if not self.fake:
                 DeletedUser.objects.delete_user(user)

             model = get_ou_model()
             fields = (
                 'name', 'default', 'username_is_unique', 'email_is_unique', 'validate_emails',
                 'show_username'
                 'show_username', 'clean_unused_accounts_alert', 'clean_unused_accounts_deletion'
+            )

         if constants.LAST_LOGIN_SESSION_KEY not in request.session:
             request.session[constants.LAST_LOGIN_SESSION_KEY] = \
                 localize(to_current_timezone(last_login), True)
         user.last_account_deletion_alert = None
         user.save()
         record_authentication_event(request, how, nonce=nonce)
         hooks.call_hooks('event', name='login', user=user, how=how, service=service_slug)
         return continue_to_next_url(request, **kwargs)

         # 5 global roles and 4 ou roles for both ous
         assert Role.objects.count() == 5 + 4 + 4
     @pytest.mark.parametrize(
         'alert,deletion', [(-1, 31), (31, -1), (0, 31), (31, 0), (None, 31), (31, None), (32, 31)]
+    )
     def test_unused_account_settings_validation(ou1, alert, deletion):
         ou1.clean_unused_accounts_alert = alert
         ou1.clean_unused_accounts_deletion = deletion
         with pytest.raises(ValidationError):
             ou1.full_clean()

tests/test_all.py
83	83	'is_staff': False,
84	84	'is_superuser': False,
85	85	'last_login': u.last_login,
	86	'last_account_deletion_alert': None,
86	87	'date_joined': u.date_joined,
87	88	'modified': u.modified,
88	89	'groups': [],

                         'first_name', 'first_name_verified', 'last_name',
                         'last_name_verified', 'date_joined', 'last_login',
                         'username', 'password', 'email', 'is_active', 'title',
                         'title_verified', 'modified', 'email_verified']) == set(resp.json.keys())
                         'title_verified', 'modified', 'email_verified',
                         'last_account_deletion_alert']) == set(resp.json.keys())
             assert resp.json['first_name'] == payload['first_name']
             assert resp.json['last_name'] == payload['last_name']
             assert resp.json['email'] == payload['email']
-...
                         'first_name', 'first_name_verified', 'last_name',
                         'last_name_verified', 'date_joined', 'last_login',
                         'username', 'password', 'email', 'is_active', 'title',
                         'title_verified', 'modified', 'email_verified']) == set(resp.json.keys())
                         'title_verified', 'modified', 'email_verified',
                         'last_account_deletion_alert']) == set(resp.json.keys())
             user = get_user_model().objects.get(pk=resp.json['id'])
             assert AttributeValue.objects.with_owner(user).filter(verified=True).count() == 3
             assert AttributeValue.objects.with_owner(user).filter(verified=False).count() == 0

     from authentic2_auth_oidc.models import OIDCProvider
     from django_rbac.utils import get_ou_model
     from utils import login
     if six.PY2:
         FileType = file
     else:
-...
         assert old_pass != simple_user.password
     def test_clean_unused_account(simple_user):
     def test_clean_unused_account(db, simple_user, mailoutbox, freezer):
         freezer.move_to('2018-01-01')
         simple_user.ou.clean_unused_accounts_alert = 2
         simple_user.ou.clean_unused_accounts_deletion = 3
         simple_user.ou.save()
         simple_user.last_login = now() - datetime.timedelta(days=2)
         simple_user.save()
         management.call_command('clean-unused-accounts', '1')
         assert DeletedUser.objects.get(user=simple_user)
         management.call_command('clean-unused-accounts')
         assert not DeletedUser.objects.filter(user=simple_user).exists()
         assert len(mailoutbox) == 1
         freezer.move_to('2018-01-01 12:00:00')
         # no new mail, no deletion
         management.call_command('clean-unused-accounts')
         assert not DeletedUser.objects.filter(user=simple_user).exists()
         assert len(mailoutbox) == 1
         freezer.move_to('2018-01-02')
         management.call_command('clean-unused-accounts')
         assert DeletedUser.objects.filter(user=simple_user).exists()
         assert len(mailoutbox) == 2
     def test_clean_unused_account_user_logs_in(app, db, simple_user, mailoutbox, freezer):
         freezer.move_to('2018-01-01')
         simple_user.ou.clean_unused_accounts_alert = 2
         simple_user.ou.clean_unused_accounts_deletion = 3
         simple_user.ou.save()
         simple_user.last_login = now() - datetime.timedelta(days=2)
         simple_user.save()
         management.call_command('clean-unused-accounts')
         assert len(mailoutbox) == 1
         login(app, simple_user)
         # the day of deletion, nothing happens
         freezer.move_to('2018-01-02')
         assert not DeletedUser.objects.filter(user=simple_user).exists()
         assert len(mailoutbox) == 1
         # when new alert delay is reached, user gets alerted again
         freezer.move_to('2018-01-04')
         management.call_command('clean-unused-accounts')
         assert not DeletedUser.objects.filter(user=simple_user).exists()
         assert len(mailoutbox) == 2
     def test_clean_unused_account_disabled_by_default(db, simple_user, mailoutbox):
         simple_user.last_login = now() - datetime.timedelta(days=2)
         simple_user.save()
         management.call_command('clean-unused-accounts')
         assert not DeletedUser.objects.filter(user=simple_user).exists()
         assert len(mailoutbox) == 0
     def test_clean_unused_account_always_alert(db, simple_user, mailoutbox, freezer):
         simple_user.ou.clean_unused_accounts_alert = 2
         simple_user.ou.clean_unused_accounts_deletion = 3  # one day between alert and actual deletion
         simple_user.ou.save()
         simple_user.last_login = now() - datetime.timedelta(days=4)
         simple_user.save()
         # even if account last login in past deletion delay, an alert is always sent first
         management.call_command('clean-unused-accounts')
         assert not len(DeletedUser.objects.filter(user=simple_user))
         assert len(mailoutbox) == 1
         # and calling again as no effect, since one day must pass before account is deleted
         management.call_command('clean-unused-accounts')
         assert not len(DeletedUser.objects.filter(user=simple_user))
         assert len(mailoutbox) == 1
     def test_cleanupauthentic(db):
+    -

Projet

Général

Profil

Produits Entr'ouvert » Authentic 2

0002-commands-per-OU-unused-user-accounts-cleaning-policy.patch