diff --git a/docs/development/testing.md b/docs/development/testing.md index 42565cb76e..ca861ea502 100644 --- a/docs/development/testing.md +++ b/docs/development/testing.md @@ -38,8 +38,24 @@ The test suite covers all major plugin functionality. Tests are organized by the | [test_coverage_device_fields.py](../../netbox_librenms_plugin/tests/test_coverage_device_fields.py) | Device field sync view—field update logic and device field mapping | | [test_coverage_list.py](../../netbox_librenms_plugin/tests/test_coverage_list.py) | Import list view—background job decision, job result loading, and GET handler | | [test_coverage_api.py](../../netbox_librenms_plugin/tests/test_coverage_api.py) | LibreNMS API client—malformed payload guards, error paths, and edge cases | +| [test_coverage_api2.py](../../netbox_librenms_plugin/tests/test_coverage_api2.py) | API views—device status, background job management, VM status endpoints | +| [test_coverage_base_views.py](../../netbox_librenms_plugin/tests/test_coverage_base_views.py) | Base view coverage tests—sync table views, context data, and data pipeline | +| [test_coverage_base_views2.py](../../netbox_librenms_plugin/tests/test_coverage_base_views2.py) | Additional base view coverage—IP address sync, cable matching, edge cases | +| [test_coverage_cache.py](../../netbox_librenms_plugin/tests/test_coverage_cache.py) | Import cache helpers—cache key generation, active search tracking, metadata | +| [test_coverage_device_operations.py](../../netbox_librenms_plugin/tests/test_coverage_device_operations.py) | Device validation—type matching, serial handling, VC detection, role lookup | +| [test_coverage_forms.py](../../netbox_librenms_plugin/tests/test_coverage_forms.py) | Import forms—filter form choices, background-job option guards, field validation | +| [test_coverage_mixins.py](../../netbox_librenms_plugin/tests/test_coverage_mixins.py) | View mixins—VLAN group scope resolution, VlanAssignmentMixin, scope priority | +| [test_coverage_sync_interfaces.py](../../netbox_librenms_plugin/tests/test_coverage_sync_interfaces.py) | Interface sync view—port caching, attribute updates, MAC handling, VC member routing | | [test_coverage_sync_view.py](../../netbox_librenms_plugin/tests/test_coverage_sync_view.py) | Sync view base class—context preparation and tab rendering | +| [test_coverage_sync_views.py](../../netbox_librenms_plugin/tests/test_coverage_sync_views.py) | Sync action views—cables, IP addresses, VLAN sync action handlers | +| [test_coverage_sync_views2.py](../../netbox_librenms_plugin/tests/test_coverage_sync_views2.py) | Additional sync action view coverage—device fields, device name/type sync | +| [test_coverage_sync_views3.py](../../netbox_librenms_plugin/tests/test_coverage_sync_views3.py) | Further sync action view coverage—location sync, VLAN assignment edge cases | +| [test_coverage_actions.py](../../netbox_librenms_plugin/tests/test_coverage_actions.py) | Import action views—bulk import, device role/cluster/rack update, validation details | | [test_coverage_filters.py](../../netbox_librenms_plugin/tests/test_coverage_filters.py) | Import filter logic—filter form processing and device count helpers | +| [test_coverage_tables.py](../../netbox_librenms_plugin/tests/test_coverage_tables.py) | Sync tables—column rendering, row data, interface and cable table helpers | +| [test_coverage_utils.py](../../netbox_librenms_plugin/tests/test_coverage_utils.py) | Utility function coverage—name matching, speed conversion, site/platform lookup | +| [test_coverage_virtual_chassis.py](../../netbox_librenms_plugin/tests/test_coverage_virtual_chassis.py) | Virtual chassis coverage—VC creation, position conflict handling, member naming | +| [test_coverage_vlans_table.py](../../netbox_librenms_plugin/tests/test_coverage_vlans_table.py) | VLAN sync table—column rendering, group assignment, VLAN comparison rows | | [test_sync_modules.py](../../netbox_librenms_plugin/tests/test_sync_modules.py) | Module sync—inventory matching, module type resolution, and normalization rules | | [test_modules_view.py](../../netbox_librenms_plugin/tests/test_modules_view.py) | Module sync view—context preparation, table rendering, and module bay mapping | | [test_tables_modules.py](../../netbox_librenms_plugin/tests/test_tables_modules.py) | Module tables—column rendering, row formatting, and action buttons | @@ -76,7 +92,7 @@ pytest netbox_librenms_plugin/tests/test_librenms_api.py::TestLibreNMSAPIConnect ```bash # API client tests -pytest netbox_librenms_plugin/tests/test_librenms_api.py -v +pytest netbox_librenms_plugin/tests/test_librenms_api.py netbox_librenms_plugin/tests/test_coverage_api.py netbox_librenms_plugin/tests/test_coverage_api2.py -v # Import and validation tests pytest netbox_librenms_plugin/tests/test_import_utils.py netbox_librenms_plugin/tests/test_import_validation_helpers.py netbox_librenms_plugin/tests/test_utils.py -v diff --git a/docs/usage_tips/custom_field.md b/docs/usage_tips/custom_field.md index 038c1773bb..1474d6826e 100644 --- a/docs/usage_tips/custom_field.md +++ b/docs/usage_tips/custom_field.md @@ -4,6 +4,9 @@ To enhance device identification and synchronization between NetBox and LibreNMS, this plugin supports using a custom field `librenms_id` on Device, Virtual Machine and Interface objects. While the plugin works without it, using this custom field is recommended for LibreNMS API lookups, and to assist with matching the remote device and remote interfaces for cable creation in Netbox. It can also be entered manually if no primary IP or FQDN is available. +!!! info "Automatic Creation" + As of version 0.4.2, the plugin **automatically creates** the `librenms_id` custom field when migrations are run. You no longer need to create it manually. The field is created for Device, Virtual Machine, Interface, and VM Interface objects. + For the Device and Virtual Machine objects the plugin will automatically populate the LibreNMS ID custom field when opening the LibreNMS Sync page if the device has been found in LibreNMS. For the Interface object, the plugin will automatically populate the LibreNMS ID custom field when the interface data is synced from LibreNMS. @@ -15,7 +18,10 @@ For the Interface object, the plugin will automatically populate the LibreNMS ID - **Efficient Synchronization:** Enhances the reliability of API lookups. - **Cable creation:** Allows better device identification for the creation of cables between NetBox devices. -## Suggested Custom Field Setup +## Manual Custom Field Setup (Legacy) + +!!! note + This section is only needed if you are running an older version of the plugin that does not auto-create the field, or if you need to recreate it after deletion. Follow these steps to create the `librenms_id` custom field in NetBox: diff --git a/netbox_librenms_plugin/__init__.py b/netbox_librenms_plugin/__init__.py index 96b9997506..299ceadd3e 100644 --- a/netbox_librenms_plugin/__init__.py +++ b/netbox_librenms_plugin/__init__.py @@ -28,6 +28,7 @@ def ready(self): super().ready() from django.conf import settings + from django.db.models.signals import post_migrate plugin_config = getattr(settings, "PLUGINS_CONFIG", {}).get(self.name, {}) @@ -37,6 +38,12 @@ def ready(self): else: self._validate_legacy_config(plugin_config) + # Auto-create the librenms_id custom field after migrations complete + post_migrate.connect( + _ensure_librenms_id_custom_field, + dispatch_uid="netbox_librenms_plugin_ensure_cf", + ) + def _validate_multi_server_config(self, servers_config): """Validate multi-server configuration.""" if not servers_config or not isinstance(servers_config, dict): @@ -61,4 +68,63 @@ def _validate_legacy_config(self, plugin_config): ) +def _ensure_librenms_id_custom_field(sender, **kwargs): + """ + Auto-create the 'librenms_id' custom field if it doesn't exist. + Runs after migrations via post_migrate signal to ensure tables exist. + Uses dispatch_uid to avoid duplicate connections. + """ + # Only run once per migrate invocation (post_migrate fires per-app). + # The _executed flag is intentionally never reset: migrations are expected to + # run in short-lived CLI processes (manage.py migrate) where the flag is + # naturally cleared on exit. Long-running processes (e.g. gunicorn workers) + # should not rely on this handler re-executing after startup. + if getattr(_ensure_librenms_id_custom_field, "_executed", False): + return + _ensure_librenms_id_custom_field._executed = True # not reset; see comment above + + try: + from django.contrib.contenttypes.models import ContentType + + from extras.models import CustomField + + cf, created = CustomField.objects.get_or_create( + name="librenms_id", + defaults={ + "type": "json", + "label": "LibreNMS ID", + "description": "LibreNMS Device ID for synchronization (auto-created by plugin)", + "required": False, + "ui_visible": "if-set", + "ui_editable": "yes", + "is_cloneable": False, + }, + ) + + # Ensure the field is assigned to the required object types + from dcim.models import Device, Interface + from virtualization.models import VirtualMachine, VMInterface + + required_models = [Device, VirtualMachine, Interface, VMInterface] + current_types = set(cf.object_types.values_list("pk", flat=True)) + + for model in required_models: + ct = ContentType.objects.get_for_model(model) + if ct.pk not in current_types: + cf.object_types.add(ct) + + if created: + import logging + + logging.getLogger("netbox_librenms_plugin").info( + "Auto-created 'librenms_id' custom field for Device, VirtualMachine, Interface, VMInterface" + ) + except Exception as e: + # Don't break startup if custom field creation fails (e.g., during initial migration), + # but log the error so it's not silently swallowed. + import logging + + logging.getLogger("netbox_librenms_plugin").exception("Failed to auto-create 'librenms_id' custom field: %s", e) + + config = LibreNMSSyncConfig diff --git a/netbox_librenms_plugin/import_utils/bulk_import.py b/netbox_librenms_plugin/import_utils/bulk_import.py index abc8754190..a01bc7562f 100644 --- a/netbox_librenms_plugin/import_utils/bulk_import.py +++ b/netbox_librenms_plugin/import_utils/bulk_import.py @@ -99,13 +99,13 @@ def bulk_import_devices_shared( if user is None and job is not None: user = getattr(job.job, "user", None) - # Check permissions at start of bulk operation — both device and VM perms are + # Check permissions at start of bulk operation — device and VM add perms are # required because any device may be flagged as import_as_vm during validation. + # change_device is needed for VC master/member updates; VMs are only created, not changed. required_perms = [ "dcim.add_device", "dcim.change_device", "virtualization.add_virtualmachine", - "virtualization.change_virtualmachine", ] require_permissions(user, required_perms, "import devices") @@ -239,51 +239,50 @@ def bulk_import_devices_shared( for m in vc_data.get("members", []) ) if member_parts: - fingerprint = hashlib.md5((f"{device_id}," + ",".join(member_parts)).encode()).hexdigest()[ - :12 - ] + fingerprint = hashlib.md5(",".join(member_parts).encode()).hexdigest()[:12] vc_domain = f"librenms-stack-{fingerprint}" else: vc_domain = f"librenms-{device_id}" - # Guard VC creation with its own permission check — the upfront check - # only covers add_device/change_device; VirtualChassis needs a separate perm. - has_vc_perm, missing_vc_perms = check_user_permissions(user, ["dcim.add_virtualchassis"]) - if not has_vc_perm: - warn_msg = ( - f"Skipping VC creation for device {device_id}: " - f"missing permissions: {', '.join(missing_vc_perms)}" - ) - if job and job.logger: - job.logger.warning(warn_msg) - else: - logger.warning(warn_msg) # Only create VC if we haven't processed this stack yet - # Add to set BEFORE attempting creation to prevent race condition - elif vc_domain not in processed_vc_domains: - processed_vc_domains.add(vc_domain) - try: - vc = create_virtual_chassis_with_members( - result["device"], - vc_data["members"], - libre_device, - server_key=api.server_key, + if vc_domain not in processed_vc_domains: + # Guard VC creation with its own permission check — the upfront check + # only covers add_device/change_device; VirtualChassis needs a separate perm. + has_vc_perm, missing_vc_perms = check_user_permissions(user, ["dcim.add_virtualchassis"]) + if not has_vc_perm: + warn_msg = ( + f"Skipping VC creation for device {device_id}: " + f"missing permissions: {', '.join(missing_vc_perms)}" ) - vc_created_count += 1 - log_msg = f"Created VC '{vc.name}' during bulk import for device {device_id}" - if job and job.logger: - job.logger.info(log_msg) - else: - logger.info(log_msg) - except Exception as vc_error: - # Remove from set on failure so retry is possible - processed_vc_domains.discard(vc_domain) - warn_msg = f"Failed to create VC for device {device_id}: {vc_error}" if job and job.logger: job.logger.warning(warn_msg) else: logger.warning(warn_msg) - # Don't fail the import, just log the warning + else: + # Add to set BEFORE attempting creation to prevent race condition + processed_vc_domains.add(vc_domain) + try: + vc = create_virtual_chassis_with_members( + result["device"], + vc_data["members"], + libre_device, + server_key=api.server_key, + ) + vc_created_count += 1 + log_msg = f"Created VC '{vc.name}' during bulk import for device {device_id}" + if job and job.logger: + job.logger.info(log_msg) + else: + logger.info(log_msg) + except Exception as vc_error: + # Remove from set on failure so retry is possible + processed_vc_domains.discard(vc_domain) + warn_msg = f"Failed to create VC for device {device_id}: {vc_error}" + if job and job.logger: + job.logger.warning(warn_msg) + else: + logger.warning(warn_msg) + # Don't fail the import, just log the warning elif result.get("device"): # Device exists skipped_list.append({"device_id": device_id, "reason": result["error"]}) @@ -386,8 +385,11 @@ def _refresh_existing_device(validation: dict, libre_device: dict = None, server elif not validation.get("import_as_vm"): validation["device_role"] = {"found": False, "role": None} remove_validation_issue(validation, "role") - validation.setdefault("issues", []).append("Device role must be manually selected before import") recalculate_validation_status(validation, is_vm=bool(validation.get("import_as_vm"))) + # Re-assert non-importable state: recalculate bases can_import on + # issues alone, but an existing matched device must never be import-ready. + validation["can_import"] = False + validation["is_ready"] = False return else: # Device was deleted since caching — recompute readiness to match @@ -403,7 +405,7 @@ def _refresh_existing_device(validation: dict, libre_device: dict = None, server except Exception as e: existing_id = getattr(existing, "pk", "unknown") if existing else "none" logger.error(f"Failed to refresh existing device (pk={existing_id}): {e}") - return + return # existing_device was None at cache time — check if device was imported since if not libre_device: @@ -468,9 +470,10 @@ def _lookup_in_model(m): actual_is_vm = found_as_cross_model != import_as_vm # XOR: cross flips the flag validation["import_as_vm"] = actual_is_vm # Update so future refreshes query correct model if not actual_is_vm and hasattr(new_device, "role") and new_device.role: - validation["device_role"] = {"found": True, "role": new_device.role} + apply_role_to_validation(validation, new_device.role, is_vm=False) elif not actual_is_vm: - validation.setdefault("device_role", {}).update({"found": False, "role": None}) + validation["device_role"] = {"found": False, "role": None} + recalculate_validation_status(validation, is_vm=actual_is_vm) except Exception as e: logger.error(f"Failed to check for newly imported device: {e}") diff --git a/netbox_librenms_plugin/import_utils/cache.py b/netbox_librenms_plugin/import_utils/cache.py index 8de881f897..b6860a3fea 100644 --- a/netbox_librenms_plugin/import_utils/cache.py +++ b/netbox_librenms_plugin/import_utils/cache.py @@ -9,6 +9,19 @@ logger = logging.getLogger(__name__) +def _build_filter_hash(filters: dict) -> str: + """ + Build a stable, collision-free hash from a filter dict. + + Removes None values (preserves valid falsy values like 0 and False), + sorts by key, and returns the first 16 hex characters of the SHA-256 + digest of the JSON-serialized result. + """ + return hashlib.sha256( + json.dumps({k: v for k, v in filters.items() if v is not None}, sort_keys=True, separators=(",", ":")).encode() + ).hexdigest()[:16] + + 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}" @@ -34,11 +47,7 @@ def get_cache_metadata_key( # valid falsy values like 0 and False (filtering only None/missing entries). # Use JSON serialization for a stable, collision-free hash (avoids issues with # values containing "=" or "_" that could collide with the key separators). - filter_hash = hashlib.sha256( - json.dumps( - {k: v for k, v in sorted(filters.items()) if v is not None}, sort_keys=True, separators=(",", ":") - ).encode() - ).hexdigest()[:16] + filter_hash = _build_filter_hash(filters) return f"librenms_filter_cache_metadata_{server_key}_{filter_hash}_{vc_enabled}_sysname={use_sysname}_strip={strip_domain}" @@ -94,9 +103,12 @@ def get_active_cached_searches(server_key: str) -> list[dict]: now = datetime.now(timezone.utc) try: cached_at_raw = metadata.get("cached_at") - cached_at = ( - datetime.fromisoformat(cached_at_raw) if cached_at_raw else datetime.fromtimestamp(0, timezone.utc) - ) + if isinstance(cached_at_raw, datetime): + cached_at = cached_at_raw + elif cached_at_raw: + cached_at = datetime.fromisoformat(cached_at_raw) + else: + cached_at = datetime.fromtimestamp(0, timezone.utc) # Normalize naive datetimes (e.g., stored without tzinfo) to UTC if cached_at.tzinfo is None: cached_at = cached_at.replace(tzinfo=timezone.utc) @@ -109,6 +121,8 @@ def get_active_cached_searches(server_key: str) -> list[dict]: # Add remaining time and cache key metadata["remaining_seconds"] = int(remaining_seconds) metadata["cache_key"] = cache_key + # Store numeric sort key so the final sort is unambiguous + metadata["cached_at_ts"] = cached_at.timestamp() # Enrich filters with human-readable display values if "filters" in metadata: @@ -132,7 +146,7 @@ def get_active_cached_searches(server_key: str) -> list[dict]: cache.set(cache_index_key, valid_cache_keys, timeout=3600) # Sort by most recent first - active_searches.sort(key=lambda x: x.get("cached_at", ""), reverse=True) + active_searches.sort(key=lambda x: x.get("cached_at_ts", 0.0), reverse=True) return active_searches @@ -167,15 +181,16 @@ def get_validated_device_cache_key( >>> key 'validated_device_default_e3b0c44298fc1c14_123_vc' """ - # Sort filters for a deterministic, cross-process stable hash - filter_hash = hashlib.sha256(json.dumps(sorted(filters.items()), sort_keys=True).encode()).hexdigest()[:16] + # Sort filters for a deterministic, cross-process stable hash; None values are excluded + # (consistent with get_cache_metadata_key). + filter_hash = _build_filter_hash(filters) vc_part = "vc" if vc_enabled else "novc" return ( f"validated_device_{server_key}_{filter_hash}_{device_id}_{vc_part}_sysname={use_sysname}_strip={strip_domain}" ) -def get_import_device_cache_key(device_id: int | str, server_key: str) -> str: +def get_import_device_cache_key(device_id: int | str, server_key: str = "default") -> str: """ Generate cache key for raw LibreNMS device data. @@ -185,7 +200,7 @@ def get_import_device_cache_key(device_id: int | str, server_key: str) -> str: Args: device_id: LibreNMS device ID - server_key: LibreNMS server identifier for multi-server setups (required) + server_key: LibreNMS server identifier for multi-server setups. Defaults to "default" for backward compatibility. Returns: str: Cache key for the device data @@ -212,12 +227,6 @@ def get_import_search_cache_key(server_key: str, api_filters: dict, client_filte Returns: str: Cache key for the import search result. """ - import hashlib - import json - - def _hash(d): - return hashlib.sha256( - json.dumps(sorted(d.items()) if isinstance(d, dict) else d, sort_keys=True).encode() - ).hexdigest()[:16] - - return f"librenms_devices_import_{server_key}_{_hash(api_filters)}_{_hash(client_filters)}" + return ( + f"librenms_devices_import_{server_key}_{_build_filter_hash(api_filters)}_{_build_filter_hash(client_filters)}" + ) diff --git a/netbox_librenms_plugin/import_utils/device_operations.py b/netbox_librenms_plugin/import_utils/device_operations.py index 23a7db9b8a..2b1e5c416e 100644 --- a/netbox_librenms_plugin/import_utils/device_operations.py +++ b/netbox_librenms_plugin/import_utils/device_operations.py @@ -1,6 +1,7 @@ """Device validation, import, and fetch operations.""" import logging +from types import SimpleNamespace from dcim.models import Device, DeviceRole, DeviceType, Rack, Site from django.core.cache import cache @@ -11,6 +12,7 @@ from ..librenms_api import LibreNMSAPI from ..utils import ( + find_by_librenms_id, find_matching_platform, find_matching_site, match_librenms_hardware_to_device_type, @@ -128,11 +130,11 @@ def validate_device_for_import( import_as_vm: bool = False, api: "LibreNMSAPI" = None, *, + server_key: str = "default", include_vc_detection: bool = True, 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. @@ -282,14 +284,10 @@ def validate_device_for_import( server_key = api.server_key if api is not None else server_key - # Check for existing VM first (by librenms_id custom field) - try: - from netbox_librenms_plugin.utils import find_by_librenms_id - - existing_vm = find_by_librenms_id(VirtualMachine, librenms_id, server_key) - except (ValueError, TypeError): - # librenms_id is not convertible to int; no match will be found - existing_vm = None + # Check for existing VM first (by librenms_id custom field). + # find_by_librenms_id() covers both the new per-server JSON format + # and legacy bare-integer values so neither is missed. + existing_vm = find_by_librenms_id(VirtualMachine, librenms_id, server_key) if existing_vm: logger.info(f"Found existing VM: {existing_vm.name} (matched by librenms_id={librenms_id})") @@ -315,15 +313,11 @@ def validate_device_for_import( result["name_sync_available"] = True result["suggested_name"] = hostname - # Check for existing Device (by librenms_id custom field) + # Check for existing Device (by librenms_id custom field). + # find_by_librenms_id() covers both the new per-server JSON format + # and legacy bare-integer values so neither is missed. if not result["existing_device"]: - try: - from netbox_librenms_plugin.utils import find_by_librenms_id - - existing_device = find_by_librenms_id(Device, librenms_id, server_key) - except (ValueError, TypeError): - # librenms_id is not convertible to int; no match will be found - existing_device = None + existing_device = find_by_librenms_id(Device, librenms_id, server_key) if existing_device: logger.info(f"Found existing device: {existing_device.name} (matched by librenms_id={librenms_id})") @@ -849,6 +843,8 @@ def import_single_device( # Generate import timestamp comment import_time = timezone.now().strftime("%Y-%m-%d %H:%M:%S %Z") + _cf_proxy = SimpleNamespace(custom_field_data={}) + set_librenms_device_id(_cf_proxy, device_id, api.server_key) device_data = { "name": device_name, "site": site, @@ -856,6 +852,7 @@ 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": _cf_proxy.custom_field_data, } # Add optional fields @@ -880,7 +877,6 @@ 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/virtual_chassis.py b/netbox_librenms_plugin/import_utils/virtual_chassis.py index f7a1a58b78..b4ef6f0676 100644 --- a/netbox_librenms_plugin/import_utils/virtual_chassis.py +++ b/netbox_librenms_plugin/import_utils/virtual_chassis.py @@ -238,7 +238,7 @@ def detect_virtual_chassis_from_inventory(api: LibreNMSAPI, device_id: int) -> d # position is already 1-based, so pass it directly (no +1). if master_name: member_data["suggested_name"] = _generate_vc_member_name( - master_name, position, serial=member_data.get("serial"), pattern=vc_name_pattern + master_name, position, serial=_norm_serial(member_data.get("serial")), pattern=vc_name_pattern ) else: member_data["suggested_name"] = f"Member-{position}" @@ -344,7 +344,7 @@ def update_vc_member_suggested_names(vc_data: dict, master_name: str) -> dict: position = idx + 1 member["position"] = position member["suggested_name"] = _generate_vc_member_name( - master_name, position, serial=member.get("serial"), pattern=vc_pattern + master_name, position, serial=_norm_serial(member.get("serial")), pattern=vc_pattern ) return vc_data @@ -393,8 +393,11 @@ def create_virtual_chassis_with_members( ] """ - # original_master_name is still referenced in warning messages inside the atomic block. + # Save originals for in-memory rollback — transaction.atomic() rolls back DB but + # not in-memory model fields. original_master_name = master_device.name + original_vc = master_device.virtual_chassis + original_vc_position = master_device.vc_position # Find master's actual VC position from members_info by serial match; default to 1 _master_pos = 1 @@ -412,7 +415,7 @@ def create_virtual_chassis_with_members( vc_pattern = _load_vc_member_name_pattern() # Rename master device to include position 1 pattern master_device_new_name = _generate_vc_member_name( - original_master_name, _master_pos, serial=master_device.serial, pattern=vc_pattern + original_master_name, _master_pos, serial=_norm_serial(master_device.serial), pattern=vc_pattern ) # Check if renamed master conflicts with existing device @@ -456,7 +459,7 @@ def create_virtual_chassis_with_members( member_pos = _safe_pos(member.get("position")) # Skip if this is the master's serial (only when both serials are non-empty) - if serial and serial == (master_device.serial or "").strip(): + if serial and serial == _norm_serial(master_device.serial): continue # Skip blank-serial entries that represent the master slot by position if ( @@ -524,7 +527,10 @@ def create_virtual_chassis_with_members( [ m for m in members_info - if not (_norm_serial(m.get("serial")) and _norm_serial(m.get("serial")) == master_device.serial) + if not ( + _norm_serial(m.get("serial")) + and _norm_serial(m.get("serial")) == _norm_serial(master_device.serial) + ) and not ( not _norm_serial(m.get("serial")) and m.get("position") is not None @@ -547,10 +553,11 @@ def create_virtual_chassis_with_members( return vc except Exception as e: - # The transaction.atomic() block above will roll back all DB changes automatically. - # Manual state restoration is redundant and the save() would fail in a broken transaction. + master_device.name = original_master_name + master_device.virtual_chassis = original_vc + master_device.vc_position = original_vc_position logger.error( - f"Virtual Chassis creation failed for device {master_device.name}: {e}", + f"Virtual Chassis creation failed for device {original_master_name}: {e}", exc_info=True, ) raise diff --git a/netbox_librenms_plugin/import_utils/vm_operations.py b/netbox_librenms_plugin/import_utils/vm_operations.py index 0190c069ba..e1c43cc434 100644 --- a/netbox_librenms_plugin/import_utils/vm_operations.py +++ b/netbox_librenms_plugin/import_utils/vm_operations.py @@ -17,9 +17,10 @@ def create_vm_from_librenms( libre_device: dict, validation: dict, - server_key: str, + server_key: str = "default", use_sysname: bool = True, strip_domain: bool = False, + role=None, ): """ Create a NetBox VirtualMachine from LibreNMS device data. @@ -44,7 +45,7 @@ def create_vm_from_librenms( # Extract matched objects from validation cluster = validation["cluster"]["cluster"] platform = validation["platform"].get("platform") - role = validation.get("device_role", {}).get("role") + role = role if role is not None else validation.get("device_role", {}).get("role") # Determine VM name - use pre-computed name if available (handles strip_domain), # falling back to the validated resolved_name before recomputing from raw fields. diff --git a/netbox_librenms_plugin/librenms_api.py b/netbox_librenms_plugin/librenms_api.py index 0cceab18b7..c4cefea76c 100644 --- a/netbox_librenms_plugin/librenms_api.py +++ b/netbox_librenms_plugin/librenms_api.py @@ -209,42 +209,42 @@ def get_librenms_id(self, obj): # Try IP address if ip_address: - librenms_id = self.get_device_id_by_ip(ip_address) + librenms_id = self._normalize_librenms_id(self.get_device_id_by_ip(ip_address)) if librenms_id is not None: - try: - librenms_id = int(librenms_id) - except (ValueError, TypeError): - librenms_id = None - if librenms_id is not None: - self._store_librenms_id(obj, librenms_id) - return librenms_id + self._store_librenms_id(obj, librenms_id) + return librenms_id # Try primary IP's DNS name if dns_name: - librenms_id = self.get_device_id_by_hostname(dns_name) + librenms_id = self._normalize_librenms_id(self.get_device_id_by_hostname(dns_name)) if librenms_id is not None: - try: - librenms_id = int(librenms_id) - except (ValueError, TypeError): - librenms_id = None - if librenms_id is not None: - self._store_librenms_id(obj, librenms_id) - return librenms_id + self._store_librenms_id(obj, librenms_id) + return librenms_id # Try hostname if FQDN if hostname: - librenms_id = self.get_device_id_by_hostname(hostname) + librenms_id = self._normalize_librenms_id(self.get_device_id_by_hostname(hostname)) if librenms_id is not None: - try: - librenms_id = int(librenms_id) - except (ValueError, TypeError): - librenms_id = None - if librenms_id is not None: - self._store_librenms_id(obj, librenms_id) - return librenms_id + self._store_librenms_id(obj, librenms_id) + return librenms_id return None + @staticmethod + def _normalize_librenms_id(value): + """Coerce a raw LibreNMS ID value to int or None. + + Treats booleans as None (LibreNMS occasionally returns True/False for + missing devices) and converts any other value to int, returning None on + failure. + """ + if value is None or isinstance(value, bool): + return None + try: + return int(value) + except (ValueError, TypeError): + return None + def _get_cache_key(self, obj): """ Generate a unique cache key for an object. @@ -300,7 +300,7 @@ def get_device_id_by_ip(self, ip_address): response.raise_for_status() device_data = response.json()["devices"][0] return device_data["device_id"] - except (requests.exceptions.RequestException, IndexError, KeyError, TypeError): + except (requests.exceptions.RequestException, ValueError, IndexError, KeyError, TypeError): return None def get_device_id_by_hostname(self, hostname): @@ -323,7 +323,7 @@ def get_device_id_by_hostname(self, hostname): response.raise_for_status() device_data = response.json()["devices"][0] return device_data["device_id"] - except (requests.exceptions.RequestException, IndexError, KeyError, TypeError): + except (requests.exceptions.RequestException, ValueError, IndexError, KeyError, TypeError): return None def get_device_info(self, device_id): @@ -346,9 +346,11 @@ def get_device_info(self, device_id): ) if response.status_code == 200: device_data = response.json()["devices"][0] + if not isinstance(device_data, dict): + return False, None return True, device_data return False, None - except (requests.exceptions.RequestException, IndexError, KeyError, TypeError): + except (requests.exceptions.RequestException, ValueError, IndexError, KeyError, TypeError): return False, None def get_ports(self, device_id, with_vlans=True): @@ -553,7 +555,7 @@ def add_location(self, location_data): location_id = result["message"].split("#")[-1] return True, {"id": location_id, "message": result["message"]} else: - return False, result.get("message", "Unexpected response format") + return False, result.get("message") or "Unexpected response format" except requests.exceptions.RequestException as e: error_message = str(e) if hasattr(e.response, "json"): @@ -591,7 +593,7 @@ def update_location(self, location_name, location_data): if result.get("status") == "ok": return True, result["message"] else: - return False, result.get("message", "Unexpected response format") + return False, result.get("message") or "Unexpected response format" except requests.exceptions.RequestException as e: error_message = str(e) if hasattr(e.response, "json"): @@ -645,7 +647,7 @@ def get_device_ips(self, device_id): message = data.get("message") if isinstance(data, dict) else None return False, message or "Unexpected response format: 'addresses' must be a list" return True, addresses - except requests.exceptions.RequestException as e: + except (requests.exceptions.RequestException, ValueError) as e: return False, str(e) def get_port_by_id(self, port_id): @@ -702,12 +704,12 @@ def get_device_inventory(self, device_id): response.raise_for_status() inventory_data = response.json() inventory = inventory_data.get("inventory") if isinstance(inventory_data, dict) else None - if not isinstance(inventory, list): + if not isinstance(inventory, list) or any(not isinstance(item, dict) for item in inventory): msg = inventory_data.get("message", "") if isinstance(inventory_data, dict) else "" logger.warning(f"Unexpected inventory response for device {device_id}: {inventory_data}") - return False, msg or "Unexpected response format: missing 'inventory' list" + return False, msg or "Unexpected response format: invalid 'inventory' payload" return True, inventory - except requests.exceptions.RequestException as e: + except (requests.exceptions.RequestException, ValueError) as e: return False, str(e) def get_poller_groups(self): @@ -739,11 +741,13 @@ def get_poller_groups(self): poller_groups = result.get("get_poller_group") if not isinstance(poller_groups, list): return False, result.get("message") or "Unexpected response format: missing 'get_poller_group' list" + if not all(isinstance(item, dict) for item in poller_groups): + return False, "Unexpected response format: invalid item shape in 'get_poller_group'" return True, poller_groups if isinstance(result, dict): return False, result.get("message") or "Unexpected response format" return False, "Unexpected response format: non-object JSON" - except requests.exceptions.RequestException as e: + except (requests.exceptions.RequestException, ValueError) as e: return False, str(e) def get_inventory_filtered(self, device_id, ent_physical_class=None, ent_physical_contained_in=None): @@ -790,9 +794,9 @@ def get_inventory_filtered(self, device_id, ent_physical_class=None, ent_physica data = response.json() if isinstance(data, dict) and data.get("status") == "ok": inventory = data.get("inventory") - if not isinstance(inventory, list): + if not isinstance(inventory, list) or any(not isinstance(item, dict) for item in inventory): msg = data.get("message") - return False, msg or "Unexpected response format: missing 'inventory' list" + return False, msg or "Unexpected response format: invalid 'inventory' payload" logger.debug(f"API returned {len(inventory)} items") # If we got results or didn't specify filters, return @@ -821,11 +825,11 @@ def get_inventory_filtered(self, device_id, ent_physical_class=None, ent_physica return True, filtered - return False, data.get("message", "Unexpected response format") if isinstance( + return False, data.get("message") or "Unexpected response format" if isinstance( data, dict ) else "Unexpected response format" - except requests.exceptions.RequestException as e: + except (requests.exceptions.RequestException, ValueError) as e: logger.warning(f"Failed to fetch filtered inventory: {e}") return False, str(e) @@ -893,12 +897,14 @@ def list_devices(self, filters=None): if not isinstance(devices, list): msg = result.get("message") return False, msg or "Unexpected response format: missing 'devices' list" + if not all(isinstance(item, dict) for item in devices): + return False, "Unexpected response format: invalid item shape in 'devices'" return True, devices - return False, result.get("message", "Unexpected response format") if isinstance( + return False, result.get("message") or "Unexpected response format" if isinstance( result, dict ) else "Unexpected response format" - except requests.exceptions.RequestException as e: + except (requests.exceptions.RequestException, ValueError) as e: return False, str(e) # ========================================================================= @@ -952,13 +958,13 @@ def get_device_vlans(self, device_id: int) -> tuple[bool, list | str]: ] return True, device_vlans if isinstance(result, dict): - return False, result.get("message", "Unexpected response format") + return False, result.get("message") or "Unexpected response format" return False, "Unexpected response format" except requests.exceptions.HTTPError as e: if e.response.status_code == 404: return False, "VLANs resource not found" return False, f"HTTP error: {str(e)}" - except requests.exceptions.RequestException as e: + except (requests.exceptions.RequestException, ValueError) as e: return False, f"Error connecting to LibreNMS: {str(e)}" def get_port_vlan_details(self, port_id: int) -> tuple[bool, dict | str]: @@ -1000,17 +1006,21 @@ def get_port_vlan_details(self, port_id: int) -> tuple[bool, dict | str]: result = response.json() if not isinstance(result, dict): return False, "Unexpected response format" - port_data = result.get("port", []) - if port_data and len(port_data) > 0: - return True, port_data[0] - return False, "Port not found" + port_data = result.get("port") + if not isinstance(port_data, list): + return False, result.get("message", "Unexpected response format: missing 'port' list") + if not port_data: + return False, "Port not found" + if not isinstance(port_data[0], dict): + return False, "Unexpected response format: invalid 'port' entry" + return True, port_data[0] return False, f"HTTP {response.status_code}" except requests.exceptions.HTTPError as e: if e.response.status_code == 404: return False, "Port not found in LibreNMS" return False, f"HTTP error: {str(e)}" - except requests.exceptions.RequestException as e: + except (requests.exceptions.RequestException, ValueError) as e: return False, f"Error connecting to LibreNMS: {str(e)}" def parse_port_vlan_data(self, port_data: dict, interface_name_field: str = "ifName") -> dict: @@ -1051,7 +1061,7 @@ def parse_port_vlan_data(self, port_data: dict, interface_name_field: str = "ifN untagged_vlan = None tagged_vlans = [] - if vlans_data: + if isinstance(vlans_data, list) and vlans_data: # Parse from detailed vlans array for vlan_entry in vlans_data: if not isinstance(vlan_entry, dict): 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 c2067eb166..39206ed1f0 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 @@ -458,6 +458,11 @@ function verifyVlanInGroup(select, deviceId, vid, vlanType, groupId) { const saveBtn = document.getElementById('saveVlanGroups'); _vlanVerifyStart(saveBtn); + // Capture safeName before the async fetch to avoid stale closure if the modal + // is opened for a different interface while this request is in flight. + const modal = document.getElementById('vlanDetailModal'); + const capturedSafeName = modal?.dataset.currentSafeName; + fetch('/plugins/librenms_plugin/verify-vlan-group/', { method: 'POST', headers: { @@ -506,10 +511,8 @@ function verifyVlanInGroup(select, deviceId, vid, vlanType, groupId) { } // Update the css in the source edit button's data-vlans - const modal = document.getElementById('vlanDetailModal'); - const safeName = modal?.dataset.currentSafeName; - if (safeName) { - const btn = document.querySelector(`.vlan-edit-btn[data-safe-name="${safeName}"]`); + if (capturedSafeName) { + const btn = document.querySelector(`.vlan-edit-btn[data-safe-name="${capturedSafeName}"]`); if (btn) { try { const btnVlans = JSON.parse(btn.dataset.vlans); @@ -1572,6 +1575,9 @@ function initializeModuleReplaceButtons() { // Fetch preview content and inject into modal body fetch(`${previewUrl}?${params.toString()}`, { signal, + headers: { + 'X-CSRFToken': document.querySelector('[name=csrfmiddlewaretoken]').value, + }, }) .then(response => { if (!response.ok) return response.text().then(t => { throw new Error(t); }); @@ -1581,6 +1587,7 @@ function initializeModuleReplaceButtons() { const modalBody = document.getElementById('htmx-modal-body'); if (modalBody) { modalBody.innerHTML = html; + htmx.process(modalBody); } }) .catch(err => { diff --git a/netbox_librenms_plugin/tables/interfaces.py b/netbox_librenms_plugin/tables/interfaces.py index 1216313da4..d99d2a7bfe 100644 --- a/netbox_librenms_plugin/tables/interfaces.py +++ b/netbox_librenms_plugin/tables/interfaces.py @@ -47,7 +47,7 @@ class Meta: "id": "librenms-interface-table", } - def __init__(self, *args, device=None, interface_name_field=None, vlan_groups=None, server_key="default", **kwargs): + def __init__(self, *args, device=None, interface_name_field=None, vlan_groups=None, server_key=None, **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() diff --git a/netbox_librenms_plugin/tests/mock_librenms_server.py b/netbox_librenms_plugin/tests/mock_librenms_server.py index 060ba62006..83a9258dba 100644 --- a/netbox_librenms_plugin/tests/mock_librenms_server.py +++ b/netbox_librenms_plugin/tests/mock_librenms_server.py @@ -201,10 +201,15 @@ def auth_error_response(self, path="/api/v0/devices"): def inventory_response(self, device_id: int, items: list, status: int = 200): """Register a plain inventory response for /api/v0/inventory/{device_id}/all.""" + payload_status = "ok" if 200 <= status < 300 else "error" + payload = ( + {"status": payload_status, "inventory": items} if payload_status == "ok" else {"status": payload_status} + ) self.register( f"/api/v0/inventory/{device_id}/all", - {"status": "ok", "inventory": items}, + payload, status=status, + method="GET", ) def vc_inventory_callable(self, device_id: int, root_items: list, children_by_parent_index: dict): @@ -225,6 +230,11 @@ def _handler(method, path, query, headers, body): if contained_in == "0": return 200, {"status": "ok", "inventory": root} if contained_in is not None: + # Require entPhysicalClass=chassis for child queries so tests catch + # any regression where the production code stops sending the class filter. + phy_class = query.get("entPhysicalClass", [None])[0] + if phy_class != "chassis": + return 200, {"status": "ok", "inventory": []} try: idx = int(contained_in) except (TypeError, ValueError): @@ -237,8 +247,8 @@ def _handler(method, path, query, headers, body): all_items.extend(v) return 200, {"status": "ok", "inventory": all_items} - self.routes[f"/api/v0/inventory/{device_id}"] = _handler - self.routes[f"/api/v0/inventory/{device_id}/all"] = _handler + self.register(f"/api/v0/inventory/{device_id}", _handler, method="GET") + self.register(f"/api/v0/inventory/{device_id}/all", _handler, method="GET") @contextmanager diff --git a/netbox_librenms_plugin/tests/test_background_jobs.py b/netbox_librenms_plugin/tests/test_background_jobs.py index 88f6931cf7..f1d1bd2274 100644 --- a/netbox_librenms_plugin/tests/test_background_jobs.py +++ b/netbox_librenms_plugin/tests/test_background_jobs.py @@ -437,9 +437,9 @@ def test_run_mixed_device_and_vm_import(self, mock_api_class, mock_bulk_devices, """Import both devices and VMs.""" from netbox_librenms_plugin.jobs import ImportDevicesJob - mock_api_instance = MagicMock() - mock_api_instance.server_key = "default" - mock_api_class.return_value = mock_api_instance + mock_api = MagicMock() + mock_api.server_key = "non-default" + mock_api_class.return_value = mock_api # Mock device imports mock_device = MagicMock() @@ -468,16 +468,20 @@ def test_run_mixed_device_and_vm_import(self, mock_api_class, mock_bulk_devices, job.run( device_ids=[1], vm_imports={10: {"cluster_id": 1}}, - server_key="default", + server_key="non-default", ) # Both should be called mock_bulk_devices.assert_called_once() mock_bulk_vms.assert_called_once() - # Verify server_key is forwarded to bulk_import_devices_shared + # Verify server_key (via api.server_key) is forwarded to bulk_import_devices_shared bulk_devices_kwargs = mock_bulk_devices.call_args[1] - assert bulk_devices_kwargs.get("server_key") == "default" + assert bulk_devices_kwargs.get("server_key") == "non-default" + + # Verify bulk_import_vms received the api with the correct server_key + bulk_vms_positional = mock_bulk_vms.call_args[0] + assert bulk_vms_positional[1].server_key == "non-default" # Verify combined results assert job.job.data["imported_device_pks"] == [100] 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..ad6dc468c0 --- /dev/null +++ b/netbox_librenms_plugin/tests/test_cable_verify.py @@ -0,0 +1,331 @@ +""" +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, server_key=None): + assert link is not None + assert hostname is not None + assert device_id is not None + assert server_key == "default" + # 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 ast + import inspect + import re + + 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) + + def extract_raw_keys(src, label): + match = re.search(r"_raw_keys\s*=\s*(\{[^}]*\})", src, re.S) + assert match, f"_raw_keys missing from {label}" + return set(ast.literal_eval(match.group(1))) + + prepare_keys = extract_raw_keys(prepare_src, "_prepare_context") + post_keys = extract_raw_keys(post_src, "post()") + assert prepare_keys == post_keys, f"_raw_keys mismatch: {prepare_keys} != {post_keys}" + def test_post_strips_derived_fields_from_cached_link(self): + """post() must strip derived fields (URLs, IDs) before re-enrichment. + + Both _prepare_context and post() define a _raw_keys set that controls + which cached fields survive into re-enrichment. This test verifies the + behavior: derived fields in the cached link must not leak through. + """ + view = _make_view() + + # Cached link with both raw and derived (stale) fields + cached_link = { + "local_port": "eth0", + "local_port_id": 100, + "remote_port": "eth1", + "remote_device": "switch-a", + "remote_port_id": 200, + "remote_device_id": 42, + # Derived fields that must be stripped: + "netbox_local_interface_id": 999, + "netbox_remote_interface_id": 888, + "netbox_remote_device_id": 777, + "local_port_url": "/stale/", + "remote_port_url": "/stale/", + "remote_device_url": "/stale/", + "cable_status": "stale", + "can_create_cable": True, + } + + # Mock process_remote_device to avoid DB access during re-enrichment; + # it should receive the link WITHOUT derived fields. + received_link = {} + + def fake_process_remote(link, hostname, device_id, server_key=None): + received_link.update(link) + return link + + view.process_remote_device = fake_process_remote + + with ( + patch("netbox_librenms_plugin.views.base.cables_view.get_object_or_404") as mock_get, + patch("netbox_librenms_plugin.views.base.cables_view.cache") as mock_cache, + patch("netbox_librenms_plugin.views.base.cables_view.get_librenms_sync_device", return_value=None), + patch("netbox_librenms_plugin.views.base.cables_view.get_token", return_value="tok"), + ): + device = MagicMock() + device.pk = 1 + device.virtual_chassis = None + device.interfaces.filter.return_value.first.return_value = None + mock_get.return_value = device + mock_cache.get.return_value = {"links": [cached_link]} + + request = MagicMock() + request.body = json.dumps( + { + "device_id": 1, + "local_port_id": 100, + "server_key": "default", + } + ) + view.post(request) + + # The link passed to process_remote_device must have derived fields stripped + assert "netbox_local_interface_id" not in received_link + assert "netbox_remote_interface_id" not in received_link + assert "netbox_remote_device_id" not in received_link + assert "local_port_url" not in received_link + assert "cable_status" not in received_link + + +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, server_key=None): + 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 "' + vc.members.all.return_value = [member] + device.virtual_chassis = vc + + table = VCCableTable([], device=device) + record = {"local_port": "eth0", "local_port_id": "42"} + + with patch( + "netbox_librenms_plugin.tables.cables.get_virtual_chassis_member", + return_value=member, + ): + html = str(table.render_device_selection(None, record)) + + # The raw