github-jwinterm/modoboa
1
0
Fork 0
mirror of https://github.com/jwinterm/modoboa.git synced 2026-10-17 05:41:25 +02:00
Code Issues Packages Projects Releases Wiki Activity

Check MX validity when adding new domain. (#1215)

Browse Source 
* Check MX validity when adding new domain.

see #1211

* Fixed unit test.

see #1215

* Added unit tests.

see #1211

* Fixed unit test.

see #1211

* Added missing file.

see #1211

* Fixed error definition.

see #1211
This commit is contained in:
Antoine Nguyen
2017-08-18 09:29:01 +02:00
committed by GitHub
parent 1221240b2f
commit 226d65c0a0
13 changed files with 330 additions and 115 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modoboa/admin/app_settings.py
+25 -1
View File
@@ -6,7 +6,7 @@ import os
from django import forms
from django.conf import settings
from django.utils.translation import ugettext_lazy
from django.utils.translation import ugettext as _, ugettext_lazy
from django.utils.encoding import force_text
from modoboa.lib.form_utils import YesNoField, SeparatorField
@@ -38,6 +38,15 @@ class AdminParametersForm(param_forms.AdminParametersForm):
required=False
)
domains_must_have_authorized_mx = YesNoField(
label=ugettext_lazy("New domains must use authorized MX(s)"),
initial=False,
help_text=ugettext_lazy(
"Prevent the creation of a new domain if its MX record does "
"not use one of the defined addresses."
)
)
enable_dnsbl_checks = YesNoField(
label=ugettext_lazy("Enable DNSBL checks"),
initial=True,
@@ -102,6 +111,7 @@ class AdminParametersForm(param_forms.AdminParametersForm):
# Visibility rules
visibility_rules = {
"valid_mxs": "enable_mx_checks=True",
"domains_must_have_authorized_mx": "enable_mx_checks=True",
"mailboxes_owner": "handle_mailboxes=True",
}
@@ -155,6 +165,20 @@ class AdminParametersForm(param_forms.AdminParametersForm):
)
return self.cleaned_data["default_mailbox_quota"]
def clean(self):
"""Check MX options."""
cleaned_data = super(AdminParametersForm, self).clean()
condition = (
cleaned_data.get("enable_mx_checks") and
cleaned_data.get("domains_must_have_authorized_mx") and
not cleaned_data.get("valid_mxs"))
if condition:
self.add_error(
"valid_mxs",
_("Define at least one authorized network / address")
)
return cleaned_data
def load_admin_settings():
"""Load admin settings."""
modoboa/admin/forms/domain.py
+21 -10
View File
@@ -21,7 +21,7 @@ from modoboa.lib.web_utils import render_to_json_response
from modoboa.parameters import tools as param_tools
from .. import signals
from ..lib import check_if_domain_exists
from ..lib import check_if_domain_exists, domain_has_authorized_mx
from ..models import (
Domain, DomainAlias, Mailbox, Alias
)
@@ -82,6 +82,25 @@ class DomainFormGeneral(forms.ModelForm, DynamicForm):
name = "aliases_%d" % (pos + 1)
self._create_field(forms.CharField, name, dalias.name, 3)
def clean_name(self):
"""Check unicity and more."""
name = self.cleaned_data["name"]
label = check_if_domain_exists(
name, [(DomainAlias, _('domain alias'))])
if label is not None:
raise forms.ValidationError(
_("A %s with this name already exists") % force_text(label))
domains_must_have_authorized_mx = (
param_tools.get_global_parameter("domains_must_have_authorized_mx")
)
if domains_must_have_authorized_mx and not self.user.is_superuser:
if not domain_has_authorized_mx(name):
raise forms.ValidationError(
_("No authorized MX record found for this domain"))
return name
def clean(self):
"""Custom fields validation.
@@ -93,14 +112,6 @@ class DomainFormGeneral(forms.ModelForm, DynamicForm):
cleaned_data = super(DomainFormGeneral, self).clean()
if self._errors:
return cleaned_data
name = cleaned_data["name"]
label = check_if_domain_exists(
name, [(DomainAlias, _('domain alias'))])
if label is not None:
self.add_error(
"name", _("A %s with this name already exists")
% force_text(label)
)
condition = (
self.cleaned_data["quota"] != 0 and
self.cleaned_data["default_mailbox_quota"] >
@@ -128,7 +139,7 @@ class DomainFormGeneral(forms.ModelForm, DynamicForm):
if cleaned_data[k] == "":
del cleaned_data[k]
continue
if cleaned_data[k] == name:
if cleaned_data[k] == self.cleaned_data["name"]:
self._errors[k] = self.error_class(
[_("A %s with this name already exists") % _("domain")]
)
modoboa/admin/lib.py
+45
View File
@@ -5,7 +5,11 @@
from __future__ import unicode_literals
from functools import wraps
import ipaddress
from itertools import chain
import socket
import dns.resolver
from django.contrib.contenttypes.models import ContentType
from django.db.models import Q
@@ -14,6 +18,7 @@ from django.utils.translation import ugettext as _
from modoboa.core.models import User
from modoboa.core import signals as core_signals
from modoboa.lib.exceptions import PermDeniedException
from modoboa.parameters import tools as param_tools
from .models import Domain, DomainAlias, Alias
@@ -161,3 +166,43 @@ def import_forward(user, row, formopts):
def import_dlist(user, row, formopts):
_import_alias(user, row)
def get_domain_mx_list(name):
"""Return a list of MX IP address for domain."""
result = []
try:
answers = dns.resolver.query(name, "MX")
except (dns.resolver.NoAnswer, dns.resolver.NXDOMAIN,
dns.resolver.NoNameservers):
return result
for answer in answers:
try:
# work if .exchange is a name or IP
address = socket.gethostbyname(str(answer.exchange))
except socket.gaierror:
pass
else:
try:
# we must have a valid IP
address = ipaddress.ip_address("{}".format(address))
except ValueError:
pass
else:
result.append((str(answer.exchange), address))
return result
def domain_has_authorized_mx(name):
"""Check if domain has authorized mx record at least."""
valid_mxs = param_tools.get_global_parameter("valid_mxs")
valid_mxs = [ipaddress.ip_network(u"{}".format(v.strip()))
for v in valid_mxs.split() if v.strip()]
domain_mxs = get_domain_mx_list(name)
if len(domain_mxs) == 0:
return True
for mx_addr, mx_ip_addr in domain_mxs:
for subnet in valid_mxs:
if mx_ip_addr in subnet:
return True
return False
modoboa/admin/management/commands/subcommands/_mx.py
+3 -48
View File
@@ -2,11 +2,8 @@
from __future__ import print_function, unicode_literals
from datetime import timedelta
import ipaddress
import dns.resolver
import gevent
from gevent import socket
@@ -17,7 +14,6 @@ from django.core.management.base import BaseCommand
from django.template.loader import render_to_string
from django.utils.functional import cached_property
from django.utils.translation import ugettext as _
from django.utils import timezone
from modoboa.admin import constants
from modoboa.admin import models
@@ -70,48 +66,6 @@ class CheckMXRecords(BaseCommand):
"--ttl", type=int, default=7200,
help="TTL for dns query.")
def get_mx_records_for_domain(self, domain, ttl=7200):
"""Return one or more `models.MXRecord` for `domain`.
DNS queries are not performed while `ttl` (in seconds) is still valid.
"""
now = timezone.now()
records = models.MXRecord.objects.filter(domain=domain,
updated__gt=now)
if records.exists():
for record in records:
yield record
raise StopIteration()
models.MXRecord.objects.filter(domain=domain).delete()
try:
answers = dns.resolver.query(domain.name, "MX")
except (dns.resolver.NoAnswer, dns.resolver.NXDOMAIN,
dns.resolver.NoNameservers):
raise StopIteration()
delta = timedelta(seconds=ttl)
for answer in answers:
try:
# work if .exchange is a name or IP
address = socket.gethostbyname(str(answer.exchange))
except socket.gaierror:
pass
else:
try:
# we must have a valid IP
address = ipaddress.ip_address(u"{}".format(address))
except ValueError:
pass
else:
record = models.MXRecord.objects.create(
domain=domain,
name=u"{}".format(str(answer.exchange).strip(".")),
address=u"{}".format(address),
updated=now + delta)
yield record
def query_dnsbl(self, mx_list, provider):
"""Check given IP against given DNSBL provider."""
results = {}
@@ -183,7 +137,7 @@ class CheckMXRecords(BaseCommand):
"""
alerts = []
check = False
mxs = [(mx, ipaddress.ip_address(u"%s" % mx.address))
mxs = [(mx, ipaddress.ip_address("%s" % mx.address))
for mx in mx_list]
valid_mxs = self.valid_mxs
if not mxs:
@@ -222,7 +176,8 @@ class CheckMXRecords(BaseCommand):
def check_domain(self, domain, timeout=3, ttl=7200, **options):
"""Check specified domain."""
mx_list = list(self.get_mx_records_for_domain(domain, ttl=ttl))
mx_list = list(
models.MXRecord.objects.get_or_create_for_domain(domain, ttl))
if param_tools.get_global_parameter("enable_mx_checks"):
self.check_valid_mx(domain, mx_list, **options)
modoboa/admin/models/__init__.py
+2 -1
View File
@@ -3,9 +3,10 @@
from __future__ import unicode_literals
from .base import AdminObject
from .domain import Domain, MXRecord, DNSBLResult
from .domain import Domain
from .domain_alias import DomainAlias
from .mailbox import Mailbox, Quota, MailboxOperation, SenderAddress
from .mxrecord import DNSBLResult, MXRecord
from .alias import Alias, AliasRecipient
__all__ = [
modoboa/admin/models/domain.py
+12 -55
View File
@@ -23,9 +23,9 @@ from modoboa.core.models import User, ObjectAccess
from modoboa.lib.exceptions import BadRequest, Conflict
from modoboa.parameters import tools as param_tools
from .. import signals
from .base import AdminObject
from .. import constants
from .. import signals
class DomainManager(Manager):
@@ -250,11 +250,22 @@ class Domain(AdminObject):
:param ``core.User`` user: user creating the domain
:param str row: a list containing domain's definition
"""
from .. import lib
if len(row) < 5:
raise BadRequest(_("Invalid line"))
self.name = row[1].strip()
if Domain.objects.filter(name=self.name).exists():
raise Conflict
domains_must_have_authorized_mx = (
param_tools.get_global_parameter("domains_must_have_authorized_mx")
)
if domains_must_have_authorized_mx and not user.is_superuser:
if not lib.domain_has_authorized_mx(self.name):
raise BadRequest(
_("No authorized MX record found for domain {}")
.format(self.name)
)
try:
self.quota = int(row[2].strip())
except ValueError:
@@ -303,57 +314,3 @@ class Domain(AdminObject):
reversion.register(Domain)
class MXQuerySet(models.QuerySet):
"""Custom manager for MXRecord."""
def has_valids(self):
"""Return managed results."""
if param_tools.get_global_parameter("valid_mxs").strip():
return self.filter(managed=True).exists()
return self.exists()
@python_2_unicode_compatible
class MXRecord(models.Model):
"""A model used to store MX records for Domain."""
domain = models.ForeignKey(Domain)
name = models.CharField(max_length=254)
address = models.GenericIPAddressField()
managed = models.BooleanField(default=False)
updated = models.DateTimeField()
objects = models.Manager.from_queryset(MXQuerySet)()
def is_managed(self):
if not param_tools.get_global_parameter("enable_mx_checks"):
return False
return bool(param_tools.get_global_parameter("valid_mxs").strip())
def __str__(self):
return "{0.name} ({0.address}) for {0.domain} ".format(self)
class DNSBLQuerySet(models.QuerySet):
"""Custom manager for DNSBLResultManager."""
def blacklisted(self):
"""Return blacklisted results."""
return self.exclude(status="")
class DNSBLResult(models.Model):
"""Store a DNSBL query result."""
domain = models.ForeignKey(Domain)
provider = models.CharField(max_length=254, db_index=True)
mx = models.ForeignKey(MXRecord)
status = models.CharField(max_length=45, blank=True, db_index=True)
objects = models.Manager.from_queryset(DNSBLQuerySet)()
class Meta:
app_label = "admin"
unique_together = [("domain", "provider", "mx")]
modoboa/admin/models/mxrecord.py
+98
View File
@@ -0,0 +1,98 @@
"""MX records storage."""
from __future__ import unicode_literals
import datetime
from django.db import models
from django.utils import timezone
from django.utils.encoding import python_2_unicode_compatible
from modoboa.parameters import tools as param_tools
class MXRecordQuerySet(models.QuerySet):
"""Custom manager for MXRecord."""
def has_valids(self):
"""Return managed results."""
if param_tools.get_global_parameter("valid_mxs").strip():
return self.filter(managed=True).exists()
return self.exists()
class MXRecordManager(models.Manager):
"""Custom manager for MXRecord."""
def get_or_create_for_domain(self, domain, ttl=7200):
"""Get or create MX record(s) for given domain.
DNS queries are not performed while `ttl` (in seconds) is still valid.
"""
from .. import lib
now = timezone.now()
records = self.get_queryset().filter(
domain=domain, updated__gt=now)
if records.exists():
for record in records:
yield record
raise StopIteration()
self.get_queryset().filter(domain=domain).delete()
delta = datetime.timedelta(seconds=ttl)
domain_mxs = lib.get_domain_mx_list(domain.name)
if len(domain_mxs) == 0:
raise StopIteration
for mx_addr, mx_ip_addr in domain_mxs:
record = self.get_queryset().create(
domain=domain,
name="{}".format(mx_addr.strip(".")),
address="{}".format(mx_ip_addr),
updated=now + delta)
yield record
@python_2_unicode_compatible
class MXRecord(models.Model):
"""A model used to store MX records for Domain."""
domain = models.ForeignKey("admin.Domain")
name = models.CharField(max_length=254)
address = models.GenericIPAddressField()
managed = models.BooleanField(default=False)
updated = models.DateTimeField()
objects = MXRecordManager.from_queryset(MXRecordQuerySet)()
def is_managed(self):
if not param_tools.get_global_parameter("enable_mx_checks"):
return False
return bool(param_tools.get_global_parameter("valid_mxs").strip())
def __str__(self):
return "{0.name} ({0.address}) for {0.domain} ".format(self)
class DNSBLQuerySet(models.QuerySet):
"""Custom manager for DNSBLResultManager."""
def blacklisted(self):
"""Return blacklisted results."""
return self.exclude(status="")
class DNSBLResult(models.Model):
"""Store a DNSBL query result."""
domain = models.ForeignKey("admin.Domain")
provider = models.CharField(max_length=254, db_index=True)
mx = models.ForeignKey(MXRecord)
status = models.CharField(max_length=45, blank=True, db_index=True)
objects = models.Manager.from_queryset(DNSBLQuerySet)()
class Meta:
app_label = "admin"
unique_together = [("domain", "provider", "mx")]
modoboa/admin/serializers.py
+14
View File
@@ -17,7 +17,9 @@ from modoboa.core import models as core_models
from modoboa.core import signals as core_signals
from modoboa.lib import exceptions as lib_exceptions
from modoboa.lib import permissions, email_utils, fields as lib_fields
from modoboa.parameters import tools as param_tools
from . import lib
from . import models
@@ -31,6 +33,18 @@ class DomainSerializer(serializers.ModelSerializer):
"pk", "name", "quota", "default_mailbox_quota", "enabled", "type",
)
def validate_name(self, value):
"""Check name constraints."""
domains_must_have_authorized_mx = (
param_tools.get_global_parameter("domains_must_have_authorized_mx")
)
user = self.context["request"].user
if domains_must_have_authorized_mx and not user.is_superuser:
if not lib.domain_has_authorized_mx(value):
raise serializers.ValidationError(
_("No authorized MX record found for this domain"))
return value
def validate(self, data):
"""Check quota values."""
quota = data.get("quota", 0)
modoboa/admin/tests/test_api.py
+33
View File
@@ -6,6 +6,9 @@ from __future__ import unicode_literals
import copy
import json
import dns.resolver
from mock import patch
from django.core.urlresolvers import reverse
from rest_framework.authtoken.models import Token
@@ -15,6 +18,7 @@ from modoboa.core import factories as core_factories
from modoboa.core import models as core_models
from modoboa.lib.tests import ModoAPITestCase
from . import utils
from .. import factories
from .. import models
@@ -68,6 +72,35 @@ class DomainAPITestCase(ModoAPITestCase):
url, {"name": "test4.com", "default_mailbox_quota": 10})
self.assertEqual(response.status_code, 403)
@patch.object(dns.resolver.Resolver, "query")
@patch("socket.gethostbyname")
def test_create_domain_with_mx_check(self, mock_gethostbyname, mock_query):
"""Check domain creation when MX check is activated."""
self.set_global_parameter("enable_admin_limits", False, app="limits")
self.set_global_parameter("valid_mxs", "1.2.3.4")
self.set_global_parameter("domains_must_have_authorized_mx", True)
reseller = core_factories.UserFactory(
username="reseller", groups=("Resellers", ))
token = Token.objects.create(user=reseller)
self.client.credentials(HTTP_AUTHORIZATION='Token ' + token.key)
url = reverse("api:domain-list")
mock_query.return_value = [utils.FakeDNSAnswer("mail.ok.com")]
mock_gethostbyname.return_value = "1.2.3.5"
response = self.client.post(
url, {"name": "test3.com", "quota": 0, "default_mailbox_quota": 10}
)
self.assertEqual(response.status_code, 400)
self.assertEqual(
response.json()["name"][0],
"No authorized MX record found for this domain")
mock_gethostbyname.return_value = "1.2.3.4"
response = self.client.post(
url, {"name": "test3.com", "quota": 0, "default_mailbox_quota": 10}
)
self.assertEqual(response.status_code, 201)
def test_update_domain(self):
"""Check domain update."""
domain = models.Domain.objects.get(name="test.com")
modoboa/admin/tests/test_domain.py
+32
View File
@@ -4,11 +4,16 @@
from __future__ import unicode_literals
import dns.resolver
from mock import patch
from django.core.urlresolvers import reverse
from modoboa.core import factories as core_factories
from modoboa.core.models import User
from modoboa.lib.tests import ModoTestCase
from . import utils
from .. import factories
from ..models import Domain, Alias
@@ -181,6 +186,33 @@ class DomainTestCase(ModoTestCase):
self.assertContains(response, "value=\"500\"")
self.assertContains(response, "value=\"50\"")
@patch.object(dns.resolver.Resolver, "query")
@patch("socket.gethostbyname")
def test_create_and_check_mx(self, mock_gethostbyname, mock_query):
"""Check for authorized MX record."""
reseller = core_factories.UserFactory(
username="reseller", groups=("Resellers", ))
self.client.force_login(reseller)
self.set_global_parameter("enable_admin_limits", False, app="limits")
self.set_global_parameter("valid_mxs", "1.2.3.4")
self.set_global_parameter("domains_must_have_authorized_mx", True)
mock_query.return_value = [utils.FakeDNSAnswer("mail.ok.com")]
mock_gethostbyname.return_value = "1.2.3.5"
values = {
"name": "pouet.com", "quota": 0, "default_mailbox_quota": 0,
"create_dom_admin": True, "dom_admin_username": "toto",
"create_aliases": True, "type": "domain", "stepid": "step3",
"with_mailbox": True
}
self.ajax_post(reverse("admin:domain_add"), values, 400)
self.assertFalse(
Domain.objects.filter(name=values["name"]).exists())
mock_gethostbyname.return_value = "1.2.3.4"
self.ajax_post(reverse("admin:domain_add"), values)
def test_modify(self):
"""Test the modification of a domain
modoboa/admin/tests/test_import_.py
+36
View File
@@ -2,12 +2,17 @@
from __future__ import unicode_literals
from mock import patch
import dns.resolver
from django.core.files.base import ContentFile
from django.core.urlresolvers import reverse
from modoboa.core import factories as core_factories
from modoboa.core.models import User
from modoboa.lib.tests import ModoTestCase
from . import utils
from .. import factories
from ..models import Domain, Alias, DomainAlias
@@ -74,6 +79,37 @@ domainalias; domalias1.com; domain1.com; True
response,
"Default mailbox quota cannot be greater than domain quota")
@patch.object(dns.resolver.Resolver, "query")
@patch("socket.gethostbyname")
def test_domain_import_with_mx_check(self, mock_gethostbyname, mock_query):
"""Check domain import when MX check is enabled."""
reseller = core_factories.UserFactory(
username="reseller", groups=("Resellers", ))
self.client.force_login(reseller)
self.set_global_parameter("valid_mxs", "1.2.3.4")
self.set_global_parameter("domains_must_have_authorized_mx", True)
mock_query.return_value = [utils.FakeDNSAnswer("mail.ok.com")]
mock_gethostbyname.return_value = "1.2.3.5"
f = ContentFile(
b"domain; domain1.com; 100; 1; True", name="domains.csv")
resp = self.client.post(
reverse("admin:domain_import"), {
"sourcefile": f
}
)
self.assertContains(resp, "No authorized MX record found for domain")
mock_gethostbyname.return_value = "1.2.3.4"
f.seek(0)
resp = self.client.post(
reverse("admin:domain_import"), {
"sourcefile": f
}
)
self.assertTrue(
Domain.objects.filter(name="domain1.com").exists())
def test_import_domains_with_conflict(self):
f = ContentFile(b"""domain;test.alias;100;10;True
domainalias;test.alias;test.com;True
modoboa/admin/tests/utils.py
+8
View File
@@ -0,0 +1,8 @@
"""Test utilities."""
class FakeDNSAnswer(object):
"""Fake answer."""
def __init__(self, exchange):
self.exchange = exchange
modoboa/core/tests/test_views.py
+1
View File
@@ -157,6 +157,7 @@ class SettingsTestCase(ModoTestCase):
"limits-deflt_user_quota_limit": "0",
"core-hide_features_widget": "False",
"core-inactive_account_threshold": "30",
"admin-domains_must_have_authorized_mx": False,
}
def test_get_settings(self):