0001-per-OU-unused-user-accounts-cleaning-policy-26909.patch

Paul Marillonnet, 29 mars 2019 18:29

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

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

 .../migrations/0020_auto_20190329_1539.py     |  35 +++++
 src/authentic2/a2_rbac/models.py              |  19 +++
 .../0017_user_last_account_deletion_alert.py  |  20 +++
 src/authentic2/custom_user/models.py          |   2 +
 .../commands/clean-unused-accounts.py         | 139 ++++++++----------
 tests/test_api.py                             |   6 +-
 tests/test_commands.py                        |  23 ++-
 7 files changed, 163 insertions(+), 81 deletions(-)
 create mode 100644 src/authentic2/a2_rbac/migrations/0020_auto_20190329_1539.py
 create mode 100644 src/authentic2/custom_user/migrations/0017_user_last_account_deletion_alert.py

     # -*- coding: utf-8 -*-
     # Generated by Django 1.11.18 on 2019-03-29 14:39
     from __future__ import unicode_literals
     from django.db import migrations, models
     class Migration(migrations.Migration):
         dependencies = [
             ('a2_rbac', '0019_organizationalunit_show_username'),
+        ]
         operations = [
             migrations.AddField(
                 model_name='organizationalunit',
                 name='clean_unused_accounts_alert',
                 field=models.IntegerField(default=0, verbose_name='Days after which the user receives an account deletion alert'),
             ),
             migrations.AddField(
                 model_name='organizationalunit',
                 name='clean_unused_accounts_deletion',
                 field=models.IntegerField(default=0, verbose_name='Delay in days before cleaning unused accounts'),
             ),
             migrations.AddField(
                 model_name='organizationalunit',
                 name='clean_unused_accounts_filter',
                 field=models.CharField(blank=True, default=None, max_length=511, null=True, verbose_name='json dict filter to apply for unused account cleanup'),
             ),
             migrations.AddField(
                 model_name='organizationalunit',
                 name='clean_unused_accounts_from_email',
                 field=models.EmailField(blank=True, default=None, max_length=255, null=True, verbose_name='sender email address for account deletion warnings and notifications'),
             ),
+        ]

     import os
     from collections import namedtuple
     from django.core.exceptions import ValidationError
     from django.utils.translation import ugettext_lazy as _
