fix(#121): Remediate security issues from ORCH-121 review

Priority Fixes (Required Before Production):

H3: Add rate limiting to webhook endpoint
- Added slowapi library for FastAPI rate limiting
- Implemented per-IP rate limiting (100 req/min) on webhook endpoint
- Added global rate limiting support via slowapi

M4: Add subprocess timeouts to all gates
- Added timeout=300 (5 minutes) to all subprocess.run() calls in gates
- Implemented proper TimeoutExpired exception handling
- Removed dead CalledProcessError handlers (check=False makes them unreachable)

M2: Add input validation on QualityCheckRequest
- Validate files array size (max 1000 files)
- Validate file paths (no path traversal, no null bytes, no absolute paths)
- Validate diff summary size (max 10KB)
- Validate taskId and agentId format (non-empty)

Additional Fixes:

H1: Fix coverage.json path resolution
- Use absolute paths resolved from project root
- Validate path is within project boundaries (prevent path traversal)

Code Review Cleanup:
- Moved imports to module level in quality_orchestrator.py
- Refactored mock detection logic into separate helper methods
- Removed dead subprocess.CalledProcessError exception handlers from all gates

Testing:
- Added comprehensive tests for all security fixes
- All 339 coordinator tests pass
- All 447 orchestrator tests pass
- Followed TDD principles (RED-GREEN-REFACTOR)

Security Impact:
- Prevents webhook DoS attacks via rate limiting
- Prevents hung processes via subprocess timeouts
- Prevents path traversal attacks via input validation
- Prevents malformed input attacks via comprehensive validation

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>

apps/coordinator/src/gates/build_gate.py

@@ -24,6 +24,7 @@ class BuildGate:
                 capture_output=True,
                 text=True,
                 check=False,  # Don't raise on non-zero exit
+                timeout=300,  # 5 minute timeout
             )
 
             if result.returncode == 0:
@@ -54,11 +55,11 @@ class BuildGate:
                 details={"error": str(e)},
             )
 
-        except subprocess.CalledProcessError as e:
+        except subprocess.TimeoutExpired as e:
             return GateResult(
                 passed=False,
-                message="Build gate failed: Error running mypy",
-                details={"error": str(e), "return_code": e.returncode},
+                message=f"Build gate failed: mypy timed out after {e.timeout} seconds",
+                details={"error": str(e), "timeout": e.timeout},
             )
 
         except Exception as e:

apps/coordinator/src/gates/coverage_gate.py

@@ -1,6 +1,7 @@
 """CoverageGate - Enforces 85% minimum test coverage via pytest-cov."""
 
 import json
+import os
 import subprocess
 from pathlib import Path
 
@@ -35,6 +36,7 @@ class CoverageGate:
                 capture_output=True,
                 text=True,
                 check=False,  # Don't raise on non-zero exit
+                timeout=300,  # 5 minute timeout
             )
 
             # Try to read coverage data from coverage.json
@@ -94,11 +96,11 @@ class CoverageGate:
                 details={"error": str(e)},
             )
 
-        except subprocess.CalledProcessError as e:
+        except subprocess.TimeoutExpired as e:
             return GateResult(
                 passed=False,
-                message="Coverage gate failed: Error running pytest",
-                details={"error": str(e), "return_code": e.returncode},
+                message=f"Coverage gate failed: pytest timed out after {e.timeout} seconds",
+                details={"error": str(e), "timeout": e.timeout},
             )
 
         except Exception as e:
@@ -111,18 +113,28 @@ class CoverageGate:
     def _extract_coverage_from_json(self) -> float | None:
         """Extract coverage percentage from coverage.json file.
 
+        Uses absolute path resolved from current working directory and validates
+        that the path is within project boundaries to prevent path traversal attacks.
+
         Returns:
             float | None: Coverage percentage or None if file not found
         """
         try:
-            coverage_file = Path("coverage.json")
+            # Get absolute path from current working directory
+            cwd = Path.cwd().resolve()
+            coverage_file = (cwd / "coverage.json").resolve()
+
+            # Validate that coverage file is within project directory (prevent path traversal)
+            if not str(coverage_file).startswith(str(cwd)):
+                return None
+
             if coverage_file.exists():
                 with open(coverage_file) as f:
                     data = json.load(f)
                     percent = data.get("totals", {}).get("percent_covered")
                     if percent is not None and isinstance(percent, (int, float)):
                         return float(percent)
-        except (FileNotFoundError, json.JSONDecodeError, KeyError):
+        except (FileNotFoundError, json.JSONDecodeError, KeyError, OSError):
             pass
         return None
 

apps/coordinator/src/gates/lint_gate.py

@@ -24,6 +24,7 @@ class LintGate:
                 capture_output=True,
                 text=True,
                 check=False,  # Don't raise on non-zero exit
+                timeout=300,  # 5 minute timeout
             )
 
             if result.returncode == 0:
@@ -54,11 +55,11 @@ class LintGate:
                 details={"error": str(e)},
             )
 
-        except subprocess.CalledProcessError as e:
+        except subprocess.TimeoutExpired as e:
             return GateResult(
                 passed=False,
-                message="Lint gate failed: Error running ruff",
-                details={"error": str(e), "return_code": e.returncode},
+                message=f"Lint gate failed: ruff timed out after {e.timeout} seconds",
+                details={"error": str(e), "timeout": e.timeout},
             )
 
         except Exception as e:

apps/coordinator/src/gates/test_gate.py

@@ -24,6 +24,7 @@ class TestGate:
                 capture_output=True,
                 text=True,
                 check=False,  # Don't raise on non-zero exit
+                timeout=300,  # 5 minute timeout
             )
 
             if result.returncode == 0:
@@ -54,11 +55,11 @@ class TestGate:
                 details={"error": str(e)},
             )
 
