diff --git a/backend/secuscan/routes.py b/backend/secuscan/routes.py index a1e48fded..21dfd8f8b 100644 --- a/backend/secuscan/routes.py +++ b/backend/secuscan/routes.py @@ -2136,6 +2136,21 @@ async def run_workflow_once(workflow_id: str, owner: str = Depends(get_current_o for step in steps: execution_context = normalize_execution_context(step.get("execution_context") or {}) target_policy = await get_target_policy(db, owner, execution_context.get("target_policy_id")) + plugin = get_plugin_manager().get_plugin(step.get("plugin_id")) + if not plugin: + logger.warning("Workflow %s: plugin %s not found, skipping step", workflow_id, step.get("plugin_id")) + continue + requires_exploit_policy = ( + plugin.safety.get("level") == "exploit" + or execution_context.get("validation_mode") == ValidationMode.CONTROLLED_EXTRACT.value + ) + if requires_exploit_policy and not (target_policy and target_policy.get("allow_exploit_validation")): + logger.warning( + "Workflow %s: skipping exploit-level step %s: no target policy allows exploit validation", + workflow_id, + step.get("plugin_id"), + ) + continue safe_mode = bool( settings.safe_mode_default and not (target_policy and target_policy.get("allow_public_targets")) diff --git a/backend/secuscan/workflows.py b/backend/secuscan/workflows.py index ebf752566..1ef03fcdd 100644 --- a/backend/secuscan/workflows.py +++ b/backend/secuscan/workflows.py @@ -12,6 +12,7 @@ from .ratelimit import workflow_rate_limiter, rate_limiter, concurrent_limiter from .executor import executor from .execution_context import normalize_execution_context +from .models import ValidationMode from .platform_resources import get_target_policy logger = logging.getLogger(__name__) class WorkflowScheduler: @@ -146,6 +147,17 @@ async def _run_workflow(self, workflow_id: str, steps: List[Dict[str, Any]], own if not plugin: logger.warning("Workflow %s: plugin %s not found, skipping step", workflow_id, plugin_id) continue + requires_exploit_policy = ( + plugin.safety.get("level") == "exploit" + or execution_context.get("validation_mode") == ValidationMode.CONTROLLED_EXTRACT.value + ) + if requires_exploit_policy and not (target_policy and target_policy.get("allow_exploit_validation")): + logger.warning( + "Workflow %s: skipping exploit-level step %s: no target policy allows exploit validation", + workflow_id, + plugin_id, + ) + continue effective_inputs = dict(inputs) effective_inputs.pop("safe_mode", None) effective_inputs["safe_mode"] = safe_mode diff --git a/testing/backend/test_workflow_api_edge_cases.py b/testing/backend/test_workflow_api_edge_cases.py index c4f8f76e4..de7193eba 100644 --- a/testing/backend/test_workflow_api_edge_cases.py +++ b/testing/backend/test_workflow_api_edge_cases.py @@ -285,3 +285,94 @@ def test_valid_creation_response_body_is_json_object(self, client): assert isinstance(response.json(), dict), ( f"Expected JSON object, got: {response.text}" ) + + +# --------------------------------------------------------------------------- +# Workflow run — exploit-validation policy gate +# --------------------------------------------------------------------------- + +class TestRunWorkflowExploitGate: + """POST /workflows/{id}/run must enforce the same exploit-validation gate as task start.""" + + @pytest.mark.asyncio + async def test_run_workflow_skips_exploit_step_without_policy(self): + from backend.secuscan.routes import run_workflow_once + + workflow_row = { + "id": "wf-exploit", + "name": "exploit-bypass", + "owner_id": "user-1", + "schedule_seconds": None, + "enabled": 1, + "steps_json": json.dumps( + [{"plugin_id": "xss_exploiter", "inputs": {"target": "https://example.com"}}] + ), + "last_run_at": None, + } + version_row = {"id": "ver-1", "version_number": 3} + + mock_db = AsyncMock() + mock_db.fetchone = AsyncMock(side_effect=[workflow_row, version_row]) + mock_db.execute = AsyncMock(return_value=None) + mock_db.record_workflow_run = AsyncMock(return_value="run-1") + + plugin = MagicMock() + plugin.safety = {"level": "exploit"} + + with patch("backend.secuscan.routes.get_db", new_callable=AsyncMock, return_value=mock_db), \ + patch("backend.secuscan.routes.workflow_rate_limiter.check_workflow_rate_limit", + new_callable=AsyncMock, return_value=(True, "")), \ + patch("backend.secuscan.routes.get_target_policy", new_callable=AsyncMock, return_value=None), \ + patch("backend.secuscan.routes.get_plugin_manager") as mock_get_pm, \ + patch("backend.secuscan.routes.executor.create_task", new_callable=AsyncMock, return_value="task-1") as mock_create, \ + patch("backend.secuscan.routes._finalize_workflow_run", new_callable=AsyncMock): + + mock_get_pm.return_value.get_plugin.return_value = plugin + + response = await run_workflow_once("wf-exploit", owner="user-1") + + mock_create.assert_not_called() + assert response["queued_task_ids"] == [] + + @pytest.mark.asyncio + async def test_run_workflow_keeps_exploit_step_when_policy_allows(self): + from backend.secuscan.routes import run_workflow_once + + workflow_row = { + "id": "wf-exploit", + "name": "exploit-bypass", + "owner_id": "user-1", + "schedule_seconds": None, + "enabled": 1, + "steps_json": json.dumps( + [{"plugin_id": "xss_exploiter", "inputs": {"target": "https://example.com"}}] + ), + "last_run_at": None, + } + version_row = {"id": "ver-1", "version_number": 3} + + mock_db = AsyncMock() + mock_db.fetchone = AsyncMock(side_effect=[workflow_row, version_row]) + mock_db.execute = AsyncMock(return_value=None) + mock_db.record_workflow_run = AsyncMock(return_value="run-1") + + plugin = MagicMock() + plugin.safety = {"level": "exploit"} + + with patch("backend.secuscan.routes.get_db", new_callable=AsyncMock, return_value=mock_db), \ + patch("backend.secuscan.routes.workflow_rate_limiter.check_workflow_rate_limit", + new_callable=AsyncMock, return_value=(True, "")), \ + patch("backend.secuscan.routes.get_target_policy", new_callable=AsyncMock, + return_value={"allow_exploit_validation": True, "allow_public_targets": True}), \ + patch("backend.secuscan.routes.get_plugin_manager") as mock_get_pm, \ + patch("backend.secuscan.routes.executor.create_task", new_callable=AsyncMock, return_value="task-1") as mock_create, \ + patch("backend.secuscan.routes.concurrent_limiter.acquire", new_callable=AsyncMock, return_value=(True, "")), \ + patch("backend.secuscan.routes.executor.execute_task", new_callable=AsyncMock), \ + patch("backend.secuscan.routes._finalize_workflow_run", new_callable=AsyncMock): + + mock_get_pm.return_value.get_plugin.return_value = plugin + + response = await run_workflow_once("wf-exploit", owner="user-1") + + mock_create.assert_called_once() + assert response["queued_task_ids"] == ["task-1"] diff --git a/testing/backend/unit/test_workflow_concurrency_ordering.py b/testing/backend/unit/test_workflow_concurrency_ordering.py index be741ff7b..d69adba6d 100644 --- a/testing/backend/unit/test_workflow_concurrency_ordering.py +++ b/testing/backend/unit/test_workflow_concurrency_ordering.py @@ -68,6 +68,7 @@ def tracking_create_task(coro, **kwargs): AsyncMock(return_value=(True, ""))), patch("backend.secuscan.routes.normalize_execution_context", return_value={}), patch("backend.secuscan.routes.get_target_policy", AsyncMock(return_value=None)), + patch("backend.secuscan.routes.get_plugin_manager") as mock_get_pm, patch("backend.secuscan.routes.concurrent_limiter.acquire", side_effect=tracking_acquire), patch("backend.secuscan.routes.executor.create_task", @@ -77,6 +78,7 @@ def tracking_create_task(coro, **kwargs): patch("backend.secuscan.routes.logger"), patch("asyncio.create_task", side_effect=tracking_create_task), ): + mock_get_pm.return_value.get_plugin.return_value = MagicMock(safety={"level": "safe"}) await run_workflow_once("wf-1", owner="owner") assert len(call_order) >= 2 @@ -111,6 +113,7 @@ async def test_rejected_acquire_marks_failed_and_skips_execution(): AsyncMock(return_value=(True, ""))), patch("backend.secuscan.routes.normalize_execution_context", return_value={}), patch("backend.secuscan.routes.get_target_policy", AsyncMock(return_value=None)), + patch("backend.secuscan.routes.get_plugin_manager") as mock_get_pm, patch("backend.secuscan.routes.concurrent_limiter.acquire", AsyncMock(return_value=(False, "Concurrency limit reached"))), patch("backend.secuscan.routes.executor.create_task", @@ -121,6 +124,7 @@ async def test_rejected_acquire_marks_failed_and_skips_execution(): new_callable=AsyncMock) as mock_execute, patch("backend.secuscan.routes.logger"), ): + mock_get_pm.return_value.get_plugin.return_value = MagicMock(safety={"level": "safe"}) result = await run_workflow_once("wf-1", owner="owner") mock_mark_failed.assert_called_once_with( @@ -160,6 +164,7 @@ async def test_rejected_acquire_does_not_block_accepted_tasks(): AsyncMock(return_value=(True, ""))), patch("backend.secuscan.routes.normalize_execution_context", return_value={}), patch("backend.secuscan.routes.get_target_policy", AsyncMock(return_value=None)), + patch("backend.secuscan.routes.get_plugin_manager") as mock_get_pm, patch("backend.secuscan.routes.concurrent_limiter.acquire", AsyncMock(side_effect=[ (False, "Concurrency limit reached"), @@ -173,6 +178,7 @@ async def test_rejected_acquire_does_not_block_accepted_tasks(): new_callable=AsyncMock) as mock_execute, patch("backend.secuscan.routes.logger"), ): + mock_get_pm.return_value.get_plugin.return_value = MagicMock(safety={"level": "safe"}) result = await run_workflow_once("wf-1", owner="owner") assert result["queued_task_ids"] == ["task-2"] diff --git a/testing/backend/unit/test_workflow_scheduler_security.py b/testing/backend/unit/test_workflow_scheduler_security.py index fba6d6c06..7a9e51ccd 100644 --- a/testing/backend/unit/test_workflow_scheduler_security.py +++ b/testing/backend/unit/test_workflow_scheduler_security.py @@ -193,6 +193,58 @@ async def test_skips_step_when_concurrency_limit_reached(self, scheduler): await scheduler._run_workflow("wf-1", steps) mock_fail.assert_called_once() + @pytest.mark.asyncio + async def test_skips_exploit_step_without_exploit_policy(self, scheduler): + """Exploit-level steps must be skipped when no target policy allows exploit validation.""" + steps = [{ + "plugin_id": "sqlmap", + "inputs": {"target": "example.com"}, + }] + with patch("backend.secuscan.workflows.get_db", new_callable=AsyncMock), \ + patch("backend.secuscan.plugins.get_plugin_manager") as mock_get_pm, \ + patch("backend.secuscan.workflows.get_target_policy", new_callable=AsyncMock, return_value=None), \ + patch("backend.secuscan.executor.executor.create_task", new_callable=AsyncMock, return_value="task-1") as mock_create: + + mock_pm = MagicMock() + plugin = MagicMock() + plugin.category = "exploit" + plugin.safety = {"level": "exploit", "rate_limit": {"max_per_hour": 5}} + plugin.fields = [] + mock_pm.get_plugin.return_value = plugin + mock_get_pm.return_value = mock_pm + + await scheduler._run_workflow("wf-1", steps, owner_id="default") + mock_create.assert_not_called() + + @pytest.mark.asyncio + async def test_runs_exploit_step_when_policy_allows(self, scheduler): + """Exploit-level steps run when the target policy explicitly allows exploit validation.""" + steps = [{ + "plugin_id": "sqlmap", + "inputs": {"target": "example.com"}, + }] + with patch("backend.secuscan.workflows.get_db", new_callable=AsyncMock), \ + patch("backend.secuscan.plugins.get_plugin_manager") as mock_get_pm, \ + patch("backend.secuscan.workflows.get_target_policy", new_callable=AsyncMock, + return_value={"allow_exploit_validation": True, "allow_public_targets": True}), \ + patch("backend.secuscan.validation.validate_target", return_value=(True, "")), \ + patch("backend.secuscan.ratelimit.rate_limiter.can_execute", return_value=(True, "")), \ + patch("backend.secuscan.ratelimit.concurrent_limiter.acquire", return_value=(True, "")), \ + patch("backend.secuscan.executor.executor.create_task", new_callable=AsyncMock, return_value="task-1") as mock_create, \ + patch("backend.secuscan.executor.executor.execute_task", new_callable=AsyncMock), \ + patch("backend.secuscan.workflows._finalize_workflow_run", new_callable=AsyncMock): + + mock_pm = MagicMock() + plugin = MagicMock() + plugin.category = "exploit" + plugin.safety = {"level": "exploit", "rate_limit": {"max_per_hour": 5}} + plugin.fields = [] + mock_pm.get_plugin.return_value = plugin + mock_get_pm.return_value = mock_pm + + await scheduler._run_workflow("wf-1", steps, owner_id="default") + mock_create.assert_called_once() + # --------------------------------------------------------------------------- # WorkflowScheduler.tick rate limit integration