feat(sandbox): hardened sandbox executor with proactive streaming and resource limits (Closes #326)

backend/secuscan/config.py

@@ -81,6 +81,8 @@ class Settings(BaseSettings):
     sandbox_timeout: int = 600  # seconds
     sandbox_cpu_quota: float = 0.5
     sandbox_memory_mb: int = 512
+    sandbox_max_output_bytes: int = 5_242_880  # 5 MB
+    sandbox_allow_network: bool = True
 
     # Task-start payload limits (tunable via env vars)
     task_start_max_body_bytes: int = 64_000       # 64 KB total JSON body

backend/secuscan/executor.py

@@ -18,8 +18,9 @@
 from .config import settings
 from .database import get_db
 from .plugins import get_plugin_manager
-from .models import TaskStatus, ScanPhase
+from .models import TaskStatus, ScanPhase, SandboxConfig
 from .ratelimit import concurrent_limiter
+from .sandbox_executor import sandbox_execute
 from .risk_scoring import compute_risk_score, compute_risk_factors
 
 
@@ -354,28 +355,33 @@ async def execute_task(self, task_id: str):
                 await self._broadcast(task_id, "status", TaskStatus.RUNNING.value)
                 await self._broadcast_phase(task_id, ScanPhase.RUNNING_COMMAND.value)
 
-                # Execute command
                 start_time = time.time()
-                output, exit_code = await self._execute_command(
+                output, exit_code, violation_reason = await self._execute_command(
                     command,
                     task_id,
                     timeout=self._resolve_execution_timeout(inputs),
                 )
                 duration = time.time() - start_time
 
-                # Save raw output
                 raw_path = Path(settings.raw_output_dir) / f"{task_id}.txt"
                 output = redact(output)
                 with open(raw_path, 'w') as f:
                     f.write(output)
 
-                # Some CLI tools use non-zero exit codes for "no result" states while still
-                # producing a complete, parseable report. Let plugin metadata opt into that.
-                final_status, error_message = self._classify_command_result(
-                    plugin=plugin,
-                    output=output,
-                    exit_code=exit_code,
-                )
+                if violation_reason:
+                    status_map = {
+                        "timeout": TaskStatus.TERMINATED_TIMEOUT.value,
+                        "memory_limit": TaskStatus.TERMINATED_MEMORY.value,
+                        "output_limit": TaskStatus.TERMINATED_OUTPUT.value,
+                    }
+                    final_status = status_map.get(violation_reason, TaskStatus.FAILED.value)
+                    error_message = f"Sandbox violation: {violation_reason}"
+                else:
+                    final_status, error_message = self._classify_command_result(
+                        plugin=plugin,
+                        output=output,
+                        exit_code=exit_code,
+                    )
 
                 await db.execute(
                     """
@@ -499,63 +505,36 @@ async def _execute_command(
         self,
         command: list,
         task_id: str,
-        timeout: int = 600
+        timeout: int = 600,
     ) -> tuple:
-        """
-        Execute command in subprocess and stream output.
+        if timeout is None:
+            timeout = settings.sandbox_timeout
+        config = SandboxConfig(
+            timeout_seconds=0,
+            max_memory_mb=settings.sandbox_memory_mb,
+            max_output_bytes=settings.sandbox_max_output_bytes,
+            allow_network=settings.sandbox_allow_network,
+        )
 
-        Args:
-            command: Command as list
-            task_id: Task identifier for logging
-            timeout: Execution timeout in seconds
+        async def _on_chunk(data: bytes, stream_name: str):
+            text = data.decode("utf-8", errors="replace")
+            await self._broadcast(task_id, "output", text)
 
-        Returns:
-            Tuple of (output, exit_code)
-        """
         try:
-            process = await asyncio.create_subprocess_exec(
-                *command,
-                stdout=subprocess.PIPE,
-                stderr=subprocess.STDOUT
+            stdout, stderr, exit_code, violation_reason = await asyncio.wait_for(
+                sandbox_execute(command, config, broadcast_callback=_on_chunk),
+                timeout=timeout,
             )
-
-            output_lines = []
-
-            async def read_stream():
-                stdout = process.stdout
-                if stdout is None:
-                    return
-
-                while not stdout.at_eof():
-                    line = await stdout.readline()
-                    if line:
-                        decoded_line = line.decode('utf-8', errors='replace')
-                        output_lines.append(decoded_line)
-                        await self._broadcast(task_id, "output", decoded_line)
-
-            try:
-                await asyncio.wait_for(read_stream(), timeout=timeout)
-                await process.wait()
-                return "".join(output_lines), process.returncode if process.returncode is not None else -1
-
-            except asyncio.TimeoutError:
-                process.kill()
-                await process.wait()
-                return "".join(output_lines) + "\nTask timed out", -1
-
-            except asyncio.CancelledError:
-                # Handle task cancellation by killing the subprocess
-                logger.warning(f"Task {task_id} cancelled. Killing process {process.pid}")
-                try:
-                    process.kill()
-                    await process.wait()
-                except Exception as e:
-                    logger.error(f"Error killing process for cancelled task {task_id}: {e}")
-                raise
-
+            if stderr:
+                stdout = stdout + "\n" + stderr if stdout else stderr
+            return stdout, exit_code, violation_reason
+        except asyncio.TimeoutError:
+            return "", -1, "timeout"
+        except asyncio.CancelledError:
+            raise
         except Exception as e:
             logger.error(f"Failed to execute command: {e}")
-            return f"Execution error: {str(e)}", -1
+            return f"Execution error: {str(e)}", -1, None
 
     def _resolve_execution_timeout(self, inputs: Dict[str, Any]) -> int:
         """Resolve per-task process timeout from plugin inputs."""

backend/secuscan/models.py

@@ -24,6 +24,17 @@ class TaskStatus(str, Enum):
     COMPLETED = "completed"
     FAILED = "failed"
     CANCELLED = "cancelled"
+    TERMINATED_TIMEOUT = "terminated:timeout"
+    TERMINATED_MEMORY = "terminated:memory_limit"
+    TERMINATED_OUTPUT = "terminated:output_limit"
+
+
+class SandboxConfig(BaseModel):
+    """Resource constraints applied to every plugin subprocess execution"""
+    timeout_seconds: int = Field(default=120, description="Max wall-clock seconds before SIGTERM")
+    max_memory_mb: int = Field(default=512, description="Max virtual memory in MB (RLIMIT_AS on Linux)")
+    max_output_bytes: int = Field(default=5_242_880, description="Max bytes captured from stdout/stderr")
+    allow_network: bool = Field(default=True, description="Whether subprocess can make network calls")
 
 
 class ScanPhase(str, Enum):
@@ -83,6 +94,8 @@ class PluginMetadata(BaseModel):
     dependencies: Optional[Dict[str, List[str]]] = None
     docker_image: Optional[str] = None
 
+    sandbox: Optional[SandboxConfig] = None
+
     checksum: Optional[str] = None
     signature: Optional[str] = None
 
@@ -171,6 +184,14 @@ class PluginListResponse(BaseModel):
     total: int
 
 
+class SandboxViolation(Exception):
+    """Raised when sandbox constraints are violated."""
+
+    def __init__(self, reason: str):
+        super().__init__(reason)
+        self.reason = reason
+
+
 class ErrorResponse(BaseModel):
     """Error response"""
     error: str