Production audit: fix race conditions, memory leaks, and security issues

Fixes:
- Fix race condition in DynamicRateLimit middleware (thread-safe implementation)
- Add missing logger field to EnhancedAnalyzer struct
- Fix SSRF vulnerability by removing file:// URL scheme support
- Fix SQL placeholder syntax (PostgreSQL $1 -> SQLite ?) in repository
- Add batch job TTL cleanup to prevent unbounded memory growth

Changes:
- ratelimit.go: Rewrite DynamicRateLimit() to not mutate shared config
- enhanced_analyzer.go: Add logger field and initialize in constructors
- validator.go: Block file:// scheme for security
- repository.go: Fix ListHLSAnalyses, ListReports, ListQualityComparisons
- main.go: Add cleanupBatchJobs goroutine (1hr TTL, 5min cleanup interval)

Author: alokemajumder

cmd/rendiff-probe/main.go

@@ -43,14 +43,16 @@ import (
 
 // Production constants
 const (
-	maxFileSize       = 5 * 1024 * 1024 * 1024 // 5GB max file size
-	maxRequestBodyMB  = 10                      // 10MB max JSON request body
-	maxBatchItems     = 100                     // Max items in batch processing
-	defaultTimeout    = 60 * time.Second
-	maxTimeout        = 30 * time.Minute
-	shutdownTimeout   = 30 * time.Second
-	wsReadBufferSize  = 1024
-	wsWriteBufferSize = 1024
+	maxFileSize        = 5 * 1024 * 1024 * 1024 // 5GB max file size
+	maxRequestBodyMB   = 10                      // 10MB max JSON request body
+	maxBatchItems      = 100                     // Max items in batch processing
+	defaultTimeout     = 60 * time.Second
+	maxTimeout         = 30 * time.Minute
+	shutdownTimeout    = 30 * time.Second
+	wsReadBufferSize   = 1024
+	wsWriteBufferSize  = 1024
+	batchJobTTL        = 1 * time.Hour  // TTL for completed batch jobs before cleanup
+	batchCleanupPeriod = 5 * time.Minute // How often to run batch job cleanup
 )
 
 // Global instances for services
@@ -167,6 +169,10 @@ func main() {
 
 	appLogger.Info().Msg("All services initialized successfully")
 
+	// Start batch job cleanup goroutine
+	go cleanupBatchJobs()
+	appLogger.Info().Dur("ttl", batchJobTTL).Dur("period", batchCleanupPeriod).Msg("Batch job cleanup started")
+
 	// Create Gin router with production settings
 	router := gin.New()
 
@@ -285,6 +291,44 @@ func closeAllWebSocketConnections() {
 	wsConnections = make(map[string]*websocket.Conn)
 }
 
+// cleanupBatchJobs periodically removes expired batch jobs to prevent memory leaks
+func cleanupBatchJobs() {
+	ticker := time.NewTicker(batchCleanupPeriod)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-shutdownCtx.Done():
+			appLogger.Debug().Msg("Batch job cleanup goroutine stopped")
+			return
+		case <-ticker.C:
+			now := time.Now()
+			var toDelete []string
+
+			batchLock.RLock()
+			for id, job := range batchJobs {
+				// Remove completed/cancelled/failed jobs older than TTL
+				if job.Status == "completed" || job.Status == "cancelled" || job.Status == "failed" {
+					if now.Sub(job.UpdatedAt) > batchJobTTL {
+						toDelete = append(toDelete, id)
+					}
+				}
+			}
+			batchLock.RUnlock()
+
+			if len(toDelete) > 0 {
+				batchLock.Lock()
+				for _, id := range toDelete {
+					delete(batchJobs, id)
+					appLogger.Debug().Str("job_id", id).Msg("Cleaned up expired batch job")
+				}
+				batchLock.Unlock()
+				appLogger.Info().Int("count", len(toDelete)).Msg("Batch job cleanup completed")
+			}
+		}
+	}
+}
+
 // requestLoggingMiddleware logs HTTP requests
 func requestLoggingMiddleware() gin.HandlerFunc {
 	return func(c *gin.Context) {

internal/database/repository.go

@@ -416,11 +416,9 @@ func (r *SQLiteRepository) ListHLSAnalyses(ctx context.Context, userID *uuid.UUI
 	baseQuery := `FROM hls_analyses h JOIN analyses a ON h.analysis_id = a.id`
 	whereClause := ""
 	args := []interface{}{}
-	argCount := 0
 
 	if userID != nil {
-		argCount++
-		whereClause = fmt.Sprintf(" WHERE a.user_id = $%d", argCount)
+		whereClause = " WHERE a.user_id = ?"
 		args = append(args, *userID)
 	}
 
@@ -431,24 +429,19 @@ func (r *SQLiteRepository) ListHLSAnalyses(ctx context.Context, userID *uuid.UUI
 		return nil, 0, err
 	}
 
-	// Get paginated results
-	argCount++
-	limitArg := argCount
-	args = append(args, limit)
-	argCount++
-	offsetArg := argCount
-	args = append(args, offset)
+	// Get paginated results - SQLite uses ? placeholders
+	paginatedArgs := append(args, limit, offset)
 
 	query := fmt.Sprintf(`
 		SELECT h.id, h.analysis_id, h.manifest_path, h.manifest_type, h.manifest_data,
 			h.segment_count, h.total_duration, h.bitrate_variants, h.segment_duration,
 			h.playlist_version, h.status, h.processing_time, h.created_at, h.completed_at, h.error_msg
 		%s %s
 		ORDER BY h.created_at DESC
-		LIMIT $%d OFFSET $%d
-	`, baseQuery, whereClause, limitArg, offsetArg)
+		LIMIT ? OFFSET ?
+	`, baseQuery, whereClause)
 
-	err = r.db.DB.SelectContext(ctx, &analyses, query, args...)
+	err = r.db.DB.SelectContext(ctx, &analyses, query, paginatedArgs...)
 	if err != nil {
 		return nil, 0, err
 	}
@@ -575,31 +568,26 @@ func (r *SQLiteRepository) ListReports(ctx context.Context, userID *uuid.UUID, a
 	baseQuery := "FROM reports"
 	whereConditions := []string{}
 	args := []interface{}{}
-	argCount := 0
 
 	if userID != nil {
-		argCount++
-		whereConditions = append(whereConditions, fmt.Sprintf("user_id = $%d", argCount))
+		whereConditions = append(whereConditions, "user_id = ?")
 		args = append(args, *userID)
 	}
 
 	if analysisID != "" {
 		if analysisUUID, err := uuid.Parse(analysisID); err == nil {
-			argCount++
-			whereConditions = append(whereConditions, fmt.Sprintf("analysis_id = $%d", argCount))
+			whereConditions = append(whereConditions, "analysis_id = ?")
 			args = append(args, analysisUUID)
 		}
 	}
 
 	if reportType != "" {
-		argCount++
-		whereConditions = append(whereConditions, fmt.Sprintf("report_type = $%d", argCount))
+		whereConditions = append(whereConditions, "report_type = ?")
 		args = append(args, reportType)
 	}
 
 	if format != "" {
-		argCount++
-		whereConditions = append(whereConditions, fmt.Sprintf("format = $%d", argCount))
+		whereConditions = append(whereConditions, "format = ?")
 		args = append(args, format)
 	}
 
@@ -615,23 +603,18 @@ func (r *SQLiteRepository) ListReports(ctx context.Context, userID *uuid.UUID, a
 		return nil, 0, err
 	}
 
-	// Get paginated results
-	argCount++
-	limitArg := argCount
-	args = append(args, limit)
-	argCount++
-	offsetArg := argCount
-	args = append(args, offset)
+	// Get paginated results - SQLite uses ? placeholders
+	paginatedArgs := append(args, limit, offset)
 
 	query := fmt.Sprintf(`
 		SELECT id, analysis_id, user_id, report_type, format, title, description,
 			file_path, file_size, download_count, is_public, expires_at, created_at, last_download
 		%s %s
 		ORDER BY created_at DESC
-		LIMIT $%d OFFSET $%d
-	`, baseQuery, whereClause, limitArg, offsetArg)
+		LIMIT ? OFFSET ?
+	`, baseQuery, whereClause)
 
-	err = r.db.DB.SelectContext(ctx, &reports, query, args...)
+	err = r.db.DB.SelectContext(ctx, &reports, query, paginatedArgs...)
 	if err != nil {
 		return nil, 0, err
 	}
@@ -701,54 +684,50 @@ func (r *SQLiteRepository) UpdateQualityComparison(ctx context.Context, comparis
 
 	_, err := r.db.DB.ExecContext(
 		ctx, query,
-		comparison.ID, comparison.Status, comparison.ResultSummary, comparison.ProcessingTime,
-		comparison.CompletedAt, comparison.ErrorMsg,
+		comparison.Status, comparison.ResultSummary, comparison.ProcessingTime,
+		comparison.CompletedAt, comparison.ErrorMsg, comparison.ID,
 	)
 	return err
 }
 
 func (r *SQLiteRepository) UpdateQualityComparisonStatus(ctx context.Context, id uuid.UUID, status models.AnalysisStatus) error {
 	query := "UPDATE quality_comparisons SET status = ?, updated_at = datetime('now') WHERE id = ?"
-	_, err := r.db.DB.ExecContext(ctx, query, id, status)
+	_, err := r.db.DB.ExecContext(ctx, query, status, id)
 	return err
 }
 
 func (r *SQLiteRepository) ListQualityComparisons(ctx context.Context, userID *uuid.UUID, referenceID, distortedID, status string, limit, offset int) ([]*models.QualityComparison, int, error) {
 	var comparisons []*models.QualityComparison
 	var total int
 
-	baseQuery := `FROM quality_comparisons qc 
+	baseQuery := `FROM quality_comparisons qc
 		LEFT JOIN analyses ref ON qc.reference_id = ref.id
 		LEFT JOIN analyses dist ON qc.distorted_id = dist.id`
 	whereConditions := []string{}
 	args := []interface{}{}
-	argCount := 0
 
 	if userID != nil {
-		argCount++
-		whereConditions = append(whereConditions, fmt.Sprintf("(ref.user_id = $%d OR dist.user_id = $%d)", argCount, argCount))
-		args = append(args, *userID)
+		// SQLite uses ? placeholders - need to add the same arg twice for OR condition
+		whereConditions = append(whereConditions, "(ref.user_id = ? OR dist.user_id = ?)")
+		args = append(args, *userID, *userID)
 	}
 
 	if referenceID != "" {
 		if refUUID, err := uuid.Parse(referenceID); err == nil {
-			argCount++
-			whereConditions = append(whereConditions, fmt.Sprintf("qc.reference_id = $%d", argCount))
+			whereConditions = append(whereConditions, "qc.reference_id = ?")
 			args = append(args, refUUID)
 		}
 	}
 
 	if distortedID != "" {
 		if distUUID, err := uuid.Parse(distortedID); err == nil {
-			argCount++
-			whereConditions = append(whereConditions, fmt.Sprintf("qc.distorted_id = $%d", argCount))
+			whereConditions = append(whereConditions, "qc.distorted_id = ?")
 			args = append(args, distUUID)
 		}
 	}
 
 	if status != "" {
-		argCount++
-		whereConditions = append(whereConditions, fmt.Sprintf("qc.status = $%d", argCount))
+		whereConditions = append(whereConditions, "qc.status = ?")
 		args = append(args, status)
 	}
 
@@ -764,23 +743,18 @@ func (r *SQLiteRepository) ListQualityComparisons(ctx context.Context, userID *u
 		return nil, 0, err
 	}
 
-	// Get paginated results
-	argCount++
-	limitArg := argCount
-	args = append(args, limit)
-	argCount++
-	offsetArg := argCount
-	args = append(args, offset)
+	// Get paginated results - SQLite uses ? placeholders
+	paginatedArgs := append(args, limit, offset)
 
 	query := fmt.Sprintf(`
 		SELECT qc.id, qc.reference_id, qc.distorted_id, qc.comparison_type, qc.status,
 			qc.result_summary, qc.processing_time, qc.created_at, qc.completed_at, qc.error_msg
 		%s %s
 		ORDER BY qc.created_at DESC
-		LIMIT $%d OFFSET $%d
-	`, baseQuery, whereClause, limitArg, offsetArg)
+		LIMIT ? OFFSET ?
+	`, baseQuery, whereClause)
 
-	err = r.db.DB.SelectContext(ctx, &comparisons, query, args...)
+	err = r.db.DB.SelectContext(ctx, &comparisons, query, paginatedArgs...)
 	if err != nil {
 		return nil, 0, err
 	}

internal/ffmpeg/enhanced_analyzer.go

@@ -30,6 +30,7 @@ type EnhancedAnalyzer struct {
 	pseAnalyzer               *PSEAnalyzer
 	streamDispositionAnalyzer *StreamDispositionAnalyzer
 	dataIntegrityAnalyzer     *DataIntegrityAnalyzer
+	logger                    zerolog.Logger
 }
 
 // NewEnhancedAnalyzer creates a new enhanced analyzer
@@ -53,6 +54,7 @@ func NewEnhancedAnalyzer(ffprobePath string, logger zerolog.Logger) *EnhancedAna
 		pseAnalyzer:               NewPSEAnalyzer(ffprobePath, logger),
 		streamDispositionAnalyzer: NewStreamDispositionAnalyzer(ffprobePath, logger),
 		dataIntegrityAnalyzer:     NewDataIntegrityAnalyzer(ffprobePath, logger),
+		logger:                    logger,
 	}
 }
 
