mirror of
https://github.com/jwinterm/modoboa.git
synced 2026-10-17 05:41:25 +02:00
@@ -1,8 +1,12 @@
|
||||
"""Admin API."""
|
||||
|
||||
from django.contrib.contenttypes.models import ContentType
|
||||
|
||||
from rest_framework import viewsets
|
||||
from rest_framework.permissions import DjangoModelPermissions
|
||||
|
||||
from modoboa.core import models as core_models
|
||||
|
||||
from . import models
|
||||
from . import serializers
|
||||
|
||||
@@ -20,3 +24,23 @@ 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.CreateAccountSerializer
|
||||
|
||||
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)
|
||||
|
||||
@@ -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":
|
||||
|
||||
@@ -160,7 +160,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
|
||||
|
||||
|
||||
|
||||
@@ -250,7 +250,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(
|
||||
|
||||
@@ -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
|
||||
|
||||
from . import models
|
||||
|
||||
@@ -12,3 +21,89 @@ class DomainSerializer(serializers.ModelSerializer):
|
||||
class Meta:
|
||||
model = models.Domain
|
||||
fields = ("pk", "name", "quota", "enabled", "type", )
|
||||
|
||||
|
||||
class MailboxSerializer(serializers.ModelSerializer):
|
||||
"""Base mailbox serializer."""
|
||||
|
||||
full_address = serializers.EmailField()
|
||||
|
||||
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 CreateAccountSerializer(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(CreateAccountSerializer, 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.")
|
||||
})
|
||||
return data
|
||||
|
||||
def create(self, validated_data):
|
||||
"""Call the set_password method."""
|
||||
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()
|
||||
user.role = role
|
||||
if mailbox_data:
|
||||
address, domain_name = email_utils.split_mailbox(
|
||||
mailbox_data.pop("full_address"))
|
||||
domain = get_object_or_404(
|
||||
admin_models.Domain, name=domain_name)
|
||||
admin_models.Mailbox.objects.create(
|
||||
user=user, address=address, domain=domain, **mailbox_data)
|
||||
return user
|
||||
|
||||
@@ -1,5 +1,7 @@
|
||||
"""Admin API related tests."""
|
||||
|
||||
import json
|
||||
|
||||
from django.core.urlresolvers import reverse
|
||||
|
||||
from rest_framework.authtoken.models import Token
|
||||
@@ -77,3 +79,66 @@ class APITestCase(ModoAPITestCase):
|
||||
response = self.client.delete(url)
|
||||
self.assertEqual(response.status_code, 204)
|
||||
self.assertFalse(models.Domain.objects.filter(pk=domain.pk).exists())
|
||||
|
||||
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."""
|
||||
data = {
|
||||
"username": "fromapi@test.com",
|
||||
"role": "SimpleUsers",
|
||||
"password": "Toto1234",
|
||||
"mailbox": {
|
||||
"full_address": "fromapi@test.com",
|
||||
"quota": 10
|
||||
}
|
||||
}
|
||||
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)
|
||||
|
||||
def test_create_account_bad_password(self):
|
||||
"""Try to create a new account."""
|
||||
data = {
|
||||
"username": "fromapi@test.com",
|
||||
"role": "SimpleUsers",
|
||||
"password": "toto", # Bad password
|
||||
"mailbox": {
|
||||
"full_address": "fromapi@test.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("password", errors)
|
||||
|
||||
def test_create_account_bad_master_user(self):
|
||||
"""Try to create a new account."""
|
||||
data = {
|
||||
"username": "fromapi@test.com",
|
||||
"role": "SimpleUsers",
|
||||
"master_user": True,
|
||||
"password": "Toto1234",
|
||||
"mailbox": {
|
||||
"full_address": "fromapi@test.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("master_user", errors)
|
||||
|
||||
@@ -7,5 +7,6 @@ from . import api
|
||||
|
||||
router = routers.SimpleRouter()
|
||||
router.register(r"domains", api.DomainViewSet, base_name="domain")
|
||||
router.register(r"accounts", api.AccountViewSet, base_name="account")
|
||||
|
||||
urlpatterns = router.urls
|
||||
|
||||
@@ -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"]
|
||||
]
|
||||
}
|
||||
|
||||
|
||||
|
||||
+15
-3
@@ -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
|
||||
@@ -407,7 +419,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 +461,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)
|
||||
|
||||
@@ -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):
|
||||
|
||||
@@ -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)
|
||||
|
||||
Reference in New Issue
Block a user