Résoudre automatiquement les problèmes SonarQube sur GitLab

stages:
  - analyze
  - remediate

variables:
  SONAR_TOKEN: "$SONAR_TOKEN"
  SONAR_ORG: "$SONAR_ORG"
  SONAR_PROJECT_KEY: "Colhodm_juice-shop"
  DEVIN_API_KEY: "$DEVIN_API_KEY"
  GIT_STRATEGY: clone

analyze:
  stage: analyze
  image: sonarsource/sonar-scanner-cli:latest
  script:
    - sonar-scanner -Dsonar.organization=$SONAR_ORG
  only:
    - branches
    - merge_requests

setup_python:
  stage: remediate
  image: python:3.9
  before_script:
    - python --version
    - pip install aiohttp
  script:
    - git config --global user.name "GitLab Runner"
    - git config --global user.email "runner@gitlab.com"
    - python3 $CI_PROJECT_DIR/.gitlab/scripts/devin_remediation.py
  only:
    - branches
    - merge_requests

python
import asyncio
import aiohttp
import os
from datetime import datetime

# Variables d'environnement
GITHUB_REPOSITORY = os.getenv("GITHUB_REPOSITORY")
SONAR_TOKEN = os.getenv("SONAR_TOKEN")
DEVIN_API_KEY = os.getenv("DEVIN_API_KEY")
SONAR_ORG = os.getenv("SONAR_ORG")
SONAR_PROJECT_KEY = os.getenv("SONAR_PROJECT_KEY")
DEVIN_API_BASE = "https://api.devin.ai/v1"

async def get_sonarcloud_issues():
    """Récupère les vulnérabilités ouvertes depuis SonarCloud."""
    url = "https://sonarcloud.io/api/issues/search"
    headers = {"Authorization": f"Bearer {SONAR_TOKEN}"}
    params = {
        "organization": SONAR_ORG,
        "projectKeys": SONAR_PROJECT_KEY,
        "types": "VULNERABILITY",
        "statuses": "OPEN"
    }

    async with aiohttp.ClientSession() as session:
        async with session.get(url, headers=headers, params=params) as response:
            if response.status != 200:
                print(f"Erreur lors de la récupération des problèmes SonarCloud : {await response.text()}")
                return []
            result = await response.json()
            print(f"{len(result.get('issues', []))} problèmes trouvés")
            return result.get('issues', [])

async def delegate_task_to_devin(issue):
    """Délègue l'ensemble de la tâche de correction, commit et push à Devin AI."""
    async with aiohttp.ClientSession() as session:
        headers = {"Authorization": f"Bearer {DEVIN_API_KEY}"}
        prompt = f"""
        Corrigez la vulnérabilité suivante dans {GITHUB_REPOSITORY} : {issue['message']} dans le fichier {issue['component']}.
        1. Créez une nouvelle branche nommée 'devin/{issue['key']}-fix-vulnerability'.
        2. Implémentez la correction.
        3. Rédigez un message de commit détaillé expliquant les modifications :
            - Clé du problème : {issue['key']}
            - Composant : {issue['component']}
            - Corrigé par Devin AI le {datetime.now().isoformat()}
            - Incluez 'Co-authored-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>'.
        4. Poussez la branche vers le dépôt distant.
        5. Ouvrez une pull request avec une description de la correction.
        """

        data = {"prompt": prompt, "idempotent": True}

        async with session.post(f"{DEVIN_API_BASE}/sessions", json=data, headers=headers) as response:
            if response.status != 200:
                print(f"Erreur lors de la délégation de la tâche à Devin : {await response.text()}")
                return None
            result = await response.json()
            print(f"Session Devin créée : {result}")
            return result

async def monitor_devin_session(session_id):
    """Surveille la progression de Devin jusqu'à ce qu'il termine la tâche."""
    async with aiohttp.ClientSession() as session:
        headers = {"Authorization": f"Bearer {DEVIN_API_KEY}"}

        while True:
            async with session.get(f"{DEVIN_API_BASE}/sessions/{session_id}", headers=headers) as response:
                if response.status != 200:
                    print(f"Erreur lors de la surveillance de la session Devin : {await response.text()}")
                    return None

                result = await response.json()
                status = result.get("status_enum")

                if status in ["completed", "stopped"]:
                    print(f"Devin a terminé la tâche : {result}")
                    return result
                elif status == "blocked":
                    print("Devin a rencontré un problème. Veuillez vérifier manuellement.")
                    return None

                await asyncio.sleep(5)

async def main():
    try:
        issues = await get_sonarcloud_issues()

        for issue in issues:
            print(f"Traitement du problème : {issue['key']}")

            # Déléguer la tâche à Devin AI
            session_data = await delegate_task_to_devin(issue)

            if session_data:
                session_id = session_data["session_id"]

                # Surveiller la progression de Devin
                await monitor_devin_session(session_id)

    except Exception as e:
        print(f"Erreur survenue : {str(e)}")
        raise

if __name__ == "__main__":
    asyncio.run(main())