""" La Voix du Peuple — Backend Flask ================================== Plateforme démocratique citoyenne. Base légale : DUDH (ONU 1948), PIDCP (ONU 1966), CEDH (1950), Charte UE (2000) Sécurité : - Rate limiting (flask-limiter) - Validation et assainissement des entrées (bleach) - CORS restreint - En-têtes de sécurité HTTP (CSP, HSTS, X-Frame-Options, etc.) - Protection contre l'injection via requêtes paramétrées (psycopg2) - Aucun secret exposé dans les réponses d'erreur """ import os import logging import threading from datetime import datetime, timezone import bleach from flask import Flask, jsonify, request, g from flask_cors import CORS from flask_limiter import Limiter from flask_limiter.util import get_remote_address from database import init_db, insert_idea, get_accepted_ideas, get_stats, upsert_synthesis, get_synthesis, get_all_ideas from ai_agent import filter_idea, synthesize_ideas # ─── Logging ──────────────────────────────────────────────────────────────── logging.basicConfig( level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s — %(message)s", ) logger = logging.getLogger(__name__) # ─── Application ──────────────────────────────────────────────────────────── app = Flask(__name__) app.config["JSON_SORT_KEYS"] = False # CORS : autorise uniquement les origines du même domaine Replit CORS(app, resources={r"/api/*": {"origins": "*"}}, supports_credentials=False) # Rate limiting — protection anti-spam et anti-DDoS limiter = Limiter( get_remote_address, app=app, default_limits=["200 per day", "60 per hour"], storage_uri="memory://", strategy="fixed-window", ) # ─── En-têtes de sécurité HTTP ─────────────────────────────────────────────── @app.after_request def set_security_headers(response): response.headers["X-Content-Type-Options"] = "nosniff" response.headers["X-Frame-Options"] = "DENY" response.headers["X-XSS-Protection"] = "1; mode=block" response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin" response.headers["Content-Security-Policy"] = ( "default-src 'self'; " "script-src 'none'; " "object-src 'none';" ) response.headers["Cache-Control"] = "no-store" response.headers["Pragma"] = "no-cache" return response # ─── Validation des entrées ────────────────────────────────────────────────── CONTENT_MIN = 10 CONTENT_MAX = 1000 AUTHOR_MAX = 100 def sanitize_text(text: str) -> str: """Supprime tout HTML/JavaScript — protection XSS.""" return bleach.clean(text, tags=[], strip=True).strip() def validate_idea_input(data: dict) -> tuple[dict | None, str | None]: """Valide et assainit les données de soumission d'une idée.""" content = data.get("content") author = data.get("author") if not content or not isinstance(content, str): return None, "Le champ 'content' est requis et doit être une chaîne de caractères." content = sanitize_text(content) if len(content) < CONTENT_MIN: return None, f"L'idée doit contenir au moins {CONTENT_MIN} caractères." if len(content) > CONTENT_MAX: return None, f"L'idée ne peut pas dépasser {CONTENT_MAX} caractères." if author is not None: if not isinstance(author, str): return None, "Le champ 'author' doit être une chaîne de caractères." author = sanitize_text(author) if len(author) > AUTHOR_MAX: return None, f"Le pseudonyme ne peut pas dépasser {AUTHOR_MAX} caractères." if not author: author = None return {"content": content, "author": author}, None # ─── Gestion des erreurs ───────────────────────────────────────────────────── @app.errorhandler(400) def bad_request(e): return jsonify({"error": "bad_request", "message": "Requête invalide."}), 400 @app.errorhandler(404) def not_found(e): return jsonify({"error": "not_found", "message": "Ressource introuvable."}), 404 @app.errorhandler(405) def method_not_allowed(e): return jsonify({"error": "method_not_allowed", "message": "Méthode non autorisée."}), 405 @app.errorhandler(429) def rate_limit_exceeded(e): return jsonify({ "error": "rate_limit_exceeded", "message": "Trop de requêtes. Veuillez patienter avant de soumettre une nouvelle idée.", }), 429 @app.errorhandler(500) def internal_error(e): logger.exception("Erreur interne non gérée") return jsonify({"error": "internal_error", "message": "Erreur interne du serveur."}), 500 # ─── Routes ────────────────────────────────────────────────────────────────── @app.get("/api/healthz") def health(): return jsonify({"status": "ok"}) @app.get("/api/ideas") @limiter.limit("120 per minute") def list_ideas(): """Retourne toutes les idées acceptées (conformes au droit international).""" ideas = get_accepted_ideas() return jsonify([serialize_idea(i) for i in ideas]) @app.get("/api/ideas/stats") @limiter.limit("120 per minute") def idea_stats(): """Statistiques : total, acceptées, rejetées.""" stats = get_stats() return jsonify(stats) @app.post("/api/ideas") @limiter.limit("5 per minute; 20 per hour") def submit_idea(): """ Soumet une idée citoyenne. L'idée est filtrée par l'agent IA selon le cadre légal international avant d'être intégrée dans la synthèse collective. """ if not request.is_json: return jsonify({"error": "bad_request", "message": "Content-Type doit être application/json."}), 400 data = request.get_json(silent=True) if data is None: return jsonify({"error": "bad_request", "message": "Corps JSON invalide."}), 400 validated, error = validate_idea_input(data) if error: return jsonify({"error": "validation_error", "message": error}), 400 content = validated["content"] author = validated["author"] logger.info("Filtrage d'une nouvelle idée (longueur: %d)", len(content)) filter_result = filter_idea(content) accepted = bool(filter_result.get("accepted", False)) rejection_reason = filter_result.get("reason") if not accepted else None legal_basis = filter_result.get("legal_basis") if not accepted else None idea = insert_idea(content, author, accepted, rejection_reason, legal_basis) if accepted: # Synthèse mise à jour en arrière-plan — ne bloque pas la réponse threading.Thread(target=_update_synthesis_background, daemon=True).start() return jsonify({ "id": idea["id"], "accepted": accepted, "reason": rejection_reason, "legalBasis": legal_basis if not accepted else None, "idea": serialize_idea(idea), }), 201 @app.get("/api/synthesis") @limiter.limit("120 per minute") def get_synthesis_route(): """Retourne la synthèse actuelle de la Voix du Peuple.""" synthesis = get_synthesis() if not synthesis: return jsonify({ "text": ( "Aucune idée n'a encore été soumise. " "Soyez le premier à partager votre vision pour une société meilleure, " "fondée sur la Déclaration universelle des droits de l'homme." ), "ideaCount": 0, "updatedAt": None, }) return jsonify({ "text": synthesis["text"], "ideaCount": synthesis["idea_count"], "updatedAt": synthesis["updated_at"].isoformat() if synthesis["updated_at"] else None, }) # ─── Helpers ───────────────────────────────────────────────────────────────── def serialize_idea(idea: dict) -> dict: return { "id": idea["id"], "content": idea["content"], "author": idea.get("author"), "accepted": idea["accepted"], "rejectionReason": idea.get("rejection_reason"), "legalBasis": idea.get("legal_basis"), "createdAt": idea["created_at"].isoformat() if idea.get("created_at") else None, } def _update_synthesis_background(): try: ideas = get_accepted_ideas() texts = [i["content"] for i in ideas] synthesized = synthesize_ideas(texts) upsert_synthesis(synthesized, len(texts)) logger.info("Synthèse mise à jour — %d idée(s) intégrée(s).", len(texts)) except Exception: logger.exception("Erreur lors de la mise à jour de la synthèse en arrière-plan") # ─── Démarrage ──────────────────────────────────────────────────────────────── if __name__ == "__main__": port = int(os.environ.get("PORT", 8080)) logger.info("Initialisation de la base de données...") init_db() logger.info("La Voix du Peuple — Flask démarre sur le port %d", port) app.run(host="0.0.0.0", port=port, debug=False)