-...
             choices=USER_ADD_PASSWD_POLICY_CHOICES,
             default=0)
         clean_unused_accounts_alert = models.IntegerField(
             verbose_name=_('Days after which the user receives an account deletion '
                 'alert'),
             default=0)
         clean_unused_accounts_deletion = models.IntegerField(
             verbose_name=_('Delay in days before cleaning unused accounts'),
             default=0)
         clean_unused_accounts_from_email = models.EmailField(
             verbose_name=_('sender email address for account deletion warnings and notifications'),
             null=True, blank=True, default=None, max_length=255)
         clean_unused_accounts_filter = models.CharField(
             verbose_name=_('json dict filter to apply for unused account cleanup'),
             null=True, blank=True, default=None, max_length=511)
         objects = managers.OrganizationalUnitManager()
         class Meta:

     # -*- coding: utf-8 -*-
     # Generated by Django 1.11.18 on 2019-03-29 14:39
     from __future__ import unicode_literals
     from django.db import migrations, models
     class Migration(migrations.Migration):
         dependencies = [
             ('custom_user', '0016_auto_20180925_1107'),
+        ]
         operations = [
             migrations.AddField(
                 model_name='user',
                 name='last_account_deletion_alert',
                 field=models.DateTimeField(blank=True, default=None, 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(
             _('last account deletion alert'), null=True, blank=True, default=None)
         objects = UserManager.from_queryset(UserQuerySet)()
         attributes = AttributesDescriptor()

     from __future__ import print_function
     import json
     import logging
     import datetime
     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.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
-...
         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):
             log = logging.getLogger(__name__)
-...
                 log.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':
-...
             log = logging.getLogger(__name__)
             n = now().replace(hour=0, minute=0, second=0, microsecond=0)
             self.fake = options['fake']
             self.from_email = options['from_email']
             if self.fake:
                 log.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)
             for ou in get_ou_model().objects.all():
                 if ou.clean_unused_accounts_filter:
                     try:
                         users = users.filter(**{key: value})
                         for key, value in json.loads(ou.clean_unused_accounts_filter):
                             users = users.filter(**{key: value})
                     except:
                         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):
                         log.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
                 log.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):
             ctx = { 'user': user, 'threshold': threshold,
                     'clean_threshold': clean_threshold }
             self.send_mail('authentic2/unused_account_alert', user, ctx)
         def send_mail(self, prefix, user, ctx):
                         raise CommandError('invalid filter {}: {}'.format(
                             key, value))
                 if not ou.clean_unused_accounts_deletion:
                     continue
                 deletion = timedelta(days=ou.clean_unused_accounts_deletion)
                 alert = timedelta(days=ou.clean_unused_accounts_alert)
                 for user in users.filter(ou=ou, last_login__lt=n-alert):
                     if n - user.last_login >= deletion and \
                             user.last_account_deletion_alert and \
                             n - user.last_account_deletion_alert >= alert:
                         log.info(
                             '%s last login more than %d days ago, deleting user',
                             user, ou.clean_unused_accounts_deletion)
                         self.delete_user(user)
                     elif not user.last_account_deletion_alert or \
                             n - user.last_account_deletion_alert >= alert:
                         log.info('%s last login %d days ago, sending alert', user,
                             ou.clean_unused_accounts_alert)
                         self.send_alert(user)
                         user.last_account_deletion_alert = n
                         user.save()
         def send_alert(self, user):
             ctx = {
                 'user': user,
                 'threshold': user.ou.clean_unused_accounts_alert,
                 'clean_threshold': user.ou.clean_unused_accounts_deletion
+            }
             self.send_mail(
+                [
                     'authentic2/unused_account_alert_{}'.format(user.ou.slug),
                     'authentic2/unused_account_alert' #default fallback template
                 ],
                 user,
                 ctx)
         def send_mail(self, template_names, user, ctx):
             log = logging.getLogger(__name__)
             if not user.email:
                 log.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)
             from_email = user.ou.clean_unused_accounts_from_email
             if not self.fake:
                 try:
                     log.debug('sending mail to %s', user.email)
                     send_mail(subject, body, self.from_email, [user.email])
                     send_templated_mail(
                         user.email, template_names, from_email=from_email)
                 except:
                     log.exception('email sending failure')
         def delete_user(self, user, threshold):
             ctx = { 'user': user, 'threshold': threshold }
             self.send_mail('authentic2/unused_account_delete', user,
                     ctx)
         def delete_user(self, user):
             ctx = {
                 'user': user,
                 'threshold': user.ou.clean_unused_accounts_deletion
+            }
             self.send_mail(
+                [
                     'authentic2/unused_account_delete_{}'.format(user.ou.slug),
                     'authentic2/unused_account_delete' #default fallback template
                 ],
                 user, ctx)
             if not self.fake:
                 DeletedUser.objects.delete_user(user)

                         '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.models import Attribute, DeletedUser
     from authentic2_auth_oidc.models import OIDCProvider
     from django.contrib.auth import get_user_model
     from django_rbac.utils import get_ou_model
-...
         assert old_pass != simple_user.password
     def test_clean_unused_account(simple_user):
         simple_user.last_login = now() - datetime.timedelta(days=2)
     def test_clean_unused_account(db, simple_user, mailoutbox):
         User = get_user_model()
         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=4)
         simple_user.save()
         management.call_command('clean-unused-accounts')
         assert not len(DeletedUser.objects.filter(user=simple_user))
         assert len(mailoutbox) == 1
         simple_user.last_account_deletion_alert = now() - datetime.timedelta(days=4)
         simple_user.last_login = now() - datetime.timedelta(days=5)
         simple_user.save()
         management.call_command('clean-unused-accounts', '1')
         management.call_command('clean-unused-accounts')
         assert DeletedUser.objects.get(user=simple_user)
         assert len(mailoutbox) == 2
     def test_cleanupauthentic(db):
+    -

Projet

Général

Profil

Produits Entr'ouvert » Authentic 2

0001-per-OU-unused-user-accounts-cleaning-policy-26909.patch