Dashboard_Admin_Sys/GENERATEUR_CERTIFICATES/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 GENERATEUR_CERTIFICATES/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/GENERATEUR_CERTIFICATES/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, re
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__))      # .../GENERATEUR_CERTIFICATES
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')

SAFE = re.compile(r'^[A-Za-z0-9._@\- ]+$')                  # noms de dossier/fichier autorisés

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)

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/GENERATEUR_CERTIFICATES/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()