🔒 Fix HIGH SEVERITY Security Vulnerabilities - Flask Debug Mode & Information Exposure

deployment/cloud-run/test_minimal_swagger.py

@@ -44,4 +44,4 @@ def get(self):
     print("- http://localhost:5003/docs (should work)")
     print("- http://localhost:5003/api/health (should work)")
 
-    app.run(host='0.0.0.0', port=5003, debug=True) 
+    app.run(host='0.0.0.0', port=5003, debug=False) 

deployment/cloud-run/test_routing_minimal.py

@@ -53,4 +53,4 @@ def root():
         print(f"API: {rule.rule} -> {rule.endpoint}")
 
     print("\n=== Starting test server ===")
-    app.run(host='0.0.0.0', port=5000, debug=True) 
+    app.run(host='0.0.0.0', port=5000, debug=False) 

deployment/cloud-run/test_swagger_debug.py

@@ -44,4 +44,4 @@ def api_root():  # Different function name to avoid conflict
     print("- http://localhost:5001/docs (should work)")
     print("- http://localhost:5001/api/health (should work)")
 
-    app.run(host='0.0.0.0', port=5001, debug=True) 
+    app.run(host='0.0.0.0', port=5001, debug=False) 

deployment/cloud-run/test_swagger_no_model.py

@@ -52,4 +52,4 @@ def get(self):
     print("- http://localhost:8083/docs (should work)")
     print("- http://localhost:8083/api/health (should work)")
 
-    app.run(host='0.0.0.0', port=8083, debug=True) 
+    app.run(host='0.0.0.0', port=8083, debug=False) 

scripts/deployment/security_deployment_fix.py

@@ -17,6 +17,7 @@
 import shlex
 import time
 import requests
+import re
 from pathlib import Path
 from typing import Dict, List, Optional
 
@@ -55,10 +56,29 @@ def __init__(self):
         self.secure_api = self.deployment_dir / "secure_api_server.py"
 
     @staticmethod
-    def log(message: str, level: str = "INFO"):
+    def log(self, message: str, level: str = "INFO"):
         """Log messages with timestamp"""
         timestamp = time.strftime("%Y-%m-%d %H:%M:%S")
-        print(f"[{timestamp}] [{level}] {message}")
+        # Sanitize sensitive information before logging
+        sanitized_message = self._sanitize_log_message(message)
+        print(f"[{timestamp}] [{level}] {sanitized_message}")
+
+    def _sanitize_log_message(self, message: str) -> str:
+        """Sanitize log messages to remove sensitive information"""
+        # Remove API keys, tokens, and other sensitive data
+        sensitive_patterns = [
+            r'api[_-]?key[=:]\s*[^\s&]+',  # API keys
+            r'token[=:]\s*[^\s&]+',         # Tokens
+            r'password[=:]\s*[^\s&]+',      # Passwords
+            r'secret[=:]\s*[^\s&]+',        # Secrets
+            r'key[=:]\s*[^\s&]+',           # Generic keys
+        ]
+
+        sanitized = message
+        for pattern in sensitive_patterns:
+            sanitized = re.sub(pattern, r'\1=***REDACTED***', sanitized, flags=re.IGNORECASE)
+
+        return sanitized
 
     def run_command(self, command: List[str], check: bool = True) -> subprocess.CompletedProcess:
         """Run shell command with error handling"""

scripts/testing/debug_model_loading.py

@@ -19,7 +19,7 @@
     print("🔍 Debugging Model Loading Issues")
     print("=" * 50)
     print(f"Testing URL: {config.base_url}")
-    print(f"API Key: {config.api_key[:20]}...")
+    print(f"API Key: {config.api_key[:8]}...***REDACTED***")
 
     # Test model status with API key
     print("\n1. Testing model status with API key...")

src/unified_ai_api.py

@@ -51,6 +51,20 @@
 # Note: Avoid monkey-patching httpx internals. Tests should handle httpx.StreamConsumed
 # or public APIs directly when dealing with closed file objects.
 
+def sanitize_exception(exc: Exception) -> str:
+    """Sanitize exception messages to prevent information exposure."""
+    # Only expose safe, generic error messages
+    if isinstance(exc, ValidationError):
+        return "Validation error occurred"
+    elif isinstance(exc, ValueError):
+        return "Invalid value provided"
+    elif isinstance(exc, KeyError):
+        return "Missing required field"
+    elif isinstance(exc, TypeError):
+        return "Type mismatch error"
+    else:
+        return "Internal server error"
+
 # Global AI models (loaded on startup)
 emotion_detector = None
 text_summarizer = None
@@ -1237,7 +1251,7 @@ async def chat_websocket(websocket: WebSocket, token: str = Query(None)) -> None
                         exc,
                         exc_info=True,
                     )
-                    response["summary_error"] = str(exc)
+                    response["summary_error"] = sanitize_exception(exc)
 
             await websocket.send_json(response)
     except WebSocketDisconnect:
@@ -1717,7 +1731,7 @@ async def batch_transcribe_voice(
                     "file_index": i,
                     "filename": audio_file.filename,
                     "success": False,
-                    "error": str(exc)
+                    "error": sanitize_exception(exc)
                 })
 
         processing_time = (time.time() - start_time) * 1000
@@ -1921,7 +1935,7 @@ async def websocket_realtime_processing(websocket: WebSocket, token: str = Query
                 except Exception as exc:
                     await websocket.send_json({
                         "type": "error",
-                        "message": str(exc)
+                        "message": sanitize_exception(exc)
                     })
             else:
                 await websocket.send_json({
@@ -2031,7 +2045,7 @@ async def detailed_health_check(
         except Exception as exc:
             health_status = "degraded"
             issues.append(f"Emotion detection model error: {exc}")
-            model_checks["emotion_detection"] = {"status": "error", "error": str(exc)}
+            model_checks["emotion_detection"] = {"status": "error", "error": sanitize_exception(exc)}
 
     if text_summarizer is None:
         health_status = "degraded"
@@ -2045,7 +2059,7 @@ async def detailed_health_check(
         except Exception as exc:
             health_status = "degraded"
             issues.append(f"Text summarization model error: {exc}")
-            model_checks["text_summarization"] = {"status": "error", "error": str(exc)}
+            model_checks["text_summarization"] = {"status": "error", "error": sanitize_exception(exc)}
 
     if voice_transcriber is None:
         health_status = "degraded"
@@ -2074,9 +2088,9 @@ async def detailed_health_check(
             "status": "healthy" if cpu_percent < 90 and memory.percent < 90 else "warning"
         }
     except Exception as exc:
-        system_checks = {"status": "error", "error": str(exc)}
+        system_checks = {"status": "error", "error": sanitize_exception(exc)}
         health_status = "degraded"
-        issues.append(f"System check failed: {exc}")
+        issues.append("System check failed: Internal error")
 
     return {
         "status": health_status,