Commits

George Notaras  committed e92a776

Added experimental support for zone cloning.

  • Participants
  • Parent commits 74dcfd3

Comments (0)

Files changed (8)

 - The application can be configured to support a user-defined subset of the
   resource records supported by PowerDNS and customize the order in which they
   appear in the administration panel.
+- Zone cloning (experimental).
 - Zone file import through web form.
 - Zone file export.
 - Command-line interfaces to import and export zones in bulk.

File docs/introduction.rst

 - The application can be configured to support a user-defined subset of the
   resource records supported by PowerDNS and customize the order in which they
   appear in the administration panel.
+- Zone cloning (experimental).
 - Zone file import through web form.
 - Zone file export.
 - Command-line interfaces to import and export zones in bulk.

File docs/usage.rst

 
 TODO
 
+
+Zone Cloning
+============
+
+TODO
+

File src/powerdns_manager/actions.py

 
 from powerdns_manager.forms import ZoneTypeSelectionForm
 from powerdns_manager.forms import TtlSelectionForm
+from powerdns_manager.forms import ClonedZoneDomainForm
+from powerdns_manager.utils import generate_serial
 from powerdns_manager.utils import generate_api_key
+from powerdns_manager.utils import interchange_domain
 
 
 
 force_serial_update.short_description = "Force serial update"
 
 
+
+def clone_zone(modeladmin, request, queryset):
+    """Actions that clones the selected zone.
+    
+    Accepts only one selected zone.
+    
+    This action first displays a page which provides an input box to enter
+    the origin of the new zone.
+    
+    It checks if the user has add & change permissions.
+    
+    It checks if a zone with the name that has been entered as new exists in
+    the database.
+    
+    Based on: https://github.com/django/django/blob/1.4.2/django/contrib/admin/actions.py
+    
+    Important
+    ---------
+    In order to work requires some special form fields (see the template).
+    
+    """
+    opts = modeladmin.model._meta
+    app_label = opts.app_label
+    
+    Domain = cache.get_model('powerdns_manager', 'Domain')
+    Record = cache.get_model('powerdns_manager', 'Record')
+
+    # Check the number of selected zones. This action can work on a single zone.
+    
+    n = queryset.count()
+    if n != 1:
+        messages.error(request, 'Only one zone may be selected for cloning.')
+        return None
+    
+    # Check permissions
+        
+    perm_domain_add = '%s.%s' % (opts.app_label, opts.get_add_permission())
+    perm_domain_change = '%s.%s' % (opts.app_label, opts.get_change_permission())
+    perm_record_add = '%s.add_record' % opts.app_label
+    perm_record_change = '%s.change_record' % opts.app_label
+    
+    if not request.user.has_perms(
+            [perm_domain_add, perm_domain_change, perm_record_add, perm_record_change]):
+        raise PermissionDenied
+    
+    # Check that the user has change permission for the add and change modeladmin forms
+    if not modeladmin.has_add_permission(request):
+        raise PermissionDenied
+    if not modeladmin.has_change_permission(request):
+        raise PermissionDenied
+    
+    # The user has set a domain name for the clone through the forms.ClonedZoneDomainForm form.
+    #if request.method == 'POST':
+    if request.POST.get('post'):
+        form = ClonedZoneDomainForm(request.POST)
+        if form.is_valid():
+            
+            clone_domain_name = form.cleaned_data['clone_domain_name']
+            
+            if not clone_domain_name:
+                return None # Should never happen
+            
+            # At this point queryset contain exactly one object. Checked above.
+            domain_obj = queryset[0]
+            
+            # Find all resource records of this domain
+            domain_rr_qs = Record.objects.filter(domain=domain_obj)
+            
+            # Create the clone (Check for uniqueness takes place in forms.ClonedZoneDomainForm 
+            clone_obj = Domain.objects.create(
+                name = clone_domain_name,
+                master = domain_obj.master,
+                #last_check = domain_obj.last_check,
+                type = domain_obj.type,
+                #notified_serial = domain_obj.notified_serial,
+                account = domain_obj.account,
+                created_by = request.user   # We deliberately do not use the domain_obj.created_by
+            )
+            modeladmin.log_addition(request, clone_obj)
+            
+            # Create the clone's RRs
+            for rr in domain_rr_qs:
+                
+                # Construct RR name with interchanged domain
+                clone_rr_name = interchange_domain(rr.name, domain_obj.name, clone_domain_name)
+                
+                # Special treatment to the content of SOA and SRV
+                if rr.type == 'SOA':
+                    content_parts = rr.content.split()
+                    # primary
+                    content_parts[0] = interchange_domain(content_parts[0], domain_obj.name, clone_domain_name)
+                    # hostmaster
+                    content_parts[1] = interchange_domain(content_parts[1], domain_obj.name, clone_domain_name)
+                    # Serial. Set new serial
+                    content_parts[2] = generate_serial()
+                    clone_rr_content = ' '.join(content_parts)
+                elif rr.type == 'SRV':
+                    content_parts = rr.content.split()
+                    # target
+                    content_parts[2] = interchange_domain(content_parts[2], domain_obj.name, clone_domain_name)
+                else:
+                    clone_rr_content = interchange_domain(rr.content, domain_obj.name, clone_domain_name)
+                
+                clone_rr = Record(
+                    domain = clone_obj,
+                    name = clone_rr_name,
+                    type = rr.type,
+                    content = clone_rr_content,
+                    ttl = rr.ttl,
+                    prio = rr.prio,
+                    auth = rr.auth,
+                    ordername = rr.ordername
+                )
+                clone_rr.save()
+                modeladmin.log_addition(request, clone_rr)
+            
+            # Update the domain serial
+            #domain_obj.update_serial()
+            
+            messages.info(request, 'Successfully cloned %s zone to %s' % \
+                (domain_obj.name, clone_domain_name))
+            # Return None to display the change list page again.
+            return None
+        
+    else:
+        form = ClonedZoneDomainForm()
+    
+    info_dict = {
+        'form': form,
+        'queryset': queryset,
+        'opts': opts,
+        'app_label': app_label,
+        'action_checkbox_name': helpers.ACTION_CHECKBOX_NAME,
+    }
+    return render_to_response(
+        'powerdns_manager/actions/clone_zone.html', info_dict, context_instance=RequestContext(request), mimetype='text/html')
+clone_zone.short_description = "Clone the selected zone"
+
+

