Ticket #3138: dns_check.2.py

# -*- coding: UTF-8 -*-

"""
Demo code for a DNS resolving validator. You need to rearrange this for your needs.

you'll need dnspython >= 1.4.0  from  http://www.dnspython.org/

requires two settings: DNS_LIFETIME (time for a single dns query)
and DNS_TIMEOUT (timeout for completely resolving one domain name).

to make use of the asynchronous lookup feature, you need to call hasResolvableDomain.prepare()
for each form field with this validator before validation starts.


I don't recommend to use this as a validator
since DNS can have transient errors.

- mir@noris.de
"""

import re
import dns.resolver, dns.name, dns.exception    # from dnspython



def is_resolvable(domain_string, rdatatypes, resolver):
    """Testet, ob der String in line auflösbar ist, und gibt entsprechend ein Boolean zurück."""
    for rdatatype in rdatatypes:
        answer = resolver.resolve(domain_string, rdatatype)
        if answer != None:
            return True
    return False

class hasResolvableDomain(object):
    """Testet, ob ein Feld eine DNS-auflösbare Domain enthält.
    Das funktioniert für Domains, Email-Adressen, auch für einen ganzen Text,
    bei dem in jeder Zeile ein Eintrag steht.

    >>> from kundebunt.popkern.async_dns import AsyncDNSResolver
    >>> resolver=AsyncDNSResolver()
    >>> nl=chr(10)
    >>> validator = hasResolvableDomain(resolver)
    >>> validator("noris.de")
    >>> validator("blazlsprazl.noris.de")
    Traceback (most recent call last):
        ...
    ValidationError: ['blazlsprazl.noris.de: Die angegebene Domain ist im DNS nicht vorhanden.']
    >>> validator("braslfasl@noris.de")
    >>> validator(nl.join(["noris.de","braslfasl@noris.de"]))
    >>> validator("")
    >>> validator("aosidjasoasdioj")
    Traceback (most recent call last):
        ...
    ValidationError: ['aosidjasoasdioj: Die angegebene Domain ist im DNS nicht vorhanden.']
    >>> validator("..")
    Traceback (most recent call last):
        ...
    ValidationError: ['..: Die angegebene Domain ist im DNS nicht vorhanden.']
    >>> validator = hasResolvableDomain(resolver, ["mx"],"fehlermeldung")
    >>> validator("noris.de")
    >>> validator(nl.join(["blablara@noris.de","service.noris.de"]))
    Traceback (most recent call last):
        ...
    ValidationError: ['service.noris.de: fehlermeldung']

    >>> validator = hasResolvableDomain(resolver, subdomain_field_name='subdomain')
    >>> validator('service.noris.net', {'subdomain': 'raxlbratzl'})
    Traceback (most recent call last):
        ...
    ValidationError: ['raxlbratzl.service.noris.net: Die angegebene Domain ist im DNS nicht vorhanden.']
    >>> validator('raxlbratzl.noris.net', {'subdomain': '.'})
    Traceback (most recent call last):
        ...
    ValidationError: ['raxlbratzl.noris.net: Die angegebene Domain ist im DNS nicht vorhanden.']

    >>> validator('noris.net', {'subdomain': 'service'})
    >>> validator('noris.net', {'subdomain': '.service'})
    """

    def __init__(self, resolver, rrtypes = None, error_message=gettext_lazy("Die angegebene Domain ist im DNS nicht vorhanden."), subdomain_field_name=None, allow_wildcard=False):
        self.resolver = resolver
        if rrtypes == None:
            rrtypes = ["A", "MX"]
        self.rrtypes = [dns.rdatatype.from_text(s) for s in rrtypes]
        self.error_message = error_message
        self.subdomain_field_name = subdomain_field_name
        self.allow_wildcard = allow_wildcard

    def _domains_to_check(self, field_data, all_data):
        """generiert alle zu überprüfenden Domains."""
        if not self.allow_wildcard or field_data!="*":
            for line in field_data.split("\n"):
                line = line.strip()
                if line:
                    at_pos = line.find('@')
                    if at_pos != -1:
                        # es ist eine email-adresse
                        domain = line[at_pos+1:]
                    else:
                        domain = line
                    if self.subdomain_field_name:
                        subdomain = all_data.get(self.subdomain_field_name,'')
                        if subdomain and subdomain[0] != '.':
                            domain = "%s.%s" % (subdomain, domain)
                    if len(domain)>=2 and domain[0] == '.' and domain[1] != '.':
                        domain = domain[1:]
                    elif len(domain)>=3 and domain[:2] == '*.' and domain[2] != '.':
                        domain = domain[2:]
                    yield domain


    def __call__(self, field_data, all_data=None):
        for domain in self._domains_to_check(field_data, all_data):
            resolvable = is_resolvable(domain + ".", self.rrtypes, self.resolver)
            if not resolvable:
                raise validators.ValidationError, "%s: %s" % (domain,self.error_message)

    def prepare(self, field_data, all_data=None):
        """submits the dns query to the resolver so that the reply can be ready in __call__"""
        for domain in self._domains_to_check(field_data, all_data):
            for rrtype in self.rrtypes:
                self.resolver.submit(domain + ".", rrtype)

import dns.resolver, threading, time
from django.conf import settings


class AsyncDNSResolver(object):
    def __init__(self):
        self.threads = {}
        self.answers = {}  # answers is modified within threads; apply locking!
        self.lock = threading.Lock()

    def submit(self, domain_string, rdatatype):
        query = (domain_string, rdatatype)
        if not query in self.threads and not query in self.answers:
            thread = threading.Thread(target=_run_thread, args=(domain_string, rdatatype, self.answers, self.lock))
            thread.setDaemon(True)
            thread.start()
            self.final_submit_time = time.time()
            self.threads[query] = thread


    def resolve(self, domain_string, rdatatype):
        self.submit(domain_string, rdatatype)
        query = (domain_string, rdatatype)
        if self.threads[query].isAlive():
            self.threads[query].join(self._compute_timeout())
        self.lock.acquire()
        try:
            answer = self.answers.get(query)
        finally:
            self.lock.release()
        return answer

    def _compute_timeout(self):
        return max(settings.DNS_LIFETIME - (time.time() - self.final_submit_time) , 0)


def _run_thread(domain_string, rdatatype, answers, lock):
    resolver = dns.resolver.Resolver()
    resolver.timeout = settings.DNS_TIMEOUT
    resolver.lifetime = settings.DNS_LIFETIME
    try:
        answer = resolver.query(domain_string, rdatatype)
    except (dns.exception.Timeout, dns.resolver.NXDOMAIN, dns.resolver.NoAnswer, dns.name.EmptyLabel):
        answer = None
    lock.acquire()
    try:
        answers[(domain_string, rdatatype)] = answer
    finally:
        lock.release()