18751d75c8
Serveur Python (http.server + ldap3) servant cmdb.html, la base SQLite partagee avec versioning optimiste, snapshots horodates et recherche AD. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
549 lines
19 KiB
Python
549 lines
19 KiB
Python
#!/usr/bin/env python3
|
|
"""Serveur CMDB : sert cmdb.html, la base partagée et la recherche AD.
|
|
|
|
Remplace `python -m http.server` ET le connecteur séparé :
|
|
pip install ldap3
|
|
python3 cmdb_server.py # http://0.0.0.0:8000
|
|
python3 cmdb_server.py 8080 # port custom
|
|
|
|
Endpoints :
|
|
GET /db → base SQLite (bytes) + entête X-DB-Version
|
|
GET /db/version → {"version": n} (poll léger)
|
|
POST /db → écrit la base ; entête X-Base-Version pour la
|
|
concurrence optimiste (409 si obsolète)
|
|
POST /ldap/search → recherche annuaire (config envoyée par la page)
|
|
|
|
La base est un vrai fichier `cmdb.db` partagé : plusieurs personnes
|
|
peuvent l'utiliser en même temps, avec détection de conflit.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import re
|
|
import ssl
|
|
import sys
|
|
import threading
|
|
import time
|
|
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__))
|
|
DB_PATH = os.path.join(HERE, "cmdb.db")
|
|
VER_PATH = DB_PATH + ".ver"
|
|
DB_LOCK = threading.Lock()
|
|
|
|
|
|
def db_version():
|
|
try:
|
|
with open(VER_PATH) as f:
|
|
return int(f.read().strip() or 0)
|
|
except (OSError, ValueError):
|
|
return 1 if os.path.exists(DB_PATH) else 0
|
|
|
|
|
|
def db_bytes():
|
|
try:
|
|
with open(DB_PATH, "rb") as f:
|
|
return f.read()
|
|
except OSError:
|
|
return b""
|
|
|
|
|
|
def db_write(data):
|
|
"""Écrit la base atomiquement et incrémente la version. Renvoie la version."""
|
|
with DB_LOCK:
|
|
ver = db_version() + 1
|
|
tmp = DB_PATH + ".tmp"
|
|
with open(tmp, "wb") as f:
|
|
f.write(data)
|
|
os.replace(tmp, DB_PATH)
|
|
with open(VER_PATH, "w") as f:
|
|
f.write(str(ver))
|
|
return ver
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Snapshots : copies horodatées de la base dans snapshots/.
|
|
# - « auto » : créées après une écriture, au plus 1/heure, rétention 72.
|
|
# - « manuel » : créées à la demande (bouton), rétention 100, label libre.
|
|
# Restaurer un snapshot crée d'abord un snapshot de sécurité de l'état courant.
|
|
# ---------------------------------------------------------------------------
|
|
SNAP_DIR = os.path.join(HERE, "snapshots")
|
|
# nom : cmdb-<AAAAMMJJ>-<HHMMSS>-v<version>[-<label>].db (pas de '/' possible → pas de traversal)
|
|
SNAP_RE = re.compile(r"^cmdb-(\d{8})-(\d{6})-v(\d+)(?:-([A-Za-z0-9_.-]+))?\.db$")
|
|
AUTO_MIN_INTERVAL = 3600 # s : au plus un snapshot auto par heure
|
|
AUTO_KEEP = 72 # rétention snapshots auto (~3 jours horaires)
|
|
MANUAL_KEEP = 100 # rétention snapshots manuels
|
|
|
|
|
|
def _sanitize_label(label):
|
|
label = re.sub(r"[^A-Za-z0-9_.-]+", "-", (label or "").strip()).strip("-")
|
|
return label[:40]
|
|
|
|
|
|
def _parse_snap(name):
|
|
"""Valide un nom de snapshot et renvoie sa méta, ou None si invalide."""
|
|
name = os.path.basename(name or "")
|
|
m = SNAP_RE.match(name)
|
|
if not m:
|
|
return None
|
|
return {
|
|
"name": name,
|
|
"date": m.group(1),
|
|
"time": m.group(2),
|
|
"version": int(m.group(3)),
|
|
"label": m.group(4) or "",
|
|
"auto": m.group(4) == "auto",
|
|
}
|
|
|
|
|
|
def list_snapshots():
|
|
"""Liste les snapshots existants, plus récents d'abord."""
|
|
out = []
|
|
try:
|
|
entries = os.listdir(SNAP_DIR)
|
|
except OSError:
|
|
return out
|
|
for n in entries:
|
|
meta = _parse_snap(n)
|
|
if not meta:
|
|
continue
|
|
try:
|
|
st = os.stat(os.path.join(SNAP_DIR, n))
|
|
except OSError:
|
|
continue
|
|
meta["size"] = st.st_size
|
|
meta["mtime"] = int(st.st_mtime * 1000)
|
|
out.append(meta)
|
|
out.sort(key=lambda s: s["mtime"], reverse=True)
|
|
return out
|
|
|
|
|
|
def _prune(auto, keep):
|
|
for s in [s for s in list_snapshots() if s["auto"] == auto][keep:]:
|
|
try:
|
|
os.remove(os.path.join(SNAP_DIR, s["name"]))
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
def create_snapshot(label="manual", auto=False):
|
|
"""Copie la base courante dans snapshots/. Renvoie la méta, ou None si base vide."""
|
|
with DB_LOCK:
|
|
data = db_bytes()
|
|
if not data:
|
|
return None
|
|
ver = db_version()
|
|
os.makedirs(SNAP_DIR, exist_ok=True)
|
|
lbl = "auto" if auto else (_sanitize_label(label) or "manual")
|
|
name = "cmdb-%s-v%d-%s.db" % (time.strftime("%Y%m%d-%H%M%S"), ver, lbl)
|
|
path = os.path.join(SNAP_DIR, name)
|
|
tmp = path + ".tmp"
|
|
with open(tmp, "wb") as f:
|
|
f.write(data)
|
|
os.replace(tmp, path)
|
|
_prune(auto, AUTO_KEEP if auto else MANUAL_KEEP)
|
|
return _parse_snap(name)
|
|
|
|
|
|
def _maybe_auto_snapshot():
|
|
"""Snapshot auto si le dernier date de plus d'AUTO_MIN_INTERVAL. Jamais dans DB_LOCK."""
|
|
autos = [s for s in list_snapshots() if s["auto"]]
|
|
if autos and (time.time() - autos[0]["mtime"] / 1000) < AUTO_MIN_INTERVAL:
|
|
return
|
|
try:
|
|
create_snapshot(auto=True)
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
def db_write_and_snapshot(data):
|
|
"""Écrit la base puis déclenche un snapshot auto (throttlé). Renvoie la version."""
|
|
ver = db_write(data)
|
|
_maybe_auto_snapshot()
|
|
return ver
|
|
|
|
|
|
def snapshot_bytes(name):
|
|
"""Contenu d'un snapshot. Lève si nom invalide ou fichier absent."""
|
|
meta = _parse_snap(name)
|
|
if not meta:
|
|
raise ValueError("nom de snapshot invalide")
|
|
with open(os.path.join(SNAP_DIR, meta["name"]), "rb") as f:
|
|
return f.read()
|
|
|
|
|
|
def delete_snapshot(name):
|
|
meta = _parse_snap(name)
|
|
if not meta:
|
|
raise ValueError("nom de snapshot invalide")
|
|
path = os.path.join(SNAP_DIR, meta["name"])
|
|
if os.path.isfile(path):
|
|
os.remove(path)
|
|
|
|
|
|
def restore_snapshot(name):
|
|
"""Remplace la base courante par le snapshot `name`.
|
|
|
|
Crée d'abord un snapshot de sécurité de l'état courant. Renvoie la nouvelle version.
|
|
"""
|
|
meta = _parse_snap(name)
|
|
if not meta:
|
|
raise ValueError("nom de snapshot invalide")
|
|
path = os.path.join(SNAP_DIR, meta["name"])
|
|
if not os.path.isfile(path):
|
|
raise FileNotFoundError("snapshot introuvable")
|
|
with open(path, "rb") as f:
|
|
data = f.read()
|
|
create_snapshot(label="pre-restore") # filet de sécurité avant écrasement
|
|
return db_write(data)
|
|
|
|
|
|
try:
|
|
from ldap3 import Server, Connection, Tls, SUBTREE, ALL_ATTRIBUTES
|
|
from ldap3.utils.conv import escape_filter_chars
|
|
|
|
HAS_LDAP = True
|
|
except ImportError:
|
|
HAS_LDAP = False
|
|
|
|
|
|
def _cn_of(dn):
|
|
if not dn:
|
|
return ""
|
|
head = dn.split(",")[0]
|
|
return head.split("=", 1)[1] if "=" in head else head
|
|
|
|
|
|
def ldap_connect(cfg):
|
|
"""Ouvre et bind une connexion LDAP selon la config de la page."""
|
|
use_ssl = cfg.get("method") == "ssl"
|
|
tls = Tls(
|
|
validate=ssl.CERT_REQUIRED if cfg.get("verify_certificates") else ssl.CERT_NONE
|
|
)
|
|
srv = Server(
|
|
cfg["host"],
|
|
port=int(cfg.get("port") or (636 if use_ssl else 389)),
|
|
use_ssl=use_ssl,
|
|
tls=tls,
|
|
get_info=None,
|
|
connect_timeout=8,
|
|
)
|
|
conn = Connection(
|
|
srv,
|
|
user=cfg.get("bind_dn") or None,
|
|
password=cfg.get("bind_password") or None,
|
|
auto_referrals=False,
|
|
receive_timeout=15,
|
|
)
|
|
if cfg.get("method") == "start_tls":
|
|
conn.open()
|
|
conn.start_tls()
|
|
if not conn.bind():
|
|
raise RuntimeError("Bind LDAP refusé : %s" % conn.result)
|
|
return conn
|
|
|
|
|
|
def ldap_computer(cfg, query):
|
|
"""Cherche un poste (objectClass=computer) par nom/dNSHostName."""
|
|
conn = ldap_connect(cfg)
|
|
q = escape_filter_chars((query or "").strip().rstrip("$"))
|
|
flt = (
|
|
"(&(objectClass=computer)(|(cn=*%s*)(name=*%s*)(dNSHostName=*%s*)"
|
|
"(sAMAccountName=*%s*)))" % (q, q, q, q)
|
|
)
|
|
base = cfg.get("base") or ""
|
|
ok = conn.search(base, flt, SUBTREE, attributes=ALL_ATTRIBUTES, size_limit=5)
|
|
res = conn.result or {}
|
|
if not ok and res.get("result") not in (0, 4):
|
|
raise RuntimeError(
|
|
"Recherche LDAP échouée : %s" % (res.get("description") or res)
|
|
)
|
|
out = []
|
|
for e in conn.entries:
|
|
d = e.entry_attributes_as_dict
|
|
|
|
def val(k):
|
|
for kk in d:
|
|
if kk.lower() == k.lower():
|
|
v = d[kk]
|
|
if isinstance(v, list):
|
|
v = v[0] if v else ""
|
|
return str(v) if v else ""
|
|
return ""
|
|
|
|
def vals(k):
|
|
for kk in d:
|
|
if kk.lower() == k.lower():
|
|
v = d[kk]
|
|
return [str(x) for x in (v if isinstance(v, list) else [v]) if x]
|
|
return []
|
|
|
|
uac = val("userAccountControl")
|
|
enabled = (not (int(uac) & 2)) if uac.isdigit() else None
|
|
out.append(
|
|
{
|
|
"name": val("cn") or val("name"),
|
|
"dns": val("dNSHostName"),
|
|
"os": val("operatingSystem"),
|
|
"os_version": val("operatingSystemVersion"),
|
|
"description": val("description"),
|
|
"location": val("location"),
|
|
"managed_by": _cn_of(val("managedBy")),
|
|
"when_created": val("whenCreated")[:8],
|
|
"enabled": enabled,
|
|
"groups": [_cn_of(x) for x in vals("memberOf")][:20],
|
|
}
|
|
)
|
|
return out
|
|
|
|
|
|
def ldap_search(cfg, query, limit=25):
|
|
"""Recherche d'utilisateurs dans l'AD selon la config de la page.
|
|
|
|
Cherche par username, nom, prénom, nom affiché ou mail.
|
|
Lève une exception explicite si le bind ou la recherche échoue
|
|
(sinon un problème de config renverrait « aucun résultat » en silence).
|
|
"""
|
|
conn = ldap_connect(cfg)
|
|
|
|
uid = cfg.get("uid") or "userPrincipalName"
|
|
raw = cfg.get("filter") or ""
|
|
if raw:
|
|
flt = raw.replace("{query}", escape_filter_chars(query))
|
|
else:
|
|
# recherche prioritairement sur displayName, multi-mots (ET) → "jean dupont"
|
|
# et "dupont jean" fonctionnent tous les deux, quel que soit l'ordre.
|
|
search_attrs = [
|
|
"displayName",
|
|
"cn",
|
|
"sAMAccountName",
|
|
"givenName",
|
|
"sn",
|
|
"userPrincipalName",
|
|
"mail",
|
|
]
|
|
tokens = [escape_filter_chars(t) for t in query.split() if t] or [""]
|
|
parts = []
|
|
for t in tokens:
|
|
ors = "".join("(%s=*%s*)" % (a, t) for a in search_attrs)
|
|
parts.append("(|%s)" % ors)
|
|
flt = "(&%s)" % "".join(parts) if len(parts) > 1 else parts[0]
|
|
# personnes uniquement, compatible AD + OpenLDAP
|
|
person = (
|
|
"(|(objectClass=user)(objectClass=inetOrgPerson)"
|
|
"(objectClass=organizationalPerson)(objectClass=person))"
|
|
)
|
|
base = cfg.get("base") or ""
|
|
|
|
ok = conn.search(
|
|
base,
|
|
"(&%s%s)" % (person, flt),
|
|
SUBTREE,
|
|
attributes=ALL_ATTRIBUTES,
|
|
size_limit=limit,
|
|
)
|
|
res = conn.result or {}
|
|
# 0 = success, 4 = sizeLimitExceeded (résultats partiels OK)
|
|
if not ok and res.get("result") not in (0, 4):
|
|
raise RuntimeError(
|
|
"Recherche LDAP échouée : %s (base=%s)"
|
|
% (res.get("description") or res, base or "∅")
|
|
)
|
|
|
|
def val(d, k):
|
|
# accès insensible à la casse, première valeur
|
|
for kk in d:
|
|
if kk.lower() == k.lower():
|
|
v = d[kk]
|
|
if isinstance(v, list):
|
|
v = v[0] if v else ""
|
|
return str(v) if v else ""
|
|
return ""
|
|
|
|
def vals(d, k):
|
|
for kk in d:
|
|
if kk.lower() == k.lower():
|
|
v = d[kk]
|
|
return [str(x) for x in (v if isinstance(v, list) else [v]) if x]
|
|
return []
|
|
|
|
def cn_of(dn):
|
|
# "CN=Jean Dupont,OU=..." -> "Jean Dupont"
|
|
if not dn:
|
|
return ""
|
|
head = dn.split(",")[0]
|
|
return head.split("=", 1)[1] if "=" in head else head
|
|
|
|
out, seen = [], set()
|
|
for e in conn.entries:
|
|
d = e.entry_attributes_as_dict
|
|
name = (
|
|
val(d, "displayName")
|
|
or val(d, "cn")
|
|
or (val(d, "givenName") + " " + val(d, "sn")).strip()
|
|
)
|
|
username = (
|
|
val(d, "sAMAccountName") or val(d, uid) or val(d, "userPrincipalName")
|
|
)
|
|
key = (name, username)
|
|
if not (name or username) or key in seen:
|
|
continue
|
|
seen.add(key)
|
|
uac = val(d, "userAccountControl")
|
|
enabled = (not (int(uac) & 2)) if uac.isdigit() else None
|
|
out.append(
|
|
{
|
|
"displayName": name or username,
|
|
"username": username,
|
|
"mail": val(d, "mail"),
|
|
"title": val(d, "title"),
|
|
"department": val(d, "department"),
|
|
"company": val(d, "company"),
|
|
"office": val(d, "physicalDeliveryOfficeName"),
|
|
"phone": val(d, "telephoneNumber"),
|
|
"mobile": val(d, "mobile") or val(d, "ipPhone"),
|
|
"manager": cn_of(val(d, "manager")),
|
|
"description": val(d, "description"),
|
|
"employee_id": val(d, "employeeID") or val(d, "employeeNumber"),
|
|
"division": val(d, "division"),
|
|
"city": val(d, "l"),
|
|
"country": val(d, "co") or val(d, "c"),
|
|
"enabled": enabled,
|
|
"groups": [cn_of(x) for x in vals(d, "memberOf")][:15],
|
|
}
|
|
)
|
|
out.sort(key=lambda u: u["displayName"].lower())
|
|
return out
|
|
|
|
|
|
class Handler(SimpleHTTPRequestHandler):
|
|
def log_message(self, *args):
|
|
pass
|
|
|
|
def _json(self, code, payload):
|
|
body = json.dumps(payload).encode()
|
|
self.send_response(code)
|
|
self.send_header("Content-Type", "application/json")
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
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 do_GET(self):
|
|
if self.path == "/db":
|
|
with DB_LOCK:
|
|
return self._send_db(200, db_bytes(), db_version())
|
|
if self.path == "/db/version":
|
|
return self._json(200, {"version": db_version()})
|
|
if self.path == "/db/snapshots":
|
|
return self._json(200, {"snapshots": list_snapshots()})
|
|
if self.path.startswith("/db/snapshot?"):
|
|
name = (parse_qs(urlparse(self.path).query).get("name") or [""])[0]
|
|
try:
|
|
data = snapshot_bytes(name)
|
|
except (ValueError, OSError):
|
|
return self._json(404, {"error": "snapshot introuvable"})
|
|
return self._send_download(data, os.path.basename(name))
|
|
if self.path in ("/", "/index.html"):
|
|
self.path = "/cmdb.html"
|
|
return super().do_GET()
|
|
|
|
def do_POST(self):
|
|
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 = db_version()
|
|
if base != cur: # quelqu'un a écrit entre-temps
|
|
return self._send_db(409, db_bytes(), cur)
|
|
ver = db_write_and_snapshot(data)
|
|
return self._send_db(200, b"", ver)
|
|
if self.path in ("/db/snapshot", "/db/snapshot/delete", "/db/restore"):
|
|
try:
|
|
body = json.loads(
|
|
self.rfile.read(int(self.headers.get("Content-Length", 0)) or 0)
|
|
or b"{}"
|
|
)
|
|
except ValueError:
|
|
body = {}
|
|
if self.path == "/db/snapshot":
|
|
meta = create_snapshot(label=body.get("label") or "manual")
|
|
if not meta:
|
|
return self._json(400, {"error": "base vide — rien à sauvegarder"})
|
|
return self._json(200, meta)
|
|
if self.path == "/db/snapshot/delete":
|
|
try:
|
|
delete_snapshot(body.get("name") or "")
|
|
except (ValueError, OSError):
|
|
return self._json(400, {"error": "nom de snapshot invalide"})
|
|
return self._json(200, {"ok": True})
|
|
# /db/restore
|
|
try:
|
|
restore_snapshot(body.get("name") or "")
|
|
except (ValueError, FileNotFoundError):
|
|
return self._json(404, {"error": "snapshot introuvable"})
|
|
with DB_LOCK:
|
|
return self._send_db(200, db_bytes(), db_version())
|
|
if self.path == "/ldap/computer":
|
|
if not HAS_LDAP:
|
|
return self._json(
|
|
500, {"error": "module ldap3 manquant — pip install ldap3"}
|
|
)
|
|
try:
|
|
body = json.loads(
|
|
self.rfile.read(int(self.headers.get("Content-Length", 0)))
|
|
)
|
|
result = ldap_computer(
|
|
body.get("config") or {}, (body.get("query") or "").strip()
|
|
)
|
|
return self._json(200, result)
|
|
except Exception as ex:
|
|
return self._json(500, {"error": str(ex)})
|
|
if self.path != "/ldap/search":
|
|
return self._json(404, {"error": "not found"})
|
|
if not HAS_LDAP:
|
|
return self._json(
|
|
500, {"error": "module ldap3 manquant — pip install ldap3"}
|
|
)
|
|
try:
|
|
body = json.loads(
|
|
self.rfile.read(int(self.headers.get("Content-Length", 0)))
|
|
)
|
|
limit = int(body.get("limit") or 25)
|
|
result = ldap_search(
|
|
body.get("config") or {}, (body.get("query") or "").strip(), limit
|
|
)
|
|
self._json(200, result)
|
|
except Exception as ex: # renvoyé tel quel à la page
|
|
self._json(500, {"error": str(ex)})
|
|
|
|
|
|
if __name__ == "__main__":
|
|
os.chdir(os.path.dirname(os.path.abspath(__file__)))
|
|
if not HAS_LDAP:
|
|
print(
|
|
"⚠ ldap3 absent : la recherche AD renverra une erreur (pip install ldap3)"
|
|
)
|
|
print("CMDB : http://127.0.0.1:%d (Ctrl+C pour arrêter)" % PORT)
|
|
ThreadingHTTPServer(("0.0.0.0", PORT), Handler).serve_forever()
|