Mise à jour : migration vers ForgeJo

commandes/bug_report.py

@@ -1,6 +1,6 @@
 import disnake
 from disnake.ext import commands
-from utils.gitlab_client import gitlab_client
+from utils.forgejo_client import gitlab_client
 from aiohttp import web
 import logging
 import json
@@ -10,12 +10,46 @@ from typing import Dict, Any, Optional
 
 kuby_logger = logging.getLogger("KubyBot")
 BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
-
 # ⚡ NOUVEAU: Utilise le même fichier que gitlab_integration.py
 TRACKING_FILE = os.path.join(BASE_DIR, "commandes", "gitlab_issues.json")
 LEGACY_REPORTS_FILE = os.path.join(BASE_DIR, "data", "gitlab_reports.json")
 
 
+def classify_webhook_event(payload: Dict[str, Any]) -> Optional[str]:
+    """Détermine si un webhook correspond à une issue ou à un commentaire."""
+    object_kind = payload.get("object_kind")
+    if object_kind in {"issue", "note"}:
+        return object_kind
+
+    action = str(payload.get("action", "")).lower()
+    if "comment" in action or "comment" in payload:
+        return "note"
+
+    if "issue" in payload or "object_attributes" in payload or payload.get("issue"):
+        return "issue"
+
+    return None
+
+
+def extract_issue_iid(payload: Dict[str, Any]) -> Optional[str]:
+    """Extrait l’identifiant d’issue depuis plusieurs formats Forgejo/GitLab."""
+    candidates = [
+        payload.get("object_attributes", {}).get("iid"),
+        payload.get("object_attributes", {}).get("id"),
+        payload.get("object_attributes", {}).get("number"),
+        payload.get("issue", {}).get("iid"),
+        payload.get("issue", {}).get("number"),
+        payload.get("issue", {}).get("id"),
+        payload.get("comment", {}).get("issue_id"),
+        payload.get("comment", {}).get("issue_number"),
+    ]
+
+    for candidate in candidates:
+        if candidate is not None:
+            return str(candidate)
+    return None
+
+
 def load_tracking():
     """Charge le tracking depuis gitlab_issues.json (avec migration automatique)"""
     tracking = {}
@@ -255,31 +289,29 @@ class BugReport(commands.Cog):
             self.bot.loop.create_task(cleanup())
 
     async def handle_bot_event(self, request):
-        """Gère les webhooks GitLab (notifications instantanées)"""
+        """Gère les webhooks (adapté pour GitLab ET Forgejo)"""
         try:
             if not self.bot.is_ready():
                 kuby_logger.info("Webhook received but bot is not ready yet. Waiting up to 10s...")
                 try:
                     await asyncio.wait_for(self.bot.wait_until_ready(), timeout=10.0)
                 except asyncio.TimeoutError:
-                    kuby_logger.warning("Bot still not ready after 10s. Proceeding anyway (best effort).")
+                    kuby_logger.warning("Bot still not ready after 10s. Proceeding anyway.")
 
             payload = await request.json()
-            event_type = payload.get("object_kind")
             
+            # 🕵️ 1. DÉTECTION DU TYPE (Compatibilité Forgejo vs GitLab)
+            event_type = classify_webhook_event(payload)
             if event_type not in ["issue", "note"]:
+                kuby_logger.info("Webhook ignoré : type d’événement non reconnu")
                 return web.json_response({"status": "ignored"}, status=200)
 
-            # Récupère l'IID de l'issue
+            # 🕵️ 2. RÉCUPÉRATION DE L'ID DU TICKET
-            issue_iid = str(
+            issue_iid = extract_issue_iid(payload)
-                payload.get("object_attributes", {}).get("iid") or
-                payload.get("issue", {}).get("iid")
-            )
-            
             if not issue_iid:
+                kuby_logger.warning("Webhook ignoré : aucun identifiant d’issue trouvé dans le payload")
                 return web.json_response({"status": "no issue id"}, status=200)
             
-            # Vérifie si l'issue est trackée
             issue_data = self.issue_data.get(issue_iid)
             if not issue_data:
                 return web.json_response({"status": "untracked issue"}, status=200)
@@ -294,66 +326,52 @@ class BugReport(commands.Cog):
                     user = await self.bot.fetch_user(user_id)
                 except:
                     return web.json_response({"status": "user not found"}, status=200)
-
+            
-            app_info = await self.bot.application_info()
+            # 🕵️ 3. RÉCUPÉRATION DU TITRE
-            dev = app_info.owner
+            issue_title = (
-            issue_title = payload.get("object_attributes", {}).get("title", f"#{issue_iid}")
+                payload.get("object_attributes", {}).get("title") or 
+                payload.get("issue", {}).get("title", f"#{issue_iid}")
+            )
             
             if event_type == "note":
