wip

validity/api/serializers.py

@@ -287,3 +287,22 @@ class Meta(NestedDeviceSerializer.Meta):
             "results_count",
             "results",
         ]
+
+
+class KeyBundleSerializer(NetBoxModelSerializer):
+    url = serializers.HyperlinkedIdentityField(view_name="plugins-api:validity-api:keybundle-detail")
+    class Meta:
+        model = models.KeyBundle
+        fields = (
+            "id",
+            "url",
+            "display",
+            "name",
+            "connection_type",
+            "public_credentials",
+            "private_credentials",
+            "tags",
+            "custom_fields",
+            "created",
+            "last_updated",
+        )

validity/api/urls.py

@@ -13,6 +13,7 @@
 router.register("serializers", views.ConfigSerializerViewSet)
 router.register("namesets", views.NameSetViewSet)
 router.register("reports", views.ComplianceReportViewSet)
+router.register("keybundles", views.KeyBundleViewSet)
 
 urlpatterns = [
     path("reports/<int:pk>/devices/", views.DeviceReportView.as_view(), name="report_devices"),

validity/api/views.py

@@ -70,6 +70,12 @@ class ComplianceReportViewSet(NetBoxModelViewSet):
     http_method_names = ["get", "head", "options", "trace", "delete"]
 
 
+class KeyBundleViewSet(NetBoxModelViewSet):
+    queryset = models.KeyBundle.objects.prefetch_related('tags')
+    serializer_class = serializers.KeyBundleSerializer
+    filterset_class = filtersets.KeyBundleFilterSet
+
+
 class DeviceReportView(ListAPIView):
     serializer_class = serializers.DeviceReportSerializer
     filterset_class = filtersets.DeviceReportFilterSet

validity/choices.py

@@ -107,3 +107,8 @@ def viewname(self) -> str:
     def pk_field(self):
         pk_path = self.value.split("__")[:-1] + ["pk"]
         return "__".join(pk_path)
+
+
+class ConnectionTypeChoices(TextChoices, metaclass=ColoredChoiceMeta):
+    NETMIKO = 'netmiko', 'blue'
+    GENERIC = 'generic', 'grey'

validity/filtersets.py

@@ -111,3 +111,10 @@ def get_filters(cls):
 
 class DeviceReportFilterSet(DeviceFilterSet):
     compliance_passed = BooleanFilter()
+
+
+class KeyBundleFilterSet(SearchMixin, NetBoxModelFilterSet):
+    class Meta:
+        model = models.KeyBundle
+        fields = ("id", "name", "connection_type")
+        search_fields = ("name",)

validity/forms/__init__.py

@@ -7,5 +7,6 @@
     NameSetFilterForm,
     ReportGroupByForm,
     TestResultFilterForm,
+    KeyBundleFilterForm
 )
-from .general import ComplianceSelectorForm, ComplianceTestForm, ConfigSerializerForm, NameSetForm
+from .general import ComplianceSelectorForm, ComplianceTestForm, ConfigSerializerForm, NameSetForm, KeyBundleForm

validity/forms/filterset.py

@@ -14,6 +14,7 @@
     DeviceGroupByChoices,
     DynamicPairsChoices,
     SeverityChoices,
+    ConnectionTypeChoices,
 )
 from .helpers import ExcludeMixin, PlaceholderChoiceField
 
@@ -138,3 +139,9 @@ class ComplianceTestFilterForm(NetBoxModelFilterSetForm):
     datasource_id = DynamicModelMultipleChoiceField(
         label=_("Data Source"), queryset=DataSource.objects.all(), required=False
     )
+
+
+class KeyBundleFilterForm(NetBoxModelFilterSetForm):
+    model = models.KeyBundle
+    name = CharField(required=False)
+    connection_type = PlaceholderChoiceField(required=False, placeholder=_("Connection Type"), choices=ConnectionTypeChoices.choices)

validity/forms/general.py

@@ -109,3 +109,9 @@ class NameSetForm(NetBoxModelForm):
     class Meta:
         model = models.NameSet
         fields = ("name", "description", "_global", "tests", "definitions", "data_source", "data_file", "tags")
