Closes #122: Implement a mechanism to enforce policy before executing branch actions

docs/configuration.md

@@ -51,3 +51,35 @@ Default: `branch_`
 The string to prefix to the unique branch ID when provisioning the PostgreSQL schema for a branch. Per [the PostgreSQL documentation](https://www.postgresql.org/docs/16/sql-syntax-lexical.html#SQL-SYNTAX-IDENTIFIERS), this string must begin with a letter or underscore.
 
 Note that a valid prefix is required, as the randomly-generated branch ID alone may begin with a digit, which would not qualify as a valid schema name.
+
+---
+
+## `sync_validators`
+
+Default: `[]` (empty list)
+
+A list of import paths to functions which validate whether a branch is permitted to be synced.
+
+---
+
+## `merge_validators`
+
+Default: `[]` (empty list)
+
+A list of import paths to functions which validate whether a branch is permitted to be merged.
+
+---
+
+## `revert_validators`
+
+Default: `[]` (empty list)
+
+A list of import paths to functions which validate whether a branch is permitted to be reverted.
+
+---
+
+## `archive_validators`
+
+Default: `[]` (empty list)
+
+A list of import paths to functions which validate whether a branch is permitted to be archived.

netbox_branching/__init__.py

@@ -1,9 +1,12 @@
 from django.conf import settings
 from django.core.exceptions import ImproperlyConfigured
+from django.utils.module_loading import import_string
 
 from netbox.plugins import PluginConfig, get_plugin_config
 from netbox.registry import registry
 
+from .constants import BRANCH_ACTIONS
+
 
 class AppConfig(PluginConfig):
     name = 'netbox_branching'
@@ -27,6 +30,12 @@ class AppConfig(PluginConfig):
 
         # This string is prefixed to the name of each new branch schema during provisioning
         'schema_prefix': 'branch_',
+
+        # Branch action validators
+        'sync_validators': [],
+        'merge_validators': [],
+        'revert_validators': [],
+        'archive_validators': [],
     }
 
     def ready(self):
@@ -44,6 +53,14 @@ def ready(self):
                 "netbox_branching: DATABASE_ROUTERS must contain 'netbox_branching.database.BranchAwareRouter'."
             )
 
+        # Validate branch action validators
+        for action in BRANCH_ACTIONS:
+            for validator_path in get_plugin_config('netbox_branching', f'{action}_validators'):
+                try:
+                    import_string(validator_path)
+                except ImportError:
+                    raise ImproperlyConfigured(f"Branch {action} validator not found: {validator_path}")
+
         # Record all object types which support branching in the NetBox registry
         exempt_models = (
             *constants.EXEMPT_MODELS,

netbox_branching/constants.py

@@ -7,6 +7,14 @@
 # HTTP header for API requests
 BRANCH_HEADER = 'X-NetBox-Branch'
 
+# Branch actions
+BRANCH_ACTIONS = (
+    'sync',
+    'merge',
+    'revert',
+    'archive',
+)
+
 # URL query parameter name
 QUERY_PARAM = '_branch'
 

netbox_branching/forms/misc.py

@@ -20,10 +20,13 @@ class BranchActionForm(forms.Form):
         help_text=_('Leave unchecked to perform a dry run')
     )
 
-    def __init__(self, branch, *args, **kwargs):
+    def __init__(self, branch, *args, allow_commit=True, **kwargs):
         self.branch = branch
         super().__init__(*args, **kwargs)
 
+        if not allow_commit:
+            self.fields['commit'].disabled = True
+
     def clean(self):
         super().clean()
 

netbox_branching/models/branches.py

@@ -12,6 +12,7 @@
 from django.test import RequestFactory
 from django.urls import reverse
 from django.utils import timezone
+from django.utils.module_loading import import_string
 from django.utils.translation import gettext_lazy as _
 
 from core.models import ObjectChange as ObjectChange_
@@ -21,6 +22,7 @@
 from netbox.models.features import JobsMixin
 from netbox.plugins import get_plugin_config
 from netbox_branching.choices import BranchEventTypeChoices, BranchStatusChoices
