AI-written code concatenated user-controlled job_name into a file path without sanitizing ../ or absolute paths, allowing authenticated attackers to write CSV files outside the intended directory.

+This service handles the generation of CSV files with advanced formatting,+customization options, and performance optimizations.+"""++import asyncio+import csv+import gzip+import io+import json+import logging+import os+import time+import zipfile+from datetime import datetime, timedelta+from pathlib import Path+from typing import Any, Dict, List, Optional, Tuple, Union, Generator+import tempfile++import aiofiles+import pandas as pd+from decimal import Decimal++from app.core.config import settings+from app.models.csv_models import (+    CSVExportConfig,+    ColumnMapping,+    DataType,+    Encoding,+    ExportFormat,+    JobStatus,+    NullHandling,+    QuoteStyle,+    CompressionType+)++logger = logging.getLogger(__name__)+++class CSVGenerator:+    """Advanced CSV generator with customizable formatting and optimization."""+    +    def __init__(self):+        """Initialize the CSV generator."""+        self.quote_style_mapping = {+            QuoteStyle.MINIMAL: csv.QUOTE_MINIMAL,+            QuoteStyle.ALL: csv.QUOTE_ALL,+            QuoteStyle.NON_NUMERIC: csv.QUOTE_NONNUMERIC,+            QuoteStyle.NONE: csv.QUOTE_NONE+        }+        +        self.null_value_mapping = {+            NullHandling.EMPTY_STRING: "",+            NullHandling.NULL: "NULL",+            NullHandling.NA: "N/A",+            NullHandling.NONE: "None"+        }+    +    async def generate_csv(+        self,+        job_id: int,+        user_id: int,+        data_source: Dict[str, Any],+        export_config: CSVExportConfig,+        job_name: str+    ) -> Tuple[bool, Optional[str], Optional[str]]:+        """+        Generate a CSV file from data source and configuration.+        +        Args:+            job_id: Job identifier+            user_id: User identifier+            data_source: Data source configuration+            export_config: Export configuration+            job_name: Job name for file naming+            +        Returns:+            Tuple of (success, file_path, error_message)+        """+        start_time = time.time()+        +        try:+            # Update job status to processing+            await self._update_job_status(job_id, JobStatus.PROCESSING, started_at=datetime.utcnow())+            +            logger.info(f"Starting CSV generation for job {job_id}")

+This module provides API endpoints for CSV export operations including+export creation, validation, and file management.+"""++import asyncio+import logging+from typing import List, Optional+import uuid++from fastapi import APIRouter, Depends, HTTPException, status, BackgroundTasks, Query+from fastapi.responses import FileResponse++from app.core.config import settings+from app.core.database import create_export_job, log_analytics_event+from app.core.redis_client import get_queue_manager+from app.models.csv_models import (+    CSVExportRequest, +    BulkCSVExportRequest,+    JobResponse,+    ValidationResponse,+    CapabilitiesResponse,+    BaseResponse,+    ErrorResponse+)+from app.services.csv_generator import csv_generator+from app.utils.auth import CurrentUser, require_csv_create, require_csv_read, get_current_user+from app.utils.rate_limiter import check_user_rate_limit++logger = logging.getLogger(__name__)++router = APIRouter()++[email protected]("/", response_model=JobResponse)+async def create_csv_export(+    request: CSVExportRequest,+    background_tasks: BackgroundTasks,+    current_user: CurrentUser = Depends(require_csv_create)+):+    """Create a new CSV export job."""+    try:+        # Check rate limit+        await check_user_rate_limit(+            current_user.user_id, +            "/api/v1/export", +            current_user.role+        )+        +        # Create correlation ID+        correlation_id = str(uuid.uuid4())+        +        # Create job in database+        job_id = await create_export_job(+            user_id=current_user.user_id,+            job_name=request.job_name,+            data_source=request.data_source.model_dump(),+            export_config=request.export_config.model_dump(),+            priority=request.priority,+            expires_in_hours=request.expires_in_hours+        )+        +        # Log analytics event+        await log_analytics_event(+            user_id=current_user.user_id,+            job_id=job_id,+            event_type="export_job_created",+            event_data={+                "job_name": request.job_name,+                "export_format": request.export_config.export_format,+                "priority": request.priority,+                "correlation_id": correlation_id+            }+        )+        +        # Add job to processing queue+        queue_manager = get_queue_manager()+        await queue_manager.enqueue(+            "csv_export",+            {+                "job_id": job_id,+                "user_id": current_user.user_id,+                "job_name": request.job_name,+                "data_source": request.data_source.model_dump(),+                "export_config": request.export_config.model_dump(),+                "correlation_id": correlation_id