Dyomedea
/
binder
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Fixes some python style issues. 

No changes in binders functionality are included in this commit, just improved
python style
This commit is contained in:
Daniel Roschka 2015-06-13 12:05:42 +02:00
parent ec85c465d8
commit 20d785840b
12 changed files with 294 additions and 250 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
binder/admin.py
View File

@ -4,6 +4,7 @@ from models import BindServer, Key
from django.contrib import admin from django.contrib import admin
from django.forms import ModelForm, ValidationError from django.forms import ModelForm, ValidationError
class BindServerAdminForm(ModelForm): class BindServerAdminForm(ModelForm):
def clean_statistics_port(self): def clean_statistics_port(self):
port = self.cleaned_data["statistics_port"] port = self.cleaned_data["statistics_port"]
@ -13,7 +14,6 @@ class BindServerAdminForm(ModelForm):
params={'port': port}) params={'port': port})
return self.cleaned_data["statistics_port"] return self.cleaned_data["statistics_port"]
def clean_dns_port(self): def clean_dns_port(self):
port = self.cleaned_data["dns_port"] port = self.cleaned_data["dns_port"]
if port < 1 or port > 65535: if port < 1 or port > 65535:
@ -31,7 +31,7 @@ class BindServerAdmin(admin.ModelAdmin):
class KeyAdminForm(ModelForm): class KeyAdminForm(ModelForm):
def clean_data(self): def clean_data(self):
try: try:
keyring = dns.tsigkeyring.from_text({'': self.cleaned_data["data"]}) dns.tsigkeyring.from_text({'': self.cleaned_data["data"]})
except binascii.Error as err: except binascii.Error as err:
raise ValidationError("Invalid key data: %(error)s", raise ValidationError("Invalid key data: %(error)s",
params={'error': err}) params={'error': err})

28
binder/exceptions.py
View File

@ -1,32 +1,34 @@
### Binder Exceptions # Binder Exceptions
class TransferException(Exception): class TransferException(Exception):
"""
Thrown when an AXFR transfer cannot be performed. """Thrown when an AXFR transfer cannot be performed."""
"""
pass pass
class ZoneException(Exception): class ZoneException(Exception):
"""
Thrown when there is an issue dealing with a """Thrown when there is an issue dealing with a DNS zone."""
DNS zone.
"""
pass pass
class RecordException(Exception): class RecordException(Exception):
"""
Thrown when there is an issue dealign with """Thrown when there is an issue dealign with a Record.
a Record.
* Adding or deleting. * Adding or deleting.
""" """
pass pass
class KeyringException(Exception): class KeyringException(Exception):
"""
Thrown when there is a problem creating the keyring. """Thrown when there is a problem creating the keyring.
* When the length/padding of the TSIG data is incorrect. * When the length/padding of the TSIG data is incorrect.
""" """

62
binder/forms.py
View File

