fix(workflow): enforce exploit-validation target policy gate on workflow runs

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"))

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

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"]

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"]

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