fix: develop-era hardening pass (security + correctness + tests)

.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/

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

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,16 +332,18 @@ 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.
                 # Direct access needed to detect legacy format for migration prompt:
                 # LibreNMSAPI.get_librenms_id() returns an int in both formats, so only the
                 # raw type check on custom_field_data reveals whether migration is needed.
                 _dev_cf_id = existing_device.custom_field_data.get("librenms_id")
-                if (isinstance(_dev_cf_id, int) and not isinstance(_dev_cf_id, bool)) or (
-                    isinstance(_dev_cf_id, str) and _dev_cf_id.isdigit()
-                ):
+                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)

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