diff --git a/.gitignore b/.gitignore index 25c298c23b..164a3d160c 100644 --- a/.gitignore +++ b/.gitignore @@ -290,3 +290,6 @@ cython_debug/ ca-bundle.crt *.pem .github/hooks/ + +# Playwright MCP scratch snapshots (accessibility/page dumps from local UI verification) +.playwright-mcp/ diff --git a/netbox_librenms_plugin/import_utils/bulk_import.py b/netbox_librenms_plugin/import_utils/bulk_import.py index 93962432e7..d4fde87631 100644 --- a/netbox_librenms_plugin/import_utils/bulk_import.py +++ b/netbox_librenms_plugin/import_utils/bulk_import.py @@ -8,7 +8,7 @@ from ..import_validation_helpers import apply_role_to_validation, recalculate_validation_status, remove_validation_issue from ..librenms_api import LibreNMSAPI -from ..utils import find_by_librenms_id +from ..utils import find_by_librenms_id, preload_normalization_rules from .cache import get_cache_metadata_key, get_import_device_cache_key, get_validated_device_cache_key from .device_operations import import_single_device, validate_device_for_import from .filters import _safe_disabled, get_librenms_devices_for_import @@ -117,6 +117,10 @@ def bulk_import_devices_shared( # Initialize API client once for all devices to avoid repeated config parsing api = LibreNMSAPI(server_key=server_key) + # Preload the device_type NormalizationRule set once so the per-device hardware→device-type + # match doesn't re-query it for every device in the loop (issue #90 / N+1 avoidance, #92). + device_type_norm_rules = preload_normalization_rules("device_type") + for idx, device_id in enumerate(device_ids, start=1): # Check for job cancellation on first iteration and every 5th thereafter. if job and (idx == 1 or idx % 5 == 0) and _is_job_cancelled(job): @@ -133,7 +137,16 @@ def bulk_import_devices_shared( libre_device = libre_devices_cache[device_id] success = True else: - success, libre_device = api.get_device_info(device_id) + # Import decisions (DeviceType match, serial-conflict, hostname) must run against + # live LibreNMS data: bypass the short device-info read cache so a value the user + # just corrected in LibreNMS isn't read back stale within the cache window. + success, libre_device = api.get_device_info(device_id, use_cache=False) + # Backfill the shared cache so the synchronous import's post-import row re-render + # (which reads the same dict via fetch_device_with_cache) doesn't issue a second + # LibreNMS round-trip per device on a cold cache. The background path passes a + # serialized copy and skips that re-render, so this is a harmless no-op there. + if success and libre_device is not None and libre_devices_cache is not None: + libre_devices_cache[device_id] = libre_device if not success or not libre_device: error_msg = f"Failed to retrieve device {device_id} from LibreNMS" @@ -156,6 +169,7 @@ def bulk_import_devices_shared( # LibreNMS inventory so stack members are created even when preview # flags are stale or omitted. include_vc_detection=True, + preloaded_device_type_rules=device_type_norm_rules, ) vc_data = validation.get("virtual_chassis", {}) @@ -410,15 +424,21 @@ def _refresh_existing_device(validation: dict, libre_device: dict = None, server match_type = None found_as_cross_model = False - def _lookup_in_model(m): - """Return (device, match_type) for model m, or (None, None).""" + def _id_lookup(m): + """librenms_id match in model m, or None.""" if librenms_id is not None and not isinstance(librenms_id, bool): - try: - dev = find_by_librenms_id(m, int(librenms_id), server_key) - if dev: - return dev, "librenms_id" - except (ValueError, TypeError): - pass + # int/str only (find_by_librenms_id's contract). Don't int()-coerce: + # int(42.9) -> 42 would bind an unrelated object by a truncated id. A float + # (or any other type) carries no exact id, so fail closed; str-digit ids are + # canonicalized by find_by_librenms_id itself (mirrors _librenms_id_q's + # lossy-coercion guard, #103). + if not isinstance(librenms_id, (int, str)): + return None + return find_by_librenms_id(m, librenms_id, server_key) + return None + + def _name_lookup(m): + """name-based match in model m → (device, match_type) or (None, None).""" resolved_name = validation.get("resolved_name") if resolved_name: dev = m.objects.filter(name__iexact=resolved_name).first() @@ -434,14 +454,24 @@ def _lookup_in_model(m): return dev, "sysname" return None, None - new_device, match_type = _lookup_in_model(Model) - - if not new_device: - # Try the opposite model: catches cross-model imports that happened - # after the cache was built (e.g. LibreNMS device imported as VM). - new_device, match_type = _lookup_in_model(CrossModel) - if new_device: - found_as_cross_model = True + # An exact librenms_id match wins across BOTH models before any name fallback. Otherwise a + # refresh could bind a name-colliding object in the preferred model even when the scanned + # LibreNMS id is already linked to the *other* model (e.g. a VM), disagreeing with + # validation and rendering actions for the wrong object. + dev = _id_lookup(Model) + if dev: + new_device, match_type = dev, "librenms_id" + else: + dev = _id_lookup(CrossModel) + if dev: + new_device, match_type, found_as_cross_model = dev, "librenms_id", True + else: + new_device, match_type = _name_lookup(Model) + if not new_device: + # Catches cross-model imports made after the cache was built. + new_device, match_type = _name_lookup(CrossModel) + if new_device: + found_as_cross_model = True if new_device: validation["existing_device"] = new_device diff --git a/netbox_librenms_plugin/import_utils/device_operations.py b/netbox_librenms_plugin/import_utils/device_operations.py index 6b87935b03..948f9a9e6f 100644 --- a/netbox_librenms_plugin/import_utils/device_operations.py +++ b/netbox_librenms_plugin/import_utils/device_operations.py @@ -15,6 +15,7 @@ find_by_librenms_id, find_matching_platform, find_matching_site, + is_legacy_librenms_id, match_librenms_hardware_to_device_type, set_librenms_device_id, ) @@ -29,7 +30,7 @@ logger = logging.getLogger(__name__) -def _try_chassis_device_type_match(api, device_id): +def _try_chassis_device_type_match(api, device_id, preloaded_device_type_rules: dict | None = None): """ Attempt device type matching using chassis inventory fields. @@ -41,8 +42,15 @@ def _try_chassis_device_type_match(api, device_id): Tries entPhysicalName first (typically the chassis part number), then entPhysicalModelName as fallback. + Args: + api (LibreNMSAPI): API client used to fetch the chassis inventory. + device_id (int): LibreNMS device ID whose chassis inventory to inspect. + preloaded_device_type_rules (dict | None): Optional device_type NormalizationRule set from + :func:`preload_normalization_rules`, threaded into the inner match so bulk imports don't + reissue the rule query per device. + Returns: - dict with matched/device_type/match_type keys, or None on failure. + dict | None: Dict with matched/device_type/match_type keys, or None on failure. """ skip_values = {"", "-", "Unspecified", "BUILTIN", "None"} @@ -56,7 +64,9 @@ def _try_chassis_device_type_match(api, device_id): for field in ("entPhysicalName", "entPhysicalModelName"): value = item.get(field) or "" if value and value not in skip_values: - chassis_match = match_librenms_hardware_to_device_type(value) + chassis_match = match_librenms_hardware_to_device_type( + value, preloaded_rules=preloaded_device_type_rules + ) if chassis_match is None: continue if chassis_match["matched"]: @@ -135,6 +145,7 @@ def validate_device_for_import( force_vc_refresh: bool = False, use_sysname: bool = True, strip_domain: bool = False, + preloaded_device_type_rules: dict | None = None, ) -> dict: """ Validate if a LibreNMS device can be imported to NetBox. @@ -301,9 +312,7 @@ def validate_device_for_import( # 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() - ): + if is_legacy_librenms_id(_vm_cf_id): result["librenms_id_needs_migration"] = True # Check if name matches resolved name (accounts for use_sysname/strip_domain) @@ -323,6 +332,10 @@ def validate_device_for_import( logger.info(f"Found existing device: {existing_device.name} (matched by librenms_id={librenms_id})") result["existing_device"] = existing_device result["existing_match_type"] = "librenms_id" + # A Device matched: force Device mode so a user-selected VM mode doesn't carry a + # Device-mapped row into the VM validation/UI path (the VM branches above set + # import_as_vm=True; the Device branches must symmetrically set it False). + result["import_as_vm"] = False result["can_import"] = False # Detect legacy bare-integer or string-digit format so UI can offer a migration action. @@ -330,9 +343,7 @@ def validate_device_for_import( # 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() - ): + if is_legacy_librenms_id(_dev_cf_id): result["librenms_id_needs_migration"] = True # Check if name matches resolved name (VC-aware: compare against VC member name) @@ -411,6 +422,9 @@ def validate_device_for_import( logger.info(f"Found existing device by hostname: {existing_device.name}") result["existing_device"] = existing_device result["existing_match_type"] = "hostname" + # Force Device mode (see the librenms_id branch above): a hostname match to a + # Device must not leave a user-selected VM mode active. + result["import_as_vm"] = False # Check for serial conflict on hostname-matched device incoming_serial = libre_device.get("serial") or "" @@ -443,7 +457,26 @@ def validate_device_for_import( if not result["existing_device"]: serial = libre_device.get("serial") or "" if serial and serial != "-" and not import_as_vm: - existing_by_serial = Device.objects.filter(serial=serial).first() + # Serial is not unique in NetBox, so .first() would bind an arbitrary row + # and the downstream serial/OOB/merge flow would derive its guidance from a + # random device. Require a unique match before binding, mirroring the + # merge-peer [:2] guard (issue #101). + serial_matches = list(Device.objects.filter(serial=serial)[:2]) + if len(serial_matches) > 1: + # Device.serial is not unique in NetBox, so several rows already share it. + # Binding to an arbitrary one is wrong, and importing anyway would mint YET + # ANOTHER same-serial device. Make it a blocking issue (issue -> can_import + # False) and flag the duplicate, rather than only warning and letting the row + # import. Not serial_action="conflict": there is no single peer to resolve + # against, so the sync-serial UI (which needs one existing_device) can't apply. + result["serial_duplicate"] = True + result["issues"].append( + f"Multiple NetBox devices share serial '{serial}'; resolve the duplicate " + "serial before importing." + ) + existing_by_serial = None + else: + existing_by_serial = serial_matches[0] if serial_matches else None if existing_by_serial: logger.info(f"Found existing device by serial: {existing_by_serial.name} (serial={serial})") result["existing_device"] = existing_by_serial @@ -528,7 +561,7 @@ def validate_device_for_import( # 3. Validate DeviceType (required) hardware = libre_device.get("hardware", "") - dt_match = match_librenms_hardware_to_device_type(hardware) + dt_match = match_librenms_hardware_to_device_type(hardware, preloaded_rules=preloaded_device_type_rules) if dt_match is None: result["device_type"]["found"] = False @@ -544,7 +577,9 @@ def validate_device_for_import( if not dt_match["matched"] and api: device_id = libre_device.get("device_id") if device_id: - chassis_match = _try_chassis_device_type_match(api, device_id) + chassis_match = _try_chassis_device_type_match( + api, device_id, preloaded_device_type_rules=preloaded_device_type_rules + ) if chassis_match and chassis_match["matched"]: dt_match = chassis_match @@ -755,9 +790,12 @@ def import_single_device( try: api = LibreNMSAPI(server_key=server_key) - # Use pre-fetched device data if provided, otherwise fetch from API + # Use pre-fetched device data if provided, otherwise fetch from API. Read LIVE + # (use_cache=False): this is the device-CREATION path — name/serial/status/hardware are + # derived from libre_device — so it must not build a NetBox device from the 60s get_device_info + # snapshot a sync-tab render may have seeded, mirroring the import flow's live-read policy. if libre_device is None: - success, libre_device = api.get_device_info(device_id) + success, libre_device = api.get_device_info(device_id, use_cache=False) if not success or not libre_device: return { "success": False, @@ -936,20 +974,23 @@ def import_single_device( } -def get_librenms_device_by_id(api: LibreNMSAPI, device_id: int) -> dict: +def get_librenms_device_by_id(api: LibreNMSAPI, device_id: int, use_cache: bool = True) -> dict: """ Retrieve a single device from LibreNMS by ID. Args: api: LibreNMSAPI instance device_id: LibreNMS device ID + use_cache: Passed through to ``get_device_info``. The import fallback passes ``False`` so + that when ``fetch_device_with_cache``'s own caches miss, the API fallback returns live + data rather than the 60s get_device_info snapshot a sync-tab render may have seeded. Returns: Device dictionary or None if not found """ try: # Use the dedicated API endpoint to get device by ID - success, device = api.get_device_info(device_id) + success, device = api.get_device_info(device_id, use_cache=use_cache) if success and device: return device @@ -1005,8 +1046,11 @@ def fetch_device_with_cache( libre_device = cache.get(cache_key) if not libre_device: - # Fallback to API fetch - libre_device = get_librenms_device_by_id(api, device_id) + # Fallback to API fetch. Read LIVE: both this function's caches (the pre-fetched dict and the + # import Django cache) have already missed, so the fallback must reflect current LibreNMS + # state — not the separate 60s get_device_info snapshot a sync-tab render may have populated — + # or the confirm modal / re-rendered row shows metadata that disagrees with what import uses. + libre_device = get_librenms_device_by_id(api, device_id, use_cache=False) if libre_device: # Cache for future use cache.set(cache_key, libre_device, timeout=api.cache_timeout) diff --git a/netbox_librenms_plugin/import_utils/vm_operations.py b/netbox_librenms_plugin/import_utils/vm_operations.py index 99ae10a8af..99f1fa0a05 100644 --- a/netbox_librenms_plugin/import_utils/vm_operations.py +++ b/netbox_librenms_plugin/import_utils/vm_operations.py @@ -65,9 +65,16 @@ def create_vm_from_librenms( # Validate device_id before creating the VM so a missing/invalid value # never leaves a VM without a librenms_id (partial persistence). raw_device_id = libre_device["device_id"] - if isinstance(raw_device_id, bool): - raise ValueError(f"device_id is a boolean ({raw_device_id!r}); expected an integer") + # Reject bool (a subclass of int → int(True) == 1) AND floats/Decimals (int(1.9) would truncate + # to a valid-looking but WRONG pk), accepting only a real int or a plain digit string. Without + # this a numeric-like device_id would create a VM with a truncated/garbage librenms_id mapping. + if isinstance(raw_device_id, bool) or not ( + isinstance(raw_device_id, int) or (isinstance(raw_device_id, str) and raw_device_id.strip().isdecimal()) + ): + raise ValueError(f"device_id {raw_device_id!r} is not a valid integer id") librenms_device_id = int(raw_device_id) + if librenms_device_id <= 0: + raise ValueError(f"device_id {raw_device_id!r} must be a positive integer") from ..utils import set_librenms_device_id diff --git a/netbox_librenms_plugin/librenms_api.py b/netbox_librenms_plugin/librenms_api.py index 86eba28f32..7b2ded8334 100644 --- a/netbox_librenms_plugin/librenms_api.py +++ b/netbox_librenms_plugin/librenms_api.py @@ -9,6 +9,12 @@ DEFAULT_API_TIMEOUT = 10 EXTENDED_API_TIMEOUT = 20 # For endpoints that may take longer (e.g., device listing) +# Short-lived cache for get_device_info(). The device-info header is fetched on every sync-tab +# render (and again on the post-action redirect), but a device's identity/metadata is stable, so +# a brief cache removes a redundant synchronous LibreNMS round-trip from each render without +# meaningfully staling the displayed values. +DEVICE_INFO_CACHE_TIMEOUT = 60 + logger = logging.getLogger(__name__) @@ -24,6 +30,22 @@ def __init__(self, server_key=None): Args: server_key: Key for specific server configuration. If None, uses selected server or default. """ + # Track whether the caller explicitly requested a specific server. A key auto-resolved + # from LibreNMSSettings.selected_server is NOT explicit, so a stale stored key falls back + # to the first available server rather than hard-failing (issue #110). + # A blank/whitespace-only string is "no key", not an explicit request: treating "" as + # explicit would mark the auto-resolved (and possibly stale) selected_server as explicit + # and defeat that fallback, raising KeyError instead. + if isinstance(server_key, str): + server_key = server_key.strip() or None + elif server_key is not None: + # A non-string key (e.g. a list/dict from a tampered payload) is not a valid server + # key and is unhashable — left as-is it would raise TypeError at the `not in + # servers_config` membership check below. Treat it as unset so it fails cleanly via + # the same fallback path as a blank string. + server_key = None + explicit_server_key = server_key is not None + # If no server_key is provided, try to get the selected server from settings if not server_key: try: @@ -46,31 +68,58 @@ def __init__(self, server_key=None): # If a specific (non-default) server_key was requested but not found, raise # immediately to avoid silently using the wrong LibreNMS instance. if servers_config and isinstance(servers_config, dict) and server_key not in servers_config: - if server_key != "default": + # Only fail closed for an *explicitly* requested non-default key (tampered or + # stale-page input). An auto-resolved/default key falls back instead (issue #110). + if explicit_server_key and server_key != "default": available = list(servers_config.keys()) raise KeyError( f"Server '{server_key}' not found in LibreNMS plugin configuration. Available servers: {available}" ) - first_key = next(iter(servers_config), None) - if first_key: - logger.info( - "Server '%s' not found in config, falling back to '%s'", - server_key, - first_key, - ) - server_key = first_key + # Pick the first *usable* mapping — a dict that actually carries a url + token. A + # non-dict or incomplete entry would otherwise be selected and then raise at the config + # read below, masking a perfectly usable later server (e.g. {"bad": {}, "prod": {...}}). + # If none are usable, fail with a clear error (issue #110). + first_key = next( + ( + key + for key, config in servers_config.items() + if isinstance(config, dict) and config.get("librenms_url") and config.get("api_token") + ), + None, + ) + if first_key is None: + raise ValueError("No valid LibreNMS server configuration entries found.") + logger.info( + "Server '%s' not found in config, falling back to '%s'", + server_key, + first_key, + ) + server_key = first_key self.server_key = server_key if servers_config and isinstance(servers_config, dict) and server_key in servers_config: # Multi-server configuration config = servers_config[server_key] - self.librenms_url = config["librenms_url"] - self.api_token = config["api_token"] + # The fallback above only guards the key-not-found case; a present-but-malformed + # entry (e.g. {"default": "not-a-dict"}) would otherwise raise an opaque TypeError + # at the key reads below. Fail with a clear configuration error instead (issue #110). + if not isinstance(config, dict): + raise ValueError(f"Invalid LibreNMS server configuration for '{server_key}': expected a mapping.") + # Read with .get() rather than direct indexing: a dict-shaped but incomplete entry + # (e.g. {"default": {}}) passes the isinstance check, so config["librenms_url"] would + # raise an opaque KeyError instead of the ValueError contract callers rely on. The + # url/token completeness is enforced by the single guard at the end of __init__. + self.librenms_url = config.get("librenms_url") + self.api_token = config.get("api_token") self.cache_timeout = config.get("cache_timeout", 300) self.verify_ssl = config.get("verify_ssl", True) else: - # Fallback to legacy single-server configuration + # Fallback to legacy single-server configuration. Legacy mode has only the implicit + # default server, so a stale/tampered request key (e.g. build_librenms_api("ghost")) + # must not survive as self.server_key — it would otherwise become the cache/redirect + # discriminator under an unconfigured value. Normalize to "default". + self.server_key = "default" self.librenms_url = get_plugin_config("netbox_librenms_plugin", "librenms_url") self.api_token = get_plugin_config("netbox_librenms_plugin", "api_token") self.cache_timeout = get_plugin_config("netbox_librenms_plugin", "cache_timeout", 300) @@ -160,8 +209,14 @@ def get_available_servers(cls): # Multi-server configuration result = {} for key, config in servers_config.items(): - display_name = config.get("display_name", key) - result[key] = display_name + # Skip non-usable entries: a non-dict raises on config.get(...) below, and a + # dict-shaped but incomplete entry (no librenms_url/api_token, e.g. {"bad": {}}) + # would pass the redirect/rebind membership check that keys off this map and then + # blow up LibreNMSAPI(server_key="bad"). A server that can't be constructed must not + # be selectable — exposing it 500s a sync POST instead of degrading to the active one. + if not isinstance(config, dict) or not config.get("librenms_url") or not config.get("api_token"): + continue + result[key] = config.get("display_name", key) return result else: # Legacy single-server configuration @@ -346,16 +401,29 @@ def get_device_id_by_hostname(self, hostname): except (requests.exceptions.RequestException, ValueError, IndexError, KeyError, TypeError): return None - def get_device_info(self, device_id): + def get_device_info(self, device_id, use_cache=True): """ Fetch device information from LibreNMS using its primary IP. + Successful lookups are cached briefly (``DEVICE_INFO_CACHE_TIMEOUT``) per + server/device so the device-info header doesn't re-hit LibreNMS on every + sync-tab render. Failures are never cached, so a transient error doesn't + persist for the cache window. + Args: device_id: LibreNMS device ID + use_cache: When False, bypass the short read cache and fetch live data + (still refreshing the cache on success). Import decisions pass False so a + value just corrected in LibreNMS isn't read back stale within the cache window. Returns: tuple: (success: bool, data: dict) """ + cache_key = f"librenms_device_info_{self.server_key}_{device_id}" + if use_cache: + cached = cache.get(cache_key) + if cached is not None: + return cached try: response = requests.get( @@ -375,7 +443,9 @@ def get_device_info(self, device_id): location = device_data.get("location") if isinstance(location, dict): device_data["location"] = location.get("location") - return True, device_data + result = (True, device_data) + cache.set(cache_key, result, timeout=DEVICE_INFO_CACHE_TIMEOUT) + return result except (requests.exceptions.RequestException, ValueError, IndexError, KeyError, TypeError): return False, None diff --git a/netbox_librenms_plugin/migrations/0011_renormalize_device_type_mappings.py b/netbox_librenms_plugin/migrations/0011_renormalize_device_type_mappings.py new file mode 100644 index 0000000000..71b4bcb602 --- /dev/null +++ b/netbox_librenms_plugin/migrations/0011_renormalize_device_type_mappings.py @@ -0,0 +1,117 @@ +"""Re-key existing DeviceTypeMapping rows through the device_type NormalizationRule scope. + +The hardware→DeviceType lookup (``utils.match_librenms_hardware_to_device_type``) now normalizes +the LibreNMS hardware string via the ``device_type`` NormalizationRule scope *before* querying +``librenms_hardware__iexact``, and newly created mappings are stored normalized. Mappings created +before that change hold the un-normalized value, so once any ``device_type`` rule exists they no +longer match the normalized lookup. This data migration applies the same normalization the runtime +uses to existing rows so they keep matching after upgrade. +""" + +import logging + +from django.db import migrations, transaction + +logger = logging.getLogger(__name__) + + +def renormalize_device_type_mappings(apps, schema_editor): + """Apply the device_type normalization to each existing DeviceTypeMapping's hardware key.""" + DeviceTypeMapping = apps.get_model("netbox_librenms_plugin", "DeviceTypeMapping") + + # Use the runtime normalizer so re-keyed values mirror the lookup path exactly. Guard the + # import/normalization: if anything is unavailable, leave rows untouched rather than corrupt + # them (a fresh install has no rows to migrate anyway). + try: + from netbox_librenms_plugin.utils import apply_normalization_rules, preload_normalization_rules + except Exception: # pragma: no cover - defensive + return + + # Preload the device_type rule chain ONCE. Without this, each per-row + # apply_normalization_rules() re-queries NormalizationRule (an avoidable N+1 that scales with + # the number of existing mappings); passing the preloaded dict makes the whole migration issue + # a constant number of rule queries. The mappings are un-scoped (manufacturer=None), matching + # the per-row calls below which pass no manufacturer. + # + # Guard the preload the same way the per-row work is guarded: it issues a DB query, so a failure + # here (connection/query error mid-upgrade) would otherwise abort the whole migration BEFORE any + # per-row savepoint runs. And because it is a DB error, catching it alone is not enough — on + # PostgreSQL it poisons the migration's outer atomic transaction, so subsequent work (or the + # migration's own commit) would still fail. Wrap it in its own savepoint and bail out leaving + # every row untouched, mirroring the import guard above and the "leave rows untouched rather + # than corrupt them" contract. + try: + with transaction.atomic(): + preloaded_rules = preload_normalization_rules(scope="device_type") + except Exception: + logger.exception( + "renormalize_device_type_mappings: failed to preload device_type normalization rules; " + "leaving all rows untouched." + ) + return + + for mapping in DeviceTypeMapping.objects.all().iterator(): + raw = mapping.librenms_hardware or "" + rekeyed = None + # One savepoint per ROW, wrapping every DB touch — the NormalizationRule queries + # inside apply_normalization_rules, the clash .exists(), AND the save. Under the + # migration's default atomic transaction, a DB-level failure in ANY of them poisons + # the whole transaction on PostgreSQL, so a bare try/except could not actually + # continue — the next row's query would error with "current transaction is + # aborted". Rolling back only this row's savepoint keeps the loop going, mirroring + # the "leave it for manual fixup" path. (The row iterator's cursor predates the + # savepoint, so a rollback doesn't invalidate it.) + try: + with transaction.atomic(): + normalized = ( + (apply_normalization_rules(value=raw, scope="device_type", preloaded_rules=preloaded_rules) or "") + .strip() + .lower() + ) + + # Stored values are already lowercased (the model lowercases on save); skip when + # the normalization is a no-op so we don't issue pointless writes or fire signals. + if not normalized or normalized == raw.strip().lower(): + continue + + # The lookup is case-insensitive (__iexact). If another row already holds the + # normalized key, re-keying this one would trip the unique constraint — and + # signals that two raw mappings collapse to the same normalized value. Leave the + # original and warn so the operator can resolve the duplicate by hand. + clash = ( + DeviceTypeMapping.objects.exclude(pk=mapping.pk) + .filter(librenms_hardware__iexact=normalized) + .exists() + ) + if clash: + logger.warning( + "renormalize_device_type_mappings: %r normalizes to %r which already maps to a " + "device type; leaving the original and skipping to avoid a unique-constraint clash " + "— resolve the duplicate DeviceTypeMapping manually.", + raw, + normalized, + ) + continue + + mapping.librenms_hardware = normalized + mapping.save(update_fields=["librenms_hardware"]) + rekeyed = normalized + except Exception: + logger.exception( + "renormalize_device_type_mappings: failed to re-key %r; leaving the original value.", + raw, + ) + continue + if rekeyed is not None: + logger.info("renormalize_device_type_mappings: re-keyed %r → %r", raw, rekeyed) + + +class Migration(migrations.Migration): + dependencies = [ + ("netbox_librenms_plugin", "0010_inventory_and_mapping_models"), + ] + + operations = [ + # Irreversible: the original (un-normalized) hardware strings cannot be reconstructed. + migrations.RunPython(renormalize_device_type_mappings, migrations.RunPython.noop), + ] diff --git a/netbox_librenms_plugin/models.py b/netbox_librenms_plugin/models.py index 16c8b68952..e8aa8dec77 100644 --- a/netbox_librenms_plugin/models.py +++ b/netbox_librenms_plugin/models.py @@ -10,6 +10,8 @@ from django.urls import reverse from netbox.models import NetBoxModel +from netbox_librenms_plugin.utils import validate_regex_field + logger = logging.getLogger(__name__) @@ -364,10 +366,7 @@ def clean(self): raise ValidationError({"netbox_bay_name": "NetBox bay name must not be empty or whitespace-only."}) self.netbox_bay_name = netbox_bay_name_stripped if self.is_regex: - try: - pattern = re.compile(self.librenms_name) - except re.error as e: - raise ValidationError({"librenms_name": f"Invalid regex: {e}"}) + pattern = validate_regex_field(self.librenms_name, "librenms_name") try: _validate_replacement_template(pattern, self.netbox_bay_name) except (re.error, IndexError) as e: @@ -487,10 +486,7 @@ def clean(self): errors["replacement"] = "This field is required." if errors: raise ValidationError(errors) - try: - compiled = re.compile(self.match_pattern) - except re.error as e: - raise ValidationError({"match_pattern": f"Invalid regex: {e}"}) + compiled = validate_regex_field(self.match_pattern, "match_pattern") # Validate the replacement template by running a dummy substitution try: _validate_replacement_template(compiled, self.replacement) @@ -621,10 +617,7 @@ def clean(self): self.__dict__.pop("_compiled_pattern", None) pattern_stripped = self.pattern.strip() if self.pattern else "" if self.match_type == self.MATCH_REGEX and pattern_stripped: - try: - re.compile(pattern_stripped) - except re.error as e: - raise ValidationError({"pattern": f"Invalid regex: {e}"}) + validate_regex_field(pattern_stripped, "pattern") if self.match_type != self.MATCH_SERIAL_DEVICE and not pattern_stripped: raise ValidationError({"pattern": "Pattern is required for name-based match types."}) # Normalize stored pattern to the stripped form so matches_name() and @@ -863,10 +856,7 @@ def clean(self): ): if not value: continue - try: - re.compile(value) - except re.error as e: - raise ValidationError({field: f"Invalid regex: {e}"}) + validate_regex_field(value, field) def get_absolute_url(self): return reverse( 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 9a9b0a0207..78a7685888 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 @@ -18,6 +18,17 @@ const TOMSELECT_INIT_DELAY_MS = 100; const COUNTDOWN_UPDATE_INTERVAL_MS = 1000; +/** + * Return the CSRF token value, or null when the hidden input is missing/empty. + * Callers MUST bail (running any needed UI cleanup) on null instead of reading + * `.value` off a missing element, which throws a TypeError and breaks the handler. + * @returns {string|null} + */ +function getCsrfToken() { + const input = document.querySelector('[name=csrfmiddlewaretoken]'); + return input && input.value ? input.value : null; +} + /** * Show a Bootstrap modal, using native Bootstrap Modal when available, * falling back to manual DOM manipulation otherwise. @@ -291,34 +302,59 @@ function initializeTableCheckboxes(tableId) { const table = document.getElementById(tableId); if (!table) return; + // Query the CURRENT checkboxes live inside every handler instead of closing over a snapshot. + // The master initializer re-runs on each htmx:afterSwap, but the dataset guards below keep the + // toggle/shift handlers from re-binding on a SURVIVING toggle; a NodeList captured once + // would then go stale, so select-all / shift-range would iterate detached checkboxes and miss + // the rows a later row-level swap injected. + const liveCheckboxes = () => Array.from(table.querySelectorAll('td input[name="select"]')); const toggleAll = table.querySelector('th input.toggle'); - const checkboxes = table.querySelectorAll('td input[name="select"]'); - let lastChecked = null; + // Persist the shift-range anchor on the TABLE element, not in a per-call closure. This + // initializer re-runs on every htmx:afterSwap: checkboxes bound in an earlier run keep their + // handlers (the dataset guard skips re-binding), so a closure-scoped anchor would leave old + // rows referencing a stale `lastChecked` while rows added by a later swap use a fresh one — + // shift-clicking between the two then uses disconnected anchors and selects nothing. One + // anchor on the shared table node keeps every row's handler in sync across swaps. + const getAnchor = () => table._lnmsLastChecked || null; + const setAnchor = (cb) => { + table._lnmsLastChecked = cb; + }; - if (toggleAll) { + // Guard against stacked handlers: register each listener at most once per element (a dataset + // flag marks it done) since the master initializer re-runs on every htmx:afterSwap. + if (toggleAll && toggleAll.dataset.tableToggleInitialized !== 'true') { + toggleAll.dataset.tableToggleInitialized = 'true'; toggleAll.addEventListener('change', function () { - checkboxes.forEach(checkbox => { + liveCheckboxes().forEach(checkbox => { checkbox.checked = toggleAll.checked; }); }); } - checkboxes.forEach(checkbox => { + liveCheckboxes().forEach(checkbox => { + if (checkbox.dataset.tableClickInitialized === 'true') return; + checkbox.dataset.tableClickInitialized = 'true'; checkbox.addEventListener('click', function (e) { - if (!lastChecked) { - lastChecked = checkbox; + const anchor = getAnchor(); + if (!anchor) { + setAnchor(checkbox); return; } if (e.shiftKey) { - const start = Array.from(checkboxes).indexOf(checkbox); - const end = Array.from(checkboxes).indexOf(lastChecked); - Array.from(checkboxes).slice(Math.min(start, end), Math.max(start, end) + 1).forEach(cb => { - cb.checked = lastChecked.checked; - }); + const current = liveCheckboxes(); + const start = current.indexOf(checkbox); + const end = current.indexOf(anchor); + // Skip the range when the prior anchor was swapped out (indexOf -1) rather than + // slicing a bogus range off the live list. + if (start !== -1 && end !== -1) { + current.slice(Math.min(start, end), Math.max(start, end) + 1).forEach(cb => { + cb.checked = anchor.checked; + }); + } } - lastChecked = checkbox; + setAnchor(checkbox); }); }); } @@ -589,11 +625,17 @@ function verifyVlanInGroup(select, deviceId, vid, vlanType, groupId) { const modal = document.getElementById('vlanDetailModal'); const capturedSafeName = modal?.dataset.currentSafeName; + const csrfToken = getCsrfToken(); + if (!csrfToken) { + _vlanVerifyEnd(saveBtn); // don't leave the Save button stuck disabled + return; + } + fetch('/plugins/librenms_plugin/verify-vlan-group/', { method: 'POST', headers: { 'Content-Type': 'application/json', - 'X-CSRFToken': document.querySelector('[name=csrfmiddlewaretoken]').value + 'X-CSRFToken': csrfToken }, body: JSON.stringify({ device_id: deviceId, @@ -774,11 +816,24 @@ function initializeVlanModalSave() { // Persist overrides in server cache so other table pages pick them up if (applyToAll && Object.keys(vidGroupMap).length > 0) { const deviceId = modalEl.dataset.currentDeviceId; + const csrfToken = getCsrfToken(); + if (!csrfToken) { + // Can't persist without a CSRF token; surface it via the same error UI the + // fetch .catch uses rather than letting a `.value`-on-null TypeError abort silently. + let alertEl = modalEl.querySelector('.vlan-override-error'); + if (!alertEl) { + alertEl = document.createElement('div'); + alertEl.className = 'vlan-override-error alert alert-danger mt-2'; + modalEl.querySelector('.modal-body')?.appendChild(alertEl); + } + alertEl.textContent = 'Failed to save VLAN group overrides: CSRF token not found.'; + return; + } fetch('/plugins/librenms_plugin/save-vlan-group-overrides/', { method: 'POST', headers: { 'Content-Type': 'application/json', - 'X-CSRFToken': document.querySelector('[name=csrfmiddlewaretoken]').value + 'X-CSRFToken': csrfToken }, body: JSON.stringify({ device_id: deviceId, @@ -916,11 +971,14 @@ function handleVRFChange(select, value) { } const deviceId = deviceInfo.id; + const csrfToken = getCsrfToken(); + if (!csrfToken) return; // missing token → abort rather than throw on `.value` + fetch('/plugins/librenms_plugin/verify-ipaddress/', { method: 'POST', headers: { 'Content-Type': 'application/json', - 'X-CSRFToken': document.querySelector('[name=csrfmiddlewaretoken]').value + 'X-CSRFToken': csrfToken }, body: JSON.stringify({ device_id: deviceId, @@ -958,11 +1016,14 @@ function handleVRFChange(select, value) { * @param {string} value - Selected device ID */ function handleInterfaceChange(select, value) { + const csrfToken = getCsrfToken(); + if (!csrfToken) return; // missing token → abort rather than throw on `.value` + fetch('/plugins/librenms_plugin/verify-interface/', { method: 'POST', headers: { 'Content-Type': 'application/json', - 'X-CSRFToken': document.querySelector('[name=csrfmiddlewaretoken]').value + 'X-CSRFToken': csrfToken }, body: JSON.stringify({ device_id: value, @@ -1004,11 +1065,14 @@ function handleInterfaceChange(select, value) { * @param {string} value - Selected device ID */ function handleCableChange(select, value) { + const csrfToken = getCsrfToken(); + if (!csrfToken) return; // missing token → abort rather than throw on `.value` + fetch('/plugins/librenms_plugin/verify-cable/', { method: 'POST', headers: { 'Content-Type': 'application/json', - 'X-CSRFToken': document.querySelector('[name=csrfmiddlewaretoken]').value + 'X-CSRFToken': csrfToken }, body: JSON.stringify({ device_id: value, @@ -1058,11 +1122,14 @@ function handleModuleChange(select, value) { const controller = new AbortController(); select._moduleVerifyController = controller; + const csrfToken = getCsrfToken(); + if (!csrfToken) return; // missing token → abort rather than throw on `.value` + fetch('/plugins/librenms_plugin/verify-module/', { method: 'POST', headers: { 'Content-Type': 'application/json', - 'X-CSRFToken': document.querySelector('[name=csrfmiddlewaretoken]').value + 'X-CSRFToken': csrfToken }, body: JSON.stringify({ device_id: value, @@ -1159,15 +1226,22 @@ function initializeBulkEditApply() { function initializeCheckboxListeners() { const interfaceTable = document.getElementById('librenms-interface-table'); if (!interfaceTable) return; - const checkboxes = interfaceTable.querySelectorAll('input[name="select"]'); - checkboxes.forEach(checkbox => { + // Query live inside the handlers: the bulkToggle guard below keeps the toggle handler from + // re-binding on a surviving toggle across htmx:afterSwap, so a captured NodeList would + // go stale and select-all would skip rows added by later row-level swaps. + const liveCheckboxes = () => interfaceTable.querySelectorAll('input[name="select"]'); + // Idempotent across htmx:afterSwap re-runs — register the change handler once per checkbox. + liveCheckboxes().forEach(checkbox => { + if (checkbox.dataset.bulkChangeInitialized === 'true') return; + checkbox.dataset.bulkChangeInitialized = 'true'; checkbox.addEventListener('change', updateBulkActionButton); }); const toggleAll = interfaceTable.querySelector('input.toggle'); - if (toggleAll) { + if (toggleAll && toggleAll.dataset.bulkToggleInitialized !== 'true') { + toggleAll.dataset.bulkToggleInitialized = 'true'; toggleAll.addEventListener('change', function () { - checkboxes.forEach(checkbox => { + liveCheckboxes().forEach(checkbox => { checkbox.checked = toggleAll.checked; }); updateBulkActionButton(); @@ -1629,6 +1703,14 @@ function initializeSyncFormSpinners() { * The form is separate from the table (to avoid nested forms), so we copy the * selected checkbox values into hidden inputs just before the form is submitted. * Guard against duplicate listeners on repeated HTMX swaps via a data attribute. + * + * NOTE: this submit-phase injection only reliably covers the NATIVE (no-htmx) submit + * fallback. When htmx drives the POST, its own submit listener can be registered on the + * form BEFORE this one (fresh page load: htmx's DOMContentLoaded processNode runs before + * initializeScripts), so it serializes the form first and these hidden inputs arrive too + * late. The htmx path is therefore injected at htmx:configRequest (see the + * DOMContentLoaded handler), which fires after serialization and replaces any + * select/device_selection values this handler managed to add. */ function handleInstallSelectedSubmit() { // Remove any previously-injected hidden inputs to avoid duplicates @@ -1943,6 +2025,33 @@ document.addEventListener('DOMContentLoaded', function () { if (csrfToken) { event.detail.headers['X-CSRFToken'] = csrfToken.value; } + // Install Selected: the checked rows live in the table OUTSIDE the form, and htmx's + // own submit listener (attached to the form at ITS DOMContentLoaded processNode, + // which on a fresh page load runs before initializeScripts registers the + // submit-phase injector on the same element — listener ORDER, not event phase, + // decides) serializes the form BEFORE the hidden inputs are injected. The first + // click after a full page load then POSTs no 'select' values and the view warns + // "No modules selected." while wiping the selection. configRequest fires AFTER + // htmx serialization, exactly to let listeners amend the outgoing parameters, so + // injecting here is ordering-independent. Replace (not append to) any + // select/device_selection values the submit-phase injector already serialized so + // rows are never posted twice. + if (event.detail.elt && event.detail.elt.id === 'install-selected-form') { + const params = event.detail.parameters; + Array.from(params.keys()) + .filter((k) => k === 'select' || k.startsWith('device_selection_')) + .forEach((k) => params.delete(k)); + const table = document.getElementById('librenms-module-table'); + if (table) { + table.querySelectorAll('input[name="select"]:checked').forEach((cb) => { + params.append('select', cb.value); + const selectedDevice = table.querySelector(`#device_selection_${cb.value}`); + if (selectedDevice) { + params.append(`device_selection_${cb.value}`, selectedDevice.value); + } + }); + } + } }); }); diff --git a/netbox_librenms_plugin/tables/cables.py b/netbox_librenms_plugin/tables/cables.py index cad4660b34..5a2d13b0be 100644 --- a/netbox_librenms_plugin/tables/cables.py +++ b/netbox_librenms_plugin/tables/cables.py @@ -1,3 +1,5 @@ +import re + import django_tables2 as tables from django.utils.html import escape, format_html from django.utils.safestring import mark_safe @@ -6,7 +8,6 @@ from netbox_librenms_plugin.utils import ( get_table_paginate_count, - get_virtual_chassis_member, ) @@ -118,17 +119,43 @@ class VCCableTable(LibreNMSCableTable): def __init__(self, *args, device=None, **kwargs): """Initialize the VC cable table with device context.""" super().__init__(*args, device=device, **kwargs) + # Cache the VC member set once so render_device_selection doesn't re-query + # members.all() (and a members.get per row via get_virtual_chassis_member) for every + # row in large cable tables. Mirrors VCModuleTable. + self._vc_members = [] + self._vc_member_by_position = {} + if getattr(self.device, "virtual_chassis", None): + self._vc_members = list(self.device.virtual_chassis.members.all()) + self._vc_member_by_position = {m.vc_position: m for m in self._vc_members} + + def _selected_member_id(self, port_name): + """ + Resolve the selected VC member id from the port name. + + Served from the cached member set, mirroring get_virtual_chassis_member's position + parse but without a per-row members.get() query. + + Args: + port_name: The LibreNMS local port name (e.g. ``Ethernet3``). + + Returns: + int: The matched member's id, or the table device's id when no member matches. + """ + match = re.match(r"^[A-Za-z]+(\d+)", port_name or "") + if match: + member = self._vc_member_by_position.get(int(match.group(1))) + if member is not None: + return member.id + return self.device.id def render_device_selection(self, value, record): """Render a dropdown to select the virtual chassis member for a port.""" - members = self.device.virtual_chassis.members.all() - chassis_member = get_virtual_chassis_member(self.device, record["local_port"]) - selected_member_id = chassis_member.id if chassis_member else self.device.id + selected_member_id = self._selected_member_id(record["local_port"]) port_id = record["local_port_id"] options = [ f'' - for member in members + for member in self._vc_members ] return format_html( diff --git a/netbox_librenms_plugin/tables/interfaces.py b/netbox_librenms_plugin/tables/interfaces.py index d99d2a7bfe..34dbc65794 100644 --- a/netbox_librenms_plugin/tables/interfaces.py +++ b/netbox_librenms_plugin/tables/interfaces.py @@ -52,7 +52,12 @@ def __init__(self, *args, device=None, interface_name_field=None, vlan_groups=No 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 + # Default the key so render_librenms_id's get_librenms_device_id(self.server_key) lookup + # falls back to the "default" server entry; a None key would miss {"default": 42} values. + self.server_key = server_key or "default" + # Lazily-built {(librenms_type, librenms_speed): mapping} cache so render_type doesn't run + # 1-2 InterfaceTypeMapping queries for every interface row (the table is small and static). + self._interface_type_mapping_cache = None # Update column accessors after initialization for column in ["selection", "name"]: @@ -178,23 +183,31 @@ def render_vlans(self, value, record): else: css = get_tagged_vlan_css_class(vid, netbox_tagged_vids, exists_in_netbox, missing_vlans, group_matches) warning = get_missing_vlan_warning(vid, missing_vlans) - inline_parts.append(f'{vid}({vlan_type}){warning}') + # Escape the LibreNMS-sourced vid/vlan_type (XSS, issue #105 class). css is an + # internal class name; warning is the static icon HTML from get_missing_vlan_warning, + # so it is marked safe rather than escaped. + inline_parts.append( + format_html('{}({}){}', css, vid, vlan_type, mark_safe(warning)) + ) - summary = ", ".join(inline_parts) + # inline_parts are already escaped SafeStrings; join them and keep the result safe. + summary = mark_safe(", ".join(str(part) for part in inline_parts)) if len(all_vlans) > MAX_INLINE: extra = len(all_vlans) - MAX_INLINE - summary += f' +{extra} more' + summary = format_html('{} +{} more', summary, extra) - # Build tooltip showing auto-selected VLAN group per VLAN + # Build tooltip showing auto-selected VLAN group per VLAN. Escape the LibreNMS-sourced + # vid/vlan_type and group_name; the " " separator is a literal newline entity for the + # title attribute, so join the escaped lines and mark the whole tooltip safe. tooltip_lines = [] for vlan_type, vid in all_vlans: if vid in missing_vlans: - tooltip_lines.append(f"VLAN {vid}({vlan_type}) → ⚠ Not in NetBox") + tooltip_lines.append(format_html("VLAN {}({}) → ⚠ Not in NetBox", vid, vlan_type)) else: group_info = vlan_group_map.get(vid, {}) group_name = group_info.get("group_name", "Global") - tooltip_lines.append(f"VLAN {vid}({vlan_type}) → {escape(group_name)}") - tooltip_text = " ".join(tooltip_lines) + tooltip_lines.append(format_html("VLAN {}({}) → {}", vid, vlan_type, group_name)) + tooltip_text = mark_safe(" ".join(str(line) for line in tooltip_lines)) # Build hidden inputs for per-VLAN group selections (submitted with form) hidden_inputs = [] @@ -282,8 +295,8 @@ def render_vlans(self, value, record): return format_html( '{}{}{}', - mark_safe(tooltip_text), - mark_safe(summary), + tooltip_text, + summary, edit_btn, hidden_inputs_html, ) @@ -355,29 +368,31 @@ def render_mtu(self, value, record): def render_librenms_id(self, value, record): """Render the 'librenms_id' field with appropriate styling based on comparison with NetBox.""" + # Same XSS guard as _render_field: value/netbox_librenms_id originate outside NetBox, so + # use format_html to auto-escape both the body and the title attribute (issue #105). if not record.get("exists_in_netbox"): - return mark_safe(f'{value}') + return format_html('{}', value) netbox_interface = record.get("netbox_interface") if not netbox_interface: - return mark_safe(f'{value}') + return format_html('{}', value) netbox_librenms_id = get_librenms_device_id(netbox_interface, self.server_key, auto_save=False) if netbox_librenms_id is None: - return mark_safe( - f'{value}' + return format_html( + '{}', value ) # Compare the IDs if str(value) != str(netbox_librenms_id): # IDs do not match - return mark_safe( - f'{value}' + return format_html( + '{}', netbox_librenms_id, value ) else: # IDs match - return mark_safe(f'{value}') + return format_html('{}', value) def _compare_mac_addresses(self, librenms_mac, netbox_interface): """ @@ -399,17 +414,20 @@ def _compare_mac_addresses(self, librenms_mac, netbox_interface): def _render_field(self, value, record, librenms_key, netbox_key): """Render a field value with appropriate styling based on the comparison with NetBox.""" + # value is an untrusted LibreNMS field (ifName, description, MAC, …). Use format_html so + # it is auto-escaped — a device reporting e.g. ifName="" must + # not render as live HTML (stored XSS, issue #105). The class names stay literal. if not record.get("exists_in_netbox"): - return mark_safe(f'{value}') + return format_html('{}', value) netbox_interface = record.get("netbox_interface") if not netbox_interface: - return mark_safe(f'{value}') + return format_html('{}', value) if librenms_key == "ifPhysAddress": mac_matches = self._compare_mac_addresses(value, netbox_interface) css_class = "text-success" if mac_matches else "text-warning" - return mark_safe(f'{value}') + return format_html('{}', css_class, value) netbox_value = getattr(netbox_interface, netbox_key, None) librenms_value = record.get(librenms_key) @@ -418,9 +436,9 @@ def _render_field(self, value, record, librenms_key, netbox_key): librenms_value = convert_speed_to_kbps(librenms_value) if librenms_value != netbox_value: - return mark_safe(f'{value}') + return format_html('{}', value) - return mark_safe(f'{value}') + return format_html('{}', value) def render_type(self, value, record): """Render interface type with appropriate styling based on comparison with NetBox""" @@ -445,18 +463,23 @@ def render_type(self, value, record): return format_html('{}', combined_display) def get_interface_mapping(self, librenms_type, speed): - """Get interface type mapping based on type and speed""" - - # First try exact match with type and speed - mapping = InterfaceTypeMapping.objects.filter(librenms_type=librenms_type, librenms_speed=speed).first() - - # If no match found, fall back to type-only match - if not mapping: - mapping = InterfaceTypeMapping.objects.filter( - librenms_type=librenms_type, librenms_speed__isnull=True - ).first() + """Get interface type mapping based on type and speed. - return mapping + Resolves from a single in-memory snapshot of the (small, static) + InterfaceTypeMapping table, built on first use, so a table render doesn't + issue 1-2 queries per interface row. + """ + if getattr(self, "_interface_type_mapping_cache", None) is None: + cache = {} + # Keep the FIRST mapping per key to match the previous .filter().first() semantics. + for m in InterfaceTypeMapping.objects.all(): + cache.setdefault((m.librenms_type, m.librenms_speed), m) + self._interface_type_mapping_cache = cache + + # Exact (type, speed) match, then the type-only (speed is NULL) fallback. + return self._interface_type_mapping_cache.get((librenms_type, speed)) or self._interface_type_mapping_cache.get( + (librenms_type, None) + ) def render_mapping_tooltip(self, value, speed, mapping): """Render tooltip for interface type mapping""" diff --git a/netbox_librenms_plugin/tables/modules.py b/netbox_librenms_plugin/tables/modules.py index b3f7f72f80..318c73b866 100644 --- a/netbox_librenms_plugin/tables/modules.py +++ b/netbox_librenms_plugin/tables/modules.py @@ -81,6 +81,40 @@ def __init__( self.can_add_module_bay_mapping = can_add_module_bay_mapping self.can_add_module_type_mapping = can_add_module_type_mapping super().__init__(*args, **kwargs) + # Batch-load the installed modules (with module_type + interface templates) referenced by + # the rows, so render_actions' VC "Report VC issue" diagnostic doesn't run a per-row + # Module.objects.get() + interfacetemplates.all() — an N+1 over the whole module table. + # Only VC devices reach that branch, so skip the query otherwise. + self._installed_modules_by_id = {} + if isinstance(getattr(self.device, "virtual_chassis_id", None), int): + data_rows = args[0] if args else kwargs.get("data") or [] + installed_ids = { + row["installed_module_id"] + for row in data_rows + if isinstance(row, dict) and row.get("installed_module_id") + } + if installed_ids: + from dcim.models import Module + from django.db import DatabaseError + + try: + self._installed_modules_by_id = { + m.pk: m + for m in Module.objects.select_related( + "module_type", + "module_type__manufacturer", + "module_bay", + "device", + "device__device_type", + "device__virtual_chassis", + ) + .prefetch_related("module_type__interfacetemplates") + .filter(pk__in=installed_ids) + } + except (DatabaseError, RuntimeError): + # RuntimeError: pytest "Database access not allowed" in unit-test contexts + # whose self.device is a MagicMock with a real-looking virtual_chassis_id. + self._installed_modules_by_id = {} if not (has_write_permission and can_add_module) and hasattr(self, "columns"): self.columns["selection"].column.visible = False self.tab = "modules" @@ -227,6 +261,42 @@ def render_module_type(self, value, record): return format_html("{}", value) def render_status(self, value, record): + """ + Render the sync-status badge alongside a hidden in-flight spinner badge. + + The live badge is wrapped so CSS (see ``_module_sync.html``) can swap it + for the spinner badge while a row-action POST is in flight — the row forms + set ``hx-indicator="closest tr"``, so HTMX marks the row with ``htmx-request`` + for the duration. The spinner label tracks the row's action: "Updating…" on an + installed-module row offering Update Serial / Update Interface, "Installing…" + on an install / install-branch / carrier-install row. It stays hidden in every + other state, including the inline verify-endpoint cell updates. + """ + # An update action (Update Serial / Update Interface) acts on an already-installed module + # and a row never offers it alongside an install-flavoured action, so a row with an update + # flag and no install/branch/carrier action shows "Updating…"; everything else "Installing…". + in_flight_label = ( + "Updating" + if ( + record.get("installed_module_id") + and (record.get("can_update_serial") or record.get("can_update_interface_binding")) + and not record.get("can_install") + and not record.get("has_installable_children") + and not record.get("carrier_install_options") + ) + else "Installing" + ) + badge = self._status_badge_html(value, record) + return format_html( + '{}' + '' + '{}…', + badge, + in_flight_label, + in_flight_label, + ) + + def _status_badge_html(self, value, record): """Render sync status with badge.""" # Promote No Bay → Missing Carrier when concrete carrier-install rules # produced suggestions for this row (one-click install offered below). @@ -429,7 +499,11 @@ def render_actions(self, value, record): url = reverse("plugins:netbox_librenms_plugin:install_module", kwargs={"pk": self.device.pk}) buttons.append( format_html( - '
' + # hx-post swaps just the module table in place (the view returns the + # table partial for HTMX); method/action keep it working without JS. + '' '' '' '' @@ -446,6 +520,7 @@ def render_actions(self, value, record): ' Install' "
", url, + url, self.csrf_token, self.server_key, record.get("selected_device_id") or self.device.pk, @@ -466,7 +541,11 @@ def render_actions(self, value, record): url = reverse("plugins:netbox_librenms_plugin:install_branch", kwargs={"pk": self.device.pk}) buttons.append( format_html( - '
' + # hx-post swaps just the module table in place (the view returns the + # table partial for HTMX); method/action keep it working without JS. + '' '' '' '' @@ -476,6 +555,7 @@ def render_actions(self, value, record): ' Install Branch' "
", url, + url, self.csrf_token, self.server_key, record.get("selected_device_id") or self.device.pk, @@ -488,7 +568,11 @@ def render_actions(self, value, record): url = reverse("plugins:netbox_librenms_plugin:update_module_serial", kwargs={"pk": self.device.pk}) buttons.append( format_html( - '
' + # hx-post swaps just the module table in place (the view returns the + # table partial for HTMX); method/action keep it working without JS. + '' '' '' '' @@ -499,6 +583,7 @@ def render_actions(self, value, record): ' Update Serial' "
", url, + url, self.csrf_token, self.server_key, record.get("selected_device_id") or self.device.pk, @@ -515,7 +600,11 @@ def render_actions(self, value, record): url = reverse("plugins:netbox_librenms_plugin:update_module_interface", kwargs={"pk": self.device.pk}) buttons.append( format_html( - '
' + # hx-post swaps just the module table in place (the view returns the + # table partial for HTMX); method/action keep it working without JS. + '' '' '' '' @@ -531,6 +620,7 @@ def render_actions(self, value, record): ' Update Interface' "
", url, + url, self.csrf_token, self.server_key, record.get("selected_device_id") or self.device.pk, @@ -613,7 +703,11 @@ def render_actions(self, value, record): for opt in record["carrier_install_options"]: buttons.append( format_html( - '
' + # hx-post swaps just the module table in place (install_module returns the + # table partial for HTMX); method/action keep it working without JS. + '' '' '' '' @@ -625,6 +719,7 @@ def render_actions(self, value, record): ' Install {} into '{}'' "
", install_url, + install_url, self.csrf_token, self.server_key, record.get("selected_device_id") or self.device.pk, @@ -791,26 +886,12 @@ def render_actions(self, value, record): # `virtual_chassis_id` is an int in production (or None); MagicMock-only tests # see a MagicMock here, which the isinstance check correctly skips. if record.get("installed_module_id") and isinstance(getattr(self.device, "virtual_chassis_id", None), int): - from dcim.models import Module - from django.db import DatabaseError - from netbox_librenms_plugin.utils import detect_vc_normalization_noop - installed_module = None - try: - installed_module = Module.objects.select_related( - "module_type", - "module_type__manufacturer", - "module_bay", - "device", - "device__device_type", - "device__virtual_chassis", - ).get(pk=record["installed_module_id"]) - except (Module.DoesNotExist, DatabaseError, RuntimeError): - # RuntimeError: pytest's "Database access not allowed" in unit-test - # contexts that supply self.device as a MagicMock with a real-looking - # virtual_chassis_id. Production rows wouldn't reach here. - installed_module = None + # Served from the __init__ batch prefetch (with interface templates), so this + # diagnostic adds no per-row query. A missing id (deleted concurrently, or a + # unit-test MagicMock device that skipped the prefetch) yields None → no button. + installed_module = self._installed_modules_by_id.get(record["installed_module_id"]) if installed_module is not None and detect_vc_normalization_noop(installed_module.device, installed_module): report_url = reverse( diff --git a/netbox_librenms_plugin/templates/netbox_librenms_plugin/_interface_sync.html b/netbox_librenms_plugin/templates/netbox_librenms_plugin/_interface_sync.html index fa1d3c0d73..7bb4aeca25 100644 --- a/netbox_librenms_plugin/templates/netbox_librenms_plugin/_interface_sync.html +++ b/netbox_librenms_plugin/templates/netbox_librenms_plugin/_interface_sync.html @@ -7,19 +7,24 @@

