Project

General

Profile

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

Paul Marillonnet, 29 Mar 2019 04:16 PM

Download (15 KB)

View differences:

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_commands.py                        |  23 ++-
 6 files changed, 159 insertions(+), 79 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
src/authentic2/a2_rbac/migrations/0020_auto_20190329_1539.py
1
# -*- coding: utf-8 -*-
2
# Generated by Django 1.11.18 on 2019-03-29 14:39
3
from __future__ import unicode_literals
4

  
5
from django.db import migrations, models
6

  
7

  
8
class Migration(migrations.Migration):
9

  
10
    dependencies = [
11
        ('a2_rbac', '0019_organizationalunit_show_username'),
12
    ]
13

  
14
    operations = [
15
        migrations.AddField(
16
            model_name='organizationalunit',
17
            name='clean_unused_accounts_alert',
18
            field=models.IntegerField(default=0, verbose_name='Days after which the user receives an account deletion alert'),
19
        ),
20
        migrations.AddField(
21
            model_name='organizationalunit',
22
            name='clean_unused_accounts_deletion',
23
            field=models.IntegerField(default=0, verbose_name='Delay in days before cleaning unused accounts'),
24
        ),
25
        migrations.AddField(
26
            model_name='organizationalunit',
27
            name='clean_unused_accounts_filter',
28
            field=models.CharField(blank=True, default=None, max_length=511, null=True, verbose_name='json dict filter to apply for unused account cleanup'),
29
        ),
30
        migrations.AddField(
31
            model_name='organizationalunit',
32
            name='clean_unused_accounts_from_email',
33
            field=models.EmailField(blank=True, default=None, max_length=255, null=True, verbose_name='sender email address for account deletion warnings and notifications'),
34
        ),
35
    ]
src/authentic2/a2_rbac/models.py
1
import os
2

  
1 3
from collections import namedtuple
2 4
from django.core.exceptions import ValidationError
3 5
from django.utils.translation import ugettext_lazy as _
......
76 78
        choices=USER_ADD_PASSWD_POLICY_CHOICES,
77 79
        default=0)
78 80

  
81
    clean_unused_accounts_alert = models.IntegerField(
82
        verbose_name=_('Days after which the user receives an account deletion '
83
            'alert'),
84
        default=0)
85

  
86
    clean_unused_accounts_deletion = models.IntegerField(
87
        verbose_name=_('Delay in days before cleaning unused accounts'),
88
        default=0)
89

  
90
    clean_unused_accounts_from_email = models.EmailField(
91
        verbose_name=_('sender email address for account deletion warnings and notifications'),
92
        null=True, blank=True, default=None, max_length=255)
93

  
94
    clean_unused_accounts_filter = models.CharField(
95
        verbose_name=_('json dict filter to apply for unused account cleanup'),
96
        null=True, blank=True, default=None, max_length=511)
97

  
79 98
    objects = managers.OrganizationalUnitManager()
80 99

  
81 100
    class Meta:
src/authentic2/custom_user/migrations/0017_user_last_account_deletion_alert.py
1
# -*- coding: utf-8 -*-
2
# Generated by Django 1.11.18 on 2019-03-29 14:39
3
from __future__ import unicode_literals
4

  
5
from django.db import migrations, models
6

  
7

  
8
class Migration(migrations.Migration):
9

  
10
    dependencies = [
11
        ('custom_user', '0016_auto_20180925_1107'),
12
    ]
13

  
14
    operations = [
15
        migrations.AddField(
16
            model_name='user',
17
            name='last_account_deletion_alert',
18
            field=models.DateTimeField(blank=True, default=None, null=True, verbose_name='last account deletion alert'),
19
        ),
20
    ]
src/authentic2/custom_user/models.py
126 126
        verbose_name=_('Last modification time'),
127 127
        db_index=True,
128 128
        auto_now=True)
129
    last_account_deletion_alert = models.DateTimeField(
130
        _('last account deletion alert'), null=True, blank=True, default=None)
129 131

  
130 132
    objects = UserManager.from_queryset(UserQuerySet)()
131 133
    attributes = AttributesDescriptor()
src/authentic2/management/commands/clean-unused-accounts.py
1 1
from __future__ import print_function
2 2

  
3
import json
3 4
import logging
4
import datetime
5 5

  
6
from datetime import timedelta
6 7
from django.contrib.auth import get_user_model
7 8
from django.core.management.base import BaseCommand, CommandError
8
from django.core.mail import send_mail
9 9
from django.utils.timezone import now
10 10
from django.template.loader import render_to_string
11
from django_rbac.utils import get_ou_model
11 12

  
12 13
from authentic2.models import DeletedUser
14
from authentic2.utils import send_templated_mail
13 15

  
14 16
from django.conf import settings
15 17

  
......
24 26
    help = '''Clean unused accounts'''