File src/powerdns_manager/admin.py

 from powerdns_manager.actions import set_ttl_bulk
 from powerdns_manager.actions import force_serial_update
 from powerdns_manager.actions import reset_api_key
+from powerdns_manager.actions import clone_zone
 from powerdns_manager.utils import generate_api_key
 
 
     verbose_name = 'zone'
     verbose_name_plural = 'zones'
     save_on_top = True
-    actions = [reset_api_key, set_domain_type_bulk, set_ttl_bulk, force_serial_update]
+    actions = [reset_api_key, set_domain_type_bulk, set_ttl_bulk, force_serial_update, clone_zone]
     change_list_template = 'powerdns_manager/domain_changelist.html'
     
     #

File src/powerdns_manager/forms.py

 class SrvRecordModelForm(BaseRecordModelForm):
     """ModelForm for SRV resource records.
     
+    Content for SRV record:
+    
+        weight port target
+    
     For details see docstrings in SoaRecordModelForm.
     
     """
     reset_zone_minimum = forms.BooleanField(required=False, label=_('Reset minimum TTL of the zones?'), help_text="""If checked, the minimum TTL of the selected zones will be reset to the new TTL value.""")
 
 
+
+class ClonedZoneDomainForm(forms.Form):
+    """This form is used in intermediate page that sets the name of the cloned zone."""
+    clone_domain_name = forms.CharField(max_length=255, required=True, label=_('Domain Name'), help_text="""Enter the domain name of the clone.""")
+    
+    def clean_clone_domain_name(self):
+        clone_domain_name = self.cleaned_data.get('clone_domain_name')
+        
+        # 1) Check for valid characters
+        validate_hostname(clone_domain_name, supports_cidr_notation=True)
+        
+        # 2) Check for uniqueness
+        Domain = cache.get_model('powerdns_manager', 'Domain')
+        try:
+            domain_obj = Domain.objects.get(name=clone_domain_name)
+        except Domain.DoesNotExist:
+            pass
+        else:
+            raise forms.ValidationError('A zone with this name already exists. Please enter a new domain name.')
+        
+        return clone_domain_name
+

