github_workflow_app/app.py

from fastapi import FastAPI, HTTPException, Query
from fastapi.staticfiles import StaticFiles
from fastapi.responses import FileResponse, JSONResponse
from pydantic import BaseModel
from typing import Optional, Dict, Any
import os
import subprocess
import dotenv
import json

# Load environment variables
dotenv.load_dotenv(os.path.join(os.path.dirname(__file__), '..', '.env'))

import sys
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
from src.config import get_config_value

app = FastAPI(title="GitHub Workflow Runner")

# Mount static files
app.mount("/static", StaticFiles(directory=os.path.dirname(__file__)), name="static")


def validate_safe_arg(value: str, param_name: str, allow_hyphen: bool = False):
    """
    Validate that a string argument is safe to pass to a subprocess.
    Rejects values starting with '-' to prevent flag injection.
    """
    if not value:
        return
        
    # Prevent argument injection
    if not allow_hyphen and value.startswith('-'):
        raise HTTPException(status_code=400, detail=f"Invalid {param_name}: Cannot start with '-'")
        
    # Basic sanity check for dangerous characters often used in shell injection
    # Even though we use shell=False, it's good practice to be strict
    dangerous_chars = [';', '&', '|', '`', '$', '(', ')', '<', '>', '\\']
    if any(char in value for char in dangerous_chars):
        raise HTTPException(status_code=400, detail=f"Invalid {param_name}: Contains illegal characters")



def get_github_token_from_git_credentials():
    """Parse ~/.git-credentials to extract GitHub token"""
    git_credentials_path = os.path.expanduser('~/.git-credentials')
    if not os.path.exists(git_credentials_path):
        return None
    
    try:
        with open(git_credentials_path, 'r') as f:
            for line in f:
                line = line.strip()
                # Format: https://username:token@github.com
                if 'github.com' in line and ':' in line and '@' in line:
                    try:
                        creds_part = line.split('@')[0]
                        if '://' in creds_part:
                            creds_part = creds_part.split('://')[1]
                        parts = creds_part.split(':')
                        if len(parts) >= 2:
                            token = parts[1]
                            if token.startswith('ghp_') or token.startswith('gho_') or token.startswith('github_pat_'):
                                return token
                    except Exception:
                        continue
    except Exception as e:
        print(f"Error reading git-credentials: {e}")
    
    return None


# Configuration - Priority: env var > git-credentials
GITHUB_TOKEN = get_config_value("GITHUB_TOKEN") or get_config_value("GITHUB_PAT") or get_github_token_from_git_credentials()
REPO_OWNER = "ElvoroLtd"
REPO_NAME = "Elvoro"
WORKFLOW_FILE = "process_csv.yml"


@app.get("/health")
async def health_check():
    """Health check endpoint for container orchestration"""
    return {"status": "healthy"}


@app.get("/")
async def index():
    return FileResponse(os.path.join(os.path.dirname(__file__), 'index.html'))


@app.get("/api/auth/status")
async def auth_status():
    """Check if we have a valid token by running `gh auth status` or similar"""
    token = GITHUB_TOKEN
    
    if not token:
        return {"authenticated": False, "message": "Token not found in .env"}
    
    try:
        env = os.environ.copy()
        env['GITHUB_TOKEN'] = token
        cmd = ['gh', 'api', 'user']
        result = subprocess.run(cmd, capture_output=True, text=True, env=env)
        
        if result.returncode == 0:
            user_data = json.loads(result.stdout)
            return {"authenticated": True, "user": user_data.get('login')}
        else:
            return {"authenticated": False, "message": "Invalid token"}
    except Exception as e:
        return {"authenticated": False, "message": str(e)}