+from netbox_branching.constants import BRANCH_ACTIONS
 from netbox_branching.contextvars import active_branch
 from netbox_branching.signals import *
 from netbox_branching.utilities import (
@@ -241,6 +243,54 @@ def get_event_history(self):
             last_time = event.time
         return history
 
+    #
+    # Branch action indicators
+    #
+
+    def _can_do_action(self, action):
+        """
+        Execute any validators configured for the specified branch
+        action. Return False if any fail; otherwise return True.
+        """
+        if action not in BRANCH_ACTIONS:
+            raise Exception(f"Unrecognized branch action: {action}")
+        for validator_path in get_plugin_config('netbox_branching', f'{action}_validators'):
+            if not import_string(validator_path)(self):
+                return False
+        return True
+
+    @property
+    def can_sync(self):
+        """
+        Indicates whether the branch can be synced.
+        """
+        return self._can_do_action('sync')
+
+    @property
+    def can_merge(self):
+        """
+        Indicates whether the branch can be merged.
+        """
+        return self._can_do_action('merge')
+
+    @property
+    def can_revert(self):
+        """
+        Indicates whether the branch can be reverted.
+        """
+        return self._can_do_action('revert')
+
+    @property
+    def can_archive(self):
+        """
+        Indicates whether the branch can be archived.
+        """
+        return self._can_do_action('archive')
+
+    #
+    # Branch actions
+    #
+
     def sync(self, user, commit=True):
         """
         Apply changes from the main schema onto the Branch's schema.
@@ -250,6 +300,8 @@ def sync(self, user, commit=True):
 
         if not self.ready:
             raise Exception(f"Branch {self} is not ready to sync")
+        if commit and not self.can_sync:
+            raise Exception(f"Syncing this branch is not permitted.")
 
         # Retrieve unsynced changes before we update the Branch's status
         if changes := self.get_unsynced_changes().order_by('time'):
@@ -305,6 +357,8 @@ def merge(self, user, commit=True):
 
         if not self.ready:
             raise Exception(f"Branch {self} is not ready to merge")
+        if commit and not self.can_merge:
+            raise Exception(f"Merging this branch is not permitted.")
 
         # Retrieve staged changes before we update the Branch's status
         if changes := self.get_unmerged_changes().order_by('time'):
@@ -374,6 +428,8 @@ def revert(self, user, commit=True):
 
         if not self.merged:
             raise Exception(f"Only merged branches can be reverted.")
+        if commit and not self.can_revert:
+            raise Exception(f"Reverting this branch is not permitted.")
 
         # Retrieve applied changes before we update the Branch's status
         if changes := self.get_changes().order_by('-time'):
@@ -530,6 +586,10 @@ def archive(self, user):
         """
         Deprovision the Branch and set its status to "archived."
         """
+        if not self.can_archive:
+            raise Exception(f"Archiving this branch is not permitted.")
+
+        # Deprovision the branch's schema
         self.deprovision()
 
         # Update the branch's status to "archived"

netbox_branching/templates/netbox_branching/branch.html

@@ -51,7 +51,7 @@
         <i class="mdi mdi-arrow-u-left-top"></i> {% trans "Revert" %}
       </button>
     {% endif %}
-    {% if perms.netbox_branching.archive_branch %}
+    {% if perms.netbox_branching.archive_branch and object.can_archive %}
       <a href="{% url 'plugins:netbox_branching:branch_archive' pk=object.pk %}" class="btn btn-primary">
         <i class="mdi mdi-archive-outline"></i> {% trans "Archive" %}
       </a>

netbox_branching/templates/netbox_branching/branch_action.html

@@ -17,6 +17,14 @@
 {% block content %}
   {# Form tab #}
   <div class="tab-pane show active" id="action-form" role="tabpanel" aria-labelledby="action-form-tab">
+    {% if not action_permitted %}
+      <div class="alert alert-warning">
+        <i class="mdi mdi-alert-circle"></i>
+        {% blocktrans %}
+          This action is disallowed per policy, however dry runs are permitted.
+        {% endblocktrans %}
+      </div>
+    {% endif %}
     {% if conflicts_table.rows %}
       <div class="alert alert-danger">
         <i class="mdi mdi-alert-circle"></i>

netbox_branching/views.py

@@ -186,18 +186,21 @@ def do_action(self, branch, request, form):
 
     def get(self, request, **kwargs):
         branch = self.get_object(**kwargs)
-        form = self.form(branch)
+        action_permitted = getattr(branch, f'can_{self.action}')
+        form = self.form(branch, allow_commit=action_permitted)
 
         return render(request, self.template_name, {
             'branch': branch,
             'action': _(f'{self.action.title()} Branch'),
             'form': form,
+            'action_permitted': action_permitted,
             'conflicts_table': self._get_conflicts_table(branch),
         })
 
     def post(self, request, **kwargs):
         branch = self.get_object(**kwargs)
-        form = self.form(branch, request.POST)
+        action_permitted = getattr(branch, f'can_{self.action}')
+        form = self.form(branch, request.POST, allow_commit=action_permitted)
 
         if branch.status not in self.valid_states:
             messages.error(request, _(
@@ -210,6 +213,7 @@ def post(self, request, **kwargs):
             'branch': branch,
             'action': _(f'{self.action.title()} Branch'),
             'form': form,
+            'action_permitted': action_permitted,
             'conflicts_table': self._get_conflicts_table(branch),
         })
 
@@ -277,14 +281,17 @@ def get_required_permission(self):
         return f'netbox_branching.archive_branch'
 
     @staticmethod
-    def _enforce_status(request, branch):
+    def _validate(request, branch):
         if branch.status != BranchStatusChoices.MERGED:
             messages.error(request, _("Only merged branches can be archived."))
             return redirect(branch.get_absolute_url())
+        if not branch.can_revert:
+            messages.error(request, _("Reverting this branch is disallowed per policy."))
+            return redirect(branch.get_absolute_url())
 
     def get(self, request, **kwargs):
         branch = self.get_object(**kwargs)
-        self._enforce_status(request, branch)
+        self._validate(request, branch)
         form = forms.ConfirmationForm()
 
         return render(request, self.template_name, {
@@ -294,7 +301,7 @@ def get(self, request, **kwargs):
 
     def post(self, request, **kwargs):
         branch = self.get_object(**kwargs)
-        self._enforce_status(request, branch)
+        self._validate(request, branch)
         form = forms.ConfirmationForm(request.POST)
 
         if form.is_valid():