TheHackdes/Dashboard_Admin_Sys
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
Dashboard_Admin_Sys/server.py
Guillaume DENIS 05df4c0dd8 chore(server): reformat code and improve mise tooling config 
- Normalise le style (guillemets doubles, black/ruff)
- mise.toml : ajoute [tools], tâches dev/clean, port par défaut
- server.py : fallback PORT via variable d'environnement
2026-07-04 18:17:22 +02:00

299 lines
11 KiB
Python
Raw Blame History

#!/usr/bin/env python3
"""
Serveur unifié du portail Infra (CMDB + Certificats + Switch).
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 3 applications + le switcher (liens relatifs OK)
- Recherche AD/LDAP de la CMDB conservée (nécessite `pip install ldap3`)
"""
import sys, os, json, base64, shutil, re, threading
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")
# 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)
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()})
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)
# 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)
# ---------- 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}
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)
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"]))
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)
self._json({"ok": True})
def cert_delete_archives(self):
self._rm_archives(safe_name(self._body()["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
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()
Reference in a new issue View git blame Copy permalink