JulieChv/ARC
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Correction QA: Validation arborescence + qualité/sécurité, ajout logique de fin complète, suppression repo non conforme

Browse Source
This commit is contained in:
Chevallier
2026-07-15 14:00:59 +02:00
parent 2ded228202
commit 9d464fb261
13 changed files with 156656 additions and 230 deletions
Download Patch File Download Diff File Expand all files Collapse all files

129
backend/app/agents/qa_agent.py
View File

@@ -3,8 +3,8 @@ from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage
import httpx
from app.core.config import settings
from app.schemas.qa_report import QAEvaluation
from app.llm.prompts import QA_AGENT_PROMPT
from app.schemas.qa_report import QAEvaluation, StructureEvaluation
from app.llm.prompts import QA_AGENT_PROMPT, QA_STRUCTURE_AGENT_PROMPT
from app.sandbox.qa_client import run_project_qa
logger = logging.getLogger(__name__)
@@ -15,8 +15,9 @@ async_client = httpx.AsyncClient(verify=False)
async def run_qa_agent(project_title: str, dev_output: dict) -> QAEvaluation:
"""
Agent QA :
Analyse le code généré à l'aide de l'API externe de la Sandbox (Semgrep uniquement)
et produit le rapport d'évaluation structuré final.
Analyse le code généré à l'aide de l'API externe de la Sandbox (Semgrep uniquement),
valide la structure minimale du projet à l'aide d'un second appel LLM,
puis produit le rapport d'évaluation fusionné.
"""
logger.info(f"[QA Agent] Début du processus d'audit statique pour le projet : {project_title}")
@@ -39,12 +40,12 @@ async def run_qa_agent(project_title: str, dev_output: dict) -> QAEvaluation:
http_async_client=async_client,
)
# 2. Exécution de l'analyse statique via le service Sandbox indépendant
# 1. Exécution de l'analyse statique via le service Sandbox indépendant
try:
repo_name = repo_url.rstrip("/").split("/")[-1]
logger.info(f"[QA Agent] Envoi du dépôt '{repo_name}' au conteneur d'analyse statique...")
raw_results = await run_project_qa(repo_name=repo_name)
raw_results = await run_project_qa(repo_url=repo_url)
logger.info(f"[QA Agent] 📊 Retour Sandbox Statis - Status : {'Succès' if raw_results.is_executable else 'Échec'}")
logger.info(f"[QA Agent] 📊 Retour Sandbox - Nombre d'alertes de sécurité/qualité (Semgrep): {len(raw_results.issues)}")
@@ -57,46 +58,84 @@ async def run_qa_agent(project_title: str, dev_output: dict) -> QAEvaluation:
technical_feedback=[f"L'API de la sandbox a échoué : {type(e).__name__}: {str(e)}"]
)
# 3. Génération de l'évaluation finale via LLM
structured_llm = llm.with_structured_output(QAEvaluation, strict=True)
messages = [
SystemMessage(content=QA_AGENT_PROMPT),
]
# Le prompt est nettoyé des mentions relatives à l'exécution de tests dynamiques
user_content = f"PROJET À AUDITER : {project_title}\n"
user_content += f"LIEN DU DÉPÔT : {repo_url}\n"
user_content += f"RÉSULTATS DE L'ANALYSE STATIQUE DU CODE (Semgrep) :\n"
user_content += f"{raw_results.model_dump_json(indent=2)}\n\n"
user_content += "Analyse ces failles, vulnérabilités et défauts de qualité de code, puis génère ton évaluation de sécurité au format structuré demandé."
messages.append(HumanMessage(content=user_content))
# 2. Premier appel LLM : Audit de sécurité (Semgrep)
qa_evaluation = None
try:
qa_evaluation = await structured_llm.ainvoke(messages)
logger.info(f"[QA Agent] 📝 Justification du LLM : {qa_evaluation.global_summary}")
logger.info("[QA Agent] Lancement de l'audit de sécurité et de qualité (Semgrep)...")
structured_security_llm = llm.with_structured_output(QAEvaluation, strict=True)
security_messages = [
SystemMessage(content=QA_AGENT_PROMPT),
HumanMessage(content=(
f"PROJET À AUDITER : {project_title}\n"
f"LIEN DU DÉPÔT : {repo_url}\n"
f"RÉSULTATS DE L'ANALYSE STATIQUE DU CODE (Semgrep) :\n"
f"{raw_results.model_dump_json(indent=2)}\n\n"
f"Analyse ces failles, vulnérabilités et défauts de qualité de code, puis génère ton évaluation de sécurité au format structuré demandé."
))
]
qa_evaluation = await structured_security_llm.ainvoke(security_messages)
if not qa_evaluation:
logger.error("[QA Agent] Le LLM a renvoyé une réponse vide (None).")
return QAEvaluation(
is_complete_and_safe=False,
global_summary="Erreur de formatage de l'évaluation par l'IA.",
technical_feedback=["L'évaluation automatique n'a pas pu être structurée correctement."]
)
if qa_evaluation.is_complete_and_safe:
logger.info(f"[QA Agent] ✅ Projet '{project_title}' VALIDÉ (Analyse statique propre).")
else:
reason = "Erreur d'exécution ou non-conformité" if "exit_code" in str(qa_evaluation) else "Failles de sécurité"
logger.warning(f"[QA Agent] ❌ Projet '{project_title}' REJETÉ ({reason}).")
return qa_evaluation
except Exception as e:
logger.error(f"[QA Agent] ❌ Échec lors de l'audit de sécurité LLM : {type(e).__name__}: {str(e)}")
qa_evaluation = QAEvaluation(
is_complete_and_safe=False,
global_summary="Erreur lors de la génération du rapport de sécurité par le LLM.",
technical_feedback=[f"Le module de sécurité a échoué : {type(e).__name__}"]
)
# 3. Second appel LLM : Validation de la structure minimale (Arborescence)
structure_evaluation = None
try:
logger.info("[QA Agent] Lancement de l'audit de conformité de l'arborescence...")
structured_structure_llm = llm.with_structured_output(StructureEvaluation, strict=True)
repo_structure = raw_results.runtime.stdout or "Arborescence non détectée ou vide."
structure_messages = [
SystemMessage(content=QA_STRUCTURE_AGENT_PROMPT),
HumanMessage(content=(
f"PROJET À ANALYSER : {project_title}\n"
f"ARBORESCENCE DU RÉPERTOIRE CLONÉ :\n"
f"{repo_structure}\n\n"
f"Vérifie si la structure respecte les exigences minimales et fournis ton évaluation structurée."
))
]
structure_evaluation = await structured_structure_llm.ainvoke(structure_messages)
logger.info(f"[QA Agent] 📁 Structure détectée : {structure_evaluation.detected_language} | Valide : {structure_evaluation.is_valid}")
except Exception as e:
logger.error(f"[QA Agent] ❌ Échec critique lors de l'analyse LLM : {type(e).__name__}: {str(e)}")
return QAEvaluation(
is_complete_and_safe=False,
global_summary="Erreur interne de l'Agent QA lors de la génération du rapport LLM.",
technical_feedback=[f"Le LLM a crashé avec l'erreur : {type(e).__name__}."]
)
logger.error(f"[QA Agent] ❌ Échec lors de la validation de structure LLM : {type(e).__name__}: {str(e)}")
structure_evaluation = StructureEvaluation(
is_valid=False,
detected_language="Inconnu",
missing_elements=["Erreur d'analyse structurelle"],
feedback="Le module d'analyse d'arborescence a rencontré une erreur de traitement."
)
# 4. Fusion des résultats des deux audits
final_is_complete_and_safe = qa_evaluation.is_complete_and_safe and structure_evaluation.is_valid
final_feedback = list(qa_evaluation.technical_feedback)
if not structure_evaluation.is_valid:
for missing in structure_evaluation.missing_elements:
final_feedback.append(f"[Structure] Élément obligatoire manquant : {missing}")
final_summary = (
f"{qa_evaluation.global_summary}\n\n"
f"--- 📁 Validation de la Structure ({structure_evaluation.detected_language}) ---\n"
f"Statut : {'Conforme' if structure_evaluation.is_valid else 'Non conforme'}\n"
f"Feedback : {structure_evaluation.feedback}"
)
if final_is_complete_and_safe:
logger.info(f"[QA Agent] ✅ Projet '{project_title}' VALIDÉ (Sécurité & Structure OK).")
else:
logger.warning(f"[QA Agent] ❌ Projet '{project_title}' REJETÉ (Non-conformité détectée).")
return QAEvaluation(
is_complete_and_safe=final_is_complete_and_safe,
global_summary=final_summary,
technical_feedback=final_feedback
)

