nsupdate.info/nsupdate/main/models.py

import re
import base64

import dns.resolver

from django.db import models
from django.contrib.auth.models import User
from django.core.exceptions import ValidationError
from django.core.validators import RegexValidator
from django.conf import settings
from django.db.models.signals import pre_delete
from django.contrib.auth.hashers import make_password
from django.utils.timezone import now

from . import dnstools


class BlacklistedDomain(models.Model):
    domain = models.CharField(
        max_length=256,
        unique=True,
        help_text='Blacklisted domain. Evaluated as regex (search).')

    last_update = models.DateTimeField(auto_now=True)
    created = models.DateTimeField(auto_now_add=True)
    created_by = models.ForeignKey(User, blank=True, null=True)

    def __unicode__(self):
        return u"%s" % (self.domain, )


def domain_blacklist_validator(value):
    for bd in BlacklistedDomain.objects.all():
        if re.search(bd.domain, value):
            raise ValidationError(u'This domain is not allowed')


UPDATE_ALGORITHMS = (
    ('HMAC_SHA512', 'HMAC_SHA512'),
    ('HMAC_SHA384', 'HMAC_SHA384'),
    ('HMAC_SHA256', 'HMAC_SHA256'),
    ('HMAC_SHA224', 'HMAC_SHA224'),
    ('HMAC_SHA1', 'HMAC_SHA1'),
    ('HMAC_MD5', 'HMAC_MD5'),
)


class Domain(models.Model):
    domain = models.CharField(
        max_length=256, unique=True,
        help_text="Name of the zone where dynamic hosts may get added")
    nameserver_ip = models.GenericIPAddressField(
        max_length=256,
        help_text="IP where the dynamic DNS updates for this zone will be sent to")
    nameserver_update_key = models.CharField(
        max_length=256,
        help_text="Shared secret that allows updating this zone (base64 encoded)")
    nameserver_update_algorithm = models.CharField(
        max_length=256, default='HMAC_SHA512', choices=UPDATE_ALGORITHMS)
    public = models.BooleanField(
        default=False,
        help_text="Check to allow any user to add dynamic hosts to this zone - "
                  "if not checked, we'll only allow the owner to add hosts")
    # available means "nameserver for domain operating and reachable" -
    # gets set to False if we have trouble reaching the nameserver
    available = models.BooleanField(
        default=True,
        help_text="Check if nameserver is available/reachable - "
                  "if not checked, we'll pause querying/updating this nameserver for a while")
    comment = models.CharField(
        max_length=256, default='', blank=True, null=True)

    last_update = models.DateTimeField(auto_now=True)
    created = models.DateTimeField(auto_now_add=True)
    created_by = models.ForeignKey(User, blank=True, null=True)

    def __unicode__(self):
        return u"%s" % (self.domain, )

    def generate_ns_secret(self):
        secret = User.objects.make_random_password(length=64)  # 512 bits
        self.nameserver_update_key = key = base64.b64encode(secret)
        self.nameserver_update_algorithm = 'HMAC_SHA512'
        self.save()
        return key

    def get_bind9_algorithm(self):
        mapping = {
            'HMAC_SHA512': 'hmac-sha512',
            'HMAC_SHA384': 'hmac-sha384',
            'HMAC_SHA256': 'hmac-sha256',
            'HMAC_SHA224': 'hmac-sha224',
            'HMAC_SHA1': 'hmac-sha1',
            'HMAC_MD5': 'hmac-md5',
        }
        return mapping.get(self.nameserver_update_algorithm)


class Host(models.Model):
    subdomain = models.CharField(max_length=256, validators=[
        RegexValidator(
            regex=r'^(([a-z0-9][a-z0-9\-]*[a-z0-9])|[a-z0-9])$',
            message='Invalid subdomain: only "a-z", "0-9" and "-" is allowed'
        ),
        domain_blacklist_validator])
    domain = models.ForeignKey(Domain, on_delete=models.CASCADE)
    update_secret = models.CharField(max_length=256)  # gets hashed on save
    comment = models.CharField(
        max_length=256, default='', blank=True, null=True)

    last_update = models.DateTimeField(auto_now=True)
    # when we received the last update for v4/v6 addr
    last_update_ipv4 = models.DateTimeField(blank=True, null=True)
    last_update_ipv6 = models.DateTimeField(blank=True, null=True)
    # how we received the last update for v4/v6 addr
    ssl_update_ipv4 = models.BooleanField()
    ssl_update_ipv6 = models.BooleanField()
    created = models.DateTimeField(auto_now_add=True)
    created_by = models.ForeignKey(
        settings.AUTH_USER_MODEL, related_name='hosts')

    def __unicode__(self):
        return u"%s.%s - %s" % (
            self.subdomain, self.domain.domain, self.comment)

    class Meta(object):
        unique_together = (('subdomain', 'domain'),)

    def get_fqdn(self):
        return '%s.%s' % (self.subdomain, self.domain.domain)

    @classmethod
    def filter_by_fqdn(cls, fqdn, **kwargs):
        # Assuming subdomain has no dots (.) the fqdn is split at the first dot
        splitted = fqdn.split('.', 1)
        if not len(splitted) == 2:
            raise NotImplemented("FQDN has to contain a dot")
        return Host.objects.filter(
            subdomain=splitted[0], domain__domain=splitted[1], **kwargs)

    def get_ipv4(self):
        try:
            return dnstools.query_ns(self.get_fqdn(), 'A', origin=self.domain.domain)
        except (dns.resolver.NXDOMAIN, dns.resolver.NoAnswer, dns.resolver.NoNameservers, dns.resolver.Timeout,
                dnstools.NameServerNotAvailable):
            return 'error'

    def get_ipv6(self):
        try:
            return dnstools.query_ns(self.get_fqdn(), 'AAAA', origin=self.domain.domain)
        except (dns.resolver.NXDOMAIN, dns.resolver.NoAnswer, dns.resolver.NoNameservers, dns.resolver.Timeout,
                dnstools.NameServerNotAvailable):
            return 'error'

    def poke(self, kind, ssl):
        if kind == 'ipv4':
            self.last_update_ipv4 = now()
            self.ssl_update_ipv4 = ssl
        else:
            self.last_update_ipv6 = now()
            self.ssl_update_ipv6 = ssl
        self.save()

    def generate_secret(self):
        # note: we use a quick hasher for the update_secret as expensive
        # more modern hashes might put too much load on the servers. also
        # many update clients might use http without ssl, so it is not too
        # secure anyway.
        secret = User.objects.make_random_password()
        self.update_secret = make_password(
            secret,
            hasher='sha1'
        )
        self.save()
        return secret


def pre_delete_host(sender, **kwargs):
    obj = kwargs['instance']
    try:
        dnstools.delete(obj.get_fqdn(), origin=obj.domain.domain)
    except (dnstools.Timeout, dnstools.NameServerNotAvailable):
        # well, we tried to clean up, but we didn't reach the nameserver
        pass

pre_delete.connect(pre_delete_host, sender=Host)