mirror of
https://github.com/jwinterm/modoboa.git
synced 2026-10-17 05:41:25 +02:00
899 lines
30 KiB
Python
899 lines
30 KiB
Python
# coding: utf-8
|
|
from django.db import models, IntegrityError
|
|
from django.contrib.contenttypes.models import ContentType
|
|
from django.contrib.contenttypes import generic
|
|
from django.contrib.auth.models import User as DUser, UserManager, Group
|
|
from django.utils.translation import ugettext as _, ugettext_lazy
|
|
from django.utils.crypto import constant_time_compare
|
|
from django.conf import settings
|
|
from modoboa.lib import parameters, events
|
|
from modoboa.lib.exceptions import PermDeniedException
|
|
from modoboa.lib.sysutils import exec_cmd, exec_as_vuser
|
|
from modoboa.lib.emailutils import split_mailbox
|
|
from modoboa.extensions import exts_pool
|
|
from exceptions import *
|
|
import os, sys
|
|
import pwd
|
|
import re
|
|
import crypt, hashlib, string, base64
|
|
from random import Random
|
|
|
|
try:
|
|
from modoboa.lib.ldaputils import *
|
|
ldap_available = True
|
|
except ImportError:
|
|
ldap_available = False
|
|
|
|
class User(DUser):
|
|
"""Proxy for the ``User`` model.
|
|
|
|
It overloads the way passwords are stored into the database. The
|
|
main reason to change this mechanism is to ensure the
|
|
compatibility with the way Dovecot stores passwords.
|
|
|
|
It also adds new attributes and methods.
|
|
"""
|
|
class Meta:
|
|
proxy = True
|
|
ordering = ["username"]
|
|
|
|
password_expr = re.compile(r'\{(\w+)\}(.+)')
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
"""Constructor
|
|
|
|
A little hack to increase the maximum length for the username
|
|
field.
|
|
"""
|
|
max_length = 254
|
|
DUser._meta.get_field('username').max_length = max_length
|
|
DUser._meta.get_field('username').validators[0].limit_value = 254
|
|
super(User, self).__init__(*args, **kwargs)
|
|
|
|
def __unicode__(self):
|
|
return u"%s %s" % (self.first_name, self.last_name)
|
|
|
|
def delete(self, fromuser, keep_mb_dir, *args, **kwargs):
|
|
"""Custom delete method
|
|
|
|
To check permissions properly, we need to make a distinction
|
|
between 2 cases:
|
|
|
|
* If the user owns a mailbox, the check is made on that object
|
|
(useful for domain admins)
|
|
|
|
* Otherwise, the check is made on the user
|
|
"""
|
|
from modoboa.lib.permissions import \
|
|
get_object_owner, grant_access_to_object, ungrant_access_to_object
|
|
|
|
if fromuser == self:
|
|
raise AdminError(_("You can't delete your own account"))
|
|
|
|
if not fromuser.can_access(self):
|
|
raise PermDeniedException
|
|
|
|
if self.has_mailbox:
|
|
mb = self.mailbox_set.all()[0]
|
|
if not fromuser.can_access(mb):
|
|
raise PermDeniedException
|
|
mb.delete(keepdir=keep_mb_dir)
|
|
|
|
owner = get_object_owner(self)
|
|
for ooentry in self.objectaccess_set.filter(is_owner=True):
|
|
if ooentry.content_object is not None:
|
|
grant_access_to_object(owner, ooentry.content_object, True)
|
|
|
|
events.raiseEvent("AccountDeleted", self)
|
|
ungrant_access_to_object(self)
|
|
super(User, self).delete(*args, **kwargs)
|
|
|
|
@staticmethod
|
|
def get_content_type():
|
|
"""An uggly hack to retrieve the appropriate content_type!
|
|
|
|
The explanation is available here:
|
|
https://code.djangoproject.com/ticket/11154
|
|
|
|
Quickly, the content_types framework does not retrieve the
|
|
appropriate content type for proxy models, it retrieves the
|
|
one of the first parent that is not a proxy.
|
|
"""
|
|
if not hasattr(User, "ct"):
|
|
User.ct = ContentType.objects.get(app_label="admin", model="user")
|
|
return User.ct
|
|
|
|
def _crypt_password(self, raw_value):
|
|
scheme = parameters.get_admin("PASSWORD_SCHEME")
|
|
if scheme == "crypt":
|
|
salt = ''.join(Random().sample(string.letters + string.digits, 2))
|
|
result = crypt.crypt(raw_value, salt)
|
|
elif scheme == "md5":
|
|
obj = hashlib.md5(raw_value)
|
|
result = obj.hexdigest()
|
|
elif scheme == "sha256":
|
|
obj = hashlib.sha256(raw_value)
|
|
result = base64.b64encode(obj.digest())
|
|
else:
|
|
scheme = "plain"
|
|
result = raw_value
|
|
return "{%s}%s" % (scheme.upper(), result)
|
|
|
|
def set_password(self, raw_value, curvalue=None):
|
|
"""Password update
|
|
|
|
Update the current mailbox's password with the given clear
|
|
value. This value is encrypted according to the defined method
|
|
before it is saved.
|
|
|
|
:param raw_value: the new password's value
|
|
:param curvalue: the current password (for LDAP authentication)
|
|
"""
|
|
if parameters.get_admin("AUTHENTICATION_TYPE") == "local":
|
|
self.password = self._crypt_password(raw_value)
|
|
return
|
|
|
|
if not ldap_available:
|
|
raise AdminError(_("Failed to update password: LDAP module not installed"))
|
|
|
|
ab = LDAPAuthBackend()
|
|
try:
|
|
ab.update_user_password(self.username, curvalue, raw_value)
|
|
except LDAPException, e:
|
|
raise AdminError(_("Failed to update password: %s" % str(e)))
|
|
|
|
def check_password(self, raw_value):
|
|
m = self.password_expr.match(self.password)
|
|
if m is None:
|
|
return False
|
|
scheme = m.group(1).lower()
|
|
val2 = m.group(2)
|
|
if scheme == "crypt":
|
|
val1 = crypt.crypt(raw_value, m.group(2))
|
|
elif scheme == "md5":
|
|
val1 = hashlib.md5(raw_value).hexdigest()
|
|
elif scheme == "sha256":
|
|
val1 = base64.b64encode(hashlib.sha256(raw_value).digest())
|
|
else:
|
|
val1 = raw_value
|
|
return constant_time_compare(val1, val2)
|
|
|
|
@property
|
|
def has_mailbox(self):
|
|
return len(self.mailbox_set.all()) != 0
|
|
|
|
@property
|
|
def fullname(self):
|
|
if self.first_name != "":
|
|
return "%s %s" % (self.first_name, self.last_name)
|
|
return self.username
|
|
|
|
@property
|
|
def identity(self):
|
|
return self.username
|
|
|
|
@property
|
|
def name_or_rcpt(self):
|
|
if self.first_name != "":
|
|
return "%s %s" % (self.first_name, self.last_name)
|
|
return "----"
|
|
|
|
@property
|
|
def group(self):
|
|
if self.is_superuser:
|
|
return "SuperAdmins"
|
|
try:
|
|
return self.groups.all()[0].name
|
|
except IndexError:
|
|
return "?"
|
|
|
|
def belongs_to_group(self, name):
|
|
"""Simple shortcut to check if this user is a member of a
|
|
specific group.
|
|
|
|
:param name: the group's name
|
|
:return: a boolean
|
|
"""
|
|
try:
|
|
self.groups.get(name=name)
|
|
except Group.DoesNotExist:
|
|
return False
|
|
return True
|
|
|
|
def get_identities(self):
|
|
from modoboa.lib.permissions import get_content_type
|
|
|
|
userct = get_content_type(self)
|
|
alct = get_content_type(Alias)
|
|
return self.objectaccess_set.filter(content_type__in=[userct, alct])
|
|
|
|
def get_domains(self):
|
|
"""Return the domains belonging to this user
|
|
|
|
The result is a ``QuerySet`` object, so this function can be used
|
|
to fill ``ModelChoiceField`` objects.
|
|
"""
|
|
if self.is_superuser:
|
|
return Domain.objects.all()
|
|
return Domain.objects.filter(owners__user=self)
|
|
|
|
def get_domaliases(self):
|
|
"""Return the domain aliases that belong to this user
|
|
|
|
The result will contain the domain aliases defined for each domain
|
|
that user can see.
|
|
|
|
:return: a list of ``DomainAlias`` objects
|
|
"""
|
|
if self.is_superuser:
|
|
return DomainAlias.objects.all()
|
|
domains = self.get_domains()
|
|
domaliases = []
|
|
for dom in domains:
|
|
domaliases += dom.domainalias_set.all()
|
|
return domaliases
|
|
|
|
def get_mailboxes(self):
|
|
"""Return the mailboxes that belong to this user
|
|
|
|
The result will contain the mailboxes defined for each domain that
|
|
user can see.
|
|
|
|
:return: a list of ``Mailbox`` objects
|
|
"""
|
|
if self.is_superuser:
|
|
return Mailbox.objects.all()
|
|
mboxes = []
|
|
domains = self.get_domains()
|
|
for dom in domains:
|
|
mboxes += dom.mailbox_set.all()
|
|
return mboxes
|
|
|
|
def get_mbaliases(self):
|
|
"""Return the mailbox aliases that belong to this user
|
|
|
|
The result will contain the mailbox aliases defined for each
|
|
domain that user can see.
|
|
|
|
:return: a list of ``Alias`` objects
|
|
"""
|
|
if self.is_superuser:
|
|
return Alias.objects.all()
|
|
domains = self.get_domains()
|
|
mbaliases = []
|
|
for dom in domains:
|
|
mbaliases += dom.alias_set.all()
|
|
return mbaliases
|
|
|
|
def is_owner(self, obj):
|
|
"""Tell is the user is the unique owner of this object
|
|
|
|
:param obj: an object inheriting from ``models.Model``
|
|
:return: a boolean
|
|
"""
|
|
from modoboa.lib.permissions import get_content_type
|
|
|
|
ct = get_content_type(obj)
|
|
try:
|
|
ooentry = self.objectaccess_set.get(content_type=ct, object_id=obj.id)
|
|
except ObjectAccess.DoesNotExist:
|
|
return False
|
|
return ooentry.is_owner
|
|
|
|
def can_access(self, obj):
|
|
"""Check if the user can access a specific object
|
|
|
|
This function is recursive : if the given user hasn't got direct
|
|
access to this object and if he has got access other ``User``
|
|
objects, we check if one of those users owns the object.
|
|
|
|
:param obj: a admin object
|
|
:return: a boolean
|
|
"""
|
|
from modoboa.lib.permissions import get_content_type
|
|
|
|
if self.is_superuser:
|
|
return True
|
|
|
|
ct = get_content_type(obj)
|
|
try:
|
|
ooentry = self.objectaccess_set.get(content_type=ct, object_id=obj.id)
|
|
except ObjectAccess.DoesNotExist:
|
|
pass
|
|
else:
|
|
return True
|
|
if ct.model == "user":
|
|
return False
|
|
|
|
ct = self.get_content_type()
|
|
qs = self.objectaccess_set.filter(content_type=ct)
|
|
for ooentry in qs.all():
|
|
if ooentry.content_object.is_owner(obj):
|
|
return True
|
|
return False
|
|
|
|
class ObjectAccess(models.Model):
|
|
user = models.ForeignKey(User)
|
|
content_type = models.ForeignKey(ContentType)
|
|
object_id = models.PositiveIntegerField()
|
|
content_object = generic.GenericForeignKey('content_type', 'object_id')
|
|
is_owner = models.BooleanField(default=False)
|
|
|
|
class Meta:
|
|
unique_together = (("user", "content_type", "object_id"),)
|
|
|
|
def __unicode__(self):
|
|
return "%s => %s (%s)" % (self.user, self.content_object, self.content_type)
|
|
|
|
class ObjectDates(models.Model):
|
|
"""Dates recording for admin objects
|
|
|
|
This table keeps creation and last modification dates for Domains,
|
|
domain aliases, mailboxes and aliases objects.
|
|
"""
|
|
creation = models.DateTimeField(auto_now_add=True)
|
|
last_modification = models.DateTimeField(auto_now=True)
|
|
|
|
@staticmethod
|
|
def set_for_object(obj):
|
|
"""Initialize or update dates for a given object.
|
|
|
|
:param obj: an admin object (Domain, Mailbox, etc)
|
|
"""
|
|
try:
|
|
dates = getattr(obj, "dates")
|
|
except ObjectDates.DoesNotExist:
|
|
dates = ObjectDates()
|
|
dates.save()
|
|
obj.dates = dates
|
|
|
|
class DatesAware(models.Model):
|
|
"""Abstract model to support dates
|
|
|
|
Inherit from this model to automatically add the "dates" feature
|
|
to another model. It defines the appropriate field and handles
|
|
saves.
|
|
"""
|
|
dates = models.ForeignKey(ObjectDates)
|
|
|
|
class Meta:
|
|
abstract = True
|
|
|
|
def save(self, *args, **kwargs):
|
|
ObjectDates.set_for_object(self)
|
|
super(DatesAware, self).save(*args, **kwargs)
|
|
|
|
@property
|
|
def creation(self):
|
|
return self.dates.creation
|
|
|
|
@property
|
|
def last_modification(self):
|
|
return self.dates.last_modification
|
|
|
|
class Domain(DatesAware):
|
|
name = models.CharField(ugettext_lazy('name'), max_length=100, unique=True,
|
|
help_text=ugettext_lazy("The domain name"))
|
|
quota = models.IntegerField(help_text=ugettext_lazy("Default quota in MB applied to mailboxes"))
|
|
enabled = models.BooleanField(ugettext_lazy('enabled'),
|
|
help_text=ugettext_lazy("Check to activate this domain"))
|
|
owners = generic.GenericRelation(ObjectAccess)
|
|
|
|
class Meta:
|
|
permissions = (
|
|
("view_domain", "View domain"),
|
|
("view_domains", "View domains"),
|
|
)
|
|
ordering = ["name"]
|
|
|
|
@property
|
|
def domainalias_count(self):
|
|
return len(self.domainalias_set.all())
|
|
|
|
@property
|
|
def mailbox_count(self):
|
|
return len(self.mailbox_set.all())
|
|
|
|
@property
|
|
def mbalias_count(self):
|
|
return len(self.alias_set.all())
|
|
|
|
@property
|
|
def admins(self):
|
|
return [oa.user for oa in self.owners.filter(user__is_superuser=False)]
|
|
|
|
def create_dir(self):
|
|
if parameters.get_admin("CREATE_DIRECTORIES") == "yes":
|
|
path = "%s/%s" % (parameters.get_admin("STORAGE_PATH"), self.name)
|
|
return exec_as_vuser("mkdir -p %s" % path)
|
|
|
|
return True
|
|
|
|
def rename_dir(self, oldname):
|
|
if parameters.get_admin("CREATE_DIRECTORIES") == "yes":
|
|
stpath = parameters.get_admin("STORAGE_PATH")
|
|
return exec_as_vuser("mv %s/%s %s/%s" \
|
|
% (stpath, oldname, stpath, self.name))
|
|
return True
|
|
|
|
def delete_dir(self):
|
|
if parameters.get_admin("CREATE_DIRECTORIES") == "yes":
|
|
return exec_as_vuser("rm -r %s/%s" \
|
|
% (parameters.get_admin("STORAGE_PATH"),
|
|
self.name))
|
|
return True
|
|
|
|
def delete(self, keepdir=False):
|
|
from modoboa.lib.permissions import ungrant_access_to_object, ungrant_access_to_objects
|
|
|
|
if len(self.domainalias_set.all()):
|
|
events.raiseEvent("DomainAliasDeleted", self.domainalias_set.all())
|
|
ungrant_access_to_objects(self.domainalias_set.all())
|
|
if len(self.mailbox_set.all()):
|
|
events.raiseEvent("DeleteMailbox", self.mailbox_set.all())
|
|
ungrant_access_to_objects(self.mailbox_set.all())
|
|
if len(self.alias_set.all()):
|
|
events.raiseEvent("MailboxAliasDelete", self.alias_set.all())
|
|
ungrant_access_to_objects(self.alias_set.all())
|
|
events.raiseEvent("DeleteDomain", self)
|
|
ungrant_access_to_object(self)
|
|
super(Domain, self).delete()
|
|
if not keepdir:
|
|
self.delete_dir()
|
|
|
|
def save(self, *args, **kwargs):
|
|
if not self.create_dir():
|
|
raise AdminError(_("Failed to initialise domain, check permissions"))
|
|
super(Domain, self).save(*args, **kwargs)
|
|
|
|
def __str__(self):
|
|
return self.name
|
|
|
|
def create_from_csv(self, user, row):
|
|
if len(row) < 3:
|
|
raise AdminError(_("Invalid line"))
|
|
self.name = row[1].strip()
|
|
self.quota = int(row[2].strip())
|
|
self.enabled = True
|
|
self.save()
|
|
|
|
class DomainAlias(DatesAware):
|
|
name = models.CharField(ugettext_lazy("name"), max_length=100, unique=True,
|
|
help_text=ugettext_lazy("The alias name"))
|
|
target = models.ForeignKey(Domain, verbose_name=ugettext_lazy('target'),
|
|
help_text=ugettext_lazy("The domain this alias points to"))
|
|
enabled = models.BooleanField(
|
|
ugettext_lazy('enabled'),
|
|
help_text=ugettext_lazy("Check to activate this alias")
|
|
)
|
|
|
|
class Meta:
|
|
permissions = (
|
|
("view_domaliases", "View domain aliases"),
|
|
)
|
|
|
|
def __unicode__(self):
|
|
return self.name
|
|
|
|
def delete(self):
|
|
from modoboa.lib.permissions import ungrant_access_to_object
|
|
events.raiseEvent("DomainAliasDeleted", self)
|
|
ungrant_access_to_object(self)
|
|
super(DomainAlias, self).delete()
|
|
|
|
class Mailbox(DatesAware):
|
|
address = models.CharField(
|
|
ugettext_lazy('address'), max_length=100,
|
|
help_text=ugettext_lazy("Mailbox address (without the @domain.tld part)")
|
|
)
|
|
quota = models.IntegerField()
|
|
uid = models.IntegerField()
|
|
gid = models.IntegerField()
|
|
path = models.CharField(max_length=255)
|
|
domain = models.ForeignKey(Domain)
|
|
user = models.ForeignKey(User)
|
|
|
|
class Meta:
|
|
permissions = (
|
|
("view_mailboxes", "View mailboxes"),
|
|
)
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
super(Mailbox, self).__init__(*args, **kwargs)
|
|
self.mbtype = parameters.get_admin("MAILBOX_TYPE")
|
|
self.mdirroot = parameters.get_admin("MAILDIR_ROOT")
|
|
if not self.mdirroot.endswith('/'):
|
|
self.mdirroot += '/'
|
|
|
|
def __str__(self):
|
|
return self.full_address
|
|
|
|
@property
|
|
def full_address(self):
|
|
return "%s@%s" % (self.address, self.domain.name)
|
|
|
|
@property
|
|
def enabled(self):
|
|
return self.user.is_active
|
|
|
|
@property
|
|
def full_path(self):
|
|
path = os.path.join(parameters.get_admin("STORAGE_PATH"),
|
|
self.domain.name, self.address)
|
|
return os.path.abspath(path)
|
|
|
|
@property
|
|
def alias_count(self):
|
|
return len(self.alias_set.all())
|
|
|
|
def create_dir(self):
|
|
if parameters.get_admin("CREATE_DIRECTORIES") == "yes":
|
|
relpath = "%s/%s" % (self.domain.name, self.address)
|
|
abspath = os.path.join(parameters.get_admin("STORAGE_PATH"), relpath)
|
|
if self.mbtype == "mbox":
|
|
self.path = "%s/Inbox" % self.address
|
|
else:
|
|
self.path = "%s/" % self.address
|
|
if os.path.exists(abspath):
|
|
return True
|
|
if not exec_as_vuser("mkdir -p %s" % abspath):
|
|
return False
|
|
if self.mbtype == "mbox":
|
|
template = ["Inbox", "Drafts", "Sent", "Trash", "Junk"]
|
|
for f in template:
|
|
exec_as_vuser("touch %s/%s" % (abspath, f))
|
|
else:
|
|
template = [".Drafts/", ".Sent/", ".Trash/", ".Junk/"]
|
|
for dir in template:
|
|
for sdir in ["cur", "new", "tmp"]:
|
|
exec_as_vuser("mkdir -p %s/%s/%s/%s" \
|
|
% (abspath, self.mdirroot, dir, sdir))
|
|
return True
|
|
|
|
def rename_dir(self, newdomain, newaddress):
|
|
if parameters.get_admin("CREATE_DIRECTORIES") == "yes":
|
|
if self.domain.name == newdomain and newaddress == self.address:
|
|
return True
|
|
oldpath = "%s/%s" % (parameters.get_admin("STORAGE_PATH"), self.domain.name)
|
|
newpath = "%s/%s" % (parameters.get_admin("STORAGE_PATH"), newdomain)
|
|
oldpath = os.path.join(oldpath, self.address)
|
|
newpath = os.path.join(newpath, newaddress)
|
|
code = exec_as_vuser("mv %s %s" % (oldpath, newpath))
|
|
if code:
|
|
self.path = "%s/" % newaddress
|
|
return code
|
|
|
|
return True
|
|
|
|
def delete_dir(self):
|
|
if parameters.get_admin("CREATE_DIRECTORIES") == "yes":
|
|
return exec_as_vuser("rm -r %s/%s/%s" \
|
|
% (parameters.get_admin("STORAGE_PATH"),
|
|
self.domain.name, self.address))
|
|
return True
|
|
|
|
def set_ownership(self):
|
|
# If we ever had numerical uid, don't resolve it
|
|
v_uid = parameters.get_admin("VIRTUAL_UID");
|
|
if v_uid.isdigit():
|
|
self.uid = v_uid
|
|
else:
|
|
try:
|
|
self.uid = pwd.getpwnam(v_uid).pw_uid
|
|
except KeyError:
|
|
raise AdminError(_("%s is not a valid uid/username" % v_uid))
|
|
# Just the same for gid
|
|
v_gid = parameters.get_admin("VIRTUAL_GID")
|
|
if v_gid.isdigit():
|
|
self.gid = v_gid
|
|
else:
|
|
try:
|
|
self.gid = pwd.getpwnam(v_gid).pw_gid
|
|
except KeyError:
|
|
raise AdminError(_("%s is not a valid gid/groupname" % v_gid))
|
|
|
|
def save(self, *args, **kwargs):
|
|
if not self.create_dir():
|
|
raise AdminError(_("Failed to initialise mailbox, check permissions"))
|
|
try:
|
|
user = getattr(self, "user")
|
|
except User.DoesNotExist:
|
|
user = User()
|
|
user.email = "%s@%s" % (self.address, self.domain.name)
|
|
if kwargs.has_key("username"):
|
|
user.username = kwargs["username"]
|
|
else:
|
|
user.username = user.email
|
|
if kwargs.has_key("enabled"):
|
|
user.is_active = kwargs["enabled"]
|
|
if kwargs.has_key("name"):
|
|
try:
|
|
fname, lname = kwargs["name"].split()
|
|
except ValueError:
|
|
fname = kwargs["name"]
|
|
lname = ""
|
|
user.first_name = fname
|
|
user.last_name = lname
|
|
if kwargs.has_key("password"):
|
|
user.set_password(kwargs["password"])
|
|
try:
|
|
user.save()
|
|
except IntegrityError, e:
|
|
raise AdminError(_("Account '%s' already exists" % user.username))
|
|
self.user = user
|
|
|
|
if len(user.groups.all()) == 0:
|
|
user.groups.add(Group.objects.get(name="SimpleUsers"))
|
|
user.save()
|
|
|
|
self.set_ownership()
|
|
|
|
if kwargs.has_key("quota") and kwargs["quota"] is not None \
|
|
and int(kwargs["quota"]) < self.domain.quota:
|
|
self.quota = kwargs["quota"]
|
|
else:
|
|
self.quota = self.domain.quota
|
|
for kw in ["username", "name", "enabled", "password", "quota"]:
|
|
try:
|
|
del kwargs[kw]
|
|
except KeyError:
|
|
pass
|
|
super(Mailbox, self).save(*args, **kwargs)
|
|
|
|
def save_from_user(self, localpart, domain, user, quota=None):
|
|
"""Simple save method called for automatic creations
|
|
|
|
:param localpart: the mailbox
|
|
:param domain: the associated Domain object
|
|
:param user: the associated User object
|
|
"""
|
|
self.address = localpart
|
|
self.domain = domain
|
|
self.user = user
|
|
if not self.create_dir():
|
|
raise AdminError(_("Failed to initialise mailbox, check permissions"))
|
|
self.set_ownership()
|
|
self.quota = self.domain.quota if quota is None else quota
|
|
super(Mailbox, self).save()
|
|
|
|
def create_from_csv(self, user, line):
|
|
"""Create a new mailbox from a CSV file entry
|
|
|
|
The expected order is the following::
|
|
|
|
[loginname, password, first name, last name, address]
|
|
|
|
:param line: a list containing the expected information
|
|
"""
|
|
if len(line) < 6:
|
|
raise AdminError(_("Invalid line"))
|
|
mailbox, domname = split_mailbox(line[5].strip())
|
|
self.address = mailbox
|
|
try:
|
|
self.domain = Domain.objects.get(name=domname)
|
|
except Domain.DoesNotExist:
|
|
raise AdminError(_("Cannot import this mailbox because the associated domain does not exist"))
|
|
if not user.can_access(self.domain):
|
|
raise PermDeniedException
|
|
|
|
name = "%s %s" % (line[3].strip(), line[4].strip())
|
|
self.save(username=line[1].strip(), name=name,
|
|
password=line[2].strip(), enabled=True)
|
|
|
|
def delete(self, keepdir=False):
|
|
from modoboa.lib.permissions import ungrant_access_to_object
|
|
|
|
events.raiseEvent("DeleteMailbox", self)
|
|
ungrant_access_to_object(self)
|
|
for alias in self.alias_set.all():
|
|
alias.mboxes.remove(self)
|
|
if len(alias.mboxes.all()) == 0:
|
|
alias.delete()
|
|
super(Mailbox, self).delete()
|
|
if not keepdir:
|
|
self.delete_dir()
|
|
|
|
|
|
class Alias(DatesAware):
|
|
address = models.CharField(
|
|
ugettext_lazy('address'), max_length=254,
|
|
help_text=ugettext_lazy("The alias address (without the domain part). For a 'catch-all' address, just enter an * character.")
|
|
)
|
|
domain = models.ForeignKey(Domain)
|
|
mboxes = models.ManyToManyField(
|
|
Mailbox, verbose_name=ugettext_lazy('mailboxes'),
|
|
help_text=ugettext_lazy("The mailboxes this alias points to")
|
|
)
|
|
extmboxes = models.TextField(blank=True)
|
|
enabled = models.BooleanField(ugettext_lazy('enabled'),
|
|
help_text=ugettext_lazy("Check to activate this alias"))
|
|
|
|
class Meta:
|
|
permissions = (
|
|
("view_aliases", "View aliases"),
|
|
)
|
|
unique_together = (("address", "domain"),)
|
|
ordering = ["domain__name", "address"]
|
|
|
|
@property
|
|
def full_address(self):
|
|
return "%s@%s" % (self.address, self.domain.name)
|
|
|
|
@property
|
|
def identity(self):
|
|
return self.full_address
|
|
|
|
@property
|
|
def name_or_rcpt(self):
|
|
rcpts = self.get_recipients()
|
|
if len(rcpts) > 1:
|
|
return "%s, ..." % rcpts[0]
|
|
return rcpts[0]
|
|
|
|
def save(self, int_rcpts, ext_rcpts, *args, **kwargs):
|
|
if len(ext_rcpts):
|
|
self.extmboxes = ",".join(ext_rcpts)
|
|
else:
|
|
self.extmboxes = ""
|
|
super(Alias, self).save(*args, **kwargs)
|
|
curmboxes = self.mboxes.all()
|
|
for t in int_rcpts:
|
|
if not t in curmboxes:
|
|
self.mboxes.add(t)
|
|
for t in curmboxes:
|
|
if not t in int_rcpts:
|
|
self.mboxes.remove(t)
|
|
|
|
def delete(self):
|
|
from modoboa.lib.permissions import ungrant_access_to_object
|
|
|
|
events.raiseEvent("MailboxAliasDeleted", self)
|
|
ungrant_access_to_object(self)
|
|
super(Alias, self).delete()
|
|
|
|
def get_recipients(self):
|
|
"""Return the recipients list
|
|
|
|
Internal and external addresses are mixed into a single list.
|
|
"""
|
|
result = map(lambda mb: mb.full_address, self.mboxes.all())
|
|
if self.extmboxes != "":
|
|
result += self.extmboxes.split(',')
|
|
return result
|
|
|
|
def ui_disabled(self, user):
|
|
if user.is_superuser:
|
|
return False
|
|
for mb in self.mboxes.all():
|
|
if not user.is_owner(mb.domain):
|
|
return True
|
|
return False
|
|
|
|
def create_from_csv(self, user, row):
|
|
if len(row) < 4:
|
|
raise AdminError(_("Invalid line"))
|
|
localpart, domname = split_mailbox(row[1].strip())
|
|
try:
|
|
domain = Domain.objects.get(name=domname)
|
|
except Domain.DoesNotExist:
|
|
raise AdminError(_("Domain does not exist"))
|
|
if not user.can_access(domain):
|
|
raise PermDeniedException
|
|
self.address = localpart
|
|
self.domain = domain
|
|
self.enabled = True
|
|
int_rcpts = []
|
|
ext_rcpts = []
|
|
for rcpt in row[2:]:
|
|
rcpt = rcpt.strip()
|
|
localpart, domname = split_mailbox(rcpt)
|
|
try:
|
|
Domain.objects.get(name=domname)
|
|
except Domain.DoesNotExist:
|
|
ext_rcpts += [rcpt]
|
|
continue
|
|
try:
|
|
int_rcpts += [Mailbox.objects.get(address=localpart,
|
|
domain__name=domname)]
|
|
except Mailbox.DoesNotExist:
|
|
raise AdminError(_("Mailbox %s does not exist" % rcpt))
|
|
self.save(int_rcpts, ext_rcpts)
|
|
|
|
class Extension(models.Model):
|
|
name = models.CharField(max_length=150)
|
|
enabled = models.BooleanField(
|
|
ugettext_lazy('enabled'),
|
|
help_text=ugettext_lazy("Check to enable this extension")
|
|
)
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
super(Extension, self).__init__(*args, **kwargs)
|
|
|
|
def __get_ext_instance(self):
|
|
if not self.name:
|
|
return None
|
|
if hasattr(self, "instance") and self.instance:
|
|
return
|
|
self.instance = exts_pool.get_extension(self.name)
|
|
if self.instance:
|
|
self.__get_ext_dir()
|
|
|
|
def __get_ext_dir(self):
|
|
modname = self.instance.__module__
|
|
path = os.path.realpath(sys.modules[modname].__file__)
|
|
self.extdir = os.path.dirname(path)
|
|
|
|
def on(self):
|
|
self.enabled = True
|
|
self.save()
|
|
|
|
self.__get_ext_instance()
|
|
self.instance.load()
|
|
self.instance.init()
|
|
|
|
if self.instance.needs_media:
|
|
path = os.path.join(settings.MEDIA_ROOT, self.name)
|
|
exec_cmd("mkdir %s" % path)
|
|
|
|
scriptdir = "%s/%s" % (self.extdir, "scripts")
|
|
if os.path.exists(scriptdir):
|
|
targetdir = os.path.join(settings.MODOBOA_DIR, "scripts")
|
|
exec_cmd("ln -s %s %s/%s" % (scriptdir, targetdir, self.name))
|
|
|
|
events.raiseEvent("ExtEnabled", self)
|
|
|
|
def off(self):
|
|
self.__get_ext_instance()
|
|
self.instance.destroy()
|
|
|
|
self.enabled = False
|
|
self.save()
|
|
|
|
if self.instance.needs_media:
|
|
path = os.path.join(settings.MEDIA_ROOT, self.name)
|
|
exec_cmd("rm -r %s" % path)
|
|
|
|
scriptdir = "%s/%s" % (self.extdir, "scripts")
|
|
if os.path.exists(scriptdir):
|
|
target = "%s/scripts/%s" % (settings.MODOBOA_DIR, self.name)
|
|
exec_cmd("rm %s" % target)
|
|
|
|
events.raiseEvent("ExtDisabled", self)
|
|
|
|
def populate_callback(sender, user=None, **kwargs):
|
|
"""Populate signal callback
|
|
|
|
If the LDAP authentication backend is in use, this callback will
|
|
be called each time a new user authenticates succesfuly to
|
|
Modoboa. This function is in charge of creating the mailbox
|
|
associated to the provided ``User`` object.
|
|
|
|
:param sender: ??
|
|
:param user: a ``User`` instance
|
|
"""
|
|
if parameters.get_admin("AUTHENTICATION_TYPE") != "ldap":
|
|
return
|
|
if user is None:
|
|
return
|
|
localpart, domname = split_mailbox(user.username)
|
|
try:
|
|
domain = Domain.objects.get(name=domname)
|
|
except Domain.DoesNotExist:
|
|
domain = Domain()
|
|
domain.name = domname
|
|
domain.enabled = True
|
|
domain.quota = 0
|
|
domain.save()
|
|
try:
|
|
mb = Mailbox.objects.get(domain=domain, address=localpart)
|
|
except Mailbox.DoesNotExist:
|
|
mb = Mailbox()
|
|
mb.save_from_user(localpart, domain, user)
|
|
|
|
try:
|
|
from django_auth_ldap.backend import populate_user
|
|
except ImportError, inst:
|
|
pass
|
|
else:
|
|
populate_user.connect(populate_callback, dispatch_uid="myuid")
|