feat(password-generator): envoi de mails avec templates HTML, images et pièces jointes

- Page de génération : sélection de template, variables {{ }} éditables
  (pré-remplies via AD), aperçu du rendu HTML
- Serveur : routes /mail/send (smtplib), upload d'images et de fichiers ;
  images inlinées en data URI, fichiers ajoutés en pièces jointes à l'envoi
- Rendu email-safe partagé (banner, listes, code, liens)
- gitignore des contenus utilisateur (images/, files/)
- Lien du générateur ajouté à la navigation des autres apps + README
- Bump du sous-module cmdb (éditeur de templates)

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Guillaume DENIS 2026-07-08 07:39:33 +02:00
parent 05df4c0dd8
commit 90f3a2ac3d
6 changed files with 1558 additions and 45 deletions

4
.gitignore vendored
View file

@ -9,6 +9,10 @@ cmdb/snapshots/
# Certificats générés (contiennent des clés privées)
cert-generator/LISTS_ALL_CERT/
# Images et fichiers importés dans les templates email (contenu utilisateur)
password-generator/images/
password-generator/files/
# === Environnement / build ===
.venv/
__pycache__/

View file

@ -5,8 +5,9 @@ Portail d'outils pour administrateurs systèmes et réseau, servi par un serveur
- **cmdb** — inventaire (machines, réseaux, IP/MAC, salles & baies), recherche Active Directory, import/export CSV, snapshots de la base
- **cert-generator** — génération de certificats SSL (clés RSA, CSR)
- **switch-config** — mapper de VLAN et export de configuration switch
- **password-generator** — génération de mots de passe aléatoires, passphrases et mots de passe temporaires (préfixe configurable, envoi par mail via l'AD + templates)
Le tout accessible depuis un seul serveur, avec navigation entre les 3 apps.
Le tout accessible depuis un seul serveur, avec navigation entre les apps.
## Prérequis

View file

@ -184,6 +184,7 @@
<a href="../cmdb/cmdb.html"><div class="psw-ico"></div><div class="psw-txt"><b>CMDB</b><small>Machines, réseaux, salles</small></div></a>
<a class="active" href="dashboard.html"><div class="psw-ico">🔐</div><div class="psw-txt"><b>Générateur certificats</b><small>Clés RSA, CSR, SSL</small></div></a>
<a href="../switch-config/switch_vlan_mapper.html"><div class="psw-ico">🔌</div><div class="psw-txt"><b>Configuration switch</b><small>VLANs, export config</small></div></a>
<a href="../password-generator/password_generator.html"><div class="psw-ico">🔑</div><div class="psw-txt"><b>Mots de passe</b><small>Générateur &amp; envoi</small></div></a>
</div>
<div class="psw" style="margin-top:auto;padding-top:14px;border-top:1px solid var(--line)">
<a href="../cmdb/preferences.html"><div class="psw-ico"></div><div class="psw-txt"><b>Préférences</b><small>Types, tickets, réglages</small></div></a>

2
cmdb

@ -1 +1 @@
Subproject commit 18751d75c88a32790a50c1f0a458a2657db6a2ad
Subproject commit 805f0fe00b1314970a95c717c2945102130a0fd4

File diff suppressed because it is too large Load diff

355
server.py
View file

@ -1,6 +1,6 @@
#!/usr/bin/env python3
"""
Serveur unifié du portail Infra (CMDB + Certificats + Switch).
Serveur unifié du portail Infra (CMDB + Certificats + Switch + Mots de passe).
Un seul serveur, lancé à la racine du projet :
@ -11,11 +11,15 @@ Un seul serveur, lancé à la racine du projet :
- CMDB : base partagée cmdb.db sourcée automatiquement (/db, versionnée)
- Générateur de certificats : écrit sur le disque dans
cert-generator/LISTS_ALL_CERT/ (créé au besoin), aucun prompt
- Sert les 3 applications + le switcher (liens relatifs OK)
- Sert les 4 applications + le switcher (liens relatifs OK)
- Recherche AD/LDAP de la CMDB conservée (nécessite `pip install ldap3`)
- Générateur de mots de passe : envoi de mails via POST /mail/send (SMTP
configuré dans la page, module standard smtplib)
"""
import sys, os, json, base64, shutil, re, threading
import sys, os, json, base64, shutil, re, smtplib, threading, mimetypes, sqlite3
from email.message import EmailMessage
from email.utils import formataddr
from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import urlparse, parse_qs
@ -27,6 +31,13 @@ CMDB_DIR = os.path.join(HERE, "cmdb")
CERT_DIR = os.path.join(HERE, "cert-generator")
WORK = os.path.join(CERT_DIR, "LISTS_ALL_CERT") # dossier de travail certs (disque)
ARCH = os.path.join(WORK, "archives")
PWGEN_DIR = os.path.join(HERE, "password-generator")
PWGEN_IMG = os.path.join(PWGEN_DIR, "images") # images des templates email (disque)
PWGEN_IMG_URL = "/password-generator/images/" # préfixe d'URL servi (chdir HERE)
PWGEN_FILES = os.path.join(PWGEN_DIR, "files") # fichiers joignables (rdp, pdf…)
PWGEN_FILES_URL = "/password-generator/files/"
MAX_IMG_BYTES = 2 * 1024 * 1024 # 2 Mo par image
MAX_FILE_BYTES = 10 * 1024 * 1024 # 10 Mo par fichier joint
# réutilise la logique CMDB existante (base partagée cmdb.db + LDAP)
sys.path.insert(0, CMDB_DIR)
@ -46,6 +57,175 @@ def ensure_work():
os.makedirs(WORK, exist_ok=True)
def read_vlans():
"""Lit les VLANs depuis la table `networks` de la base CMDB partagée
(cmdb.db). Un réseau devient un VLAN dès qu'il porte un numéro de VLAN.
Renvoie une liste triée : [{"vlan": int, "name": str, "color": str}, ...].
Source de vérité unique pour le mapper de switch (remplace vlan_database.js)."""
out = []
try:
con = sqlite3.connect("file:%s?mode=ro" % C.DB_PATH, uri=True)
except sqlite3.Error:
return out
try:
con.row_factory = sqlite3.Row
for r in con.execute(
"SELECT vlan, name, color FROM networks "
"WHERE vlan IS NOT NULL ORDER BY vlan"
):
out.append(
{
"vlan": r["vlan"],
"name": r["name"] or ("VLAN_%s" % r["vlan"]),
"color": r["color"] or "#8B5CF6",
}
)
except sqlite3.Error:
pass # table absente / base illisible → liste vide, repli côté client
finally:
con.close()
return out
# ---------------------------------------------------------------------------
# Cache du listing des certificats.
# Les certs sont des fichiers sur disque (LISTS_ALL_CERT/<dir>/<fichiers>).
# Sans cache, chaque GET /api/list re-scanne tout le disque (N×M syscalls) →
# lent quand il y a beaucoup de certs. On mémorise le résultat et on ne le
# reconstruit que si le contenu a changé :
# - empreinte légère (mtime des sous-dossiers) → détecte ajout/suppression
# et toute écriture passant par le serveur (le mtime du dossier bouge) ;
# - invalidation explicite après chaque write/delete → couvre l'écrasement
# d'un fichier en place (même nom), que le mtime du dossier ne voit pas.
# ---------------------------------------------------------------------------
_CERT_CACHE = {"sig": None, "payload": None}
_CERT_CACHE_LOCK = threading.Lock()
def invalidate_cert_cache():
with _CERT_CACHE_LOCK:
_CERT_CACHE["sig"] = None
_CERT_CACHE["payload"] = None
def _cert_signature():
"""Empreinte du dossier certs : (nom, mtime_ns) de chaque sous-dossier.
Ne descend pas dans les fichiers peu de syscalls. Renvoie None si le
dossier est illisible (force alors une reconstruction)."""
sig = []
try:
with os.scandir(WORK) as it:
for e in it:
try:
sig.append((e.name, e.stat().st_mtime_ns))
except OSError:
continue
except OSError:
return None
sig.sort()
return tuple(sig)
def _build_cert_list():
"""Reconstruit le listing complet via os.scandir (le stat est replié dans la
lecture du dossier bien moins de syscalls que listdir + stat séparés)."""
certs, archives = [], []
try:
with os.scandir(WORK) as it:
entries = sorted(it, key=lambda e: e.name)
except OSError:
entries = []
for entry in entries:
if entry.name == "archives":
if entry.is_dir():
try:
with os.scandir(entry.path) as ait:
for a in sorted(ait, key=lambda e: e.name):
if a.is_file():
st = a.stat()
archives.append(
{
"name": a.name,
"size": st.st_size,
"mtime": int(st.st_mtime * 1000),
}
)
except OSError:
pass
continue
if not entry.is_dir():
continue
files, date = [], 0
try:
with os.scandir(entry.path) as sit:
subs = sorted(sit, key=lambda e: e.name)
except OSError:
continue
for f in subs:
if not f.is_file():
continue
st = f.stat()
m = int(st.st_mtime * 1000)
date = max(date, m)
files.append(
{
"name": f.name,
"ext": (f.name.rsplit(".", 1)[-1] if "." in f.name else "").lower(),
"size": st.st_size,
"mtime": m,
}
)
if files:
certs.append({"dir": entry.name, "files": files, "date": date})
archives.sort(key=lambda a: -a["mtime"])
certs.sort(key=lambda c: -c["date"])
return {"ok": True, "certs": certs, "archives": archives}
# src d'images pointant vers le dossier local des templates (chemin absolu ou relatif)
_LOCAL_IMG_SRC = re.compile(r'src="([^"]*?/password-generator/images/([^"?#]+))"', re.I)
def inline_local_images(html):
"""Remplace les <img src="…/password-generator/images/x.png"> par des data
URI (lecture du fichier sur disque). Les images data:/http(s): sont laissées
telles quelles. Rend le mail HTML autonome au moment de l'envoi."""
def repl(m):
fname = os.path.basename(m.group(2)) # anti-traversal
path = os.path.join(PWGEN_IMG, fname)
if not os.path.isfile(path):
return m.group(0)
mime = mimetypes.guess_type(path)[0] or "application/octet-stream"
with open(path, "rb") as f:
data = base64.b64encode(f.read()).decode("ascii")
return 'src="data:%s;base64,%s"' % (mime, data)
return _LOCAL_IMG_SRC.sub(repl, html)
def attach_files(msg, urls):
"""Joint au mail les fichiers du dossier password-generator/files/ dont l'URL
figure dans la liste `urls` (pièces jointes du template). Le nom présenté au
destinataire est le nom d'origine (suffixe aléatoire de stockage retiré)."""
seen = set()
for url in urls or []:
fname = os.path.basename(str(url)) # anti-traversal
if not fname or fname in seen:
continue
seen.add(fname)
path = os.path.join(PWGEN_FILES, fname)
if not os.path.isfile(path):
continue
ctype = mimetypes.guess_type(path)[0] or "application/octet-stream"
maintype, _, subtype = ctype.partition("/")
disp = re.sub(r"-[0-9a-f]{8}(\.[^.]+)$", r"\1", fname) # nom d'origine lisible
with open(path, "rb") as f:
msg.add_attachment(
f.read(), maintype=maintype, subtype=subtype or "octet-stream", filename=disp
)
class Handler(SimpleHTTPRequestHandler):
def log_message(self, *a):
pass
@ -106,6 +286,10 @@ class Handler(SimpleHTTPRequestHandler):
return self._send_db(200, C.db_bytes(), C.db_version())
if self.path == "/db/version":
return self._json({"version": C.db_version()})
# VLANs = réseaux CMDB (table networks) → consommés par le mapper switch
if self.path == "/api/vlans":
with C.DB_LOCK:
return self._json(read_vlans())
if self.path == "/db/snapshots":
return self._json({"snapshots": C.list_snapshots()})
if self.path.startswith("/db/snapshot?"):
@ -177,6 +361,23 @@ class Handler(SimpleHTTPRequestHandler):
)
except Exception as ex:
return self._json({"error": str(ex)}, 500)
# envoi de mail (générateur de mots de passe)
if self.path == "/mail/send":
try:
return self.mail_send()
except Exception as e:
return self._err(e, 500)
# upload d'image / de fichier joint (générateur de mots de passe)
if self.path == "/pwgen/upload-image":
try:
return self.pwgen_upload_image()
except Exception as e:
return self._err(e, 500)
if self.path == "/pwgen/upload-file":
try:
return self.pwgen_upload_file()
except Exception as e:
return self._err(e, 500)
# API certs (écriture)
api = self._cert_api()
try:
@ -192,49 +393,113 @@ class Handler(SimpleHTTPRequestHandler):
return self._err(e, 500)
self._err("route inconnue: %s" % self.path, 404)
# ---------- mail (générateur de mots de passe) ----------
def mail_send(self):
"""Envoie un mail texte via le serveur SMTP fourni par la page.
Body JSON : { smtp: {host, port, security, user, pass, from, from_name},
to, subject, body }
security : none | starttls | ssl
"""
d = self._body()
smtp = d.get("smtp") or {}
to = (d.get("to") or "").strip()
host = (smtp.get("host") or "").strip()
if not to or "@" not in to:
raise ValueError("destinataire invalide: %r" % to)
if not host:
raise ValueError("serveur SMTP non configuré")
security = (smtp.get("security") or "starttls").lower()
try:
port = int(smtp.get("port") or 0)
except (TypeError, ValueError):
port = 0
if not port:
port = {"ssl": 465, "starttls": 587}.get(security, 25)
sender = (smtp.get("from") or smtp.get("user") or "").strip()
if not sender:
raise ValueError("adresse expéditeur non configurée")
msg = EmailMessage()
from_name = (smtp.get("from_name") or "").strip()
msg["From"] = formataddr((from_name, sender)) if from_name else sender
msg["To"] = to
msg["Subject"] = d.get("subject") or ""
msg.set_content(d.get("body") or "")
html = (d.get("html") or "").strip()
if html:
# les images de templates stockées sur disque sont ré-intégrées en
# data URI pour que le mail reçu soit autonome (pas de lien vers le
# serveur, images visibles hors du réseau).
html = inline_local_images(html)
msg.add_alternative(html, subtype="html")
# pièces jointes du template (ex : .rdp de connexion)
attach_files(msg, d.get("attachments"))
if security == "ssl":
server = smtplib.SMTP_SSL(host, port, timeout=15)
else:
server = smtplib.SMTP(host, port, timeout=15)
try:
if security == "starttls":
server.starttls()
user = (smtp.get("user") or "").strip()
if user:
server.login(user, smtp.get("pass") or "")
server.send_message(msg)
finally:
try:
server.quit()
except Exception:
pass
self._json({"ok": True, "to": to})
# ---------- uploads de templates (disque réel) ----------
def _save_upload(self, dest_dir, url_prefix, max_bytes, default_ext, kind):
"""Stocke un fichier posté ({name, b64}) sur le disque avec un nom sûr
(radical assaini + suffixe aléatoire) et renvoie son URL servie."""
d = self._body()
raw = (d.get("name") or kind).strip()
try:
data = base64.b64decode(d.get("b64") or "", validate=True)
except (ValueError, TypeError):
raise ValueError("%s invalide (base64)" % kind)
if not data:
raise ValueError("%s vide" % kind)
if len(data) > max_bytes:
raise ValueError("%s trop lourd (max %d Mo)" % (kind, max_bytes // (1024 * 1024)))
base, _, ext = raw.rpartition(".")
ext = re.sub(r"[^A-Za-z0-9]", "", ext).lower()[:8] or default_ext
stem = re.sub(r"[^A-Za-z0-9._-]", "_", base or kind)[:40].strip("._-") or kind
fname = "%s-%s.%s" % (stem, os.urandom(4).hex(), ext)
os.makedirs(dest_dir, exist_ok=True)
with open(os.path.join(dest_dir, fname), "wb") as f:
f.write(data)
self._json({"ok": True, "url": url_prefix + fname, "name": fname})
def pwgen_upload_image(self):
"""Image de template → password-generator/images/ (ré-intégrée en data URI à l'envoi)."""
self._save_upload(PWGEN_IMG, PWGEN_IMG_URL, MAX_IMG_BYTES, "png", "image")
def pwgen_upload_file(self):
"""Fichier joignable (rdp, pdf…) → password-generator/files/ (joint au mail à l'envoi)."""
self._save_upload(PWGEN_FILES, PWGEN_FILES_URL, MAX_FILE_BYTES, "bin", "fichier")
# ---------- certificats (disque réel) ----------
def cert_list(self):
ensure_work()
certs, archives = [], []
for entry in sorted(os.listdir(WORK)):
full = os.path.join(WORK, entry)
if entry == "archives":
if os.path.isdir(full):
for a in sorted(os.listdir(full)):
af = os.path.join(full, a)
if os.path.isfile(af):
st = os.stat(af)
archives.append(
{
"name": a,
"size": st.st_size,
"mtime": int(st.st_mtime * 1000),
}
)
continue
if not os.path.isdir(full):
continue
files, date = [], 0
for fn in sorted(os.listdir(full)):
ff = os.path.join(full, fn)
if not os.path.isfile(ff):
continue
st = os.stat(ff)
m = int(st.st_mtime * 1000)
date = max(date, m)
files.append(
{
"name": fn,
"ext": (fn.rsplit(".", 1)[-1] if "." in fn else "").lower(),
"size": st.st_size,
"mtime": m,
}
)
if files:
certs.append({"dir": entry, "files": files, "date": date})
archives.sort(key=lambda a: -a["mtime"])
certs.sort(key=lambda c: -c["date"])
return {"ok": True, "certs": certs, "archives": archives}
sig = _cert_signature()
with _CERT_CACHE_LOCK:
if (
sig is not None
and sig == _CERT_CACHE["sig"]
and _CERT_CACHE["payload"] is not None
):
return _CERT_CACHE["payload"] # inchangé → réponse mémorisée
payload = _build_cert_list()
_CERT_CACHE["sig"] = sig
_CERT_CACHE["payload"] = payload
return payload
def cert_write_text(self):
d = self._body()
@ -249,6 +514,7 @@ class Handler(SimpleHTTPRequestHandler):
newline="",
) as f:
f.write(content)
invalidate_cert_cache()
self._json({"ok": True, "path": "LISTS_ALL_CERT/%s/" % dname})
def cert_write_bin(self):
@ -260,6 +526,7 @@ class Handler(SimpleHTTPRequestHandler):
os.makedirs(os.path.dirname(target), exist_ok=True)
with open(target, "wb") as f:
f.write(base64.b64decode(d["b64"]))
invalidate_cert_cache()
self._json({"ok": True, "path": "LISTS_ALL_CERT/%s" % "/".join(parts)})
def cert_delete_dir(self):
@ -270,10 +537,12 @@ class Handler(SimpleHTTPRequestHandler):
shutil.rmtree(target)
if d.get("archives"):
self._rm_archives(dname)
invalidate_cert_cache()
self._json({"ok": True})
def cert_delete_archives(self):
self._rm_archives(safe_name(self._body()["dir"]))
invalidate_cert_cache()
self._json({"ok": True})
def _rm_archives(self, dname):