From 321fa813b26cd8d7efc1b32351aa32669065c553 Mon Sep 17 00:00:00 2001 From: Antoine Nguyen Date: Wed, 14 Sep 2016 15:18:51 +0200 Subject: [PATCH] Few changes. --- .../management/commands/subcommands/_mx.py | 150 ++++++++++-------- 1 file changed, 80 insertions(+), 70 deletions(-) diff --git a/modoboa/admin/management/commands/subcommands/_mx.py b/modoboa/admin/management/commands/subcommands/_mx.py index 0b9a459af..92dabb429 100644 --- a/modoboa/admin/management/commands/subcommands/_mx.py +++ b/modoboa/admin/management/commands/subcommands/_mx.py @@ -41,17 +41,17 @@ class CheckMXRecords(BaseCommand): def add_arguments(self, parser): """Add extra arguments to command.""" parser.add_argument( - "--valid-mx", type=str, action='append', default=[], + "--valid-mx", type=str, action="append", default=[], help="Valid MX ip(s) or subnet(s) used to check that domain's MX " "are your own.") parser.add_argument( - "--no-dnsbl", action='store_true', default=False, + "--no-dnsbl", action="store_true", default=False, help="Skip DNSBL queries.") parser.add_argument( - "--email", type=str, action='append', default=[], + "--email", type=str, action="append", default=[], help="One or more email to notify") parser.add_argument( - "--skip-admin-emails", action='store_true', + "--skip-admin-emails", action="store_true", default=False, help="Skip domain's admins email notification.") parser.add_argument( @@ -62,18 +62,19 @@ class CheckMXRecords(BaseCommand): 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""" + """Return one or more `models.MXRecord` for `domain`. + + DNS queries are not performed while `ttl` (in seconds) is still valid. + """ now = datetime.now() records = models.MXRecord.objects.filter(domain=domain, updated__gt=now) - if records.count(): + if records.exists(): for record in records: yield record raise StopIteration() - else: - models.MXRecord.objects.filter(domain=domain).delete() + models.MXRecord.objects.filter(domain=domain).delete() try: answers = dns.resolver.query(domain.name, "MX") except dns.resolver.NoAnswer: @@ -96,7 +97,7 @@ class CheckMXRecords(BaseCommand): if address is not None: record = models.MXRecord.objects.create( domain=domain, - name=str(answer.exchange).strip('.'), + name=str(answer.exchange).strip("."), address=address, updated=now + delta) yield record @@ -137,90 +138,99 @@ class CheckMXRecords(BaseCommand): models.DNSBLResult.objects.bulk_create(to_create) if not alerts: return - emails = options['email'] - if not options['skip_admin_emails']: - emails.extend([a.email for a in domain.admins if a.email]) - if emails: - content = render_to_string( - "admin/notifications/domain_in_dnsbl.html", { - "domain": domain, "alerts": alerts - }) - subject = _("[modoboa] DNSBL issue(s) for domain {}").format( - domain.name) - for email in emails: - status, msg = email_utils.sendmail_simple( - self.sender, email, - subject=subject, content=content) - if not status: - print(msg) + emails = options["email"] + if not options["skip_admin_emails"]: + emails.extend( + domain.admins.exclude(email="").values_list("email", flat=True) + ) + if not len(emails): + return + content = render_to_string( + "admin/notifications/domain_in_dnsbl.html", { + "domain": domain, "alerts": alerts + }) + subject = _("[modoboa] DNSBL issue(s) for domain {}").format( + domain.name) + for email in emails: + status, msg = email_utils.sendmail_simple( + self.sender, email, + subject=subject, content=content) + if not status: + print(msg) def check_valid_mx(self, domain, mx_list, valid_mx=None, **options): - """Check that domain's MX record exist and is contains in `valid_mx` if - the option is provided""" + """Check that domain's MX record exist. + + If `valid_mx` is provided, retrieved MX records must be + contained in it. + """ alerts = [] check = False mxs = [ipaddress.ip_address(u"%s" % mx.address) for mx in mx_list] - if valid_mx: + if not mxs: + alerts.append(_("Domain {} as no MX record").format(domain)) + elif valid_mx: for subnet in valid_mx: check = True in [mx in subnet for mx in mxs] if check is True: break - if not mxs: - alerts.append(_("Domain {} as no MX record").format(domain)) - elif valid_mx and check is False: - mx_names = ["{0.name} ({0.address})".format(mx) for mx in mx_list] - args = (domain, ", ".join(mx_names)) - alerts.append( - _("Domain {0} got an invalid MX record. Got {1}").format(*args) - ) + if check is False: + mx_names = [ + "{0.name} ({0.address})".format(mx) for mx in mx_list] + alerts.append( + _("Domain {0} got an invalid MX record. Got {1}").format( + domain, ", ".join(mx_names)) + ) if not alerts: return - emails = options['email'] - if not options['skip_admin_emails']: - emails.extend([a.email for a in domain.admins if a.email]) - if emails: - content = render_to_string( - "admin/notifications/domain_invalid_mx.html", { - "domain": domain, "valid_mx": valid_mx, "alerts": alerts - }) - subject = _("[modoboa] MX issue(s) for domain {}").format( - domain.name) - for email in emails: - status, msg = email_utils.sendmail_simple( - self.sender, email, - subject=subject, content=content) - if not status: - print(msg) + emails = options["email"] + if not options["skip_admin_emails"]: + emails.extend( + domain.admins.exclude(email="").values_list("email", flat=True) + ) + if not len(emails): + return + content = render_to_string( + "admin/notifications/domain_invalid_mx.html", { + "domain": domain, "valid_mx": valid_mx, "alerts": alerts + }) + subject = _("[modoboa] MX issue(s) for domain {}").format( + domain.name) + for email in emails: + status, msg = email_utils.sendmail_simple( + self.sender, email, + subject=subject, content=content) + if not status: + print(msg) 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)) if parameters.get_admin("ENABLE_MX_CHECKS") != "no": self.check_valid_mx(domain, mx_list, **options) - if parameters.get_admin("ENABLE_DNSBL_CHECKS") == "no": + condition = ( + parameters.get_admin("ENABLE_DNSBL_CHECKS") == "no" or + options["no_dnsbl"] is True) + if condition or not mx_list: return - elif options['no_dnsbl'] is True: - return - elif mx_list: - jobs = [ - gevent.spawn(self.query_dnsbl, mx_list, provider) - for provider in self.providers] - gevent.joinall(jobs, timeout) - for job in jobs: - if not job.successful(): - continue - provider, results = job.value - self.store_dnsbl_result(domain, provider, results, **options) + jobs = [ + gevent.spawn(self.query_dnsbl, mx_list, provider) + for provider in self.providers] + gevent.joinall(jobs, timeout) + for job in jobs: + if not job.successful(): + continue + provider, results = job.value + self.store_dnsbl_result(domain, provider, results, **options) def handle(self, *args, **options): """Command entry point.""" # Check that user provide valid network addresses - valid_mx = options['valid_mx'] - options['valid_mx'] = [ipaddress.ip_network(u"%s" % v) - for v in valid_mx] + valid_mx = options["valid_mx"] + options["valid_mx"] = [ + ipaddress.ip_network(u"{}".format(v)) for v in valid_mx] # Remove deprecated records first models.DNSBLResult.objects.exclude(