+
+
+class KeyBundleForm(NetBoxModelForm):
+    class Meta:
+        model = models.KeyBundle
+        fields = ("name", "connection_type", "public_credentials", "private_credentials", "tags")

validity/graphql.py

@@ -134,6 +134,20 @@ def get_queryset(cls, queryset, info):
         return queryset.annotate_result_stats().count_devices_and_tests()
 
 
+class KeyBundleType(NetBoxObjectType):
+    class Meta:
+        model = models.KeyBundle
+        fields = (
+            "id",
+            "name",
+            "connection_type",
+            "public_credentials",
+            "private_credentials",
+            "created",
+            "last_updated",
+        )
+
+
 class Query(ObjectType):
     nameset = ObjectField(NameSetType)
     nameset_list = ObjectListField(NameSetType)
@@ -153,5 +167,8 @@ class Query(ObjectType):
     report = ObjectField(ReportType)
     report_list = ObjectListField(ReportType)
 
+    keybundle = ObjectField(KeyBundleType)
+    keybundle_list = ObjectListField(KeyBundleType)
+
 
 schema = Query

validity/managers.py

@@ -1,3 +1,4 @@
+from functools import partialmethod
 from itertools import chain
 from typing import TypeVar
 
@@ -182,38 +183,58 @@ def prefetch_datasource(self: _QS, prefetch_config_files: bool = False) -> _QS:
             datasource_qs = datasource_qs.prefetch_config_files()
         return self.annotate_datasource_id().custom_prefetch("data_source", datasource_qs)
 
-    def annotate_serializer_id(self: _QS) -> _QS:
+    def annotate_cf(self, cf: str, annotation: str = ""):
+        """
+        Annotates CF value (in decreasing precedence):
+            1) From device itself
+            2) from device type
+            3) from manufacturer
+        """
+        annotation = annotation or cf
+        device_cf = f"device_{cf}"
+        devtype_cf = f"devtype_{cf}"
+        manuf_cf = f"manuf_{cf}"
         return (
-            self.annotate(device_s=KeyTextTransform("serializer", "custom_field_data"))
+            self.annotate(**{device_cf: KeyTextTransform(cf, "custom_field_data")})
             .annotate(
-                devtype_s=KeyTextTransform("serializer", "device_type__custom_field_data"),
+                **{devtype_cf: KeyTextTransform(cf, "device_type__custom_field_data")},
             )
-            .annotate(manufacturer_s=KeyTextTransform("serializer", "device_type__manufacturer__custom_field_data"))
+            .annotate(**{manuf_cf: KeyTextTransform(cf, "device_type__manufacturer__custom_field_data")})
             .annotate(
-                serializer_id=Case(
-                    When(device_s__isnull=False, then=Cast(F("device_s"), BigIntegerField())),
-                    When(devtype_s__isnull=False, then=Cast(F("devtype_s"), BigIntegerField())),
-                    When(manufacturer_s__isnull=False, then=Cast(F("manufacturer_s"), BigIntegerField())),
-                )
+                **{
+                    annotation: Case(
+                        When(**{f"{device_cf}__isnull": False, "then": Cast(F(device_cf), BigIntegerField())}),
+                        When(**{f"{devtype_cf}__isnull": False, "then": Cast(F(devtype_cf), BigIntegerField())}),
+                        When(**{f"{manuf_cf}__isnull": False, "then": Cast(F(manuf_cf), BigIntegerField())}),
+                    )
+                }
             )
         )
 
-    def prefetch_serializer(self: _QS) -> _QS:
+    annotate_serializer_id = partialmethod(annotate_cf, "serializer", "serializer_id")
+    annotate_keybundle_id = partialmethod(annotate_cf, "keybundle", "keybundle_id")
+
+    def prefetch_serializer(self):
         from validity.models import ConfigSerializer
 
         return self.annotate_serializer_id().custom_prefetch(
             "serializer", ConfigSerializer.objects.select_related("data_file")
         )
 
