diff --git a/commandes/bug_report.py b/commandes/bug_report.py index d2656d2..a0b7fd5 100644 --- a/commandes/bug_report.py +++ b/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 - issue_iid = str( - payload.get("object_attributes", {}).get("iid") or - payload.get("issue", {}).get("iid") - ) - + # 🕵️ 2. RÉCUPÉRATION DE L'ID DU TICKET + issue_iid = extract_issue_iid(payload) 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() - dev = app_info.owner - issue_title = payload.get("object_attributes", {}).get("title", f"#{issue_iid}") + + # 🕵️ 3. RÉCUPÉRATION DU TITRE + issue_title = ( + 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é) - note_attr = payload.get("object_attributes", {}) - is_system = note_attr.get("system", False) - - if not is_system: + # 🕵️ 4. EXTRACTION DU COMMENTAIRE FORGEJO / GITLAB + if "comment" in payload: + # Mode Forgejo + note_body = payload["comment"].get("body", "") + 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: - 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" - + if not is_system and note_body: try: embed = disnake.Embed( - title="🛠️ Mise à jour de 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}", - color=disnake.Color.blue() + title="💬 Nouveau commentaire sur votre signalement", + description=f"L'équipe 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) + 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 label: {e}") + kuby_logger.info(f"Commentaire 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}") + + 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}") diff --git a/serveur_flask.py b/serveur_flask.py index 190e4c8..c28f392 100644 --- a/serveur_flask.py +++ b/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' - if request.headers.get('X-Gitea-Token') != WEBHOOK_SECRET: + # 1. Vérification de la signature HMAC + 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 - received_token = request.headers.get('X-Gitea-Token') - - print(f"[DEBUG] Webhook d'événements reçu. Token reçu: '{received_token}'") + # 1. On récupère la signature envoyée par Forgejo/Gitea + signature_recue = None + for header_name in ('X-Hub-Signature-256', 'X-Gitea-Signature', 'X-Gogs-Signature'): + signature_recue = request.headers.get(header_name) + if signature_recue: + break - if received_token != WEBHOOK_SECRET: - abort(403) + # 2. On lit les données brutes envoyées par Forgejo + 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) \ No newline at end of file + app.run(host='0.0.0.0', port=5000) +