Dynamic Remediation Functions and Mapping to override default hier_config

nautobot_golden_config/api/serializers.py

@@ -125,3 +125,23 @@ class Meta:
         model = models.ConfigPlan
         fields = "__all__"
         read_only_fields = ["device", "plan_type", "feature", "config_set"]
+
+
+class DynamicRemediationFunctionSerializer(NautobotModelSerializer, TaggedModelSerializerMixin):
+    """Serializer for DynamicRemediationFunction object."""
+
+    class Meta:
+        """Set Meta Data for DynamicRemediationFunction, will serialize all fields."""
+
+        model = models.DynamicRemediationFunction
+        fields = "__all__"
+
+
+class DynamicRemediationMappingSerializer(NautobotModelSerializer, TaggedModelSerializerMixin):
+    """Serializer for DynamicRemediationMapping object."""
+
+    class Meta:
+        """Set Meta Data for DynamicRemediationMapping, will serialize all fields."""
+
+        model = models.DynamicRemediationMapping
+        fields = "__all__"

nautobot_golden_config/api/urls.py

@@ -17,6 +17,8 @@
 router.register("remediation-setting", views.RemediationSettingViewSet)
 router.register("config-postprocessing", views.ConfigToPushViewSet)
 router.register("config-plan", views.ConfigPlanViewSet)
+router.register("dynamic-remediation-function", views.DynamicRemediationFunctionViewSet)
+router.register("dynamic-remediation-mapping", views.DynamicRemediationMappingViewSet)
 urlpatterns = router.urls
 urlpatterns.append(
     path(

nautobot_golden_config/api/views.py

@@ -164,3 +164,19 @@ def get_serializer_context(self):
             }
         )
         return context
+
+
+class DynamicRemediationFunctionViewSet(NautobotModelViewSet):  # pylint:disable=too-many-ancestors
+    """API viewset for interacting with ConfigCompliance objects."""
+
+    queryset = models.DynamicRemediationFunction.objects.all()
+    serializer_class = serializers.DynamicRemediationFunctionSerializer
+    filterset_class = filters.DynamicRemediationFunctionFilterSet
+
+
+class DynamicRemediationMappingViewSet(NautobotModelViewSet):  # pylint:disable=too-many-ancestors
+    """API viewset for interacting with ConfigCompliance objects."""
+
+    queryset = models.DynamicRemediationMapping.objects.all()
+    serializer_class = serializers.DynamicRemediationMappingSerializer
+    filterset_class = filters.DynamicRemediationMappingFilterSet

nautobot_golden_config/choices.py

@@ -42,3 +42,19 @@ class ConfigPlanTypeChoice(ChoiceSet):
         (TYPE_REMEDIATION, "Remediation"),
         (TYPE_MANUAL, "Manual"),
     )
+
+
+class DynamicRemediationExpressionChoice(ChoiceSet):
+    """Choiceset used for Dynamic Remediation Expression Choices."""
+
+    STARTS_WITH = "startswith"
+    ENDS_WITH = "endswith"
+    EQUALS = "equals"
+    CONTAINS = "contains"
+
+    CHOICES = (
+        (STARTS_WITH, "startswith"),
+        (ENDS_WITH, "endswith"),
+        (EQUALS, "equals"),
+        (CONTAINS, "contains"),
+    )

nautobot_golden_config/datasources.py

@@ -8,7 +8,13 @@
 from nautobot.extras.registry import DatasourceContent
 
 from nautobot_golden_config.exceptions import MissingReference
-from nautobot_golden_config.models import ComplianceFeature, ComplianceRule, ConfigRemove, ConfigReplace
+from nautobot_golden_config.models import (
+    ComplianceFeature,
+    ComplianceRule,
+    ConfigRemove,
+    ConfigReplace,
+    DynamicRemediationFunction,
+)
 from nautobot_golden_config.utilities.constant import ENABLE_BACKUP, ENABLE_COMPLIANCE, ENABLE_INTENDED
 
 
@@ -36,6 +42,51 @@ def refresh_git_backup(repository_record, job_result, delete=False):  # pylint:
     )
 
 