@@ -78,6 +80,7 @@ func NewEnhancedAnalyzerWithContentAnalysis(ffmpegPath string, ffprobePath strin
 		pseAnalyzer:               NewPSEAnalyzer(ffprobePath, logger),
 		streamDispositionAnalyzer: NewStreamDispositionAnalyzer(ffprobePath, logger),
 		dataIntegrityAnalyzer:     NewDataIntegrityAnalyzer(ffprobePath, logger),
+		logger:                    logger,
 	}
 }
 
@@ -158,7 +161,7 @@ func (ea *EnhancedAnalyzer) AnalyzeResultWithAdvancedQC(ctx context.Context, res
 		timecodeAnalysis, err := ea.timecodeAnalyzer.AnalyzeTimecode(ctx, filePath, result.Streams)
 		if err != nil {
 			// Log error but don't fail entire analysis - some files may not have timecode
-			fmt.Printf("Warning: timecode analysis failed: %v\n", err)
+			ea.logger.Warn().Err(err).Msg("timecode analysis failed")
 		} else {
 			result.EnhancedAnalysis.TimecodeAnalysis = timecodeAnalysis
 		}
@@ -169,7 +172,7 @@ func (ea *EnhancedAnalyzer) AnalyzeResultWithAdvancedQC(ctx context.Context, res
 		afdAnalysis, err := ea.afdAnalyzer.AnalyzeAFD(ctx, filePath, result.Streams)
 		if err != nil {
 			// Log error but don't fail entire analysis - some files may not have AFD
-			fmt.Printf("Warning: AFD analysis failed: %v\n", err)
+			ea.logger.Warn().Err(err).Msg("AFD analysis failed")
 		} else {
 			result.EnhancedAnalysis.AFDAnalysis = afdAnalysis
 		}
@@ -180,7 +183,7 @@ func (ea *EnhancedAnalyzer) AnalyzeResultWithAdvancedQC(ctx context.Context, res
 		transportAnalysis, err := ea.transportStreamAnalyzer.AnalyzeTransportStream(ctx, filePath, result.Streams, result.Format)
 		if err != nil {
 			// Log error but don't fail entire analysis - only applies to transport streams
-			fmt.Printf("Warning: transport stream analysis failed: %v\n", err)
+			ea.logger.Warn().Err(err).Msg("transport stream analysis failed")
 		} else {
 			result.EnhancedAnalysis.TransportStreamAnalysis = transportAnalysis
 		}
@@ -191,7 +194,7 @@ func (ea *EnhancedAnalyzer) AnalyzeResultWithAdvancedQC(ctx context.Context, res
 		endiannessAnalysis, err := ea.endiannessAnalyzer.AnalyzeEndianness(ctx, filePath, result.Streams, result.Format)
 		if err != nil {
 			// Log error but don't fail entire analysis - endianness may not be detectable for all formats
-			fmt.Printf("Warning: endianness analysis failed: %v\n", err)
+			ea.logger.Warn().Err(err).Msg("endianness analysis failed")
 		} else {
 			result.EnhancedAnalysis.EndiannessAnalysis = endiannessAnalysis
 		}
@@ -202,7 +205,7 @@ func (ea *EnhancedAnalyzer) AnalyzeResultWithAdvancedQC(ctx context.Context, res
 		audioWrappingAnalysis, err := ea.audioWrappingAnalyzer.AnalyzeAudioWrapping(ctx, filePath, result.Streams, result.Format)
 		if err != nil {
 			// Log error but don't fail entire analysis - not all formats have professional audio wrapping
-			fmt.Printf("Warning: audio wrapping analysis failed: %v\n", err)
+			ea.logger.Warn().Err(err).Msg("audio wrapping analysis failed")
 		} else {
 			result.EnhancedAnalysis.AudioWrappingAnalysis = audioWrappingAnalysis
 		}
@@ -213,7 +216,7 @@ func (ea *EnhancedAnalyzer) AnalyzeResultWithAdvancedQC(ctx context.Context, res
 		imfAnalysis, err := ea.imfAnalyzer.AnalyzeIMF(ctx, filePath)
 		if err != nil {
 			// Log error but don't fail entire analysis - only applies to IMF packages
-			fmt.Printf("Warning: IMF analysis failed: %v\n", err)
+			ea.logger.Warn().Err(err).Msg("IMF analysis failed")
 		} else {
 			result.EnhancedAnalysis.IMFAnalysis = imfAnalysis
 		}
@@ -224,7 +227,7 @@ func (ea *EnhancedAnalyzer) AnalyzeResultWithAdvancedQC(ctx context.Context, res
 		mxfAnalysis, err := ea.mxfAnalyzer.AnalyzeMXF(ctx, filePath)
 		if err != nil {
 			// Log error but don't fail entire analysis - only applies to MXF files
-			fmt.Printf("Warning: MXF analysis failed: %v\n", err)
+			ea.logger.Warn().Err(err).Msg("MXF analysis failed")
 		} else {
 			result.EnhancedAnalysis.MXFAnalysis = mxfAnalysis
 		}
@@ -235,7 +238,7 @@ func (ea *EnhancedAnalyzer) AnalyzeResultWithAdvancedQC(ctx context.Context, res
 		deadPixelAnalysis, err := ea.deadPixelAnalyzer.AnalyzeDeadPixels(ctx, filePath)
 		if err != nil {
 			// Log error but don't fail entire analysis - analysis may fail on some video types
-			fmt.Printf("Warning: dead pixel analysis failed: %v\n", err)
+			ea.logger.Warn().Err(err).Msg("dead pixel analysis failed")
 		} else {
 			result.EnhancedAnalysis.DeadPixelAnalysis = deadPixelAnalysis
 		}
@@ -246,7 +249,7 @@ func (ea *EnhancedAnalyzer) AnalyzeResultWithAdvancedQC(ctx context.Context, res
 		pseAnalysis, err := ea.pseAnalyzer.AnalyzePSERisk(ctx, filePath)
 		if err != nil {
 			// Log error but don't fail entire analysis - analysis may fail on some video types
-			fmt.Printf("Warning: PSE analysis failed: %v\n", err)
+			ea.logger.Warn().Err(err).Msg("PSE analysis failed")
 		} else {
 			result.EnhancedAnalysis.PSEAnalysis = pseAnalysis
 		}
@@ -257,7 +260,7 @@ func (ea *EnhancedAnalyzer) AnalyzeResultWithAdvancedQC(ctx context.Context, res
 		dispositionAnalysis, err := ea.streamDispositionAnalyzer.AnalyzeStreamDisposition(ctx, filePath, result.Streams)
 		if err != nil {
 			// Log error but don't fail entire analysis
-			fmt.Printf("Warning: stream disposition analysis failed: %v\n", err)
+			ea.logger.Warn().Err(err).Msg("stream disposition analysis failed")
 		} else {
 			result.EnhancedAnalysis.StreamDispositionAnalysis = dispositionAnalysis
 		}