Dashboard_Admin_Sys/cert-generator/cert_server.py

#!/usr/bin/env python3
"""
Serveur du Générateur de Certificats SSL.

Remplace `python -m http.server` : sert les pages ET écrit les certificats
sur le disque dans cert-generator/LISTS_ALL_CERT/ (créé au besoin),
sans aucune sélection de dossier côté navigateur.

Usage :
    python3 cert_server.py            # http://127.0.0.1:8000/cert-generator/dashboard.html
    python3 cert_server.py 8080       # port custom

Sert la RACINE du projet (parent), donc les 3 apps + le switcher fonctionnent.
"""

import sys, os, json, base64, shutil
from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer

PORT = int(sys.argv[1]) if len(sys.argv) > 1 else 8000

SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))  # .../cert-generator
ROOT = os.path.dirname(SCRIPT_DIR)  # racine projet
WORK = os.path.join(SCRIPT_DIR, "LISTS_ALL_CERT")  # dossier de travail (disque)
ARCH = os.path.join(WORK, "archives")

# Blacklist plutôt que whitelist : on autorise tout nom (DN avec « , = » et
# espaces, y compris insécables/Unicode venant de Windows) SAUF ce qui permet
# de sortir du dossier ou de casser le FS.
def safe_name(n):
    n = (n or "").strip()
    if (not n or n in (".", "..")
            or "/" in n or "\\" in n
            or "\x00" in n
            or any(ord(c) < 32 or ord(c) == 0x7f for c in n)):  # caractères de contrôle
        raise ValueError("nom invalide: %r" % n)
    return n


def ensure_work():
    os.makedirs(WORK, exist_ok=True)


class Handler(SimpleHTTPRequestHandler):
    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 _body(self):
        n = int(self.headers.get("Content-Length", 0))
        return json.loads(self.rfile.read(n) or b"{}")

    def _api(self):
        p = self.path.split("?", 1)[0]
        return p.split("/api/", 1)[1] if "/api/" in p else None

    # ---------- GET : API list/ping, sinon fichiers statiques ----------
    def do_GET(self):
        api = self._api()
        if api == "ping":
            return self._json({"ok": True, "workdir": "LISTS_ALL_CERT"})
        if api == "list":
            return self._json(self.list_all())
        return super().do_GET()

    def do_POST(self):
        api = self._api()
        try:
            if api == "write":
                return self.write_text()
            if api == "write_bin":
                return self.write_bin()
            if api == "delete":
                return self.delete_dir()
            if api == "delete_file":
                return self.delete_file()
            if api == "delete_archives":
                return self.delete_archives()
        except Exception as e:
            return self._err(e, 500)
        self._err("route inconnue: %s" % self.path, 404)

    # ---------- implémentations ----------
    def list_all(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}

    def 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():
            fn = safe_name(fname)
            with open(os.path.join(target, fn), "w", encoding="utf-8", newline="") as f:
                f.write(content)
        self._json({"ok": True, "path": "LISTS_ALL_CERT/%s/" % dname})

    def write_bin(self):
        d = self._body()
        # path = "archives/<fichier>.zip" OU "<dir>/<fichier>"
        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"]))
        self._json({"ok": True, "path": "LISTS_ALL_CERT/%s" % "/".join(parts)})

    def 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)
        self._json({"ok": True})

    def delete_file(self):
        d = self._body()
        target = os.path.join(WORK, safe_name(d["dir"]), safe_name(d["name"]))
        if os.path.isfile(target):
            os.remove(target)
        self._json({"ok": True})

    def delete_archives(self):
        d = self._body()
        self._rm_archives(safe_name(d["dir"]))
        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

    def log_message(self, *a):
        pass  # silencieux


if __name__ == "__main__":
    ensure_work()
    os.chdir(ROOT)
    url = "http://127.0.0.1:%d/cert-generator/dashboard.html" % PORT
    print("Générateur Certificats : %s" % url)
    print("Dossier disque : %s  (créé automatiquement)" % WORK)
    print("Ctrl+C pour arrêter.")
    ThreadingHTTPServer(("0.0.0.0", PORT), Handler).serve_forever()