Merge pull request #1285 from MervinPraison/claude/pr-1284-20260407-2042

PR #1284: Changes from Claude

Author: MervinPraison

src/praisonai-agents/praisonaiagents/context/monitor.py

@@ -542,7 +542,9 @@ def _get_output_path(self, agent_name: str = "") -> Path:
         # Verify the resolved path stays within the expected parent dir
         resolved = result.resolve()
         parent_resolved = self.path.parent.resolve()
-        if not str(resolved).startswith(str(parent_resolved) + '/') and resolved.parent != parent_resolved:
+        try:
+            resolved.relative_to(parent_resolved)
+        except ValueError:
             # Fallback to a safe hash-based name
             import hashlib
             hash_name = hashlib.sha256(agent_name.encode()).hexdigest()[:16]

src/praisonai-ts/src/workflows/yaml-parser.ts

@@ -36,6 +36,12 @@ export interface ParsedWorkflow {
   errors: string[];
 }
 
+// Whitelist of allowed step keys to prevent injection
+const ALLOWED_STEP_KEYS = new Set([
+  'type', 'agent', 'tool', 'input', 'output', 'condition',
+  'onError', 'maxRetries', 'timeout', 'loopCondition', 'maxIterations',
+]);
+
 /**
  * Parse YAML string into workflow definition
  */
@@ -85,10 +91,6 @@ export function parseYAMLWorkflow(yamlContent: string): YAMLWorkflowDefinition {
       };
     } else if (currentStep) {
       // Step properties — whitelist allowed keys to prevent injection
-      const ALLOWED_STEP_KEYS = new Set([
-        'type', 'agent', 'tool', 'input', 'output', 'condition',
-        'onError', 'maxRetries', 'timeout', 'loopCondition', 'maxIterations',
-      ]);
       if (!ALLOWED_STEP_KEYS.has(key)) {
         // Ignore unknown keys — do not allow arbitrary property injection
         continue;
@@ -282,27 +284,33 @@ export async function loadWorkflowFromFile(
 
   // SECURITY: Prevent path traversal
   const normalizedPath = path.normalize(filePath);
-  if (normalizedPath.includes('..')) {
-    throw new Error('Path traversal detected: ".." is not allowed in file paths');
+  // Check for '..' as path segments (not just substring)
+  const pathSegments = normalizedPath.split(path.sep);
+  if (pathSegments.includes('..')) {
+    throw new Error('Path traversal detected: ".." path segments are not allowed');
   }
 
+  let effectivePath: string;
   // If basePath is specified, ensure resolvedPath stays within it
   if (options.basePath) {
     const resolvedBase = path.resolve(options.basePath);
     const resolvedFile = path.resolve(options.basePath, normalizedPath);
     if (!resolvedFile.startsWith(resolvedBase + path.sep) && resolvedFile !== resolvedBase) {
       throw new Error(`File path must be within base directory: ${options.basePath}`);
     }
+    effectivePath = resolvedFile;
+  } else {
+    effectivePath = path.resolve(normalizedPath);
   }
 
   // SECURITY: Enforce file size limit (default 1 MB)
   const maxSize = options.maxFileSizeBytes ?? 1_048_576;
-  const stat = await fs.stat(filePath);
+  const stat = await fs.stat(effectivePath);
   if (stat.size > maxSize) {
     throw new Error(`File too large: ${stat.size} bytes exceeds limit of ${maxSize} bytes`);
   }
 
-  const content = await fs.readFile(filePath, 'utf-8');
+  const content = await fs.readFile(effectivePath, 'utf-8');
   const definition = parseYAMLWorkflow(content);
   return createWorkflowFromYAML(definition, agents, tools);
 }

src/praisonai/praisonai/cli/features/agent_tools.py

@@ -8,20 +8,20 @@
 import asyncio
 import json
 import logging
-from typing import Callable, Dict, List, TYPE_CHECKING
+from typing import Callable, Dict, List, Optional, TYPE_CHECKING
 
 if TYPE_CHECKING:
     from .interactive_runtime import InteractiveRuntime
     from .code_intelligence import CodeIntelligenceRouter
     from .action_orchestrator import ActionOrchestrator
 
 import os
-import re
+from pathlib import Path
 
 logger = logging.getLogger(__name__)
 
 
-def _sanitize_filepath(filepath: str, workspace: str = None) -> str:
+def _sanitize_filepath(filepath: str, workspace: Optional[str] = None) -> str:
     """Validate and sanitize a filepath against injection attacks.
     
     Raises ValueError if the path is unsafe.
@@ -45,9 +45,11 @@ def _sanitize_filepath(filepath: str, workspace: str = None) -> str:
 
     # If we have a workspace, ensure the resolved path stays within it
     if workspace:
-        resolved = os.path.realpath(os.path.join(workspace, normalized))
-        ws_resolved = os.path.realpath(workspace)
-        if not resolved.startswith(ws_resolved + os.sep) and resolved != ws_resolved:
+        resolved = Path(os.path.realpath(os.path.join(workspace, normalized)))
+        ws_resolved = Path(os.path.realpath(workspace))
+        try:
+            resolved.relative_to(ws_resolved)
+        except ValueError:
             raise ValueError(
                 f"Path {filepath!r} resolves outside workspace {workspace!r}"
             )
@@ -70,9 +72,9 @@ def _sanitize_command(command: str) -> str:
     DANGEROUS_PATTERNS = [
         '$(', '`',      # Command substitution
         '&&', '||',     # Command chaining
-        ';',            # Command separator
-        '>>', '> ',     # Output redirection
-        '| ',           # Pipe
+        '>>', '>',      # Output redirection
+        '|', ';', '&',  # Pipe and separators
+        '\n', '\r'      # Line breaks
     ]
     for pattern in DANGEROUS_PATTERNS:
         if pattern in command:
@@ -513,7 +515,9 @@ def read_file(filepath: str) -> str:
             # SECURITY: Ensure resolved path stays within workspace
             resolved = path.resolve()
             ws_resolved = Path(runtime.config.workspace).resolve()
-            if not str(resolved).startswith(str(ws_resolved) + os.sep) and resolved != ws_resolved:
+            try:
+                resolved.relative_to(ws_resolved)
+            except ValueError:
                 return json.dumps({"error": f"Path escapes workspace: {filepath}"})
 
             if not resolved.exists():
@@ -556,7 +560,9 @@ def list_files(directory: str = ".", pattern: str = "*") -> str:
             # SECURITY: Ensure resolved path stays within workspace
             resolved = path.resolve()
             ws_resolved = Path(runtime.config.workspace).resolve()
-            if not str(resolved).startswith(str(ws_resolved) + os.sep) and resolved != ws_resolved:
+            try:
+                resolved.relative_to(ws_resolved)
+            except ValueError:
                 return json.dumps({"error": f"Directory escapes workspace: {directory}"})
 
             if not resolved.exists():