Initial commit: CMDB dashboard
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>
This commit is contained in:
Guillaume DENIS
commit
18751d75c8
7 changed files with 8589 additions and 0 deletions
40
.gitignore
vendored
Normal file
40
.gitignore
vendored
Normal file
|
|
@ -0,0 +1,40 @@
|
|||
# Database (runtime data)
|
||||
cmdb.db
|
||||
cmdb.db.ver
|
||||
*.db
|
||||
*.db-journal
|
||||
*.db-wal
|
||||
*.db-shm
|
||||
|
||||
# Snapshots (generated backups)
|
||||
snapshots/
|
||||
|
||||
# Python
|
||||
__pycache__/
|
||||
*.py[cod]
|
||||
*.egg-info/
|
||||
.eggs/
|
||||
build/
|
||||
dist/
|
||||
|
||||
# Virtual environments
|
||||
.venv/
|
||||
venv/
|
||||
env/
|
||||
|
||||
# uv
|
||||
.uv/
|
||||
|
||||
# Env / secrets
|
||||
.env
|
||||
.env.*
|
||||
*.secret
|
||||
|
||||
# IDE
|
||||
.vscode/
|
||||
.idea/
|
||||
*.swp
|
||||
|
||||
# OS
|
||||
.DS_Store
|
||||
Thumbs.db
|
||||
76
README.md
Normal file
76
README.md
Normal file
|
|
@ -0,0 +1,76 @@
|
|||
<div align="center">
|
||||
<img src="sources/logo.png" alt="Manage OpenLDAP Logo" width="220"/>
|
||||
|
||||
<h1>Noctâme</h1>
|
||||
<p><i>CMDB légère : inventaire de parc et recherche Active Directory</i></p>
|
||||
</div>
|
||||
|
||||
---
|
||||
|
||||
## À propos
|
||||
|
||||
Noctâme est une CMDB minimaliste. Un seul serveur Python sert l'interface
|
||||
(`cmdb.html`), partage une base SQLite entre plusieurs utilisateurs et
|
||||
interroge l'annuaire Active Directory.
|
||||
|
||||
## Démarrage
|
||||
|
||||
Prérequis : [mise](https://mise.jdx.dev/) (gère la version de Python) et
|
||||
[uv](https://docs.astral.sh/uv/) (gère les dépendances).
|
||||
|
||||
```bash
|
||||
mise install # installe Python (>=3.13)
|
||||
uv sync # installe les dépendances (ldap3)
|
||||
|
||||
uv run cmdb_server.py # http://0.0.0.0:8000
|
||||
uv run cmdb_server.py 8080 # port custom
|
||||
```
|
||||
|
||||
## Fonctionnalités
|
||||
|
||||
- Inventaire du parc dans une base SQLite partagée (`cmdb.db`)
|
||||
- Détection de conflit à l'écriture (concurrence optimiste)
|
||||
- Snapshots automatiques et manuels de la base
|
||||
- Recherche Active Directory / LDAP intégrée
|
||||
|
||||
---
|
||||
|
||||
## Annexe
|
||||
|
||||
<details>
|
||||
<summary>Génération du logo</summary>
|
||||
|
||||
</br>
|
||||
|
||||
1. **Prompt pour ChatGPT (DALL-E)** :
|
||||
|
||||
```
|
||||
A high-quality, original creature inspired by Pokémon style, unevolved stage (basic form). The creature has balanced proportions (can be cute, neutral, or slightly serious), with a clean and memorable silhouette, expressive eyes (not overly exaggerated), and a distinct personality that can range from friendly to calm, mysterious, or determined.
|
||||
|
||||
The creature has a unique elemental theme (choose one: fire, water, electric, grass, ghost, etc.), with subtle and well-integrated visual cues reflecting its type (glowing sparks, leaves, flames, mist, energy patterns, etc.).
|
||||
|
||||
Art style: official Pokémon-style illustration, soft but controlled shading, vibrant yet natural color palette, smooth gradients, clean and precise outlines, slightly stylized but grounded anatomy.
|
||||
|
||||
Pose: simple, confident, and readable, facing slightly toward the viewer, with a composed or expressive stance depending on its personality.
|
||||
|
||||
Background: transparent.
|
||||
|
||||
Lighting: soft studio lighting with gentle contrast to enhance form and details.
|
||||
|
||||
Composition: centered, full body visible, no cropping.
|
||||
|
||||
Aspect ratio: 1:1
|
||||
|
||||
High resolution, sharp details, no text, no watermark.
|
||||
|
||||
Theme: Astro, Star, Milkyway
|
||||
|
||||
Format d'image : **1:1**
|
||||
```
|
||||
|
||||
2. **Supprimer le fond** : [PhotoRoom Background Remover](https://www.photoroom.com/tools/background-remover)
|
||||
|
||||
3. **Size Logo :** 220x200
|
||||
|
||||
</details>
|
||||
|
||||
549
cmdb_server.py
Normal file
549
cmdb_server.py
Normal file
|
|
@ -0,0 +1,549 @@
|
|||
#!/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()
|
||||
3860
preferences.html
Normal file
3860
preferences.html
Normal file
File diff suppressed because it is too large
Load diff
9
pyproject.toml
Normal file
9
pyproject.toml
Normal file
|
|
@ -0,0 +1,9 @@
|
|||
[project]
|
||||
name = "cmdb"
|
||||
version = "0.1.0"
|
||||
description = "Add your description here"
|
||||
readme = "README.md"
|
||||
requires-python = ">=3.13"
|
||||
dependencies = [
|
||||
"ldap3>=2.9.1",
|
||||
]
|
||||
BIN
sources/logo.png
Normal file
BIN
sources/logo.png
Normal file
Binary file not shown.
|
After Width: | Height: | Size: 33 KiB |
Loading…
Add table
Add a link
Reference in a new issue