-        except subprocess.CalledProcessError as e:
+        except subprocess.TimeoutExpired as e:
             return GateResult(
                 passed=False,
-                message="Test gate failed: Error running pytest",
-                details={"error": str(e), "return_code": e.returncode},
+                message=f"Test gate failed: pytest timed out after {e.timeout} seconds",
+                details={"error": str(e), "timeout": e.timeout},
             )
 
         except Exception as e:

apps/coordinator/src/main.py

@@ -7,8 +7,11 @@ from contextlib import asynccontextmanager
 from pathlib import Path
 from typing import Any
 
-from fastapi import FastAPI
+from fastapi import FastAPI, Request
 from pydantic import BaseModel
+from slowapi import Limiter, _rate_limit_exceeded_handler
+from slowapi.errors import RateLimitExceeded
+from slowapi.util import get_remote_address
 
 from .config import settings
 from .coordinator import Coordinator
@@ -104,6 +107,9 @@ async def lifespan(app: FastAPI) -> AsyncIterator[dict[str, Any]]:
     logger.info("Mosaic-coordinator shutdown complete")
 
 
+# Initialize rate limiter
+limiter = Limiter(key_func=get_remote_address)
+
 # Create FastAPI application
 app = FastAPI(
     title="Mosaic Coordinator",
@@ -112,6 +118,10 @@ app = FastAPI(
     lifespan=lifespan,
 )
 
+# Register rate limiter
+app.state.limiter = limiter
+app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
+
 
 class HealthResponse(BaseModel):
     """Health check response model."""

apps/coordinator/src/quality_orchestrator.py

@@ -1,7 +1,9 @@
 """Quality Orchestrator service for coordinating quality gate execution."""
 
 import asyncio
-from typing import Any
+import inspect
+from typing import Any, cast
+from unittest.mock import Mock
 
 from pydantic import BaseModel, Field
 
@@ -127,37 +129,51 @@ class QualityOrchestrator:
             Production gates are run in a thread pool to avoid blocking the event loop.
             Test mocks can be async functions or lambdas returning coroutines.
         """
-        import inspect
-        from typing import cast
-        from unittest.mock import Mock
-
         # Check if gate.check is an async function
         if inspect.iscoroutinefunction(gate.check):
             return cast(GateResult, await gate.check())
 
-        # Check if gate.check is a Mock/MagicMock (testing scenario)
+        # Check if it's a real production gate instance
+        if self._is_real_gate(gate):
+            # Real gate - run in thread pool to avoid blocking event loop
+            return cast(GateResult, await asyncio.to_thread(gate.check))
+
+        # Handle test mocks and callables
+        return await self._handle_test_mock(gate)
+
+    def _is_real_gate(self, gate: Any) -> bool:
+        """Check if gate is a real production gate instance.
+
+        Args:
+            gate: Gate instance to check
+
+        Returns:
+            bool: True if gate is a real production gate
+        """
+        if not inspect.ismethod(gate.check):
+            return False
+
+        gate_class_name = gate.__class__.__name__
+        return gate_class_name in ("BuildGate", "LintGate", "TestGate", "CoverageGate")
+
+    async def _handle_test_mock(self, gate: Any) -> GateResult:
+        """Handle test mocks and callables.
+
+        Args:
+            gate: Gate mock or callable to handle
+
+        Returns:
+            GateResult: Result from the mock
+        """
+        # Check if it's a Mock/MagicMock (testing scenario)
         mock_types = ("Mock", "MagicMock", "AsyncMock")
         if isinstance(gate.check, Mock) or type(gate.check).__name__ in mock_types:
-            # It's a mock - call it and handle the result
             result_or_coro = gate.check()
             if asyncio.iscoroutine(result_or_coro):
                 return cast(GateResult, await result_or_coro)
             return cast(GateResult, result_or_coro)
 
-        # Check if gate.check is a lambda or other callable (could be test or production)
-        # For lambdas in tests that return coroutines, we need to call and await
-        # But we need to avoid calling real production gates outside of to_thread
-        # The distinguishing factor: real gates are methods on BuildGate/LintGate/etc classes
-
-        # Check if it's a bound method on a real gate class
-        if inspect.ismethod(gate.check):
-            # Check if the class is one of our real gate classes
-            gate_class_name = gate.__class__.__name__
-            if gate_class_name in ("BuildGate", "LintGate", "TestGate", "CoverageGate"):
-                # It's a real gate - run in thread pool
-                return cast(GateResult, await asyncio.to_thread(gate.check))
-
-        # For any other callable (lambdas, functions), try calling and see what it returns
+        # For any other callable (lambdas, functions), call and check result
         result_or_coro = gate.check()
         if asyncio.iscoroutine(result_or_coro):
             return cast(GateResult, await result_or_coro)

apps/coordinator/src/webhook.py

@@ -5,6 +5,8 @@ from typing import Any
 
 from fastapi import APIRouter, Header, HTTPException, Request
 from pydantic import BaseModel, Field
+from slowapi import Limiter
+from slowapi.util import get_remote_address
 
 from .config import settings
 from .security import verify_signature
@@ -13,6 +15,9 @@ logger = logging.getLogger(__name__)
 
 router = APIRouter()
 
+# Initialize limiter for this module
+limiter = Limiter(key_func=get_remote_address)
+
 
 class WebhookResponse(BaseModel):
     """Response model for webhook endpoint."""
@@ -34,6 +39,7 @@ class GiteaWebhookPayload(BaseModel):
 
 
 @router.post("/webhook/gitea", response_model=WebhookResponse)
+@limiter.limit("100/minute")  # Per-IP rate limit: 100 requests per minute
 async def handle_gitea_webhook(
     request: Request,
     payload: GiteaWebhookPayload,