90f3a2ac3d
- 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>
568 lines
22 KiB
Python
568 lines
22 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Serveur unifié du portail Infra (CMDB + Certificats + Switch + Mots de passe).
|
||
|
||
Un seul serveur, lancé à la racine du projet :
|
||
|
||
python3 server.py # http://127.0.0.1:8000 → CMDB
|
||
python3 server.py 8080 # port custom
|
||
|
||
- http://127.0.0.1:8000/ → redirige vers la CMDB
|
||
- 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 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, 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
|
||
|
||
_arg = sys.argv[1].strip() if len(sys.argv) > 1 else ""
|
||
PORT = int(_arg) if _arg else int(os.environ.get("PORT", 8000))
|
||
|
||
HERE = os.path.dirname(os.path.abspath(__file__)) # racine projet
|
||
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)
|
||
import cmdb_server as C # ne démarre rien (guarded __main__)
|
||
|
||
SAFE = re.compile(r"^[A-Za-z0-9._@\- ]+$")
|
||
|
||
|
||
def safe_name(n):
|
||
n = (n or "").strip()
|
||
if not n or n in (".", "..") or "/" in n or "\\" in n or not SAFE.match(n):
|
||
raise ValueError("nom invalide: %r" % n)
|
||
return n
|
||
|
||
|
||
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
|
||
|
||
# ---------- helpers ----------
|
||
def _json(self, obj, code=200):
|
||
body = json.dumps(obj).encode("utf-8")
|
||
self.send_response(code)
|
||
self.send_header("Content-Type", "application/json")
|
||
self.send_header("Content-Length", str(len(body)))
|
||
self.send_header("Cache-Control", "no-store")
|
||
self.end_headers()
|
||
self.wfile.write(body)
|
||
|
||
def _err(self, msg, code=400):
|
||
self._json({"ok": False, "error": str(msg)}, code)
|
||
|
||
def _send_db(self, code, data, version):
|
||
self.send_response(code)
|
||
self.send_header("Content-Type", "application/octet-stream")
|
||
self.send_header("X-DB-Version", str(version))
|
||
self.send_header("Cache-Control", "no-store")
|
||
self.send_header("Content-Length", str(len(data)))
|
||
self.end_headers()
|
||
if data:
|
||
self.wfile.write(data)
|
||
|
||
def _send_download(self, data, filename):
|
||
self.send_response(200)
|
||
self.send_header("Content-Type", "application/octet-stream")
|
||
self.send_header("Content-Disposition", 'attachment; filename="%s"' % filename)
|
||
self.send_header("Cache-Control", "no-store")
|
||
self.send_header("Content-Length", str(len(data)))
|
||
self.end_headers()
|
||
self.wfile.write(data)
|
||
|
||
def _body(self):
|
||
n = int(self.headers.get("Content-Length", 0))
|
||
return json.loads(self.rfile.read(n) or b"{}")
|
||
|
||
def _cert_api(self):
|
||
p = self.path.split("?", 1)[0]
|
||
if "/cert-generator/api/" in p:
|
||
return p.split("/api/", 1)[1]
|
||
return None
|
||
|
||
# ---------- GET ----------
|
||
def do_GET(self):
|
||
# racine → CMDB
|
||
if self.path in ("/", "/index.html"):
|
||
self.send_response(302)
|
||
self.send_header("Location", "/cmdb/cmdb.html")
|
||
self.end_headers()
|
||
return
|
||
# base CMDB partagée
|
||
if self.path == "/db":
|
||
with C.DB_LOCK:
|
||
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?"):
|
||
name = (parse_qs(urlparse(self.path).query).get("name") or [""])[0]
|
||
try:
|
||
data = C.snapshot_bytes(name)
|
||
except (ValueError, OSError):
|
||
return self._json({"error": "snapshot introuvable"}, 404)
|
||
return self._send_download(data, os.path.basename(name))
|
||
# API certs (lecture)
|
||
api = self._cert_api()
|
||
if api == "ping":
|
||
return self._json({"ok": True, "workdir": "LISTS_ALL_CERT"})
|
||
if api == "list":
|
||
return self._json(self.cert_list())
|
||
return super().do_GET()
|
||
|
||
# ---------- POST ----------
|
||
def do_POST(self):
|
||
# base CMDB
|
||
if self.path == "/db":
|
||
data = self.rfile.read(int(self.headers.get("Content-Length", 0)))
|
||
base = int(self.headers.get("X-Base-Version", "0") or 0)
|
||
cur = C.db_version()
|
||
if base != cur:
|
||
return self._send_db(409, C.db_bytes(), cur)
|
||
return self._send_db(200, b"", C.db_write_and_snapshot(data))
|
||
# snapshots CMDB (création / suppression / restauration)
|
||
if self.path in ("/db/snapshot", "/db/snapshot/delete", "/db/restore"):
|
||
body = self._body()
|
||
if self.path == "/db/snapshot":
|
||
meta = C.create_snapshot(label=body.get("label") or "manual")
|
||
if not meta:
|
||
return self._json({"error": "base vide — rien à sauvegarder"}, 400)
|
||
return self._json(meta)
|
||
if self.path == "/db/snapshot/delete":
|
||
try:
|
||
C.delete_snapshot(body.get("name") or "")
|
||
except (ValueError, OSError):
|
||
return self._json({"error": "nom de snapshot invalide"}, 400)
|
||
return self._json({"ok": True})
|
||
try:
|
||
C.restore_snapshot(body.get("name") or "")
|
||
except (ValueError, FileNotFoundError):
|
||
return self._json({"error": "snapshot introuvable"}, 404)
|
||
with C.DB_LOCK:
|
||
return self._send_db(200, C.db_bytes(), C.db_version())
|
||
# LDAP (délégué à cmdb_server)
|
||
if self.path in ("/ldap/search", "/ldap/computer"):
|
||
if not C.HAS_LDAP:
|
||
return self._json(
|
||
{"error": "module ldap3 manquant — pip install ldap3"}, 500
|
||
)
|
||
try:
|
||
body = json.loads(
|
||
self.rfile.read(int(self.headers.get("Content-Length", 0)))
|
||
)
|
||
fn = C.ldap_computer if self.path == "/ldap/computer" else C.ldap_search
|
||
if self.path == "/ldap/search":
|
||
return self._json(
|
||
fn(
|
||
body.get("config") or {},
|
||
(body.get("query") or "").strip(),
|
||
int(body.get("limit") or 25),
|
||
)
|
||
)
|
||
return self._json(
|
||
fn(body.get("config") or {}, (body.get("query") or "").strip())
|
||
)
|
||
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:
|
||
if api == "write":
|
||
return self.cert_write_text()
|
||
if api == "write_bin":
|
||
return self.cert_write_bin()
|
||
if api == "delete":
|
||
return self.cert_delete_dir()
|
||
if api == "delete_archives":
|
||
return self.cert_delete_archives()
|
||
except Exception as e:
|
||
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()
|
||
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()
|
||
dname = safe_name(d["dir"])
|
||
target = os.path.join(WORK, dname)
|
||
os.makedirs(target, exist_ok=True)
|
||
for fname, content in (d.get("files") or {}).items():
|
||
with open(
|
||
os.path.join(target, safe_name(fname)),
|
||
"w",
|
||
encoding="utf-8",
|
||
newline="",
|
||
) as f:
|
||
f.write(content)
|
||
invalidate_cert_cache()
|
||
self._json({"ok": True, "path": "LISTS_ALL_CERT/%s/" % dname})
|
||
|
||
def cert_write_bin(self):
|
||
d = self._body()
|
||
parts = [safe_name(x) for x in d["path"].split("/") if x != ""]
|
||
if not parts:
|
||
raise ValueError("path vide")
|
||
target = os.path.join(WORK, *parts)
|
||
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):
|
||
d = self._body()
|
||
dname = safe_name(d["dir"])
|
||
target = os.path.join(WORK, dname)
|
||
if os.path.isdir(target):
|
||
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):
|
||
low = dname.lower()
|
||
if os.path.isdir(ARCH):
|
||
for a in os.listdir(ARCH):
|
||
if low in a.lower():
|
||
try:
|
||
os.remove(os.path.join(ARCH, a))
|
||
except OSError:
|
||
pass
|
||
|
||
|
||
if __name__ == "__main__":
|
||
ensure_work()
|
||
os.chdir(HERE)
|
||
if not C.HAS_LDAP:
|
||
print("⚠ ldap3 absent : recherche AD de la CMDB désactivée (pip install ldap3)")
|
||
print("Portail Infra : http://127.0.0.1:%d → CMDB" % PORT)
|
||
print("CMDB base : %s" % C.DB_PATH)
|
||
print("Certificats : %s (créé automatiquement)" % WORK)
|
||
print("Ctrl+C pour arrêter.")
|
||
ThreadingHTTPServer(("0.0.0.0", PORT), Handler).serve_forever()
|