feat: plugin configuration improvements (#30)

* feat: plugin config: diode target and disallow diode target override

Signed-off-by: Michal Fiedorowicz <[email protected]>

* feat: read secrets for api keys from mounted path

Signed-off-by: Michal Fiedorowicz <[email protected]>

* tests for views and forms

Signed-off-by: Michal Fiedorowicz <[email protected]>

---------

Signed-off-by: Michal Fiedorowicz <[email protected]>

Author: mfiedorowicz

README.md

@@ -40,6 +40,17 @@ PLUGINS = [
 ]
 ```
 
+Also in your `configuration.py` file, add `netbox_diode_plugin`to the `PLUGINS_CONFIG` dictionary, e.g.:
+
+```python
+PLUGINS_CONFIG = {
+    "netbox_diode_plugin": {
+        "diode_target": "grpc://localhost:8080/diode", # The Diode gRPC target for communication with Diode server, default: "grpc://localhost:8080/diode"
+        "disallow_diode_target_override": True, # Disallow the Diode target to be overridden by the user, default: False
+    }
+}
+```
+
 Restart NetBox services to load the plugin:
 
 ```

docker/netbox/configuration/plugins.py

@@ -6,8 +6,9 @@
 
 PLUGINS = ["netbox_diode_plugin"]
 
-# PLUGINS_CONFIG = {
-#     "netbox_diode_plugin": {
-#
-#     }
-# }
+PLUGINS_CONFIG = {
+    "netbox_diode_plugin": {
+        "diode_target": "grpc://localhost:8080/diode", # The Diode gRPC target for communication with Diode server, default: "grpc://localhost:8080/diode"
+        "disallow_diode_target_override": False, # Disallow the Diode target to be overridden by the user, default: False
+    }
+}

netbox_diode_plugin/forms.py

@@ -1,7 +1,7 @@
 # !/usr/bin/env python
 # Copyright 2024 NetBox Labs Inc
 """Diode NetBox Plugin - Forms."""
-
+from django.conf import settings as netbox_settings
 from netbox.forms import NetBoxModelForm
 from utilities.forms.rendering import FieldSet
 
@@ -24,3 +24,17 @@ class Meta:
 
         model = Setting
         fields = ("diode_target",)
+
+    def __init__(self, *args, **kwargs):
+        """Initialize the form."""
+        super().__init__(*args, **kwargs)
+
+        disallow_diode_target_override = netbox_settings.PLUGINS_CONFIG.get(
+            "netbox_diode_plugin", {}
+        ).get("disallow_diode_target_override", False)
+
+        if disallow_diode_target_override:
+            self.fields["diode_target"].disabled = True
+            self.fields["diode_target"].help_text = (
+                "This field is not allowed to be overridden."
+            )

netbox_diode_plugin/migrations/0001_initial.py

@@ -11,6 +11,17 @@
 from users.models import Token as NetBoxToken
 
 
+# Read secret from file
+def _read_secret(secret_name, default=None):
+    try:
+        f = open("/run/secrets/" + secret_name, encoding="utf-8")
+    except OSError:
+        return default
+    else:
+        with f:
+            return f.readline().strip()
+
+
 def _create_user_with_token(apps, username, group, is_superuser: bool = False):
     User = apps.get_model(settings.AUTH_USER_MODEL)
     """Create a user with the given username and API key if it does not exist."""
@@ -27,7 +38,8 @@ def _create_user_with_token(apps, username, group, is_superuser: bool = False):
     Token = apps.get_model("users", "Token")
 
     if not Token.objects.filter(user=user).exists():
-        api_key = os.getenv(f"{username}_API_KEY")
+        key = f"{username}_API_KEY"
+        api_key = _read_secret(key.lower(), os.getenv(key))
         if api_key is None:
             api_key = NetBoxToken.generate_key()
         Token.objects.create(user=user, key=api_key)

netbox_diode_plugin/migrations/0002_setting.py

@@ -3,13 +3,19 @@
 """Diode Netbox Plugin - Database migrations."""
 
 import utilities.json
+from django.conf import settings as netbox_settings
 from django.db import migrations, models
 
 
 def create_settings_entity(apps, schema_editor):
     """Create a Setting entity."""
     Setting = apps.get_model("netbox_diode_plugin", "Setting")
-    Setting.objects.create(diode_target="grpc://localhost:8080/diode")
+
+    diode_target = netbox_settings.PLUGINS_CONFIG.get(
+        "netbox_diode_plugin", {}
+    ).get("diode_target", "grpc://localhost:8080/diode")
+
+    Setting.objects.create(diode_target=diode_target)
 
 
 class Migration(migrations.Migration):

netbox_diode_plugin/tests/test_forms.py

@@ -0,0 +1,41 @@
+#!/usr/bin/env python
+# Copyright 2024 NetBox Labs Inc
+"""Diode NetBox Plugin - Tests."""
+from unittest import mock
+
+from django.test import TestCase
+
+from netbox_diode_plugin.forms import SettingsForm
+from netbox_diode_plugin.models import Setting
+
+
+class SettingsFormTestCase(TestCase):
+    """Test case for the SettingsForm."""
+
+    def setUp(self):
+        """Set up the test case."""
+        self.setting = Setting.objects.create(diode_target="grpc://localhost:8080/diode")
+
+    def test_form_initialization_with_override_allowed(self):
+        """Test form initialization when override is allowed."""
+        with mock.patch("netbox_diode_plugin.forms.netbox_settings") as mock_settings:
+            mock_settings.PLUGINS_CONFIG = {
+                "netbox_diode_plugin": {
+                    "disallow_diode_target_override": False
+                }
+            }
+            form = SettingsForm(instance=self.setting)
+            self.assertFalse(form.fields["diode_target"].disabled)
+            self.assertNotIn("This field is not allowed to be overridden.", form.fields["diode_target"].help_text)
+
+    def test_form_initialization_with_override_disallowed(self):
+        """Test form initialization when override is disallowed."""
+        with mock.patch("netbox_diode_plugin.forms.netbox_settings") as mock_settings:
+            mock_settings.PLUGINS_CONFIG = {
+                "netbox_diode_plugin": {
+                    "disallow_diode_target_override": True
+                }
+            }
+            form = SettingsForm(instance=self.setting)
+            self.assertTrue(form.fields["diode_target"].disabled)
+            self.assertEqual("This field is not allowed to be overridden.", form.fields["diode_target"].help_text)

netbox_diode_plugin/tests/test_views.py

@@ -6,10 +6,12 @@
 from django.contrib.auth import get_user_model
 from django.contrib.auth.models import AnonymousUser
 from django.contrib.messages.middleware import MessageMiddleware
+from django.contrib.messages.storage.fallback import FallbackStorage
 from django.contrib.sessions.middleware import SessionMiddleware
 from django.core.cache import cache
 from django.test import RequestFactory, TestCase
 from django.urls import reverse
+from rest_framework import status
 
 from netbox_diode_plugin.models import Setting
 from netbox_diode_plugin.reconciler.sdk.v1 import ingester_pb2, reconciler_pb2
@@ -35,7 +37,7 @@ def test_returns_200_for_authenticated(self):
         self.request.user.is_staff = True
 
         response = self.view.get(self.request)
-        self.assertEqual(response.status_code, 200)
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
 
     def test_redirects_to_login_page_for_unauthenticated_user(self):
         """Test that the view returns 200 for an authenticated user."""
@@ -44,7 +46,7 @@ def test_redirects_to_login_page_for_unauthenticated_user(self):
 
         response = IngestionLogsView.as_view()(self.request)
 
-        self.assertEqual(response.status_code, 302)
+        self.assertEqual(response.status_code, status.HTTP_302_FOUND)
         self.assertEqual(response.url, f"/netbox/login/?next={self.path}")
 
     def test_ingestion_logs_failed_to_retrieve(self):
@@ -53,7 +55,7 @@ def test_ingestion_logs_failed_to_retrieve(self):
         self.request.user.is_staff = True
 
         response = self.view.get(self.request)
-        self.assertEqual(response.status_code, 200)
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
         self.assertIn(
             "UNAVAILABLE: failed to connect to all addresses;", str(response.content)
         )
@@ -100,7 +102,7 @@ def test_ingestion_logs_retrieve_logs(self):
             response = self.view.get(self.request)
             mock_retrieve_ingestion_logs.assert_called()
             self.assertEqual(mock_retrieve_ingestion_logs.call_count, 2)
-            self.assertEqual(response.status_code, 200)
+            self.assertEqual(response.status_code, status.HTTP_200_OK)
             self.assertNotIn("Server Error", str(response.content))
 
     def test_cached_metrics(self):
@@ -153,7 +155,7 @@ def test_cached_metrics(self):
             response = self.view.get(self.request)
             mock_retrieve_ingestion_logs.assert_called()
             self.assertEqual(mock_retrieve_ingestion_logs.call_count, 1)
-            self.assertEqual(response.status_code, 200)
+            self.assertEqual(response.status_code, status.HTTP_200_OK)
             self.assertNotIn("Server Error", str(response.content))
 
 
@@ -173,7 +175,7 @@ def test_returns_200_for_authenticated(self):
         self.request.user.is_staff = True
 
         response = self.view.get(self.request)
-        self.assertEqual(response.status_code, 200)
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
 
     def test_redirects_to_login_page_for_unauthenticated_user(self):
         """Test that the view returns 200 for an authenticated user."""
@@ -182,7 +184,7 @@ def test_redirects_to_login_page_for_unauthenticated_user(self):
 
         response = SettingsView.as_view()(self.request)
 
-        self.assertEqual(response.status_code, 302)
+        self.assertEqual(response.status_code, status.HTTP_302_FOUND)
         self.assertEqual(response.url, f"/netbox/login/?next={self.path}")
 
     def test_settings_created_if_not_found(self):
@@ -194,7 +196,7 @@ def test_settings_created_if_not_found(self):
             mock_get.side_effect = Setting.DoesNotExist
 
             response = self.view.get(self.request)
-            self.assertEqual(response.status_code, 200)
+            self.assertEqual(response.status_code, status.HTTP_200_OK)
             self.assertIn(
                 "grpc://localhost:8080/diode", str(response.content)
             )
@@ -218,7 +220,7 @@ def test_returns_200_for_authenticated(self):
         self.view.setup(request)
 
         response = self.view.get(request)
-        self.assertEqual(response.status_code, 200)
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
 
     def test_redirects_to_login_page_for_unauthenticated_user(self):
         """Test that the view redirects an authenticated user to login page."""
@@ -227,7 +229,7 @@ def test_redirects_to_login_page_for_unauthenticated_user(self):
         self.view.setup(request)
 
         response = self.view.get(request)
-        self.assertEqual(response.status_code, 302)
+        self.assertEqual(response.status_code, status.HTTP_302_FOUND)
         self.assertEqual(response.url, f"/netbox/login/?next={self.path}")
 
     def test_settings_updated(self):
@@ -241,7 +243,7 @@ def test_settings_updated(self):
         self.view.setup(request)
 
         response = self.view.get(request)
-        self.assertEqual(response.status_code, 200)
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
         self.assertIn("grpc://localhost:8080/diode", str(response.content))
 
         request = self.request_factory.post(self.path)
@@ -258,7 +260,7 @@ def test_settings_updated(self):
         request.session.save()
 
         response = self.view.post(request)
-        self.assertEqual(response.status_code, 302)
+        self.assertEqual(response.status_code, status.HTTP_302_FOUND)
         self.assertEqual(response.url, reverse("plugins:netbox_diode_plugin:settings"))
 
         request = self.request_factory.get(self.path)
@@ -267,7 +269,7 @@ def test_settings_updated(self):
         self.view.setup(request)
 
         response = self.view.get(request)
-        self.assertEqual(response.status_code, 200)
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
         self.assertIn("grpc://localhost:8090/diode", str(response.content))
 
     def test_settings_update_post_redirects_to_login_page_for_unauthenticated_user(
@@ -280,5 +282,45 @@ def test_settings_update_post_redirects_to_login_page_for_unauthenticated_user(
         request.POST = {"diode_target": "grpc://localhost:8090/diode"}
 
         response = self.view.post(request)
-        self.assertEqual(response.status_code, 302)
+        self.assertEqual(response.status_code, status.HTTP_302_FOUND)
         self.assertEqual(response.url, f"/netbox/login/?next={self.path}")
+
+    def test_settings_update_disallowed(self):
+        """Test that the Diode target cannot be overridden."""
+        with mock.patch("netbox_diode_plugin.views.netbox_settings") as mock_settings:
+            mock_settings.PLUGINS_CONFIG = {
+                "netbox_diode_plugin": {
+                    "disallow_diode_target_override": True
+                }
+            }
+
+            user = User.objects.create_user("foo", password="pass")
+            user.is_staff = True
+
+            request = self.request_factory.post(self.path)
+            request.user = user
+            request.htmx = None
+            request.POST = {"diode_target": "grpc://localhost:8090/diode"}
+
+            middleware = SessionMiddleware(get_response=lambda request: None)
+            middleware.process_request(request)
+            request.session.save()
+
+            middleware = MessageMiddleware(get_response=lambda request: None)
+            middleware.process_request(request)
+            request.session.save()
+
+            setattr(request, 'session', 'session')
+            messages = FallbackStorage(request)
+            request._messages = messages
+
+            self.view.setup(request)
+            response = self.view.post(request)
+
+            self.assertEqual(response.status_code, status.HTTP_302_FOUND)
+            self.assertEqual(response.url, reverse("plugins:netbox_diode_plugin:settings"))
+            self.assertEqual(len(request._messages._queued_messages), 1)
+            self.assertEqual(
+                str(request._messages._queued_messages[0]),
+                "The Diode target is not allowed to be overridden.",
+            )

netbox_diode_plugin/views.py

@@ -2,6 +2,7 @@
 # Copyright 2024 NetBox Labs Inc
 """Diode NetBox Plugin - Views."""
 from django.conf import settings as netbox_settings
+from django.contrib import messages
 from django.contrib.auth import get_user_model
 from django.core.cache import cache
 from django.shortcuts import redirect, render
@@ -51,7 +52,10 @@ def get(self, request):
             table = IngestionLogsTable(resp.logs)
 
             cached_ingestion_metrics = cache.get(self.INGESTION_METRICS_CACHE_KEY)
-            if cached_ingestion_metrics is not None and cached_ingestion_metrics["total"] == resp.metrics.total:
+            if (
+                cached_ingestion_metrics is not None
+                and cached_ingestion_metrics["total"] == resp.metrics.total
+            ):
                 metrics = cached_ingestion_metrics
             else:
                 ingestion_metrics = reconciler_client.retrieve_ingestion_logs(
@@ -150,6 +154,15 @@ def post(self, request, *args, **kwargs):
         if not request.user.is_authenticated or not request.user.is_staff:
             return redirect(f"{netbox_settings.LOGIN_URL}?next={request.path}")
 
+        if netbox_settings.PLUGINS_CONFIG.get("netbox_diode_plugin", {}).get(
+            "disallow_diode_target_override", False
+        ):
+            messages.error(
+                request,
+                "The Diode target is not allowed to be overridden.",
+            )
+            return redirect("plugins:netbox_diode_plugin:settings")
+
         settings = Setting.objects.get()
         kwargs["pk"] = settings.pk