fix(security): re-land CWE-209 response protections from closed PR #831

src/agents/specialized/code_generator.py

@@ -22,6 +22,10 @@ def _load_templates(self) -> dict[str, str]:
         return {
             "fastapi_endpoint": textwrap.dedent(
                 """
+                import logging
+
+                logger = logging.getLogger(__name__)
+
                 @app.post("/api/v1/{endpoint_name}")
                 async def {function_name}({parameters}):
                     \"\"\"
@@ -42,8 +46,12 @@ async def {function_name}({parameters}):
                         }}
                     except ValidationError as e:
                         raise HTTPException(status_code=400, detail=str(e))
-                    except Exception as e:
-                        raise HTTPException(status_code=500, detail=str(e))
+                    except Exception:
+                        logger.exception("Generated endpoint failed")
+                        raise HTTPException(
+                            status_code=500,
+                            detail="Internal server error",
+                        )
             """
             ),
             "rest_api": textwrap.dedent(

src/youtube_extension/backend/cloud_ai_routes.py

@@ -260,7 +260,7 @@ async def analyze_video(request: VideoAnalysisRequest):
         # Must precede CloudAIError: RateLimitError subclasses it, so catching
         # the base first would shadow this handler and return a 503 instead.
         logger.warning(f"Rate limit exceeded: {e}")
-        raise HTTPException(status_code=429, detail=f"Rate limit exceeded: {str(e)}")
+        raise HTTPException(status_code=429, detail="Rate limit exceeded")
     except ConfigurationError as e:
         # Must precede CloudAIError (same subclassing reason) so configuration
         # failures reach this sanitized 500 rather than the dynamic 503 below.

src/youtube_extension/backend/cloud_api_endpoints.py

@@ -30,6 +30,50 @@
 router = APIRouter()
 
 
+def _client_safe_error(error_message: Any) -> Optional[str]:
+    """Return a stable client-safe value for persisted processor failures.
+
+    Older Firestore records may predate the write-side sanitizer, so every read
+    boundary must treat stored error text as untrusted rather than echoing it.
+    """
+    return "Internal server error" if error_message else None
+
+
+def _sanitize_error_list(value: Any) -> Any:
+    """Replace scalar diagnostic entries while preserving structured shapes."""
+    if isinstance(value, list):
+        return [_sanitize_error_list(item) for item in value]
+    if isinstance(value, tuple):
+        return tuple(_sanitize_error_list(item) for item in value)
+    if isinstance(value, dict):
+        return _sanitize_response_errors(value)
+    # Every remaining leaf is a scalar diagnostic. Only ``None`` (absence of an
+    # error) is preserved; any other non-null leaf — including non-string types
+    # such as bytes, ints, or bools that FastAPI can still serialize — is
+    # replaced so it cannot bypass the scalar sanitization invariant.
+    if value is None:
+        return None
+    return "Internal server error"
+
+
+def _sanitize_response_errors(value: Any) -> Any:
+    """Copy a client response tree and remove persisted diagnostic messages."""
+    if isinstance(value, dict):
+        sanitized: Dict[Any, Any] = {}
+        for key, item in value.items():
+            if key in {"error", "error_message"}:
+                sanitized[key] = _client_safe_error(item)
+            elif key == "errors":
+                sanitized[key] = _sanitize_error_list(item)
+            else:
+                sanitized[key] = _sanitize_response_errors(item)
+        return sanitized
+    if isinstance(value, list):
+        return [_sanitize_response_errors(item) for item in value]
+    if isinstance(value, tuple):
+        return tuple(_sanitize_response_errors(item) for item in value)
+    return value
+
 
 # Pydantic models for API requests/responses
 class CloudVideoProcessingRequest(BaseModel):
@@ -128,12 +172,12 @@ async def process_video_cloud(
                 video_url=result.video_url,
                 success=result.success,
                 status='completed' if result.success else 'failed',
-                metadata=result.metadata,
-                transcript=result.transcript,
-                ai_analysis=result.ai_analysis,
+                metadata=_sanitize_response_errors(result.metadata),
+                transcript=_sanitize_response_errors(result.transcript),
+                ai_analysis=_sanitize_response_errors(result.ai_analysis),
                 processing_time=result.processing_time,
                 from_cache=result.from_cache,
-                error=result.error_message,
+                error=_client_safe_error(result.error_message),
             )
 
     except Exception as e:
@@ -235,8 +279,8 @@ async def process_video_task_handler(
                 status='failed',
                 error_message="Task processing failed"
             )
-        except Exception as state_error:
-            logger.error(f"Failed to update error state: {state_error}")
+        except Exception:
+            logger.error("Failed to update error state", exc_info=True)
 
         raise HTTPException(status_code=500, detail="Internal server error")
 
@@ -297,7 +341,7 @@ async def get_video_status(video_id: str):
             created_at=state.created_at,
             updated_at=state.updated_at,
             processing_time=state.processing_time,
-            error_message=state.error_message,
+            error_message=_client_safe_error(state.error_message),
         )
 
     except HTTPException:
@@ -329,13 +373,13 @@ async def get_video_result(video_id: str):
             "video_url": state.video_url,
             "status": state.status,
             "current_stage": state.current_stage,
-            "metadata": state.metadata,
-            "transcript": state.transcript,
-            "ai_analysis": state.ai_analysis,
+            "metadata": _sanitize_response_errors(state.metadata),
+            "transcript": _sanitize_response_errors(state.transcript),
+            "ai_analysis": _sanitize_response_errors(state.ai_analysis),
             "processing_time": state.processing_time,
             "created_at": state.created_at,
             "updated_at": state.updated_at,
-            "error_message": state.error_message,
+            "error_message": _client_safe_error(state.error_message),
         }
 
     except HTTPException:
@@ -362,11 +406,11 @@ async def get_queue_stats():
             "timestamp": datetime.now(timezone.utc).isoformat(),
         }
 
-    except Exception as e:
-        logger.error(f"Error getting queue stats: {e}")
+    except Exception:
+        logger.error("Error getting queue stats", exc_info=True)
         return {
             "success": False,
-            "error": str(e),
+            "error": "Internal server error",
             "timestamp": datetime.now(timezone.utc).isoformat(),
         }
 
@@ -389,10 +433,11 @@ async def get_cloud_status():
                 "status": "operational",
                 "enabled": True,
             }
-        except Exception as e:
+        except Exception:
+            logger.error("firestore status check failed", exc_info=True)
             status["services"]["firestore"] = {
                 "status": "error",
-                "error": str(e),
+                "error": "Service unavailable",
             }
             status["overall_status"] = "degraded"
 
@@ -405,10 +450,11 @@ async def get_cloud_status():
                 "enabled": True,
                 "queue_stats": stats,
             }
-        except Exception as e:
+        except Exception:
+            logger.error("cloud_tasks status check failed", exc_info=True)
             status["services"]["cloud_tasks"] = {
                 "status": "error",
-                "error": str(e),
+                "error": "Service unavailable",
             }
             status["overall_status"] = "degraded"
 
@@ -419,20 +465,21 @@ async def get_cloud_status():
                 "status": "operational",
                 "enabled": True,
             }
-        except Exception as e:
+        except Exception:
+            logger.error("vertex_ai status check failed", exc_info=True)
             status["services"]["vertex_ai"] = {
                 "status": "error",
-                "error": str(e),
+                "error": "Service unavailable",
             }
             status["overall_status"] = "degraded"
 
         return status
 
-    except Exception as e:
-        logger.error(f"Error getting cloud status: {e}")
+    except Exception:
+        logger.error("Error getting cloud status", exc_info=True)
         return {
             "overall_status": "error",
-            "error": str(e),
+            "error": "Internal server error",
             "timestamp": datetime.now(timezone.utc).isoformat(),
         }
 

