Suggestion fix

Author: Sahilbhatane

.github/workflows/codeql.yml

@@ -213,9 +213,9 @@ def parallel_log_callback(message: str, level: str = "info"):
                         # Calculate total duration from tasks
                         total_duration = 0
                         if parallel_tasks:
-                            max_end = max((t.end_time or 0) for t in parallel_tasks)
-                            min_start = min((t.start_time or time.time()) for t in parallel_tasks)
-                            if max_end and min_start:
+                            max_end = max((t.end_time for t in parallel_tasks if t.end_time is not None), default=None)
+                            min_start = min((t.start_time for t in parallel_tasks if t.start_time is not None), default=None)
+                            if max_end is not None and min_start is not None:
                                 total_duration = max_end - min_start
 
                         if success:

cortex/cli.py

@@ -9,23 +9,9 @@
 from datetime import datetime
 import logging
 
-logger = logging.getLogger(__name__)
+from cortex.validators import DANGEROUS_PATTERNS
 
-# Dangerous patterns that should never be executed
-DANGEROUS_PATTERNS = [
-    r'rm\s+-rf\s+[/\*]',
-    r'rm\s+--no-preserve-root',
-    r'dd\s+if=.*of=/dev/',
-    r'curl\s+.*\|\s*sh',
-    r'curl\s+.*\|\s*bash',
-    r'wget\s+.*\|\s*sh',
-    r'wget\s+.*\|\s*bash',
-    r'\beval\s+',
-    r'base64\s+-d\s+.*\|',
-    r'>\s*/etc/',
-    r'chmod\s+777',
-    r'chmod\s+\+s',
-]
+logger = logging.getLogger(__name__)
 
 
 class StepStatus(Enum):

cortex/coordinator.py

@@ -7,6 +7,8 @@
 from dataclasses import dataclass, field
 from enum import Enum
 
+from cortex.validators import DANGEROUS_PATTERNS
+
 
 class TaskStatus(Enum):
     PENDING = "pending"