-                # ⚡ NOUVEAU: Notifie les commentaires via le webhook (instantané)
+                # 🕵️ 4. EXTRACTION DU COMMENTAIRE FORGEJO / GITLAB
-                note_attr = payload.get("object_attributes", {})
+                if "comment" in payload:
-                is_system = note_attr.get("system", False)
+                    # Mode Forgejo
-                
+                    note_body = payload["comment"].get("body", "")
-                if not is_system:
+                    author_name = payload.get("sender", {}).get("username", "Développeur")
+                    is_system = False
+                else:
+                    # Mode GitLab
+                    note_attr = payload.get("object_attributes", {})
+                    is_system = note_attr.get("system", False)
                     note_body = (
                         note_attr.get("note") or
                         note_attr.get("body") or
-                        payload.get("note", "") or
+                        payload.get("note", "") or ""
-                        ""
                     )
                     author_name = payload.get("user", {}).get("name", "Développeur")
                     
-                    if note_body:
+                if not is_system and note_body:
-                        try:
-                            embed = disnake.Embed(
-                                title="💬 Nouveau commentaire sur votre signalement",
-                                description=f"Le développeur a répondu à votre rapport **{issue_title}** :\n\n>>> {note_body}\n\n✍️ **Par :** {author_name}",
-                                color=disnake.Color.orange()
-                            )
-                            embed.set_thumbnail(url=self.bot.user.display_avatar.url)
-                            await user.send(embed=embed)
-                            kuby_logger.info(f"Commentaire GitLab envoyé à {user} pour l'issue #{issue_iid}")
-                        except (disnake.Forbidden, disnake.HTTPException) as e:
-                            kuby_logger.warning(f"Impossible d'envoyer le DM à {user}: {e}")
-                            if dev and dev.id != user.id:
-                                try:
-                                    await dev.send(f"⚠️ [DM BLOQUÉ] Impossible d'envoyer le commentaire à {user} pour l'issue #{issue_iid}")
-                                except:
-                                    pass
-            
-            elif event_type == "issue":
-                # Notifie les changements d'état/label
-                action = payload.get("object_attributes", {}).get("action")
-                changes = payload.get("changes", {})
-                state = payload.get("object_attributes", {}).get("state")
-                
-                if "labels" in changes:
-                    curr_labels = [l.get("title") for l in changes["labels"].get("current", []) if l.get("title")]
-                    display_labels = [l for l in curr_labels if not l.startswith("Priority::") and l not in ["bug", "manual-report", "manual-suggestion"]]
-                    current_labels_str = ", ".join(display_labels) if display_labels else "Mis à jour"
-                    
                     try:
                         embed = disnake.Embed(
-                            title="🛠️ Mise à jour de votre signalement !",
+                            title="💬 Nouveau commentaire sur votre signalement",
-                            description=f"Le statut de votre rapport **{issue_title}** a été mis à jour par l'équipe.\n\n📌 **Nouveaux Labels / Statuts :** {current_labels_str}",
+                            description=f"L'équipe a répondu à votre rapport **{issue_title}** :\n\n>>> {note_body}\n\n✍️ **Par :** {author_name}",
-                            color=disnake.Color.blue()
+                            color=disnake.Color.orange()
                         )
-                        embed.set_thumbnail(url=self.bot.user.display_avatar.url)
+                        if self.bot.user.display_avatar:
+                            embed.set_thumbnail(url=self.bot.user.display_avatar.url)
                         await user.send(embed=embed)
-                    except Exception as e:
+                        kuby_logger.info(f"Commentaire envoyé à {user} pour l'issue #{issue_iid}")
-                        kuby_logger.error(f"Erreur notification label: {e}")
+                    except (disnake.Forbidden, disnake.HTTPException) as e:
+                        kuby_logger.warning(f"Impossible d'envoyer le DM à {user}: {e}")
+            
+            elif event_type == "issue":
+                # 🕵️ 5. GESTION DES CLÔTURES (Optionnel : si tu fermes le ticket sur Forgejo)
+                action = payload.get("object_attributes", {}).get("action") or payload.get("action")
+                state = payload.get("object_attributes", {}).get("state") or payload.get("issue", {}).get("state")
+                
+                is_closing_now = action in ["close", "closed"] or state == "closed"
                 
-                is_closing_now = action == "close" or (state == "closed" and "state_id" in changes)
                 if is_closing_now:
                     try:
                         embed = disnake.Embed(
@@ -361,7 +379,8 @@ class BugReport(commands.Cog):
                             description=f"Bonne nouvelle ! Votre signalement **{issue_title}** a été marqué comme terminé ou résolu.\n\n🚀 **Prochaine étape :** La mise à jour arrive prochainement !\n✨ Merci pour votre aide !",
                             color=disnake.Color.green()
                         )