src/youtube_extension/backend/real_api_endpoints.py

@@ -177,6 +177,65 @@ def _read_video_analysis_sync(cache_path: Path) -> Any:
     return raw
 
 
+_PUBLIC_PROCESSING_ERROR = "Video processing failed"
+
+
+def _sanitize_public_error(value: Any) -> Optional[str]:
+    """Return a client-safe processing error without exposing provider details."""
+    return None if value is None else _PUBLIC_PROCESSING_ERROR
+
+
+def _sanitize_error_list(value: Any) -> Any:
+    """Sanitize a plural ``errors`` collection.
+
+    The AI processor records per-step failures as scalar strings that embed
+    exception text (e.g. ``"content_analysis: <exception>"`` in
+    ``real_ai_processor.analyze_video_content``).  Replace those scalar strings
+    with the public message, while preserving — and recursing into — structured
+    error records so batch result shapes stay intact.
+    """
+    if isinstance(value, list):
+        return [_sanitize_error_list(item) for item in value]
+    if isinstance(value, tuple):
+        return tuple(_sanitize_error_list(item) for item in value)
+    if isinstance(value, dict):
+        return _sanitize_response_errors(value)
+    # Every remaining leaf is a scalar diagnostic. Only ``None`` (absence of an
+    # error) is preserved; any other non-null leaf — including non-string types
+    # such as bytes, ints, or bools that FastAPI can still serialize — is
+    # replaced so it cannot bypass the scalar sanitization invariant.
+    if value is None:
+        return None
+    return _PUBLIC_PROCESSING_ERROR
+
+
+def _sanitize_response_errors(value: Any) -> Any:
+    """Copy a response tree while replacing every diagnostic error value.
+
+    Processor results can contain exception text at the top level and inside
+    batch, step, or provider records.  Sanitizing the complete response tree at
+    the HTTP boundary also protects cached legacy records without mutating the
+    processor's internal diagnostic value.  The plural ``errors`` collection is
+    handled by :func:`_sanitize_error_list` because it carries scalar exception
+    strings that a plain recursion would pass through unchanged.
+    """
+    if isinstance(value, dict):
+        sanitized: dict[Any, Any] = {}
+        for key, item in value.items():
+            if key in {"error", "error_message"}:
+                sanitized[key] = _sanitize_public_error(item)
+            elif key == "errors":
+                sanitized[key] = _sanitize_error_list(item)
+            else:
+                sanitized[key] = _sanitize_response_errors(item)
+        return sanitized
+    if isinstance(value, list):
+        return [_sanitize_response_errors(item) for item in value]
+    if isinstance(value, tuple):
+        return tuple(_sanitize_response_errors(item) for item in value)
+    return value
+
+
 # Pydantic models for API requests/responses
 class VideoProcessingRequest(BaseModel):
     video_url: str = Field(..., description="YouTube video URL or ID")
@@ -246,22 +305,23 @@ async def process_video_real_api(request: VideoProcessingRequest, background_tas
                 video_url=request.video_url,
                 force_refresh=request.force_refresh
             )
+            safe_result = _sanitize_response_errors(result)
 
             # Track metrics
             processing_time = (datetime.now(timezone.utc) - start_time).total_seconds()
 
             # Format response
             response = VideoAnalysisResponse(
-                video_id=result.get('video_id', ''),
+                video_id=safe_result.get('video_id', ''),
                 video_url=request.video_url,
-                success=result.get('success', False),
-                metadata=result.get('metadata'),
-                transcript=result.get('transcript'),
-                ai_analysis=result.get('ai_analysis'),
-                cost_breakdown=result.get('cost_breakdown'),
+                success=safe_result.get('success', False),
+                metadata=safe_result.get('metadata'),
+                transcript=safe_result.get('transcript'),
+                ai_analysis=safe_result.get('ai_analysis'),
+                cost_breakdown=safe_result.get('cost_breakdown'),
                 processing_time=processing_time,
-                cached=result.get('cached', False),
-                error=result.get('error')
+                cached=safe_result.get('cached', False),
+                error=_sanitize_public_error(safe_result.get('error'))
             )
 
             logger.info(f"✅ Real API processing completed: {result.get('video_id')} - ${result.get('cost_breakdown', {}).get('total_cost', 0):.4f}")
