ARC/backend/app/sandbox/main.py

import subprocess
import json
import time
import tempfile
import os
import logging
import traceback
from fastapi import FastAPI

app = FastAPI()
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

SEMGREP_TIMEOUT = 300.0
SEMGREP_RULES_DIR = "/app/rules/"

def generate_tree_string(path: str, max_depth: int = 3) -> str:
    """Génère une représentation visuelle de l'arborescence (exclut le bruit)."""
    lines = []
    exclude_dirs = {".git", "__pycache__", "node_modules", "venv", ".venv", "env"}
    
    def _walk(current_path, depth):
        if depth > max_depth:
            return
        try:
            entries = sorted(os.listdir(current_path))
        except Exception:
            return
        
        for entry in entries:
            if entry in exclude_dirs or entry.startswith('.'):
                continue
            full_path = os.path.join(current_path, entry)
            indent = "  " * depth
            if os.path.isdir(full_path):
                lines.append(f"{indent}📁 {entry}/")
                _walk(full_path, depth + 1)
            else:
                lines.append(f"{indent}📄 {entry}")
                
    _walk(path, 0)
    return "\n".join(lines)

@app.post("/scan")
async def scan_repository(repo_url: str):
    start_time = time.time()
    issues_list = []
    
    try:
        # 1. CRÉATION DU DOSSIER TEMPORAIRE ET CLONAGE DU PROJET À SCANNER
        with tempfile.TemporaryDirectory() as tmp_dir:
            repo_path = os.path.join(tmp_dir, "repo")
            logger.info(f"[Sandbox] Clonage dynamique du projet {repo_url} dans {repo_path}...")
            
            clone_cmd = f"git clone --depth 1 {repo_url} {repo_path}"
            clone_result = subprocess.run(clone_cmd, shell=True, capture_output=True, text=True)
            
            if clone_result.returncode != 0:
                logger.error(f"[Sandbox] Échec du git clone du projet : {clone_result.stderr}")
                return {
                    "results": {
                        "is_executable": False,
                        "runtime": {
                            "exit_code": clone_result.returncode,
                            "stdout": "",
                            "stderr": f"Impossible de cloner le projet : {clone_result.stderr}",
                            "duration_seconds": round(time.time() - start_time, 2),
                            "timeout_triggered": False
                        },
                        "issues": []
                    }
                }
            
            repo_tree = generate_tree_string(repo_path)
            
            try:
                cloned_files = os.listdir(repo_path)
                logger.info(f"[Sandbox] 📁 Fichiers récupérés après clone : {cloned_files}")
                if not cloned_files or cloned_files == ['.git']:
                    logger.warning("[Sandbox] ⚠️ Le dossier cloné est vide (ou ne contient que .git) !")
            except Exception as dir_err:
                logger.error(f"[Sandbox] Impossible de lister le contenu du dossier cloné : {dir_err}")

            # 2. EXÉCUTION DE SEMGREP AVEC LA RÈGLE UNIQUE
            try:
                if os.path.exists(SEMGREP_RULES_DIR):
                    logger.info(f"[Sandbox] Répertoire de règles détecté : {SEMGREP_RULES_DIR}")
                else:
                    logger.error(f"[Sandbox] ❌ Répertoire de règles introuvable à l'emplacement : {SEMGREP_RULES_DIR}")

                logger.info(f"[Sandbox] Lancement du scan Semgrep avec l'audit unique ({SEMGREP_RULES_DIR})...")
                secure_command = f"semgrep scan --config={SEMGREP_RULES_DIR} --json --quiet ."
                
                result = subprocess.run(
                    secure_command,
                    shell=True,
                    cwd=repo_path,
                    capture_output=True,
                    text=True,
                    timeout=SEMGREP_TIMEOUT
                )
                
                duration = round(time.time() - start_time, 2)
                exit_code = result.returncode
                stderr_output = result.stderr
                timeout_triggered = False

            except subprocess.TimeoutExpired:
                logger.error("[Sandbox] Timeout Semgrep dépassé.")
                return {
                    "results": {
                        "is_executable": False,
                        "runtime": {
                            "exit_code": -1,
                            "stdout": "",
                            "stderr": "L'analyse statique a dépassé le timeout global.",
                            "duration_seconds": round(time.time() - start_time, 2),
                            "timeout_triggered": True
                        },
                        "issues": []
                    }
                }
            
            # 3. PARSING DES RÉSULTATS
            try:
                if result.returncode == 0:
                    is_executable = True
                    stdout_msg = f"Scan statique réussi.\n\n[STRUCTURE DETECTEE] :\n{repo_tree}"
                    if result.stdout.strip():
                        scan_data = json.loads(result.stdout)
                        for item in scan_data.get("results", []):
                            issues_list.append({
                                "tool": "Semgrep",
                                "file": item.get("path", "").replace(f"{repo_path}/", ""),
                                "line": item.get("start", {}).get("line"),
                                "code": item.get("check_id"),
                                "severity": item.get("extra", {}).get("severity", "MEDIUM").upper(),
                                "message": item.get("extra", {}).get("message", "")
                            })
                else:
                    is_executable = False
                    stdout_msg = f"Échec du moteur de scan Semgrep (Exit code {result.returncode})."
                    logger.error(f"[Sandbox] Semgrep a échoué ! Code: {result.returncode}\nSTDOUT: {result.stdout}\nSTDERR: {result.stderr}")
                
            except Exception as e:
                is_executable = False
                stdout_msg = "Erreur lors du traitement des résultats du scan."
                stderr_output = f"Erreur parsing JSON Semgrep: {str(e)}\n{result.stderr}"

            return {
                "results": {
                    "is_executable": is_executable,
                    "runtime": {
                        "exit_code": exit_code,
                        "stdout": stdout_msg,
                        "stderr": stderr_output,
                        "duration_seconds": duration,
                        "timeout_triggered": timeout_triggered
                    },
                    "issues": issues_list
                }
            }

    except Exception as global_e:
        error_trace = traceback.format_exc()
        logger.error(f"[Sandbox] CRASH INTERNE CRITIQUE :\n{error_trace}")
        return {
            "results": {
                "is_executable": False,
                "runtime": {
                    "exit_code": -99,
                    "stdout": "Crash critique du conteneur de la Sandbox.",
                    "stderr": f"Exception Python : {str(global_e)}\n\nTraceback complet :\n{error_trace}",
                    "duration_seconds": round(time.time() - start_time, 2),
                    "timeout_triggered": False
                },
                "issues": []
            }
        }