+def refresh_git_gc_dynamic_remediations(repository_record, job_result, delete=False):  # pylint: disable=unused-argument
+    """Callback for gitrepository updates on Hier Config Dynamic Remediation repo."""
+    job_result.log(
+        "Successfully Pulled git repo test",
+        level_choice=LogLevelChoices.LOG_DEBUG,
+    )
+
+    dynamic_remediation_path = os.path.join(repository_record.filesystem_path, "hier_config_dynamic_remediations")
+    if not os.path.isdir(dynamic_remediation_path):
+        job_result.log(
+            f"Skipping sync for {dynamic_remediation_path} because directory doesn't exist.",
+            level_choice=LogLevelChoices.LOG_INFO,
+        )
+        return
+
+    file_info = []
+    for root, _, files in os.walk(dynamic_remediation_path):
+        for file_name in files:
+            if not file_name.endswith(".py") or "__init__" in file_name:
+                continue
+            file_info.append({"root": root, "file_name": file_name})
+
+    for details in file_info:
+        dynamic_remediation_function, created = DynamicRemediationFunction.objects.get_or_create(
+            dynamic_remediation_repository=repository_record,
+            file_name=details["file_name"],
+        )
+        job_result.log(
+            f"{'Created' if created else 'Updated'} record {details['file_name']} -- {dynamic_remediation_function}",
+            level_choice=LogLevelChoices.LOG_INFO,
+        )
+
+    active_file_names = [details["file_name"] for details in file_info]
+    for function in DynamicRemediationFunction.objects.filter(dynamic_remediation_repository=repository_record):
+        if function.file_name not in active_file_names:
+            try:
+                job_result.log(f"Deleting {function}...")
+                function.delete()
+            except IntegrityError:
+                job_result.log(
+                    f"File {function.file_name} was not found in this repo {active_file_names} while there are still Remediation Mappings associated to this file.  This may cause an issue while generating remediation.",
+                    level_choice=LogLevelChoices.LOG_WARNING,
+                )
+
+
 def refresh_git_gc_properties(repository_record, job_result, delete=False):  # pylint: disable=unused-argument
     """Callback for gitrepository updates on Git Configuration repo.
 
@@ -244,6 +295,18 @@ def update_git_gc_properties(golden_config_path, job_result, gc_config_item):  #
             ),
         )
     )
+if ENABLE_COMPLIANCE:
+    datasource_contents.append(
+        (
+            "extras.gitrepository",
+            DatasourceContent(
+                name="Hier Config Dynamic Remediations",
+                content_identifier="nautobot_golden_config.hierconfigdynamicremediations",
+                icon="mdi-file-code",
+                callback=refresh_git_gc_dynamic_remediations,
+            ),
+        )
+    )
 
 datasource_contents.append(
     (

nautobot_golden_config/filters.py

@@ -441,3 +441,32 @@ class Meta:
 
         model = models.ConfigPlan
         fields = ["id", "created", "change_control_id", "plan_type", "tags"]
+
+
+class DynamicRemediationFunctionFilterSet(NautobotFilterSet):
+    """Inherits Base Class NautobotFilterSet."""
+
+    q = SearchFilter(
+        filter_predicates={
+            "file_name": {
+                "lookup_expr": "icontains",
+                "preprocessor": str,
+            },
+        },
+    )
+
+    class Meta:
+        """Boilerplate filter Meta data for Dynamic Remediation Function."""
+
+        model = models.DynamicRemediationFunction
+        fields = ["id", "file_name", "dynamic_remediation_repository"]
+
+
+class DynamicRemediationMappingFilterSet(NautobotFilterSet):
+    """Inherits Base Class NautobotFilterSet."""
+
+    class Meta:
+        """Boilerplate filter Meta data for Dynamic Remediation Mapping."""
+
+        model = models.DynamicRemediationMapping
+        fields = ["id", "expression_choice", "config_string", "remediation_function", "platform", "enabled"]

nautobot_golden_config/forms.py

@@ -12,7 +12,12 @@
 from nautobot.tenancy.models import Tenant, TenantGroup
 
 from nautobot_golden_config import models
-from nautobot_golden_config.choices import ComplianceRuleConfigTypeChoice, ConfigPlanTypeChoice, RemediationTypeChoice
+from nautobot_golden_config.choices import (
+    ComplianceRuleConfigTypeChoice,
+    ConfigPlanTypeChoice,
+    RemediationTypeChoice,
+    DynamicRemediationExpressionChoice,
+)
 
 # ConfigCompliance
 
@@ -644,3 +649,54 @@ class Meta:
             "change_control_url",
             "tags",
         ]
+
+
+class DynamicRemediationMappingForm(NautobotModelForm):
+    """Form for ConfigPlan instances."""
+
+    remediation_function = forms.DynamicModelChoiceField(
+        queryset=models.DynamicRemediationFunction.objects.all(),
+    )
+    platform = forms.DynamicModelChoiceField(queryset=Platform.objects.all())
+
+    class Meta:
+        """Boilerplate form Meta data for ConfigPlan."""
+
+        model = models.DynamicRemediationMapping
+        fields = ("expression_choice", "config_string", "remediation_function", "platform", "enabled")
+
+
+class DynamicRemediationMappingBulkEditForm(NautobotBulkEditForm):
+    """BulkEdit form for DynamicRemediationMapping instances."""
+
+    pk = django_forms.ModelMultipleChoiceField(
+        queryset=models.DynamicRemediationMapping.objects.all(), widget=django_forms.MultipleHiddenInput
+    )
+
+    expression_choice = django_forms.ChoiceField(
+        choices=forms.add_blank_choice(DynamicRemediationExpressionChoice),
+        required=False,
+        widget=django_forms.Select(),
+        label="Plan Type",
+    )
+
+    config_string = django_forms.CharField(required=False)
+
+    remediation_function = forms.DynamicModelChoiceField(
+        queryset=models.DynamicRemediationFunction.objects.all(),
+        required=False,
+    )
+    platform = forms.DynamicModelChoiceField(
+        queryset=Platform.objects.all(),
+        required=False,
+    )
+    enabled = django_forms.BooleanField(required=False)
+
+    field_order = ("expression_choice", "config_string", "remediation_function", "platform", "enabled", "note")
+
+    class Meta:
+        """Boilerplate form Meta data for DynamicRemediationMapping."""
+
+        model = models.DynamicRemediationMapping
+
+        nullable_fields = []

nautobot_golden_config/migrations/0031_dynamicremediation.py

@@ -0,0 +1,120 @@
+# Generated by Django 3.2.24 on 2024-06-10 14:23
+
+import django.core.serializers.json
+from django.db import migrations, models
+import django.db.models.deletion
+import nautobot.core.models.fields
+import nautobot.extras.models.mixins
+import uuid
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("extras", "0099_remove_dangling_note_objects"),
+        ("nautobot_golden_config", "0030_alter_goldenconfig_device"),
+        ("dcim", "0052_fix_interface_redundancy_group_created"),
+        ("extras", "0103_add_db_indexes_to_object_change"),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name="DynamicRemediationFunction",
+            fields=[
+                (
+                    "id",
+                    models.UUIDField(
+                        default=uuid.uuid4, editable=False, primary_key=True, serialize=False, unique=True
+                    ),
+                ),
+                ("created", models.DateTimeField(auto_now_add=True, null=True)),
+                ("last_updated", models.DateTimeField(auto_now=True, null=True)),
+                (
+                    "_custom_field_data",
+                    models.JSONField(blank=True, default=dict, encoder=django.core.serializers.json.DjangoJSONEncoder),
+                ),
+                ("file_name", models.CharField(max_length=200)),
+                (
+                    "dynamic_remediation_repository",
+                    models.ForeignKey(
+                        limit_choices_to={
+                            "provided_contents__contains": "nautobot_golden_config.hierconfigdynamicremedation"
+                        },
+                        on_delete=django.db.models.deletion.PROTECT,
+                        related_name="dynamic_remediation_repository",
+                        to="extras.gitrepository",
+                    ),
+                ),
+                (
+                    "tags",
+                    nautobot.core.models.fields.TagsField(
+                        help_text="A comma-separated list of tags.",
+                        through="extras.TaggedItem",
+                        to="extras.Tag",
+                        verbose_name="Tags",
+                    ),
+                ),
+            ],
+            options={
+                "abstract": False,
+                "unique_together": {("file_name", "dynamic_remediation_repository")},
+            },
+            bases=(
+                models.Model,
+                nautobot.extras.models.mixins.DynamicGroupMixin,
+                nautobot.extras.models.mixins.NotesMixin,
+            ),
+        ),
+        migrations.CreateModel(
+            name="DynamicRemediationMapping",
+            fields=[
+                (
+                    "id",
+                    models.UUIDField(
+                        default=uuid.uuid4, editable=False, primary_key=True, serialize=False, unique=True
+                    ),
+                ),
+                ("created", models.DateTimeField(auto_now_add=True, null=True)),
+                ("last_updated", models.DateTimeField(auto_now=True, null=True)),
+                (
+                    "_custom_field_data",
+                    models.JSONField(blank=True, default=dict, encoder=django.core.serializers.json.DjangoJSONEncoder),
+                ),
+                ("expression_choice", models.CharField(max_length=50)),
+                ("config_string", models.CharField(max_length=256)),
+                ("enabled", models.BooleanField(default=True)),
+                (
+                    "platform",
+                    models.ForeignKey(
+                        on_delete=django.db.models.deletion.CASCADE,
+                        related_name="dynamic_remediation_mapping",
+                        to="dcim.platform",
+                    ),
+                ),
+                (
+                    "remediation_function",
+                    models.ForeignKey(
+                        on_delete=django.db.models.deletion.PROTECT,
+                        to="nautobot_golden_config.dynamicremediationfunction",
+                    ),
+                ),
+                (
+                    "tags",
+                    nautobot.core.models.fields.TagsField(
+                        help_text="A comma-separated list of tags.",
+                        through="extras.TaggedItem",
+                        to="extras.Tag",
+                        verbose_name="Tags",
+                    ),
+                ),
+            ],
+            options={
+                "abstract": False,
+                "unique_together": {("expression_choice", "config_string", "remediation_function")},
+            },
+            bases=(
+                models.Model,
+                nautobot.extras.models.mixins.DynamicGroupMixin,
+                nautobot.extras.models.mixins.NotesMixin,
+            ),
+        ),
+    ]