diff --git a/modoboa/admin/management/commands/subcommands/_repair.py b/modoboa/admin/management/commands/subcommands/_repair.py index d41013fcf..ff958714a 100644 --- a/modoboa/admin/management/commands/subcommands/_repair.py +++ b/modoboa/admin/management/commands/subcommands/_repair.py @@ -1,4 +1,6 @@ """Management command to check and fix known problems.""" + +from django.conf import settings from django.core.management.base import BaseCommand from modoboa.lib.permissions import grant_access_to_object @@ -7,20 +9,88 @@ from modoboa.lib.permissions import get_object_owner from modoboa.core.models import User from modoboa.admin import models -try: - from modoboa_postfix_autoreply.models import ARmessage -except ImportError: - ARmessage = None - - known_problems = [] def known_problem(func): - known_problems.append(func.__name__) + """Simple decorator to register a known problem.""" + known_problems.append(func) return func +def log(message, quiet=False, **options): + if not quiet: + print(message) + + +def fix_owner(qs, dry_run=False, **options): + """Fix ownership for orphan objects.""" + model = qs.model + for obj in qs: + kw = {"cls": model.__name__, "obj": obj} + if get_object_owner(obj) is not None: + continue + if dry_run: + log(" {cls} {obj} has no owner".format(**kw), **options) + continue + if isinstance(obj, User): + admin = User.objects.filter( + is_superuser=True, is_active=True).first() + elif isinstance(obj, models.Domain): + admin = obj.admins.first() + elif isinstance(obj, models.DomainAlias): + admin = obj.target.admins.first() + else: + admin = obj.domain.admins.first() + if not admin: + # Fallback: use the first superuser found + admin = User.objects.filter( + is_superuser=True, is_active=True).first() + grant_access_to_object(admin, obj, is_owner=True) + kw["admin"] = admin + log(" {cls} {obj} is now owned by {admin}".format(**kw), + **options) + + +@known_problem +def sometimes_objects_have_no_owner(**options): + """Sometime objects have no owner.""" + owned_models = ( + User.objects.all(), + models.Domain.objects.all(), + models.DomainAlias.objects.all(), + models.Alias.objects.filter(domain__isnull=False), + models.Mailbox.objects.filter(domain__isnull=False), + ) + for qs in owned_models: + fix_owner(qs, **options) + + +@known_problem +def sometimes_mailbox_have_no_alias(**options): + """Sometime mailboxes have no alias.""" + alias_created = 0 + recipient_created = 0 + for instance in models.Mailbox.objects.select_related("domain").all(): + alias, created = models.Alias.objects.get_or_create( + address=instance.full_address, + domain=instance.domain, + internal=True) + if created: + alias_created += 1 + log("Alias {0} created".format(alias)) + recipient, created = models.AliasRecipient.objects.get_or_create( + alias=alias, + address=instance.full_address, + r_mailbox=instance) + if created: + recipient_created += 1 + log("AliasRecipient {0} created".format(recipient)) + if alias_created or recipient_created: + log("{0} alias created. {1} alias recipient created".format( + alias_created, recipient_created)) + + class Repair(BaseCommand): """Command class.""" @@ -37,107 +107,14 @@ class Repair(BaseCommand): def handle(self, *args, **options): """Command entry point.""" - for problem in known_problems: - func = getattr(self, problem) + # Load known problems from extensions. + for ext in settings.MODOBOA_APPS: + try: + __import__(ext, locals(), globals(), ["known_problems"]) + except ImportError: + pass + for func in known_problems: title = func.__doc__.strip() - self.log("", **options) - self.log("Checking for... {}...".format(title), **options) + log("", **options) + log("Checking for... {}...".format(title), **options) func(**options) - - def log(self, message, quiet=False, **options): - if not quiet: - print(message) - - def fix_owner(self, qs, dry_run=False, **options): - model = qs.model - for obj in qs: - kw = dict( - cls=model.__name__, - obj=obj - ) - if get_object_owner(obj) is None: - if dry_run: - self.log( - " {cls} {obj} has no owner".format(**kw), - **options) - else: - if isinstance(obj, User): - admin = User.objects.filter(is_superuser=True, - is_active=True).first() - elif isinstance(obj, models.Domain): - admin = obj.admins.first() - elif isinstance(obj, models.DomainAlias): - admin = obj.target.admins.first() - else: - admin = obj.domain.admins.first() - if not admin: - # domain has no admin. use the first superuser found - admin = User.objects.filter(is_superuser=True, - is_active=True).first() - grant_access_to_object(admin, obj, is_owner=True) - kw['admin'] = admin - self.log( - " {cls} {obj} is now owned by {admin}".format(**kw), - **options) - - @known_problem - def sometimes_objects_have_no_owner(self, **options): - """Sometime objects have no owner""" - owned_models = ( - User.objects.all(), - models.Domain.objects.all(), - models.DomainAlias.objects.all(), - models.Alias.objects.filter(domain__isnull=False), - models.Mailbox.objects.filter(domain__isnull=False), - ) - for qs in owned_models: - self.fix_owner(qs, **options) - - @known_problem - def ensure_autoreplies_recipents_are_valids(self, **options): - """Sometime autoreply alias exists when ARmessage is not enabled""" - if ARmessage is None: - return - deleted = 0 - qs = models.AliasRecipient.objects.filter( - address__contains='@autoreply.') - for alias in qs: - address, domain = alias.alias.address.split('@') - arqs = ARmessage.objects.filter( - mbox__address=address, - mbox__domain__name=domain) - if not arqs.count(): - self.log('Delete {0} (No AR found)'.format(alias)) - deleted += 1 - else: - for ar in arqs: - if not ar.enabled: - self.log('Delete {0} (AR disabled)'.format(alias)) - alias.delete() - deleted += 1 - if deleted: - self.log('{0} alias recipient deleted'.format(deleted)) - - @known_problem - def sometimes_mailbox_have_no_alias(self, **options): - """Sometime mailboxes have no alias""" - alias_created = 0 - recipient_created = 0 - for instance in models.Mailbox.objects.all(): - alias, created = models.Alias.objects.get_or_create( - address=instance.full_address, - domain=instance.domain, - internal=True) - if created: - alias_created += 1 - self.log('Alias {0} created'.format(alias)) - recipient, created = models.AliasRecipient.objects.get_or_create( - alias=alias, - address=instance.full_address, - r_mailbox=instance) - if created: - recipient_created += 1 - self.log('AliasRecipient {0} created'.format(recipient)) - if alias_created or recipient_created: - self.log('{0} alias created. {1} alias recipient created'.format( - alias_created, recipient_created)) diff --git a/modoboa/admin/tests/test_repair.py b/modoboa/admin/tests/test_repair.py index 0674abdfa..58e66e1e2 100644 --- a/modoboa/admin/tests/test_repair.py +++ b/modoboa/admin/tests/test_repair.py @@ -45,7 +45,7 @@ class RepairTestCase(ModoTestCase): self.assertIs(get_object_owner(mbox), None) def test_management_command_with_nul_domain(self): - """ Just assume nothing raise when an alias has no domain""" + """Just assume nothing raise when an alias has no domain.""" models.Alias.objects.create(address='@modoboa.xxx') ret = management.call_command("modo", "repair", "--quiet") assert ret is None