Interface Sync

{% csrf_token %} + {# Carry the active server_key so the refresh hits the right LibreNMS server/cache (else a non-default server tab uses the fallback). #} + {% if server_key %}{% endif %} {% if has_librenms_id %} {% with model_name=object|meta:"model_name" %} {% if model_name == "device" %} - {% elif model_name == "virtualmachine" %} - 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 c8a51bdcad..5620026c9b 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 @@ -305,7 +305,8 @@

Module Sync

+ +
{% include 'netbox_librenms_plugin/_module_sync_content.html' %} diff --git a/netbox_librenms_plugin/templates/netbox_librenms_plugin/_module_sync_content.html b/netbox_librenms_plugin/templates/netbox_librenms_plugin/_module_sync_content.html index eccff430e8..5b2c8d0776 100644 --- a/netbox_librenms_plugin/templates/netbox_librenms_plugin/_module_sync_content.html +++ b/netbox_librenms_plugin/templates/netbox_librenms_plugin/_module_sync_content.html @@ -17,14 +17,17 @@
{% if has_write_permission %} -{# Separate form for Install Selected — uses JS to collect checked rows before submit #} +{# Install Selected: checked rows are injected into the HTMX POST at htmx:configRequest (ordering-independent — htmx's own submit listener serializes first on a fresh load); the submit-phase injector covers only the native no-JS fallback. hx-post swaps the table in place. #} + action="{% url 'plugins:netbox_librenms_plugin:install_selected' pk=module_sync.object.pk %}" + hx-post="{% url 'plugins:netbox_librenms_plugin:install_selected' pk=module_sync.object.pk %}" + hx-target="#module-sync-content" hx-swap="innerHTML" hx-disabled-elt="find button"> {% csrf_token %}
diff --git a/netbox_librenms_plugin/templates/netbox_librenms_plugin/htmx/bulk_import_confirm.html b/netbox_librenms_plugin/templates/netbox_librenms_plugin/htmx/bulk_import_confirm.html index a83ff7c0ff..a6a473a858 100644 --- a/netbox_librenms_plugin/templates/netbox_librenms_plugin/htmx/bulk_import_confirm.html +++ b/netbox_librenms_plugin/templates/netbox_librenms_plugin/htmx/bulk_import_confirm.html @@ -15,6 +15,15 @@
{% endif %} + {% if cache_expired_count %} + + {% endif %} + {% for entry in devices %} {% with vc=entry.validation.virtual_chassis %}
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 7cedbc8c67..b9e2c3837a 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 @@ -497,6 +497,8 @@
hx-swap="none" hx-include="#use-sysname-toggle, #strip-domain-toggle"> {% csrf_token %} + {# server_key keeps the action scoped to the active LibreNMS server (issue #106). #} + {% if validation.device_type_mismatch %} @@ -546,6 +548,8 @@
hx-swap="none" hx-include="#use-sysname-toggle, #strip-domain-toggle"> {% csrf_token %} + {# server_key keeps the action scoped to the active LibreNMS server (issue #106). #} + {% if validation.device_type_mismatch %} diff --git a/netbox_librenms_plugin/templates/netbox_librenms_plugin/htmx/module_mismatch_modal.html b/netbox_librenms_plugin/templates/netbox_librenms_plugin/htmx/module_mismatch_modal.html index 98dc0d013a..1f7206aa59 100644 --- a/netbox_librenms_plugin/templates/netbox_librenms_plugin/htmx/module_mismatch_modal.html +++ b/netbox_librenms_plugin/templates/netbox_librenms_plugin/htmx/module_mismatch_modal.html @@ -24,7 +24,10 @@ {{ installed_module.module_type.model }} {% else %}-{% endif %} - {{ librenms_model }} + + {{ librenms_model }} + {% if type_matched %}{% endif %} + Serial @@ -40,10 +43,17 @@ Different module type — the installed module type does not match LibreNMS. Replacing will delete the current module and install {{ librenms_model }}.
-{% elif serial_mismatch %} +{% elif serial_mismatch and type_matched %}
- - Same module type, different serial — the module may have been physically replaced. + + Same module type — the LibreNMS model is recognised as this module type (see the badge above). + Only the serial differs, so the module may have been physically replaced. +
+{% elif serial_mismatch %} +
+ + The LibreNMS model isn't recognised as a NetBox module type, so only the serial can be compared — + it differs from the installed module.
{% endif %} diff --git a/netbox_librenms_plugin/tests/conftest.py b/netbox_librenms_plugin/tests/conftest.py index a6bba487f7..05ca8062fd 100644 --- a/netbox_librenms_plugin/tests/conftest.py +++ b/netbox_librenms_plugin/tests/conftest.py @@ -4,6 +4,25 @@ import pytest + +@pytest.fixture(autouse=True) +def _clear_device_info_cache(): + """Clear get_device_info()'s short-lived cache between tests. + + get_device_info() caches successful lookups in the shared cache. Without this, + a cached success from one test leaks into another that reuses the same + (server_key, device_id) but mocks a different response — e.g. the failure-path + tests keyed on device_id=123 that run after test_get_device_info_success. + """ + from django.core.cache import cache + + try: + cache.delete_pattern("librenms_device_info_*") + except (AttributeError, NotImplementedError): + cache.clear() + yield + + # ============================================================================= # Configuration Fixtures # ============================================================================= diff --git a/netbox_librenms_plugin/tests/test_cable_verify.py b/netbox_librenms_plugin/tests/test_cable_verify.py index 36aaf3952d..0c02515dce 100644 --- a/netbox_librenms_plugin/tests/test_cable_verify.py +++ b/netbox_librenms_plugin/tests/test_cable_verify.py @@ -1,311 +1,382 @@ """ Regression tests for SingleCableVerifyView.post(). +Fully DB-backed: the verify path re-enriches purely from live NetBox state +(get_device_by_id_or_name, enrich_remote_port and check_cable_status are ORM lookups — the +LibreNMS client is never touched once the links cache is warm), so these exercise real +Device/Interface/Cable objects, a real links cache entry and a real object-scoped request end to +end. The only patched boundary is get_available_servers() — i.e. which LibreNMS servers are +configured — so the POSTed server_key resolves deterministically. + 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. +- Stale derived fields (ids/URLs from a previous enrichment) in the cached link are stripped and + recomputed from current NetBox state. +- LibreNMS-sourced labels (local_port, remote_device) are HTML-escaped to prevent XSS. """ import json from unittest.mock import MagicMock, patch +import pytest +from django.core.cache import cache as real_cache +from django.test import RequestFactory -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 +SERVER_KEY = "default" -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 +def _cable_device(tag, ifaces): + """Create a real Device named *tag* with named interfaces. + ``ifaces`` is a list of ``(name, librenms_port_id)`` — a non-None port id binds the interface's + ``librenms_id`` custom field for that server so the view's ``_librenms_id_q`` lookup resolves it. + Returns ``(device, {name: interface})``. + """ + from dcim.models import Device, DeviceRole, DeviceType, Interface, Manufacturer, Site -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, - } + mfr, _ = Manufacturer.objects.get_or_create(name=f"CblMfr-{tag}", slug=f"cblmfr-{tag}") + dt, _ = DeviceType.objects.get_or_create(manufacturer=mfr, model=f"CblDT-{tag}", slug=f"cbldt-{tag}") + role, _ = DeviceRole.objects.get_or_create(name="CblRole", slug="cblrole") + site, _ = Site.objects.get_or_create(name="CblSite", slug="cblsite") + device = Device.objects.create(name=tag, device_type=dt, role=role, site=site, status="active") + made = {} + for name, port_id in ifaces: + iface = Interface.objects.create(device=device, name=name, type="1000base-t") + if port_id is not None: + iface.custom_field_data = {"librenms_id": {SERVER_KEY: port_id}} + iface.save() + made[name] = iface + return device, made - 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_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, - } +def _cable_view_and_request(tag, body): + """A real SingleCableVerifyView plus a real superuser request (so the perm gate + restrict resolve).""" + from django.contrib.auth import get_user_model - # Mock process_remote_device to avoid DB access during re-enrichment; - # it should receive the link WITHOUT derived fields. - received_link = {} + from netbox_librenms_plugin.views.base.cables_view import SingleCableVerifyView - def fake_process_remote(link, hostname, device_id, server_key=None): - received_link.update(link) - return link + view = SingleCableVerifyView() + request = RequestFactory().post("/verify-cable/", data=json.dumps(body), content_type="application/json") + request.user = get_user_model().objects.create_superuser(username=f"cbl-{tag}", email="", password="x") + view.request = request + view.kwargs = {} + view.args = () + return view, request + + +def _post_with_links(view, request, device, links): + """Warm the real links cache for *device*, run post() (with 'default' configured), clean up, return JSON.""" + key = view.get_cache_key(device, "links", SERVER_KEY) + real_cache.set(key, {"links": links}) + try: + with patch( + "netbox_librenms_plugin.librenms_api.LibreNMSAPI.get_available_servers", + return_value={SERVER_KEY: "Default"}, + ): + response = view.post(request) + finally: + real_cache.delete(key) + return json.loads(response.content) - 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( +@pytest.mark.django_db +class TestStaleFieldStripping: + """Cached derived fields (URLs/ids from a previous enrichment) must be recomputed from live state.""" + + def test_stale_remote_fields_recomputed_from_live_netbox(self): + """A stale remote_device_url/id in cache is dropped; the row links the REAL remote device + cable.""" + from dcim.models import Cable + + local, lifaces = _cable_device("stale-local", [("eth0", 100)]) + remote, rifaces = _cable_device("switch-remote", [("eth1", 200)]) + cable = Cable(a_terminations=[lifaces["eth0"]], b_terminations=[rifaces["eth1"]]) + cable.save() + + view, request = _cable_view_and_request( + "stale", {"device_id": local.pk, "local_port_id": 100, "server_key": SERVER_KEY} + ) + content = _post_with_links( + view, + request, + local, + [ { - "device_id": 1, + "local_port": "eth0", "local_port_id": 100, - "server_key": "default", + "remote_port": "eth1", + "remote_port_id": 200, + "remote_device": "switch-remote", + "remote_device_id": None, + # Stale derived fields from a previous enrichment (a different remote object then): + "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, } - ) - view.post(request) + ], + ) + + row = content["formatted_row"] + # Remote side recomputed to the REAL device, not the stale 999. + assert f"/dcim/devices/{remote.pk}/" in row["remote_device"] + assert "/dcim/devices/999/" not in row["remote_device"] + # The real cable is found, so the stale "No Cable" is replaced. + assert "Cable Found" in row["cable_status"] + + def test_stale_local_and_status_fields_recomputed(self): + """Stale local interface id / cable_status in cache are stripped; the row uses live values.""" + local, lifaces = _cable_device("stale2-local", [("eth0", 100)]) + + view, request = _cable_view_and_request( + "stale2", {"device_id": local.pk, "local_port_id": 100, "server_key": SERVER_KEY} + ) + content = _post_with_links( + view, + request, + local, + [ + { + "local_port": "eth0", + "local_port_id": 100, + "remote_port": "eth1", + "remote_port_id": 200, + "remote_device": "", # no remote side → no remote enrichment / cable check + "remote_device_id": None, + # Stale derived fields that must not survive the strip-to-raw-keys step: + "netbox_local_interface_id": 999, + "local_port_url": "/stale/", + "cable_status": "stale-status", + "can_create_cable": True, + } + ], + ) - # 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 + row = content["formatted_row"] + # Local port link recomputed to the REAL interface, not the stale id/url. + assert f"/dcim/interfaces/{lifaces['eth0'].pk}/" in row["local_port"] + assert "/stale/" not in row["local_port"] + # cable_status was stripped (not a raw key) → falls back to the default, never the stale value. + assert "stale-status" not in row["cable_status"] + assert row["cable_status"] == "Missing Ports" +@pytest.mark.django_db class TestXSSEscaping: - """LibreNMS-sourced labels must be HTML-escaped in cable verify output.""" + """LibreNMS-sourced labels must be HTML-escaped in cable verify output (real render, real objects).""" 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", "") + local, lifaces = _cable_device("xss-local", [("eth0", 100)]) + xss = '' + + view, request = _cable_view_and_request( + "xsslocal", {"device_id": local.pk, "local_port_id": 100, "server_key": SERVER_KEY} + ) + content = _post_with_links( + view, + request, + local, + [ + { + "local_port": xss, # interface still resolves by librenms_id below + "local_port_id": 100, + "remote_port": "eth1", + "remote_port_id": 200, + "remote_device": "", + "remote_device_id": None, + } + ], + ) - # The raw script tag must NOT appear unescaped + local_port_html = content["formatted_row"]["local_port"] assert "', + device_type=dt, + role=role, + site=site, + status="active", + virtual_chassis=vc, + vc_position=1, + ) + + # The device IS a VC member, so device.virtual_chassis.members.all() includes it and its + # (malicious) name flows into the dropdown options cached in __init__. + table = VCCableTable([], device=member) + html = str(table.render_device_selection(None, {"local_port": "eth0", "local_port_id": "42"})) + + # The raw ' - 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 ' - 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)) - - assert "