test(plugins): cover duplicate plugin identifier diagnostics #2168

backend/secuscan/plugins.py

@@ -102,6 +102,7 @@ class PluginManager:
     def __init__(self, plugins_dir: str):
         self.plugins_dir = Path(plugins_dir)
         self.plugins: Dict[str, PluginMetadata] = {}
+        self.plugin_locations: Dict[str, Path] = {}
 
     def _scan_plugin_dirs(self) -> List[Path]:
         """Scan the plugins directory for plugin directories."""
@@ -123,6 +124,8 @@ async def load_plugins(self) -> int:
         Returns:
             Number of successfully loaded plugins
         """
+        self.plugins.clear()
+        self.plugin_locations.clear()
         plugin_dirs = self._scan_plugin_dirs()
         loaded = 0
 
@@ -135,9 +138,19 @@ async def load_plugins(self) -> int:
             try:
                 plugin_meta = await self._load_plugin_metadata(metadata_file)
 
+                # Check for duplicate plugin identifier
+                if plugin_meta.id in self.plugins:
+                    existing_loc = self.plugin_locations.get(plugin_meta.id)
+                    logger.error(
+                        f"Duplicate plugin identifier '{plugin_meta.id}' found in {plugin_dir} "
+                        f"(conflicts with existing plugin at {existing_loc})"
+                    )
+                    continue
+
                 # Validate plugin
                 if await self._validate_plugin(plugin_meta, plugin_dir):
                     self.plugins[plugin_meta.id] = plugin_meta
+                    self.plugin_locations[plugin_meta.id] = plugin_dir
                     loaded += 1
                     logger.info(f"✓ Loaded plugin: {plugin_meta.name} v{plugin_meta.version}")
                 else:

frontend/package.json

@@ -55,6 +55,7 @@
     "react-router": "^6.30.4",
     "dompurify": "^3.4.10",
     "@babel/core": "^7.29.7",
-    "undici": "^8.10.0"
+    "undici": "^8.10.0",
+    "nanoid": "^3.3.18"
   }
 }

testing/backend/unit/test_plugins.py

@@ -1,4 +1,6 @@
 import asyncio
+import json
+import logging
 from pathlib import Path
 
 import pytest
@@ -475,3 +477,60 @@ def test_plugin_build_command_allows_legitimate_targets(setup_test_environment):
     )
     assert command is not None
     assert "https://example.com" in command
+
+
+def test_plugin_loader_duplicate_identifier_diagnostics(tmp_path, caplog):
+    """Loader must log error diagnostics identifying both conflicting plugin locations when duplicate IDs exist."""
+    dir1 = tmp_path / "plugin_alpha"
+    dir1.mkdir()
+    dir2 = tmp_path / "plugin_beta"
+    dir2.mkdir()
+
+    meta1 = {
+        "id": "duplicate_scanner",
+        "name": "Alpha Scanner",
+        "description": "First plugin instance",
+        "version": "1.0.0",
+        "category": "web",
+        "icon": "shield",
+        "engine": {"type": "cli", "binary": "echo"},
+        "command_template": ["echo", "test"],
+        "fields": [],
+        "presets": {},
+        "output": {"parser": "text"},
+        "safety": {"level": "safe"},
+    }
+
+    meta2 = {
+        "id": "duplicate_scanner",
+        "name": "Beta Scanner",
+        "description": "Second plugin instance with duplicate identifier",
+        "version": "2.0.0",
+        "category": "web",
+        "icon": "shield",
+        "engine": {"type": "cli", "binary": "echo"},
+        "command_template": ["echo", "test"],
+        "fields": [],
+        "presets": {},
+        "output": {"parser": "text"},
+        "safety": {"level": "safe"},
+    }
+
+    (dir1 / "metadata.json").write_text(json.dumps(meta1), encoding="utf-8")
+    (dir2 / "metadata.json").write_text(json.dumps(meta2), encoding="utf-8")
+
+    manager = PluginManager(str(tmp_path))
+    with caplog.at_level(logging.ERROR):
+        loaded = asyncio.run(manager.load_plugins())
+
+    # Only 1 plugin should be loaded successfully (the duplicate is skipped)
+    assert loaded == 1
+    loaded_plugin = manager.get_plugin("duplicate_scanner")
+    assert loaded_plugin is not None
+    assert loaded_plugin.name == "Alpha Scanner"
+
+    # Verify that an error log was captured identifying the duplicate ID and BOTH locations
+    error_logs = [rec.message for rec in caplog.records if rec.levelno == logging.ERROR]
+    assert any("duplicate_scanner" in msg for msg in error_logs)
+    assert any(str(dir1) in msg or dir1.name in msg for msg in error_logs)
+    assert any(str(dir2) in msg or dir2.name in msg for msg in error_logs)