+    def prefetch_keybundle(self):
+        from validity.models import KeyBundle
+
+        return self.annotate_keybundle_id().custom_prefetch("keybundle", KeyBundle.objects.all())
+
     def _count_per_something(self, field: str, annotate_method: str) -> dict[int | None, int]:
         qs = getattr(self, annotate_method)().values(field).annotate(cnt=Count("id", distinct=True))
         result = {}
         for values in qs:
             result[values[field]] = values["cnt"]
         return result
 
-    def count_per_serializer(self) -> dict[int | None, int]:
-        return self._count_per_something("serializer_id", "annotate_serializer_id")
+    count_per_serializer = partialmethod(_count_per_something, "serializer_id", "annotate_serializer_id")
+    count_per_keybundle = partialmethod(_count_per_something, "keybundle_id", "annotate_keybundle_id")
 
     def annotate_result_stats(self, report_id: int, severity_ge: SeverityChoices = SeverityChoices.LOW):
         results_filter = Q(results__report__pk=report_id) & self._severity_filter(severity_ge, "results")

validity/migrations/0005_datasources.py

@@ -1,7 +1,7 @@
 from itertools import chain
 from django.db import migrations
 from django.utils.translation import gettext_lazy as _
-from validity.utils.password import EncryptedString
+from validity.utils.dbfields import EncryptedString
 from django.db import migrations, models
 import django.db.models.deletion
 from validity.utils.misc import datasource_sync

validity/migrations/0007_keybundle.py

@@ -0,0 +1,71 @@
+# Generated by Django 4.2.7 on 2023-12-03 20:00
+
+from django.db import migrations, models
+import taggit.managers
+import utilities.json
+import validity.models.base
+import validity.utils.dbfields
+from django.utils.translation import gettext_lazy as _
+
+
+
+def create_cf(apps, schema_editor):
+    ContentType = apps.get_model("contenttypes", "ContentType")
+    CustomField = apps.get_model("extras", "CustomField")
+    KeyBundle = apps.get_model("validity", "KeyBundle")
+    Device = apps.get_model("dcim", "Device")
+    DeviceType = apps.get_model("dcim", "DeviceType")
+    Manufacturer = apps.get_model("dcim", "Manufacturer")
+    db_alias = schema_editor.connection.alias
+
+    cf = CustomField.objects.using(db_alias).create(
+        name="keybundle",
+        label=_("Key Bundle"),
+        description=_("Required by Validity. Defines properties of device connection"),
+        type="object",
+        object_type=ContentType.objects.get_for_model(KeyBundle),
+        required=False,
+    )
+    cf.content_types.set([
+        ContentType.objects.get_for_model(Device),
+        ContentType.objects.get_for_model(DeviceType),
+        ContentType.objects.get_for_model(Manufacturer),
+    ])
+
+
+def delete_cf(apps, schema_editor):
+    CustomField = apps.get_model("extras", "CustomField")
+    db_alias = schema_editor.connection.alias
+    CustomField.objects.using(db_alias).filter(name="keybundle").delete()
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("extras", "0098_webhook_custom_field_data_webhook_tags"),
+        ("validity", "0006_script_change"),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name="KeyBundle",
+            fields=[
+                ("id", models.BigAutoField(auto_created=True, primary_key=True, serialize=False)),
+                ("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=utilities.json.CustomFieldJSONEncoder),
+                ),
+                ("name", models.CharField(max_length=255)),
+                ("connection_type", models.CharField(max_length=50)),
+                ("public_credentials", models.JSONField(default=dict)),
+                ("private_credentials", validity.utils.dbfields.EncryptedDictField()),
+                ("tags", taggit.managers.TaggableManager(through="extras.TaggedItem", to="extras.Tag")),
+            ],
+            options={
+                "abstract": False,
+            },
+            bases=(validity.models.base.URLMixin, models.Model),
+        ),
+        migrations.RunPython(create_cf, delete_cf)
+    ]

validity/models/__init__.py