@app.get("/api/env-vars")
async def get_env_vars(workflow: str = Query(default="process_csv.yml")):
    """
    Parse env file based on selected workflow.
    Strategy:
    1. Load actual values from .env (Source of Truth)
    2. Load keys from template file (publisher.env / video_generate.env)
    3. Return keys from template populated with values from .env
    """
    # 1. Load Source of Truth (.env)
    root_dir = os.path.join(os.path.dirname(__file__), '..')
    dotenv_path = os.path.join(root_dir, '.env')
    actual_values = dotenv.dotenv_values(dotenv_path)
    
    # 2. Determine Template File
    if workflow == 'publisher.yml':
        template_filename = 'publisher.env'
    else:
        template_filename = 'video_generate.env'
        
    template_path = os.path.join(root_dir, template_filename)
    
    vars_dict = {}
    
    # 3. Specific keys from template, populate with actual values
    if os.path.exists(template_path):
        try:
            with open(template_path, 'r') as f:
                for line in f:
                    line = line.strip()
                    if not line or line.startswith('#'):
                        continue
                    
                    if '=' in line:
                        key = line.split('=')[0].strip()
                    else:
                        key = line.strip()
                        
                    vars_dict[key] = actual_values.get(key, '')
                    
        except Exception as e:
            print(f"Error reading template {template_filename}: {e}")
    else:
        vars_dict = dict(actual_values)

    return {"vars": vars_dict}


class TriggerRequest(BaseModel):
    token: Optional[str] = None
    inputs: Optional[Dict[str, Any]] = {}
    ref: Optional[str] = "feature/video-revamp"
    workflow: Optional[str] = "process_csv.yml"


@app.post("/api/trigger")
async def trigger_workflow(data: TriggerRequest):
    token = data.token or GITHUB_TOKEN
    
    if not token:
        raise HTTPException(status_code=401, detail="No GitHub token provided")
        
    inputs = data.inputs or {}
    ref = data.ref
    workflow_file = data.workflow
    
    # Validate inputs to prevent argument injection
    validate_safe_arg(ref, "ref")
    validate_safe_arg(workflow_file, "workflow")
    
    cmd = [
        'gh', 'workflow', 'run', workflow_file,
        '--repo', f"{REPO_OWNER}/{REPO_NAME}",
        '--ref', ref
    ]
    
    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:
        print(f"Executing: {' '.join(cmd)}")
        
        env = os.environ.copy()
        env['GITHUB_TOKEN'] = token
        
        result = subprocess.run(cmd, capture_output=True, text=True, env=env)
        
        if result.returncode == 0:
            return {"success": True, "message": "Workflow triggered successfully"}
        else:
            print(f"Error triggering workflow: {result.stderr}")
            raise HTTPException(status_code=400, detail=f"Failed: {result.stderr}")
            
    except HTTPException:
        raise
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


@app.get("/api/runs")
async def list_runs(
    token: Optional[str] = Query(default=None),
    workflow: str = Query(default="process_csv.yml")
):
    """List recent workflow runs using gh"""
    token = token or GITHUB_TOKEN
    
    if not token:
        return {"runs": []}
        
    # Validate inputs
    validate_safe_arg(workflow, "workflow")
        
    cmd = [
        'gh', 'run', 'list',
        '--workflow', workflow,
        '--repo', f"{REPO_OWNER}/{REPO_NAME}",
        '--limit', '5',
        '--json', 'number,status,conclusion,createdAt,url,name'
    ]
    
    try:
        env = os.environ.copy()
        env['GITHUB_TOKEN'] = token
        
        result = subprocess.run(cmd, capture_output=True, text=True, env=env)
        
        if result.returncode == 0:
            runs = json.loads(result.stdout)
            normalized_runs = []
            for run in runs:
                normalized_runs.append({
                    'run_number': run.get('number'),
                    'name': run.get('name'),
                    'status': run.get('status'),
                    'conclusion': run.get('conclusion'),
                    'created_at': run.get('createdAt'),
                    'html_url': run.get('url')
                })
                
            return {"workflow_runs": normalized_runs}
        else:
            return {"runs": [], "error": result.stderr}
    except Exception as e:
        return {"runs": [], "error": str(e)}


if __name__ == '__main__':
    import uvicorn
    print("GitHub Workflow Runner (FastAPI)")
    print("Open your browser to: http://localhost:5002")
    uvicorn.run(app, host="127.0.0.1", port=5002)