fix: Remediate CodeQL security vulnerabilities

github_workflow_app/app.py

@@ -189,6 +189,9 @@ async def trigger_workflow(data: TriggerRequest):
     
     for key, value in inputs.items():
         if value:
+            # Validate input key and value to prevent command injection
+            validate_safe_arg(key, "input key")
+            validate_safe_arg(str(value), "input value")
             cmd.extend(['-f', f"{key}={value}"])
             
     try:

social_media_publishers/app.py

@@ -205,8 +205,8 @@ async def get_account_videos(
         result = publisher.get_uploaded_videos(account_id, limit, page_token, start_date, end_date)
         return result
     except Exception as e:
-        logger.error(f"Get videos error: {e}")
-        return JSONResponse(content={"error": str(e)}, status_code=500)
+        logger.error(f"Get videos error: {e}", exc_info=True)
+        return JSONResponse(content={"error": "An error occurred while fetching videos"}, status_code=500)
 
 @app.get("/api/{platform}/accounts/{account_id}/stats")
 async def get_account_stats(
@@ -224,8 +224,8 @@ async def get_account_stats(
         result = publisher.get_account_stats(account_id, start_date, end_date)
         return result
     except Exception as e:
-        logger.error(f"Get stats error: {e}")
-        return JSONResponse(content={"error": str(e)}, status_code=500)
+        logger.error(f"Get stats error: {e}", exc_info=True)
+        return JSONResponse(content={"error": "An error occurred while fetching stats"}, status_code=500)
 
 @app.get("/api/{platform}/accounts/{account_id}/creator_info")
 async def get_creator_info(
@@ -245,8 +245,8 @@ async def get_creator_info(
         
         return {} # Return empty for others
     except Exception as e:
-        logger.error(f"Get creator info error: {e}")
-        return JSONResponse(content={"error": str(e)}, status_code=500)
+        logger.error(f"Get creator info error: {e}", exc_info=True)
+        return JSONResponse(content={"error": "An error occurred while fetching creator info"}, status_code=500)
 
 @app.post("/api/publish")
 async def publish_video(
@@ -275,7 +275,9 @@ async def publish_video(
             # Save to temp
             temp_dir = os.path.join(PROJECT_ROOT, 'temp_uploads')
             os.makedirs(temp_dir, exist_ok=True)
-            filename = f"{platform}_{os.urandom(4).hex()}_{file.filename}"
+            # Sanitize filename to prevent path traversal
+            safe_filename = os.path.basename(file.filename or "upload")
+            filename = f"{platform}_{os.urandom(4).hex()}_{safe_filename}"
             file_path = os.path.join(temp_dir, filename)
             
             with open(file_path, "wb") as buffer:
@@ -334,7 +336,7 @@ async def publish_video(
 
     except Exception as e:
         logger.error(f"Publish Error: {e}", exc_info=True)
-        return JSONResponse(content={'error': f"Request processing failed: {str(e)}"}, status_code=500)
+        return JSONResponse(content={'error': "An error occurred during publishing. Please try again."}, status_code=500)
     finally:
         # Cleanup
         if file_path and os.path.exists(file_path):
@@ -453,9 +455,8 @@ async def verify_app_review(
 
     except Exception as e:
         log(f"❌ Verification script error: {e}")
-        import traceback
-        traceback.print_exc()
-        return JSONResponse(content={"success": False, "logs": logs, "error": str(e)}, status_code=500)
+        logger.error(f"Verify App Review Error: {e}", exc_info=True)
+        return JSONResponse(content={"success": False, "logs": logs, "error": "Verification encountered an error"}, status_code=500)
 
 @app.delete("/api/accounts/{filename}")
 async def delete_account(filename: str, platform: str = 'youtube'):

social_media_publishers/base.py

@@ -2,6 +2,7 @@ import os
 import sys
 import json
 import glob
+import tempfile
 from abc import ABC, abstractmethod
 from typing import Dict, List, Optional, Any
 
@@ -273,8 +274,9 @@ class SocialPublisher(ABC):
                  print(f"⚠️ Unsafe content path rejected: {content_path}")
                  return None
                  
-             if os.path.exists(content_path):
-                return content_path
+             normalized_path = os.path.abspath(content_path)
+             if os.path.exists(normalized_path):
+                return normalized_path
 
             
         # Case 4: GCS Download (existing logic)

social_media_publishers/facebook/publisher.py

@@ -137,7 +137,13 @@ class FacebookPublisher(SocialPublisher):
         """
         Upload video to Facebook Page using resumable upload API.
         """
-        file_size = os.path.getsize(video_path)
+        # Sanitize path to prevent traversal attacks
+        safe_path = os.path.abspath(video_path)
+        allowed_prefixes = [os.path.abspath(os.getcwd()), '/tmp', os.path.expanduser('~')]
+        if not any(safe_path.startswith(prefix) for prefix in allowed_prefixes):
+            return {"error": "Invalid video path: access denied"}
+        
+        file_size = os.path.getsize(safe_path)
         
         # Step 1: Start upload session
         print(f"🎬 Starting Facebook video upload...")
@@ -166,7 +172,7 @@ class FacebookPublisher(SocialPublisher):
         chunk_size = 10 * 1024 * 1024  # 10MB chunks
         start_offset = 0
         
-        with open(video_path, 'rb') as video_file:
+        with open(safe_path, 'rb') as video_file:
             while start_offset < file_size:
                 chunk = video_file.read(chunk_size)
                 end_offset = start_offset + len(chunk)

social_media_publishers/instagram/publisher.py

@@ -140,6 +140,14 @@ class InstagramPublisher(SocialPublisher):
         is_url = content_source and (content_source.startswith('http://') or content_source.startswith('https://'))
         is_local = content_source and os.path.exists(content_source)
         
+        # Sanitize local path to prevent traversal attacks
+        if is_local:
+            safe_path = os.path.abspath(content_source)
+            allowed_prefixes = [os.path.abspath(os.getcwd()), '/tmp', os.path.expanduser('~')]
+            if not any(safe_path.startswith(prefix) for prefix in allowed_prefixes):
+                return {"error": "Invalid path: access denied"}
+            content_source = safe_path
+        
         video_url = None
         
         if is_url:

social_media_publishers/threads/publisher.py

@@ -122,6 +122,14 @@ class ThreadsPublisher(SocialPublisher):
         is_url = content_source and (content_source.startswith('http://') or content_source.startswith('https://'))
         is_local = content_source and os.path.exists(content_source)
         
+        # Sanitize local path to prevent traversal attacks
+        if is_local:
+            safe_path = os.path.abspath(content_source)
+            allowed_prefixes = [os.path.abspath(os.getcwd()), '/tmp', os.path.expanduser('~')]
+            if not any(safe_path.startswith(prefix) for prefix in allowed_prefixes):
+                return {"error": "Invalid path: access denied"}
+            content_source = safe_path
+        
         if is_local or is_url:
             # Check extension or MIME type logic?
             # For simplicity, assume video if typical extensions, else check.

social_media_publishers/tiktok/publisher.py

@@ -120,7 +120,13 @@ class TikTokPublisher(SocialPublisher):
         # Prepare content (handles GCS download if needed)
         local_path = self.prepare_content(content_path)
         
-        if not os.path.exists(local_path):
+        # Sanitize path to prevent traversal attacks
+        safe_path = os.path.abspath(local_path) if local_path else None
+        allowed_prefixes = [os.path.abspath(os.getcwd()), '/tmp', os.path.expanduser('~')]
+        if not safe_path or not any(safe_path.startswith(prefix) for prefix in allowed_prefixes):
+            return {"error": "Invalid video path: access denied"}
+        
+        if not os.path.exists(safe_path):
             return {"error": f"Video file not found: {local_path}"}
         
         title = metadata.get('title', metadata.get('description', metadata.get('caption', '')))
@@ -146,7 +152,13 @@ class TikTokPublisher(SocialPublisher):
         Upload video to TikTok using the Content Posting API.
         3-step process: Initialize → Upload → Check Status
         """
-        video_size = os.path.getsize(video_path)
+        # Sanitize path to prevent traversal attacks
+        safe_path = os.path.abspath(video_path)
+        allowed_prefixes = [os.path.abspath(os.getcwd()), '/tmp', os.path.expanduser('~')]
+        if not any(safe_path.startswith(prefix) for prefix in allowed_prefixes):
+            return {"error": "Invalid video path: access denied"}
+        
+        video_size = os.path.getsize(safe_path)
         print(f"🎬 Uploading TikTok video: {title[:50]}... ({video_size / 1024 / 1024:.1f} MB)")
         
         # Step 1: Initialize upload
@@ -175,7 +187,7 @@ class TikTokPublisher(SocialPublisher):
         print(f"✅ Upload initialized: {publish_id}")
         
         # Step 2: Upload video file
-        upload_success = self._upload_file(upload_url, video_path)
+        upload_success = self._upload_file(upload_url, safe_path)
         if not upload_success:
             return {"error": "Video upload failed"}
         

social_media_publishers/youtube/publisher.py

@@ -190,11 +190,18 @@ class YoutubePublisher(SocialPublisher):
             print(f"Uploading: {title}")
             print(f"File: {video_path}")
             
-            if not os.path.exists(video_path):
+            # Sanitize path to prevent traversal attacks
+            # Normalize and validate against allowed directories
+            safe_path = os.path.abspath(video_path)
+            allowed_prefixes = [os.path.abspath(os.getcwd()), '/tmp', os.path.expanduser('~')]
+            if not any(safe_path.startswith(prefix) for prefix in allowed_prefixes):
+                return {'error': 'Invalid video path: access denied'}
+            
+            if not os.path.exists(safe_path):
                 return {'error': f'Video file not found: {video_path}'}
             
             media = MediaFileUpload(
-                video_path, 
+                safe_path, 
                 chunksize=10 * 1024 * 1024,  # 10MB chunks
                 resumable=True,
                 mimetype='video/*'