2fb3621053
Rename CMDB -> cmdb (submodule), GENERATEUR_CERTIFICATES -> cert-generator, GENERATEUR_SWITCH -> switch-config Remove deleted CMDB files, switch config examples, proxy.sh, and pycache artifacts Update all internal path references across HTML, JS, and Python files
174 lines
6.5 KiB
Python
174 lines
6.5 KiB
Python
#!/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, 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__)) # .../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')
|
|
|
|
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/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()
|