@@ -6,3 +6,4 @@
 from .serializer import ConfigSerializer
 from .test import ComplianceTest
 from .test_result import ComplianceTestResult
+from .keybundle import KeyBundle

validity/models/keybundle.py

@@ -0,0 +1,27 @@
+from dcim.models import Device
+from django.db import models
+from django.utils.translation import gettext_lazy as _
+
+from validity.choices import ConnectionTypeChoices
+from validity.utils.dbfields import EncryptedDictField
+from .base import BaseModel
+
+
+class KeyBundle(BaseModel):
+    name = models.CharField(_("Name"), max_length=255)
+    connection_type = models.CharField(_("Connection Type"), max_length=50, choices=ConnectionTypeChoices.choices)
+    public_credentials = models.JSONField(_("Public Credentials"), default=dict, blank=True)
+    private_credentials = EncryptedDictField(_("Private Credentials"), blank=True)
+
+    @property
+    def credentials(self):
+        return self.public_credentials | self.private_credentials.decrypted
+
+    def get_connection_type_color(self):
+        return ConnectionTypeChoices.colors.get(self.connection_type)
+
+    @property
+    def bound_devices(self) -> models.QuerySet[Device]:
+        from .device import VDevice
+
+        return VDevice.objects.annotate_keybundle_id().filter(keybundle_id=self.pk)

validity/navigation.py

@@ -59,6 +59,18 @@
             ),
         ],
     ),
+    PluginMenuItem(
+        link="plugins:validity:keybundle_list",
+        link_text="Key Bundles",
+        buttons=[
+            PluginMenuButton(
+                link="plugins:validity:keybundle_add",
+                title="Add",
+                icon_class="mdi mdi-plus-thick",
+                color=ButtonColorChoices.GREEN,
+            ),
+        ],
+    ),
 )
 
 menu = PluginMenu(

validity/search.py

@@ -31,3 +31,9 @@ class SerializerIndex(SearchIndex):
 class TestIndex(SearchIndex):
     model = models.ComplianceTest
     fields = (("name", 100), ("expression", 1000))
+
+
+@register_search
+class KeyBundleIndex(SearchIndex):
+    model = models.KeyBundle
+    fields = (("name", 100),)

validity/tables.py

@@ -75,22 +75,42 @@ class Meta(NetBoxTable.Meta):
         default_columns = fields
 
 
-class ConfigSerializerTable(NetBoxTable):
+
+class TotalDevicesMixin(NetBoxTable):
+    total_devices = Column(empty_values=())
+
+    count_per: str
+
+    def __init__(self, *args, extra_columns=None, **kwargs):
+        super().__init__(*args, extra_columns=extra_columns, **kwargs)
+        self.total_devices_map = getattr(models.VDevice.objects, f'count_per_{self.count_per}')()
+
+    def render_total_devices(self, record):
+        return self.total_devices_map.get(record.id, 0)
+
+
+class ConfigSerializerTable(TotalDevicesMixin, NetBoxTable):
     name = Column(linkify=True)
     extraction_method = ChoiceFieldColumn()
-    total_devices = Column(empty_values=())
+
+    count_per = 'serializer'
 
     class Meta(NetBoxTable.Meta):
         model = models.ConfigSerializer
         fields = ("name", "extraction_method", "total_devices")
         default_columns = fields
 
-    def __init__(self, *args, extra_columns=None, **kwargs):
-        super().__init__(*args, extra_columns=extra_columns, **kwargs)
-        self.total_devices_map = models.VDevice.objects.count_per_serializer()
 
-    def render_total_devices(self, record):
-        return self.total_devices_map.get(record.id, 0)
+class KeyBundleTable(TotalDevicesMixin, NetBoxTable):
+    name = Column(linkify=True)
+    connection_type = ChoiceFieldColumn()
+
+    count_per = 'keybundle'
+
+    class Meta(NetBoxTable.Meta):
+        model = models.KeyBundle
+        fields = ("name", "connection_type", "total_devices")
+        default_columns = fields
 
 
 class ExplanationColumn(Column):