mirror of
https://github.com/jwinterm/modoboa.git
synced 2026-10-17 05:41:25 +02:00
WIP: new policy daemon.
This commit is contained in:
Antoine Nguyen
@@ -27,6 +27,7 @@ cache: pip
|
||||
services:
|
||||
- postgres
|
||||
- mysql
|
||||
- redis-server
|
||||
|
||||
before_install:
|
||||
- pip install codecov
|
||||
|
||||
@@ -238,6 +238,13 @@ class AccountFormMail(forms.Form, DynamicForm):
|
||||
widget=forms.widgets.TextInput(attrs={"class": "form-control"})
|
||||
)
|
||||
quota_act = forms.BooleanField(required=False)
|
||||
message_limit = forms.IntegerField(
|
||||
label=ugettext_lazy("Message sending limit"),
|
||||
required=False,
|
||||
min_value=0,
|
||||
help_text=ugettext_lazy(
|
||||
"Number of messages this mailbox can send per day")
|
||||
)
|
||||
aliases = lib_fields.UTF8AndEmptyUserEmailField(
|
||||
label=ugettext_lazy("Alias(es)"),
|
||||
required=False,
|
||||
@@ -261,7 +268,7 @@ class AccountFormMail(forms.Form, DynamicForm):
|
||||
def __init__(self, user, *args, **kwargs):
|
||||
self.mb = kwargs.pop("instance", None)
|
||||
self.user = user
|
||||
super(AccountFormMail, self).__init__(*args, **kwargs)
|
||||
super().__init__(*args, **kwargs)
|
||||
self.field_widths = {
|
||||
"quota": 3
|
||||
}
|
||||
@@ -282,6 +289,8 @@ class AccountFormMail(forms.Form, DynamicForm):
|
||||
self.fields["quota_act"].initial = self.mb.use_domain_quota
|
||||
if not self.mb.use_domain_quota and self.mb.quota:
|
||||
self.fields["quota"].initial = self.mb.quota
|
||||
if self.mb.message_limit:
|
||||
self.fields["message_limit"].initial = self.mb.message_limit
|
||||
self.fields["create_alias_with_old_address"].initial = (
|
||||
param_tools.get_global_parameter(
|
||||
"create_alias_on_mbox_rename")
|
||||
@@ -367,7 +376,9 @@ class AccountFormMail(forms.Form, DynamicForm):
|
||||
self.__class__, context=user, klass=models.Mailbox)
|
||||
self.mb = models.Mailbox(
|
||||
address=self.locpart, domain=self.domain, user=account,
|
||||
use_domain_quota=self.cleaned_data["quota_act"])
|
||||
use_domain_quota=self.cleaned_data["quota_act"],
|
||||
message_limit=self.cleaned_data.get("message_limit")
|
||||
)
|
||||
override_rules = (
|
||||
user.is_superuser or
|
||||
user.has_perm("admin.add_domain") and
|
||||
|
||||
@@ -57,8 +57,8 @@ class DomainFormGeneral(forms.ModelForm, DynamicForm):
|
||||
class Meta:
|
||||
model = Domain
|
||||
fields = (
|
||||
"name", "type", "quota", "default_mailbox_quota", "aliases",
|
||||
"enabled", "enable_dns_checks", "enable_dkim",
|
||||
"name", "type", "quota", "default_mailbox_quota", "message_limit",
|
||||
"aliases", "enabled", "enable_dns_checks", "enable_dkim",
|
||||
"dkim_key_selector", "dkim_key_length"
|
||||
)
|
||||
|
||||
|
||||
@@ -0,0 +1,23 @@
|
||||
# Generated by Django 2.2.12 on 2020-06-02 10:01
|
||||
|
||||
from django.db import migrations, models
|
||||
|
||||
|
||||
class Migration(migrations.Migration):
|
||||
|
||||
dependencies = [
|
||||
('admin', '0015_rename_view_permissions'),
|
||||
]
|
||||
|
||||
operations = [
|
||||
migrations.AddField(
|
||||
model_name='domain',
|
||||
name='message_limit',
|
||||
field=models.PositiveIntegerField(blank=True, help_text='Number of messages this domain can send per day', null=True, verbose_name='Message sending limit'),
|
||||
),
|
||||
migrations.AddField(
|
||||
model_name='mailbox',
|
||||
name='message_limit',
|
||||
field=models.PositiveIntegerField(blank=True, help_text='Number of messages this mailbox can send per day', null=True, verbose_name='Message sending limit'),
|
||||
),
|
||||
]
|
||||
@@ -43,13 +43,13 @@ class AdminObject(models.Model):
|
||||
|
||||
def __init__(self, *args, **kwargs):
|
||||
"""Custom constructor."""
|
||||
super(AdminObject, self).__init__(*args, **kwargs)
|
||||
super().__init__(*args, **kwargs)
|
||||
self._loaded_values = {}
|
||||
|
||||
@classmethod
|
||||
def from_db(cls, db, field_names, values):
|
||||
"""Store loaded values."""
|
||||
instance = super(AdminObject, cls).from_db(db, field_names, values)
|
||||
instance = super().from_db(db, field_names, values)
|
||||
instance._loaded_values = dict(zip(field_names, values))
|
||||
return instance
|
||||
|
||||
|
||||
@@ -6,9 +6,7 @@ from reversion import revisions as reversion
|
||||
|
||||
from django.db import models
|
||||
from django.utils import timezone
|
||||
from django.utils.encoding import (
|
||||
force_text, python_2_unicode_compatible, smart_text
|
||||
)
|
||||
from django.utils.encoding import force_text, smart_text
|
||||
from django.utils.functional import cached_property
|
||||
from django.utils.translation import ugettext as _, ugettext_lazy
|
||||
|
||||
@@ -16,12 +14,13 @@ from modoboa.core import signals as core_signals
|
||||
from modoboa.core.models import User
|
||||
from modoboa.lib.exceptions import BadRequest, Conflict
|
||||
from modoboa.parameters import tools as param_tools
|
||||
|
||||
from .. import constants
|
||||
from .base import AdminObject
|
||||
from . import mixins
|
||||
|
||||
|
||||
@python_2_unicode_compatible
|
||||
class Domain(AdminObject):
|
||||
class Domain(mixins.MessageLimitMixin, AdminObject):
|
||||
"""Mail domain."""
|
||||
|
||||
name = models.CharField(ugettext_lazy("name"), max_length=100, unique=True,
|
||||
@@ -33,6 +32,13 @@ class Domain(AdminObject):
|
||||
verbose_name=ugettext_lazy("Default mailbox quota"),
|
||||
default=0
|
||||
)
|
||||
message_limit = models.PositiveIntegerField(
|
||||
ugettext_lazy("Message sending limit"),
|
||||
null=True, blank=True,
|
||||
help_text=ugettext_lazy(
|
||||
"Number of messages this domain can send per day"
|
||||
)
|
||||
)
|
||||
enabled = models.BooleanField(
|
||||
ugettext_lazy("enabled"),
|
||||
help_text=ugettext_lazy("Check to activate this domain"),
|
||||
@@ -213,6 +219,11 @@ class Domain(AdminObject):
|
||||
return 0
|
||||
return int(self.used_quota / float(self.quota) * 100)
|
||||
|
||||
@property
|
||||
def message_counter_key(self):
|
||||
"""Return the key used to store messages count."""
|
||||
return self.name
|
||||
|
||||
@property
|
||||
def bind_format_dkim_public_key(self):
|
||||
"""Return DKIM public key using Bind format."""
|
||||
|
||||
@@ -20,6 +20,7 @@ from modoboa.parameters import tools as param_tools
|
||||
|
||||
from .base import AdminObject
|
||||
from .domain import Domain
|
||||
from . import mixins
|
||||
|
||||
|
||||
class QuotaManager(models.Manager):
|
||||
@@ -87,8 +88,7 @@ class MailboxManager(Manager):
|
||||
return self.get_queryset().select_related().filter(qf)
|
||||
|
||||
|
||||
@python_2_unicode_compatible
|
||||
class Mailbox(AdminObject):
|
||||
class Mailbox(mixins.MessageLimitMixin, AdminObject):
|
||||
"""User mailbox."""
|
||||
|
||||
address = models.CharField(
|
||||
@@ -98,6 +98,11 @@ class Mailbox(AdminObject):
|
||||
)
|
||||
quota = models.PositiveIntegerField(default=0)
|
||||
use_domain_quota = models.BooleanField(default=False)
|
||||
message_limit = models.PositiveIntegerField(
|
||||
ugettext_lazy("Message sending limit"), null=True, blank=True,
|
||||
help_text=ugettext_lazy(
|
||||
"Number of messages this mailbox can send per day")
|
||||
)
|
||||
domain = models.ForeignKey(Domain, on_delete=models.CASCADE)
|
||||
user = models.OneToOneField(User, on_delete=models.CASCADE)
|
||||
|
||||
@@ -107,9 +112,10 @@ class Mailbox(AdminObject):
|
||||
app_label = "admin"
|
||||
|
||||
def __init__(self, *args, **kwargs):
|
||||
super(Mailbox, self).__init__(*args, **kwargs)
|
||||
super().__init__(*args, **kwargs)
|
||||
self.__mail_home = None
|
||||
self.old_full_address = self.full_address
|
||||
self.old_message_limit = self.message_limit
|
||||
|
||||
def __str__(self):
|
||||
return smart_text(self.full_address)
|
||||
@@ -190,6 +196,11 @@ class Mailbox(AdminObject):
|
||||
"""Set the ``Quota`` for this mailbox."""
|
||||
self._quota_value = instance
|
||||
|
||||
@property
|
||||
def message_counter_key(self):
|
||||
"""Return the key used to store messages count."""
|
||||
return self.full_address
|
||||
|
||||
def rename_dir(self, old_mail_home):
|
||||
"""Rename local directory if needed."""
|
||||
hm = param_tools.get_global_parameter(
|
||||
@@ -319,6 +330,7 @@ class Mailbox(AdminObject):
|
||||
not user.userobjectlimit_set.get(name="quota").max_value
|
||||
)
|
||||
self.set_quota(values["quota"], override_rules)
|
||||
self.message_limit = values.get("message_limit")
|
||||
if newaddress:
|
||||
self.rename(local_part, domain)
|
||||
self.save()
|
||||
|
||||
@@ -0,0 +1,27 @@
|
||||
"""Admin model mixins."""
|
||||
|
||||
from modoboa.policyd.utils import get_message_counter
|
||||
|
||||
|
||||
class MessageLimitMixin:
|
||||
"""A mixin to add message limit support."""
|
||||
|
||||
@property
|
||||
def message_counter_key(self):
|
||||
raise NotImplementedError
|
||||
|
||||
@property
|
||||
def sent_messages(self):
|
||||
"""Return number of sent messages for the current day."""
|
||||
if self.message_limit is None:
|
||||
return None
|
||||
return (
|
||||
self.message_limit - get_message_counter(self.message_counter_key)
|
||||
)
|
||||
|
||||
@property
|
||||
def sent_messages_in_percent(self):
|
||||
"""Return number of sent messages as a percentage."""
|
||||
if self.message_limit is None:
|
||||
return None
|
||||
return int(self.sent_messages / float(self.message_limit) * 100)
|
||||
@@ -45,7 +45,8 @@ class DomainSerializer(serializers.ModelSerializer):
|
||||
"pk", "name", "quota", "default_mailbox_quota", "enabled", "type",
|
||||
"enable_dkim", "dkim_key_selector", "dkim_key_length",
|
||||
"dkim_public_key", "dkim_private_key_path",
|
||||
"mailbox_count", "mbalias_count", "domainalias_count"
|
||||
"mailbox_count", "mbalias_count", "domainalias_count",
|
||||
"message_limit"
|
||||
)
|
||||
read_only_fields = (
|
||||
"pk", "mailbox_count", "mbalias_count", "domainalias_count"
|
||||
@@ -137,7 +138,9 @@ class MailboxSerializer(serializers.ModelSerializer):
|
||||
|
||||
class Meta:
|
||||
model = models.Mailbox
|
||||
fields = ("pk", "full_address", "use_domain_quota", "quota", )
|
||||
fields = (
|
||||
"pk", "full_address", "use_domain_quota", "quota", "message_limit"
|
||||
)
|
||||
|
||||
def validate_full_address(self, value):
|
||||
"""Lower case address."""
|
||||
|
||||
@@ -45,6 +45,12 @@
|
||||
<th>{% trans "Quota" %}</th>
|
||||
<td>{% include "common/progressbar.html" with id='quotabar' value=object.mailbox.get_quota_in_percent %}</td>
|
||||
</tr>
|
||||
{% if object.mailbox.message_limit is not None %}
|
||||
<tr>
|
||||
<th>{% trans "Message sending limit" %}</a></th>
|
||||
<td>{% include "common/progressbar.html" with id='messagelimitbar' value=object.mailbox.sent_messages_in_percent %}</td>
|
||||
</tr>
|
||||
{% endif %}
|
||||
{% endif %}
|
||||
</table>
|
||||
</div>
|
||||
@@ -59,7 +65,7 @@
|
||||
<tr><td>{{ domain }}</td></tr>
|
||||
{% empty %}
|
||||
<tr><td>{% trans "No domain defined." %}</td></tr>
|
||||
{% endfor %}
|
||||
{% endfor %}
|
||||
</table>
|
||||
</div>
|
||||
{% endif %}
|
||||
|
||||
@@ -64,6 +64,20 @@
|
||||
<th>{% trans "Default mailbox quota" %}</a></th>
|
||||
<td>{{ domain.default_mailbox_quota }} {% trans "MB" %}</td>
|
||||
</tr>
|
||||
{% if domain.message_limit != None %}
|
||||
<tr>
|
||||
<th>{% trans "Message sending limit" %}</a></th>
|
||||
<td>{{ domain.message_limit }} {% trans "per day" %}</td>
|
||||
</tr>
|
||||
<tr>
|
||||
<th>{% trans "Usage" %}</a></th>
|
||||
<td>
|
||||
<div class="progress">
|
||||
<div class="{% progress_color domain.sent_messages_in_percent %}" style="width: {{ domain.sent_messages_in_percent }}%" title="{{ domain.sent_messages_in_percent }}% ({{ domain.sent_messages }}) {% trans 'used' context 'message count' %}"></div>
|
||||
</div>
|
||||
</td>
|
||||
</tr>
|
||||
{% endif %}
|
||||
</table>
|
||||
</div>
|
||||
|
||||
@@ -141,7 +155,7 @@
|
||||
</tbody>
|
||||
</table>
|
||||
</div>
|
||||
|
||||
|
||||
{% for template in templates.right %}
|
||||
{% include template %}
|
||||
{% endfor %}
|
||||
|
||||
@@ -3,6 +3,7 @@
|
||||
{% render_field form.type %}
|
||||
{% render_field form.quota %}
|
||||
{% render_field form.default_mailbox_quota %}
|
||||
{% render_field form.message_limit %}
|
||||
{% render_fields_group form "aliases" %}
|
||||
{% render_field form.enabled %}
|
||||
{% render_field form.enable_dns_checks %}
|
||||
|
||||
@@ -5,5 +5,6 @@
|
||||
{% render_field form.create_alias_with_old_address %}
|
||||
{% endif %}
|
||||
{% include "common/generic_field.html" with field=form.quota help_display_mode="tooltip" label_width="col-sm-4" activator=True activator_value=form.quota_act.value deactivate_if_empty=False activator_label=_("Use domain default value") activator_size="col-sm-5" %}
|
||||
{% render_field form.message_limit %}
|
||||
{% render_fields_group form "aliases" %}
|
||||
{% render_fields_group form "senderaddress" %}
|
||||
|
||||
@@ -0,0 +1 @@
|
||||
default_app_config = "modoboa.policyd.apps.PolicydConfig"
|
||||
@@ -0,0 +1,13 @@
|
||||
"""AppConfig for policyd."""
|
||||
|
||||
from django.apps import AppConfig
|
||||
|
||||
|
||||
class PolicydConfig(AppConfig):
|
||||
"""App configuration."""
|
||||
|
||||
name = "modoboa.policyd"
|
||||
verbose_name = "Modoboa policy daemon"
|
||||
|
||||
def ready(self):
|
||||
from . import handlers # NOQA:F401
|
||||
@@ -0,0 +1,3 @@
|
||||
"""App. related constants."""
|
||||
|
||||
REDIS_HASHNAME = "messages_count"
|
||||
@@ -0,0 +1,219 @@
|
||||
"""Core components of the policy daemon."""
|
||||
|
||||
import asyncio
|
||||
import concurrent.futures
|
||||
from email.message import EmailMessage
|
||||
import logging
|
||||
|
||||
import aiosmtplib
|
||||
from dateutil.relativedelta import relativedelta
|
||||
import aioredis
|
||||
|
||||
from django.conf import settings
|
||||
from django.template.loader import render_to_string
|
||||
from django.utils import timezone
|
||||
from django.utils import translation
|
||||
from django.utils.translation import ugettext as _, ugettext_lazy
|
||||
|
||||
from modoboa.admin import models as admin_models
|
||||
from modoboa.core import models as core_models
|
||||
from modoboa.lib.email_utils import split_mailbox
|
||||
|
||||
from . import constants
|
||||
|
||||
logger = logging.getLogger("modoboa.policyd")
|
||||
|
||||
|
||||
SUCCESS_ACTION = b"dunno"
|
||||
FAILURE_ACTION = b"defer_if_permit Daily limit reached, retry later"
|
||||
|
||||
|
||||
async def wait_for(dt):
|
||||
"""sleep until the specified datetime."""
|
||||
one_day = 86400
|
||||
while True:
|
||||
now = timezone.now()
|
||||
remaining = (dt - now).total_seconds()
|
||||
if remaining < one_day:
|
||||
break
|
||||
# asyncio.sleep doesn't like long sleeps, so don't sleep more
|
||||
# than a day at a time
|
||||
await asyncio.sleep(one_day)
|
||||
await asyncio.sleep(remaining)
|
||||
|
||||
|
||||
async def run_at(dt, coro):
|
||||
"""Run coroutine at given datetime."""
|
||||
await wait_for(dt)
|
||||
return await coro()
|
||||
|
||||
|
||||
def get_local_config():
|
||||
"""Return local configuration."""
|
||||
return core_models.LocalConfig.objects.first()
|
||||
|
||||
|
||||
def get_notification_recipients():
|
||||
"""Return superadmins with a mailbox."""
|
||||
return (
|
||||
core_models.User.objects
|
||||
.filter(is_superuser=True, mailbox__isnull=False)
|
||||
)
|
||||
|
||||
|
||||
async def notify_limit_reached(ltype, name):
|
||||
"""Send a notification to super admins about item."""
|
||||
ltype_translations = {
|
||||
"account": ugettext_lazy("account"),
|
||||
"domain": ugettext_lazy("domain")
|
||||
}
|
||||
# We're going to execute sync code so we need an executor
|
||||
executor = concurrent.futures.ThreadPoolExecutor(max_workers=3)
|
||||
loop = asyncio.get_running_loop()
|
||||
futures = [
|
||||
loop.run_in_executor(executor, get_local_config),
|
||||
loop.run_in_executor(executor, get_notification_recipients)
|
||||
]
|
||||
lc, recipients = await asyncio.gather(*futures)
|
||||
sender = lc.parameters.get_value("sender_address", app="core")
|
||||
for recipient in recipients:
|
||||
with translation.override(recipient.language):
|
||||
content = render_to_string(
|
||||
"policyd/notifications/limit_reached.html", {
|
||||
"ltype": ltype_translations[ltype], "name": name
|
||||
})
|
||||
subject = _("[modoboa] Sending limit reached")
|
||||
msg = EmailMessage()
|
||||
msg["From"] = sender
|
||||
msg["To"] = recipient.email
|
||||
msg["Subject"] = subject
|
||||
msg.set_content(content)
|
||||
await aiosmtplib.send(msg)
|
||||
|
||||
|
||||
async def decrement_limit(rclient, ltype, name):
|
||||
"""Decrement the given limit by one."""
|
||||
new_counter = await rclient.hincrby(constants.REDIS_HASHNAME, name, -1)
|
||||
if new_counter <= 0:
|
||||
logger.info("Limit reached for {} {}".format(ltype, name))
|
||||
loop = asyncio.get_running_loop()
|
||||
loop.create_task(notify_limit_reached(ltype, name))
|
||||
|
||||
|
||||
async def apply_policies(attributes):
|
||||
"""Apply defined policies to received request."""
|
||||
sasl_username = attributes.get("sasl_username")
|
||||
if not sasl_username:
|
||||
return SUCCESS_ACTION
|
||||
rclient = await aioredis.create_redis_pool(settings.REDIS_URL)
|
||||
decr_domain = False
|
||||
decr_user = False
|
||||
localpart, domain = split_mailbox(sasl_username)
|
||||
if await rclient.hexists(constants.REDIS_HASHNAME, domain):
|
||||
counter = await rclient.hget(constants.REDIS_HASHNAME, domain)
|
||||
logger.info("Domain {} current counter: {}".format(domain, counter))
|
||||
if int(counter) <= 0:
|
||||
return FAILURE_ACTION
|
||||
decr_domain = True
|
||||
if await rclient.hexists(constants.REDIS_HASHNAME, sasl_username):
|
||||
counter = await rclient.hget(constants.REDIS_HASHNAME, sasl_username)
|
||||
logger.info("Account {} current counter: {}".format(
|
||||
sasl_username, counter))
|
||||
if int(counter) <= 0:
|
||||
return FAILURE_ACTION
|
||||
decr_user = True
|
||||
if decr_domain:
|
||||
await decrement_limit(rclient, "domain", domain)
|
||||
if decr_user:
|
||||
await decrement_limit(rclient, "account", sasl_username)
|
||||
return SUCCESS_ACTION
|
||||
|
||||
|
||||
async def process_messages(queue):
|
||||
"""Pull from queue and process messages."""
|
||||
while True:
|
||||
transport, message = await queue.get()
|
||||
result = await apply_policies(message)
|
||||
transport.write(b"action=" + result + b"\n\n")
|
||||
queue.task_done()
|
||||
|
||||
|
||||
class APDProtocol(asyncio.Protocol):
|
||||
"""Access Policy Delegation protocol (postfix)."""
|
||||
|
||||
def __init__(self, *args, **kwargs):
|
||||
"""Store queue."""
|
||||
self.queue = kwargs.pop("queue")
|
||||
super().__init__(*args, **kwargs)
|
||||
|
||||
def connection_made(self, transport):
|
||||
peername = transport.get_extra_info("peername")
|
||||
logger.info("Connection from {}".format(peername))
|
||||
self.transport = transport
|
||||
|
||||
def data_received(self, data):
|
||||
attributes = {}
|
||||
for line in data.decode().split("\n"):
|
||||
if not line:
|
||||
continue
|
||||
try:
|
||||
name, value = line.split("=")
|
||||
except ValueError:
|
||||
continue
|
||||
attributes[name] = value
|
||||
|
||||
# action = SUCCESS_ACTION
|
||||
state = attributes.get("protocol_state")
|
||||
if state == "RCPT":
|
||||
self.queue.put_nowait((self.transport, attributes))
|
||||
else:
|
||||
self.transport.write(b"action=" + SUCCESS_ACTION + b"\n\n")
|
||||
|
||||
|
||||
def get_next_execution_dt():
|
||||
"""Return next execution date and time."""
|
||||
return (timezone.now() + relativedelta(days=1)).replace(
|
||||
hour=0, minute=0, second=0)
|
||||
|
||||
|
||||
def get_domains_to_reset():
|
||||
"""Return a list of domain to reset."""
|
||||
return admin_models.Domain.objects.filter(message_limit__isnull=False)
|
||||
|
||||
|
||||
def get_mailboxes_to_reset():
|
||||
"""Return a list of mailboxes to reset."""
|
||||
return (
|
||||
admin_models.Mailbox.objects.filter(message_limit__isnull=False)
|
||||
.select_related("domain")
|
||||
)
|
||||
|
||||
|
||||
async def reset_counters():
|
||||
"""Reset all counters."""
|
||||
rclient = await aioredis.create_redis_pool(settings.REDIS_URL)
|
||||
logger.info("Resetting all counters")
|
||||
# We're going to execute sync code so we need an executor
|
||||
executor = concurrent.futures.ThreadPoolExecutor(max_workers=3)
|
||||
loop = asyncio.get_running_loop()
|
||||
futures = [
|
||||
loop.run_in_executor(executor, get_domains_to_reset),
|
||||
loop.run_in_executor(executor, get_mailboxes_to_reset)
|
||||
]
|
||||
domains, mboxes = await asyncio.gather(*futures)
|
||||
for domain in domains:
|
||||
rclient.hset(
|
||||
constants.REDIS_HASHNAME, domain.name, domain.message_limit)
|
||||
for mb in mboxes:
|
||||
rclient.hset(
|
||||
constants.REDIS_HASHNAME, mb.full_address, mb.message_limit)
|
||||
# reschedule
|
||||
loop = asyncio.get_running_loop()
|
||||
loop.create_task(run_at(get_next_execution_dt(), reset_counters))
|
||||
|
||||
|
||||
def start_reset_counters_coro(loop):
|
||||
"""Start coroutine."""
|
||||
first_time = (timezone.now() + relativedelta(days=1)).replace(
|
||||
hour=0, minute=0, second=0)
|
||||
loop.create_task(run_at(first_time, reset_counters))
|
||||
@@ -0,0 +1,45 @@
|
||||
"""App related signal handlers."""
|
||||
|
||||
import redis
|
||||
|
||||
from django.conf import settings
|
||||
from django.db.models import signals
|
||||
from django.dispatch import receiver
|
||||
|
||||
from modoboa.admin import models as admin_models
|
||||
|
||||
from . import constants
|
||||
|
||||
|
||||
def set_message_limit(instance, key):
|
||||
"""Store message limit in Redis."""
|
||||
old_message_limit = instance._loaded_values.get("message_limit")
|
||||
if old_message_limit == instance.message_limit:
|
||||
return
|
||||
rclient = redis.Redis(
|
||||
host=settings.REDIS_HOST,
|
||||
port=settings.REDIS_PORT,
|
||||
db=settings.REDIS_QUOTA_DB
|
||||
)
|
||||
if instance.message_limit is None:
|
||||
# delete existing key
|
||||
if rclient.hexists(constants.REDIS_HASHNAME, key):
|
||||
rclient.hdel(constants.REDIS_HASHNAME, key)
|
||||
return
|
||||
if old_message_limit is not None:
|
||||
diff = instance.message_limit - old_message_limit
|
||||
else:
|
||||
diff = instance.message_limit
|
||||
rclient.hincrby(constants.REDIS_HASHNAME, key, diff)
|
||||
|
||||
|
||||
@receiver(signals.post_save, sender=admin_models.Domain)
|
||||
def set_domain_message_limit(sender, instance, created, **kwargs):
|
||||
"""Store domain message limit in Redis."""
|
||||
set_message_limit(instance, instance.name)
|
||||
|
||||
|
||||
@receiver(signals.post_save, sender=admin_models.Mailbox)
|
||||
def set_mailbox_message_limit(sender, instance, created, **kwargs):
|
||||
"""Store mailbox message limit in Redis."""
|
||||
set_message_limit(instance, instance.full_address)
|
||||
@@ -0,0 +1,36 @@
|
||||
# SOME DESCRIPTIVE TITLE.
|
||||
# Copyright (C) YEAR THE PACKAGE'S COPYRIGHT HOLDER
|
||||
# This file is distributed under the same license as the PACKAGE package.
|
||||
# FIRST AUTHOR <EMAIL@ADDRESS>, YEAR.
|
||||
#
|
||||
#, fuzzy
|
||||
msgid ""
|
||||
msgstr ""
|
||||
"Project-Id-Version: PACKAGE VERSION\n"
|
||||
"Report-Msgid-Bugs-To: \n"
|
||||
"POT-Creation-Date: 2020-06-11 20:26+0200\n"
|
||||
"PO-Revision-Date: YEAR-MO-DA HO:MI+ZONE\n"
|
||||
"Last-Translator: FULL NAME <EMAIL@ADDRESS>\n"
|
||||
"Language-Team: LANGUAGE <LL@li.org>\n"
|
||||
"Language: \n"
|
||||
"MIME-Version: 1.0\n"
|
||||
"Content-Type: text/plain; charset=UTF-8\n"
|
||||
"Content-Transfer-Encoding: 8bit\n"
|
||||
"Plural-Forms: nplurals=2; plural=(n != 1);\n"
|
||||
|
||||
#: core.py:74
|
||||
msgid "account"
|
||||
msgstr ""
|
||||
|
||||
#: core.py:75
|
||||
msgid "domain"
|
||||
msgstr ""
|
||||
|
||||
#: core.py:92
|
||||
msgid "[modoboa] Sending limit reached"
|
||||
msgstr ""
|
||||
|
||||
#: templates/policyd/notifications/limit_reached.html:3
|
||||
#, python-format
|
||||
msgid "Daily limit of the following %(ltype)s has been reached: %(name)s."
|
||||
msgstr ""
|
||||
@@ -0,0 +1,35 @@
|
||||
# SOME DESCRIPTIVE TITLE.
|
||||
# Copyright (C) YEAR THE PACKAGE'S COPYRIGHT HOLDER
|
||||
# This file is distributed under the same license as the PACKAGE package.
|
||||
# FIRST AUTHOR <EMAIL@ADDRESS>, YEAR.
|
||||
#
|
||||
msgid ""
|
||||
msgstr ""
|
||||
"Project-Id-Version: PACKAGE VERSION\n"
|
||||
"Report-Msgid-Bugs-To: \n"
|
||||
"POT-Creation-Date: 2020-06-11 20:26+0200\n"
|
||||
"PO-Revision-Date: 2020-06-11 20:27+0200\n"
|
||||
"Last-Translator: FULL NAME <EMAIL@ADDRESS>\n"
|
||||
"Language-Team: LANGUAGE <LL@li.org>\n"
|
||||
"Language: \n"
|
||||
"MIME-Version: 1.0\n"
|
||||
"Content-Type: text/plain; charset=UTF-8\n"
|
||||
"Content-Transfer-Encoding: 8bit\n"
|
||||
"Plural-Forms: nplurals=2; plural=(n > 1);\n"
|
||||
|
||||
#: core.py:74
|
||||
msgid "account"
|
||||
msgstr "compte"
|
||||
|
||||
#: core.py:75
|
||||
msgid "domain"
|
||||
msgstr "domaine"
|
||||
|
||||
#: core.py:92
|
||||
msgid "[modoboa] Sending limit reached"
|
||||
msgstr "[modoboa] Limite d'envoi atteinte"
|
||||
|
||||
#: templates/policyd/notifications/limit_reached.html:3
|
||||
#, python-format
|
||||
msgid "Daily limit of the following %(ltype)s has been reached: %(name)s."
|
||||
msgstr "La limite journalière du %(ltype)s suivant a été atteinte : %(name)s."
|
||||
@@ -0,0 +1,67 @@
|
||||
"""Policy daemon management command."""
|
||||
|
||||
import asyncio
|
||||
from contextlib import suppress
|
||||
import functools
|
||||
import logging
|
||||
import signal
|
||||
|
||||
from django.core.management.base import BaseCommand
|
||||
|
||||
from ... import core
|
||||
|
||||
logger = logging.getLogger("modoboa.policyd")
|
||||
|
||||
|
||||
def ask_exit(signame, loop):
|
||||
"""Stop event loop."""
|
||||
loop.stop()
|
||||
|
||||
|
||||
class Command(BaseCommand):
|
||||
"""Management command for policy daemon."""
|
||||
|
||||
help = "Launch Modoboa policy daemon"
|
||||
|
||||
def add_arguments(self, parser):
|
||||
"""Add command line arguments."""
|
||||
parser.add_argument("--host", type=str, default="127.0.0.1")
|
||||
parser.add_argument("--port", type=int, default=9999)
|
||||
|
||||
def handle(self, *args, **options):
|
||||
"""Entry point."""
|
||||
queue = asyncio.Queue()
|
||||
loop = asyncio.get_event_loop()
|
||||
coro = loop.create_server(
|
||||
lambda: core.APDProtocol(queue=queue),
|
||||
options["host"], options["port"]
|
||||
)
|
||||
server = loop.run_until_complete(coro)
|
||||
|
||||
loop.create_task(core.process_messages(queue))
|
||||
|
||||
# Schedule reset task
|
||||
core.start_reset_counters_coro(loop)
|
||||
|
||||
for signame in {'SIGINT', 'SIGTERM'}:
|
||||
loop.add_signal_handler(
|
||||
getattr(signal, signame),
|
||||
functools.partial(ask_exit, signame, loop)
|
||||
)
|
||||
|
||||
logger.info("Serving on {}".format(server.sockets[0].getsockname()))
|
||||
|
||||
loop.run_forever()
|
||||
|
||||
logger.info("Stopping policy daemon...")
|
||||
# Close the server
|
||||
server.close()
|
||||
loop.run_until_complete(server.wait_closed())
|
||||
# Cancel pending tasks
|
||||
for task in asyncio.Task.all_tasks():
|
||||
task.cancel()
|
||||
# Await task to execute it's cancellation. Cancelled task
|
||||
# raises asyncio.CancelledError that we can suppress
|
||||
with suppress(asyncio.CancelledError):
|
||||
loop.run_until_complete(task)
|
||||
loop.close()
|
||||
@@ -0,0 +1,5 @@
|
||||
{% load i18n %}
|
||||
|
||||
{% blocktrans trimmed with ltype=ltype name=name %}
|
||||
Daily limit of the following {{ ltype }} has been reached: {{ name }}.
|
||||
{% endblocktrans %}
|
||||
@@ -0,0 +1,196 @@
|
||||
"""Policy daemon related tests."""
|
||||
|
||||
import asyncio
|
||||
from mock import patch
|
||||
from multiprocessing import Process
|
||||
import socket
|
||||
import time
|
||||
|
||||
import redis
|
||||
|
||||
from django.conf import settings
|
||||
from django.core.management import call_command
|
||||
|
||||
from modoboa.admin import factories as admin_factories
|
||||
from modoboa.admin import models as admin_models
|
||||
from modoboa.core import models as core_models
|
||||
from modoboa.lib.tests import ModoTestCase
|
||||
from modoboa.policyd import core as policyd_core
|
||||
|
||||
from . import constants
|
||||
|
||||
|
||||
def start_policy_daemon():
|
||||
call_command("policy_daemon")
|
||||
|
||||
|
||||
class PolicyDaemonTestCase(ModoTestCase):
|
||||
"""Test cases for policy daemon.
|
||||
|
||||
A redis instance is required to run those tests.
|
||||
"""
|
||||
|
||||
@classmethod
|
||||
def setUpTestData(cls): # NOQA:N802
|
||||
"""Create test data."""
|
||||
super().setUpTestData()
|
||||
admin_factories.populate_database()
|
||||
# FIXME: all database modifications must be made before the
|
||||
# daemon is started otherwise they won't be seen by the other
|
||||
# process. I think it's just a sqlite issue...
|
||||
cls.admin = core_models.User.objects.get(username="user@test2.com")
|
||||
cls.admin.role = "SuperAdmins"
|
||||
|
||||
def setUp(self):
|
||||
super().setUp()
|
||||
self.rclient = redis.Redis(
|
||||
host=settings.REDIS_HOST,
|
||||
port=settings.REDIS_PORT,
|
||||
db=settings.REDIS_QUOTA_DB
|
||||
)
|
||||
self.rclient.set_response_callback("HGET", int)
|
||||
self.rclient.delete(constants.REDIS_HASHNAME)
|
||||
patcher = patch("aiosmtplib.send")
|
||||
self.send_mock = patcher.start()
|
||||
self.process = Process(target=start_policy_daemon)
|
||||
self.process.daemon = True
|
||||
self.process.start()
|
||||
# Wait a bit for the daemon to start
|
||||
self.process.join(0.1)
|
||||
|
||||
def set_domain_limit(self, name, value):
|
||||
"""Set daily limit for domain."""
|
||||
domain = admin_models.Domain.objects.get(name=name)
|
||||
domain.message_limit = value
|
||||
domain.save()
|
||||
self.assertEqual(
|
||||
self.rclient.hget(constants.REDIS_HASHNAME, domain.name),
|
||||
domain.message_limit
|
||||
)
|
||||
return domain
|
||||
|
||||
def set_account_limit(self, name, value):
|
||||
"""Set daily limit for account."""
|
||||
account = core_models.User.objects.get(username=name)
|
||||
mb = account.mailbox
|
||||
self.rclient.hdel(constants.REDIS_HASHNAME, account.email)
|
||||
mb.message_limit = value
|
||||
mb.save()
|
||||
self.assertEqual(
|
||||
self.rclient.hget(constants.REDIS_HASHNAME, account.email),
|
||||
mb.message_limit
|
||||
)
|
||||
return account
|
||||
|
||||
def tearDown(self):
|
||||
self.process.terminate()
|
||||
self.process.join()
|
||||
|
||||
def connect_to_daemon(self):
|
||||
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||||
s.connect(("127.0.0.1", 9999))
|
||||
return s
|
||||
|
||||
def test_daemon_starts(self):
|
||||
s = self.connect_to_daemon()
|
||||
s.send(b"protocol_state=RCPT\n\n")
|
||||
res = s.recv(1024)
|
||||
self.assertEqual(res, b"action=dunno\n\n")
|
||||
s.close()
|
||||
|
||||
def test_domain_limit(self):
|
||||
domain = self.set_domain_limit("test.com", 2)
|
||||
s = self.connect_to_daemon()
|
||||
|
||||
# This one will be accepted
|
||||
s.send(b"""protocol_state=RCPT
|
||||
sasl_username=user@test.com
|
||||
|
||||
""")
|
||||
res = s.recv(1024)
|
||||
self.assertEqual(res, b"action=dunno\n\n")
|
||||
self.assertEqual(
|
||||
self.rclient.hget(constants.REDIS_HASHNAME, domain.name),
|
||||
domain.message_limit - 1
|
||||
)
|
||||
|
||||
# This one will be accepted and will trigger a notification
|
||||
# but I can't check it yet
|
||||
s.send(b"""protocol_state=RCPT
|
||||
sasl_username=user@test.com
|
||||
|
||||
""")
|
||||
time.sleep(0.1)
|
||||
res = s.recv(1024)
|
||||
self.assertEqual(res, b"action=dunno\n\n")
|
||||
|
||||
# This one will be denied
|
||||
s.send(b"""protocol_state=RCPT
|
||||
sasl_username=user@test.com
|
||||
|
||||
""")
|
||||
res = s.recv(1024)
|
||||
self.assertEqual(
|
||||
res, b"action=defer_if_permit Daily limit reached, retry later\n\n"
|
||||
)
|
||||
|
||||
s.close()
|
||||
|
||||
def test_account_limit(self):
|
||||
account = self.set_account_limit("user@test.com", 2)
|
||||
s = self.connect_to_daemon()
|
||||
|
||||
# This one will be accepted
|
||||
s.send(b"""protocol_state=RCPT
|
||||
sasl_username=user@test.com
|
||||
|
||||
""")
|
||||
res = s.recv(1024)
|
||||
self.assertEqual(res, b"action=dunno\n\n")
|
||||
self.assertEqual(
|
||||
self.rclient.hget(constants.REDIS_HASHNAME, account.email),
|
||||
account.mailbox.message_limit - 1
|
||||
)
|
||||
|
||||
# This one will be accepted and will trigger a notification
|
||||
# but I can't check it yet
|
||||
s.send(b"""protocol_state=RCPT
|
||||
sasl_username=user@test.com
|
||||
|
||||
""")
|
||||
time.sleep(0.1)
|
||||
res = s.recv(1024)
|
||||
self.assertEqual(res, b"action=dunno\n\n")
|
||||
|
||||
# This one will be denied
|
||||
s.send(b"""protocol_state=RCPT
|
||||
sasl_username=user@test.com
|
||||
|
||||
""")
|
||||
res = s.recv(1024)
|
||||
self.assertEqual(
|
||||
res, b"action=defer_if_permit Daily limit reached, retry later\n\n"
|
||||
)
|
||||
|
||||
s.close()
|
||||
|
||||
def test_reset_counters(self):
|
||||
domain = self.set_domain_limit("test.com", 20)
|
||||
account = self.set_account_limit("user@test.com", 10)
|
||||
self.rclient.hset(constants.REDIS_HASHNAME, domain.name, 10)
|
||||
self.rclient.hset(constants.REDIS_HASHNAME, account.email, 0)
|
||||
event_loop = asyncio.new_event_loop()
|
||||
asyncio.set_event_loop(event_loop)
|
||||
|
||||
async def run_test():
|
||||
await policyd_core.reset_counters()
|
||||
|
||||
# Run the async test
|
||||
coro = asyncio.coroutine(run_test)
|
||||
event_loop.run_until_complete(coro())
|
||||
event_loop.close()
|
||||
|
||||
self.assertEqual(
|
||||
self.rclient.hget(constants.REDIS_HASHNAME, domain.name), 20)
|
||||
self.assertEqual(
|
||||
self.rclient.hget(constants.REDIS_HASHNAME, account.email), 10)
|
||||
@@ -0,0 +1,24 @@
|
||||
"""Tooling."""
|
||||
|
||||
import redis
|
||||
|
||||
from django.conf import settings
|
||||
|
||||
from . import constants
|
||||
|
||||
|
||||
def get_redis_connection():
|
||||
"""Return a client connection to Redis server."""
|
||||
rclient = redis.Redis(
|
||||
host=settings.REDIS_HOST,
|
||||
port=settings.REDIS_PORT,
|
||||
db=settings.REDIS_QUOTA_DB
|
||||
)
|
||||
rclient.set_response_callback("HGET", int)
|
||||
return rclient
|
||||
|
||||
|
||||
def get_message_counter(key):
|
||||
"""Return current counter for given key."""
|
||||
rclient = get_redis_connection()
|
||||
return rclient.hget(constants.REDIS_HASHNAME, key)
|
||||
@@ -27,3 +27,7 @@ lxml
|
||||
chardet
|
||||
ovh
|
||||
oath
|
||||
redis
|
||||
|
||||
aiosmtplib
|
||||
aioredis
|
||||
|
||||
@@ -0,0 +1,7 @@
|
||||
version: '3'
|
||||
|
||||
services:
|
||||
redis:
|
||||
image: redis:5-alpine
|
||||
ports:
|
||||
- 6379:6379
|
||||
@@ -78,6 +78,7 @@ MODOBOA_APPS = (
|
||||
'modoboa.limits',
|
||||
'modoboa.parameters',
|
||||
'modoboa.dnstools',
|
||||
'modoboa.policyd',
|
||||
# Modoboa extensions here.
|
||||
)
|
||||
|
||||
@@ -186,6 +187,13 @@ REST_FRAMEWORK = {
|
||||
|
||||
MODOBOA_API_URL = 'https://api.modoboa.org/1/'
|
||||
|
||||
# REDIS
|
||||
|
||||
REDIS_URL = 'redis://localhost/0'
|
||||
REDIS_HOST = 'localhost'
|
||||
REDIS_PORT = 6379
|
||||
REDIS_QUOTA_DB = 0
|
||||
|
||||
# Password validation
|
||||
# https://docs.djangoproject.com/en/2.2/ref/settings/#auth-password-validators
|
||||
|
||||
|
||||
Reference in New Issue
Block a user