@ -1,4 +1,4 @@
### Binder Forms # Binder Forms
# 3rd Party # 3rd Party
from django import forms from django import forms
@ -8,10 +8,11 @@ from django.forms import ValidationError
# App Imports # App Imports
from models import Key from models import Key
### Custom Form Fields
class CustomUnicodeListField(forms.CharField): class CustomUnicodeListField(forms.CharField):
""" Convert unicode item list to list of strings. """
"""Convert unicode item list to list of strings."""
def clean(self, value): def clean(self, value):
try: try:
string_list = [str(cur_rr) for cur_rr in eval(value)] string_list = [str(cur_rr) for cur_rr in eval(value)]
@ -19,12 +20,17 @@ class CustomUnicodeListField(forms.CharField):
raise ValidationError("Error in converting Unicode list to list of Strings: %r" % value) raise ValidationError("Error in converting Unicode list to list of Strings: %r" % value)
return string_list return string_list
class CustomStringPeriodSuffix(forms.CharField): class CustomStringPeriodSuffix(forms.CharField):
""" Convert unicode to string and make sure period is last character. """
### This seems very unclean. Need a better to way to complete the fqdn """Convert unicode to string and make sure period is last character.
### depending on if it ends in a period.
### TODO(jforman): Add Regex check in here for valid rr data This seems very unclean. Need a better to way to complete the fqdn
### http://www.zytrax.com/books/dns/apa/names.html depending on if it ends in a period.
TODO(jforman): Add Regex check in here for valid rr data
http://www.zytrax.com/books/dns/apa/names.html
"""
def clean(self, value): def clean(self, value):
try: try:
new_string = str(value) new_string = str(value)
@ -34,42 +40,60 @@ class CustomStringPeriodSuffix(forms.CharField):
raise ValidationError("Unable to stick a period on the end of your input: %r" % value) raise ValidationError("Unable to stick a period on the end of your input: %r" % value)
return new_string return new_string
### Form Models
class FormAddForwardRecord(forms.Form): class FormAddForwardRecord(forms.Form):
""" Form used to add a Forward DNS record. """
"""Form used to add a Forward DNS record."""
dns_server = forms.CharField(max_length=100) dns_server = forms.CharField(max_length=100)
record_name = forms.RegexField(max_length=100, regex="^[a-zA-Z0-9-_]+$", required=False) record_name = forms.RegexField(max_length=100,
regex="^[a-zA-Z0-9-_]+$",
required=False)
record_type = forms.ChoiceField(choices=settings.RECORD_TYPE_CHOICES) record_type = forms.ChoiceField(choices=settings.RECORD_TYPE_CHOICES)
zone_name = forms.CharField(max_length=100) zone_name = forms.CharField(max_length=100)
record_data = forms.GenericIPAddressField() record_data = forms.GenericIPAddressField()
ttl = forms.ChoiceField(choices=settings.TTL_CHOICES) ttl = forms.ChoiceField(choices=settings.TTL_CHOICES)
create_reverse = forms.BooleanField(required=False) create_reverse = forms.BooleanField(required=False)
key_name = forms.ModelChoiceField(queryset=Key.objects.all(), required=False) key_name = forms.ModelChoiceField(queryset=Key.objects.all(),
required=False)
class FormAddReverseRecord(forms.Form): class FormAddReverseRecord(forms.Form):
""" Form used to add a Reverse (PTR) DNS record. """
"""Form used to add a Reverse (PTR) DNS record."""
dns_server = forms.CharField(max_length=100) dns_server = forms.CharField(max_length=100)
record_name = forms.IntegerField(min_value=0, max_value=255) record_name = forms.IntegerField(min_value=0, max_value=255)
record_type = forms.RegexField(regex=r"^PTR$",error_messages={"invalid" : "The only valid choice here is PTR."}) record_type = forms.RegexField(regex=r"^PTR$",
error_messages={"invalid": "The only valid choice here is PTR."})
zone_name = forms.CharField(max_length=100) zone_name = forms.CharField(max_length=100)
record_data = CustomStringPeriodSuffix(required=True) record_data = CustomStringPeriodSuffix(required=True)
ttl = forms.ChoiceField(choices=settings.TTL_CHOICES) ttl = forms.ChoiceField(choices=settings.TTL_CHOICES)
key_name = forms.ModelChoiceField(queryset=Key.objects.all(), required=False) key_name = forms.ModelChoiceField(queryset=Key.objects.all(),
required=False)
create_reverse = forms.BooleanField(required=False) create_reverse = forms.BooleanField(required=False)
class FormAddCnameRecord(forms.Form): class FormAddCnameRecord(forms.Form):
""" Form used to add a CNAME record. """
"""Form used to add a CNAME record."""
dns_server = forms.CharField(max_length=100) dns_server = forms.CharField(max_length=100)
originating_record = forms.CharField(max_length=100) originating_record = forms.CharField(max_length=100)
cname = forms.RegexField(max_length=100, regex="^[a-zA-Z0-9-_]+$") cname = forms.RegexField(max_length=100, regex="^[a-zA-Z0-9-_]+$")
zone_name = forms.CharField(max_length=256) zone_name = forms.CharField(max_length=256)
ttl = forms.ChoiceField(choices=settings.TTL_CHOICES) ttl = forms.ChoiceField(choices=settings.TTL_CHOICES)
key_name = forms.ModelChoiceField(queryset=Key.objects.all(), required=False) key_name = forms.ModelChoiceField(queryset=Key.objects.all(),
required=False)
class FormDeleteRecord(forms.Form): class FormDeleteRecord(forms.Form):
""" Final form to delete DNS record(s). """
"""Final form to delete DNS record(s)."""
dns_server = forms.CharField(max_length=100) dns_server = forms.CharField(max_length=100)
zone_name = forms.CharField(max_length=256) zone_name = forms.CharField(max_length=256)
rr_list = CustomUnicodeListField() rr_list = CustomUnicodeListField()
key_name = forms.ModelChoiceField(queryset=Key.objects.all(), required=False) key_name = forms.ModelChoiceField(queryset=Key.objects.all(),
required=False)

72
binder/helpers.py
View File

@ -1,10 +1,8 @@
### Binder Helpers # Binder Helpers
# Standard Imports # Standard Imports
import binascii
import re import re
import socket import socket
import sys
# 3rd Party # 3rd Party
import dns.query import dns.query
@ -14,10 +12,12 @@ import dns.tsigkeyring
import dns.update import dns.update
# App Imports # App Imports
from binder import exceptions, models from binder import models
def add_record(dns_server, zone_name, record_name, record_type, record_data, ttl, key_name, create_reverse=False):
""" Parse passed elements and determine which records to create. def add_record(dns_server, zone_name, record_name, record_type, record_data,
ttl, key_name, create_reverse=False):
"""Parse passed elements and determine which records to create.
Args: Args:
String dns_server String dns_server
@ -32,10 +32,10 @@ def add_record(dns_server, zone_name, record_name, record_type, record_data, ttl
Return: Return:
Dict containing {description, output} from record creation Dict containing {description, output} from record creation
""" """
response = [] response = []
response.append({ "description" : "Forward Record Creation: %s.%s" % (record_name, zone_name), response.append({"description": "Forward Record Creation: %s.%s" %
"output" : create_update(dns_server, (record_name, zone_name),
"output": create_update(dns_server,
zone_name, zone_name,
record_name, record_name,
record_type, record_type,
@ -43,8 +43,10 @@ def add_record(dns_server, zone_name, record_name, record_type, record_data, ttl
ttl, ttl,
key_name)}) key_name)})
""" If requested, create a reverse PTR record. """If requested, create a reverse PTR record.
Given the forward record created, resolve its underlying IP. Use that to create the reverse record.
Given the forward record created, resolve its underlying IP.
Use that to create the reverse record.
reverse_ip_fqdn ex: 5.0.20.10.in-addr.arpa. reverse_ip_fqdn ex: 5.0.20.10.in-addr.arpa.
reverse_ip: 5 reverse_ip: 5
reverse_domain: 0.20.10.in-addr.arpa. reverse_domain: 0.20.10.in-addr.arpa.
@ -55,8 +57,8 @@ def add_record(dns_server, zone_name, record_name, record_type, record_data, ttl
# for this reverse DNS record parsing. # for this reverse DNS record parsing.
reverse_ip = re.search(r"([0-9]+).(.*)$", reverse_ip_fqdn).group(1) reverse_ip = re.search(r"([0-9]+).(.*)$", reverse_ip_fqdn).group(1)
reverse_domain = re.search(r"([0-9]+).(.*)$", reverse_ip_fqdn).group(2) reverse_domain = re.search(r"([0-9]+).(.*)$", reverse_ip_fqdn).group(2)
response.append({ "description" : "Reverse Record Creation: %s" % record_data, response.append({"description": "Reverse Record Creation: %s" % record_data,
"output" : create_update(dns_server, "output": create_update(dns_server,
reverse_domain, reverse_domain,
reverse_ip, reverse_ip,
"PTR", "PTR",
@ -66,9 +68,10 @@ def add_record(dns_server, zone_name, record_name, record_type, record_data, ttl
return response return response
def add_cname_record(dns_server, zone_name, cname, originating_record, ttl, key_name):
"""Add a Cname record."""
def add_cname_record(dns_server, zone_name, cname, originating_record, ttl,
key_name):
"""Add a CNAME record."""
output = create_update(dns_server, output = create_update(dns_server,
zone_name, zone_name,
cname, cname,
@ -77,12 +80,13 @@ def add_cname_record(dns_server, zone_name, cname, originating_record, ttl, key_
ttl, ttl,
key_name) key_name)
return [{ "description" : "CNAME %s.%s points to %s" % (cname, zone_name, originating_record), return [{"description": "CNAME %s.%s points to %s" %
"output" : output}] (cname, zone_name, originating_record),
"output": output}]
def delete_record(dns_server, rr_list, key_name): def delete_record(dns_server, rr_list, key_name):
"""Delete a list of DNS records passed as strings in rr_items.""" """Delete a list of DNS records passed as strings in rr_items."""
server = models.BindServer.objects.get(hostname=dns_server) server = models.BindServer.objects.get(hostname=dns_server)
try: try:
@ -99,18 +103,24 @@ def delete_record(dns_server, rr_list, key_name):
record_list = current_rr.split(".", 1) record_list = current_rr.split(".", 1)
record = record_list[0] record = record_list[0]
domain = record_list[1] domain = record_list[1]
dns_update = dns.update.Update(domain, keyring=keyring, keyalgorithm=algorithm) dns_update = dns.update.Update(domain,
keyring=keyring,
keyalgorithm=algorithm)
dns_update.delete(record) dns_update.delete(record)
output = send_dns_update(dns_update, dns_server, server.dns_port, key_name) output = send_dns_update(dns_update,
dns_server,
server.dns_port,
key_name)
delete_response.append({ "description" : "Delete Record: %s" % current_rr, delete_response.append({"description": "Delete Record: %s" % current_rr,
"output" : output }) "output": output})
return delete_response return delete_response
def create_update(dns_server, zone_name, record_name, record_type, record_data, ttl, key_name):
""" Update/Create DNS record of name and type with passed data and ttl. """
def create_update(dns_server, zone_name, record_name, record_type, record_data,
ttl, key_name):
"""Update/Create DNS record of name and type with passed data and ttl."""
server = models.BindServer.objects.get(hostname=dns_server) server = models.BindServer.objects.get(hostname=dns_server)
try: try:
@ -122,14 +132,18 @@ def create_update(dns_server, zone_name, record_name, record_type, record_data,
keyring = transfer_key.create_keyring() keyring = transfer_key.create_keyring()
algorithm = transfer_key.algorithm algorithm = transfer_key.algorithm
dns_update = dns.update.Update(zone_name, keyring=keyring, keyalgorithm=algorithm) dns_update = dns.update.Update(zone_name,
keyring=keyring,
keyalgorithm=algorithm)
dns_update.replace(record_name, ttl, record_type, record_data) dns_update.replace(record_name, ttl, record_type, record_data)
output = send_dns_update(dns_update, dns_server, server.dns_port, key_name) output = send_dns_update(dns_update, dns_server, server.dns_port, key_name)
return output return output
def ip_info(host_name): def ip_info(host_name):
"""Create a dictionary mapping address types to their IP's. """Create a dictionary mapping address types to their IP's.
If an error is encountered, key to error is "Error". If an error is encountered, key to error is "Error".
""" """
info = [] info = []
@ -148,8 +162,9 @@ def ip_info(host_name):
return info return info
def send_dns_update(dns_message, dns_server, port, key_name): def send_dns_update(dns_message, dns_server, port, key_name):
""" Send DNS message to server and return response. """Send DNS message to server and return response.
Args: Args:
Update dns_update Update dns_update
@ -159,14 +174,13 @@ def send_dns_update(dns_message, dns_server, port, key_name):
Returns: Returns:
String output String output
""" """
try: try:
output = dns.query.tcp(dns_message, dns_server, port=port) output = dns.query.tcp(dns_message, dns_server, port=port)
except dns.tsig.PeerBadKey: except dns.tsig.PeerBadKey:
output = ("DNS server %s is not configured for TSIG key: %s." % output = ("DNS server %s is not configured for TSIG key: %s." %
(dns_server, key_name)) (dns_server, key_name))
except dns.tsig.PeerBadSignature: except dns.tsig.PeerBadSignature:
output = ("DNS server %s did like the TSIG signature we sent. Check key %s " output = ("DNS server %s did like the TSIG signature we sent. Check "
"for correctness." % (dns_server, key_name)) "key %s for correctness." % (dns_server, key_name))
return output return output

