292 lines
13 KiB
Python
292 lines
13 KiB
Python
|
#!/usr/bin/env python3
|
||
|
|
# -*- coding: utf-8 -*-
|
||
|
|
"""
|
||
|
|
OmegaKube — Formulaire de contact CGI
|
||
|
|
"""
|
||
|
|
|
||
|
|
import os, sys, json, time, re, sqlite3, smtplib
|
||
|
|
from urllib.parse import urlparse
|
||
|
|
from datetime import datetime
|
||
|
|
from email.mime.text import MIMEText
|
||
|
|
from email.header import Header
|
||
|
|
|
||
|
|
sys.stdout.reconfigure(encoding='utf-8')
|
||
|
|
|
||
|
|
# ─── CONFIGURATION SMTP ────────────────────────────────────────────────────────
|
||
|
|
# IMPORTANT: Configurez ces paramètres selon votre hébergeur SMTP
|
||
|
|
#
|
||
|
|
# Exemples de configuration pour différents fournisseurs :
|
||
|
|
#
|
||
|
|
# --- Local (Postfix/Exim sur le serveur) ---
|
||
|
|
# SMTP_HOST = "localhost"
|
||
|
|
# SMTP_PORT = 25
|
||
|
|
# SMTP_SSL = False
|
||
|
|
# SMTP_USER = "" # Pas d'authentification
|
||
|
|
# SMTP_PASSWORD = ""
|
||
|
|
#
|
||
|
|
# --- Gmail ---
|
||
|
|
# SMTP_HOST = "smtp.gmail.com"
|
||
|
|
# SMTP_PORT = 587
|
||
|
|
# SMTP_SSL = False # Utilise STARTTLS
|
||
|
|
# SMTP_USER = "votre@email.com"
|
||
|
|
# SMTP_PASSWORD = "votre_mot_de_passe_ou_app_password"
|
||
|
|
#
|
||
|
|
# --- SMTP Gmail avec SSL ---
|
||
|
|
# SMTP_HOST = "smtp.gmail.com"
|
||
|
|
# SMTP_PORT = 465
|
||
|
|
# SMTP_SSL = True # SSL direct
|
||
|
|
# SMTP_USER = "votre@email.com"
|
||
|
|
# SMTP_PASSWORD = "votre_mot_de_passe_ou_app_password"
|
||
|
|
#
|
||
|
|
# --- OVH ---
|
||
|
|
# SMTP_HOST = "ssl0.ovh.net" # ou votre serveur OVH
|
||
|
|
# SMTP_PORT = 587
|
||
|
|
# SMTP_SSL = False
|
||
|
|
# SMTP_USER = "contact@omegakube.fr"
|
||
|
|
# SMTP_PASSWORD = "votre_mot_de_passe"
|
||
|
|
#
|
||
|
|
# --- Mailjet ---
|
||
|
|
# SMTP_HOST = "in-v3.mailjet.com"
|
||
|
|
# SMTP_PORT = 587
|
||
|
|
# SMTP_SSL = False
|
||
|
|
# SMTP_USER = "votre_api_key"
|
||
|
|
# SMTP_PASSWORD = "votre_api_secret"
|
||
|
|
#
|
||
|
|
# --- SendGrid ---
|
||
|
|
# SMTP_HOST = "smtp.sendgrid.net"
|
||
|
|
# SMTP_PORT = 587
|
||
|
|
# SMTP_SSL = False
|
||
|
|
# SMTP_USER = "apikey"
|
||
|
|
# SMTP_PASSWORD = "votre_sendgrid_api_key"
|
||
|
|
#
|
||
|
|
# Pour tester la connexion SMTP, exécutez :
|
||
|
|
# python3 -c "import smtplib; srv = smtplib.SMTP('localhost', 25); srv.quit(); print('OK')"
|
||
|
|
|
||
|
|
TO_EMAIL = "contact@omegakube.fr"
|
||
|
|
FROM_EMAIL = "noreply@omegakube.fr"
|
||
|
|
SMTP_HOST = "localhost" # ⚠️ À MODIFIER selon votre hébergeur
|
||
|
|
SMTP_PORT = 25 # ⚠️ Port SMTP (25, 465, 587)
|
||
|
|
SMTP_SSL = False # True pour SSL direct (port 465), False pour STARTTLS (port 587)
|
||
|
|
SMTP_USER = "" # ⚠️ Nom d'utilisateur si authentification requise
|
||
|
|
SMTP_PASSWORD = "" # ⚠️ Mot de passe si authentification requise
|
||
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
||
|
|
|
||
|
|
def cgi_response(status_code, status_text, body, extra_headers=None):
|
||
|
|
"""Émet une réponse CGI complète (Status en premier, ligne vide obligatoire)."""
|
||
|
|
print(f"Status: {status_code} {status_text}")
|
||
|
|
if extra_headers:
|
||
|
|
for h in extra_headers:
|
||
|
|
print(h)
|
||
|
|
print("Content-Type: application/json; charset=utf-8")
|
||
|
|
print()
|
||
|
|
print(json.dumps(body, ensure_ascii=False))
|
||
|
|
sys.exit(0)
|
||
|
|
|
||
|
|
# ─── CORS ─────────────────────────────────────────────────────────────────────
|
||
|
|
ALLOWED_ORIGINS = ["https://omegakube.fr", "https://www.omegakube.fr"]
|
||
|
|
|
||
|
|
origin = os.environ.get("HTTP_ORIGIN", "")
|
||
|
|
referer = os.environ.get("HTTP_REFERER", "")
|
||
|
|
|
||
|
|
if origin:
|
||
|
|
if origin not in ALLOWED_ORIGINS:
|
||
|
|
cgi_response(403, "Forbidden", {"success": False, "error": "Origine non autorisée"})
|
||
|
|
cors_origin = origin
|
||
|
|
elif referer:
|
||
|
|
host = urlparse(referer).netloc
|
||
|
|
if host not in ("omegakube.fr", "www.omegakube.fr"):
|
||
|
|
cgi_response(403, "Forbidden", {"success": False, "error": "Référent non autorisé"})
|
||
|
|
cors_origin = "https://omegakube.fr"
|
||
|
|
else:
|
||
|
|
cors_origin = "https://omegakube.fr" # curl/test direct → on accepte
|
||
|
|
|
||
|
|
cors_headers = [
|
||
|
|
f"Access-Control-Allow-Origin: {cors_origin}",
|
||
|
|
"Access-Control-Allow-Methods: POST, OPTIONS",
|
||
|
|
"Access-Control-Allow-Headers: Content-Type",
|
||
|
|
]
|
||
|
|
|
||
|
|
# ─── PREFLIGHT OPTIONS ────────────────────────────────────────────────────────
|
||
|
|
method = os.environ.get("REQUEST_METHOD", "GET")
|
||
|
|
if method == "OPTIONS":
|
||
|
|
print("Status: 204 No Content")
|
||
|
|
for h in cors_headers:
|
||
|
|
print(h)
|
||
|
|
print()
|
||
|
|
sys.exit(0)
|
||
|
|
|
||
|
|
if method != "POST":
|
||
|
|
cgi_response(405, "Method Not Allowed", {"success": False, "error": "Méthode POST requise"}, cors_headers)
|
||
|
|
|
||
|
|
# ─── IP CLIENT ────────────────────────────────────────────────────────────────
|
||
|
|
ip = (os.environ.get("HTTP_CF_CONNECTING_IP")
|
||
|
|
or os.environ.get("HTTP_X_FORWARDED_FOR", "").split(",")[0].strip()
|
||
|
|
or os.environ.get("REMOTE_ADDR", "unknown"))
|
||
|
|
|
||
|
|
# ─── RATE LIMITING ────────────────────────────────────────────────────────────
|
||
|
|
DB_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "rate_limits")
|
||
|
|
DB_FILE = os.path.join(DB_DIR, "rate_limit.db")
|
||
|
|
|
||
|
|
def check_rate_limit(client_ip):
|
||
|
|
try:
|
||
|
|
os.makedirs(DB_DIR, exist_ok=True)
|
||
|
|
conn = sqlite3.connect(DB_FILE)
|
||
|
|
cur = conn.cursor()
|
||
|
|
cur.execute("CREATE TABLE IF NOT EXISTS submits (ip TEXT, ts REAL)")
|
||
|
|
conn.commit()
|
||
|
|
now = time.time()
|
||
|
|
cur.execute("DELETE FROM submits WHERE ts < ?", (now - 60,))
|
||
|
|
conn.commit()
|
||
|
|
cur.execute("SELECT COUNT(*) FROM submits WHERE ip = ?", (client_ip,))
|
||
|
|
if cur.fetchone()[0] >= 3:
|
||
|
|
conn.close()
|
||
|
|
return False, "Trop de requêtes. Attendez une minute avant de réessayer."
|
||
|
|
cur.execute("SELECT MAX(ts) FROM submits WHERE ip = ?", (client_ip,))
|
||
|
|
last = cur.fetchone()[0]
|
||
|
|
if last and (now - last) < 5:
|
||
|
|
conn.close()
|
||
|
|
return False, "Veuillez patienter quelques secondes entre chaque envoi."
|
||
|
|
cur.execute("INSERT INTO submits (ip, ts) VALUES (?, ?)", (client_ip, now))
|
||
|
|
conn.commit()
|
||
|
|
conn.close()
|
||
|
|
return True, ""
|
||
|
|
except Exception:
|
||
|
|
return True, "" # SQLite indisponible → on laisse passer
|
||
|
|
|
||
|
|
ok, msg = check_rate_limit(ip)
|
||
|
|
if not ok:
|
||
|
|
cgi_response(429, "Too Many Requests", {"success": False, "error": msg}, cors_headers)
|
||
|
|
|
||
|
|
# ─── LECTURE JSON ─────────────────────────────────────────────────────────────
|
||
|
|
try:
|
||
|
|
length = int(os.environ.get("CONTENT_LENGTH", 0))
|
||
|
|
data = json.loads(sys.stdin.read(length) if length > 0 else "{}")
|
||
|
|
except Exception:
|
||
|
|
cgi_response(400, "Bad Request", {"success": False, "error": "Données JSON invalides"}, cors_headers)
|
||
|
|
|
||
|
|
if not data:
|
||
|
|
cgi_response(400, "Bad Request", {"success": False, "error": "Aucune donnée reçue"}, cors_headers)
|
||
|
|
|
||
|
|
# ─── VALIDATION ───────────────────────────────────────────────────────────────
|
||
|
|
def clean(s, max_len=None):
|
||
|
|
s = re.sub(r"[\r\n]|%0[adAD]", "", str(s)).strip()
|
||
|
|
return s[:max_len] if max_len else s
|
||
|
|
|
||
|
|
# Honeypot
|
||
|
|
if str(data.get("website", "")):
|
||
|
|
cgi_response(200, "OK", {"success": True, "message": "Message envoyé !"}, cors_headers)
|
||
|
|
|
||
|
|
name = clean(data.get("name", ""), 100)
|
||
|
|
email = clean(data.get("email", ""), 254).lower()
|
||
|
|
subject = clean(data.get("subject", ""), 150)
|
||
|
|
message = str(data.get("message", ""))[:5000].strip()
|
||
|
|
|
||
|
|
errors = []
|
||
|
|
if len(name) < 2:
|
||
|
|
errors.append("Le nom doit contenir au moins 2 caractères.")
|
||
|
|
if not re.match(r"^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$", email):
|
||
|
|
errors.append("L'adresse email n'est pas valide.")
|
||
|
|
if len(message) < 10:
|
||
|
|
errors.append("Le message doit contenir au moins 10 caractères.")
|
||
|
|
if errors:
|
||
|
|
cgi_response(400, "Bad Request", {"success": False, "errors": errors}, cors_headers)
|
||
|
|
|
||
|
|
# ─── EMAIL ────────────────────────────────────────────────────────────────────
|
||
|
|
sep = "=" * 40
|
||
|
|
subject_line = f"[OmegaKube] {name} — {datetime.now().strftime('%d/%m/%Y %H:%M')}"
|
||
|
|
if subject:
|
||
|
|
subject_line += f" ({subject})"
|
||
|
|
|
||
|
|
body = f"OmegaKube — Nouveau message de contact\n{sep}\n\n"
|
||
|
|
body += f"Nom : {name}\nEmail : {email}\nObjet : {subject or 'Non précisé'}\n\n"
|
||
|
|
body += f"{sep}\n\nMESSAGE :\n\n{message}\n\n{sep}\n\n"
|
||
|
|
body += f"Répondre : {email}\nReçu le : {datetime.now().strftime('%d/%m/%Y %H:%M')}\n"
|
||
|
|
body += f"IP : {ip}\n\n{sep}\nOmegaKube — omegakube.fr\n"
|
||
|
|
|
||
|
|
def send_smtp(to_addr, subj, content, from_addr, reply_to):
|
||
|
|
"""Envoie un email via SMTP avec gestion d'erreur détaillée."""
|
||
|
|
msg = MIMEText(content, "plain", "utf-8")
|
||
|
|
msg["Subject"] = Header(subj, "utf-8")
|
||
|
|
msg["From"] = f"OmegaKube <{from_addr}>"
|
||
|
|
msg["To"] = to_addr
|
||
|
|
msg["Reply-To"] = reply_to
|
||
|
|
|
||
|
|
srv = None
|
||
|
|
try:
|
||
|
|
# Log de tentative de connexion
|
||
|
|
print(f"[send_mail] Tentative de connexion SMTP: host={SMTP_HOST}, port={SMTP_PORT}, ssl={SMTP_SSL}, user={bool(SMTP_USER)}", file=sys.stderr)
|
||
|
|
|
||
|
|
if SMTP_SSL:
|
||
|
|
srv = smtplib.SMTP_SSL(SMTP_HOST, SMTP_PORT, timeout=10)
|
||
|
|
else:
|
||
|
|
srv = smtplib.SMTP(SMTP_HOST, SMTP_PORT, timeout=10)
|
||
|
|
srv.starttls()
|
||
|
|
|
||
|
|
# Authentification si nécessaire
|
||
|
|
if SMTP_USER and SMTP_PASSWORD:
|
||
|
|
print(f"[send_mail] Authentification SMTP...", file=sys.stderr)
|
||
|
|
srv.login(SMTP_USER, SMTP_PASSWORD)
|
||
|
|
|
||
|
|
print(f"[send_mail] Envoi de l'email à: {to_addr}", file=sys.stderr)
|
||
|
|
srv.sendmail(from_addr, [to_addr], msg.as_string())
|
||
|
|
srv.quit()
|
||
|
|
print(f"[send_mail] Email envoyé avec succès !", file=sys.stderr)
|
||
|
|
return True, None
|
||
|
|
|
||
|
|
except smtplib.SMTPConnectError as e:
|
||
|
|
error_msg = f"Impossible de se connecter au serveur SMTP. Vérifiez la configuration: host={SMTP_HOST}, port={SMTP_PORT}"
|
||
|
|
print(f"[send_mail] SMTPConnectError: {e}", file=sys.stderr)
|
||
|
|
return False, error_msg
|
||
|
|
except smtplib.SMTPAuthenticationError as e:
|
||
|
|
error_msg = "Authentification SMTP échouée. Vérifiez SMTP_USER et SMTP_PASSWORD."
|
||
|
|
print(f"[send_mail] SMTPAuthenticationError: {e}", file=sys.stderr)
|
||
|
|
return False, error_msg
|
||
|
|
except smtplib.SMTPException as e:
|
||
|
|
error_msg = f"Erreur SMTP: {str(e)}"
|
||
|
|
print(f"[send_mail] SMTPException: {e}", file=sys.stderr)
|
||
|
|
return False, error_msg
|
||
|
|
except TimeoutError as e:
|
||
|
|
error_msg = f"Timeout lors de la connexion au serveur SMTP ({SMTP_HOST}:{SMTP_PORT}). Vérifiez que le serveur est accessible."
|
||
|
|
print(f"[send_mail] TimeoutError: {e}", file=sys.stderr)
|
||
|
|
return False, error_msg
|
||
|
|
except Exception as e:
|
||
|
|
error_msg = f"Erreur inattendue lors de l'envoi: {str(e)}"
|
||
|
|
print(f"[send_mail] Exception: {e}", file=sys.stderr)
|
||
|
|
return False, error_msg
|
||
|
|
finally:
|
||
|
|
if srv:
|
||
|
|
try:
|
||
|
|
srv.quit()
|
||
|
|
except:
|
||
|
|
pass
|
||
|
|
|
||
|
|
# ─── TEST DE CONNEXION SMTP (optionnel) ────────────────────────────────────────
|
||
|
|
def test_smtp_connection():
|
||
|
|
"""Teste la connexion SMTP avant l'envoi (pour le débogage)."""
|
||
|
|
try:
|
||
|
|
if SMTP_SSL:
|
||
|
|
srv = smtplib.SMTP_SSL(SMTP_HOST, SMTP_PORT, timeout=5)
|
||
|
|
else:
|
||
|
|
srv = smtplib.SMTP(SMTP_HOST, SMTP_PORT, timeout=5)
|
||
|
|
srv.starttls()
|
||
|
|
if SMTP_USER and SMTP_PASSWORD:
|
||
|
|
srv.login(SMTP_USER, SMTP_PASSWORD)
|
||
|
|
srv.quit()
|
||
|
|
return True
|
||
|
|
except Exception as e:
|
||
|
|
print(f"[send_mail] TEST SMTP échoué: {e}", file=sys.stderr)
|
||
|
|
return False
|
||
|
|
|
||
|
|
# ─── RÉPONSE ──────────────────────────────────────────────────────────────────
|
||
|
|
success, error_msg = send_smtp(TO_EMAIL, subject_line, body, FROM_EMAIL, email)
|
||
|
|
if success:
|
||
|
|
cgi_response(200, "OK",
|
||
|
|
{"success": True, "message": "Votre message a bien été envoyé ! Nous vous répondrons rapidement."},
|
||
|
|
cors_headers)
|
||
|
|
else:
|
||
|
|
# Message d'erreur détaillé mais sécurisé (pas d'info sensible)
|
||
|
|
user_error = error_msg or "Erreur lors de l'envoi. Vérifiez la configuration SMTP ou contactez-nous directement sur Discord."
|
||
|
|
cgi_response(500, "Internal Server Error",
|
||
|
|
{"success": False, "error": user_error},
|
||
|
|
cors_headers)
|