File src/powerdns_manager/templates/powerdns_manager/actions/clone_zone.html

+{% extends "admin/base_site.html" %}
+{% load i18n l10n static %}
+{% load url from future %}
+{% load admin_urls %}
+
+{% block extrastyle %}{{ block.super }}<link rel="stylesheet" type="text/css" href="{% static "admin/css/forms.css" %}" />{% endblock %}
+
+{% block breadcrumbs %}
+	<div class="breadcrumbs">
+		<a href="{% url 'admin:index' %}">{% trans 'Home' %}</a>
+		&rsaquo; <a href="{% url 'admin:app_list' app_label=app_label %}">{{ app_label|capfirst|escape }}</a>
+		&rsaquo; <a href="{% url opts|admin_urlname:'changelist' %}">{{ opts.verbose_name_plural|capfirst }}</a>
+		&rsaquo; {% trans 'Set the domain name of the cloned zone' %}
+	</div>
+{% endblock %}
+
+{% block title %}{% trans 'Set the domain name of the cloned zone' %}{% endblock %}
+
+{% block content %}
+    <div id="content-main">
+        
+        <form action="" method="post">{% csrf_token %}
+        <div>
+            {% if form.errors %}
+                <p class="errornote">
+                {% blocktrans count counter=form.errors.items|length %}Please correct the error below.{% plural %}Please correct the errors below.{% endblocktrans %}
+                </p>
+            {% endif %}
+
+            <h1>{% trans 'Set the domain name of the cloned zone' %}</h1>
+            <p>{% trans "Enter a domain name for the cloned zone." %}</p>
+            
+            <fieldset class="module aligned">
+    
+                <div class="form-row">
+                    {{ form.clone_domain_name.errors }}
+                    <label for="id_clone_domain_name" class="">{% trans 'Domain Name' %}:</label>{{ form.clone_domain_name }}
+                </div>
+                
+            </fieldset>
+
+            {# Special Fields #}
+            {# These are needed for the action code to work. This an undocumented Django feature #}
+            {% for obj in queryset %}
+                <input type="hidden" name="{{ action_checkbox_name }}" value="{{ obj.pk|unlocalize }}" />
+            {% endfor %}
+            <input type="hidden" name="action" value="clone_zone" />
+            <input type="hidden" name="post" value="yes" />
+            
+            <div class="submit-row">
+                <input type="submit" value="{% trans 'Save' %}" class="default" />
+            </div>
+
+            <script type="text/javascript">document.getElementById("id_clone_domain_name").focus();</script>
+        </div>
+        </form>
+
+    </div> <!-- content-main -->
+{% endblock %}

File src/powerdns_manager/utils.py

         raise ValidationError('The hostname contains illegal characters')
 
 
+def interchange_domain(data, domain1, domain2):
+    """Replaces domain1 with domain2 in data.
+    
+    data: RR's name or content
+    
+    Replacement occurs only in the base domain.
+    
+    TODO: improve this description.
+    
+    """
+    if len(domain1) > len(data):
+        return data
+    elif not data.endswith(domain1):
+        return data
+    elif data == domain1:
+        return domain2
+    
+    data_parts = data.split('.')
+    domain1_parts = domain1.split('.')
+    
+    new_data_parts = data_parts[:-len(domain1_parts)]
+    return '%s.%s' % ('.'.join(new_data_parts), domain2) 
+
+
 def generate_serial(serial_old=None):
     """Return a serial number for the zone in the form YYYYMMDDNN.