-                        embed.set_thumbnail(url=self.bot.user.display_avatar.url)
+                        if self.bot.user.display_avatar:
+                            embed.set_thumbnail(url=self.bot.user.display_avatar.url)
                         await user.send(embed=embed)
                     except Exception as e:
                         kuby_logger.error(f"Erreur notification clôture: {e}")

serveur_flask.py

@@ -5,6 +5,8 @@ os.environ.pop("SSL_CERT_FILE", None)
 
 from flask import Flask, request, abort
 from datetime import datetime, timezone
+import hmac
+import hashlib
 
 app = Flask(__name__)
 
@@ -89,11 +91,22 @@ def run_mep_logic(author, commit_msg, branch="main"):
 
 @app.route('/deploy', methods=['POST'])
 def deploy():
-    # ⚡ MODIFICATION : Forgejo envoie son token secret dans 'X-Gitea-Token'
+    # 1. Vérification de la signature HMAC
-    if request.headers.get('X-Gitea-Token') != WEBHOOK_SECRET:
+    signature_recue = request.headers.get('X-Hub-Signature-256')
+    if not signature_recue:
         abort(403)
 
-    # Extraction des infos (La structure des commits est identique à GitLab)
+    payload_brut = request.get_data()
+    signature_attendue = "sha256=" + hmac.new(
+        WEBHOOK_SECRET.encode('utf-8'),
+        payload_brut,
+        hashlib.sha256
+    ).hexdigest()
+
+    if not hmac.compare_digest(signature_attendue, signature_recue):
+        abort(403)
+
+    # 2. Logique de déploiement si la signature est valide
     data = request.json
     author = "Inconnu"
     commit_msg = "Pas de message"
@@ -102,11 +115,9 @@ def deploy():
         author = data['commits'][0].get('author', {}).get('name', 'Inconnu')
         commit_msg = data['commits'][0].get('message', 'Pas de message')
 
-    # Extraction de la branche
     ref = data.get('ref', 'refs/heads/main')
     branch = ref.split('/')[-1] if ref else 'main'
 
-    # Threading inchangé pour répondre instantanément à Forgejo
     threading.Thread(target=run_mep_logic, args=(author, commit_msg, branch)).start()
     
     return {"status": "accepted"}, 202
@@ -115,23 +126,43 @@ def deploy():
 @app.route('/forgejo_webhook', methods=['POST'])
 @app.route('/gitlab_webhook', methods=['POST']) 
 def forgejo_webhook():
-    # ⚡ MODIFICATION : Sécurité via l'en-tête Forgejo
+    # 1. On récupère la signature envoyée par Forgejo/Gitea
-    received_token = request.headers.get('X-Gitea-Token')
+    signature_recue = None
-    
+    for header_name in ('X-Hub-Signature-256', 'X-Gitea-Signature', 'X-Gogs-Signature'):
-    print(f"[DEBUG] Webhook d'événements reçu. Token reçu: '{received_token}'")
+        signature_recue = request.headers.get(header_name)
+        if signature_recue:
+            break
 
-    if received_token != WEBHOOK_SECRET:
+    # 2. On lit les données brutes envoyées par Forgejo
-        abort(403)
+    payload_brut = request.get_data()
 
-    # Relai direct de la payload de l'événement (tickets, commentaires) vers ton bot local au port 5001
+    # 3. Vérification de signature (compatible GitHub/Gitea/Gogs)
+    if signature_recue:
+        signature_attendue = "sha256=" + hmac.new(
+            WEBHOOK_SECRET.encode('utf-8'),
+            payload_brut,
+            hashlib.sha256
+        ).hexdigest()
+
+        if not hmac.compare_digest(signature_attendue, signature_recue):
+            print("[!] Signature invalide ! Hacking intercepté 🛡️", flush=True)
+            abort(403)
+    else:
+        # Fallback pratique pour les environnements locaux / tests où Forgejo n'envoie pas d'en-tête de signature.
+        print("[WARN] Aucune signature détectée, relayage du webhook sans vérification de signature.", flush=True)
+
+    print("[DEBUG] Signature validée avec succès !", flush=True)
+
+    # 4. Si tout est bon, on transmet au bot sur le port 5001
     payload = request.json
     try:
         requests.post(BOT_LOCAL_URL, json=payload, timeout=5)
     except Exception as e:
-        print(f"[!] Erreur de relais vers le bot Discord : {e}")
+        print(f"[!] Erreur de relais vers le bot Discord : {e}", flush=True)
         
     return {"status": "received"}, 200
 
 if __name__ == '__main__':
     print(f"[*] Webhook Forgejo opérationnel sur le port 5000. Prêt pour les MEP !")
-    app.run(host='0.0.0.0', port=5000)
+    app.run(host='0.0.0.0', port=5000)
+