27
binder/models.py
View File

@ -3,7 +3,6 @@
# Standard Imports # Standard Imports
import binascii import binascii
import socket import socket
import urllib2
# 3rd Party # 3rd Party
from pybindxml import reader as bindreader from pybindxml import reader as bindreader
@ -22,20 +21,24 @@ TSIG_ALGORITHMS = (('HMAC-MD5.SIG-ALG.REG.INT', 'MD5'),
('hmac-sha384', 'SHA384'), ('hmac-sha384', 'SHA384'),
('hmac-sha512', 'SHA512')) ('hmac-sha512', 'SHA512'))
class Key(models.Model): class Key(models.Model):
""" Store and reference TSIG keys.
"""Store and reference TSIG keys.
TODO: Should/Can we encrypt these DNS keys in the DB? TODO: Should/Can we encrypt these DNS keys in the DB?
""" """
name = models.CharField(max_length=255, name = models.CharField(max_length=255,
unique=True, unique=True,
help_text="A human readable name for the key to store, used for " help_text="A human readable name for the key to "
"further references to the key.") "store, used for further references to the key.")
data = models.CharField(max_length=255, data = models.CharField(max_length=255,
help_text="The private part of the TSIG key.") help_text="The private part of the TSIG key.")
algorithm = models.CharField(max_length=255, algorithm = models.CharField(max_length=255,
choices=TSIG_ALGORITHMS, choices=TSIG_ALGORITHMS,
help_text="The algorithm which has been used for the key.") help_text="The algorithm which has been used "
"for the key.")
def __unicode__(self): def __unicode__(self):
return self.name return self.name
@ -56,10 +59,12 @@ class Key(models.Model):
class BindServer(models.Model): class BindServer(models.Model):
""" Store DNS servers and attributes for referencing their
statistics ports. Also reference FK for TSIG transfer keys, """Store DNS servers and attributes for referencing their statistics ports.
if required.
Also reference FK for TSIG transfer keys, if required.
""" """
hostname = models.CharField(max_length=255, hostname = models.CharField(max_length=255,
unique=True, unique=True,
help_text="Host name or IP address of the BIND server.") help_text="Host name or IP address of the BIND server.")
@ -85,7 +90,7 @@ class BindServer(models.Model):
ordering = ["hostname"] ordering = ["hostname"]
def list_zones(self): def list_zones(self):
""" List the DNS zones and attributes. """List the DNS zones and attributes.
TODO: Parse these XML more intelligently. Grab the view name. Any other data available? TODO: Parse these XML more intelligently. Grab the view name. Any other data available?
@ -100,7 +105,7 @@ class BindServer(models.Model):
return zone_data return zone_data
def list_zone_records(self, zone_name): def list_zone_records(self, zone_name):
""" List all records in a specific zone. """List all records in a specific zone.
TODO: Print out current_record in the loop and see if we can parse this more programatically, TODO: Print out current_record in the loop and see if we can parse this more programatically,
rather than just splitting on space. What is the difference between class and type? rather than just splitting on space. What is the difference between class and type?
@ -110,7 +115,6 @@ class BindServer(models.Model):
Returns: Returns:
List of Dicts { String rr_name, String rr_ttl, String rr_class, String rr_type, String rr_data } List of Dicts { String rr_name, String rr_ttl, String rr_class, String rr_type, String rr_data }
""" """
try: try:
transfer_key = Key.objects.get(name=self.default_transfer_key) transfer_key = Key.objects.get(name=self.default_transfer_key)
except Key.DoesNotExist: except Key.DoesNotExist:
@ -154,4 +158,3 @@ class BindServer(models.Model):
record_array.append(rr_dict) record_array.append(rr_dict)
return record_array return record_array

