import os, subprocess, threading, requests # Nettoyer les variables de certificats SSL corrompues ou obsolètes pour éviter les crashs TLS os.environ.pop("REQUESTS_CA_BUNDLE", None) os.environ.pop("SSL_CERT_FILE", None) from flask import Flask, request, abort from datetime import datetime, timezone import hashlib import hmac app = Flask(__name__) # --- CONFIGURATION --- PROJECT_PATH = "/home/discord/Bot/Kuby" WEBHOOK_SECRET = "Nois2" PROCESS_NAME = "kuby" # webhook de logs Discord DISCORD_LOG_URL = "https://discord.com/api/webhooks/1482148910888652910/IA9CcOWtjGswbuxMaOu_6uaclv5zojZo4ttxtV6RYyZzJ9gW7BF7xp_Zv3oPIjYEuh8o" # Bot Local Proxy (Relai vers le Cog de ton bot) BOT_LOCAL_URL = "http://127.0.0.1:5001/bot_event" def relay_payload_to_bot(payload): try: requests.post(BOT_LOCAL_URL, json=payload, timeout=5) except Exception as e: print(f"[!] Erreur de relais vers le bot Discord : {e}") def verify_signature(payload_brut, received_signature): if not received_signature: return False expected_signature = "sha256=" + hmac.new( WEBHOOK_SECRET.encode("utf-8"), payload_brut, hashlib.sha256 ).hexdigest() return hmac.compare_digest(expected_signature, received_signature) def send_discord_log(status, message, color, author=None, commit_msg=None): embed = { "title": f"🚀 Kuby V2 [Forgejo] : {status}", "description": message, "color": color, "timestamp": datetime.now(timezone.utc).isoformat(), "fields": [] } if author: embed["fields"].append({"name": "👤 Auteur", "value": author, "inline": True}) if commit_msg: embed["fields"].append({"name": "📝 Commit", "value": commit_msg, "inline": True}) payload = {"embeds": [embed]} try: requests.post(DISCORD_LOG_URL, json=payload, timeout=10) except Exception as e: print(f"[!] Erreur envoi Discord : {e}") def run_mep_logic(author, commit_msg, branch="main"): start_time = datetime.now() send_discord_log("EN COURS", f"Déploiement branch **{branch}** lancé sur la machine de guerre...", 3447003, author, commit_msg) try: target_path = PROJECT_PATH use_rsync = False if not os.path.exists(os.path.join(target_path, ".git")): alt_path = "/home/discord/Bot/Kuby" if os.path.exists(os.path.join(alt_path, ".git")): print(f"[!] {target_path} n'est pas un dépôt Git. Utilisation du dépôt {alt_path} et rsync.") target_path = alt_path use_rsync = True else: raise RuntimeError(f"Aucun dépôt Git trouvé dans {target_path} ou {alt_path}") os.chdir(target_path) # 1. GIT (Va maintenant pull depuis ton Forgejo auto-hébergé) subprocess.run(["git", "fetch", "origin"], check=True) subprocess.run(["git", "reset", "--hard", f"origin/{branch}"], check=True) # 2. RSYNC (Fallback) if use_rsync: subprocess.run(["rsync", "-a", "--exclude=.git", "--exclude=venv", "--exclude=.venv", "--exclude=__pycache__", f"{target_path}/", f"{PROJECT_PATH}/"], check=True) # 3. Redémarrage via systemd subprocess.run(["systemctl", "restart", "kuby-bot.service"], check=True, capture_output=True, text=True) duration = (datetime.now() - start_time).total_seconds() msg = f"✅ **Déploiement réussi en {duration:.2f}s**\nLe bot est à jour sur Forgejo et redémarré." send_discord_log("VALIDÉ", msg, 3066993, author, commit_msg) except subprocess.CalledProcessError as e: error_stderr = e.stderr if e.stderr else "Pas de log d'erreur disponible." error_msg = f"❌ **ERREUR de commande (Status {e.returncode})**\nCommande : `{ ' '.join(e.cmd) }`\n\n**Détails :**\n```\n{error_stderr}\n```" send_discord_log("ÉCHEC", error_msg, 15158332, author, commit_msg) except Exception as e: error_msg = f"❌ **ERREUR de déploiement**\n```python\n{str(e)}\n```" send_discord_log("ÉCHEC", error_msg, 15158332, author, commit_msg) @app.route('/deploy', methods=['POST']) def deploy(): # 1. Vérification de la signature HMAC signature_recue = request.headers.get('X-Hub-Signature-256') if not signature_recue: abort(403) 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" if data and 'commits' in data and len(data['commits']) > 0: author = data['commits'][0].get('author', {}).get('name', 'Inconnu') commit_msg = data['commits'][0].get('message', 'Pas de message') ref = data.get('ref', 'refs/heads/main') branch = ref.split('/')[-1] if ref else 'main' threading.Thread(target=run_mep_logic, args=(author, commit_msg, branch)).start() return {"status": "accepted"}, 202 # Nouvelle route renommée pour la clarté, accepte aussi l'ancienne au cas où @app.route('/forgejo_webhook', methods=['POST']) @app.route('/gitlab_webhook', methods=['POST']) def forgejo_webhook(): # 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 # 2. On lit les données brutes envoyées par Forgejo payload_brut = request.get_data() # 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}", 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)