diff --git a/netbox_librenms_plugin/forms.py b/netbox_librenms_plugin/forms.py index f3e3d075e5..4116eea15b 100644 --- a/netbox_librenms_plugin/forms.py +++ b/netbox_librenms_plugin/forms.py @@ -577,19 +577,19 @@ def _populate_librenms_locations(self): """Fetch and populate LibreNMS locations in the dropdown.""" from django.core.cache import cache + from netbox_librenms_plugin.import_utils.cache import get_location_choices_cache_key from netbox_librenms_plugin.librenms_api import LibreNMSAPI try: - # Use caching to avoid repeated API calls - cache_key = "librenms_locations_choices" + # Instantiate the API client to resolve the authoritative server_key + api = LibreNMSAPI() + cache_key = get_location_choices_cache_key(api.server_key) cached_choices = cache.get(cache_key) - if cached_choices: self.fields["librenms_location"].choices = cached_choices return # Fetch locations from LibreNMS - api = LibreNMSAPI() success, locations = api.get_locations() if success and locations: diff --git a/netbox_librenms_plugin/import_utils/bulk_import.py b/netbox_librenms_plugin/import_utils/bulk_import.py index 8a4ce16cae..58e893d8f4 100644 --- a/netbox_librenms_plugin/import_utils/bulk_import.py +++ b/netbox_librenms_plugin/import_utils/bulk_import.py @@ -129,6 +129,7 @@ def bulk_import_devices_shared( api=api, use_sysname=use_sysname_opt, strip_domain=strip_domain_opt, + server_key=api.server_key, ) # Build manual mappings from validation + any provided overrides @@ -148,7 +149,7 @@ def bulk_import_devices_shared( result = import_single_device( device_id, - server_key=server_key, + server_key=api.server_key, # use resolved key, not raw parameter (may be None) sync_options=sync_options, manual_mappings=device_mappings if device_mappings else None, libre_device=libre_device, @@ -503,6 +504,7 @@ def process_device_filters( api=api_for_validation, include_vc_detection=vc_detection_enabled, force_vc_refresh=clear_cache, + server_key=api.server_key, use_sysname=use_sysname, strip_domain=strip_domain, ) diff --git a/netbox_librenms_plugin/import_utils/cache.py b/netbox_librenms_plugin/import_utils/cache.py index fe0896a302..6cc914e4d9 100644 --- a/netbox_librenms_plugin/import_utils/cache.py +++ b/netbox_librenms_plugin/import_utils/cache.py @@ -7,6 +7,11 @@ logger = logging.getLogger(__name__) +def get_location_choices_cache_key(server_key: str) -> str: + """Return the cache key for LibreNMS location choices for a given server.""" + return f"librenms_locations_choices:{server_key}" + + def get_cache_metadata_key(server_key: str, filters: dict, vc_enabled: bool) -> str: """ Generate a consistent cache metadata key from filter parameters. @@ -62,7 +67,7 @@ def get_active_cached_searches(server_key: str) -> list[dict]: } # Get cached location choices for enrichment - location_cache_key = "librenms_locations_choices" + location_cache_key = get_location_choices_cache_key(server_key) cached_locations = cache.get(location_cache_key) if cached_locations: location_choices = dict(cached_locations) diff --git a/netbox_librenms_plugin/import_utils/device_operations.py b/netbox_librenms_plugin/import_utils/device_operations.py index c25fff7b8e..165cb765de 100644 --- a/netbox_librenms_plugin/import_utils/device_operations.py +++ b/netbox_librenms_plugin/import_utils/device_operations.py @@ -14,6 +14,7 @@ find_matching_platform, find_matching_site, match_librenms_hardware_to_device_type, + set_librenms_device_id, ) from .cache import get_import_device_cache_key from .virtual_chassis import ( @@ -129,6 +130,7 @@ def validate_device_for_import( force_vc_refresh: bool = False, use_sysname: bool = True, strip_domain: bool = False, + server_key: str = "default", ) -> dict: """ Validate if a LibreNMS device can be imported to NetBox. @@ -203,6 +205,7 @@ def validate_device_for_import( "serial_action": None, # None, "link", "conflict", "update_serial", "hostname_differs" "serial_confirmed": False, # True when librenms_id match and serial matches "serial_duplicate": False, # True when incoming serial is already on a different device + "librenms_id_needs_migration": False, # True when existing device has legacy bare-int ID "name_matches": False, # True when existing device name matches LibreNMS sysName "name_sync_available": False, # True when existing device name differs from sysName "suggested_name": None, # sysName to suggest when name_sync_available is True @@ -275,10 +278,13 @@ def validate_device_for_import( from virtualization.models import VirtualMachine + server_key = api.server_key if api is not None else server_key + # Check for existing VM first (by librenms_id custom field) - # Always query with int to match custom field type try: - existing_vm = VirtualMachine.objects.filter(custom_field_data__librenms_id=int(librenms_id)).first() + from netbox_librenms_plugin.utils import find_by_librenms_id + + existing_vm = find_by_librenms_id(VirtualMachine, int(librenms_id), server_key) except (ValueError, TypeError): # librenms_id is not convertible to int; no match will be found existing_vm = None @@ -290,7 +296,17 @@ def validate_device_for_import( result["import_as_vm"] = True # Force VM mode since VM exists result["can_import"] = False - # Check if name matches sysName + # Detect legacy bare-integer or string-digit format so UI can offer a migration action. + # Direct access needed to detect legacy format for migration prompt: + # LibreNMSAPI.get_librenms_id() returns an int in both formats, so only the + # raw type check on custom_field_data reveals whether migration is needed. + _vm_cf_id = existing_vm.custom_field_data.get("librenms_id") + if (isinstance(_vm_cf_id, int) and not isinstance(_vm_cf_id, bool)) or ( + isinstance(_vm_cf_id, str) and _vm_cf_id.isdigit() + ): + result["librenms_id_needs_migration"] = True + + # Check if name matches resolved name (accounts for use_sysname/strip_domain) # Note: name_sync_available/suggested_name are intentionally not set for VMs # because UpdateDeviceNameView only supports Device objects; VM name-sync # would require a separate implementation. @@ -298,10 +314,11 @@ def validate_device_for_import( result["name_matches"] = True # Check for existing Device (by librenms_id custom field) - # Always query with int to match custom field type if not result["existing_device"]: try: - existing_device = Device.objects.filter(custom_field_data__librenms_id=int(librenms_id)).first() + from netbox_librenms_plugin.utils import find_by_librenms_id + + existing_device = find_by_librenms_id(Device, int(librenms_id), server_key) except (ValueError, TypeError): # librenms_id is not convertible to int; no match will be found existing_device = None @@ -312,6 +329,16 @@ def validate_device_for_import( result["existing_match_type"] = "librenms_id" result["can_import"] = False + # Detect legacy bare-integer or string-digit format so UI can offer a migration action. + # Direct access needed to detect legacy format for migration prompt: + # LibreNMSAPI.get_librenms_id() returns an int in both formats, so only the + # raw type check on custom_field_data reveals whether migration is needed. + _dev_cf_id = existing_device.custom_field_data.get("librenms_id") + if (isinstance(_dev_cf_id, int) and not isinstance(_dev_cf_id, bool)) or ( + isinstance(_dev_cf_id, str) and _dev_cf_id.isdigit() + ): + result["librenms_id_needs_migration"] = True + # Check if name matches resolved name (VC-aware: compare against VC member name) if hostname and existing_device.virtual_chassis and existing_device.vc_position: incoming_serial = libre_device.get("serial") or "" @@ -726,6 +753,7 @@ def import_single_device( api=api, use_sysname=use_sysname_opt, strip_domain=strip_domain_opt, + server_key=api.server_key, ) # Check if device already exists @@ -811,7 +839,6 @@ def import_single_device( "role": device_role, "status": "active" if libre_device.get("status") == 1 else "offline", "comments": f"Imported from LibreNMS by netbox-librenms-plugin on {import_time}", - "custom_field_data": {"librenms_id": int(device_id)}, } # Add optional fields @@ -836,6 +863,7 @@ def import_single_device( # Create the device device = Device(**device_data) + set_librenms_device_id(device, device_id, api.server_key) device.full_clean() device.save() diff --git a/netbox_librenms_plugin/import_utils/vm_operations.py b/netbox_librenms_plugin/import_utils/vm_operations.py index b75b69f7a9..d35ea1dd4b 100644 --- a/netbox_librenms_plugin/import_utils/vm_operations.py +++ b/netbox_librenms_plugin/import_utils/vm_operations.py @@ -3,6 +3,7 @@ import logging from dcim.models import DeviceRole +from django.db import transaction from django.utils import timezone from virtualization.models import Cluster @@ -13,7 +14,9 @@ logger = logging.getLogger(__name__) -def create_vm_from_librenms(libre_device: dict, validation: dict, use_sysname: bool = True, role=None): +def create_vm_from_librenms( + libre_device: dict, validation: dict, use_sysname: bool = True, role=None, server_key: str = "default" +): """ Create a NetBox VirtualMachine from LibreNMS device data. @@ -22,6 +25,7 @@ def create_vm_from_librenms(libre_device: dict, validation: dict, use_sysname: b validation: Validation result from validate_device_for_import with import_as_vm=True use_sysname: If True, prefer sysName; if False, use hostname role: Optional DeviceRole to assign to the VM + server_key: LibreNMS server key used to store the librenms_id custom field Returns: Created VirtualMachine instance @@ -58,15 +62,20 @@ def create_vm_from_librenms(libre_device: dict, validation: dict, use_sysname: b raise ValueError(f"device_id is a boolean ({raw_device_id!r}); expected an integer") librenms_device_id = int(raw_device_id) - # Create the VM with librenms_id custom field - vm = VirtualMachine.objects.create( - name=vm_name, - cluster=cluster, - role=role, # Optional VM role - platform=platform, - comments=f"Imported from LibreNMS (device_id={librenms_device_id}) by netbox-librenms-plugin on {import_time}", - custom_field_data={"librenms_id": librenms_device_id}, - ) + from ..utils import set_librenms_device_id + + # Create the VM and assign its LibreNMS ID atomically so a failure in + # set_librenms_device_id never leaves a VM without a mapping. + with transaction.atomic(): + vm = VirtualMachine.objects.create( + name=vm_name, + cluster=cluster, + role=role, # Optional VM role + platform=platform, + comments=f"Imported from LibreNMS (device_id={librenms_device_id}) by netbox-librenms-plugin on {import_time}", + ) + set_librenms_device_id(vm, librenms_device_id, server_key) + vm.save() logger.info(f"Created VM {vm.name} (ID: {vm.pk}) from LibreNMS device {libre_device['device_id']}") return vm @@ -169,6 +178,7 @@ def bulk_import_vms( api=api, use_sysname=use_sysname_opt, strip_domain=strip_domain_opt, + server_key=api.server_key, ) # Check if VM already exists @@ -213,7 +223,9 @@ def bulk_import_vms( libre_device["_computed_name"] = vm_name # Create VM - vm = create_vm_from_librenms(libre_device, validation, use_sysname=use_sysname, role=role) + vm = create_vm_from_librenms( + libre_device, validation, use_sysname=use_sysname, role=role, server_key=api.server_key + ) result["success"].append( { diff --git a/netbox_librenms_plugin/librenms_api.py b/netbox_librenms_plugin/librenms_api.py index 02f7e3459e..8e2b2b3ac0 100644 --- a/netbox_librenms_plugin/librenms_api.py +++ b/netbox_librenms_plugin/librenms_api.py @@ -190,16 +190,11 @@ def get_librenms_id(self, obj): If found via API, stores ID in custom field if available, otherwise caches the value. """ - librenms_id = obj.cf.get("librenms_id") - if librenms_id is not None: - if isinstance(librenms_id, str): - try: - librenms_id = int(librenms_id) - self._store_librenms_id(obj, librenms_id) - except (ValueError, TypeError): - librenms_id = None # empty or invalid string — fall through to discovery - if librenms_id: - return librenms_id + from netbox_librenms_plugin.utils import get_librenms_device_id + + librenms_id = get_librenms_device_id(obj, self.server_key) + if librenms_id: + return librenms_id # Check cache cache_key = self._get_cache_key(obj) @@ -261,7 +256,9 @@ def _store_librenms_id(self, obj, librenms_id): None """ if "librenms_id" in obj.cf: - obj.custom_field_data["librenms_id"] = librenms_id + from netbox_librenms_plugin.utils import set_librenms_device_id + + set_librenms_device_id(obj, librenms_id, self.server_key) obj.save() else: # Use cache as fallback diff --git a/netbox_librenms_plugin/static/netbox_librenms_plugin/js/librenms_sync.js b/netbox_librenms_plugin/static/netbox_librenms_plugin/js/librenms_sync.js index c56f9dee97..e45c54a3c2 100644 --- a/netbox_librenms_plugin/static/netbox_librenms_plugin/js/librenms_sync.js +++ b/netbox_librenms_plugin/static/netbox_librenms_plugin/js/librenms_sync.js @@ -624,7 +624,8 @@ function initializeVlanModalSave() { }, body: JSON.stringify({ device_id: deviceId, - vid_group_map: vidGroupMap + vid_group_map: vidGroupMap, + server_key: document.getElementById('current-server-key')?.value || null }) }).then(response => { if (!response.ok) { @@ -774,7 +775,8 @@ function handleVRFChange(select, value) { body: JSON.stringify({ device_id: deviceId, ip_address: fullIpAddress, // Use full IP address with prefix - vrf_id: value + vrf_id: value, + server_key: document.getElementById('current-server-key')?.value || null }) }) .then(response => { @@ -813,7 +815,8 @@ function handleInterfaceChange(select, value) { body: JSON.stringify({ device_id: value, interface_name: select.dataset.interface, - interface_name_field: document.querySelector('input[name="interface_name_field"]:checked')?.value || null + interface_name_field: document.querySelector('input[name="interface_name_field"]:checked')?.value || null, + server_key: document.getElementById('current-server-key')?.value || null }) }) .then(response => { @@ -859,7 +862,8 @@ function handleCableChange(select, value) { }, body: JSON.stringify({ device_id: value, - local_port_id: select.dataset.interface + local_port_id: select.dataset.interface, + server_key: document.getElementById('current-server-key')?.value || null }) }) .then(response => { diff --git a/netbox_librenms_plugin/tables/interfaces.py b/netbox_librenms_plugin/tables/interfaces.py index 5ceb30bf51..f95fdcd012 100644 --- a/netbox_librenms_plugin/tables/interfaces.py +++ b/netbox_librenms_plugin/tables/interfaces.py @@ -13,6 +13,7 @@ convert_speed_to_kbps, format_mac_address, get_interface_name_field, + get_librenms_device_id, get_missing_vlan_warning, get_table_paginate_count, get_tagged_vlan_css_class, @@ -46,11 +47,12 @@ class Meta: "id": "librenms-interface-table", } - def __init__(self, *args, device=None, interface_name_field=None, vlan_groups=None, **kwargs): + def __init__(self, *args, device=None, interface_name_field=None, vlan_groups=None, server_key="default", **kwargs): """Initialize table with device context and interface name field.""" self.device = device self.interface_name_field = interface_name_field or get_interface_name_field() self.vlan_groups = vlan_groups or [] + self.server_key = server_key # Update column accessors after initialization for column in ["selection", "name"]: @@ -360,7 +362,7 @@ def render_librenms_id(self, value, record): if not netbox_interface: return mark_safe(f'{value}') - netbox_librenms_id = netbox_interface.custom_field_data.get("librenms_id") + netbox_librenms_id = get_librenms_device_id(netbox_interface, self.server_key, auto_save=False) if netbox_librenms_id is None: return mark_safe( diff --git a/netbox_librenms_plugin/templates/netbox_librenms_plugin/_cable_sync_content.html b/netbox_librenms_plugin/templates/netbox_librenms_plugin/_cable_sync_content.html index c42534a73d..485676360e 100644 --- a/netbox_librenms_plugin/templates/netbox_librenms_plugin/_cable_sync_content.html +++ b/netbox_librenms_plugin/templates/netbox_librenms_plugin/_cable_sync_content.html @@ -5,6 +5,7 @@ {% if cable_sync.table %}
{% csrf_token %} + {% if cable_sync.server_key %}{% endif %}
diff --git a/netbox_librenms_plugin/templates/netbox_librenms_plugin/_interface_sync_content.html b/netbox_librenms_plugin/templates/netbox_librenms_plugin/_interface_sync_content.html index 3d7e1406cd..acce51661b 100644 --- a/netbox_librenms_plugin/templates/netbox_librenms_plugin/_interface_sync_content.html +++ b/netbox_librenms_plugin/templates/netbox_librenms_plugin/_interface_sync_content.html @@ -9,6 +9,7 @@ action="{% url 'plugins:netbox_librenms_plugin:sync_selected_interfaces' object_type=model_name object_id=interface_sync.object.pk %}?interface_name_field={{ interface_name_field }}"> {% endwith %} {% csrf_token %} + {% if interface_sync.server_key %}{% endif %} {% block table_actions %}
diff --git a/netbox_librenms_plugin/templates/netbox_librenms_plugin/_ipaddress_sync_content.html b/netbox_librenms_plugin/templates/netbox_librenms_plugin/_ipaddress_sync_content.html index a31114de1f..ffc1bb83ef 100644 --- a/netbox_librenms_plugin/templates/netbox_librenms_plugin/_ipaddress_sync_content.html +++ b/netbox_librenms_plugin/templates/netbox_librenms_plugin/_ipaddress_sync_content.html @@ -7,6 +7,7 @@ {% endwith %} {% csrf_token %} + {% if ip_sync.server_key %}{% endif %}
diff --git a/netbox_librenms_plugin/templates/netbox_librenms_plugin/_vlan_sync_content.html b/netbox_librenms_plugin/templates/netbox_librenms_plugin/_vlan_sync_content.html index 017f888ec1..8221f29e38 100644 --- a/netbox_librenms_plugin/templates/netbox_librenms_plugin/_vlan_sync_content.html +++ b/netbox_librenms_plugin/templates/netbox_librenms_plugin/_vlan_sync_content.html @@ -21,6 +21,7 @@ action="{% url 'plugins:netbox_librenms_plugin:sync_selected_vlans' object_type=model_name object_id=vlan_sync.object.pk %}"> {% endwith %} {% csrf_token %} + {% if vlan_sync.server_key %}{% endif %}
diff --git a/netbox_librenms_plugin/templates/netbox_librenms_plugin/htmx/device_validation_details.html b/netbox_librenms_plugin/templates/netbox_librenms_plugin/htmx/device_validation_details.html index 8141c846d0..9296ec0506 100644 --- a/netbox_librenms_plugin/templates/netbox_librenms_plugin/htmx/device_validation_details.html +++ b/netbox_librenms_plugin/templates/netbox_librenms_plugin/htmx/device_validation_details.html @@ -376,7 +376,13 @@
{% if validation.existing_device %} {% if validation.existing_match_type == 'librenms_id' %}
- Linked — ID {{ libre_device.device_id }} + {% if existing_id_servers %} + {% for srv in existing_id_servers %} + Linked — ID {{ srv.device_id }} @ {{ srv.display_name }} + {% endfor %} + {% else %} + Linked — ID {{ libre_device.device_id }} + {% endif %} {% if validation.name_matches %} Name match {% elif validation.name_sync_available %} @@ -392,7 +398,38 @@
{% if validation.device_type_mismatch %} Type mismatch {% endif %} + {% if validation.librenms_id_needs_migration %} + Legacy ID format + {% endif %}
+ {% if validation.librenms_id_needs_migration %} +
+ + {% csrf_token %} + + + + + {% if validation.existing_device.cluster %} + + {% endif %} + {% if not validation.serial_confirmed %} +
+ + +
+ {% endif %} + + +
+ {% endif %} {% elif validation.existing_match_type == 'hostname' %}
diff --git a/netbox_librenms_plugin/templates/netbox_librenms_plugin/librenms_sync_base.html b/netbox_librenms_plugin/templates/netbox_librenms_plugin/librenms_sync_base.html index 82fd96ff03..a756c29a9f 100644 --- a/netbox_librenms_plugin/templates/netbox_librenms_plugin/librenms_sync_base.html +++ b/netbox_librenms_plugin/templates/netbox_librenms_plugin/librenms_sync_base.html @@ -29,7 +29,72 @@ {% block content %} -{% if librenms_server_info %} +{% if all_server_mappings %} +
+
+ LibreNMS Connections + {% if librenms_server_info and not librenms_server_info.is_legacy %} + + Change Server + + {% endif %} +
+
+ + + {% for mapping in all_server_mappings %} + + + + + + {% endfor %} + +
+ {% if mapping.is_active %} + + {% elif mapping.is_configured %} + + {% else %} + + {% endif %} + {% if mapping.is_configured %} + {{ mapping.display_name }} + {% else %} + {{ mapping.server_key }} + Not configured + {% endif %} + + {% if mapping.device_url %} + + ID {{ mapping.device_id }} + + + {% else %} + ID {{ mapping.device_id }} + {% endif %} + + {% if not mapping.is_configured %} + {% if lookup_device_model_name == "device" or lookup_device_model_name == "virtualmachine" %} +
+ {% csrf_token %} + + + +
+ {% endif %} + {% endif %} +
+
+
+{% elif librenms_server_info %}
@@ -88,7 +153,29 @@
ID - {{ librenms_device_id }} + + {{ librenms_device_id }} + {% if librenms_id_is_legacy %} + {% with model_name=object|meta:"model_name" %} +
+ {% csrf_token %} + + {% if librenms_id_serial_confirmed %} + + {% else %} + + {% endif %} +
+ {% endwith %} + {% endif %} + @@ -249,7 +336,7 @@
Device Information Sync
{{ object.name }}
- {% if sysName and sysName != object.name %} + {% if sysName and sysName != "-" and sysName != object.name %}
@@ -259,7 +346,7 @@
Device Information Sync
Sync to NetBox
- {% elif sysName %} + {% elif sysName and sysName != "-" %} @@ -574,7 +661,6 @@
Device Information Sync
{% endwith %}
- {% else %}
diff --git a/netbox_librenms_plugin/tests/conftest.py b/netbox_librenms_plugin/tests/conftest.py index c897c5ef56..a6bba487f7 100644 --- a/netbox_librenms_plugin/tests/conftest.py +++ b/netbox_librenms_plugin/tests/conftest.py @@ -49,12 +49,10 @@ def mock_librenms_api(mock_multi_server_config): """Pre-configured LibreNMSAPI instance with mocked dependencies.""" with patch("netbox_librenms_plugin.librenms_api.get_plugin_config") as mock_config: mock_config.return_value = mock_multi_server_config - with patch("netbox_librenms_plugin.librenms_api.LibreNMSSettings") as mock_settings: - mock_settings.objects.filter.return_value.first.return_value = None - from netbox_librenms_plugin.librenms_api import LibreNMSAPI + from netbox_librenms_plugin.librenms_api import LibreNMSAPI - api = LibreNMSAPI(server_key="default") - yield api + api = LibreNMSAPI(server_key="default") + yield api # ============================================================================= @@ -291,3 +289,48 @@ def mock_netbox_rack(): rack.name = "Rack A1" rack.site = MagicMock(id=1, name="DC1") return rack + + +# ============================================================================= +# Server Mapping Fixtures (used by test_sync_view_mismatch.py) +# ============================================================================= + + +@pytest.fixture +def mock_plugins_config_single_server(): + """PLUGINS_CONFIG with a single 'production' server (for _build_all_server_mappings tests).""" + return { + "netbox_librenms_plugin": { + "servers": { + "production": { + "display_name": "Production LibreNMS", + "librenms_url": "https://librenms.example.com", + }, + } + } + } + + +@pytest.fixture +def mock_plugins_config_empty_servers(): + """PLUGINS_CONFIG with no configured servers (simulates all orphaned).""" + return {"netbox_librenms_plugin": {"servers": {}}} + + +@pytest.fixture +def mock_plugins_config_multi_server_mapping(): + """PLUGINS_CONFIG with 'production' and 'mock-dev' servers (for multi-server mapping tests).""" + return { + "netbox_librenms_plugin": { + "servers": { + "production": { + "display_name": "Production LibreNMS", + "librenms_url": "https://librenms.example.com", + }, + "mock-dev": { + "display_name": "Mock", + "librenms_url": "http://mock.example.com", + }, + } + } + } diff --git a/netbox_librenms_plugin/tests/test_cable_verify.py b/netbox_librenms_plugin/tests/test_cable_verify.py new file mode 100644 index 0000000000..be4bb0d603 --- /dev/null +++ b/netbox_librenms_plugin/tests/test_cable_verify.py @@ -0,0 +1,261 @@ +"""Regression tests for SingleCableVerifyView.post(). + +Covers: +- Stale derived fields are stripped before re-enrichment (prevents + DoesNotExist when remote objects are deleted after caching). +- LibreNMS-sourced labels are HTML-escaped to prevent XSS. +""" + +import json +from unittest.mock import MagicMock, patch + + +def _make_view(server_key="default"): + """Create a SingleCableVerifyView instance without database access.""" + from netbox_librenms_plugin.views.base.cables_view import SingleCableVerifyView + + view = object.__new__(SingleCableVerifyView) + view._librenms_api = MagicMock() + view._librenms_api.server_key = server_key + view.request = MagicMock() + return view + + +def _make_request(body_dict): + """Create a mock POST request with JSON body.""" + request = MagicMock() + request.method = "POST" + request.body = json.dumps(body_dict).encode() + request.META = {"HTTP_X_REQUESTED_WITH": "XMLHttpRequest"} + return request + + +class TestStaleFieldStripping: + """Cached link data with stale derived fields must be stripped before use.""" + + def test_stale_remote_fields_stripped_before_enrichment(self): + """Stale netbox_remote_device_id / remote_device_url must not reach check_cable_status().""" + view = _make_view() + + # Cached link with stale derived fields (from a previous enrichment) + cached_link = { + "local_port": "eth0", + "local_port_id": 100, + "remote_port": "eth1", + "remote_device": "switch-remote", + "remote_port_id": 200, + "remote_device_id": 42, + # Stale derived fields — remote device was deleted after caching + "netbox_remote_device_id": 999, + "remote_device_url": "/dcim/devices/999/", + "netbox_remote_interface_id": 888, + "remote_port_url": "/dcim/interfaces/888/", + "cable_status": "No Cable", + "can_create_cable": True, + } + + cached_data = {"links": [cached_link]} + + device = MagicMock() + device.pk = 1 + device.id = 1 + device.virtual_chassis = None + interface_mock = MagicMock() + interface_mock.pk = 10 + + # Track what link_data check_cable_status receives + received_link_data = {} + + def fake_check_cable_status(link): + received_link_data.update(link) + link["cable_status"] = "No Cable" + link["can_create_cable"] = True + return link + + def fake_process_remote_device(link, hostname, device_id): + # Simulate successful remote enrichment with fresh IDs + link["remote_device_url"] = "/dcim/devices/777/" + link["netbox_remote_device_id"] = 777 + link["remote_port_url"] = "/dcim/interfaces/666/" + link["netbox_remote_interface_id"] = 666 + link["remote_port_name"] = "eth1" + return link + + request = _make_request({"device_id": 1, "local_port_id": 100}) + + with ( + patch("netbox_librenms_plugin.views.base.cables_view.get_object_or_404", return_value=device), + patch("netbox_librenms_plugin.views.base.cables_view.cache") as mock_cache, + patch.object(view, "get_cache_key", return_value="test_key"), + patch.object(view, "check_cable_status", side_effect=fake_check_cable_status), + patch.object(view, "process_remote_device", side_effect=fake_process_remote_device), + patch("netbox_librenms_plugin.views.base.cables_view.get_librenms_sync_device", return_value=device), + patch("netbox_librenms_plugin.views.base.cables_view.get_virtual_chassis_member", return_value=device), + patch("netbox_librenms_plugin.views.base.cables_view._librenms_id_q", return_value=MagicMock()), + patch("netbox_librenms_plugin.views.base.cables_view.get_token", return_value="csrf123"), + patch("netbox_librenms_plugin.views.base.cables_view.reverse", return_value="/fake/"), + ): + mock_cache.get.return_value = cached_data + # Make the interface filter return our mock + device.interfaces.filter.return_value.first.return_value = interface_mock + + view.post(request) + + # check_cable_status should have received fresh IDs from process_remote_device, + # NOT the stale 999/888 from cache + assert received_link_data.get("netbox_remote_device_id") == 777 + assert received_link_data.get("netbox_remote_interface_id") == 666 + + def test_raw_keys_match_prepare_context(self): + """The _raw_keys set in post() must match the one in _prepare_context().""" + import inspect + + from netbox_librenms_plugin.views.base.cables_view import BaseCableTableView, SingleCableVerifyView + + # Extract _raw_keys from _prepare_context source + prepare_src = inspect.getsource(BaseCableTableView._prepare_context) + post_src = inspect.getsource(SingleCableVerifyView.post) + + # Both should contain the same set of raw keys + expected_keys = { + "local_port", + "local_port_id", + "remote_port", + "remote_device", + "remote_port_id", + "remote_device_id", + } + for key in expected_keys: + assert f'"{key}"' in prepare_src, f"{key} missing from _prepare_context _raw_keys" + assert f'"{key}"' in post_src, f"{key} missing from post() _raw_keys" + + +class TestXSSEscaping: + """LibreNMS-sourced labels must be HTML-escaped in cable verify output.""" + + def test_xss_in_local_port_name_escaped(self): + """A malicious local_port name must be escaped in the HTML output.""" + view = _make_view() + + xss_port_name = '' + cached_link = { + "local_port": xss_port_name, + "local_port_id": 100, + "remote_port": "eth1", + "remote_device": "safe-switch", + "remote_port_id": 200, + "remote_device_id": 42, + } + + cached_data = {"links": [cached_link]} + + device = MagicMock() + device.pk = 1 + device.id = 1 + device.virtual_chassis = None + interface_mock = MagicMock() + interface_mock.pk = 10 + + def fake_process_remote_device(link, hostname, device_id): + link["remote_device_url"] = "/dcim/devices/2/" + link["netbox_remote_device_id"] = 2 + link["remote_port_url"] = "/dcim/interfaces/20/" + link["netbox_remote_interface_id"] = 20 + link["remote_port_name"] = "eth1" + return link + + def fake_check_cable_status(link): + link["cable_status"] = "No Cable" + link["can_create_cable"] = False + return link + + request = _make_request({"device_id": 1, "local_port_id": 100}) + + with ( + patch("netbox_librenms_plugin.views.base.cables_view.get_object_or_404", return_value=device), + patch("netbox_librenms_plugin.views.base.cables_view.cache") as mock_cache, + patch.object(view, "get_cache_key", return_value="test_key"), + patch.object(view, "check_cable_status", side_effect=fake_check_cable_status), + patch.object(view, "process_remote_device", side_effect=fake_process_remote_device), + patch("netbox_librenms_plugin.views.base.cables_view.get_librenms_sync_device", return_value=device), + patch("netbox_librenms_plugin.views.base.cables_view._librenms_id_q", return_value=MagicMock()), + patch("netbox_librenms_plugin.views.base.cables_view.get_token", return_value="csrf123"), + patch("netbox_librenms_plugin.views.base.cables_view.reverse", return_value="/fake/"), + ): + mock_cache.get.return_value = cached_data + device.interfaces.filter.return_value.first.return_value = interface_mock + + response = view.post(request) + + content = json.loads(response.content) + row = content.get("formatted_row", {}) + local_port_html = row.get("local_port", "") + + # The raw script tag must NOT appear unescaped + assert "