Merge branch 'master' of github.com:tonioo/modoboa into mysql_tests

This commit is contained in:
Gael Pasgrimaud
2016-03-10 10:44:56 +01:00
20 changed files with 804 additions and 192 deletions
+61
View File
@@ -1,7 +1,14 @@
"""Admin API."""
from django.contrib.contenttypes.models import ContentType
from rest_framework import viewsets
from rest_framework.decorators import list_route
from rest_framework.exceptions import ParseError
from rest_framework.permissions import DjangoModelPermissions
from rest_framework.response import Response
from modoboa.core import models as core_models
from . import models
from . import serializers
@@ -20,3 +27,57 @@ class DomainViewSet(viewsets.ModelViewSet):
def perform_destroy(self, instance):
"""Add custom args to delete call."""
instance.delete(self.request.user)
class AccountViewSet(viewsets.ModelViewSet):
"""ViewSet for User/Mailbox."""
permission_classes = [DjangoModelPermissions, ]
def get_serializer_class(self):
"""Return a serializer."""
if self.request.method == "GET":
return serializers.AccountSerializer
return serializers.WritableAccountSerializer
def get_queryset(self):
"""Filter queryset based on current user."""
user = self.request.user
ids = user.objectaccess_set \
.filter(content_type=ContentType.objects.get_for_model(user)) \
.values_list('object_id', flat=True)
return core_models.User.objects.filter(pk__in=ids)
def perform_destroy(self, instance):
"""Add custom args to delete call."""
instance.delete(self.request.user)
@list_route()
def exists(self, request):
"""Check if account exists."""
email = request.GET.get("email")
if not email:
raise ParseError("email not provided")
if not core_models.User.objects.filter(email=email).exists():
data = {"exists": False}
else:
data = {"exists": True}
serializer = serializers.AccountExistsSerializer(data)
return Response(serializer.data)
class AliasViewSet(viewsets.ModelViewSet):
"""ViewSet for Alias."""
permission_classes = [DjangoModelPermissions, ]
serializer_class = serializers.AliasSerializer
def get_queryset(self):
"""Filter queryset based on current user."""
user = self.request.user
ids = (
user.objectaccess_set.filter(
content_type=ContentType.objects.get_for_model(models.Alias))
.values_list('object_id', flat=True)
)
return models.Alias.objects.filter(pk__in=ids)
+3 -43
View File
@@ -12,43 +12,6 @@ from .models import (
)
PERMISSIONS = {
"DomainAdmins": [
["core", "user", "add_user"],
["core", "user", "change_user"],
["core", "user", "delete_user"],
["admin", "domain", "view_domains"],
["admin", "domain", "view_domain"],
["admin", "mailbox", "add_mailbox"],
["admin", "mailbox", "change_mailbox"],
["admin", "mailbox", "delete_mailbox"],
["admin", "alias", "add_alias"],
["admin", "alias", "change_alias"],
["admin", "alias", "delete_alias"],
["admin", "mailbox", "view_mailboxes"],
["admin", "alias", "view_aliases"],
]
}
@events.observe("GetExtraRoles")
def extra_roles(user, account):
"""Return new roles."""
filters = events.raiseQueryEvent(
"UserCanSetRole", user, "DomainAdmins", account
)
if user.has_perm("admin.add_domain") and \
(not filters or True in filters):
return [("DomainAdmins", _("Domain administrator"))]
return []
@events.observe("GetExtraRolePermissions")
def extra_permissions(rolename):
"""Return extra permissions for :kw:`rolename`."""
return PERMISSIONS.get(rolename, [])
@events.observe("AdminMenuDisplay")
def admin_menu(target, user):
if target != "top_menu":
@@ -151,12 +114,9 @@ def import_account_mailbox(user, account, row):
)
if not user.can_access(domain):
raise PermDeniedException
try:
mb = Mailbox.objects.get(address=mailbox, domain=domain)
except Mailbox.DoesNotExist:
pass
else:
raise Conflict(_("Mailbox %s already exists" % account.email))
if Mailbox.objects.filter(address=mailbox, domain=domain).exists():
raise Conflict(
_("Mailbox {} already exists").format(account.email))
if len(row) == 1:
quota = None
else:
+20 -46
View File
@@ -12,7 +12,7 @@ from passwords.fields import PasswordField
from modoboa.core.models import User
from modoboa.lib import events, parameters
from modoboa.lib.email_utils import split_mailbox
from modoboa.lib.exceptions import PermDeniedException, Conflict, NotFound
from modoboa.lib.exceptions import PermDeniedException
from modoboa.lib import fields as lib_fields
from modoboa.lib.form_utils import (
DynamicForm, TabForms, WizardForm, WizardStep
@@ -43,6 +43,7 @@ class AccountFormGeneral(forms.ModelForm):
password1 = PasswordField(
label=ugettext_lazy("Password"), widget=forms.widgets.PasswordInput
)
password2 = PasswordField(
label=ugettext_lazy("Confirmation"),
widget=forms.widgets.PasswordInput,
@@ -123,6 +124,7 @@ class AccountFormGeneral(forms.ModelForm):
return self.cleaned_data["role"]
def clean_username(self):
"""username must be a valid email address for simple users."""
if "role" not in self.cleaned_data:
return self.cleaned_data["username"]
if self.cleaned_data["role"] != "SimpleUsers":
@@ -160,7 +162,7 @@ class AccountFormGeneral(forms.ModelForm):
and self.cleaned_data["password1"] != "":
account.set_password(self.cleaned_data["password1"])
account.save()
account.set_role(self.cleaned_data["role"])
account.role = self.cleaned_data["role"]
return account
@@ -230,7 +232,15 @@ class AccountFormMail(forms.Form, DynamicForm):
def clean_email(self):
"""Ensure lower case emails"""
return self.cleaned_data["email"].lower()
email = self.cleaned_data["email"].lower()
self.locpart, domname = split_mailbox(email)
if not domname:
return email
try:
self.domain = Domain.objects.get(name=domname)
except Domain.DoesNotExist:
raise forms.ValidationError(_("Domain does not exist"))
return email
def clean(self):
"""Custom fields validation.
@@ -246,54 +256,16 @@ class AccountFormMail(forms.Form, DynamicForm):
def create_mailbox(self, user, account):
"""Create a mailbox associated to :kw:`account`."""
locpart, domname = split_mailbox(self.cleaned_data["email"])
try:
domain = Domain.objects.get(name=domname)
except Domain.DoesNotExist:
raise NotFound(_("Domain does not exist"))
if not user.can_access(domain):
if not user.can_access(self.domain):
raise PermDeniedException
try:
Mailbox.objects.get(address=locpart, domain=domain)
except Mailbox.DoesNotExist:
pass
else:
raise Conflict(
_("Mailbox %s already exists" % self.cleaned_data["email"])
)
events.raiseEvent("CanCreate", user, "mailboxes")
self.mb = Mailbox(address=locpart, domain=domain, user=account,
use_domain_quota=self.cleaned_data["quota_act"])
self.mb = Mailbox(
address=self.locpart, domain=self.domain, user=account,
use_domain_quota=self.cleaned_data["quota_act"])
self.mb.set_quota(self.cleaned_data["quota"],
user.has_perm("admin.add_domain"))
self.mb.save(creator=user)
def update_mailbox(self, user, account):
newaddress = None
if self.cleaned_data["email"] != self.mb.full_address:
newaddress = self.cleaned_data["email"]
elif (account.group == "SimpleUsers" and
account.username != self.mb.full_address):
newaddress = account.username
if newaddress is not None:
self.mb.old_full_address = self.mb.full_address
local_part, domname = split_mailbox(newaddress)
try:
domain = Domain.objects.get(name=domname)
except Domain.DoesNotExist:
raise NotFound(_("Domain does not exist"))
if not user.can_access(domain):
raise PermDeniedException
self.mb.rename(local_part, domain)
self.mb.use_domain_quota = self.cleaned_data["quota_act"]
override_rules = True \
if not self.mb.quota or user.has_perm("admin.add_domain") \
else False
self.mb.set_quota(self.cleaned_data["quota"], override_rules)
self.mb.save()
events.raiseEvent('MailboxModified', self.mb)
def _update_aliases(self, user, account):
"""Update mailbox aliases."""
aliases = []
@@ -342,7 +314,9 @@ class AccountFormMail(forms.Form, DynamicForm):
if not hasattr(self, "mb") or self.mb is None:
self.create_mailbox(user, account)
else:
self.update_mailbox(user, account)
self.cleaned_data["use_domain_quota"] = (
self.cleaned_data["quota_act"])
self.mb.update_from_dict(user, self.cleaned_data)
events.raiseEvent(
'SaveExtraFormFields', 'mailform', self.mb, self.cleaned_data
)
+2 -24
View File
@@ -16,7 +16,7 @@ from modoboa.lib.web_utils import render_to_json_response
from ..lib import check_if_domain_exists
from ..models import (
Domain, DomainAlias, Mailbox, Alias, Quota
Domain, DomainAlias, Mailbox, Alias
)
DOMAIN_TYPES = [
@@ -121,28 +121,6 @@ class DomainFormGeneral(forms.ModelForm, DynamicForm):
)
return cleaned_data
def update_mailbox_quotas(self, domain):
"""Update all quota records associated to this domain.
This method must be called only when a domain gets renamed. As
the primary key used for a quota is an email address, rename a
domain will change all associated email addresses, so it will
change the primary keys used for quotas. The consequence is we
can't issue regular UPDATE queries using the .save() method of
a Quota instance (it will trigger an INSERT as the primary key
has changed).
So, we use this ugly hack to bypass this behaviour. It is not
perfomant at all as it will generate one query per quota
record to update.
"""
for q in Quota.objects.filter(username__contains="@%s" % self.oldname):
username = q.username.replace(
'@%s' % self.oldname, '@%s' % domain.name)
Quota.objects.create(
username=username, bytes=q.bytes, messages=q.messages)
q.delete()
def save(self, user, commit=True, domalias_post_create=False):
"""Custom save method.
@@ -250,7 +228,7 @@ class DomainFormOptions(forms.Form):
da = User(username=username, email=username, is_active=True)
da.set_password("password")
da.save()
da.set_role("DomainAdmins")
da.role = "DomainAdmins"
da.post_create(user)
mb = Mailbox(
@@ -36,14 +36,12 @@ class ExportCommand(BaseCommand):
u.to_csv(self.csvwriter)
dumped_aliases = []
qset = (
models.AliasRecipient.objects.filter(r_alias__isnull=False)
.distinct("r_alias")
.select_related("r_alias")
.prefetch_related("r_alias__aliasrecipient_set")
models.Alias.objects.exclude(alias_recipient_aliases=None)
.distinct().prefetch_related("aliasrecipient_set")
)
for alr in qset:
alr.r_alias.to_csv(self.csvwriter)
dumped_aliases += [alr.r_alias.pk]
for alias in qset:
alias.to_csv(self.csvwriter)
dumped_aliases += [alias.pk]
qset = (
models.Alias.objects.exclude(pk__in=dumped_aliases)
.prefetch_related('aliasrecipient_set')
+5 -2
View File
@@ -147,7 +147,10 @@ class Alias(AdminObject):
@property
def recipients(self):
"""Return the recipient list."""
return self.aliasrecipient_set.values_list("address", flat=True)
return (
self.aliasrecipient_set.order_by("address")
.values_list("address", flat=True)
)
@property
def recipients_count(self):
@@ -210,5 +213,5 @@ class AliasRecipient(models.Model):
def __str__(self):
"""Return alias and recipient."""
return smart_text(
"{} -> {}".format(self.alias.address, self.alias)
"{} -> {}".format(self.alias.address, self.address)
)
+48 -11
View File
@@ -15,8 +15,9 @@ import reversion
from .base import AdminObject
from .domain import Domain
from modoboa.core.models import User
from modoboa.lib import parameters
from modoboa.lib.exceptions import BadRequest, InternalError
from modoboa.lib import events, parameters
from modoboa.lib import exceptions as lib_exceptions
from modoboa.lib.email_utils import split_mailbox
from modoboa.lib.sysutils import exec_cmd
@@ -96,6 +97,8 @@ class Mailbox(AdminObject):
def __init__(self, *args, **kwargs):
super(Mailbox, self).__init__(*args, **kwargs)
self.__mail_home = None
self.old_full_address = self.full_address
self.old_mail_home = self.mail_home
def __str__(self):
return smart_text(self.full_address)
@@ -142,7 +145,7 @@ class Mailbox(AdminObject):
"doveadm user %s -f home" % self.full_address, **options
)
if code:
raise InternalError(
raise lib_exceptions.InternalError(
_("Failed to retrieve mailbox location (%s)" % output))
self.__mail_home = output.strip()
return self.__mail_home
@@ -179,11 +182,12 @@ class Mailbox(AdminObject):
self._quota_value = instance
def rename_dir(self, old_mail_home):
"""Rename local directory if needed."""
hm = parameters.get_admin("HANDLE_MAILBOXES", raise_error=False)
if hm is None or hm == "no":
return
MailboxOperation.objects.create(
mailbox=self, type='rename', argument=old_mail_home
mailbox=self, type="rename", argument=old_mail_home
)
def rename(self, address, domain):
@@ -197,7 +201,6 @@ class Mailbox(AdminObject):
:param Domain domain: the new mailbox's domain
"""
old_mail_home = self.mail_home
old_qvalue = self.quota_value
self.address = address
self.domain = domain
@@ -206,7 +209,7 @@ class Mailbox(AdminObject):
messages=old_qvalue.messages
)
old_qvalue.delete()
self.rename_dir(old_mail_home)
self.rename_dir(self.old_mail_home)
def delete_dir(self):
hm = parameters.get_admin("HANDLE_MAILBOXES", raise_error=False)
@@ -231,14 +234,14 @@ class Mailbox(AdminObject):
else:
self.quota = 0
elif int(value) > self.domain.quota and not override_rules:
raise BadRequest(
_("Quota is greater than the allowed domain's limit (%dM)"
% self.domain.quota)
raise lib_exceptions.BadRequest(
_("Quota is greater than the allowed domain's limit (%dM)")
% self.domain.quota
)
else:
self.quota = value
if not self.quota and self.domain.quota and not override_rules:
raise BadRequest(_("A quota is required"))
raise lib_exceptions.BadRequest(_("A quota is required"))
def get_quota(self):
"""Get quota limit.
@@ -277,11 +280,45 @@ class Mailbox(AdminObject):
grant_access_to_object(admin, self)
grant_access_to_object(admin, self.user)
def update_from_dict(self, user, values):
"""Update mailbox from a dictionary."""
newaddress = None
if values["email"] != self.full_address:
newaddress = values["email"]
elif (self.user.group == "SimpleUsers" and
self.user.username != self.full_address):
newaddress = self.user.username
if newaddress is not None:
local_part, domname = split_mailbox(newaddress)
domain = Domain.objects.filter(name=domname).first()
if domain is None:
raise lib_exceptions.NotFound(_("Domain does not exist"))
if not user.can_access(domain):
raise lib_exceptions.PermDeniedException
if "use_domain_quota" in values:
self.use_domain_quota = values["use_domain_quota"]
override_rules = True \
if not self.quota or user.has_perm("admin.add_domain") \
else False
self.set_quota(values["quota"], override_rules)
if newaddress:
self.rename(local_part, domain)
self.save()
events.raiseEvent("MailboxModified", self)
def save(self, *args, **kwargs):
"""Custom save.
We just make sure a quota record is defined for this mailbox.
We check that the address is unique and we make sure a quota
record is defined for this mailbox.
"""
qset = Mailbox.objects.filter(address=self.address, domain=self.domain)
if self.pk:
qset = qset.exclude(pk=self.pk)
if qset.exists():
raise lib_exceptions.Conflict(
_("Mailbox {} already exists").format(self))
if self.quota_value is None:
self.quota_value, created = Quota.objects.get_or_create(
username=self.full_address)
+190
View File
@@ -1,6 +1,15 @@
"""Admin serializers."""
from django.core.exceptions import ValidationError
from django.shortcuts import get_object_or_404
from django.utils.translation import ugettext as _
from rest_framework import serializers
from passwords import validators
from modoboa.admin import models as admin_models
from modoboa.core import models as core_models
from modoboa.lib import permissions, email_utils, fields as lib_fields
from . import models
@@ -12,3 +21,184 @@ class DomainSerializer(serializers.ModelSerializer):
class Meta:
model = models.Domain
fields = ("pk", "name", "quota", "enabled", "type", )
def create(self, validated_data):
"""Set permissions."""
domain = models.Domain(**validated_data)
domain.save(creator=self.context["request"].user)
return domain
class MailboxSerializer(serializers.ModelSerializer):
"""Base mailbox serializer."""
full_address = lib_fields.DRFEmailFieldUTF8()
class Meta:
model = models.Mailbox
fields = ("full_address", "use_domain_quota", "quota", )
class AccountSerializer(serializers.ModelSerializer):
"""Base account serializer."""
role = serializers.SerializerMethodField()
mailbox = MailboxSerializer(required=False)
class Meta:
model = core_models.User
fields = (
"pk", "username", "first_name", "last_name", "is_active",
"master_user", "mailbox", "role", "language", "phone_number",
"secondary_email",
)
def __init__(self, *args, **kwargs):
"""Adapt fields to current user."""
super(AccountSerializer, self).__init__(*args, **kwargs)
user = self.context["request"].user
if not user.is_superuser:
del self.fields["master_user"]
def get_role(self, account):
"""Return role."""
return account.group
class AccountExistsSerializer(serializers.Serializer):
"""Simple serializer used with existence checks."""
exists = serializers.BooleanField()
class WritableAccountSerializer(AccountSerializer):
"""Serializer to create account."""
class Meta(AccountSerializer.Meta):
fields = AccountSerializer.Meta.fields + (
"password", )
def __init__(self, *args, **kwargs):
"""Adapt fields to current user."""
super(WritableAccountSerializer, self).__init__(*args, **kwargs)
user = self.context["request"].user
self.fields["role"] = serializers.ChoiceField(
choices=permissions.get_account_roles(user))
def validate_password(self, value):
"""Check password constraints."""
try:
validators.validate_length(value)
validators.complexity(value)
except ValidationError as exc:
raise serializers.ValidationError(exc.messages[0])
return value
def validate(self, data):
"""Check constraints."""
master_user = data.get("master_user", False)
if master_user and data["role"] != "SuperAdmins":
raise serializers.ValidationError({
"master_user": _("Not allowed for this role.")
})
if data["role"] == "SimpleUsers":
mailbox = data.get("mailbox")
if mailbox is None:
data["mailbox"] = {
"full_address": data["username"], "use_domain_quota": True
}
elif mailbox["full_address"] != data["username"]:
raise serializers.ValidationError({
"username": _("Must be equal to mailbox full_address")
})
return data
def _create_mailbox(self, creator, account, data):
"""Create a new Mailbox instance."""
full_address = data.pop("full_address")
address, domain_name = email_utils.split_mailbox(full_address)
domain = get_object_or_404(
admin_models.Domain, name=domain_name)
if not creator.can_access(domain):
raise serializers.ValidationError({
"domain": _("Permission denied.")})
mb = admin_models.Mailbox(
user=account, address=address, domain=domain, **data)
mb.set_quota(
data.get("quota"), creator.has_perm("admin.add_domain")
)
mb.save(creator=creator)
account.email = full_address
return mb
def create(self, validated_data):
"""Create appropriate objects."""
creator = self.context["request"].user
mailbox_data = validated_data.pop("mailbox", None)
role = validated_data.pop("role")
user = core_models.User(**validated_data)
user.set_password(validated_data["password"])
user.save(creator=creator)
if mailbox_data:
self._create_mailbox(creator, user, mailbox_data)
user.role = role
return user
def update(self, instance, validated_data):
"""Update account and associated objects."""
mailbox_data = validated_data.pop("mailbox")
password = validated_data.pop("password")
for key, value in validated_data.items():
setattr(instance, key, value)
instance.set_password(password)
instance.save()
if mailbox_data:
creator = self.context["request"].user
if instance.mailbox:
# FIXME: compat, to remove ASAP.
mailbox_data["email"] = mailbox_data["full_address"]
instance.mailbox.update_from_dict(creator, mailbox_data)
else:
self._create_mailbox(creator, instance, mailbox_data)
return instance
class AliasSerializer(serializers.ModelSerializer):
"""Base Alias serializer."""
address = lib_fields.DRFEmailFieldUTF8AndEmptyUser()
recipients = serializers.ListField(
child=lib_fields.DRFEmailFieldUTF8AndEmptyUser(),
allow_empty=False)
class Meta:
model = admin_models.Alias
fields = ("pk", "address", "enabled", "internal", "recipients")
def validate_address(self, value):
"""Check domain."""
local_part, domain = email_utils.split_mailbox(value)
self.domain = admin_models.Domain.objects.filter(name=domain).first()
if self.domain is None:
raise serializers.ValidationError(_("Domain not found."))
if not self.context["request"].user.can_access(self.domain):
raise serializers.ValidationError(_("Permission denied."))
return value
def create(self, validated_data):
"""Create appropriate objects."""
creator = self.context["request"].user
recipients = validated_data.pop("recipients", None)
alias = admin_models.Alias(domain=self.domain, **validated_data)
alias.save(creator=creator)
alias.set_recipients(recipients)
return alias
def update(self, instance, validated_data):
"""Update objects."""
recipients = validated_data.pop("recipients", None)
for key, value in validated_data.items():
setattr(instance, key, value)
instance.save()
instance.set_recipients(recipients)
return instance
+5
View File
@@ -6,6 +6,11 @@
* @classdesc This class contains utility methods used by the admin
* application.
*/
/*global
$ Listing
*/
var Admin = function(options) {
Listing.call(this, options);
};
+4 -2
View File
@@ -2,7 +2,7 @@
from .account import AccountTestCase, PermissionsTestCase
from .alias import AliasTestCase
from .api import APITestCase
from .api import DomainAPITestCase, AccountAPITestCase, AliasAPITestCase
from .domain import DomainTestCase
from .domain_alias import DomainAliasTestCase
from .export import ExportTestCase
@@ -12,9 +12,11 @@ from .password_schemes import PasswordSchemesTestCase
from .user import ForwardTestCase
__all__ = [
'AccountAPITestCase',
'AccountTestCase',
'AliasAPITestCase',
'AliasTestCase',
'APITestCase',
'DomainAPITestCase',
'DomainTestCase',
'DomainAliasTestCase',
'ExportTestCase',
+27 -2
View File
@@ -44,11 +44,11 @@ class AccountTestCase(ModoTestCase):
alias__internal=True).exists()
)
values["username"] = "pouet@test.com"
values.update({"username": "pouet@test.com"})
self.ajax_post(
reverse("admin:account_change", args=[account.id]), values
)
mb = models.Mailbox.objects.get(pk=mb.id)
mb = models.Mailbox.objects.get(pk=mb.pk)
self.assertEqual(mb.full_address, "pouet@test.com")
self.assertEqual(mb.quota_value.username, "pouet@test.com")
# Check if self alias has been updated
@@ -67,6 +67,31 @@ class AccountTestCase(ModoTestCase):
alias__internal=True).exists()
)
def test_conflicts(self):
"""Check if unicity constraints are respected."""
values = {
"username": "user@test.com",
"password1": "Toto1234", "password2": "Toto1234",
"role": "SimpleUsers", "quota_act": True,
"is_active": True, "email": "user@test.com",
"stepid": "step2"
}
self.ajax_post(reverse("admin:account_add"), values, status=400)
values.update({"username": "fakeuser@test.com",
"email": "fakeuser@test.com"})
self.ajax_post(reverse("admin:account_add"), values)
account = User.objects.get(username="fakeuser@test.com")
values = {
"username": "user@test.com",
"role": "SimpleUsers", "quota_act": True,
"is_active": True, "email": "user@test.com",
}
self.ajax_post(
reverse("admin:account_change", args=[account.pk]), values,
status=400
)
def test_utf8_username(self):
"""Create an account with non-ASCII characters."""
values = dict(
+320 -3
View File
@@ -1,5 +1,9 @@
# coding: utf-8
"""Admin API related tests."""
import copy
import json
from django.core.urlresolvers import reverse
from rest_framework.authtoken.models import Token
@@ -11,14 +15,13 @@ from .. import factories
from .. import models
class APITestCase(ModoAPITestCase):
class DomainAPITestCase(ModoAPITestCase):
"""Check API."""
@classmethod
def setUpTestData(cls):
"""Create test data."""
super(APITestCase, cls).setUpTestData()
super(DomainAPITestCase, cls).setUpTestData()
factories.populate_database()
cls.da_token = Token.objects.create(
user=core_models.User.objects.get(username="admin@test.com"))
@@ -70,6 +73,9 @@ class APITestCase(ModoAPITestCase):
response = self.client.put(url, {"name": "test42.com", "quota": 1000})
self.assertEqual(response.status_code, 200)
self.assertTrue(
models.Mailbox.objects.filter(
address="user", domain__name="test42.com").exists())
def test_delete_domain(self):
"""Try to delete a domain."""
@@ -78,3 +84,314 @@ class APITestCase(ModoAPITestCase):
response = self.client.delete(url)
self.assertEqual(response.status_code, 204)
self.assertFalse(models.Domain.objects.filter(pk=domain.pk).exists())
class AccountAPITestCase(ModoAPITestCase):
"""Check Account API."""
ACCOUNT_DATA = {
"username": "fromapi@test.com",
"role": "SimpleUsers",
"password": "Toto1234",
"mailbox": {
"full_address": "fromapi@test.com",
"quota": 10
}
}
@classmethod
def setUpTestData(cls):
"""Create test data."""
super(AccountAPITestCase, cls).setUpTestData()
factories.populate_database()
cls.da_token = Token.objects.create(
user=core_models.User.objects.get(username="admin@test.com"))
def test_get_accounts(self):
"""Retrieve a list of accounts."""
url = reverse("external_api:account-list")
response = self.client.get(url)
self.assertEqual(response.status_code, 200)
response = json.loads(response.content)
self.assertEqual(len(response), 5)
def test_create_account(self):
"""Try to create a new account."""
url = reverse("external_api:account-list")
response = self.client.post(url, self.ACCOUNT_DATA, format="json")
self.assertEqual(response.status_code, 201)
account = json.loads(response.content)
user = core_models.User.objects.filter(pk=account["pk"]).first()
self.assertIsNot(user, None)
self.assertIsNot(user.mailbox, None)
domadmin = core_models.User.objects.get(username="admin@test.com")
self.assertTrue(domadmin.can_access(user))
data = copy.deepcopy(self.ACCOUNT_DATA)
data["username"] = "fromapi_ééé@test.com"
data["mailbox"]["full_address"] = data["username"]
url = reverse("external_api:account-list")
response = self.client.post(url, data, format="json")
self.assertEqual(response.status_code, 201)
def test_create_domainadmin_account(self):
"""Try to create a domain admin."""
data = copy.deepcopy(self.ACCOUNT_DATA)
data["role"] = "DomainAdmins"
url = reverse("external_api:account-list")
response = self.client.post(url, data, format="json")
self.assertEqual(response.status_code, 201)
data["username"] = "domain_admin"
del data["mailbox"]
response = self.client.post(url, data, format="json")
self.assertEqual(response.status_code, 201)
def test_create_account_with_no_mailbox(self):
"""Try to create a new account."""
data = copy.deepcopy(self.ACCOUNT_DATA)
del data["mailbox"]
url = reverse("external_api:account-list")
response = self.client.post(url, data, format="json")
self.assertEqual(response.status_code, 201)
account = json.loads(response.content)
user = core_models.User.objects.filter(pk=account["pk"]).first()
self.assertIsNot(user, None)
self.assertIsNot(user.mailbox, None)
self.assertEqual(user.mailbox.quota, user.mailbox.domain.quota)
def test_create_existing_account(self):
"""Check if unicity is respected."""
data = copy.deepcopy(self.ACCOUNT_DATA)
data["username"] = "user@test.com"
url = reverse("external_api:account-list")
response = self.client.post(url, data, format="json")
self.assertEqual(response.status_code, 400)
data.update({"username": "domainadmin", "role": "DomainAdmins"})
data["mailbox"]["full_address"] = "admin@test.com"
response = self.client.post(url, data, format="json")
self.assertEqual(response.status_code, 409)
def test_create_account_bad_password(self):
"""Try to create a new account."""
data = copy.deepcopy(self.ACCOUNT_DATA)
data["password"] = "toto"
url = reverse("external_api:account-list")
response = self.client.post(url, data, format="json")
self.assertEqual(response.status_code, 400)
errors = json.loads(response.content)
self.assertIn("password", errors)
def test_create_account_as_domadmin(self):
"""As DomainAdmin, try to create a new account."""
self.client.credentials(
HTTP_AUTHORIZATION='Token ' + self.da_token.key)
data = copy.deepcopy(self.ACCOUNT_DATA)
data["mailbox"]["quota"] = 20
url = reverse("external_api:account-list")
response = self.client.post(url, data, format="json")
self.assertEqual(response.status_code, 400)
data["username"] = "fromapi@test2.com"
data["mailbox"].update({"full_address": "fromapi@test2.com",
"quota": 10})
url = reverse("external_api:account-list")
response = self.client.post(url, data, format="json")
self.assertEqual(response.status_code, 400)
errors = json.loads(response.content)
self.assertIn("domain", errors)
def test_create_account_bad_master_user(self):
"""Try to create a new account."""
data = copy.deepcopy(self.ACCOUNT_DATA)
data["master_user"] = True
url = reverse("external_api:account-list")
response = self.client.post(url, data, format="json")
self.assertEqual(response.status_code, 400)
errors = json.loads(response.content)
self.assertIn("master_user", errors)
def test_update_account(self):
"""Try to update an account."""
account = core_models.User.objects.get(username="user@test.com")
url = reverse("external_api:account-detail", args=[account.pk])
data = {
"username": "fromapi@test.com",
"role": account.role,
"password": "Toto1234",
"mailbox": {
"full_address": "fromapi@test.com",
"quota": account.mailbox.quota
}
}
response = self.client.put(url, data, format="json")
self.assertEqual(response.status_code, 200)
account.refresh_from_db()
self.assertEqual(account.email, account.mailbox.full_address)
def test_update_account_wrong_address(self):
"""Try to update an account."""
account = core_models.User.objects.get(username="user@test.com")
url = reverse("external_api:account-detail", args=[account.pk])
data = {
"username": "fromapi@test3.com",
"role": account.role,
"password": "Toto1234",
"mailbox": {
"full_address": "fromapi@test3.com",
"quota": account.mailbox.quota
}
}
response = self.client.put(url, data, format="json")
self.assertEqual(response.status_code, 404)
def test_delete_account(self):
"""Try to delete an account."""
account = core_models.User.objects.get(username="user@test.com")
domadmin = core_models.User.objects.get(username="admin@test.com")
self.assertTrue(domadmin.can_access(account))
url = reverse("external_api:account-detail", args=[account.pk])
response = self.client.delete(url)
self.assertEqual(response.status_code, 204)
self.assertFalse(
core_models.User.objects.filter(pk=account.pk).exists())
self.assertFalse(domadmin.can_access(account))
def test_account_exists(self):
"""Validate /exists/ service."""
url = reverse("external_api:account-exists")
response = self.client.get(
"{}?email={}".format(url, "user@test.com"))
self.assertEqual(response.status_code, 200)
content = json.loads(response.content)
self.assertTrue(content["exists"])
response = self.client.get(
"{}?email={}".format(url, "pipo@test.com"))
self.assertEqual(response.status_code, 200)
content = json.loads(response.content)
self.assertFalse(content["exists"])
response = self.client.get(url)
self.assertEqual(response.status_code, 400)
class AliasAPITestCase(ModoAPITestCase):
"""Check Alias API."""
ALIAS_DATA = {
"address": "alias_fromapi@test.com",
"recipients": [
"user@test.com", "postmaster@test.com", "user_éé@nonlocal.com"
]
}
@classmethod
def setUpTestData(cls):
"""Create test data."""
super(AliasAPITestCase, cls).setUpTestData()
factories.populate_database()
cls.da_token = Token.objects.create(
user=core_models.User.objects.get(username="admin@test.com"))
def test_get_aliases(self):
"""Retrieve a list of aliases."""
url = reverse("external_api:alias-list")
response = self.client.get(url)
self.assertEqual(response.status_code, 200)
response = json.loads(response.content)
self.assertEqual(len(response), 3)
def test_get_alias(self):
"""Retrieve an alias."""
al = models.Alias.objects.get(address="alias@test.com")
url = reverse("external_api:alias-detail", args=[al.pk])
response = self.client.get(url)
self.assertEqual(response.status_code, 200)
response = json.loads(response.content)
self.assertEqual(response["recipients"], ["user@test.com"])
def test_create_alias(self):
"""Try to create a new alias."""
url = reverse("external_api:alias-list")
response = self.client.post(url, self.ALIAS_DATA, format="json")
self.assertEqual(response.status_code, 201)
alias = json.loads(response.content)
alias = models.Alias.objects.filter(pk=alias["pk"]).first()
domadmin = core_models.User.objects.get(username="admin@test.com")
self.assertTrue(domadmin.can_access(alias))
self.assertEqual(alias.aliasrecipient_set.count(), 3)
self.assertTrue(alias.aliasrecipient_set.filter(
address="user@test.com", r_mailbox__isnull=False).exists())
self.assertTrue(alias.aliasrecipient_set.filter(
address="postmaster@test.com", r_alias__isnull=False).exists())
self.assertTrue(alias.aliasrecipient_set.filter(
address="user_éé@nonlocal.com",
r_mailbox__isnull=True, r_alias__isnull=True).exists())
# Create catchall alias
data = copy.deepcopy(self.ALIAS_DATA)
data["address"] = "@test.com"
response = self.client.post(url, data, format="json")
self.assertEqual(response.status_code, 201)
def test_create_alias_as_domadmin(self):
"""As DomainAdmin, try to create a new alias."""
self.client.credentials(
HTTP_AUTHORIZATION='Token ' + self.da_token.key)
url = reverse("external_api:alias-list")
response = self.client.post(url, self.ALIAS_DATA, format="json")
self.assertEqual(response.status_code, 201)
data = copy.deepcopy(self.ALIAS_DATA)
data["address"] = "alias_fromapi@test2.com"
response = self.client.post(url, data, format="json")
self.assertEqual(response.status_code, 400)
errors = json.loads(response.content)
self.assertIn("address", errors)
def test_update_alias(self):
"""Try to update an alias."""
alias = models.Alias.objects.get(address="alias@test.com")
url = reverse("external_api:alias-detail", args=[alias.pk])
data = {
"address": "alias@test.com",
"recipients": ["user@test.com", "user@nonlocal.com"]
}
response = self.client.put(url, data, format="json")
self.assertEqual(response.status_code, 200)
alias.refresh_from_db()
self.assertEqual(alias.aliasrecipient_set.count(), 2)
data = {
"address": "alias@test.com",
"recipients": ["user@nonlocal.com"]
}
response = self.client.put(url, data, format="json")
self.assertEqual(response.status_code, 200)
alias.refresh_from_db()
self.assertEqual(alias.aliasrecipient_set.count(), 1)
data = {
"address": "alias@test.com",
"recipients": []
}
response = self.client.put(url, data, format="json")
self.assertEqual(response.status_code, 400)
def test_delete_alias(self):
"""Try to delete an existing alias."""
alias = models.Alias.objects.get(address="alias@test.com")
domadmin = core_models.User.objects.get(username="admin@test.com")
self.assertTrue(domadmin.can_access(alias))
url = reverse("external_api:alias-detail", args=[alias.pk])
response = self.client.delete(url)
self.assertEqual(response.status_code, 204)
self.assertFalse(
models.Alias.objects.filter(pk=alias.pk).exists())
self.assertFalse(domadmin.can_access(alias))
self.assertFalse(
models.AliasRecipient.objects.filter(
address="user@test.com", alias__address="alias@test.com")
.exists()
)
+2
View File
@@ -7,5 +7,7 @@ from . import api
router = routers.SimpleRouter()
router.register(r"domains", api.DomainViewSet, base_name="domain")
router.register(r"accounts", api.AccountViewSet, base_name="account")
router.register(r"aliases", api.AliasViewSet, base_name="alias")
urlpatterns = router.urls
+35 -1
View File
@@ -6,7 +6,41 @@ from modoboa.lib import events
PERMISSIONS = {
"SimpleUsers": []
"SimpleUsers": [],
"DomainAdmins": [
["core", "user", "add_user"],
["core", "user", "change_user"],
["core", "user", "delete_user"],
["admin", "domain", "view_domains"],
["admin", "domain", "view_domain"],
["admin", "mailbox", "add_mailbox"],
["admin", "mailbox", "change_mailbox"],
["admin", "mailbox", "delete_mailbox"],
["admin", "alias", "add_alias"],
["admin", "alias", "change_alias"],
["admin", "alias", "delete_alias"],
["admin", "mailbox", "view_mailboxes"],
["admin", "alias", "view_aliases"],
],
"Resellers": [
["core", "user", "add_user"],
["core", "user", "change_user"],
["core", "user", "delete_user"],
["admin", "domain", "view_domains"],
["admin", "domain", "view_domain"],
["admin", "mailbox", "add_mailbox"],
["admin", "mailbox", "change_mailbox"],
["admin", "mailbox", "delete_mailbox"],
["admin", "alias", "add_alias"],
["admin", "alias", "change_alias"],
["admin", "alias", "delete_alias"],
["admin", "mailbox", "view_mailboxes"],
["admin", "alias", "view_aliases"],
["admin", "domain", "view_domains"],
["admin", "domain", "add_domain"],
["admin", "domain", "change_domain"],
["admin", "domain", "delete_domain"]
]
}
+16 -3
View File
@@ -234,6 +234,7 @@ class User(PermissionsMixin):
@property
def group(self):
"""FIXME: DEPRECATED"""
if self.is_superuser:
return "SuperAdmins"
try:
@@ -312,7 +313,18 @@ class User(PermissionsMixin):
return True
return False
def set_role(self, role):
@property
def role(self):
"""Return user role."""
if self.is_superuser:
return "SuperAdmins"
try:
return self.groups.all()[0].name
except IndexError:
return "---"
@role.setter
def role(self, role):
"""Set administrative role for this account
:param string role: the role to set
@@ -337,6 +349,7 @@ class User(PermissionsMixin):
self.save()
def post_create(self, creator):
"""Grant permission on this user to creator."""
from modoboa.lib.permissions import grant_access_to_object
grant_access_to_object(creator, self, is_owner=True)
events.raiseEvent("AccountCreated", self)
@@ -407,7 +420,7 @@ class User(PermissionsMixin):
self.last_name = row[4].strip()
self.is_active = (row[5].strip() in ["True", "1", "yes", "y"])
self.save()
self.set_role(desired_role)
self.role = desired_role
self.post_create(user)
if len(row) < 8:
return
@@ -449,7 +462,7 @@ def populate_callback(user, group='SimpleUsers'):
from modoboa.lib.permissions import grant_access_to_object
sadmins = User.objects.filter(is_superuser=True)
user.set_role(group)
user.role = group
user.post_create(sadmins[0])
for su in sadmins[1:]:
grant_access_to_object(su, user)
+30 -3
View File
@@ -3,11 +3,12 @@
from django import forms
from django.utils.translation import ugettext_lazy
from rest_framework import serializers
from . import validators
class DomainNameField(forms.fields.CharField):
"""A subclass of CharField that only accepts a valid domain name."""
default_error_messages = {
@@ -22,17 +23,43 @@ class DomainNameField(forms.fields.CharField):
class UTF8EmailField(forms.fields.EmailField):
"""A subclass of EmailField to support UTF8 addresses."""
default_validators = [validators.validate_utf8_email]
class UTF8AndEmptyUserEmailField(forms.fields.EmailField):
"""
A subclass of EmailField to support UTF8 addresses (with or without
local part).
"""
default_validators = [validators.validate_utf8_and_empty_user_email]
class DRFEmailFieldUTF8(serializers.CharField):
"""Custom DRF email field to support UTF8."""
default_error_messages = {
"invalid": ugettext_lazy("Enter a valid email address.")
}
def __init__(self, **kwargs):
super(DRFEmailFieldUTF8, self).__init__(**kwargs)
validator = validators.UTF8EmailValidator(
message=self.error_messages["invalid"])
self.validators.append(validator)
class DRFEmailFieldUTF8AndEmptyUser(serializers.CharField):
"""Custom DRF email field to support UTF8 and empty local part."""
default_error_messages = {
"invalid": ugettext_lazy("Enter a valid email address.")
}
def __init__(self, **kwargs):
super(DRFEmailFieldUTF8AndEmptyUser, self).__init__(**kwargs)
validator = validators.UTF8AndEmptyUserEmailValidator(
message=self.error_messages["invalid"])
self.validators.append(validator)
+11 -9
View File
@@ -21,21 +21,23 @@ class AjaxLoginRedirect(object):
class CommonExceptionCatcher(object):
"""Modoboa exceptions catcher."""
def process_exception(self, request, exception):
if not isinstance(exception, ModoboaException):
return None
if not request.is_ajax():
return _render_error(
request, user_context=dict(error=str(exception))
if request.is_ajax() or "/api/" in request.path:
if exception.http_code is None:
return ajax_response(
request, status="ko", respmsg=unicode(exception),
norefresh=True
)
return render_to_json_response(
unicode(exception), status=exception.http_code
)
if exception.http_code is None:
return ajax_response(
request, status="ko", respmsg=unicode(exception), norefresh=True
)
return render_to_json_response(
unicode(exception), status=exception.http_code
return _render_error(
request, user_context=dict(error=str(exception))
)
+16 -5
View File
@@ -2,11 +2,16 @@
from django.contrib.auth.models import Group, Permission
from django.contrib.contenttypes.models import ContentType
from django.utils.translation import ugettext as _
from django.utils.translation import ugettext_lazy
from modoboa.core.models import ObjectAccess, User
from modoboa.lib import events
SIMPLEUSERS_ROLE = ("SimpleUsers", ugettext_lazy("Simple user"))
DOMAINADMINS_ROLE = ("DomainAdmins", ugettext_lazy("Domain administrator"))
RESELLERS_ROLE = ("Resellers", ugettext_lazy("Reseller"))
SUPERADMINS_ROLE = ("SuperAdmins", ugettext_lazy("Super administrator"))
def get_account_roles(user, account=None):
"""Return the list of available account roles.
@@ -18,11 +23,17 @@ def get_account_roles(user, account=None):
:param ``User`` account: account beeing modified (None on creation)
:return: list of strings
"""
std_roles = [("SimpleUsers", _("Simple user"))]
result = [SIMPLEUSERS_ROLE]
filters = events.raiseQueryEvent(
"UserCanSetRole", user, "DomainAdmins", account
)
if user.has_perm("admin.add_domain") and \
(not filters or True in filters):
result += [DOMAINADMINS_ROLE]
if user.is_superuser:
std_roles += [("SuperAdmins", _("Super administrator"))]
std_roles += events.raiseQueryEvent("GetExtraRoles", user, account)
return sorted(std_roles, key=lambda role: role[1])
result += [RESELLERS_ROLE, SUPERADMINS_ROLE]
result += events.raiseQueryEvent("GetExtraRoles", user, account)
return sorted(result, key=lambda role: role[1])
def grant_access_to_object(user, obj, is_owner=False):
+2 -2
View File
@@ -53,11 +53,11 @@ class UTF8EmailValidator(EmailValidator):
validate_utf8_email = UTF8EmailValidator()
class UT8AndEmptyUserEmailValidator(UTF8EmailValidator):
class UTF8AndEmptyUserEmailValidator(UTF8EmailValidator):
"""Same as upper + allows empty local part."""
user_regex = re.compile(
ur"(^$|" + UTF8EmailValidator.user_regex_raw, re.IGNORECASE)
validate_utf8_and_empty_user_email = UT8AndEmptyUserEmailValidator()
validate_utf8_and_empty_user_email = UTF8AndEmptyUserEmailValidator()
+2 -29
View File
@@ -4,37 +4,10 @@ from django.utils.translation import ugettext as _
from modoboa.lib import events
from modoboa.admin.callbacks import PERMISSIONS as ADMIN_PERMS
from .forms import ResourcePoolForm
from .models import LimitTemplates
PERMISSIONS = {
"Resellers": (
ADMIN_PERMS.get("DomainAdmins") +
[["admin", "domain", "view_domains"],
["admin", "domain", "add_domain"],
["admin", "domain", "change_domain"],
["admin", "domain", "delete_domain"]]
)
}
@events.observe("GetExtraRoles")
def get_extra_roles(user, account):
"""Return additional roles."""
if user.is_superuser:
return [("Resellers", _("Reseller")), ]
return []
@events.observe("GetExtraRolePermissions")
def extra_permissions(rolename):
"""Return extra permissions for Resellers."""
return PERMISSIONS.get(rolename, [])
@events.observe("ExtraAdminContent")
def display_pool_usage(user, target, currentpage):
from django.template.loader import render_to_string
@@ -49,8 +22,8 @@ def display_pool_usage(user, target, currentpage):
names = [
tpl[0] for tpl in LimitTemplates().templates
if tpl[0] not in ["domain_admins_limit", "mailboxes_limit",
"mailbox_aliases_limit"]
and (len(tpl) == 3 or tpl[3] == user.group)
"mailbox_aliases_limit"] and
(len(tpl) == 3 or tpl[3] == user.group)
]
limits = user.limitspool.limit_set.filter(name__in=names, maxvalue__gt=0)