62
binder/tests/disabled_testIntegration.py
View File

@ -2,10 +2,11 @@ from django.core.urlresolvers import reverse
from django.test import TestCase from django.test import TestCase
from django.test.client import Client from django.test.client import Client
from binder import helpers, models from binder import models
class Integration_Tests(TestCase): class Integration_Tests(TestCase):
fixtures = [ "binder/fixtures/binder_test.json" ] fixtures = ["binder/fixtures/binder_test.json"]
def setUp(self): def setUp(self):
self.client = Client() self.client = Client()
@ -13,13 +14,13 @@ class Integration_Tests(TestCase):
def test_Integration_Add_Record(self): def test_Integration_Add_Record(self):
"""Add forward and reverse record on domain1.local.""" """Add forward and reverse record on domain1.local."""
add_dict = { "dns_server" : self.testserver, add_dict = {"dns_server": self.testserver,
"record_name" : "record1", "record_name": "record1",
"record_type" : "A", "record_type": "A",
"zone_name" : "domain1.local", "zone_name": "domain1.local",
"record_data" : "10.254.1.101", "record_data": "10.254.1.101",
"ttl" : 86400, "ttl": 86400,
"create_reverse" : True} "create_reverse": True}
response = self.client.post(reverse("add_record_result"), add_dict) response = self.client.post(reverse("add_record_result"), add_dict)
self.assertEqual(response.status_code, 200) self.assertEqual(response.status_code, 200)
# Make sure that we get two responses (fwd/rev) back from the server. # Make sure that we get two responses (fwd/rev) back from the server.
@ -32,10 +33,9 @@ class Integration_Tests(TestCase):
def test_Integration_Delete_Record(self): def test_Integration_Delete_Record(self):
"""Delete record1.domain1.local""" """Delete record1.domain1.local"""
delete_dict = { "dns_server" : self.testserver, delete_dict = {"dns_server": self.testserver,
"zone_name" : "domain1.local", "zone_name": "domain1.local",
"rr_list" : '[u"record1.domain1.local", u"record2.domain1.local"]', "rr_list": '[u"record1.domain1.local", u"record2.domain1.local"]'}
}
response = self.client.post(reverse("delete_record_result"), delete_dict) response = self.client.post(reverse("delete_record_result"), delete_dict)
self.assertEqual(response.status_code, 200) self.assertEqual(response.status_code, 200)
self.assertEqual(len(response.context["response"]), 2) self.assertEqual(len(response.context["response"]), 2)
@ -44,16 +44,15 @@ class Integration_Tests(TestCase):
self.assertRegexpMatches(dns_update_output, "opcode UPDATE") self.assertRegexpMatches(dns_update_output, "opcode UPDATE")
self.assertRegexpMatches(dns_update_output, "rcode NOERROR") self.assertRegexpMatches(dns_update_output, "rcode NOERROR")
def test_Integration_Add_Cname(self): def test_Integration_Add_Cname(self):
""" Add CNAME cnametest1 after adding associated A record record1.""" """Add CNAME cnametest1 after adding associated A record record1."""
add_dict = { "dns_server" : self.testserver, add_dict = {"dns_server": self.testserver,
"record_name" : "record1", "record_name": "record1",
"record_type" : "A", "record_type": "A",
"zone_name" : "domain1.local", "zone_name": "domain1.local",
"record_data" : "10.254.1.101", "record_data": "10.254.1.101",
"ttl" : 86400, "ttl": 86400,
"create_reverse" : False} "create_reverse": False}
response = self.client.post(reverse("add_record_result"), add_dict) response = self.client.post(reverse("add_record_result"), add_dict)
self.assertEqual(response.status_code, 200) self.assertEqual(response.status_code, 200)
# Make sure that we get two responses (fwd/rev) back from the server. # Make sure that we get two responses (fwd/rev) back from the server.
@ -64,12 +63,11 @@ class Integration_Tests(TestCase):
self.assertRegexpMatches(dns_update_output, "opcode UPDATE") self.assertRegexpMatches(dns_update_output, "opcode UPDATE")
self.assertRegexpMatches(dns_update_output, "rcode NOERROR") self.assertRegexpMatches(dns_update_output, "rcode NOERROR")
cname_dict = { "dns_server" : self.testserver, cname_dict = {"dns_server": self.testserver,
"originating_record" : "record1.domain1.local", "originating_record": "record1.domain1.local",
"cname" : "cnametest1", "cname": "cnametest1",
"zone_name" : "domain1.local", "zone_name": "domain1.local",
"ttl" : 86400, "ttl": 86400}
}
response = self.client.post(reverse("add_cname_result"), cname_dict) response = self.client.post(reverse("add_cname_result"), cname_dict)
self.assertEqual(response.status_code, 200) self.assertEqual(response.status_code, 200)
for current_response in response.context["response"]: for current_response in response.context["response"]:
@ -77,7 +75,6 @@ class Integration_Tests(TestCase):
self.assertRegexpMatches(dns_update_output, "opcode UPDATE") self.assertRegexpMatches(dns_update_output, "opcode UPDATE")
self.assertRegexpMatches(dns_update_output, "rcode NOERROR") self.assertRegexpMatches(dns_update_output, "rcode NOERROR")
def test_Integration_ServerZoneList_ConnectionRefused(self): def test_Integration_ServerZoneList_ConnectionRefused(self):
"""Confirm connection refused on a server zone list.""" """Confirm connection refused on a server zone list."""
dns_server = models.BindServer.objects.get(hostname="testserver1") dns_server = models.BindServer.objects.get(hostname="testserver1")
@ -94,11 +91,14 @@ class Integration_Tests(TestCase):
def test_Integration_ZoneList_MissingTransferKey(self): def test_Integration_ZoneList_MissingTransferKey(self):
"""Attempt to list a zone's records with missing TSIG key. """Attempt to list a zone's records with missing TSIG key.
domain3.local should be configured to require a TSIG key domain3.local should be configured to require a TSIG key
for transfers.""" for transfers.
"""
dns_server = models.BindServer.objects.get(hostname="testserver1") dns_server = models.BindServer.objects.get(hostname="testserver1")
response = self.client.get(reverse("zone_list", args=("testserver1", "domain3.local"))) response = self.client.get(reverse("zone_list", args=("testserver1", "domain3.local")))
self.assertEqual(response.status_code, 200) self.assertEqual(response.status_code, 200)
self.assertEqual(response.context["zone_name"], "domain3.local") self.assertEqual(response.context["zone_name"], "domain3.local")
self.assertEqual(response.context["dns_server"], "testserver1") self.assertEqual(response.context["dns_server"], "testserver1")
self.assertRegexpMatches(str(response.context["errors"]), "Unable to perform AXFR to list zone records. Did you forget to specify a default transfer key?") self.assertRegexpMatches(str(response.context["errors"]),
"Unable to perform AXFR to list zone records. Did you forget to specify a default transfer key?")