25 27

  
26 28
    def add_arguments(self, parser):
27
        parser.add_argument('clean_threshold', type=int)
28
        parser.add_argument(
29
            "--alert-thresholds",
30
            help='list of durations before sending an alert '
31
                 'message for unused account, default is none',
32
            default=None)
33
        parser.add_argument(
34
            "--period", type=int,
35
            help='period between two calls to '
36
                 'clean-unused-accounts as days, default is 1',
37
            default=1
38
        )
39 29
        parser.add_argument("--fake", action='store_true', help='do nothing', default=False)
40
        parser.add_argument(
41
            "--filter", help='filter to apply to the user queryset, '
42
            'the Django filter key and value are separated by character =', action='append',
43
            default=[]
44
        )
45
        parser.add_argument(
46
            '--from-email', default=settings.DEFAULT_FROM_EMAIL,
47
            help='sender address for notifications, default is DEFAULT_FROM_EMAIL from settings'
48
        )
49 30

  
50 31
    def handle(self, *args, **options):
51 32
        log = logging.getLogger(__name__)
......
55 36
            log.exception('failure while cleaning unused accounts')
56 37

  
57 38
    def clean_unused_acccounts(self, *args, **options):
58
        if options['period'] < 1:
59
            raise CommandError('period must be > 0')
60

  
61
        clean_threshold = options['clean_threshold']
62
        if clean_threshold < 1:
63
            raise CommandError('clean_threshold must be an integer > 0')
64

  
65 39
        if options['verbosity'] == '0':
66 40
            logging.basicConfig(level=logging.CRITICAL)
67 41
        if options['verbosity'] == '1':
......
74 48
        log = logging.getLogger(__name__)
75 49
        n = now().replace(hour=0, minute=0, second=0, microsecond=0)
76 50
        self.fake = options['fake']
77
        self.from_email = options['from_email']
78 51
        if self.fake:
79 52
            log.info('fake call to clean-unused-accounts')
80 53
        users = get_user_model().objects.all()
81
        if options['filter']:
82
            for f in options['filter']:
83
                key, value = f.split('=', 1)
54
        for ou in get_ou_model().objects.all():
55
            if ou.clean_unused_accounts_filter:
84 56
                try:
85
                    users = users.filter(**{key: value})
57
                    for key, value in json.loads(ou.clean_unused_accounts_filter):
58
                        users = users.filter(**{key: value})
86 59
                except:
87
                    raise CommandError('invalid --filter %s' % f)
88
        if options['alert_thresholds']:
89
            alert_thresholds = options['alert_thresholds']
90
            alert_thresholds = alert_thresholds.split(',')
91
            try:
92
                alert_thresholds = map(int, alert_thresholds)
93
            except ValueError:
94
                raise CommandError('alert_thresholds must be a comma '
95
                        'separated list of integers')
96
            for threshold in alert_thresholds:
97
                if not (0 < threshold < clean_threshold):
98
                    raise CommandError('alert-threshold must a positive integer '
99
                            'inferior to clean-threshold: 0 < %d < %d' % (
100
                                threshold, clean_threshold))
101
            for threshold in alert_thresholds:
102
                a = n - datetime.timedelta(days=threshold)
103
                b = n - datetime.timedelta(days=threshold-options['period'])
104
                for user in users.filter(last_login__lt=b, last_login__gte=a):
105
                    log.info('%s last login %d days ago, sending alert', user, threshold)
106
                    self.send_alert(user, threshold, clean_threshold-threshold)
107
        threshold = n - datetime.timedelta(days=clean_threshold)
108
        for user in users.filter(last_login__lt=threshold):
109
            d = n - user.last_login
110
            log.info('%s last login %d days ago, deleting user', user, d.days)
111
            self.delete_user(user, clean_threshold)
112

  
113

  
114
    def send_alert(self, user, threshold, clean_threshold):
115
        ctx = { 'user': user, 'threshold': threshold,
116
                'clean_threshold': clean_threshold }
117
        self.send_mail('authentic2/unused_account_alert', user, ctx)
118

  
119

  
120
    def send_mail(self, prefix, user, ctx):
60
                    raise CommandError('invalid filter {}: {}'.format(
61
                        key, value))
