diff --git a/modoboa/admin/callbacks.py b/modoboa/admin/callbacks.py index b894673c0..163bf49aa 100644 --- a/modoboa/admin/callbacks.py +++ b/modoboa/admin/callbacks.py @@ -3,6 +3,7 @@ from django.core.urlresolvers import reverse from django.utils.translation import ugettext as _, ugettext_lazy +from modoboa.core import signals as core_signals from modoboa.lib import parameters, events from modoboa.lib.email_utils import split_mailbox from modoboa.lib.exceptions import PermDeniedException, BadRequest, Conflict @@ -103,7 +104,6 @@ def import_account_mailbox(user, account, row): """ account.email = row[0].strip() if account.email: - account.save() mailbox, domname = split_mailbox(account.email) try: domain = Domain.objects.get(name=domname) @@ -114,6 +114,11 @@ def import_account_mailbox(user, account, row): ) if not user.can_access(domain): raise PermDeniedException + core_signals.can_create_object.send( + sender="import", context=user, object_type="mailboxes") + core_signals.can_create_object.send( + sender="import", context=domain, object_type="mailboxes") + account.save() if Mailbox.objects.filter(address=mailbox, domain=domain).exists(): raise Conflict( _("Mailbox {} already exists").format(account.email)) diff --git a/modoboa/admin/forms/account.py b/modoboa/admin/forms/account.py index 098c6cc56..455290f6f 100644 --- a/modoboa/admin/forms/account.py +++ b/modoboa/admin/forms/account.py @@ -13,7 +13,7 @@ from modoboa.core.models import User from modoboa.core import signals as core_signals from modoboa.lib import events, parameters from modoboa.lib.email_utils import split_mailbox -from modoboa.lib.exceptions import PermDeniedException +from modoboa.lib import exceptions as lib_exceptions from modoboa.lib import fields as lib_fields from modoboa.lib.form_utils import ( DynamicForm, TabForms, WizardForm, WizardStep @@ -161,7 +161,8 @@ class AccountFormGeneral(forms.ModelForm): def save(self, commit=True): account = super(AccountFormGeneral, self).save(commit=False) if self.user == account and not self.cleaned_data["is_active"]: - raise PermDeniedException(_("You can't disable your own account")) + raise lib_exceptions.PermDeniedException( + _("You can't disable your own account")) if commit: if "password1" in self.cleaned_data \ and self.cleaned_data["password1"] != "": @@ -198,11 +199,7 @@ class AccountFormMail(forms.Form, DynamicForm): ) def __init__(self, *args, **kwargs): - if "instance" in kwargs: - self.mb = kwargs["instance"] - del kwargs["instance"] - else: - self.mb = None + self.mb = kwargs.pop("instance", None) super(AccountFormMail, self).__init__(*args, **kwargs) self.field_widths = { "quota": 3 @@ -245,6 +242,13 @@ class AccountFormMail(forms.Form, DynamicForm): self.domain = Domain.objects.get(name=domname) except Domain.DoesNotExist: raise forms.ValidationError(_("Domain does not exist")) + if not self.mb: + try: + core_signals.can_create_object.send( + sender=self.__class__, context=self.domain, + object_type="mailboxes") + except lib_exceptions.ModoboaException as inst: + raise forms.ValidationError(inst) return email def clean(self): @@ -262,9 +266,9 @@ class AccountFormMail(forms.Form, DynamicForm): def create_mailbox(self, user, account): """Create a mailbox associated to :kw:`account`.""" if not user.can_access(self.domain): - raise PermDeniedException + raise lib_exceptions.PermDeniedException core_signals.can_create_object.send( - self.__class__, user=user, object_type="mailboxes") + self.__class__, context=user, object_type="mailboxes") self.mb = Mailbox( address=self.locpart, domain=self.domain, user=account, use_domain_quota=self.cleaned_data["quota_act"]) @@ -296,8 +300,11 @@ class AccountFormMail(forms.Form, DynamicForm): if not aliases: return core_signals.can_create_object.send( - self.__class__, user=user, object_type="mailbox_aliases", + self.__class__, context=user, object_type="mailbox_aliases", count=len(aliases)) + core_signals.can_create_object.send( + self.__class__, context=self.mb.domain, + object_type="mailbox_aliases", count=len(aliases)) for alias in aliases: if self.mb.aliasrecipient_set.select_related("alias").filter( alias__address=alias).exists(): diff --git a/modoboa/admin/forms/alias.py b/modoboa/admin/forms/alias.py index 76fa537ed..8195d5613 100644 --- a/modoboa/admin/forms/alias.py +++ b/modoboa/admin/forms/alias.py @@ -6,6 +6,8 @@ from django import forms from django.http import QueryDict from django.utils.translation import ugettext as _, ugettext_lazy +from modoboa.core import signals as core_signals +from modoboa.lib import exceptions as lib_exceptions from modoboa.lib.email_utils import split_mailbox from modoboa.lib import fields as lib_fields from modoboa.lib.form_utils import ( @@ -16,7 +18,6 @@ from ..models import Domain, Alias class AliasForm(forms.ModelForm, DynamicForm): - """A form to create/modify an alias.""" address = lib_fields.UTF8AndEmptyUserEmailField( @@ -74,6 +75,13 @@ class AliasForm(forms.ModelForm, DynamicForm): raise forms.ValidationError( _("You don't have access to this domain") ) + if not self.instance.pk: + try: + core_signals.can_create_object.send( + sender=self.__class__, context=domain, + object_type="mailbox_aliases") + except lib_exceptions.ModoboaException as inst: + raise forms.ValidationError(inst) return self.cleaned_data["address"].lower() def clean(self): diff --git a/modoboa/admin/forms/domain.py b/modoboa/admin/forms/domain.py index 636923d77..52e4b286a 100644 --- a/modoboa/admin/forms/domain.py +++ b/modoboa/admin/forms/domain.py @@ -101,6 +101,7 @@ class DomainFormGeneral(forms.ModelForm, DynamicForm): "name", _("A %s with this name already exists") % unicode(label) ) + self.aliases = [] for k in cleaned_data.keys(): if not k.startswith("aliases"): continue @@ -120,6 +121,8 @@ class DomainFormGeneral(forms.ModelForm, DynamicForm): k, _("A %s with this name already exists") % unicode(label) ) + else: + self.aliases.append(cleaned_data[k]) return cleaned_data def save(self, user, commit=True, domalias_post_create=False): @@ -134,28 +137,20 @@ class DomainFormGeneral(forms.ModelForm, DynamicForm): if not commit: return d d.save() - aliases = [] - for k, v in self.cleaned_data.iteritems(): - if not k.startswith("aliases"): - continue - if v in ["", None]: - continue - aliases.append(v) for dalias in d.domainalias_set.all(): - if dalias.name not in aliases: + if dalias.name not in self.aliases: dalias.delete() else: - aliases.remove(dalias.name) - if aliases: + self.aliases.remove(dalias.name) + if self.aliases: core_signals.can_create_object.send( - self.__class__, user=user, object_type="domain_aliases", - count=len(aliases)) - for alias in aliases: - try: - d.domainalias_set.get(name=alias) - except DomainAlias.DoesNotExist: - pass - else: + self.__class__, context=user, object_type="domain_aliases", + count=len(self.aliases)) + core_signals.can_create_object.send( + self.__class__, context=d, object_type="domain_aliases", + count=len(self.aliases)) + for alias in self.aliases: + if d.domainalias_set.filter(name=alias).exists(): continue options = {"creator": user} if domalias_post_create else {} DomainAlias(name=alias, target=d, enabled=d.enabled).save( @@ -225,9 +220,9 @@ class DomainFormOptions(forms.Form): except User.DoesNotExist: pass else: - raise Conflict(_("User '%s' already exists" % username)) + raise Conflict(_("User '%s' already exists") % username) core_signals.can_create_object.send( - self.__class__, user=user, object_type="mailboxes") + self.__class__, context=user, object_type="mailboxes") da = User(username=username, email=username, is_active=True) da.set_password("password") da.save() @@ -245,7 +240,7 @@ class DomainFormOptions(forms.Form): if domain.type == "domain" and \ self.cleaned_data["create_aliases"] == "yes": core_signals.can_create_object.send( - self.__class__, user=user, object_type="mailbox_aliases") + self.__class__, context=user, object_type="mailbox_aliases") alias = Alias( address=u"postmaster@{}".format(domain.name), domain=domain, enabled=True diff --git a/modoboa/admin/lib.py b/modoboa/admin/lib.py index 547d487ca..075663131 100644 --- a/modoboa/admin/lib.py +++ b/modoboa/admin/lib.py @@ -10,6 +10,7 @@ from django.db.models import Q from django.utils.translation import ugettext as _ from modoboa.core.models import User +from modoboa.core import signals as core_signals from modoboa.lib import events from modoboa.lib.exceptions import PermDeniedException @@ -118,6 +119,8 @@ def import_domain(user, row, formopts): """Specific code for domains import""" if not user.has_perm("admin.add_domain"): raise PermDeniedException(_("You are not allowed to import domains")) + core_signals.can_create_object.send( + sender="import", context=user, object_type="domains") dom = Domain() dom.from_csv(user, row) @@ -127,6 +130,8 @@ def import_domainalias(user, row, formopts): if not user.has_perm("admin.add_domainalias"): raise PermDeniedException( _("You are not allowed to import domain aliases.")) + core_signals.can_create_object.send( + sender="import", context=user, object_type="domain_aliases") domalias = DomainAlias() domalias.from_csv(user, row) diff --git a/modoboa/admin/models/alias.py b/modoboa/admin/models/alias.py index 3948966b7..c8626267c 100644 --- a/modoboa/admin/models/alias.py +++ b/modoboa/admin/models/alias.py @@ -6,6 +6,7 @@ from django.utils.translation import ugettext as _, ugettext_lazy import reversion +from modoboa.core import signals as core_signals from modoboa.lib.email_utils import split_mailbox from modoboa.lib.exceptions import ( PermDeniedException, BadRequest, Conflict, NotFound @@ -171,6 +172,10 @@ class Alias(AdminObject): raise BadRequest(_("Domain '%s' does not exist" % domname)) if not user.can_access(domain): raise PermDeniedException + core_signals.can_create_object.send( + sender="import", context=user, object_type="mailbox_aliases") + core_signals.can_create_object.send( + sender="import", context=domain, object_type="mailbox_aliases") if Alias.objects.filter(address=address).exists(): raise Conflict self.address = address diff --git a/modoboa/admin/models/domain.py b/modoboa/admin/models/domain.py index 65267a6a9..bf5197adf 100644 --- a/modoboa/admin/models/domain.py +++ b/modoboa/admin/models/domain.py @@ -8,11 +8,13 @@ from django.utils.translation import ugettext as _, ugettext_lazy import reversion -from .base import AdminObject from modoboa.core.models import User, ObjectAccess +from modoboa.core import signals as core_signals from modoboa.lib import events, parameters from modoboa.lib.exceptions import BadRequest, Conflict +from .base import AdminObject + class DomainManager(Manager): @@ -85,22 +87,28 @@ class Domain(AdminObject): @property def admins(self): - """Return the domain administrators of this domain + """Return the domain administrators of this domain. :return: a list of User objects """ - return [oa.user for oa in self.owners.filter(user__is_superuser=False)] + return User.objects.filter( + is_superuser=False, + objectaccess__content_type__model="domain", + objectaccess__object_id=self.pk) @property def aliases(self): return self.domainalias_set def add_admin(self, account): - """Add a new administrator for this domain + """Add a new administrator to this domain. - :param User account: the administrotor to add + :param User account: the administrator """ from modoboa.lib.permissions import grant_access_to_object + + core_signals.can_create_object.send( + sender=self.__class__, context=self, object_type="domain_admins") grant_access_to_object(account, self) for mb in self.mailbox_set.all(): if mb.user.has_perm("admin.add_domain"): diff --git a/modoboa/admin/models/domain_alias.py b/modoboa/admin/models/domain_alias.py index 52bed916d..1acf2ff92 100644 --- a/modoboa/admin/models/domain_alias.py +++ b/modoboa/admin/models/domain_alias.py @@ -8,10 +8,12 @@ from django.contrib.contenttypes import generic import reversion +from modoboa.core import models as core_models +from modoboa.core import signals as core_signals +from modoboa.lib.exceptions import BadRequest, Conflict + from .base import AdminObject from .domain import Domain -from modoboa.core import models as core_models -from modoboa.lib.exceptions import BadRequest, Conflict class DomainAliasManager(models.Manager): @@ -76,6 +78,8 @@ class DomainAlias(AdminObject): self.target = Domain.objects.get(name=domname) except Domain.DoesNotExist: raise BadRequest(_("Unknown domain %s") % domname) + core_signals.can_create_object.send( + sender="import", context=self.target, object_type="domain_aliases") self.enabled = row[3].strip() in ["True", "1", "yes", "y"] self.save(creator=user) diff --git a/modoboa/admin/serializers.py b/modoboa/admin/serializers.py index b61dc9acd..d7f12c966 100644 --- a/modoboa/admin/serializers.py +++ b/modoboa/admin/serializers.py @@ -9,6 +9,8 @@ from passwords import validators from modoboa.admin import models as admin_models from modoboa.core import models as core_models +from modoboa.core import signals as core_signals +from modoboa.lib import exceptions as lib_exceptions from modoboa.lib import permissions, email_utils, fields as lib_fields from . import models @@ -42,6 +44,23 @@ class DomainAliasSerializer(serializers.ModelSerializer): raise serializers.ValidationError(_("Permission denied.")) return value + def create(self, validated_data): + """Custom creation.""" + domain_alias = models.DomainAlias(**validated_data) + creator = self.context["request"].user + try: + core_signals.can_create_object.send( + sender=self.__class__, context=creator, + object_type="domain_aliases") + core_signals.can_create_object.send( + sender=self.__class__, context=domain_alias.target, + object_type="domain_aliases") + except lib_exceptions.ModoboaException as inst: + raise serializers.ValidationError({ + "domain": unicode(inst)}) + domain_alias.save(creator=creator) + return domain_alias + class MailboxSerializer(serializers.ModelSerializer): """Base mailbox serializer.""" @@ -175,6 +194,16 @@ class WritableAccountSerializer(AccountSerializer): if not creator.can_access(domain): raise serializers.ValidationError({ "domain": _("Permission denied.")}) + try: + core_signals.can_create_object.send( + sender=self.__class__, context=creator, + object_type="mailboxes") + core_signals.can_create_object.send( + sender=self.__class__, context=domain, + object_type="mailboxes") + except lib_exceptions.ModoboaException as inst: + raise serializers.ValidationError({ + "domain": unicode(inst)}) mb = admin_models.Mailbox( user=account, address=address, domain=domain, **data) mb.set_quota( @@ -241,6 +270,15 @@ class AliasSerializer(serializers.ModelSerializer): def create(self, validated_data): """Create appropriate objects.""" creator = self.context["request"].user + try: + core_signals.can_create_object.send( + sender=self.__class__, context=creator, + object_type="mailbox_aliases") + core_signals.can_create_object.send( + sender=self.__class__, context=self.domain, + object_type="mailbox_aliases") + except lib_exceptions.ModoboaException as inst: + raise serializers.ValidationError(unicode(inst)) recipients = validated_data.pop("recipients", None) alias = admin_models.Alias(domain=self.domain, **validated_data) alias.save(creator=creator) diff --git a/modoboa/admin/tests/test_account.py b/modoboa/admin/tests/test_account.py index 522899025..4ecd3ae5e 100644 --- a/modoboa/admin/tests/test_account.py +++ b/modoboa/admin/tests/test_account.py @@ -161,9 +161,9 @@ class PermissionsTestCase(ModoTestCase): from modoboa.limits import utils as limits_utils super(PermissionsTestCase, cls).setUpTestData() - for name, tpl in limits_utils.get_limit_templates(): + for name, tpl in limits_utils.get_user_limit_templates(): parameters.save_admin( - "DEFLT_{0}_LIMIT".format(name.upper()), 2, + "DEFLT_USER_{0}_LIMIT".format(name.upper()), 2, app="limits" ) factories.populate_database() diff --git a/modoboa/admin/tests/test_api.py b/modoboa/admin/tests/test_api.py index 144073442..58204920e 100644 --- a/modoboa/admin/tests/test_api.py +++ b/modoboa/admin/tests/test_api.py @@ -9,6 +9,7 @@ from django.core.urlresolvers import reverse from rest_framework.authtoken.models import Token from modoboa.core import models as core_models +from modoboa.lib import parameters from modoboa.lib.tests import ModoAPITestCase from .. import factories @@ -401,6 +402,7 @@ class AliasAPITestCase(ModoAPITestCase): def setUpTestData(cls): """Create test data.""" super(AliasAPITestCase, cls).setUpTestData() + parameters.save_admin("ENABLE_ADMIN_LIMITS", "no", app="limits") factories.populate_database() cls.da_token = Token.objects.create( user=core_models.User.objects.get(username="admin@test.com")) diff --git a/modoboa/admin/tests/test_import_.py b/modoboa/admin/tests/test_import_.py index b917959ad..8e0610f60 100644 --- a/modoboa/admin/tests/test_import_.py +++ b/modoboa/admin/tests/test_import_.py @@ -3,6 +3,7 @@ from django.core.files.base import ContentFile from django.core.urlresolvers import reverse from modoboa.core.models import User +from modoboa.lib import parameters from modoboa.lib.tests import ModoTestCase from .. import factories @@ -15,6 +16,7 @@ class ImportTestCase(ModoTestCase): def setUpTestData(cls): """Create test data.""" super(ImportTestCase, cls).setUpTestData() + parameters.save_admin("ENABLE_ADMIN_LIMITS", "no", app="limits") factories.populate_database() def test_domains_import(self): diff --git a/modoboa/admin/views/alias.py b/modoboa/admin/views/alias.py index a40bd29cf..ef771b8f1 100644 --- a/modoboa/admin/views/alias.py +++ b/modoboa/admin/views/alias.py @@ -37,7 +37,7 @@ def _validate_alias(request, form, successmsg, callback=None): def _new_alias(request, title, action, successmsg, tplname="admin/aliasform.html"): core_signals.can_create_object.send( - "new_alias", user=request.user, object_type="mailbox_aliases") + "new_alias", context=request.user, object_type="mailbox_aliases") if request.method == "POST": def callback(user, alias): alias.post_create(user) diff --git a/modoboa/admin/views/domain.py b/modoboa/admin/views/domain.py index 0d222542f..2e3f57289 100644 --- a/modoboa/admin/views/domain.py +++ b/modoboa/admin/views/domain.py @@ -100,7 +100,7 @@ def domains_list(request): @reversion.create_revision() def newdomain(request): core_signals.can_create_object.send( - "newdomain", user=request.user, object_type="domains") + "newdomain", context=request.user, object_type="domains") return DomainWizard(request).process() diff --git a/modoboa/core/__init__.py b/modoboa/core/__init__.py index adbda48b9..fd3f335b2 100644 --- a/modoboa/core/__init__.py +++ b/modoboa/core/__init__.py @@ -39,7 +39,10 @@ PERMISSIONS = { ["admin", "domain", "view_domains"], ["admin", "domain", "add_domain"], ["admin", "domain", "change_domain"], - ["admin", "domain", "delete_domain"] + ["admin", "domain", "delete_domain"], + ["admin", "domainalias", "add_domainalias"], + ["admin", "domainalias", "change_domainalias"], + ["admin", "domainalias", "delete_domainalias"], ] } diff --git a/modoboa/core/serializers.py b/modoboa/core/serializers.py deleted file mode 100644 index e81354c30..000000000 --- a/modoboa/core/serializers.py +++ /dev/null @@ -1,41 +0,0 @@ -"""Modoboa core serializers.""" - -from django.core.exceptions import ValidationError - -from rest_framework import serializers -from passwords import validators - -from . import models - - -class UserPasswordSerializer(serializers.ModelSerializer): - - """A serializer used to change a user password.""" - - new_password = serializers.CharField() - - class Meta: - model = models.User - fields = ( - "password", "new_password", ) - - def validate_password(self, value): - """Check password.""" - if not self.instance.check_password(value): - raise serializers.ValidationError("Password not correct") - return value - - def validate_new_password(self, value): - """Check new password.""" - try: - validators.validate_length(value) - validators.complexity(value) - except ValidationError as exc: - raise serializers.ValidationError(exc.messages[0]) - return value - - def update(self, instance, validated_data): - """Set new password.""" - instance.set_password(validated_data["new_password"]) - instance.save() - return instance diff --git a/modoboa/core/signals.py b/modoboa/core/signals.py index 1171e3747..1275ffe1b 100644 --- a/modoboa/core/signals.py +++ b/modoboa/core/signals.py @@ -3,4 +3,4 @@ import django.dispatch can_create_object = django.dispatch.Signal( - providing_args=["user", "object_type", "count"]) + providing_args=["context", "object_type", "count"]) diff --git a/modoboa/lib/migrations/0005_auto_20160416_1449.py b/modoboa/lib/migrations/0005_auto_20160416_1449.py new file mode 100644 index 000000000..6c594dec3 --- /dev/null +++ b/modoboa/lib/migrations/0005_auto_20160416_1449.py @@ -0,0 +1,23 @@ +# -*- coding: utf-8 -*- +from __future__ import unicode_literals + +from django.db import migrations + + +def rename_limits_parameters(apps, schema_editor): + """Rename parameters.""" + Parameter = apps.get_model("lib", "Parameter") + for param in Parameter.objects.filter(name__startswith="DEFLT_"): + param.name = param.name.replace("DEFLT_", "DEFLT_USER_") + param.save() + + +class Migration(migrations.Migration): + + dependencies = [ + ('lib', '0004_auto_20151114_1409'), + ] + + operations = [ + migrations.RunPython(rename_limits_parameters) + ] diff --git a/modoboa/limits/app_settings.py b/modoboa/limits/app_settings.py index 8c2129ccb..ff729a668 100644 --- a/modoboa/limits/app_settings.py +++ b/modoboa/limits/app_settings.py @@ -3,16 +3,24 @@ from django import forms from django.utils.translation import ugettext_lazy as _ -from modoboa.lib.form_utils import SeparatorField +from modoboa.lib.form_utils import SeparatorField, YesNoField from modoboa.lib.parameters import AdminParametersForm class ParametersForm(AdminParametersForm): + """Available admin parameters.""" + app = "limits" - defv_sep = SeparatorField(label=_("Default limits")) + defv_sep = SeparatorField(label=_("Default per-admin limits")) - deflt_domain_admins_limit = forms.IntegerField( + enable_admin_limits = YesNoField( + label=_("Enable per-admin limits"), + initial="yes", + help_text=_("Enable or disable per-admin limits") + ) + + deflt_user_domain_admins_limit = forms.IntegerField( label=_("Domain admins"), initial=0, help_text=_( @@ -23,7 +31,7 @@ class ParametersForm(AdminParametersForm): widget=forms.widgets.TextInput( attrs={"class": "col-md-1 form-control"}) ) - deflt_domains_limit = forms.IntegerField( + deflt_user_domains_limit = forms.IntegerField( label=_("Domains"), initial=0, help_text=_( @@ -33,7 +41,7 @@ class ParametersForm(AdminParametersForm): widget=forms.widgets.TextInput( attrs={"class": "col-md-1 form-control"}) ) - deflt_domain_aliases_limit = forms.IntegerField( + deflt_user_domain_aliases_limit = forms.IntegerField( label=_("Domain aliases"), initial=0, help_text=_( @@ -44,7 +52,7 @@ class ParametersForm(AdminParametersForm): widget=forms.widgets.TextInput( attrs={"class": "col-md-1 form-control"}) ) - deflt_mailboxes_limit = forms.IntegerField( + deflt_user_mailboxes_limit = forms.IntegerField( label=_("Mailboxes"), initial=0, help_text=_( @@ -54,7 +62,7 @@ class ParametersForm(AdminParametersForm): widget=forms.widgets.TextInput( attrs={"class": "col-md-1 form-control"}) ) - deflt_mailbox_aliases_limit = forms.IntegerField( + deflt_user_mailbox_aliases_limit = forms.IntegerField( label=_("Mailbox aliases"), initial=0, help_text=_( @@ -65,6 +73,69 @@ class ParametersForm(AdminParametersForm): attrs={"class": "col-md-1 form-control"}) ) + domain_limits_sep = SeparatorField(label=_("Default per-domain limits")) + + enable_domain_limits = YesNoField( + label=_("Enable per-domain limits"), + initial="no", + help_text=_("Enable or disable per-domain limits") + ) + + deflt_domain_domain_admins_limit = forms.IntegerField( + label=_("Domain admins"), + initial=0, + help_text=_( + "Maximum number of allowed domain administrators for a new " + "domain. (0 to deny any creation, -1 to allow unlimited " + "creations)" + ), + widget=forms.widgets.TextInput( + attrs={"class": "col-md-1 form-control"}) + ) + deflt_domain_domain_aliases_limit = forms.IntegerField( + label=_("Domain aliases"), + initial=0, + help_text=_( + "Maximum number of allowed domain aliases for a new " + "domain. (0 to deny any creation, -1 to allow " + "unlimited creations)" + ), + widget=forms.widgets.TextInput( + attrs={"class": "col-md-1 form-control"}) + ) + deflt_domain_mailboxes_limit = forms.IntegerField( + label=_("Mailboxes"), + initial=0, + help_text=_( + "Maximum number of allowed mailboxes for a new domain. " + "(0 to deny any creation, -1 to allow unlimited creations)" + ), + widget=forms.widgets.TextInput( + attrs={"class": "col-md-1 form-control"}) + ) + deflt_domain_mailbox_aliases_limit = forms.IntegerField( + label=_("Mailbox aliases"), + initial=0, + help_text=_( + "Maximum number of allowed aliases for a new domain. " + "(0 to deny any creation, -1 to allow unlimited creations)" + ), + widget=forms.widgets.TextInput( + attrs={"class": "col-md-1 form-control"}) + ) + + visibility_rules = { + "deflt_user_domains_limit": "enable_admin_limits=yes", + "deflt_user_domain_aliases_limit": "enable_admin_limits=yes", + "deflt_user_mailboxes_limit": "enable_admin_limits=yes", + "deflt_user_mailbox_aliases_limit": "enable_admin_limits=yes", + "deflt_user_domain_admins_limit": "enable_admin_limits=yes", + "deflt_domain_mailboxes_limit": "enable_domain_limits=yes", + "deflt_domain_mailbox_aliases_limit": "enable_domain_limits=yes", + "deflt_domain_domain_aliases_limit": "enable_domain_limits=yes", + "deflt_domain_domain_admins_limit": "enable_domain_limits=yes", + } + def __init__(self, *args, **kwargs): super(AdminParametersForm, self).__init__(*args, **kwargs) self._load_extra_parameters('A') diff --git a/modoboa/limits/constants.py b/modoboa/limits/constants.py index 947adbe62..7bdb3f38d 100644 --- a/modoboa/limits/constants.py +++ b/modoboa/limits/constants.py @@ -5,7 +5,7 @@ import collections from django.utils.translation import ugettext_lazy as _ -DEFAULT_LIMITS = collections.OrderedDict(( +DEFAULT_USER_LIMITS = collections.OrderedDict(( ("domains", { "content_type": "admin.domain", "label": _("Domains"), "help": _("Maximum number of domains this user can create"), @@ -28,3 +28,24 @@ DEFAULT_LIMITS = collections.OrderedDict(( "required_role": "Resellers", "extra_filters": {"groups__name": "DomainAdmins"}}) )) + + +DEFAULT_DOMAIN_LIMITS = collections.OrderedDict(( + ("domain_aliases", { + "relation": "domainalias_set", "label": _("Domain aliases"), + "help": _("Maximum number of domain aliases allowed for this domain.") + }), + ("mailboxes", { + "relation": "mailbox_set", "label": _("Mailboxes"), + "help": _("Maximum number of mailboxes allowed for this domain.")}), + ("mailbox_aliases", { + "relation": "alias_set", "label": _("Mailbox aliases"), + "help": _( + "Maximum number of mailbox aliases allowed for this domain."), + "extra_filters": {"internal": False} + }), + ("domain_admins", { + "relation": "admins", "label": _("Domain admins"), + "help": _("Maximum number of domain admins allowed for this domain."), + }) +)) diff --git a/modoboa/limits/controls.py b/modoboa/limits/controls.py index 4d0c20c35..fa914072f 100644 --- a/modoboa/limits/controls.py +++ b/modoboa/limits/controls.py @@ -5,7 +5,7 @@ """ -from modoboa.lib import events +from modoboa.lib import events, parameters from modoboa.lib.permissions import get_object_owner from . import utils @@ -18,11 +18,11 @@ def move_pool_resource(owner, user): associated resource to the pool of its owner. """ if not owner.is_superuser: - for name, ltpl in utils.get_limit_templates(): - l = user.objectlimit_set.get(name=name) + for name, ltpl in utils.get_user_limit_templates(): + l = user.userobjectlimit_set.get(name=name) if l.max_value < 0: continue - ol = owner.objectlimit_set.get(name=name) + ol = owner.userobjectlimit_set.get(name=name) ol.max_value += l.max_value ol.save() @@ -39,12 +39,15 @@ def user_can_set_role(user, role, account=None): :param ``User`` account: account modified (None on creation) :param str newrole: role to check """ - if role != "DomainAdmins": + condition = ( + parameters.get_admin("ENABLE_ADMIN_LIMITS") == "no" or + role != "DomainAdmins") + if condition: return [True] lname = "domain_admins" condition = ( user.is_superuser or - not user.objectlimit_set.get(name=lname).is_exceeded() + not user.userobjectlimit_set.get(name=lname).is_exceeded() ) if condition: return [True] diff --git a/modoboa/limits/forms.py b/modoboa/limits/forms.py index 8d3285cf2..0af239697 100644 --- a/modoboa/limits/forms.py +++ b/modoboa/limits/forms.py @@ -5,7 +5,6 @@ from django import forms from django.utils.translation import ugettext as _ -from modoboa.lib import parameters from modoboa.lib.permissions import get_object_owner from .lib import BadLimitValue, UnsufficientResource @@ -16,12 +15,9 @@ class ResourcePoolForm(forms.Form): """Dynamic form to display user limits.""" def __init__(self, *args, **kwargs): - self.account = None - if "instance" in kwargs: - self.account = kwargs["instance"] - del kwargs["instance"] + self.account = kwargs.pop("instance", None) super(ResourcePoolForm, self).__init__(*args, **kwargs) - for name, tpl in utils.get_limit_templates(): + for name, tpl in utils.get_user_limit_templates(): if "required_role" in tpl: condition = ( self.account is not None and @@ -32,7 +28,7 @@ class ResourcePoolForm(forms.Form): self.fields["{}_limit".format(name)] = forms.IntegerField( label=tpl["label"], help_text=tpl["help"] ) - if hasattr(self, "account"): + if self.account: self.load_from_user(self.account) def clean(self): @@ -50,7 +46,7 @@ class ResourcePoolForm(forms.Form): for fieldname in self.fields.keys(): lname = fieldname.replace("_limit", "") self.fields[fieldname].initial = ( - user.objectlimit_set.get(name=lname).max_value) + user.userobjectlimit_set.get(name=lname).max_value) def allocate_from_user(self, limit, user): """Allocate resource using an existing user. @@ -60,7 +56,7 @@ class ResourcePoolForm(forms.Form): also be limited, the resource he gives is taken from its own pool. """ - ol = user.objectlimit_set.get(name=limit.name) + ol = user.userobjectlimit_set.get(name=limit.name) fieldname = "{}_limit".format(limit.name) newvalue = self.cleaned_data[fieldname] if newvalue == -1 and ol.max_value != -1: @@ -80,12 +76,51 @@ class ResourcePoolForm(forms.Form): def save(self): owner = get_object_owner(self.account) - for name, ltpl in utils.get_limit_templates(): + for name, ltpl in utils.get_user_limit_templates(): fieldname = "{}_limit".format(name) if fieldname not in self.cleaned_data: continue - l = self.account.objectlimit_set.get(name=name) + l = self.account.userobjectlimit_set.get(name=name) if not owner.is_superuser: self.allocate_from_user(l, owner) l.max_value = self.cleaned_data[fieldname] l.save() + + +class DomainLimitsForm(forms.Form): + """Per-domain limits form.""" + + def __init__(self, *args, **kwargs): + """Define limits as fields.""" + self.domain = kwargs.pop("instance", None) + super(DomainLimitsForm, self).__init__(*args, **kwargs) + for name, tpl in utils.get_domain_limit_templates(): + self.fields["{}_limit".format(name)] = forms.IntegerField( + label=tpl["label"], help_text=tpl["help"] + ) + if not self.domain: + return + for fieldname in self.fields.keys(): + lname = fieldname.replace("_limit", "") + self.fields[fieldname].initial = ( + self.domain.domainobjectlimit_set.get(name=lname).max_value) + + def clean(self): + """Ensure limit values are correct.""" + cleaned_data = super(DomainLimitsForm, self).clean() + if self.errors: + return cleaned_data + for lname in self.fields.keys(): + if cleaned_data[lname] < -1: + self.add_error(lname, _("Invalid limit")) + return cleaned_data + + def save(self, user): + """Set limits.""" + for name, ltpl in utils.get_domain_limit_templates(): + fieldname = "{}_limit".format(name) + if fieldname not in self.cleaned_data: + continue + l = self.domain.domainobjectlimit_set.get(name=name) + l.max_value = self.cleaned_data[fieldname] + l.save() diff --git a/modoboa/limits/general_callbacks.py b/modoboa/limits/general_callbacks.py index 655497c82..3d7ab6780 100644 --- a/modoboa/limits/general_callbacks.py +++ b/modoboa/limits/general_callbacks.py @@ -2,9 +2,9 @@ from django.utils.translation import ugettext as _ -from modoboa.lib import events +from modoboa.lib import events, parameters -from .forms import ResourcePoolForm +from . import forms from . import utils @@ -12,7 +12,10 @@ from . import utils def display_pool_usage(user, target, currentpage): from django.template.loader import render_to_string - if target != "leftcol" or user.is_superuser: + condition = ( + parameters.get_admin("ENABLE_ADMIN_LIMITS") == "no" or + target != "leftcol" or user.is_superuser) + if condition: return [] if currentpage == "identities": names = ["mailboxes", "mailbox_aliases"] @@ -21,13 +24,13 @@ def display_pool_usage(user, target, currentpage): else: exceptions = ["domain_admins", "mailboxes", "mailbox_aliases"] names = [ - name for name, tpl in utils.get_limit_templates() + name for name, tpl in utils.get_user_limit_templates() if name not in exceptions and ("required_role" not in tpl or tpl["required_role"] == user.group) ] - limits = user.objectlimit_set.filter(name__in=names, max_value__gt=0) + limits = user.userobjectlimit_set.filter(name__in=names, max_value__gt=0) if len(limits) == 0: return [] return [ @@ -38,17 +41,41 @@ def display_pool_usage(user, target, currentpage): @events.observe("ExtraAccountForm") def extra_account_form(user, account=None): + if parameters.get_admin("ENABLE_ADMIN_LIMITS") == "no": + return [] if user.group not in ["SuperAdmins", "Resellers"]: return [] if account is not None and \ account.group not in ["Resellers", "DomainAdmins"]: return [] - return [ - dict( - id="resources", title=_("Resources"), cls=ResourcePoolForm - ) - ] + return [{ + "id": "resources", "title": _("Resources"), + "cls": forms.ResourcePoolForm + }] + + +@events.observe("ExtraDomainForm") +def extra_domain_form(user, domain): + """Include domain limits form.""" + if parameters.get_admin("ENABLE_DOMAIN_LIMITS") == "no": + return [] + if not user.has_perm("admin.change_domain"): + return [] + return [{ + "id": "resources", "title": _("Resources"), + "cls": forms.DomainLimitsForm + }] + + +@events.observe("FillDomainInstances") +def fill_domain_instances(user, domain, instances): + """Set domain instance for resources form.""" + if parameters.get_admin("ENABLE_DOMAIN_LIMITS") == "no": + return + if not user.has_perm("admin.change_domain"): + return + instances["resources"] = domain @events.observe("CheckExtraAccountForm") @@ -63,16 +90,21 @@ def check_form_access(account, form): @events.observe("FillAccountInstances") def fill_account_instances(user, account, instances): - if not user.is_superuser and not user.belongs_to_group("Resellers"): + condition = ( + parameters.get_admin("ENABLE_ADMIN_LIMITS") == "no" or + (not user.is_superuser and user.group != "Resellers") + ) + if condition: return - if not account.belongs_to_group("Resellers") and \ - not account.belongs_to_group("DomainAdmins"): + if account.group not in ["Resellers", "DomainAdmins"]: return instances["resources"] = account @events.observe("GetStaticContent") def get_static_content(caller, st_type, user): + if parameters.get_admin("ENABLE_ADMIN_LIMITS") == "no": + return [] if caller not in ['domains', 'identities']: return [] if user.group == "SimpleUsers": diff --git a/modoboa/limits/handlers.py b/modoboa/limits/handlers.py index 834c36bab..fc847dc28 100644 --- a/modoboa/limits/handlers.py +++ b/modoboa/limits/handlers.py @@ -5,6 +5,7 @@ from django.dispatch import receiver from django.contrib.contenttypes.models import ContentType +from modoboa.admin import models as admin_models from modoboa.core import models as core_models from modoboa.core import signals as core_signals from modoboa.lib import signals as lib_signals @@ -16,11 +17,20 @@ from . import utils @receiver(core_signals.can_create_object) -def check_object_limit(sender, user, object_type, **kwargs): +def check_object_limit(sender, context, object_type, **kwargs): """Check if user can create a new object.""" - if user.is_superuser: - return True - limit = user.objectlimit_set.get(name=object_type) + if context.__class__.__name__ == "User": + if parameters.get_admin("ENABLE_ADMIN_LIMITS") == "no": + return + if context.is_superuser: + return True + limit = context.userobjectlimit_set.get(name=object_type) + elif context.__class__.__name__ == "Domain": + if parameters.get_admin("ENABLE_DOMAIN_LIMITS") == "no": + return + limit = context.domainobjectlimit_set.get(name=object_type) + else: + raise NotImplementedError count = kwargs.get("count", 1) if limit.is_exceeded(count): raise lib.LimitReached(limit) @@ -33,13 +43,26 @@ def create_user_limits(sender, instance, **kwargs): return request = lib_signals.get_request() creator = request.user if request else None - for name, definition in utils.get_limit_templates(): + for name, definition in utils.get_user_limit_templates(): ct = ContentType.objects.get_by_natural_key( *definition["content_type"].split(".")) max_value = 0 # creator can be None if user was created by a factory if not creator or creator.is_superuser: max_value = int( - parameters.get_admin("DEFLT_{}_LIMIT".format(name.upper()))) - models.ObjectLimit.objects.create( + parameters.get_admin( + "DEFLT_USER_{}_LIMIT".format(name.upper()))) + models.UserObjectLimit.objects.create( user=instance, name=name, content_type=ct, max_value=max_value) + + +@receiver(signals.post_save, sender=admin_models.Domain) +def create_domain_limits(sender, instance, **kwargs): + """Create limits for new domain.""" + if not kwargs.get("created"): + return + for name, definition in utils.get_domain_limit_templates(): + max_value = int( + parameters.get_admin("DEFLT_DOMAIN_{}_LIMIT".format(name.upper()))) + models.DomainObjectLimit.objects.create( + domain=instance, name=name, max_value=max_value) diff --git a/modoboa/limits/migrations/0003_auto_20160413_1046.py b/modoboa/limits/migrations/0003_auto_20160413_1046.py index 5a5fa13d7..438479931 100644 --- a/modoboa/limits/migrations/0003_auto_20160413_1046.py +++ b/modoboa/limits/migrations/0003_auto_20160413_1046.py @@ -8,7 +8,7 @@ from django.conf import settings def move_limits_to_user(apps, schema_editor): """Move limits from Pool to User.""" LimitsPool = apps.get_model("limits", "LimitsPool") - ObjectLimit = apps.get_model("limits", "ObjectLimit") + UserObjectLimit = apps.get_model("limits", "UserObjectLimit") ContentType = apps.get_model("contenttypes", "ContentType") if not ContentType.objects.exists(): @@ -34,12 +34,12 @@ def move_limits_to_user(apps, schema_editor): .prefetch_related("limit_set")) for pool in qset: for limit in pool.limit_set.all(): - objlimit = ObjectLimit( + objlimit = UserObjectLimit( user=pool.user, max_value=limit.maxvalue, name=limit.name.replace("_limit", ""), content_type=name_to_content_type[limit.name]) to_create.append(objlimit) - ObjectLimit.objects.bulk_create(to_create) + UserObjectLimit.objects.bulk_create(to_create) class Migration(migrations.Migration): @@ -53,7 +53,7 @@ class Migration(migrations.Migration): operations = [ migrations.CreateModel( - name='ObjectLimit', + name='UserObjectLimit', fields=[ ('id', models.AutoField(verbose_name='ID', serialize=False, auto_created=True, primary_key=True)), ('name', models.CharField(max_length=254)), @@ -63,7 +63,7 @@ class Migration(migrations.Migration): ], ), migrations.AlterUniqueTogether( - name='objectlimit', + name='userobjectlimit', unique_together=set([('user', 'name')]), ), migrations.RunPython(move_limits_to_user) diff --git a/modoboa/limits/migrations/0005_auto_20160415_1654.py b/modoboa/limits/migrations/0005_auto_20160415_1654.py new file mode 100644 index 000000000..ab97e7807 --- /dev/null +++ b/modoboa/limits/migrations/0005_auto_20160415_1654.py @@ -0,0 +1,50 @@ +# -*- coding: utf-8 -*- +from __future__ import unicode_literals + +from django.db import migrations, models +import modoboa.limits.models + +from .. import utils + + +def create_limits(apps, schema_editor): + """Create limits for existing domains.""" + Domain = apps.get_model("admin", "Domain") + DomainObjectLimit = apps.get_model("limits", "DomainObjectLimit") + to_create = [] + for domain in Domain.objects.all(): + for name, tpl in utils.get_domain_limit_templates(): + to_create.append( + DomainObjectLimit(domain=domain, name=name, max_value=-1)) + DomainObjectLimit.objects.bulk_create(to_create) + + +class Migration(migrations.Migration): + + dependencies = [ + ('admin', '0003_auto_20151118_1215'), + ('limits', '0004_auto_20160413_1312'), + ] + + operations = [ + migrations.CreateModel( + name='DomainObjectLimit', + fields=[ + ('id', models.AutoField(verbose_name='ID', serialize=False, auto_created=True, primary_key=True)), + ('name', models.CharField(max_length=254)), + ('max_value', models.IntegerField(default=0)), + ('domain', models.ForeignKey(to='admin.Domain')), + ], + bases=(modoboa.limits.models.ObjectLimitMixin, models.Model), + ), + migrations.AlterField( + model_name='userobjectlimit', + name='max_value', + field=models.IntegerField(default=0), + ), + migrations.AlterUniqueTogether( + name='domainobjectlimit', + unique_together=set([('domain', 'name')]), + ), + migrations.RunPython(create_limits) + ] diff --git a/modoboa/limits/models.py b/modoboa/limits/models.py index c97deadff..aca757b42 100644 --- a/modoboa/limits/models.py +++ b/modoboa/limits/models.py @@ -12,14 +12,42 @@ from modoboa.core import models as core_models from . import utils +class ObjectLimitMixin(object): + """Common methods.""" + + @property + def label(self): + """Return display name.""" + return self.definition["label"] + + @property + def usage(self): + """Return current limit usage in %.""" + if self.max_value < 0: + return -1 + if self.max_value == 0: + return 100 + return int(float(self.current_value) / self.max_value * 100) + + def is_exceeded(self, count=1): + """Check if limit will be reached if we add count object(s).""" + return self.current_value + 1 > self.max_value + + def __str__(self): + """Display current usage.""" + if self.max_value == -1: + return _("unlimited") + return "{}%".format(self.usage) + + @python_2_unicode_compatible -class ObjectLimit(models.Model): - """Per-user limits on object creation.""" +class UserObjectLimit(ObjectLimitMixin, models.Model): + """Object level limit.""" user = models.ForeignKey(settings.AUTH_USER_MODEL) name = models.CharField(max_length=254) content_type = models.ForeignKey("contenttypes.ContentType") - max_value = models.IntegerField(default=-2) + max_value = models.IntegerField(default=0) class Meta: unique_together = (("user", "name"), ) @@ -27,7 +55,7 @@ class ObjectLimit(models.Model): @property def definition(self): """Return the definition of this limit.""" - for name, tpl in utils.get_limit_templates(): + for name, tpl in utils.get_user_limit_templates(): if name == self.name: return tpl return None @@ -46,6 +74,11 @@ class ObjectLimit(models.Model): return model_class.objects.filter( pk__in=id_list, **self.definition["extra_filters"]).count() + @property + def label(self): + """Return display name.""" + return self.definition["label"] + @property def usage(self): """Return current limit usage in %.""" @@ -55,19 +88,42 @@ class ObjectLimit(models.Model): return 100 return int(float(self.current_value) / self.max_value * 100) - @property - def label(self): - """Return display name.""" - return self.definition["label"] - def is_exceeded(self, count=1): """Check if limit will be reached if we add count object(s).""" return self.current_value + 1 > self.max_value def __str__(self): """Display current usage.""" - if self.max_value == -2: - return _("undefined") if self.max_value == -1: return _("unlimited") return "{}%".format(self.usage) + + +class DomainObjectLimit(ObjectLimitMixin, models.Model): + """Per-domain limits on object creation.""" + + domain = models.ForeignKey("admin.Domain") + name = models.CharField(max_length=254) + max_value = models.IntegerField(default=0) + + class Meta: + unique_together = (("domain", "name"), ) + + @property + def definition(self): + """Return the definition of this limit.""" + for name, tpl in utils.get_domain_limit_templates(): + if name == self.name: + return tpl + return None + + @property + def current_value(self): + """Return the current number of objects.""" + definition = self.definition + if not definition: + raise RuntimeError("Bad limit {}".format(self.name)) + relation = getattr(self.domain, definition["relation"]) + if "extra_filters" in definition: + relation = relation.filter(**definition["extra_filters"]) + return relation.count() diff --git a/modoboa/limits/tests/__init__.py b/modoboa/limits/tests/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/modoboa/limits/tests/test_api.py b/modoboa/limits/tests/test_api.py new file mode 100644 index 000000000..5d7c99a7d --- /dev/null +++ b/modoboa/limits/tests/test_api.py @@ -0,0 +1,237 @@ +# coding: utf-8 +"""Test cases for the limits extension.""" + +from django.core.urlresolvers import reverse + +from rest_framework.authtoken.models import Token + +from modoboa.admin.factories import populate_database +from modoboa.admin.models import Domain +from modoboa.core import factories as core_factories +from modoboa.core.models import User +from modoboa.lib import parameters +from modoboa.lib import tests as lib_tests + +from .. import utils + + +class APIAdminLimitsTestCase(lib_tests.ModoAPITestCase): + """Check that limits are used also by the API.""" + + @classmethod + def setUpTestData(cls): + """Create test data.""" + super(APIAdminLimitsTestCase, cls).setUpTestData() + for name, tpl in utils.get_user_limit_templates(): + parameters.save_admin( + "DEFLT_USER_{}_LIMIT".format(name.upper()), 2) + populate_database() + cls.user = User.objects.get(username="admin@test.com") + cls.da_token = Token.objects.create(user=cls.user) + cls.reseller = core_factories.UserFactory( + username="reseller", groups=("Resellers", ), + ) + cls.r_token = Token.objects.create(user=cls.reseller) + + def test_domadmins_limit(self): + """Check domain admins limit.""" + self.client.credentials( + HTTP_AUTHORIZATION='Token ' + self.r_token.key) + + limit = self.reseller.userobjectlimit_set.get(name="domain_admins") + url = reverse("external_api:account-list") + data = { + "username": "fromapi@test.com", + "role": "DomainAdmins", + "password": "Toto1234", + } + response = self.client.post(url, data, format="json") + self.assertEqual(response.status_code, 201) + self.assertFalse(limit.is_exceeded()) + + data["username"] = "fromapi2@test.com" + response = self.client.post(url, data, format="json") + self.assertEqual(response.status_code, 201) + self.assertTrue(limit.is_exceeded()) + + data["username"] = "fromapi3@test.com" + response = self.client.post(url, data, format="json") + self.assertEqual(response.status_code, 400) + + user = User.objects.get(username="user@test.com") + domain = Domain.objects.get(name="test.com") + domain.add_admin(self.reseller) + url = reverse("external_api:account-detail", args=[user.pk]) + data = { + "username": user.username, + "role": "DomainAdmins", + "password": "Toto1234", + "mailbox": { + "full_address": user.username, + "quota": user.mailbox.quota + } + } + response = self.client.put(url, data, format="json") + self.assertEqual(response.status_code, 400) + + def test_domains_limit(self): + """Check domains limit.""" + self.client.credentials( + HTTP_AUTHORIZATION='Token ' + self.r_token.key) + limit = self.reseller.userobjectlimit_set.get(name="domains") + url = reverse("external_api:domain-list") + data = {"name": "test3.com", "quota": 10} + response = self.client.post(url, data, format="json") + self.assertEqual(response.status_code, 201) + self.assertFalse(limit.is_exceeded()) + + data["name"] = "test4.com" + response = self.client.post(url, data, format="json") + self.assertEqual(response.status_code, 201) + self.assertTrue(limit.is_exceeded()) + + data["username"] = "test5.com" + response = self.client.post(url, data, format="json") + self.assertEqual(response.status_code, 400) + + def test_domain_aliases_limit(self): + """Check domain aliases limit.""" + self.client.credentials( + HTTP_AUTHORIZATION='Token ' + self.r_token.key) + domain = Domain.objects.get(name="test.com") + domain.add_admin(self.reseller) + limit = self.reseller.userobjectlimit_set.get(name="domain_aliases") + url = reverse("external_api:domain_alias-list") + data = {"name": "dalias1.com", "target": domain.pk} + response = self.client.post(url, data, format="json") + self.assertEqual(response.status_code, 201) + self.assertFalse(limit.is_exceeded()) + + data["name"] = "dalias2.com" + response = self.client.post(url, data, format="json") + self.assertEqual(response.status_code, 201) + self.assertTrue(limit.is_exceeded()) + + data["username"] = "dalias3.com" + response = self.client.post(url, data, format="json") + self.assertEqual(response.status_code, 400) + + def test_mailboxes_limit(self): + """Check mailboxes limit.""" + self.client.credentials( + HTTP_AUTHORIZATION='Token ' + self.da_token.key) + + limit = self.user.userobjectlimit_set.get(name="mailboxes") + url = reverse("external_api:account-list") + data = { + "username": "fromapi@test.com", + "role": "SimpleUsers", + "password": "Toto1234", + "mailbox": { + "full_address": "fromapi@test.com", + "quota": 10 + } + } + response = self.client.post(url, data, format="json") + self.assertEqual(response.status_code, 201) + self.assertFalse(limit.is_exceeded()) + + data["username"] = "fromapi2@test.com" + data["mailbox"]["full_address"] = "fromapi2@test.com" + response = self.client.post(url, data, format="json") + self.assertEqual(response.status_code, 201) + self.assertTrue(limit.is_exceeded()) + + data["username"] = "fromapi3@test.com" + data["mailbox"]["full_address"] = "fromapi3@test.com" + response = self.client.post(url, data, format="json") + self.assertEqual(response.status_code, 400) + + def test_aliases_limit(self): + """Check mailbox aliases limit.""" + self.client.credentials( + HTTP_AUTHORIZATION='Token ' + self.da_token.key) + + limit = self.user.userobjectlimit_set.get(name="mailbox_aliases") + url = reverse("external_api:alias-list") + data = { + "address": "alias_fromapi@test.com", + "recipients": [ + "user@test.com", "postmaster@test.com", "user_éé@nonlocal.com" + ] + } + response = self.client.post(url, data, format="json") + self.assertEqual(response.status_code, 201) + self.assertFalse(limit.is_exceeded()) + + data["address"] = "alias_fromapi2@test.com" + response = self.client.post(url, data, format="json") + self.assertEqual(response.status_code, 201) + self.assertTrue(limit.is_exceeded()) + + data["address"] = "alias_fromapi3@test.com" + response = self.client.post(url, data, format="json") + self.assertEqual(response.status_code, 400) + + +class APIDomainLimitsTestCase(lib_tests.ModoAPITestCase): + """Check that limits are used also by the API.""" + + @classmethod + def setUpTestData(cls): + """Create test data.""" + super(APIDomainLimitsTestCase, cls).setUpTestData() + parameters.save_admin("ENABLE_DOMAIN_LIMITS", "yes") + for name, tpl in utils.get_domain_limit_templates(): + parameters.save_admin( + "DEFLT_DOMAIN_{}_LIMIT".format(name.upper()), 2) + populate_database() + + def test_mailboxes_limit(self): + """Check mailboxes limit.""" + domain = Domain.objects.get(name="test.com") + limit = domain.domainobjectlimit_set.get(name="mailboxes") + self.assertTrue(limit.is_exceeded()) + url = reverse("external_api:account-list") + data = { + "username": "fromapi@test.com", + "role": "SimpleUsers", + "password": "Toto1234", + "mailbox": { + "full_address": "fromapi@test.com", + "quota": 10 + } + } + response = self.client.post(url, data, format="json") + self.assertEqual(response.status_code, 400) + + def test_domain_aliases_limit(self): + """Check domain_aliases limit.""" + domain = Domain.objects.get(name="test.com") + limit = domain.domainobjectlimit_set.get(name="domain_aliases") + url = reverse("external_api:domain_alias-list") + data = {"name": "dalias1.com", "target": domain.pk} + response = self.client.post(url, data, format="json") + self.assertEqual(response.status_code, 201) + data["name"] = "dalias2.com" + response = self.client.post(url, data, format="json") + self.assertEqual(response.status_code, 201) + self.assertTrue(limit.is_exceeded()) + data["name"] = "dalias3.com" + response = self.client.post(url, data, format="json") + self.assertEqual(response.status_code, 400) + + def test_mailbox_aliases_limit(self): + """Check mailbox_aliases limit.""" + domain = Domain.objects.get(name="test.com") + limit = domain.domainobjectlimit_set.get(name="mailbox_aliases") + self.assertTrue(limit.is_exceeded()) + url = reverse("external_api:alias-list") + data = { + "address": "alias_fromapi@test.com", + "recipients": [ + "user@test.com", "postmaster@test.com", "user_éé@nonlocal.com" + ] + } + response = self.client.post(url, data, format="json") + self.assertEqual(response.status_code, 400) diff --git a/modoboa/limits/tests/test_domain_limits.py b/modoboa/limits/tests/test_domain_limits.py new file mode 100644 index 000000000..12be480f2 --- /dev/null +++ b/modoboa/limits/tests/test_domain_limits.py @@ -0,0 +1,160 @@ +# coding: utf-8 +"""Test cases for the limits extension.""" + +from django.core.urlresolvers import reverse + +from modoboa.admin import factories as admin_factories +from modoboa.admin.models import Domain +from modoboa.core import factories as core_factories +from modoboa.core.models import User +from modoboa.lib import parameters +from modoboa.lib import tests as lib_tests + +from .. import utils + + +class DomainLimitsTestCase(lib_tests.ModoTestCase): + """Per-domain limits tests.""" + + @classmethod + def setUpTestData(cls): + """Create test data.""" + super(DomainLimitsTestCase, cls).setUpTestData() + parameters.save_admin("ENABLE_ADMIN_LIMITS", "no") + parameters.save_admin("ENABLE_DOMAIN_LIMITS", "yes") + for name, tpl in utils.get_domain_limit_templates(): + parameters.save_admin( + "DEFLT_DOMAIN_{}_LIMIT".format(name.upper()), 2) + admin_factories.populate_database() + + def test_set_limits(self): + """Try to set limits for a given domain.""" + domain = Domain.objects.get(name="test.com") + values = { + "name": domain.name, "quota": domain.quota, + "enabled": domain.enabled, "type": "domain", + "mailboxes_limit": 3, "mailbox_aliases_limit": 3, + "domain_aliases_limit": 3, "domain_admins_limit": 3 + } + self.ajax_post( + reverse("admin:domain_change", args=[domain.id]), + values + ) + domain.refresh_from_db() + self.assertEqual( + domain.domainobjectlimit_set.get(name="mailboxes").max_value, 3) + self.assertEqual( + domain.domainobjectlimit_set.get( + name="mailbox_aliases").max_value, 3) + self.assertEqual( + domain.domainobjectlimit_set.get( + name="domain_aliases").max_value, 3) + self.assertEqual( + domain.domainobjectlimit_set.get( + name="domain_admins").max_value, 3) + + def test_domain_aliases_limit(self): + """Try to exceed defined limit.""" + domain = Domain.objects.get(name="test.com") + limit = domain.domainobjectlimit_set.get(name="domain_aliases") + self.assertFalse(limit.is_exceeded()) + values = { + "name": domain.name, "quota": domain.quota, + "enabled": domain.enabled, "type": "domain", + "mailboxes_limit": 2, "mailbox_aliases_limit": 2, + "domain_aliases_limit": 2, "domain_admins_limit": 2, + "aliases": "alias1.com", "aliases_1": "alias2.com" + } + self.ajax_post( + reverse("admin:domain_change", args=[domain.id]), + values + ) + self.assertTrue(limit.is_exceeded()) + values["aliases_2"] = "alias3.com" + self.ajax_post( + reverse("admin:domain_change", args=[domain.id]), + values, 403 + ) + + def test_domain_admins_limit(self): + """Try to exceed defined limit.""" + domain = Domain.objects.get(name="test.com") + limit = domain.domainobjectlimit_set.get(name="domain_admins") + self.assertFalse(limit.is_exceeded()) + user = User.objects.get(username="admin@test2.com") + values = { + "username": user.username, "role": user.group, + "is_active": user.is_active, "email": user.email, + "quota_act": True, "domains": "test2.com", + "domains_1": "test.com" + } + self.ajax_post( + reverse("admin:account_change", args=[user.id]), + values + ) + self.assertTrue(limit.is_exceeded()) + user = core_factories.UserFactory( + username="admin1000@test.com", groups=("DomainAdmins", )) + self.ajax_post( + reverse("admin:account_change", args=[user.id]), + values, 400 + ) + + def test_mailboxes_limit(self): + """Try to exceed defined limits.""" + domain = Domain.objects.get(name="test.com") + domain.domainobjectlimit_set.filter(name="mailboxes").update( + max_value=3) + limit = domain.domainobjectlimit_set.get(name="mailboxes") + self.assertFalse(limit.is_exceeded()) + username = "toto@test.com" + values = { + "username": username, + "first_name": "Tester", "last_name": "Toto", + "password1": "Toto1234", "password2": "Toto1234", + "role": "SimpleUsers", "quota_act": True, + "is_active": True, "email": username, "stepid": "step2", + } + self.ajax_post(reverse("admin:account_add"), values, 200) + self.assertTrue(limit.is_exceeded()) + + username = "titi@test.com" + self.ajax_post(reverse("admin:account_add"), values, 400) + + def test_mailbox_aliases_limit(self): + """Try to exceed defined limits.""" + domain = Domain.objects.get(name="test.com") + user = User.objects.get(username="user@test.com") + limit = domain.domainobjectlimit_set.get(name="mailbox_aliases") + limit.max_value = 4 + limit.save() + self.assertFalse(limit.is_exceeded()) + values = { + "username": user.username, "role": user.group, + "is_active": user.is_active, "email": user.email, + "quota_act": True, + "aliases": "alias@test.com", "aliases_1": "alias1@test.com" + } + self.ajax_post( + reverse("admin:account_change", args=[user.id]), + values + ) + self.assertTrue(limit.is_exceeded()) + + values["aliases_2"] = "alias2@test.com" + self.ajax_post( + reverse("admin:account_change", args=[user.id]), + values, 403 + ) + + limit.max_value = 5 + limit.save() + values = { + "address": "forward2@test.com", "recipients": "user@test.com", + "enabled": True + } + self.ajax_post(reverse("admin:alias_add"), values) + self.assertTrue(limit.is_exceeded()) + + values["address"] = "forward3@test.com" + self.ajax_post(reverse("admin:alias_add"), values, 400) diff --git a/modoboa/limits/tests/test_import.py b/modoboa/limits/tests/test_import.py new file mode 100644 index 000000000..e58139a1b --- /dev/null +++ b/modoboa/limits/tests/test_import.py @@ -0,0 +1,221 @@ +# coding: utf-8 +from django.core.files.base import ContentFile +from django.core.urlresolvers import reverse + +from modoboa.admin import factories as admin_factories +from modoboa.admin import models as admin_models +from modoboa.core import factories as core_factories +from modoboa.core import models as core_models +from modoboa.lib import parameters +from modoboa.lib.tests import ModoTestCase + +from .. import utils + + +class LimitImportTestCase(ModoTestCase): + """Base class to test limits.""" + + @classmethod + def setUpTestData(cls): + """Create test data.""" + super(LimitImportTestCase, cls).setUpTestData() + admin_factories.populate_database() + cls.reseller = core_factories.UserFactory( + username="reseller", groups=("Resellers", )) + cls.dadmin = core_models.User.objects.get(username="admin@test.com") + cls.domain = admin_models.Domain.objects.get(name="test.com") + + def _test_domain_alias_import(self, limit): + """Check domain aliases limit.""" + self.client.login( + username=self.reseller.username, password="toto") + self.assertFalse(limit.is_exceeded()) + f = ContentFile(b"""domainalias; domalias1.com; {domain}; True +domainalias; domalias2.com; {domain}; True +""".format(domain=self.domain), name="domains.csv") + self.client.post( + reverse("admin:domain_import"), { + "sourcefile": f + } + ) + self.assertTrue(limit.is_exceeded()) + + f = ContentFile( + "domainalias; domalias3.com; {}; True".format(self.domain), + name="domains.csv") + response = self.client.post( + reverse("admin:domain_import"), { + "sourcefile": f + } + ) + self.assertContains(response, "Domain aliases: limit reached") + + def _test_domain_admins_import(self, limit, initial_count=0): + """Check domain admins limit.""" + self.client.login( + username=self.reseller.username, password="toto") + self.assertFalse(limit.is_exceeded()) + content = "" + for cpt in xrange(initial_count, 2): + content += ( + "account; admin{cpt}@{domain}; toto; User; One; True; " + "DomainAdmins; user{cpt}@{domain}; 5; {domain}\n" + .format(domain=self.domain, cpt=cpt) + ) + f = ContentFile(content, name="domain_admins.csv") + response = self.client.post( + reverse("admin:identity_import"), { + "sourcefile": f + } + ) + self.assertTrue(limit.is_exceeded()) + + f = ContentFile(b"""account; admin3@{domain}; toto; User; One; True; DomainAdmins; admin3@{domain}; 5; {domain} +""".format(domain=self.domain), name="domain_admins.csv") + response = self.client.post( + reverse("admin:identity_import"), { + "sourcefile": f + } + ) + self.assertContains(response, "Permission denied") + + def _test_mailboxes_import(self, limit): + """Check mailboxes limit.""" + self.client.login( + username=self.dadmin.username, password="toto") + self.assertFalse(limit.is_exceeded()) + f = ContentFile(b"""account; user1@{domain}; toto; User; One; True; SimpleUsers; user1@{domain}; 5 +account; truc@{domain}; toto; René; Truc; True; SimpleUsers; truc@{domain}; 5 +""".format(domain=self.domain), name="domains.csv") + response = self.client.post( + reverse("admin:identity_import"), { + "sourcefile": f + } + ) + self.assertTrue(limit.is_exceeded()) + + f = ContentFile(""" +account; user3@{domain}; toto; User; One; True; SimpleUsers; user3@{domain}; 5 +""".format(domain=self.domain), name="domains.csv") + response = self.client.post( + reverse("admin:identity_import"), { + "sourcefile": f + } + ) + self.assertContains(response, "Mailboxes: limit reached") + + def _test_mailbox_aliases_import(self, limit): + """Check mailbox aliases limit.""" + self.client.login( + username=self.dadmin.username, password="toto") + self.assertFalse(limit.is_exceeded()) + f = ContentFile(b"""alias; alias1@{domain}; True; user@{domain} +alias; alias2@{domain}; True; user@{domain} +""".format(domain=self.domain), name="aliases.csv") + response = self.client.post( + reverse("admin:identity_import"), { + "sourcefile": f + } + ) + self.assertTrue(limit.is_exceeded()) + + f = ContentFile(""" +alias; alias3@{domain}; True; user@{domain} +""".format(domain=self.domain), name="aliases.csv") + response = self.client.post( + reverse("admin:identity_import"), { + "sourcefile": f + } + ) + self.assertContains(response, "Mailbox aliases: limit reached") + + +class UserLimitImportTestCase(LimitImportTestCase): + + @classmethod + def setUpTestData(cls): + """Create test data.""" + for name, tpl in utils.get_user_limit_templates(): + parameters.save_admin( + "DEFLT_USER_{}_LIMIT".format(name.upper()), 2) + super(UserLimitImportTestCase, cls).setUpTestData() + + def test_domains_import(self): + """Check domains limit.""" + self.client.login( + username=self.reseller.username, password="toto") + limit = self.reseller.userobjectlimit_set.get(name="domains") + self.assertFalse(limit.is_exceeded()) + f = ContentFile(b"""domain; domain1.com; 100; True +domain; domain2.com; 200; False +""", name="domains.csv") + self.client.post( + reverse("admin:domain_import"), { + "sourcefile": f + } + ) + self.assertTrue(limit.is_exceeded()) + + f = ContentFile(b"""domain; domain3.com; 100; True +""", name="domains.csv") + response = self.client.post( + reverse("admin:domain_import"), { + "sourcefile": f + } + ) + self.assertContains(response, "Domains: limit reached") + + def test_domain_alias_import(self): + """Check domain aliases limit.""" + limit = self.reseller.userobjectlimit_set.get(name="domain_aliases") + self._test_domain_alias_import(limit) + + def test_domain_admins_import(self): + """Check domain admins limit.""" + self.domain.add_admin(self.reseller) + limit = self.reseller.userobjectlimit_set.get(name="domain_admins") + self._test_domain_admins_import(limit) + + def test_mailboxes_import(self): + """Check mailboxes limit.""" + limit = self.dadmin.userobjectlimit_set.get(name="mailboxes") + self._test_mailboxes_import(limit) + + def test_mailbox_aliases_import(self): + """Check mailbox aliases limit.""" + limit = self.dadmin.userobjectlimit_set.get(name="mailbox_aliases") + self._test_mailbox_aliases_import(limit) + + +class DomainLimitImportTestCase(LimitImportTestCase): + + @classmethod + def setUpTestData(cls): + """Create test data.""" + parameters.save_admin("ENABLE_DOMAIN_LIMITS", "yes") + parameters.save_admin("ENABLE_ADMIN_LIMITS", "no") + for name, tpl in utils.get_domain_limit_templates(): + parameters.save_admin( + "DEFLT_DOMAIN_{}_LIMIT".format(name.upper()), 2) + super(DomainLimitImportTestCase, cls).setUpTestData() + mb = admin_factories.MailboxFactory( + user__username="user@test4.com", + domain__name="test4.com", address="user") + cls.domain = mb.domain + + def test_domain_alias_import(self): + """Check domain aliases limit.""" + limit = self.domain.domainobjectlimit_set.get(name="domain_aliases") + self._test_domain_alias_import(limit) + + def test_mailboxes_import(self): + """Check mailboxes limit.""" + self.domain.add_admin(self.dadmin) + limit = self.domain.domainobjectlimit_set.get(name="mailboxes") + self._test_mailboxes_import(limit) + + def test_mailbox_aliases_import(self): + """Check mailbox aliases limit.""" + self.domain.add_admin(self.dadmin) + limit = self.domain.domainobjectlimit_set.get(name="mailbox_aliases") + self._test_mailbox_aliases_import(limit) diff --git a/modoboa/limits/tests.py b/modoboa/limits/tests/test_user_limits.py similarity index 95% rename from modoboa/limits/tests.py rename to modoboa/limits/tests/test_user_limits.py index 40af6cc93..b00ace977 100644 --- a/modoboa/limits/tests.py +++ b/modoboa/limits/tests/test_user_limits.py @@ -1,3 +1,4 @@ +# coding: utf-8 """Test cases for the limits extension.""" from django.core.urlresolvers import reverse @@ -7,12 +8,12 @@ from modoboa.admin.models import Alias, Domain from modoboa.core.factories import UserFactory from modoboa.core.models import User from modoboa.lib import parameters -from modoboa.lib.tests import ModoTestCase +from modoboa.lib import tests as lib_tests -from . import utils +from .. import utils -class PermissionsTestCase(ModoTestCase): +class PermissionsTestCase(lib_tests.ModoTestCase): @classmethod def setUpTestData(cls): @@ -41,15 +42,15 @@ class PermissionsTestCase(ModoTestCase): self.assertEqual(resp, "Permission denied") -class ResourceTestCase(ModoTestCase): +class ResourceTestCase(lib_tests.ModoTestCase): @classmethod def setUpTestData(cls): """Custom setUpTestData method.""" super(ResourceTestCase, cls).setUpTestData() - for name, tpl in utils.get_limit_templates(): + for name, tpl in utils.get_user_limit_templates(): parameters.save_admin( - "DEFLT_{0}_LIMIT".format(name.upper()), 2, + "DEFLT_USER_{0}_LIMIT".format(name.upper()), 2, app="limits" ) populate_database() @@ -107,7 +108,7 @@ class ResourceTestCase(ModoTestCase): ) def _check_limit(self, name, curvalue, maxvalue): - l = self.user.objectlimit_set.get(name=name) + l = self.user.userobjectlimit_set.get(name=name) self.assertEqual(l.current_value, curvalue) self.assertEqual(l.max_value, maxvalue) @@ -119,7 +120,7 @@ class DomainAdminTestCase(ResourceTestCase): """Create test data.""" super(DomainAdminTestCase, cls).setUpTestData() cls.user = User.objects.get(username='admin@test.com') - cls.user.objectlimit_set.filter( + cls.user.userobjectlimit_set.filter( name__in=["mailboxes", "mailbox_aliases"]).update(max_value=2) def setUp(self): @@ -240,7 +241,8 @@ class ResellerTestCase(ResourceTestCase): ) self._check_limit('domain_admins', 2, 2) - self.user.objectlimit_set.filter(name="mailboxes").update(max_value=3) + self.user.userobjectlimit_set.filter( + name="mailboxes").update(max_value=3) self._create_account('user1@domain.tld') user = User.objects.get(username='user1@domain.tld') values = { @@ -265,11 +267,12 @@ class ResellerTestCase(ResourceTestCase): domadmin = User.objects.get(username='admin1@domain.tld') for l in ['mailboxes', 'mailbox_aliases']: self.assertEqual( - domadmin.objectlimit_set.get(name=l).max_value, 0 + domadmin.userobjectlimit_set.get(name=l).max_value, 0 ) def test_domain_admins_limit_from_domain_tpl(self): - self.user.objectlimit_set.filter(name="domains").update(max_value=3) + self.user.userobjectlimit_set.filter( + name="domains").update(max_value=3) self._create_domain('domain1.tld', withtpl=True) self._create_domain('domain2.tld', withtpl=True) self._check_limit('domain_admins', 2, 2) diff --git a/modoboa/limits/utils.py b/modoboa/limits/utils.py index 1a8e4343f..6cbf3d1b0 100644 --- a/modoboa/limits/utils.py +++ b/modoboa/limits/utils.py @@ -3,7 +3,12 @@ from . import constants -def get_limit_templates(): +def get_user_limit_templates(): """Return defined templates.""" # FIXME: change GetExtraLimitTemplates - return constants.DEFAULT_LIMITS.items() + return constants.DEFAULT_USER_LIMITS.items() + + +def get_domain_limit_templates(): + """Return defined templates.""" + return constants.DEFAULT_DOMAIN_LIMITS.items() diff --git a/modoboa/relaydomains/tests.py b/modoboa/relaydomains/tests.py index 9e9dbbcb2..1c78f4ce3 100644 --- a/modoboa/relaydomains/tests.py +++ b/modoboa/relaydomains/tests.py @@ -50,7 +50,7 @@ class Operations(object): ) def _check_limit(self, name, curvalue, maxvalue): - l = self.user.objectlimit_set.get(name=name) + l = self.user.userobjectlimit_set.get(name=name) self.assertEqual(l.current_value, curvalue) self.assertEqual(l.max_value, maxvalue) @@ -233,9 +233,9 @@ class LimitsTestCase(ModoTestCase, Operations): """Create test data.""" super(LimitsTestCase, cls).setUpTestData() - for name, tpl in limits_utils.get_limit_templates(): + for name, tpl in limits_utils.get_user_limit_templates(): parameters.save_admin( - "DEFLT_{0}_LIMIT".format(name.upper()), 2, app="limits" + "DEFLT_USER_{0}_LIMIT".format(name.upper()), 2, app="limits" ) cls.user = UserFactory.create( username='reseller', groups=('Resellers',)