diff --git a/README.md b/README.md index a551366..1963664 100644 --- a/README.md +++ b/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: ``` diff --git a/docker/netbox/configuration/plugins.py b/docker/netbox/configuration/plugins.py index da31c76..b5c275c 100644 --- a/docker/netbox/configuration/plugins.py +++ b/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 + } +} diff --git a/netbox_diode_plugin/forms.py b/netbox_diode_plugin/forms.py index de05e77..4ae4a47 100644 --- a/netbox_diode_plugin/forms.py +++ b/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." + ) diff --git a/netbox_diode_plugin/migrations/0001_initial.py b/netbox_diode_plugin/migrations/0001_initial.py index 3441166..f900bd0 100644 --- a/netbox_diode_plugin/migrations/0001_initial.py +++ b/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) diff --git a/netbox_diode_plugin/migrations/0002_setting.py b/netbox_diode_plugin/migrations/0002_setting.py index 0225897..d7a7671 100644 --- a/netbox_diode_plugin/migrations/0002_setting.py +++ b/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): diff --git a/netbox_diode_plugin/tests/test_forms.py b/netbox_diode_plugin/tests/test_forms.py new file mode 100644 index 0000000..16de118 --- /dev/null +++ b/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) diff --git a/netbox_diode_plugin/tests/test_views.py b/netbox_diode_plugin/tests/test_views.py index b4584b1..8eef138 100644 --- a/netbox_diode_plugin/tests/test_views.py +++ b/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.", + ) diff --git a/netbox_diode_plugin/views.py b/netbox_diode_plugin/views.py index 37c89a0..51814d6 100644 --- a/netbox_diode_plugin/views.py +++ b/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