98
binder/tests/testForms.py
View File

@ -2,56 +2,55 @@ from django.test import TestCase
from binder import forms from binder import forms
class Form_Tests(TestCase): class Form_Tests(TestCase):
def test_Valid_FormAddRecord(self): def test_Valid_FormAddRecord(self):
"""Test FormAddRecord with valid data, with/without create_reverse.""" """Test FormAddRecord with valid data, with/without create_reverse."""
form_data = {"dns_server":"server1", form_data = {"dns_server": "server1",
"record_name":"record1", "record_name": "record1",
"record_type":"A", "record_type": "A",
"zone_name":"domain.local", "zone_name": "domain.local",
"record_data":"10.254.254.254", "record_data": "10.254.254.254",
"ttl":3600, "ttl": 3600,
"key_name": None, "key_name": None,
"create_reverse" : False} "create_reverse": False}
testform_1 = forms.FormAddForwardRecord(form_data) testform_1 = forms.FormAddForwardRecord(form_data)
self.assertTrue(testform_1.is_valid()) self.assertTrue(testform_1.is_valid())
form_data = {"dns_server":"server1", form_data = {"dns_server": "server1",
"record_name":"record1", "record_name": "record1",
"record_type":"A", "record_type": "A",
"zone_name":"domain.local", "zone_name": "domain.local",
"record_data":"10.254.254.254", "record_data": "10.254.254.254",
"ttl":3600, "ttl": 3600,
"key_name":None, "key_name": None,
"create_reverse":True} "create_reverse": True}
testform_2 = forms.FormAddForwardRecord(form_data) testform_2 = forms.FormAddForwardRecord(form_data)
self.assertTrue(testform_2.is_valid()) self.assertTrue(testform_2.is_valid())
form_data = { "dns_server" : "server1", form_data = {"dns_server": "server1",
"record_name" : 41, "record_name": 41,
"record_type" : "PTR", "record_type": "PTR",
"zone_name" : "1.254.10.in-addr.arpa", "zone_name": "1.254.10.in-addr.arpa",
"record_data" : "reverse41.domain1.local", "record_data": "reverse41.domain1.local",
"ttl" : 3600, "ttl": 3600,
"key_name" : None } "key_name": None}
reverseform_1 = forms.FormAddReverseRecord(form_data) reverseform_1 = forms.FormAddReverseRecord(form_data)
reverseform_1.is_valid() reverseform_1.is_valid()
self.assertTrue(reverseform_1.is_valid()) self.assertTrue(reverseform_1.is_valid())
def test_MissingData_FormAddRecord(self): def test_MissingData_FormAddRecord(self):
""" Submit FormAddRecord with missing record_data.""" """Submit FormAddRecord with missing record_data."""
form_data = {"dns_server":"server1", form_data = {"dns_server": "server1",
"record_name":"record1", "record_name": "record1",
"record_type": "A", "record_type": "A",
"zone_name":"domain.local", "zone_name": "domain.local",
"record_data":"", "record_data": "",
"ttl": 300, "ttl": 300,
"key_name":None, "key_name": None,
"create_reverse":True} "create_reverse": True}
expected_form_errors = {"record_data": [u"This field is required."]} expected_form_errors = {"record_data": [u"This field is required."]}
testform = forms.FormAddForwardRecord(form_data) testform = forms.FormAddForwardRecord(form_data)
@ -60,15 +59,15 @@ class Form_Tests(TestCase):
self.assertEquals(expected_form_errors, testform.errors) self.assertEquals(expected_form_errors, testform.errors)
def test_InvalidValue_FormAddRecord(self): def test_InvalidValue_FormAddRecord(self):
""" Pass FormAddRecord invalid values, compare error response dicts.""" """Pass FormAddRecord invalid values, compare error response dicts."""
form_data = {"dns_server":"server1", form_data = {"dns_server": "server1",
"record_name":"record1$$$", "record_name": "record1$$$",
"record_type":123, "record_type": 123,
"zone_name":"domain.local", "zone_name": "domain.local",
"record_data":"A.B.C.D", "record_data": "A.B.C.D",
"ttl":"A", "ttl": "A",
"key_name":None, "key_name": None,
"create_reverse":True} "create_reverse": True}
expected_form_errors = {"record_data": [u"Enter a valid IPv4 or IPv6 address."], expected_form_errors = {"record_data": [u"Enter a valid IPv4 or IPv6 address."],
"record_name": [u"Enter a valid value."], "record_name": [u"Enter a valid value."],
@ -81,10 +80,9 @@ class Form_Tests(TestCase):
def test_Validation_FormDeleteRecord(self): def test_Validation_FormDeleteRecord(self):
"""Validate good data in the FormDeleteRecord form.""" """Validate good data in the FormDeleteRecord form."""
delete_dict = { "dns_server" : "foo.net", delete_dict = {"dns_server": "foo.net",
"zone_name" : "domain1.local", "zone_name": "domain1.local",
"rr_list" : '[u"record1.domain1.local", u"record2.domain1.local"]', "rr_list": '[u"record1.domain1.local", u"record2.domain1.local"]'}
}
testform_1 = forms.FormDeleteRecord(delete_dict) testform_1 = forms.FormDeleteRecord(delete_dict)
testform_1.is_valid testform_1.is_valid
self.assertFalse(testform_1.errors) self.assertFalse(testform_1.errors)
@ -95,12 +93,12 @@ class Form_Tests(TestCase):
def test_MissingName_AddCnameForm(self): def test_MissingName_AddCnameForm(self):
"""Attempt to submit a cname add form missing the cname value.""" """Attempt to submit a cname add form missing the cname value."""
form_dict = { "dns_server" : "testserver1", form_dict = {"dns_server": "testserver1",
"zone_name" : "domain1.local", "zone_name": "domain1.local",
"originating_record" : "record1.domain1.local", "originating_record": "record1.domain1.local",
"cname" : "", "cname": "",
"ttl" : 300 } "ttl": 300}
expected_form_errors = { "cname" : [u"This field is required."] } expected_form_errors = {"cname": [u"This field is required."]}
testform_1 = forms.FormAddCnameRecord(form_dict) testform_1 = forms.FormAddCnameRecord(form_dict)
testform_1.is_valid() testform_1.is_valid()
self.assertTrue(testform_1.errors) self.assertTrue(testform_1.errors)

5
binder/tests/testModels.py
View File

@ -3,6 +3,7 @@ from django.db import IntegrityError
from binder import models from binder import models
class Model_BindServer_Tests(TestCase): class Model_BindServer_Tests(TestCase):
def test_BindServerModel(self): def test_BindServerModel(self):
"""Test that adding a well-formed BindServer works.""" """Test that adding a well-formed BindServer works."""
@ -28,7 +29,7 @@ class Model_BindServer_Tests(TestCase):
class Model_Key_Tests(TestCase): class Model_Key_Tests(TestCase):
def test_KeyModel(self): def test_KeyModel(self):
""" Test that adding a well-formed Key works.""" """Test that adding a well-formed Key works."""
self.assertEqual(models.Key.objects.count(), 0) self.assertEqual(models.Key.objects.count(), 0)
key_1 = models.Key(name="testkey1", key_1 = models.Key(name="testkey1",
data="abc123", data="abc123",
@ -38,4 +39,4 @@ class Model_Key_Tests(TestCase):
def test_NonExistantKey(self): def test_NonExistantKey(self):
with self.assertRaisesMessage(models.Key.DoesNotExist, "Key matching query does not exist"): with self.assertRaisesMessage(models.Key.DoesNotExist, "Key matching query does not exist"):
this_key = models.Key.objects.get(name="does_not_exist") models.Key.objects.get(name="does_not_exist")

37
binder/tests/testViews.py
View File

@ -2,11 +2,13 @@ from django.test import TestCase
from django.test.client import Client from django.test.client import Client
from django.core.urlresolvers import reverse from django.core.urlresolvers import reverse
from binder import models, helpers from binder import models
class GetTests(TestCase): class GetTests(TestCase):
""" Unit Tests that exercise HTTP GET. """
"""Unit Tests that exercise HTTP GET."""
def setUp(self): def setUp(self):
self.client = Client() self.client = Client()
@ -19,7 +21,7 @@ class GetTests(TestCase):
self.assertEqual(response.status_code, 200) self.assertEqual(response.status_code, 200)
def test_GetResultRedirects(self): def test_GetResultRedirects(self):
""" GETing a /result/ URL should always redirect to /. """ """GETing a /result/ URL should always redirect to /."""
response = self.client.get(reverse("add_record_result"), follow=True) response = self.client.get(reverse("add_record_result"), follow=True)
self.assertRedirects(response, reverse("index")) self.assertRedirects(response, reverse("index"))
self.assertEqual(response.status_code, 200) self.assertEqual(response.status_code, 200)
@ -31,25 +33,28 @@ class GetTests(TestCase):
self.assertEqual(response.status_code, 200) self.assertEqual(response.status_code, 200)
def test_GetInvalidServer(self): def test_GetInvalidServer(self):
""" Get a zone list for a server not in the database.""" """Get a zone list for a server not in the database."""
server_name = "unconfigured.server.net" server_name = "unconfigured.server.net"
response = self.client.get(reverse("server_zone_list", args=(server_name, ))) response = self.client.get(reverse("server_zone_list",
args=(server_name, )))
self.assertEqual(response.status_code, 404) self.assertEqual(response.status_code, 404)
class PostTests(TestCase): class PostTests(TestCase):
""" Unit Tests that exercise HTTP POST. """
"""Unit Tests that exercise HTTP POST."""
def setUp(self): def setUp(self):
self.client = Client() self.client = Client()
models.BindServer(hostname="testserver.test.net", models.BindServer(hostname="testserver.test.net",
statistics_port=1234).save() statistics_port=1234).save()
def test_DeleteRecordInitial_Empty(self): def test_DeleteRecordInitial_Empty(self):
""" Ensure the initial deletion form works as expected with no RR list. """ """Ensure the initial deletion form works as expected with no RR list."""
response = self.client.post(reverse("delete_record"), { "dns_server" : "testserver.test.net", response = self.client.post(reverse("delete_record"),
"zone_name" : "testzone1.test.net", {"dns_server": "testserver.test.net",
"rr_list" : [] }) "zone_name": "testzone1.test.net",
"rr_list": []})
self.assertContains(response, self.assertContains(response,
'<input type="text" class="form-control hidden" name="zone_name" value="testzone1.test.net"/>', '<input type="text" class="form-control hidden" name="zone_name" value="testzone1.test.net"/>',
@ -63,11 +68,11 @@ class PostTests(TestCase):
def test_DeleteRecordInitial(self): def test_DeleteRecordInitial(self):
""" Ensure the initial deletion form works as expected with RRs mentioned. """ """Ensure the initial deletion form works as expected with RRs mentioned."""
response = self.client.post(reverse("delete_record"), {"dns_server" : "testserver.test.net", response = self.client.post(reverse("delete_record"), {"dns_server": "testserver.test.net",
"zone_name" : "testzone1.test.net", "zone_name": "testzone1.test.net",
"rr_list" : ["testrecord1.testzone1.test.net", "rr_list": ["testrecord1.testzone1.test.net",
"testrecord2.testzone1.test.net"] }) "testrecord2.testzone1.test.net"]})
self.assertContains(response, self.assertContains(response,
'<input type="text" class="form-control hidden" name="zone_name" value="testzone1.test.net"/>', html=True) '<input type="text" class="form-control hidden" name="zone_name" value="testzone1.test.net"/>', html=True)
self.assertContains(response, self.assertContains(response,

2
binder/urls.py
View File

@ -1,6 +1,4 @@
from django.conf.urls import patterns, include, url from django.conf.urls import patterns, include, url
from django.core.urlresolvers import reverse
from django.contrib import admin from django.contrib import admin
admin.autodiscover() admin.autodiscover()

108
binder/views.py
View File

@ -1,4 +1,4 @@
### Binder VIews # Binder VIews
# 3rd Party # 3rd Party
from django.conf import settings from django.conf import settings
@ -7,24 +7,26 @@ from django.shortcuts import get_object_or_404, redirect, render
# App Imports # App Imports
from binder import exceptions, forms, helpers, models from binder import exceptions, forms, helpers, models
def home_index(request): def home_index(request):
""" List the main index page for Binder. """ """List the main index page for Binder."""
return render(request, "index.html") return render(request, "index.html")
def view_server_list(request): def view_server_list(request):
""" List the DNS servers configured in the database. """ """List the DNS servers configured in the database."""
server_list = models.BindServer.objects.all().order_by("hostname") server_list = models.BindServer.objects.all().order_by("hostname")
server_info = [] server_info = []
for current in server_list: for current in server_list:
server_info.append({"host_name" : current, "ip_address" : helpers.ip_info(current.hostname)}) server_info.append({"host_name": current,
"ip_address": helpers.ip_info(current.hostname)})
return render(request, "bcommon/list_servers.html", return render(request, "bcommon/list_servers.html",
{ "server_info" : server_info}) {"server_info": server_info})
def view_server_zones(request, dns_server): def view_server_zones(request, dns_server):
""" Display the list of DNS zones a particular DNS host provides. """ """Display the list of DNS zones a particular DNS host provides."""
errors = "" errors = ""
zone_array = {} zone_array = {}
@ -36,13 +38,13 @@ def view_server_zones(request, dns_server):
errors = "Unable to list server zones. Error: %s" % err errors = "Unable to list server zones. Error: %s" % err
return render(request, "bcommon/list_server_zones.html", return render(request, "bcommon/list_server_zones.html",
{ "errors" : errors, {"errors": errors,
"dns_server" : this_server, "dns_server": this_server,
"zone_array" : zone_array}) "zone_array": zone_array})
def view_zone_records(request, dns_server, zone_name): def view_zone_records(request, dns_server, zone_name):
""" Display the list of records for a particular zone. """ """Display the list of records for a particular zone."""
errors = "" errors = ""
zone_array = {} zone_array = {}
@ -52,32 +54,31 @@ def view_zone_records(request, dns_server, zone_name):
zone_array = this_server.list_zone_records(zone_name) zone_array = this_server.list_zone_records(zone_name)
except exceptions.TransferException, err: except exceptions.TransferException, err:
return render(request, "bcommon/list_zone.html", return render(request, "bcommon/list_zone.html",
{ "errors" : err, {"errors": err,
"zone_name" : zone_name, "zone_name": zone_name,
"dns_server" : this_server}) "dns_server": this_server})
return render(request, "bcommon/list_zone.html", return render(request, "bcommon/list_zone.html",
{ "zone_array" : zone_array, {"zone_array": zone_array,
"dns_server" : this_server, "dns_server": this_server,
"zone_name" : zone_name, "zone_name": zone_name,
"errors" : errors}) "errors": errors})
def view_add_record(request, dns_server, zone_name): def view_add_record(request, dns_server, zone_name):
""" View to provide form to add a DNS record. """ """View to provide form to add a DNS record."""
this_server = get_object_or_404(models.BindServer, hostname=dns_server) this_server = get_object_or_404(models.BindServer, hostname=dns_server)
return render(request, "bcommon/add_record_form.html", return render(request, "bcommon/add_record_form.html",
{ "dns_server" : this_server, {"dns_server": this_server,
"zone_name" : zone_name, "zone_name": zone_name,
"tsig_keys" : models.Key.objects.all(), "tsig_keys": models.Key.objects.all(),
"ttl_choices": settings.TTL_CHOICES, "ttl_choices": settings.TTL_CHOICES,
"record_type_choices": settings.RECORD_TYPE_CHOICES, "record_type_choices": settings.RECORD_TYPE_CHOICES})
})
def view_add_record_result(request): def view_add_record_result(request):
""" Process the input given to add a DNS record. """ """Process the input given to add a DNS record."""
errors = "" errors = ""
if request.method == "GET": if request.method == "GET":
return redirect("/") return redirect("/")
@ -108,36 +109,35 @@ def view_add_record_result(request):
errors = err errors = err
return render(request, "bcommon/response_result.html", return render(request, "bcommon/response_result.html",
{ "errors" : errors, {"errors": errors,
"response" : response }) "response": response})
dns_server = models.BindServer.objects.get(hostname=request.POST["dns_server"]) dns_server = models.BindServer.objects.get(hostname=request.POST["dns_server"])
return render(request, "bcommon/add_record_form.html", return render(request, "bcommon/add_record_form.html",
{ "dns_server" : dns_server, {"dns_server": dns_server,
"zone_name" : request.POST["zone_name"], "zone_name": request.POST["zone_name"],
"tsig_keys" : models.Key.objects.all(), "tsig_keys": models.Key.objects.all(),
"ttl_choices": settings.TTL_CHOICES, "ttl_choices": settings.TTL_CHOICES,
"record_type_choices": settings.RECORD_TYPE_CHOICES, "record_type_choices": settings.RECORD_TYPE_CHOICES,
"form_errors" : form.errors, "form_errors": form.errors,
"form_data" : request.POST }) "form_data": request.POST})
def view_add_cname_record(request, dns_server, zone_name, record_name): def view_add_cname_record(request, dns_server, zone_name, record_name):
""" Process given input to add a CNAME pointer.""" """Process given input to add a CNAME pointer."""
this_server = get_object_or_404(models.BindServer, hostname=dns_server) this_server = get_object_or_404(models.BindServer, hostname=dns_server)
return render(request, "bcommon/add_cname_record_form.html", return render(request, "bcommon/add_cname_record_form.html",
{ "dns_server" : this_server, {"dns_server": this_server,
"originating_record" : "%s.%s" % (record_name, zone_name), "originating_record": "%s.%s" % (record_name, zone_name),
"zone_name" : zone_name, "zone_name": zone_name,
"ttl_choices": settings.TTL_CHOICES, "ttl_choices": settings.TTL_CHOICES,
"tsig_keys" : models.Key.objects.all() }) "tsig_keys": models.Key.objects.all()})
def view_add_cname_result(request): def view_add_cname_result(request):
""" Process input on the CNAME form and provide a response.""" """Process input on the CNAME form and provide a response."""
if request.method == "GET": if request.method == "GET":
return redirect("/") return redirect("/")
@ -158,24 +158,24 @@ def view_add_cname_result(request):
errors = err errors = err
return render(request, "bcommon/response_result.html", return render(request, "bcommon/response_result.html",
{"response" : add_cname_response, {"response": add_cname_response,
"errors" : errors }) "errors": errors})
dns_server = models.BindServer.objects.get(hostname=request.POST["dns_server"]) dns_server = models.BindServer.objects.get(hostname=request.POST["dns_server"])
return render(request, "bcommon/add_cname_record_form.html", return render(request, "bcommon/add_cname_record_form.html",
{ "dns_server" : dns_server, {"dns_server": dns_server,
"zone_name" : request.POST["zone_name"], "zone_name": request.POST["zone_name"],
"record_name" : request.POST["cname"], "record_name": request.POST["cname"],
"originating_record" : request.POST["originating_record"], "originating_record": request.POST["originating_record"],
"form_data" : request.POST, "form_data": request.POST,
"form_errors" : form.errors, "form_errors": form.errors,
"ttl_choices": settings.TTL_CHOICES, "ttl_choices": settings.TTL_CHOICES,
"tsig_keys" : models.Key.objects.all() }) "tsig_keys": models.Key.objects.all()})
def view_delete_record(request): def view_delete_record(request):
""" Provide the initial form for deleting records. """ """Provide the initial form for deleting records."""
if request.method == "GET": if request.method == "GET":
return redirect("/") return redirect("/")
@ -184,14 +184,14 @@ def view_delete_record(request):
rr_list = request.POST.getlist("rr_list") rr_list = request.POST.getlist("rr_list")
return render(request, "bcommon/delete_record_initial.html", return render(request, "bcommon/delete_record_initial.html",
{ "dns_server" : dns_server, {"dns_server": dns_server,
"zone_name" : zone_name, "zone_name": zone_name,
"rr_list" : rr_list, "rr_list": rr_list,
"tsig_keys" : models.Key.objects.all() }) "tsig_keys": models.Key.objects.all()})
def view_delete_result(request): def view_delete_result(request):
""" View that deletes records and returns the response. """ """View that deletes records and returns the response."""
if request.method == "GET": if request.method == "GET":
return redirect("/") return redirect("/")
@ -209,4 +209,4 @@ def view_delete_result(request):
clean_form["key_name"]) clean_form["key_name"])
return render(request, "bcommon/response_result.html", return render(request, "bcommon/response_result.html",
{ "response" : delete_result }) {"response": delete_result})

1
wsgi.py
View File

@ -8,7 +8,6 @@ https://docs.djangoproject.com/en/1.7/howto/deployment/wsgi/
""" """
import os import os
import sys
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "binder.settings") os.environ.setdefault("DJANGO_SETTINGS_MODULE", "binder.settings")
from django.core.wsgi import get_wsgi_application from django.core.wsgi import get_wsgi_application