django-orchestra-test/orchestra/contrib/domains/models.py

342 lines
13 KiB
Python
Raw Normal View History

from django.core.exceptions import ValidationError
2014-05-08 16:59:35 +00:00
from django.db import models
2014-11-27 19:17:26 +00:00
from django.utils.translation import ungettext, ugettext_lazy as _
2014-05-08 16:59:35 +00:00
2014-10-30 16:34:02 +00:00
from orchestra.core.validators import validate_ipv4_address, validate_ipv6_address, validate_ascii
2014-09-26 15:05:20 +00:00
from orchestra.utils.python import AttrDict
2014-05-08 16:59:35 +00:00
from . import settings, validators, utils
class DomainQuerySet(models.QuerySet):
def get_parent(self, name, top=False):
""" get the next domain on the chain """
split = name.split('.')
parent = None
for i in range(1, len(split)-1):
name = '.'.join(split[i:])
domain = Domain.objects.filter(name=name)
if domain:
parent = domain.get()
if not top:
return parent
return parent
2014-05-08 16:59:35 +00:00
class Domain(models.Model):
2016-02-19 10:11:28 +00:00
name = models.CharField(_("name"), max_length=256, unique=True, db_index=True,
2015-04-05 10:46:24 +00:00
help_text=_("Domain or subdomain name."),
validators=[
validators.validate_domain_name,
validators.validate_allowed_domain
])
2014-11-14 15:51:18 +00:00
account = models.ForeignKey('accounts.Account', verbose_name=_("Account"), blank=True,
2015-04-05 10:46:24 +00:00
related_name='domains', help_text=_("Automatically selected for subdomains."))
2014-11-14 15:51:18 +00:00
top = models.ForeignKey('domains.Domain', null=True, related_name='subdomain_set',
editable=False, verbose_name=_("top domain"))
2015-07-15 10:35:21 +00:00
serial = models.IntegerField(_("serial"), default=utils.generate_zone_serial, editable=False,
help_text=_("A revision number that changes whenever this domain is updated."))
refresh = models.CharField(_("refresh"), max_length=16, blank=True,
2015-07-15 10:35:21 +00:00
validators=[validators.validate_zone_interval],
help_text=_("The time a secondary DNS server waits before querying the primary DNS "
"server's SOA record to check for changes. When the refresh time expires, "
"the secondary DNS server requests a copy of the current SOA record from "
"the primary. The primary DNS server complies with this request. "
"The secondary DNS server compares the serial number of the primary DNS "
"server's current SOA record and the serial number in it's own SOA record. "
"If they are different, the secondary DNS server will request a zone "
"transfer from the primary DNS server. "
"The default value is <tt>%s</tt>.") % settings.DOMAINS_DEFAULT_REFRESH)
retry = models.CharField(_("retry"), max_length=16, blank=True,
2015-07-15 10:35:21 +00:00
validators=[validators.validate_zone_interval],
help_text=_("The time a secondary server waits before retrying a failed zone transfer. "
"Normally, the retry time is less than the refresh time. "
"The default value is <tt>%s</tt>.") % settings.DOMAINS_DEFAULT_RETRY)
expire = models.CharField(_("expire"), max_length=16, blank=True,
2015-07-15 10:35:21 +00:00
validators=[validators.validate_zone_interval],
help_text=_("The time that a secondary server will keep trying to complete a zone "
"transfer. If this time expires prior to a successful zone transfer, "
"the secondary server will expire its zone file. This means the secondary "
"will stop answering queries. "
"The default value is <tt>%s</tt>.") % settings.DOMAINS_DEFAULT_EXPIRE)
min_ttl = models.CharField(_("min TTL"), max_length=16, blank=True,
2015-07-15 10:35:21 +00:00
validators=[validators.validate_zone_interval],
help_text=_("The minimum time-to-live value applies to all resource records in the "
"zone file. This value is supplied in query responses to inform other "
"servers how long they should keep the data in cache. "
"The default value is <tt>%s</tt>.") % settings.DOMAINS_DEFAULT_MIN_TTL)
2014-05-08 16:59:35 +00:00
objects = DomainQuerySet.as_manager()
2015-04-02 16:14:55 +00:00
def __str__(self):
2014-05-08 16:59:35 +00:00
return self.name
2014-10-03 14:02:11 +00:00
@property
2014-05-08 16:59:35 +00:00
def origin(self):
return self.top or self
2014-10-03 14:02:11 +00:00
@property
2014-07-17 16:09:24 +00:00
def is_top(self):
2014-10-04 09:29:18 +00:00
# don't cache, don't replace by top_id
2016-03-31 16:02:50 +00:00
try:
return not bool(self.top)
except Domain.DoesNotExist:
return False
2014-07-17 16:09:24 +00:00
2014-11-05 20:22:01 +00:00
@property
def subdomains(self):
return Domain.objects.filter(name__regex='\.%s$' % self.name)
2015-03-01 11:56:54 +00:00
def clean(self):
self.name = self.name.lower()
def save(self, *args, **kwargs):
""" create top relation """
update = False
if not self.pk:
2015-03-29 16:10:07 +00:00
top = self.get_parent(top=True)
2015-03-01 11:56:54 +00:00
if top:
self.top = top
self.account_id = self.account_id or top.account_id
else:
update = True
super(Domain, self).save(*args, **kwargs)
if update:
for domain in self.subdomains.exclude(pk=self.pk):
# queryset.update() is not used because we want to trigger backend to delete ex-topdomains
domain.top = self
2016-01-28 11:18:28 +00:00
domain.save(update_fields=('top',))
2015-03-01 11:56:54 +00:00
2014-11-27 19:17:26 +00:00
def get_description(self):
if self.is_top:
num = self.subdomains.count()
return ungettext(
_("top domain with one subdomain"),
_("top domain with %d subdomains") % num,
num)
return _("subdomain")
2014-10-23 15:38:46 +00:00
def get_absolute_url(self):
return 'http://%s' % self.name
2015-05-05 19:42:55 +00:00
def get_declared_records(self):
2014-10-03 14:02:11 +00:00
""" proxy method, needed for input validation, see helpers.domain_for_validation """
2014-05-08 16:59:35 +00:00
return self.records.all()
2014-10-04 09:29:18 +00:00
def get_subdomains(self):
2014-10-03 14:02:11 +00:00
""" proxy method, needed for input validation, see helpers.domain_for_validation """
2014-11-10 15:40:51 +00:00
return self.origin.subdomain_set.all().prefetch_related('records')
2014-05-08 16:59:35 +00:00
2015-03-29 16:10:07 +00:00
def get_parent(self, top=False):
return type(self).objects.get_parent(self.name, top=top)
2014-05-08 16:59:35 +00:00
def render_zone(self):
origin = self.origin
zone = origin.render_records()
2015-03-04 21:06:16 +00:00
tail = []
2014-11-10 15:40:51 +00:00
for subdomain in origin.get_subdomains():
2015-03-04 21:06:16 +00:00
if subdomain.name.startswith('*'):
# This subdomains needs to be rendered last in order to avoid undesired matches
tail.append(subdomain)
else:
zone += subdomain.render_records()
for subdomain in sorted(tail, key=lambda x: len(x.name), reverse=True):
2014-05-08 16:59:35 +00:00
zone += subdomain.render_records()
return zone.strip()
2014-05-08 16:59:35 +00:00
def refresh_serial(self):
""" Increases the domain serial number by one """
serial = utils.generate_zone_serial()
if serial <= self.serial:
num = int(str(self.serial)[8:]) + 1
if num >= 99:
raise ValueError('No more serial numbers for today')
serial = str(self.serial)[:8] + '%.2d' % num
serial = int(serial)
self.serial = serial
2016-01-28 11:18:28 +00:00
self.save(update_fields=('serial',))
2014-05-08 16:59:35 +00:00
2016-03-09 10:06:26 +00:00
def get_default_soa(self):
return ' '.join([
"%s." % settings.DOMAINS_DEFAULT_NAME_SERVER,
utils.format_hostmaster(settings.DOMAINS_DEFAULT_HOSTMASTER),
str(self.serial),
self.refresh or settings.DOMAINS_DEFAULT_REFRESH,
self.retry or settings.DOMAINS_DEFAULT_RETRY,
self.expire or settings.DOMAINS_DEFAULT_EXPIRE,
self.min_ttl or settings.DOMAINS_DEFAULT_MIN_TTL,
])
def get_default_records(self):
defaults = []
if self.is_top:
for ns in settings.DOMAINS_DEFAULT_NS:
defaults.append(AttrDict(
type=Record.NS,
value=ns
))
soa = self.get_default_soa()
defaults.insert(0, AttrDict(
type=Record.SOA,
value=soa
))
for mx in settings.DOMAINS_DEFAULT_MX:
defaults.append(AttrDict(
type=Record.MX,
value=mx
))
default_a = settings.DOMAINS_DEFAULT_A
if default_a:
defaults.append(AttrDict(
type=Record.A,
value=default_a
))
default_aaaa = settings.DOMAINS_DEFAULT_AAAA
if default_aaaa:
defaults.append(AttrDict(
type=Record.AAAA,
value=default_aaaa
))
return defaults
def record_is_implicit(self, record, types):
if record.type not in types:
if record.type is Record.NS:
if self.is_top:
return True
elif record.type is Record.SOA:
if self.is_top:
return True
else:
has_a = Record.A in types
has_aaaa = Record.AAAA in types
is_host = self.is_top or not types or has_a or has_aaaa
if is_host:
if record.type is Record.MX:
return True
elif not has_a and not has_aaaa:
return True
return False
2015-05-05 19:42:55 +00:00
def get_records(self):
2016-03-09 10:06:26 +00:00
types = set()
2015-05-05 19:42:55 +00:00
records = utils.RecordStorage()
for record in self.get_declared_records():
2016-03-09 10:06:26 +00:00
types.add(record.type)
2014-05-08 16:59:35 +00:00
if record.type == record.SOA:
# Update serial and insert at 0
value = record.value.split()
value[2] = str(self.serial)
2014-11-14 15:51:18 +00:00
records.insert(0, AttrDict(
type=record.SOA,
ttl=record.get_ttl(),
value=' '.join(value)
))
2014-05-08 16:59:35 +00:00
else:
2014-11-14 15:51:18 +00:00
records.append(AttrDict(
type=record.type,
ttl=record.get_ttl(),
value=record.value
))
2016-03-09 10:06:26 +00:00
for record in self.get_default_records():
if self.record_is_implicit(record, types):
if record.type is Record.SOA:
records.insert(0, record)
else:
records.append(record)
2015-05-05 19:42:55 +00:00
return records
def render_records(self):
2014-05-08 16:59:35 +00:00
result = ''
2015-05-05 19:42:55 +00:00
for record in self.get_records():
2014-09-26 15:05:20 +00:00
name = '{name}.{spaces}'.format(
2014-11-14 15:51:18 +00:00
name=self.name,
spaces=' ' * (37-len(self.name))
2014-09-26 15:05:20 +00:00
)
ttl = record.get('ttl', settings.DOMAINS_DEFAULT_TTL)
ttl = '{spaces}{ttl}'.format(
2014-11-14 15:51:18 +00:00
spaces=' ' * (7-len(ttl)),
ttl=ttl
2014-09-26 15:05:20 +00:00
)
type = '{type} {spaces}'.format(
2014-11-14 15:51:18 +00:00
type=record.type,
spaces=' ' * (7-len(record.type))
2014-09-26 15:05:20 +00:00
)
result += '{name} {ttl} IN {type} {value}\n'.format(
2014-11-14 15:51:18 +00:00
name=name,
ttl=ttl,
type=type,
value=record.value
2014-09-26 15:05:20 +00:00
)
2014-05-08 16:59:35 +00:00
return result
2015-05-05 19:42:55 +00:00
def has_default_mx(self):
records = self.get_records()
for record in records.by_type('MX'):
for default in settings.DOMAINS_DEFAULT_MX:
if record.value.endswith(' %s' % default.split()[-1]):
return True
return False
2014-05-08 16:59:35 +00:00
class Record(models.Model):
""" Represents a domain resource record """
MX = 'MX'
NS = 'NS'
CNAME = 'CNAME'
A = 'A'
AAAA = 'AAAA'
SRV = 'SRV'
TXT = 'TXT'
SPF = 'SPF'
2014-05-08 16:59:35 +00:00
TYPE_CHOICES = (
(MX, "MX"),
(NS, "NS"),
(CNAME, "CNAME"),
(A, _("A (IPv4 address)")),
(AAAA, _("AAAA (IPv6 address)")),
(SRV, "SRV"),
(TXT, "TXT"),
(SPF, "SPF"),
2014-05-08 16:59:35 +00:00
)
VALIDATORS = {
MX: (validators.validate_mx_record,),
NS: (validators.validate_zone_label,),
A: (validate_ipv4_address,),
AAAA: (validate_ipv6_address,),
CNAME: (validators.validate_zone_label,),
TXT: (validate_ascii, validators.validate_quoted_record),
SPF: (validate_ascii, validators.validate_quoted_record),
SRV: (validators.validate_srv_record,),
}
2014-05-08 16:59:35 +00:00
domain = models.ForeignKey(Domain, verbose_name=_("domain"), related_name='records')
2014-09-26 15:05:20 +00:00
ttl = models.CharField(_("TTL"), max_length=8, blank=True,
2015-04-05 10:46:24 +00:00
help_text=_("Record TTL, defaults to %s") % settings.DOMAINS_DEFAULT_TTL,
validators=[validators.validate_zone_interval])
2014-09-26 15:05:20 +00:00
type = models.CharField(_("type"), max_length=32, choices=TYPE_CHOICES)
value = models.CharField(_("value"), max_length=256,
help_text=_("MX, NS and CNAME records sould end with a dot."))
2014-05-08 16:59:35 +00:00
2015-04-02 16:14:55 +00:00
def __str__(self):
2014-09-26 15:05:20 +00:00
return "%s %s IN %s %s" % (self.domain, self.get_ttl(), self.type, self.value)
2014-05-08 16:59:35 +00:00
def clean(self):
""" validates record value based on its type """
# validate value
2015-05-12 12:38:40 +00:00
if self.type != self.TXT:
self.value = self.value.lower().strip()
if self.type:
for validator in self.VALIDATORS.get(self.type, []):
try:
validator(self.value)
except ValidationError as error:
raise ValidationError({
'value': error,
})
2014-09-26 15:05:20 +00:00
def get_ttl(self):
return self.ttl or settings.DOMAINS_DEFAULT_TTL