13
backend/app/llm/prompts.py
View File

@@ -436,4 +436,17 @@ Tu disposes des données brutes suivantes :
### Ton Style de Feedback :
Sois ultra-précis et technique. Ne dis pas "Il y a une erreur dans le code", dis plutôt : "La fonction X à la ligne Y lève une exception de type ValueError car la variable Z est passée à None". Traduis chaque erreur technique brute en une instruction claire et actionnable pour l'Agent Dev.
"""
QA_STRUCTURE_AGENT_PROMPT = """
Tu es un agent expert en architecture et standardisation logicielle.
Ton rôle est d'analyser l'arborescence de fichiers d'un dépôt Git et de valider si la structure respecte rigoureusement les standards minimaux de livraison.
Critères stricts d'un projet conforme :
1. DOCUMENTATION : Présence obligatoire d'un fichier de documentation (ex: README.md, README.txt).
2. DÉPENDANCES : Présence obligatoire d'un fichier de gestion des dépendances adapté au langage (ex: requirements.txt ou pyproject.toml pour Python, package.json pour Node, go.mod pour Go, cargo.toml pour Rust, etc.). Ce fichier doit exister à la racine, même s'il est vide.
3. POINT D'ENTRÉE : Un script principal ou fichier de démarrage cohérent (ex: main.py/app.py pour Python, index.js/server.js pour Node, main.go pour Go, etc.).
4. SUITE DE TESTS : Présence obligatoire d'un dossier dédié aux tests (généralement nommé 'tests' ou 'test') contenant au moins un script de test (ex: test_script.py, app.test.js, etc.).
IMPORTANT : Sois intransigeant sur ces quatre piliers. Si le fichier de dépendances est absent (même si le script n'a pas de dépendances externes) ou si le dossier 'tests' est manquant ou vide, le projet doit être déclaré NON conforme (is_valid = False).
"""

6
backend/app/sandbox/Dockerfile.qa
View File

@@ -3,7 +3,7 @@ FROM semgrep/semgrep:latest
USER root
RUN sed -i 's/https/http/g' /etc/apk/repositories && \
apk update && apk add --no-cache python3 py3-pip
apk update && apk add --no-cache python3 py3-pip git
RUN pip install --no-cache-dir \
--trusted-host pypi.org \
@@ -13,6 +13,10 @@ RUN pip install --no-cache-dir \
WORKDIR /app
# COPY r2c-security-audit.yaml /app/r2c-security-audit.yaml
COPY rules/ /app/rules/
RUN sed -i 's/adjust_for_docker()/pass/g' /usr/lib/python3.12/site-packages/semgrep/commands/scan.py
COPY main.py .
EXPOSE 8004

228
backend/app/sandbox/main.py
View File

@@ -1,85 +1,179 @@
import subprocess
import json
import time
from fastapi import FastAPI
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
@app.post("/scan/{repo_name}")
async def scan_repository(repo_name: str):
repo_path = f"/src/{repo_name}"
issues_list = []
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:
secure_command = f"ulimit -f 10240 && semgrep scan --config=p/r2c-security-audit --json --quiet {repo_path}"
result = subprocess.run(
secure_command,
shell=True,
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
is_executable = True
# 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}")
except subprocess.TimeoutExpired as te:
logger.error(f"[Sandbox] Semgrep a dépassé le timeout sur {repo_name}")
duration = round(time.time() - start_time, 2)
exit_code = -1
stderr_output = f"L'analyse statique a été coupée : Timeout de {SEMGREP_TIMEOUT}s dépassé."
timeout_triggered = True
is_executable = False
# 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": is_executable,
"is_executable": False,
"runtime": {
"exit_code": exit_code,
"stdout": "",
"stderr": stderr_output,
"duration_seconds": duration,
"timeout_triggered": timeout_triggered
"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": []
}
}
try:
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", "")
})
is_executable = True
stderr_output = result.stderr
except Exception as e:
is_executable = False
stderr_output = f"Erreur parsing JSON Semgrep: {str(e)}\n{result.stderr}"
return {
"results": {
"is_executable": is_executable,
"runtime": {
"exit_code": exit_code,
"stdout": "Scan statique universel effectué avec succès.",
"stderr": stderr_output,
"duration_seconds": duration,
"timeout_triggered": timeout_triggered
},
"issues": issues_list
}
}
}

35
backend/app/sandbox/qa_client.py
View File

@@ -4,23 +4,42 @@ from app.core.config import settings
from app.schemas.qa_report import QARawResults, RuntimeOutput
logger = logging.getLogger(__name__)
async_client = httpx.AsyncClient(verify=False)
async def run_project_qa(repo_name: str) -> QARawResults:
async def run_project_qa(repo_url: str) -> QARawResults:
"""
Sollicite l'API du conteneur persistant arc-sandbox pour effectuer
une analyse de qualité et sécurité statique (Semgrep).
Sollicite l'API du conteneur arc-sandbox pour effectuer
une analyse de qualité et sécurité statique (Semgrep) en clonant le dépôt.
"""
sandbox_url = f"http://arc-sandbox:8004/scan/{repo_name}"
logger.info(f"[QA Client] Envoi de la requête de scan à la sandbox : {sandbox_url}")
sandbox_url = "http://arc-sandbox:8004/scan"
logger.info(f"[QA Client] Envoi de la requête de scan pour le dépôt : {repo_url}")
try:
response = await async_client.post(sandbox_url, timeout=120.0)
response = await async_client.post(sandbox_url, params={"repo_url": repo_url}, timeout=150.0)
if response.status_code == 200:
data = response.json()
return QARawResults.model_validate(data.get("results"))
if data is None:
logger.error("[QA Client] ❌ La Sandbox a répondu 200 OK mais a renvoyé une réponse vide (null).")
return QARawResults(
is_executable=False,
runtime=RuntimeOutput(
exit_code=-1,
stdout="",
stderr="Erreur interne Sandbox : Réponse JSON vide.",
duration_seconds=0,
timeout_triggered=False
),
issues=[]
)
results = data.get("results", {})
if not results.get("is_executable"):
logger.error(f"[QA Client] ❌ Erreur interne Sandbox (Semgrep STDERR) : {results.get('runtime', {}).get('stderr')}")
return QARawResults.model_validate(results)
else:
logger.error(f"[QA Client] Erreur de la Sandbox (Status {response.status_code}): {response.text}")
return QARawResults(

10094
backend/app/sandbox/rules/ci.yaml Normal file
View File

File diff suppressed because it is too large Load Diff

20054
backend/app/sandbox/rules/cwe-top-25.yaml Normal file
View File

File diff suppressed because it is too large Load Diff

70779
backend/app/sandbox/rules/default.yaml Normal file
View File

File diff suppressed because it is too large Load Diff

41421
backend/app/sandbox/rules/owasp-top-ten.yaml Normal file
View File

File diff suppressed because it is too large Load Diff

13698
backend/app/sandbox/rules/security-audit.yaml Normal file
View File

File diff suppressed because it is too large Load Diff

9
backend/app/schemas/qa_report.py
View File

@@ -29,4 +29,11 @@ class QAEvaluation(BaseModel):
technical_feedback: List[str] = Field(
default_factory=list,
description="Instructions de correction ultra-précises destinées au Dev Agent"
)
)
class StructureEvaluation(BaseModel):
is_valid: bool = Field(description="True si le projet possède un README, un fichier de dépendance cohérent et un point d'entrée d'exécution.")
detected_language: str = Field(description="Le langage principal identifié d'après la structure (ex: Python, Node, Go, Rust...).")
missing_elements: list[str] = Field(description="Liste des fichiers obligatoires manquants (ex: ['README.md', 'requirements.txt']). Vide si tout est OK.")
feedback: str = Field(description="Courte explication des manquements ou félicitations sur la structure.")