62
            if not ou.clean_unused_accounts_deletion:
63
                continue
64

  
65
            deletion = timedelta(days=ou.clean_unused_accounts_deletion)
66
            alert = timedelta(days=ou.clean_unused_accounts_alert)
67

  
68
            for user in users.filter(ou=ou, last_login__lt=n-alert):
69
                if n - user.last_login >= deletion and \
70
                        user.last_account_deletion_alert and \
71
                        n - user.last_account_deletion_alert >= alert:
72
                    log.info(
73
                        '%s last login more than %d days ago, deleting user',
74
                        user, ou.clean_unused_accounts_deletion)
75
                    self.delete_user(user)
76
                elif not user.last_account_deletion_alert or \
77
                        n - user.last_account_deletion_alert >= alert:
78
                    log.info('%s last login %d days ago, sending alert', user,
79
                        ou.clean_unused_accounts_alert)
80
                    self.send_alert(user)
81
                    user.last_account_deletion_alert = n
82
                    user.save()
83

  
84

  
85
    def send_alert(self, user):
86
        ctx = {
87
            'user': user,
88
            'threshold': user.ou.clean_unused_accounts_alert,
89
            'clean_threshold': user.ou.clean_unused_accounts_deletion
90
        }
91
        self.send_mail(
92
            [
93
                'authentic2/unused_account_alert_{}'.format(user.ou.slug),
94
                'authentic2/unused_account_alert' #default fallback template
95
            ],
96
            user,
97
            ctx)
98

  
99

  
100
    def send_mail(self, template_names, user, ctx):
121 101
        log = logging.getLogger(__name__)
122 102

  
123 103
        if not user.email:
124 104
            log.debug('%s has no email, no mail sent', user)
125
        subject = render_to_string(prefix + '_subject.txt', ctx).strip()
126
        body = render_to_string(prefix + '_body.txt', ctx)
105
        from_email = user.ou.clean_unused_accounts_from_email
127 106
        if not self.fake:
128 107
            try:
129 108
                log.debug('sending mail to %s', user.email)
130
                send_mail(subject, body, self.from_email, [user.email])
109
                send_templated_mail(
110
                    user.email, template_names, from_email=from_email)
131 111
            except:
132 112
                log.exception('email sending failure')
133 113

  
134 114

  
135
    def delete_user(self, user, threshold):
136
        ctx = { 'user': user, 'threshold': threshold }
137
        self.send_mail('authentic2/unused_account_delete', user,
138
                ctx)
115
    def delete_user(self, user):
116
        ctx = {
117
            'user': user,
118
            'threshold': user.ou.clean_unused_accounts_deletion
119
        }
120
        self.send_mail(
121
            [
122
                'authentic2/unused_account_delete_{}'.format(user.ou.slug),
123
                'authentic2/unused_account_delete' #default fallback template
124
            ],
125
            user, ctx)
139 126
        if not self.fake:
140 127
            DeletedUser.objects.delete_user(user)
tests/test_commands.py
8 8

  
9 9
from authentic2.models import Attribute, DeletedUser
10 10
from authentic2_auth_oidc.models import OIDCProvider
11
from django.contrib.auth import get_user_model
11 12
from django_rbac.utils import get_ou_model
12 13

  
13 14

  
......
24 25
    assert old_pass != simple_user.password
25 26

  
26 27

  
27
def test_clean_unused_account(simple_user):
28
    simple_user.last_login = now() - datetime.timedelta(days=2)
28
def test_clean_unused_account(db, simple_user, mailoutbox):
29
    User = get_user_model()
30

  
31
    simple_user.ou.clean_unused_accounts_alert = 2
32
    simple_user.ou.clean_unused_accounts_deletion = 3
33
    simple_user.ou.save()
34

  
35
    simple_user.last_login = now() - datetime.timedelta(days=4)
36
    simple_user.save()
37

  
38
    management.call_command('clean-unused-accounts')
39
    assert not len(DeletedUser.objects.filter(user=simple_user))
40
    assert len(mailoutbox) == 1
41

  
42
    simple_user.last_account_deletion_alert = now() - datetime.timedelta(days=4)
43
    simple_user.last_login = now() - datetime.timedelta(days=5)
29 44
    simple_user.save()
30
    management.call_command('clean-unused-accounts', '1')
45

  
46
    management.call_command('clean-unused-accounts')
31 47
    assert DeletedUser.objects.get(user=simple_user)
48
    assert len(mailoutbox) == 2
32 49

  
33 50

  
34 51
def test_cleanupauthentic(db):
35
-