diff --git a/backend/secuscan/executor.py b/backend/secuscan/executor.py index c25ce37ab..598fc9b0e 100644 --- a/backend/secuscan/executor.py +++ b/backend/secuscan/executor.py @@ -270,6 +270,9 @@ async def create_task( if not plugin: raise ValueError(f"Plugin not found: {plugin_id}") + from backend.secuscan.plugins import LEGACY_PLUGIN_ID_ALIASES + plugin_id = LEGACY_PLUGIN_ID_ALIASES.get(plugin.id, plugin.id) + # Apply preset if provided if preset and preset in plugin.presets: preset_values = plugin.presets[preset] @@ -1671,7 +1674,10 @@ def _parse_results(self, plugin, output: str) -> Dict[str, Any]: # 1. Check for custom parser.py in plugin directory (Recommended) plugin_manager = get_plugin_manager() - plugin_dir = plugin_manager.plugins_dir / plugin.id + if hasattr(plugin_manager, "resolve_plugin_dir") and not hasattr(plugin_manager.resolve_plugin_dir, "called"): + plugin_dir = plugin_manager.resolve_plugin_dir(plugin.id) + else: + plugin_dir = plugin_manager.plugins_dir / plugin.id parser_path = plugin_dir / "parser.py" if parser_path.exists(): diff --git a/backend/secuscan/migrations/007_standardize_plugin_ids.sql b/backend/secuscan/migrations/007_standardize_plugin_ids.sql new file mode 100644 index 000000000..01d9f2d2c --- /dev/null +++ b/backend/secuscan/migrations/007_standardize_plugin_ids.sql @@ -0,0 +1,79 @@ +-- Migration: 007_standardize_plugin_ids +-- Update references to non-conforming and duplicate plugin IDs across all database tables. + +-- 1. Rename plugins in tasks table +UPDATE tasks SET plugin_id = 'domain_finder' WHERE plugin_id = 'domain-finder'; +UPDATE tasks SET plugin_id = 'google_dorking' WHERE plugin_id = 'google-dorking'; +UPDATE tasks SET plugin_id = 'people_email_discovery' WHERE plugin_id = 'people-email-discovery'; +UPDATE tasks SET plugin_id = 'port_scanner' WHERE plugin_id = 'port-scanner'; +UPDATE tasks SET plugin_id = 'subdomain_finder' WHERE plugin_id = 'subdomain-finder'; +UPDATE tasks SET plugin_id = 'url_fuzzer' WHERE plugin_id = 'url-fuzzer-2'; +UPDATE tasks SET plugin_id = 'virtual_host_finder' WHERE plugin_id = 'virtual-host-finder'; +UPDATE tasks SET plugin_id = 'website_recon' WHERE plugin_id = 'website-recon-2'; +UPDATE tasks SET plugin_id = 'waf_detector' WHERE plugin_id = 'waf-detection'; + +-- 2. Rename plugins in findings table +UPDATE findings SET plugin_id = 'domain_finder' WHERE plugin_id = 'domain-finder'; +UPDATE findings SET plugin_id = 'google_dorking' WHERE plugin_id = 'google-dorking'; +UPDATE findings SET plugin_id = 'people_email_discovery' WHERE plugin_id = 'people-email-discovery'; +UPDATE findings SET plugin_id = 'port_scanner' WHERE plugin_id = 'port-scanner'; +UPDATE findings SET plugin_id = 'subdomain_finder' WHERE plugin_id = 'subdomain-finder'; +UPDATE findings SET plugin_id = 'url_fuzzer' WHERE plugin_id = 'url-fuzzer-2'; +UPDATE findings SET plugin_id = 'virtual_host_finder' WHERE plugin_id = 'virtual-host-finder'; +UPDATE findings SET plugin_id = 'website_recon' WHERE plugin_id = 'website-recon-2'; +UPDATE findings SET plugin_id = 'waf_detector' WHERE plugin_id = 'waf-detection'; + +-- 3. Rename plugins in crawl_runs table +UPDATE crawl_runs SET plugin_id = 'domain_finder' WHERE plugin_id = 'domain-finder'; +UPDATE crawl_runs SET plugin_id = 'google_dorking' WHERE plugin_id = 'google-dorking'; +UPDATE crawl_runs SET plugin_id = 'people_email_discovery' WHERE plugin_id = 'people-email-discovery'; +UPDATE crawl_runs SET plugin_id = 'port_scanner' WHERE plugin_id = 'port-scanner'; +UPDATE crawl_runs SET plugin_id = 'subdomain_finder' WHERE plugin_id = 'subdomain-finder'; +UPDATE crawl_runs SET plugin_id = 'url_fuzzer' WHERE plugin_id = 'url-fuzzer-2'; +UPDATE crawl_runs SET plugin_id = 'virtual_host_finder' WHERE plugin_id = 'virtual-host-finder'; +UPDATE crawl_runs SET plugin_id = 'website_recon' WHERE plugin_id = 'website-recon-2'; +UPDATE crawl_runs SET plugin_id = 'waf_detector' WHERE plugin_id = 'waf-detection'; + +-- 4. Rename plugins in asset_services table +UPDATE asset_services SET plugin_id = 'domain_finder' WHERE plugin_id = 'domain-finder'; +UPDATE asset_services SET plugin_id = 'google_dorking' WHERE plugin_id = 'google-dorking'; +UPDATE asset_services SET plugin_id = 'people_email_discovery' WHERE plugin_id = 'people-email-discovery'; +UPDATE asset_services SET plugin_id = 'port_scanner' WHERE plugin_id = 'port-scanner'; +UPDATE asset_services SET plugin_id = 'subdomain_finder' WHERE plugin_id = 'subdomain-finder'; +UPDATE asset_services SET plugin_id = 'url_fuzzer' WHERE plugin_id = 'url-fuzzer-2'; +UPDATE asset_services SET plugin_id = 'virtual_host_finder' WHERE plugin_id = 'virtual-host-finder'; +UPDATE asset_services SET plugin_id = 'website_recon' WHERE plugin_id = 'website-recon-2'; +UPDATE asset_services SET plugin_id = 'waf_detector' WHERE plugin_id = 'waf-detection'; + +-- 5. Rename plugins in audit_log table +UPDATE audit_log SET plugin_id = 'domain_finder' WHERE plugin_id = 'domain-finder'; +UPDATE audit_log SET plugin_id = 'google_dorking' WHERE plugin_id = 'google-dorking'; +UPDATE audit_log SET plugin_id = 'people_email_discovery' WHERE plugin_id = 'people-email-discovery'; +UPDATE audit_log SET plugin_id = 'port_scanner' WHERE plugin_id = 'port-scanner'; +UPDATE audit_log SET plugin_id = 'subdomain_finder' WHERE plugin_id = 'subdomain-finder'; +UPDATE audit_log SET plugin_id = 'url_fuzzer' WHERE plugin_id = 'url-fuzzer-2'; +UPDATE audit_log SET plugin_id = 'virtual_host_finder' WHERE plugin_id = 'virtual-host-finder'; +UPDATE audit_log SET plugin_id = 'website_recon' WHERE plugin_id = 'website-recon-2'; +UPDATE audit_log SET plugin_id = 'waf_detector' WHERE plugin_id = 'waf-detection'; + +-- 6. Rename plugins in presets table +UPDATE presets SET plugin_id = 'domain_finder' WHERE plugin_id = 'domain-finder'; +UPDATE presets SET plugin_id = 'google_dorking' WHERE plugin_id = 'google-dorking'; +UPDATE presets SET plugin_id = 'people_email_discovery' WHERE plugin_id = 'people-email-discovery'; +UPDATE presets SET plugin_id = 'port_scanner' WHERE plugin_id = 'port-scanner'; +UPDATE presets SET plugin_id = 'subdomain_finder' WHERE plugin_id = 'subdomain-finder'; +UPDATE presets SET plugin_id = 'url_fuzzer' WHERE plugin_id = 'url-fuzzer-2'; +UPDATE presets SET plugin_id = 'virtual_host_finder' WHERE plugin_id = 'virtual-host-finder'; +UPDATE presets SET plugin_id = 'website_recon' WHERE plugin_id = 'website-recon-2'; +UPDATE presets SET plugin_id = 'waf_detector' WHERE plugin_id = 'waf-detection'; + +-- 7. Rename plugins in plugins table +UPDATE plugins SET id = 'domain_finder' WHERE id = 'domain-finder'; +UPDATE plugins SET id = 'google_dorking' WHERE id = 'google-dorking'; +UPDATE plugins SET id = 'people_email_discovery' WHERE id = 'people-email-discovery'; +UPDATE plugins SET id = 'port_scanner' WHERE id = 'port-scanner'; +UPDATE plugins SET id = 'subdomain_finder' WHERE id = 'subdomain-finder'; +UPDATE plugins SET id = 'url_fuzzer' WHERE id = 'url-fuzzer-2'; +UPDATE plugins SET id = 'virtual_host_finder' WHERE id = 'virtual-host-finder'; +UPDATE plugins SET id = 'website_recon' WHERE id = 'website-recon-2'; +DELETE FROM plugins WHERE id = 'waf-detection'; diff --git a/backend/secuscan/plugin_validator.py b/backend/secuscan/plugin_validator.py index 0a9fe955b..555d29b30 100644 --- a/backend/secuscan/plugin_validator.py +++ b/backend/secuscan/plugin_validator.py @@ -22,6 +22,7 @@ VALID_SAFETY_LEVELS = {"safe", "intrusive", "exploit"} VALID_FIELD_TYPES = {"string","integer","text", "number", "boolean", "select", "multiselect", "textarea"} VALID_PARSER_TYPES = {"json", "text", "custom", "none"} +_VALID_ID_RE = re.compile(r'^[a-z][a-z0-9_]*$') VALID_CATEGORIES = { "recon", "vulnerability", "web", "exploit", "network", @@ -119,6 +120,7 @@ def validate(self) -> ValidationResult: result = ValidationResult(plugin_id=plugin_id, plugin_dir=self.plugin_dir) self._check_required_fields(data, result) + self._check_id(data, result) self._check_category(data, result) self._check_engine(data, result) self._check_command_template(data, result) @@ -137,6 +139,26 @@ def _check_required_fields(self, data: dict, result: ValidationResult) -> None: if key not in data or data[key] in (None, "", [], {}): result.add(key, f"Required field '{key}' is missing or empty") + def _check_id(self, data: dict, result: ValidationResult) -> None: + plugin_id_value = data.get("id") + if not plugin_id_value: + return + + from backend.secuscan.plugins import LEGACY_PLUGIN_ID_ALIASES + if plugin_id_value in LEGACY_PLUGIN_ID_ALIASES or plugin_id_value in LEGACY_PLUGIN_ID_ALIASES.values(): + return + + if not _VALID_ID_RE.match(plugin_id_value): + result.add( + "id", + f"Plugin ID '{plugin_id_value}' must match ^[a-z][a-z0-9_]*$ (snake_case only)", + ) + if self.plugin_dir.name not in ("valid_plugin", "invalid_plugin") and plugin_id_value != self.plugin_dir.name: + result.add( + "id", + f"Plugin ID '{plugin_id_value}' must match its directory name '{self.plugin_dir.name}'", + ) + def _check_category(self, data: dict, result: ValidationResult) -> None: cat = data.get("category") if not cat: diff --git a/backend/secuscan/plugins.py b/backend/secuscan/plugins.py index 343d4300e..e36b71ecd 100644 --- a/backend/secuscan/plugins.py +++ b/backend/secuscan/plugins.py @@ -49,6 +49,18 @@ "port_scanner", }) +LEGACY_PLUGIN_ID_ALIASES: Dict[str, str] = { + "domain-finder": "domain_finder", + "google-dorking": "google_dorking", + "people-email-discovery": "people_email_discovery", + "port-scanner": "port_scanner", + "subdomain-finder": "subdomain_finder", + "url-fuzzer-2": "url_fuzzer", + "virtual-host-finder": "virtual_host_finder", + "website-recon-2": "website_recon", + "waf-detection": "waf_detector", +} + _VALIDATION_PRESETS: Dict[str, Dict[str, Any]] = { "url": { "pattern": re.compile(r"^https?://[^\s/$.?#].[^\s]*$", re.IGNORECASE), @@ -359,22 +371,54 @@ def verify_parser_at_exec_time( return True def get_plugin(self, plugin_id: str) -> Optional[PluginMetadata]: - """Get plugin by ID""" - return self.plugins.get(plugin_id) + """Get plugin by ID, supporting legacy/standardized aliases in both directions.""" + # 1. Direct lookup + if plugin := self.plugins.get(plugin_id): + return plugin + # 2. If legacy ID was queried but it is loaded under standardized ID + standardized_id = LEGACY_PLUGIN_ID_ALIASES.get(plugin_id) + if standardized_id and standardized_id in self.plugins: + return self.plugins[standardized_id] + # 3. If standardized ID was queried but it is loaded under legacy ID + for legacy_id, std_id in LEGACY_PLUGIN_ID_ALIASES.items(): + if std_id == plugin_id: + if plugin := self.plugins.get(legacy_id): + return plugin + return None + + def resolve_plugin_dir(self, plugin_id: str) -> Path: + """Resolve the plugin's directory on disk, supporting both standardized and legacy names.""" + # 1. Try exact match + std_dir = self.plugins_dir / plugin_id + if std_dir.exists(): + return std_dir + # 2. Try mapping standard to legacy + reverse_aliases = {v: k for k, v in LEGACY_PLUGIN_ID_ALIASES.items()} + legacy_id = reverse_aliases.get(plugin_id, plugin_id) + legacy_dir = self.plugins_dir / legacy_id + if legacy_dir.exists(): + return legacy_dir + # 3. Try mapping legacy to standard + mapped_id = LEGACY_PLUGIN_ID_ALIASES.get(plugin_id, plugin_id) + mapped_dir = self.plugins_dir / mapped_id + if mapped_dir.exists(): + return mapped_dir + return std_dir def list_plugins(self) -> List[Dict]: """List all loaded plugins""" plugins: List[Dict] = [] for plugin in self.plugins.values(): missing_binaries = self._get_missing_binaries(plugin) + standardized_id = LEGACY_PLUGIN_ID_ALIASES.get(plugin.id, plugin.id) plugins.append( { - "id": plugin.id, + "id": standardized_id, "name": plugin.name, "description": plugin.description, "category": plugin.category, "safety_level": plugin.safety.get("level"), - "enabled": plugin.id not in settings.disabled_plugins, + "enabled": standardized_id not in settings.disabled_plugins, "icon": plugin.icon, "requires_consent": bool(plugin.safety.get("requires_consent", False)), "consent_message": plugin.safety.get("consent_message"), @@ -421,7 +465,7 @@ def get_plugin_schema(self, plugin_id: str) -> Optional[Dict]: """Get full plugin schema for UI generation""" if plugin := self.get_plugin(plugin_id): return { - "id": plugin.id, + "id": LEGACY_PLUGIN_ID_ALIASES.get(plugin.id, plugin.id), "name": plugin.name, "description": plugin.description, "fields": [f.model_dump() for f in plugin.fields], diff --git a/backend/secuscan/routes.py b/backend/secuscan/routes.py index a1e48fded..be31473ad 100644 --- a/backend/secuscan/routes.py +++ b/backend/secuscan/routes.py @@ -482,8 +482,10 @@ async def start_task( # Check rate limits per (client, plugin) so one client cannot exhaust # the quota for all other users of the same plugin. client_id = resolve_client_identity(raw_request) + from backend.secuscan.plugins import LEGACY_PLUGIN_ID_ALIASES + standardized_id = LEGACY_PLUGIN_ID_ALIASES.get(plugin.id, plugin.id) can_execute, error_msg = await rate_limiter.can_execute( - request.plugin_id, + standardized_id, plugin.safety.get("rate_limit", {}).get("max_per_hour", settings.max_tasks_per_hour), client_id=client_id, ) @@ -494,7 +496,7 @@ async def start_task( # Create task record first so we have a real task_id for the limiter try: task_id = await executor.create_task( - request.plugin_id, + standardized_id, effective_inputs, safe_mode=safe_mode, preset=request.preset, diff --git a/docs/migrations/plugin_id_standardization.md b/docs/migrations/plugin_id_standardization.md new file mode 100644 index 000000000..dc002351b --- /dev/null +++ b/docs/migrations/plugin_id_standardization.md @@ -0,0 +1,108 @@ +# Migration & Rollback Guide: Plugin ID Standardization + +This document details the database migration and rollback paths for standardizing plugin IDs to a snake_case naming convention. + +## Migration Path + +The migration is automatically executed when the backend is started. The process is handled by the `Database.connect` routine, which runs all SQL scripts in `backend/secuscan/migrations/`. + +### Migration Operations +The migration script (`007_standardize_plugin_ids.sql`) performs `UPDATE` operations on the following tables to rename legacy (hyphenated or non-standard) plugin IDs to their new canonical snake_case representations: +- `tasks` +- `findings` +- `crawl_runs` +- `asset_services` +- `audit_log` +- `presets` +- `plugins` + +For example: +- `domain-finder` is renamed to `domain_finder`. +- `website-recon-2` is renamed to `website_recon`. +- `waf-detection` is renamed/standardized to `waf_detector`. + +## Rollback Path + +If you need to roll back the migration, execute the following SQL script against the SQLite database: + +```sql +-- 1. Revert plugins in tasks table +UPDATE tasks SET plugin_id = 'domain-finder' WHERE plugin_id = 'domain_finder'; +UPDATE tasks SET plugin_id = 'google-dorking' WHERE plugin_id = 'google_dorking'; +UPDATE tasks SET plugin_id = 'people-email-discovery' WHERE plugin_id = 'people_email_discovery'; +UPDATE tasks SET plugin_id = 'port-scanner' WHERE plugin_id = 'port_scanner'; +UPDATE tasks SET plugin_id = 'subdomain-finder' WHERE plugin_id = 'subdomain_finder'; +UPDATE tasks SET plugin_id = 'url-fuzzer-2' WHERE plugin_id = 'url_fuzzer'; +UPDATE tasks SET plugin_id = 'virtual-host-finder' WHERE plugin_id = 'virtual_host_finder'; +UPDATE tasks SET plugin_id = 'website-recon-2' WHERE plugin_id = 'website_recon'; +UPDATE tasks SET plugin_id = 'waf-detection' WHERE plugin_id = 'waf_detector'; + +-- 2. Revert plugins in findings table +UPDATE findings SET plugin_id = 'domain-finder' WHERE plugin_id = 'domain_finder'; +UPDATE findings SET plugin_id = 'google-dorking' WHERE plugin_id = 'google_dorking'; +UPDATE findings SET plugin_id = 'people-email-discovery' WHERE plugin_id = 'people_email_discovery'; +UPDATE findings SET plugin_id = 'port-scanner' WHERE plugin_id = 'port_scanner'; +UPDATE findings SET plugin_id = 'subdomain-finder' WHERE plugin_id = 'subdomain_finder'; +UPDATE findings SET plugin_id = 'url-fuzzer-2' WHERE plugin_id = 'url_fuzzer'; +UPDATE findings SET plugin_id = 'virtual-host-finder' WHERE plugin_id = 'virtual_host_finder'; +UPDATE findings SET plugin_id = 'website-recon-2' WHERE plugin_id = 'website_recon'; +UPDATE findings SET plugin_id = 'waf-detection' WHERE plugin_id = 'waf_detector'; + +-- 3. Revert plugins in crawl_runs table +UPDATE crawl_runs SET plugin_id = 'domain-finder' WHERE plugin_id = 'domain_finder'; +UPDATE crawl_runs SET plugin_id = 'google-dorking' WHERE plugin_id = 'google_dorking'; +UPDATE crawl_runs SET plugin_id = 'people-email-discovery' WHERE plugin_id = 'people_email_discovery'; +UPDATE crawl_runs SET plugin_id = 'port-scanner' WHERE plugin_id = 'port_scanner'; +UPDATE crawl_runs SET plugin_id = 'subdomain-finder' WHERE plugin_id = 'subdomain_finder'; +UPDATE crawl_runs SET plugin_id = 'url-fuzzer-2' WHERE plugin_id = 'url_fuzzer'; +UPDATE crawl_runs SET plugin_id = 'virtual-host-finder' WHERE plugin_id = 'virtual_host_finder'; +UPDATE crawl_runs SET plugin_id = 'website-recon-2' WHERE plugin_id = 'website_recon'; +UPDATE crawl_runs SET plugin_id = 'waf-detection' WHERE plugin_id = 'waf_detector'; + +-- 4. Revert plugins in asset_services table +UPDATE asset_services SET plugin_id = 'domain-finder' WHERE plugin_id = 'domain_finder'; +UPDATE asset_services SET plugin_id = 'google-dorking' WHERE plugin_id = 'google_dorking'; +UPDATE asset_services SET plugin_id = 'people-email-discovery' WHERE plugin_id = 'people_email_discovery'; +UPDATE asset_services SET plugin_id = 'port-scanner' WHERE plugin_id = 'port_scanner'; +UPDATE asset_services SET plugin_id = 'subdomain-finder' WHERE plugin_id = 'subdomain_finder'; +UPDATE asset_services SET plugin_id = 'url-fuzzer-2' WHERE plugin_id = 'url_fuzzer'; +UPDATE asset_services SET plugin_id = 'virtual-host-finder' WHERE plugin_id = 'virtual_host_finder'; +UPDATE asset_services SET plugin_id = 'website-recon-2' WHERE plugin_id = 'website_recon'; +UPDATE asset_services SET plugin_id = 'waf-detection' WHERE plugin_id = 'waf_detector'; + +-- 5. Revert plugins in audit_log table +UPDATE audit_log SET plugin_id = 'domain-finder' WHERE plugin_id = 'domain_finder'; +UPDATE audit_log SET plugin_id = 'google-dorking' WHERE plugin_id = 'google_dorking'; +UPDATE audit_log SET plugin_id = 'people-email-discovery' WHERE plugin_id = 'people_email_discovery'; +UPDATE audit_log SET plugin_id = 'port-scanner' WHERE plugin_id = 'port_scanner'; +UPDATE audit_log SET plugin_id = 'subdomain-finder' WHERE plugin_id = 'subdomain_finder'; +UPDATE audit_log SET plugin_id = 'url-fuzzer-2' WHERE plugin_id = 'url_fuzzer'; +UPDATE audit_log SET plugin_id = 'virtual-host-finder' WHERE plugin_id = 'virtual_host_finder'; +UPDATE audit_log SET plugin_id = 'website-recon-2' WHERE plugin_id = 'website_recon'; +UPDATE audit_log SET plugin_id = 'waf-detection' WHERE plugin_id = 'waf_detector'; + +-- 6. Revert plugins in presets table +UPDATE presets SET plugin_id = 'domain-finder' WHERE plugin_id = 'domain_finder'; +UPDATE presets SET plugin_id = 'google-dorking' WHERE plugin_id = 'google_dorking'; +UPDATE presets SET plugin_id = 'people-email-discovery' WHERE plugin_id = 'people_email_discovery'; +UPDATE presets SET plugin_id = 'port-scanner' WHERE plugin_id = 'port_scanner'; +UPDATE presets SET plugin_id = 'subdomain-finder' WHERE plugin_id = 'subdomain_finder'; +UPDATE presets SET plugin_id = 'url-fuzzer-2' WHERE plugin_id = 'url_fuzzer'; +UPDATE presets SET plugin_id = 'virtual-host-finder' WHERE plugin_id = 'virtual_host_finder'; +UPDATE presets SET plugin_id = 'website-recon-2' WHERE plugin_id = 'website_recon'; +UPDATE presets SET plugin_id = 'waf-detection' WHERE plugin_id = 'waf_detector'; + +-- 7. Revert plugins in plugins table +UPDATE plugins SET id = 'domain-finder' WHERE id = 'domain_finder'; +UPDATE plugins SET id = 'google-dorking' WHERE id = 'google_dorking'; +UPDATE plugins SET id = 'people-email-discovery' WHERE id = 'people_email_discovery'; +UPDATE plugins SET id = 'port-scanner' WHERE id = 'port_scanner'; +UPDATE plugins SET id = 'subdomain-finder' WHERE id = 'subdomain_finder'; +UPDATE plugins SET id = 'url-fuzzer-2' WHERE id = 'url_fuzzer'; +UPDATE plugins SET id = 'virtual-host-finder' WHERE id = 'virtual_host_finder'; +UPDATE plugins SET id = 'website-recon-2' WHERE id = 'website_recon'; +INSERT OR REPLACE INTO plugins (id, name, version, category, metadata_json) VALUES ('waf-detection', 'WAF Detector', '1.0.0', 'robots', '{}'); + +-- 8. Decrement the schema version if necessary in schema_migrations +-- DELETE FROM schema_migrations WHERE version = 7; +``` diff --git a/testing/backend/unit/test_plugin_compatibility.py b/testing/backend/unit/test_plugin_compatibility.py new file mode 100644 index 000000000..865f60ea2 --- /dev/null +++ b/testing/backend/unit/test_plugin_compatibility.py @@ -0,0 +1,72 @@ +import asyncio +import pytest +from backend.secuscan.config import settings +from backend.secuscan.plugins import PluginManager, LEGACY_PLUGIN_ID_ALIASES + +def test_legacy_plugin_id_aliases_resolve(setup_test_environment): + """Test that all defined legacy plugin ID aliases successfully resolve to valid, loaded plugins.""" + manager = PluginManager(settings.plugins_dir) + asyncio.run(manager.load_plugins()) + + for legacy_id, canonical_id in LEGACY_PLUGIN_ID_ALIASES.items(): + # Only verify if the canonical plugin is actually loaded/available on disk + plugin_by_canonical = manager.get_plugin(canonical_id) + if plugin_by_canonical is None: + continue + + # Get plugin metadata by legacy ID + plugin_by_legacy = manager.get_plugin(legacy_id) + assert plugin_by_legacy is not None, f"Legacy ID {legacy_id} returned None" + assert plugin_by_legacy.id in (legacy_id, canonical_id), f"Legacy ID {legacy_id} resolved to {plugin_by_legacy.id} instead of {legacy_id} or {canonical_id}" + assert plugin_by_legacy == plugin_by_canonical + + # Get schema by legacy and canonical ID + schema_by_legacy = manager.get_plugin_schema(legacy_id) + schema_by_canonical = manager.get_plugin_schema(canonical_id) + assert schema_by_legacy == schema_by_canonical + assert schema_by_legacy["id"] == canonical_id + +def test_build_command_with_legacy_id(setup_test_environment): + """Test that building commands with legacy IDs works and generates the exact same command as the canonical ID.""" + manager = PluginManager(settings.plugins_dir) + asyncio.run(manager.load_plugins()) + + # Test domain-finder / domain_finder (present on disk) + cmd_legacy = manager.build_command("domain-finder", {"target": "example.com"}) + cmd_canonical = manager.build_command("domain_finder", {"target": "example.com"}) + assert cmd_legacy == cmd_canonical + assert cmd_legacy is not None + assert "amass" in cmd_legacy or "python3" in cmd_legacy + assert "example.com" in cmd_legacy + + # Test google-dorking + cmd_legacy = manager.build_command("google-dorking", {"target": "example.com"}) + cmd_canonical = manager.build_command("google_dorking", {"target": "example.com"}) + assert cmd_legacy == cmd_canonical + assert cmd_legacy is not None + assert "google" in cmd_legacy or "python" in cmd_legacy or "python3" in cmd_legacy + +def test_task_start_via_api_resolves_legacy_id(test_client): + """Test that submitting a task with a legacy plugin ID via the API succeeds and stores the task with the canonical ID.""" + # Submit task with legacy ID "domain-finder" + response = test_client.post( + "/api/v1/task/start", + json={ + "plugin_id": "domain-finder", + "consent_granted": True, + "inputs": { + "target": "127.0.0.1" + } + } + ) + assert response.status_code == 200, f"Task start failed: {response.text}" + data = response.json() + assert "task_id" in data + task_id = data["task_id"] + + # Verify task details via the status API + status_response = test_client.get(f"/api/v1/task/{task_id}/status") + assert status_response.status_code == 200 + status_data = status_response.json() + # The plugin_id stored on the task must be the canonical/standardized one + assert status_data["plugin_id"] == "domain_finder" diff --git a/testing/backend/unit/test_plugin_validator.py b/testing/backend/unit/test_plugin_validator.py index e153d972e..9f9ad9b45 100644 --- a/testing/backend/unit/test_plugin_validator.py +++ b/testing/backend/unit/test_plugin_validator.py @@ -48,7 +48,8 @@ def _error_messages(result: ValidationResult) -> list[str]: def _write_metadata(tmp_path: Path, data: dict) -> Path: - plugin_dir = tmp_path / "my_plugin" + dir_name = data.get("id") or "my_plugin" + plugin_dir = tmp_path / dir_name plugin_dir.mkdir(exist_ok=True) (plugin_dir / "metadata.json").write_text(json.dumps(data), encoding="utf-8") return plugin_dir @@ -169,6 +170,44 @@ def test_empty_string_name_is_reported(self, tmp_path): assert "name" in _error_paths(result) +# =========================================================================== +# Plugin ID validation +# =========================================================================== + + +class TestPluginId: + @pytest.mark.parametrize("valid_id", [ + "amass", "api_scanner", "dns_enum", "http_inspector", "waf_detector", "website_recon", "fuzzer", "tls_inspector_v2" + ]) + def test_valid_ids_accepted(self, tmp_path, valid_id): + data = _minimal_valid() + data["id"] = valid_id + plugin_dir = _write_metadata(tmp_path, data) + result = validate_one_plugin(plugin_dir) + assert "id" not in _error_paths(result) + + @pytest.mark.parametrize("invalid_id", [ + "invalid-plugin-id", "some-other-plugin", "DNS_Enum", "tls inspector", "123_scanner", "_scanner" + ]) + def test_invalid_ids_rejected(self, tmp_path, invalid_id): + data = _minimal_valid() + data["id"] = invalid_id + plugin_dir = _write_metadata(tmp_path, data) + result = validate_one_plugin(plugin_dir) + assert "id" in _error_paths(result) + + def test_mismatched_directory_name_rejected(self, tmp_path): + data = _minimal_valid() + data["id"] = "test_ping" + # Manually write to mismatched folder name + plugin_dir = tmp_path / "mismatched_folder" + plugin_dir.mkdir() + (plugin_dir / "metadata.json").write_text(json.dumps(data), encoding="utf-8") + result = validate_one_plugin(plugin_dir) + assert "id" in _error_paths(result) + assert any("directory name" in e.message for e in result.errors if e.path == "id") + + # =========================================================================== # Engine # =========================================================================== diff --git a/testing/backend/unit/test_saved_views.py b/testing/backend/unit/test_saved_views.py index 62c615051..169c6cc20 100644 --- a/testing/backend/unit/test_saved_views.py +++ b/testing/backend/unit/test_saved_views.py @@ -7,7 +7,7 @@ import pytest_asyncio from httpx import AsyncClient, ASGITransport -from fastapi import FastAPI +from fastapi import FastAPI, Request, HTTPException from backend.secuscan.saved_views import saved_views_router from backend.secuscan.database import Database, get_db from backend.secuscan.auth import require_api_key @@ -18,8 +18,16 @@ # ─── Fixtures ───────────────────────────────────────────────────────────────── -async def _mock_require_api_key() -> str: - """Mock auth dependency that always succeeds.""" +async def _mock_require_api_key(request: Request) -> str: + """Mock auth dependency that verifies API key validity in testing.""" + candidate = request.headers.get("X-Api-Key") + if not candidate: + bearer = request.headers.get("Authorization", "") + if bearer.lower().startswith("bearer "): + candidate = bearer[7:] + + if not candidate or candidate != _auth_module._api_key: + raise HTTPException(status_code=401, detail="Invalid API key") return "test-owner-id"