@@ -319,7 +379,7 @@ async def batch_process_videos(request: BatchProcessingRequest):
                 max_concurrent=request.max_concurrent
             )
 
-            return result
+            return _sanitize_response_errors(result)
 
         except HTTPException:
             # Preserve explicit 4xx responses (e.g. the 400 batch-size guard above).
@@ -377,10 +437,11 @@ async def get_video_analysis(video_id: str):
                     detail=f"Video analysis not found: {video_id}"
                 )
 
-            # Raw passthrough: the cache entry already is the response JSON.
-            # Returning a Response skips FastAPI's jsonable_encoder/json.dumps
-            # round-trip, which would otherwise re-serialise the payload on
-            # the event loop in proportion to its size.
+            # The helper already parsed, sanitized, and re-encoded the entry in
+            # the worker thread. Returning a Response skips FastAPI's
+            # jsonable_encoder/json.dumps round-trip, which would otherwise
+            # re-serialise the payload on the event loop in proportion to its
+            # size.
             return Response(content=video_data, media_type="application/json")
 
         except HTTPException:
@@ -401,10 +462,10 @@ async def get_cost_dashboard():
             dashboard = await cost_monitor.get_cost_dashboard()
             return dashboard
 
-        except Exception as e:
-            logger.error(f"Error getting cost dashboard: {e}")
+        except Exception:
+            logger.error("Error getting cost dashboard", exc_info=True)
             return {
-                "error": str(e),
+                "error": "Internal server error",
                 "timestamp": datetime.now(timezone.utc).isoformat()
             }
 
@@ -425,10 +486,10 @@ async def get_usage_analytics(days: int = 7):
 
         except HTTPException:
             raise
-        except Exception as e:
-            logger.error(f"Error getting usage analytics: {e}")
+        except Exception:
+            logger.error("Error getting usage analytics", exc_info=True)
             return {
-                "error": str(e),
+                "error": "Internal server error",
                 "timestamp": datetime.now(timezone.utc).isoformat()
             }
 
@@ -441,10 +502,10 @@ async def get_optimization_recommendations():
             recommendations = await cost_monitor.optimize_api_usage()
             return recommendations
 
-        except Exception as e:
-            logger.error(f"Error getting optimization recommendations: {e}")
+        except Exception:
+            logger.error("Error getting optimization recommendations", exc_info=True)
             return {
-                "error": str(e),
+                "error": "Internal server error",
                 "recommendations": [],
                 "timestamp": datetime.now(timezone.utc).isoformat()
             }
@@ -472,7 +533,7 @@ async def get_service_status():
             return {
                 "overall_status": "operational" if processor_status.get('service_status') == 'operational' else "degraded",
                 "timestamp": datetime.now(timezone.utc).isoformat(),
-                "processor": processor_status,
+                "processor": _sanitize_response_errors(processor_status),
                 "cost_monitoring": {
                     "status": "operational",
                     "today_cost": cost_dashboard.get('today_summary', {}).get('total_cost', 0.0),
@@ -490,11 +551,11 @@ async def get_service_status():
                 "version": "2.0.0-real-api-integration"
             }
 
-        except Exception as e:
-            logger.error(f"Error getting service status: {e}")
+        except Exception:
+            logger.error("Error getting service status", exc_info=True)
             return {
                 "overall_status": "error",
-                "error": str(e),
+                "error": "Internal server error",
                 "timestamp": datetime.now(timezone.utc).isoformat()
             }