@@ -55,36 +57,23 @@ async def run_single_task(task: ParallelTask, executor, timeout: int, log_callba
         log_callback(f"Starting {task.name}…", "info")
 
     # Validate command for dangerous patterns
-    DANGEROUS_PATTERNS = [
-        r'rm\s+-rf\s+[/\*]',
-        r'rm\s+--no-preserve-root',
-        r'dd\s+if=.*of=/dev/',
-        r'curl\s+.*\|\s*sh',
-        r'curl\s+.*\|\s*bash',
-        r'wget\s+.*\|\s*sh',
-        r'wget\s+.*\|\s*bash',
-        r'\beval\s+',
-        r'base64\s+-d\s+.*\|',
-        r'>\s*/etc/',
-        r'chmod\s+777',
-        r'chmod\s+\+s',
-    ]
-    
     for pattern in DANGEROUS_PATTERNS:
         if re.search(pattern, task.command, re.IGNORECASE):
             task.status = TaskStatus.FAILED
-            task.error = f"Command blocked: matches dangerous pattern"
+            task.error = "Command blocked: matches dangerous pattern"
             task.end_time = time.time()
             if log_callback:
                 log_callback(f"Finished {task.name} (failed)", "error")
             return False
 
     try:
         # Run command in executor (thread pool) to avoid blocking the event loop
-        loop = asyncio.get_event_loop()
+        loop = asyncio.get_running_loop()
         result = await asyncio.wait_for(
             loop.run_in_executor(
                 executor,
+                # Use shell=True carefully - commands are validated against dangerous patterns above.
+                # shell=True is required to support complex shell commands (e.g., pipes, redirects).
                 lambda: subprocess.run(
                     task.command,
                     shell=True,
@@ -148,7 +137,7 @@ async def run_parallel_install(
         log_callback: Optional callback for logging (called with message and level)
     
     Returns:
-        Tuple of (success: bool, tasks: List[ParallelTask])
+        tuple[bool, List[ParallelTask]]: Success status and list of all tasks
     """
     if not commands:
         return True, []
@@ -190,14 +179,31 @@ async def run_parallel_install(
         while pending or running:
             # Start tasks whose dependencies are met
             ready_to_start = []
-            for task_name in list(pending):
+            for task_name in pending.copy():
                 task = tasks[task_name]
-                deps_met = all(dep in completed for dep in task.dependencies)
+                # When stop_on_error=False, accept both completed and failed dependencies
+                if stop_on_error:
+                    deps_met = all(dep in completed for dep in task.dependencies)
+                else:
+                    deps_met = all(dep in completed or dep in failed for dep in task.dependencies)
 
                 if deps_met:
                     ready_to_start.append(task_name)
                     pending.remove(task_name)
 
+            # If no tasks can be started and none are running, we're stuck (deadlock/cycle detection)
+            if not ready_to_start and not running and pending:
+                # Mark remaining tasks as skipped - they have unresolvable dependencies
+                for task_name in pending:
+                    task = tasks[task_name]
+                    if task.status == TaskStatus.PENDING:
+                        task.status = TaskStatus.SKIPPED
+                        task.error = "Task could not run because dependencies never completed"
+                        if log_callback:
+                            log_callback(f"{task_name} skipped due to unresolved dependencies", "error")
+                failed.update(pending)
+                break
+            
             # Create tasks for ready items
             for task_name in ready_to_start:
                 coro = run_single_task(tasks[task_name], executor, timeout, log_callback)
@@ -217,10 +223,20 @@ async def run_parallel_install(
                 # Process completed tasks
                 for task_coro in done:
                     # Find which task this is
-                    for task_name, running_coro in list(running.items()):
+                    for task_name, running_coro in running.items():
                         if running_coro is task_coro:
                             task = tasks[task_name]
-                            success = task_coro.result()
+                            
+                            # Handle cancelled tasks
+                            try:
+                                success = task_coro.result()
+                            except asyncio.CancelledError:
+                                # Task was cancelled due to stop_on_error
+                                task.status = TaskStatus.SKIPPED
+                                task.error = "Task cancelled due to dependency failure"
+                                failed.add(task_name)
+                                del running[task_name]
+                                break
 
                             if success:
                                 completed.add(task_name)

cortex/install_parallel.py

@@ -9,6 +9,23 @@
 from typing import Optional, Tuple
 
 
+# Dangerous command patterns to block for security
+DANGEROUS_PATTERNS = [
+    r'rm\s+-rf\s+[/\*]',
+    r'rm\s+--no-preserve-root',
+    r'dd\s+if=.*of=/dev/',
+    r'curl\s+.*\|\s*sh',
+    r'curl\s+.*\|\s*bash',
+    r'wget\s+.*\|\s*sh',
+    r'wget\s+.*\|\s*bash',
+    r'\beval\s+',
+    r'base64\s+-d\s+.*\|',
+    r'>\s*/etc/',
+    r'chmod\s+777',
+    r'chmod\s+\+s',
+]
+
+
 class ValidationError(Exception):
     """Custom exception for validation errors with user-friendly messages"""
 

cortex/validators.py

@@ -9,3 +9,4 @@ rich>=13.0.0
 
 # Type hints for older Python versions
 typing-extensions>=4.0.0
+PyYAML>=6.0.0

requirements.txt

@@ -5,7 +5,6 @@
 import time
 from cortex.install_parallel import (
     run_parallel_install,
-    ParallelTask,
     TaskStatus
 )
 
@@ -64,10 +63,6 @@ async def run_test():
 
             assert success
             assert all(t.status == TaskStatus.SUCCESS for t in tasks)
-            
-            # Verify all tasks completed (they were sequential)
-            assert len(tasks) == 3
-            assert all(t.status == TaskStatus.SUCCESS for t in tasks)
 
         asyncio.run(run_test())
 
@@ -178,7 +173,7 @@ async def run_test():
 
             assert not success
             assert tasks[0].status == TaskStatus.FAILED
-            assert "timed out" in tasks[0].error.lower() or tasks[0].error != ""
+            assert "timed out" in tasks[0].error.lower() or "timeout" in tasks[0].error.lower()
 
         asyncio.run(run_test())
 
@@ -242,7 +237,7 @@ async def run_test():
             def log_callback(message: str, level: str = "info"):
                 log_messages.append((message, level))
 
-            success, tasks = await run_parallel_install(
+            success, _tasks = await run_parallel_install(
                 commands,
                 timeout=10,
                 log_callback=log_callback